filter: add grpc upstream stats

api/envoy/extensions/filters/http/grpc_stats/v3/config.proto

@@ -54,6 +54,14 @@ message FilterConfig {
     //   `envoy.deprecated_features.grpc_stats_filter_enable_stats_for_all_methods_by_default`.
     google.protobuf.BoolValue stats_for_all_methods = 3;
   }
+
+  // If true, the filter will gather a histogram for the request time of the upstream.
+  // It works with :ref:`stats_for_all_methods
+  // <envoy_api_field_extensions.filters.http.grpc_stats.v3.FilterConfig.stats_for_all_methods>`
+  // and :ref:`individual_method_stats_allowlist
+  // <envoy_api_field_extensions.filters.http.grpc_stats.v3.FilterConfig.individual_method_stats_allowlist>` the same way
+  // request_message_count and response_message_count works.
+  bool enable_upstream_stats = 4;
 }
 
 // gRPC statistics filter state object in protobuf form.

docs/root/configuration/http/http_filters/grpc_stats_filter.rst

@@ -21,6 +21,8 @@ are shown in this form. See the documentation for
 :ref:`individual_method_stats_allowlist <envoy_api_field_config.filter.http.grpc_stats.v2alpha.FilterConfig.individual_method_stats_allowlist>`
 and :ref:`stats_for_all_methods <envoy_api_field_config.filter.http.grpc_stats.v2alpha.FilterConfig.stats_for_all_methods>`.
 
+To enable *upstream_rq_time* (v3 API only) see :ref:`enable_upstream_stats <envoy_v3_api_field_extensions.filters.http.grpc_stats.v3.FilterConfig.enable_upstream_stats>`.
+
 
 .. csv-table::
   :header: Name, Type, Description
@@ -31,3 +33,4 @@ and :ref:`stats_for_all_methods <envoy_api_field_config.filter.http.grpc_stats.v
   <grpc service>.<grpc method>.total, Counter, Total service/method calls
   <grpc service>.<grpc method>.request_message_count, Counter, Total request message count for service/method calls
   <grpc service>.<grpc method>.response_message_count, Counter, Total response message count for service/method calls
+  <grpc service>.<grpc method>.upstream_rq_time, Histogram, Request time milliseconds

docs/root/version_history/current.rst

@@ -7,6 +7,8 @@ Changes
 * access loggers: added GRPC_STATUS operator on logging format.
 * fault: added support for controlling the percentage of requests that abort, delay and response rate limits faults 
   are applied to using :ref:`HTTP headers <config_http_filters_fault_injection_http_header>` to the HTTP fault filter.
+* filter: add `upstram_rq_time` stats to the GPRC stats filter.
+  Disabled by default and can be enabled via :ref:`enable_upstream_stats <envoy_v3_api_field_extensions.filters.http.grpc_stats.v3.FilterConfig.enable_upstream_stats>`.
 * http: fixed a bug where the upgrade header was not cleared on responses to non-upgrade requests.
   Can be reverted temporarily by setting runtime feature `envoy.reloadable_features.fix_upgrade_response` to false.
 * tracing: tracing configuration has been made fully dynamic and every HTTP connection manager

include/envoy/grpc/context.h

@@ -84,6 +84,16 @@ class Context {
                                          const absl::optional<RequestStatNames>& request_names,
                                          uint64_t amount) PURE;
 
+  /**
+   * Charge upstream stat to a cluster/service/method.
+   * @param cluster supplies the target cluster.
+   * @param request_names supplies the request names.
+   * @param duration supplies the duration of the upstream request.
+   */
+  virtual void chargeUpstreamStat(const Upstream::ClusterInfo& cluster,
+                                  const absl::optional<RequestStatNames>& request_names,
+                                  std::chrono::milliseconds duration) PURE;
+
   /**
    * @return a struct containing StatNames for gRPC stat tokens.
    */

source/common/grpc/context_impl.cc

@@ -15,7 +15,7 @@ ContextImpl::ContextImpl(Stats::SymbolTable& symbol_table)
       total_(stat_name_pool_.add("total")), zero_(stat_name_pool_.add("0")),
       request_message_count_(stat_name_pool_.add("request_message_count")),
       response_message_count_(stat_name_pool_.add("response_message_count")),
-      stat_names_(symbol_table) {}
+      upstream_rq_time_(stat_name_pool_.add("upstream_rq_time")), stat_names_(symbol_table) {}
 
 // Makes a stat name from a string, if we don't already have one for it.
 // This always takes a lock on mutex_, and if we haven't seen the name
@@ -114,6 +114,21 @@ void ContextImpl::chargeResponseMessageStat(const Upstream::ClusterInfo& cluster
       .add(amount);
 }
 
+void ContextImpl::chargeUpstreamStat(const Upstream::ClusterInfo& cluster,
+                                     const absl::optional<RequestStatNames>& request_names,
+                                     std::chrono::milliseconds duration) {
+  auto prefix_and_storage = getPrefix(Protocol::Grpc, request_names);
+  Stats::StatName prefix = prefix_and_storage.first;
+
+  const Stats::SymbolTable::StoragePtr upstream_rq_time =
+      symbol_table_.join({prefix, upstream_rq_time_});
+
+  cluster.statsScope()
+      .histogramFromStatName(Stats::StatName(upstream_rq_time.get()),
+                             Stats::Histogram::Unit::Milliseconds)
+      .recordValue(duration.count());
+}
+
 absl::optional<ContextImpl::RequestStatNames>
 ContextImpl::resolveDynamicServiceAndMethod(const Http::HeaderEntry* path) {
   absl::optional<Common::RequestNames> request_names = Common::resolveServiceAndMethod(path);

source/common/grpc/context_impl.h

@@ -38,6 +38,9 @@ class ContextImpl : public Context {
   void chargeResponseMessageStat(const Upstream::ClusterInfo& cluster,
                                  const absl::optional<RequestStatNames>& request_names,
                                  uint64_t amount) override;
+  void chargeUpstreamStat(const Upstream::ClusterInfo& cluster,
+                          const absl::optional<RequestStatNames>& request_names,
+                          std::chrono::milliseconds duration) override;
 
   /**
    * Resolve the gRPC service and method from the HTTP2 :path header.
@@ -83,6 +86,7 @@ class ContextImpl : public Context {
   const Stats::StatName zero_;
   const Stats::StatName request_message_count_;
   const Stats::StatName response_message_count_;
+  const Stats::StatName upstream_rq_time_;
 
   StatNames stat_names_;
 };

source/extensions/filters/http/grpc_stats/grpc_stats_filter.cc

@@ -92,7 +92,8 @@ class GrpcServiceMethodToRequestNamesMap {
 struct Config {
   Config(const envoy::extensions::filters::http::grpc_stats::v3::FilterConfig& proto_config,
          Server::Configuration::FactoryContext& context)
-      : context_(context.grpcContext()), emit_filter_state_(proto_config.emit_filter_state()) {
+      : context_(context.grpcContext()), emit_filter_state_(proto_config.emit_filter_state()),
+        enable_upstream_stats_(proto_config.enable_upstream_stats()) {
 
     switch (proto_config.per_method_stat_specifier_case()) {
     case envoy::extensions::filters::http::grpc_stats::v3::FilterConfig::
@@ -134,7 +135,8 @@ struct Config {
     }
   }
   Grpc::Context& context_;
-  bool emit_filter_state_;
+  const bool emit_filter_state_;
+  const bool enable_upstream_stats_;
   bool stats_for_all_methods_{false};
   absl::optional<GrpcServiceMethodToRequestNamesMap> allowlist_;
 };
@@ -226,6 +228,16 @@ class GrpcStatsFilter : public Http::PassThroughFilter {
     if (doStatTracking()) {
       config_->context_.chargeStat(*cluster_, Grpc::Context::Protocol::Grpc, request_names_,
                                    trailers.GrpcStatus());
+
+      if (config_->enable_upstream_stats_ &&
+          decoder_callbacks_->streamInfo().lastUpstreamTxByteSent().has_value() &&
+          decoder_callbacks_->streamInfo().lastUpstreamRxByteReceived().has_value()) {
+        std::chrono::milliseconds chrono_duration =
+            std::chrono::duration_cast<std::chrono::milliseconds>(
+                decoder_callbacks_->streamInfo().lastUpstreamRxByteReceived().value() -
+                decoder_callbacks_->streamInfo().lastUpstreamTxByteSent().value());
+        config_->context_.chargeUpstreamStat(*cluster_, request_names_, chrono_duration);
+      }
     }
     return Http::FilterTrailersStatus::Continue;
   }

test/extensions/filters/http/grpc_stats/config_test.cc

@@ -14,6 +14,7 @@
 #include "gtest/gtest.h"
 
 using testing::_;
+using testing::Property;
 using testing::Return;
 
 namespace Envoy {
@@ -34,6 +35,10 @@ class GrpcStatsFilterConfigTest : public testing::Test {
     cb(filter_callback);
 
     ON_CALL(decoder_callbacks_, streamInfo()).WillByDefault(testing::ReturnRef(stream_info_));
+
+    ON_CALL(*decoder_callbacks_.cluster_info_, statsScope())
+        .WillByDefault(testing::ReturnRef(stats_store_));
+
     filter_->setDecoderFilterCallbacks(decoder_callbacks_);
   }
 
@@ -61,7 +66,8 @@ class GrpcStatsFilterConfigTest : public testing::Test {
   NiceMock<Server::Configuration::MockFactoryContext> context_;
   std::shared_ptr<Http::StreamFilter> filter_;
   NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks_;
-  TestStreamInfo stream_info_;
+  NiceMock<StreamInfo::MockStreamInfo> stream_info_;
+  NiceMock<Stats::MockIsolatedStatsStore> stats_store_;
 };
 
 TEST_F(GrpcStatsFilterConfigTest, StatsHttp2HeaderOnlyResponse) {
@@ -404,6 +410,32 @@ TEST_F(GrpcStatsFilterConfigTest, MessageCounts) {
   EXPECT_EQ(3U, filter_object.response_message_count());
 }
 
+TEST_F(GrpcStatsFilterConfigTest, UpstreamStats) {
+  config_.mutable_stats_for_all_methods()->set_value(true);
+  config_.set_emit_filter_state(true);
+  config_.set_enable_upstream_stats(true);
+  initialize();
+
+  Http::TestRequestHeaderMapImpl request_headers{
+      {"content-type", "application/grpc+proto"},
+      {":path", "/lyft.users.BadCompanions/GetBadCompanions"}};
+
+  ON_CALL(stream_info_, lastUpstreamRxByteReceived())
+      .WillByDefault(testing::Return(
+          absl::optional<std::chrono::nanoseconds>(std::chrono::nanoseconds(30000000))));
+  ON_CALL(stream_info_, lastUpstreamTxByteSent())
+      .WillByDefault(testing::Return(
+          absl::optional<std::chrono::nanoseconds>(std::chrono::nanoseconds(20000000))));
+
+  EXPECT_CALL(stats_store_,
+              deliverHistogramToSinks(
+                  Property(&Stats::Metric::name,
+                           "grpc.lyft.users.BadCompanions.GetBadCompanions.upstream_rq_time"),
+                  10ul));
+
+  doRequestResponse(request_headers);
+}
+
 } // namespace
 } // namespace GrpcStats
 } // namespace HttpFilters