istio stats: remove the dependency to the direction of factory context (#46906)

Commit Message: istio stats: remove the dependency to the direction of
factory context
Additional Description:

Now, the the istio stats won't depends on the factory context's
trafficDirection(), like what we did at the
#46893

After all these filters be updated, we could remove the direction() from
the FactoryContext and the these filters also could works correctly at
cross-listeners shared filter chain (FCDS).

Risk Level: low.
Testing: n/a.
Docs Changes: n/a.
Release Notes: n/a.
Platform Specific Features: n/a.

---------

Signed-off-by: wbpcode <wbphub@gmail.com>

Author: wbpcode

contrib/istio/filters/http/istio_stats/source/istio_stats.cc

@@ -114,6 +114,10 @@ enum class Reporter {
   ServerSidecar,
   // Gateway listener for a set of destination workloads.
   ServerGateway,
+  // The configuration does not specify the
+  // reporter type and the traffic direction
+  // will be used to infer the reporter type.
+  Unspecified,
 };
 
 // Detect if peer info read is completed by TCP metadata exchange.
@@ -474,30 +478,22 @@ struct MetricOverrides : public Logger::Loggable<Logger::Id::filter> {
 
 struct Config : public Logger::Loggable<Logger::Id::filter> {
   Config(const stats::PluginConfig& proto_config,
-         Server::Configuration::FactoryContext& factory_context)
-      : context_(factory_context.serverFactoryContext().singletonManager().getTyped<Context>(
+         Server::Configuration::ServerFactoryContext& context, Stats::Scope& stats_scope)
+      : context_(context.singletonManager().getTyped<Context>(
             SINGLETON_MANAGER_REGISTERED_NAME(Context),
-            [&factory_context] {
-              return std::make_shared<Context>(factory_context.serverFactoryContext().scope(),
-                                               factory_context.serverFactoryContext().localInfo());
+            [&context] {
+              return std::make_shared<Context>(context.scope(), context.localInfo());
             })),
         disable_host_header_fallback_(proto_config.disable_host_header_fallback()),
         report_duration_(
             PROTOBUF_GET_MS_OR_DEFAULT(proto_config, tcp_reporting_duration, /* 5s */ 5000)) {
-    recordVersion(factory_context);
+    recordVersion(stats_scope);
     reporter_ = Reporter::ClientSidecar;
     switch (proto_config.reporter()) {
     case stats::Reporter::UNSPECIFIED:
-      switch (factory_context.direction()) {
-      case envoy::config::core::v3::TrafficDirection::INBOUND:
-        reporter_ = Reporter::ServerSidecar;
-        break;
-      case envoy::config::core::v3::TrafficDirection::OUTBOUND:
-        reporter_ = Reporter::ClientSidecar;
-        break;
-      default:
-        break;
-      }
+      // Mark as unspecified to allow the filter to infer the reporter type based on traffic
+      // direction.
+      reporter_ = Reporter::Unspecified;
       break;
     case stats::Reporter::SERVER_GATEWAY:
       reporter_ = Reporter::ServerGateway;
@@ -743,14 +739,13 @@ struct Config : public Logger::Loggable<Logger::Id::filter> {
     bool evaluated_{false};
   };
 
-  void recordVersion(Server::Configuration::FactoryContext& factory_context) {
+  void recordVersion(Stats::Scope& scope) {
     Stats::StatNameTagVector tags;
     tags.push_back({context_->component_, context_->proxy_});
     tags.push_back({context_->tag_, context_->istio_version_.empty() ? context_->unknown_
                                                                      : context_->istio_version_});
 
-    Stats::Utility::gaugeFromStatNames(factory_context.scope(),
-                                       {context_->stat_namespace_, context_->istio_build_},
+    Stats::Utility::gaugeFromStatNames(scope, {context_->stat_namespace_, context_->istio_build_},
                                        Stats::Gauge::ImportMode::Accumulate, tags)
         .set(1);
   }
@@ -776,9 +771,46 @@ class IstioStatsFilter : public Http::PassThroughFilter,
 public:
   IstioStatsFilter(ConfigSharedPtr config)
       : config_(config), context_(*config->context_), pool_(config->scope().symbolTable()),
-        stream_(*config_, pool_) {
+        stream_(*config_, pool_) {}
+  ~IstioStatsFilter() override { ASSERT(report_timer_ == nullptr); }
+
+  // The reporter type is only set explicitly by gateways/waypoints. Sidecars leave it
+  // unspecified in the config and it is inferred here from the direction of the listener
+  // that owns this stream or connection.
+  Reporter resolveReporter() const {
+    ASSERT(decoder_callbacks_ != nullptr || network_read_callbacks_ != nullptr);
+    if (config_->reporter() != Reporter::Unspecified) {
+      return config_->reporter();
+    }
+
+    OptRef<const Network::ListenerInfo> listener_info;
+    if (decoder_callbacks_ != nullptr) {
+      listener_info = decoder_callbacks_->streamInfo().downstreamAddressProvider().listenerInfo();
+    } else if (network_read_callbacks_ != nullptr) {
+      listener_info = network_read_callbacks_->connection().connectionInfoProvider().listenerInfo();
+    }
+    if (listener_info.has_value()) {
+      switch (listener_info->direction()) {
+      case envoy::config::core::v3::TrafficDirection::INBOUND:
+        return Reporter::ServerSidecar;
+      case envoy::config::core::v3::TrafficDirection::OUTBOUND:
+        return Reporter::ClientSidecar;
+      default:
+        break;
+      }
+    }
+    // Retain the historical default for listeners without a traffic direction.
+    return Reporter::ClientSidecar;
+  }
+
+  // Resolved reporter type. Only valid once initializeTags() has run, which is the first
+  // thing both the HTTP and the network entry point do.
+  Reporter reporter() const { return reporter_; }
+
+  void initializeTags() {
+    reporter_ = resolveReporter();
     tags_.reserve(25);
-    switch (config_->reporter()) {
+    switch (reporter_) {
     case Reporter::ServerSidecar:
       tags_.push_back({context_.reporter_, context_.destination_});
       break;
@@ -788,11 +820,19 @@ class IstioStatsFilter : public Http::PassThroughFilter,
     case Reporter::ClientSidecar:
       tags_.push_back({context_.reporter_, context_.source_});
       break;
+    case Reporter::Unspecified:
+      // Unreachable: resolveReporter() never returns Unspecified.
+      IS_ENVOY_BUG("unresolved istio stats reporter");
+      tags_.push_back({context_.reporter_, context_.unknown_});
+      break;
     }
   }
-  ~IstioStatsFilter() override { ASSERT(report_timer_ == nullptr); }
 
   // Http::StreamDecoderFilter
+  void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override {
+    decoder_callbacks_ = &callbacks;
+    initializeTags();
+  }
   Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap& request_headers, bool) override {
     is_grpc_ = Grpc::Common::isGrpcRequestHeaders(request_headers);
     if (is_grpc_) {
@@ -866,6 +906,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
   }
   void initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) override {
     network_read_callbacks_ = &callbacks;
+    initializeTags();
     network_read_callbacks_->connection().addConnectionCallbacks(*this);
   }
   // Network::ConnectionCallbacks
@@ -893,7 +934,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
     if (decoder_callbacks_) {
       if (!peer_read_) {
         const auto& info = decoder_callbacks_->streamInfo();
-        peer_read_ = peerInfoRead(config_->reporter(), info.filterState());
+        peer_read_ = peerInfoRead(reporter(), info.filterState());
         if (peer_read_ || end_stream) {
           ENVOY_LOG(trace, "Populating peer metadata from HTTP MX.");
           populatePeerInfo(info, info.filterState());
@@ -922,7 +963,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
     const auto& info = network_read_callbacks_->connection().streamInfo();
     // TCP MX writes to upstream stream info instead.
     OptRef<const StreamInfo::UpstreamInfo> upstream_info;
-    if (config_->reporter() == Reporter::ClientSidecar) {
+    if (reporter() == Reporter::ClientSidecar) {
       upstream_info = info.upstreamInfo();
     }
     const StreamInfo::FilterState& filter_state =
@@ -931,7 +972,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
             : info.filterState();
 
     if (!peer_read_) {
-      peer_read_ = peerInfoRead(config_->reporter(), filter_state);
+      peer_read_ = peerInfoRead(reporter(), filter_state);
       // Report connection open once peer info is read or connection is closed.
       if (peer_read_ || end_stream) {
         ENVOY_LOG(trace, "Populating peer metadata from TCP MX.");
@@ -979,10 +1020,10 @@ class IstioStatsFilter : public Http::PassThroughFilter,
                         const StreamInfo::FilterState& filter_state) {
     // Compute peer info with client-side fallbacks.
     std::optional<Istio::Common::WorkloadMetadataObject> peer;
-    auto object = peerInfo(config_->reporter(), filter_state);
+    auto object = peerInfo(reporter(), filter_state);
     if (object) {
       peer.emplace(object.value());
-    } else if (config_->reporter() == Reporter::ClientSidecar) {
+    } else if (reporter() == Reporter::ClientSidecar) {
       if (auto label_obj = extractEndpointMetadata(info); label_obj) {
         peer.emplace(label_obj.value());
       }
@@ -1044,7 +1085,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
 
     std::string peer_san;
     absl::string_view local_san;
-    switch (config_->reporter()) {
+    switch (reporter()) {
     case Reporter::ServerSidecar:
     case Reporter::ServerGateway: {
       auto peer_principal =
@@ -1078,7 +1119,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
         peer_san = ssl_info->uriSanPeerCertificate()[0];
       }
       if (peer_san.empty()) {
-        auto endpoint_object = peerInfo(config_->reporter(), filter_state);
+        auto endpoint_object = peerInfo(reporter(), filter_state);
         if (endpoint_object) {
           endpoint_peer.emplace(endpoint_object.value());
           peer_san = endpoint_peer->identity_;
@@ -1091,6 +1132,8 @@ class IstioStatsFilter : public Http::PassThroughFilter,
       }
       break;
     }
+    case Reporter::Unspecified:
+      break;
     }
     // Implements fallback from using the namespace from SAN if available to
     // using peer metadata, otherwise.
@@ -1104,7 +1147,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
     if (peer_namespace.empty() && peer) {
       peer_namespace = peer->namespace_name_;
     }
-    switch (config_->reporter()) {
+    switch (reporter()) {
     case Reporter::ServerSidecar:
     case Reporter::ServerGateway: {
       tags_.push_back({context_.source_workload_, peer && !peer->workload_name_.empty()
@@ -1130,7 +1173,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
       tags_.push_back({context_.source_cluster_, peer && !peer->cluster_name_.empty()
                                                      ? pool_.add(peer->cluster_name_)
                                                      : context_.unknown_});
-      switch (config_->reporter()) {
+      switch (reporter()) {
       case Reporter::ServerGateway: {
         std::optional<Istio::Common::WorkloadMetadataObject> endpoint_peer;
         auto endpoint_object = peerInfo(Reporter::ClientSidecar, filter_state);
@@ -1256,8 +1299,9 @@ class IstioStatsFilter : public Http::PassThroughFilter,
   Context& context_;
   Stats::StatNameDynamicPool pool_;
   Stats::StatNameTagVector tags_;
+  Reporter reporter_{Reporter::ClientSidecar};
   Event::TimerPtr report_timer_{nullptr};
-  Network::ReadFilterCallbacks* network_read_callbacks_;
+  Network::ReadFilterCallbacks* network_read_callbacks_{nullptr};
   bool peer_read_{false};
   uint64_t bytes_sent_{0};
   uint64_t bytes_received_{0};
@@ -1272,12 +1316,12 @@ class IstioStatsFilter : public Http::PassThroughFilter,
 } // namespace
 
 absl::StatusOr<Http::FilterFactoryCb>
-IstioStatsFilterConfigFactory::createFilterFactoryFromProtoTyped(
-    const stats::PluginConfig& proto_config, const std::string&,
-    Server::Configuration::FactoryContext& factory_context) {
-  factory_context.serverFactoryContext().api().customStatNamespaces().registerStatNamespace(
-      CustomStatNamespace);
-  ConfigSharedPtr config = std::make_shared<Config>(proto_config, factory_context);
+IstioStatsFilterConfigFactory::createHttpFilterFactoryFromProtoTyped(
+    const stats::PluginConfig& proto_config, Server::Configuration::ServerFactoryContext& context,
+    Server::Configuration::ExtraFactoryContext& extra_context) {
+  context.api().customStatNamespaces().registerStatNamespace(CustomStatNamespace);
+  ConfigSharedPtr config =
+      std::make_shared<Config>(proto_config, context, extra_context.scopeOr(context));
   return [config](Http::FilterChainFactoryCallbacks& callbacks) {
     auto filter = std::make_shared<IstioStatsFilter>(config);
     callbacks.addStreamFilter(filter);
@@ -1295,8 +1339,9 @@ IstioStatsNetworkFilterConfigFactory::createFilterFactoryFromProto(
     const Protobuf::Message& proto_config, Server::Configuration::FactoryContext& factory_context) {
   factory_context.serverFactoryContext().api().customStatNamespaces().registerStatNamespace(
       CustomStatNamespace);
-  ConfigSharedPtr config = std::make_shared<Config>(
-      dynamic_cast<const stats::PluginConfig&>(proto_config), factory_context);
+  ConfigSharedPtr config =
+      std::make_shared<Config>(dynamic_cast<const stats::PluginConfig&>(proto_config),
+                               factory_context.serverFactoryContext(), factory_context.scope());
   return [config](Network::FilterManager& filter_manager) {
     filter_manager.addReadFilter(std::make_shared<IstioStatsFilter>(config));
   };

contrib/istio/filters/http/istio_stats/source/istio_stats.h

@@ -13,14 +13,15 @@ namespace Extensions {
 namespace HttpFilters {
 namespace IstioStats {
 
-class IstioStatsFilterConfigFactory : public Common::ExceptionFreeFactoryBase<stats::PluginConfig> {
+class IstioStatsFilterConfigFactory : public Common::UnifiedFactoryBase<stats::PluginConfig> {
 public:
-  IstioStatsFilterConfigFactory() : ExceptionFreeFactoryBase("envoy.filters.http.istio_stats") {}
+  IstioStatsFilterConfigFactory() : UnifiedFactoryBase("envoy.filters.http.istio_stats") {}
 
 private:
   absl::StatusOr<Http::FilterFactoryCb>
-  createFilterFactoryFromProtoTyped(const stats::PluginConfig& proto_config, const std::string&,
-                                    Server::Configuration::FactoryContext&) override;
+  createHttpFilterFactoryFromProtoTyped(const stats::PluginConfig& proto_config,
+                                        Server::Configuration::ServerFactoryContext&,
+                                        Server::Configuration::ExtraFactoryContext&) override;
 };
 
 class IstioStatsNetworkFilterConfigFactory