network_ext_proc: add more counters and tests (envoyproxy#39265)

Risk Level: low
Testing: unit & integration tests
Docs Changes:
Release Notes:
Platform Specific Features:

---------

Signed-off-by: Boteng Yao <[email protected]>

Author: botengyao

api/envoy/extensions/filters/network/ext_proc/v3/ext_proc.proto

@@ -34,6 +34,7 @@ option (xds.annotations.v3.file_status).work_in_progress = true;
 //
 // By using the filter's processing mode, you can selectively choose which data
 // directions to process (read, write or both), allowing for efficient processing.
+// [#next-free-field: 6]
 message NetworkExternalProcessor {
   // The gRPC service that will process network traffic.
   // This service must implement the NetworkExternalProcessor service
@@ -61,6 +62,8 @@ message NetworkExternalProcessor {
     lte {seconds: 3600}
     gte {}
   }];
+
+  string stat_prefix = 5 [(validate.rules).string = {min_len: 1}];
 }
 
 // Options for controlling processing behavior.

source/extensions/filters/network/ext_proc/config.cc

@@ -42,8 +42,8 @@ Network::FilterFactoryCb NetworkExtProcConfigFactory::createFilterFactoryFromPro
   if (!result.ok()) {
     throw EnvoyException(std::string(result.message()));
   }
-
-  ConfigConstSharedPtr ext_proc_config = std::make_shared<const Config>(proto_config);
+  ConfigConstSharedPtr ext_proc_config =
+      std::make_shared<const Config>(proto_config, context.scope());
 
   return [ext_proc_config, &context](Network::FilterManager& filter_manager) -> void {
     auto client = createExternalProcessorClient(

source/extensions/filters/network/ext_proc/ext_proc.cc

@@ -7,8 +7,8 @@ namespace ExtProc {
 
 NetworkExtProcFilter::NetworkExtProcFilter(ConfigConstSharedPtr config,
                                            ExternalProcessorClientPtr&& client)
-    : config_(config), client_(std::move(client)), config_with_hash_key_(config_->grpcService()),
-      downstream_callbacks_(*this) {}
+    : config_(config), stats_(config->stats()), client_(std::move(client)),
+      config_with_hash_key_(config_->grpcService()), downstream_callbacks_(*this) {}
 
 NetworkExtProcFilter::~NetworkExtProcFilter() { closeStream(); }
 
@@ -81,6 +81,7 @@ Network::FilterStatus NetworkExtProcFilter::handleStreamError() {
 
   if (config_->failureModeAllow()) {
     // In failure allow mode, continue processing despite stream errors
+    stats_.failure_mode_allowed_.inc();
     return Network::FilterStatus::Continue;
   } else {
     // In strict mode, close the connection and stop processing
@@ -140,10 +141,12 @@ NetworkExtProcFilter::StreamOpenState NetworkExtProcFilter::openStream() {
   if (stream_object == nullptr) {
     ENVOY_CONN_LOG(error, "Failed to create gRPC stream to external processor",
                    read_callbacks_->connection());
+    stats_.stream_open_failures_.inc();
     return StreamOpenState::Error;
   }
 
   stream_ = std::move(stream_object);
+  stats_.streams_started_.inc();
   return StreamOpenState::Ok;
 }
 
@@ -166,14 +169,17 @@ void NetworkExtProcFilter::sendRequest(Buffer::Instance& data, bool end_stream,
     auto* read_data = request.mutable_read_data();
     read_data->set_data(data.toString());
     read_data->set_end_of_stream(end_stream);
+    stats_.read_data_sent_.inc();
   } else {
     auto* write_data = request.mutable_write_data();
     write_data->set_data(data.toString());
     write_data->set_end_of_stream(end_stream);
+    stats_.write_data_sent_.inc();
   }
 
   // Send to external processor
   stream_->send(std::move(request), false);
+  stats_.stream_msgs_sent_.inc();
 
   // Clear data buffer after sending
   data.drain(data.length());
@@ -183,11 +189,13 @@ void NetworkExtProcFilter::onReceiveMessage(std::unique_ptr<ProcessingResponse>&
   if (processing_complete_) {
     ENVOY_CONN_LOG(debug, "Ignoring response message: processing already completed",
                    read_callbacks_->connection());
+    stats_.spurious_msgs_received_.inc();
     return;
   }
 
   auto response = std::move(res);
   ENVOY_CONN_LOG(debug, "Received response from external processor", read_callbacks_->connection());
+  stats_.stream_msgs_received_.inc();
 
   if (response->has_read_data()) {
     const auto& data = response->read_data();
@@ -197,15 +205,18 @@ void NetworkExtProcFilter::onReceiveMessage(std::unique_ptr<ProcessingResponse>&
     Buffer::OwnedImpl buffer(data.data());
     read_callbacks_->injectReadDataToFilterChain(buffer, data.end_of_stream());
     updateCloseCallbackStatus(false, true);
+    stats_.read_data_injected_.inc();
   } else if (response->has_write_data()) {
     const auto& data = response->write_data();
     ENVOY_CONN_LOG(trace, "Processing WRITE data response: {} bytes, end_stream={}",
                    read_callbacks_->connection(), data.data().size(), data.end_of_stream());
     Buffer::OwnedImpl buffer(data.data());
     write_callbacks_->injectWriteDataToFilterChain(buffer, data.end_of_stream());
     updateCloseCallbackStatus(false, false);
+    stats_.write_data_injected_.inc();
   } else {
     ENVOY_CONN_LOG(debug, "Response contained no data, continuing", read_callbacks_->connection());
+    stats_.empty_response_received_.inc();
   }
 }
 
@@ -216,6 +227,7 @@ void NetworkExtProcFilter::onGrpcError(Grpc::Status::GrpcStatus status,
   // Mark processing as complete to avoid further gRPC calls
   processing_complete_ = true;
   closeStream();
+  stats_.streams_grpc_error_.inc();
 
   // If failure mode is not to allow, close the connection
   if (!config_->failureModeAllow()) {
@@ -224,11 +236,14 @@ void NetworkExtProcFilter::onGrpcError(Grpc::Status::GrpcStatus status,
     closeConnection("ext_proc_grpc_error");
     return;
   }
+
+  stats_.failure_mode_allowed_.inc();
 }
 
 void NetworkExtProcFilter::onGrpcClose() {
   ENVOY_CONN_LOG(debug, "gRPC stream closed by peer", read_callbacks_->connection());
   processing_complete_ = true;
+  stats_.streams_grpc_close_.inc();
   closeStream();
 }
 
@@ -238,6 +253,9 @@ void NetworkExtProcFilter::closeStream() {
   }
 
   bool closed = stream_->close();
+  if (closed) {
+    stats_.streams_closed_.inc();
+  }
   ENVOY_CONN_LOG(debug, "Stream closed: {}", read_callbacks_->connection(), closed);
   stream_ = nullptr;
 }
@@ -249,6 +267,7 @@ void NetworkExtProcFilter::closeConnection(const std::string& reason) {
   read_callbacks_->disableClose(false);
   write_callbacks_->disableClose(false);
   read_callbacks_->connection().close(Network::ConnectionCloseType::FlushWrite, reason);
+  stats_.connections_closed_.inc();
 }
 
 } // namespace ExtProc

source/extensions/filters/network/ext_proc/ext_proc.h

@@ -19,14 +19,37 @@ namespace Extensions {
 namespace NetworkFilters {
 namespace ExtProc {
 
+#define ALL_NETWORK_EXT_PROC_FILTER_STATS(COUNTER)                                                 \
+  COUNTER(streams_started)                                                                         \
+  COUNTER(stream_msgs_sent)                                                                        \
+  COUNTER(stream_msgs_received)                                                                    \
+  COUNTER(read_data_sent)                                                                          \
+  COUNTER(write_data_sent)                                                                         \
+  COUNTER(read_data_injected)                                                                      \
+  COUNTER(write_data_injected)                                                                     \
+  COUNTER(empty_response_received)                                                                 \
+  COUNTER(spurious_msgs_received)                                                                  \
+  COUNTER(streams_closed)                                                                          \
+  COUNTER(streams_grpc_error)                                                                      \
+  COUNTER(streams_grpc_close)                                                                      \
+  COUNTER(connections_closed)                                                                      \
+  COUNTER(failure_mode_allowed)                                                                    \
+  COUNTER(stream_open_failures)
+
+struct NetworkExtProcStats {
+  ALL_NETWORK_EXT_PROC_FILTER_STATS(GENERATE_COUNTER_STRUCT)
+};
+
 /**
  * Global configuration for Network ExtProc filter.
  */
 class Config {
 public:
-  Config(const envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor& config)
+  Config(const envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor& config,
+         Stats::Scope& scope)
       : failure_mode_allow_(config.failure_mode_allow()),
-        processing_mode_(config.processing_mode()), grpc_service_(config.grpc_service()) {};
+        processing_mode_(config.processing_mode()), grpc_service_(config.grpc_service()),
+        stats_(generateStats(config.stat_prefix(), scope)) {};
 
   bool failureModeAllow() const { return failure_mode_allow_; }
 
@@ -36,10 +59,18 @@ class Config {
 
   const envoy::config::core::v3::GrpcService& grpcService() const { return grpc_service_; }
 
+  const NetworkExtProcStats& stats() const { return stats_; }
+
 private:
+  NetworkExtProcStats generateStats(const std::string& prefix, Stats::Scope& scope) {
+    const std::string final_prefix = absl::StrCat("network_ext_proc.", prefix);
+    return {ALL_NETWORK_EXT_PROC_FILTER_STATS(POOL_COUNTER_PREFIX(scope, final_prefix))};
+  }
+
   const bool failure_mode_allow_;
   const envoy::extensions::filters::network::ext_proc::v3::ProcessingMode processing_mode_;
   const envoy::config::core::v3::GrpcService grpc_service_;
+  NetworkExtProcStats stats_;
 };
 
 using ConfigConstSharedPtr = std::shared_ptr<const Config>;
@@ -114,6 +145,7 @@ class NetworkExtProcFilter : public Envoy::Network::Filter,
   Envoy::Network::WriteFilterCallbacks* write_callbacks_{nullptr};
 
   const ConfigConstSharedPtr config_;
+  const NetworkExtProcStats& stats_;
   ExternalProcessorClientPtr client_;
   ExternalProcessorStreamPtr stream_;
   const Envoy::Grpc::GrpcServiceConfigWithHashKey config_with_hash_key_;

test/extensions/filters/network/ext_proc/config_test.cc

@@ -13,8 +13,17 @@ namespace NetworkFilters {
 namespace ExtProc {
 namespace {
 
+class ConfigTest : public testing::Test {
+public:
+  ConfigTest() {}
+
+protected:
+  NiceMock<Stats::MockIsolatedStatsStore> store_;
+  Stats::Scope& scope_{*store_.rootScope()};
+};
+
 // Test the basic config setter and getter.
-TEST(ConfigTest, BasicConfigTest) {
+TEST_F(ConfigTest, BasicConfigTest) {
   envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor proto_config;
   proto_config.set_failure_mode_allow(true);
 
@@ -24,7 +33,7 @@ TEST(ConfigTest, BasicConfigTest) {
   processing_mode->set_process_write(
       envoy::extensions::filters::network::ext_proc::v3::ProcessingMode::STREAMED);
 
-  Config config(proto_config);
+  Config config(proto_config, scope_);
 
   EXPECT_TRUE(config.failureModeAllow());
 
@@ -35,10 +44,10 @@ TEST(ConfigTest, BasicConfigTest) {
             envoy::extensions::filters::network::ext_proc::v3::ProcessingMode::STREAMED);
 }
 
-TEST(ConfigTest, DefaultValues) {
+TEST_F(ConfigTest, DefaultValues) {
   envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor proto_config;
 
-  Config config(proto_config);
+  Config config(proto_config, scope_);
 
   // Test the default value for failureModeAllow (should be false by default in protobuf)
   EXPECT_FALSE(config.failureModeAllow());
@@ -52,7 +61,7 @@ TEST(ConfigTest, DefaultValues) {
 }
 
 // Test when both read and write are set to SKIP
-TEST(ConfigTest, BothSkipMode) {
+TEST_F(ConfigTest, BothSkipMode) {
   // Create a protobuf config with both read and write set to SKIP
   envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor proto_config;
 
@@ -62,7 +71,7 @@ TEST(ConfigTest, BothSkipMode) {
   processing_mode->set_process_write(
       envoy::extensions::filters::network::ext_proc::v3::ProcessingMode::SKIP);
 
-  Config config(proto_config);
+  Config config(proto_config, scope_);
 
   const auto& mode = config.processingMode();
   EXPECT_EQ(mode.process_read(),
@@ -77,6 +86,7 @@ TEST(NetworkExtProcConfigTest, SimpleConfig) {
   grpc_service:
     envoy_grpc:
       cluster_name: "ext_proc_server"
+  stat_prefix: "test_ext_proc"
   )EOF";
 
   envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor proto_config;
@@ -97,6 +107,7 @@ TEST(NetworkExtProcConfigTest, ConfigWithOptions) {
       cluster_name: "ext_proc_server"
   failure_mode_allow: true
   message_timeout: 2s
+  stat_prefix: "test_ext_proc"
   )EOF";
 
   envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor proto_config;
@@ -112,6 +123,7 @@ TEST(NetworkExtProcConfigTest, ConfigWithOptions) {
 // Test the config without a gRPC service.
 TEST(NetworkExtProcConfigFactoryTest, MissingGrpcService) {
   envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor proto_config;
+  proto_config.set_stat_prefix("test_ext_proc");
 
   NiceMock<Server::Configuration::MockFactoryContext> context;
   NetworkExtProcConfigFactory factory;
@@ -124,6 +136,7 @@ TEST(NetworkExtProcConfigFactoryTest, MissingGrpcService) {
 TEST(NetworkExtProcConfigFactoryTest, BothModesSkipped) {
   envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor proto_config;
   proto_config.mutable_grpc_service()->mutable_envoy_grpc()->set_cluster_name("ext_proc_server");
+  proto_config.set_stat_prefix("test_ext_proc");
 
   auto* processing_mode = proto_config.mutable_processing_mode();
   processing_mode->set_process_read(
@@ -143,6 +156,7 @@ TEST(NetworkExtProcConfigFactoryTest, BothModesSkipped) {
 TEST(NetworkExtProcConfigFactoryTest, DefaultProcessingMode) {
   envoy::extensions::filters::network::ext_proc::v3::NetworkExternalProcessor proto_config;
   proto_config.mutable_grpc_service()->mutable_envoy_grpc()->set_cluster_name("ext_proc_server");
+  proto_config.set_stat_prefix("test_ext_proc");
 
   NiceMock<Server::Configuration::MockFactoryContext> context;
   NetworkExtProcConfigFactory factory;