Network::FilterStatus Filter::initializeUpstreamConnection() {
  ASSERT(upstream_ == nullptr);

  route_ = pickRoute();

  const std::string& cluster_name = route_ ? route_->clusterName() : EMPTY_STRING;

  Upstream::ThreadLocalCluster* thread_local_cluster =
      cluster_manager_.getThreadLocalCluster(cluster_name);

  if (thread_local_cluster) {
    ENVOY_CONN_LOG(debug, "Creating connection to cluster {}", read_callbacks_->connection(),
                   cluster_name);
  } else {
    ENVOY_CONN_LOG(debug, "Cluster not found {}", read_callbacks_->connection(), cluster_name);
    config_->stats().downstream_cx_no_route_.inc();
    getStreamInfo().setResponseFlag(StreamInfo::ResponseFlag::NoClusterFound);
    onInitFailure(UpstreamFailureReason::NoRoute);
    return Network::FilterStatus::StopIteration;
  }

  Upstream::ClusterInfoConstSharedPtr cluster = thread_local_cluster->info();
  getStreamInfo().setUpstreamClusterInfo(cluster);

  // Check this here because the TCP conn pool will queue our request waiting for a connection that
  // will never be released.
  if (!cluster->resourceManager(Upstream::ResourcePriority::Default).connections().canCreate()) {
    getStreamInfo().setResponseFlag(StreamInfo::ResponseFlag::UpstreamOverflow);
    cluster->stats().upstream_cx_overflow_.inc();
    onInitFailure(UpstreamFailureReason::ResourceLimitExceeded);
    return Network::FilterStatus::StopIteration;
  }

  const uint32_t max_connect_attempts = config_->maxConnectAttempts();
  if (connect_attempts_ >= max_connect_attempts) {
    getStreamInfo().setResponseFlag(StreamInfo::ResponseFlag::UpstreamRetryLimitExceeded);
    cluster->stats().upstream_cx_connect_attempts_exceeded_.inc();
    onInitFailure(UpstreamFailureReason::ConnectFailed);
    return Network::FilterStatus::StopIteration;
  }

  if (auto downstream_connection = downstreamConnection(); downstream_connection != nullptr) {
    if (!read_callbacks_->connection()
             .streamInfo()
             .filterState()
             ->hasData<Network::ProxyProtocolFilterState>(
                 Network::ProxyProtocolFilterState::key())) {
      read_callbacks_->connection().streamInfo().filterState()->setData(
          Network::ProxyProtocolFilterState::key(),
          std::make_shared<Network::ProxyProtocolFilterState>(Network::ProxyProtocolData{
              downstream_connection->connectionInfoProvider().remoteAddress(),
              downstream_connection->connectionInfoProvider().localAddress()}),
          StreamInfo::FilterState::StateType::ReadOnly,
          StreamInfo::FilterState::LifeSpan::Connection);
    }
    transport_socket_options_ = Network::TransportSocketOptionsUtility::fromFilterState(
        downstream_connection->streamInfo().filterState());

    if (auto typed_state = downstream_connection->streamInfo()
                               .filterState()
                               .getDataReadOnly<Network::UpstreamSocketOptionsFilterState>(
                                   Network::UpstreamSocketOptionsFilterState::key());
        typed_state != nullptr) {
      auto downstream_options = typed_state->value();
      if (!upstream_options_) {
        upstream_options_ = std::make_shared<Network::Socket::Options>();
      }
      Network::Socket::appendOptions(upstream_options_, downstream_options);
    }
  }

  if (!maybeTunnel(*thread_local_cluster)) {
    // Either cluster is unknown or there are no healthy hosts. tcpConnPool() increments
    // cluster->stats().upstream_cx_none_healthy in the latter case.
    getStreamInfo().setResponseFlag(StreamInfo::ResponseFlag::NoHealthyUpstream);
    onInitFailure(UpstreamFailureReason::NoHealthyUpstream);
  }
  return Network::FilterStatus::StopIteration;
}