[stats] Add experiment to use transport reported request latency whenever available (#32168)

* [stats] Add experiment to use transport reported request latency whenever available

* redux: add a new metric

* review-feedback

* Automated change: Fix sanity tests

* fix

* fix

Co-authored-by: ctiller <ctiller@users.noreply.github.com>

BUILD

@@ -2185,6 +2185,7 @@ grpc_cc_library(
         "//src/core:channel_args",
         "//src/core:channel_stack_type",
         "//src/core:closure",
+        "//src/core:experiments",
         "//src/core:slice",
         "//src/core:slice_buffer",
         "//src/core:slice_refcount",

bazel/experiments.bzl

@@ -20,6 +20,9 @@ EXPERIMENTS = {
     "dbg": {
     },
     "off": {
+        "census_test": [
+            "transport_supplies_client_latency",
+        ],
         "core_end2end_test": [
             "promise_based_client_call",
         ],

include/grpcpp/opencensus.h

@@ -65,6 +65,7 @@ extern const absl::string_view kRpcClientStartedRpcsMeasureName;
 extern const absl::string_view kRpcClientRetriesPerCallMeasureName;
 extern const absl::string_view kRpcClientTransparentRetriesPerCallMeasureName;
 extern const absl::string_view kRpcClientRetryDelayPerCallMeasureName;
+extern const absl::string_view kRpcClientTransportLatencyMeasureName;
 
 extern const absl::string_view kRpcServerSentMessagesPerRpcMeasureName;
 extern const absl::string_view kRpcServerSentBytesPerRpcMeasureName;
@@ -81,6 +82,7 @@ const ::opencensus::stats::ViewDescriptor&
 ClientSentCompressedMessageBytesPerRpc();
 const ::opencensus::stats::ViewDescriptor&
 ClientReceivedCompressedMessageBytesPerRpc();
+const ::opencensus::stats::ViewDescriptor& ClientTransportLatency();
 
 const ::opencensus::stats::ViewDescriptor& ServerStartedRpcs();
 const ::opencensus::stats::ViewDescriptor& ServerCompletedRpcs();

src/core/lib/experiments/experiments.cc

@@ -48,6 +48,9 @@ const char* const description_promise_based_client_call =
     "(ie when all filters in a stack are promise based)";
 const char* const description_free_large_allocator =
     "If set, return all free bytes from a \042big\042 allocator";
+const char* const description_transport_supplies_client_latency =
+    "If set, use the transport represented value for client latency in "
+    "opencensus";
 }  // namespace
 
 namespace grpc_core {
@@ -66,6 +69,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
     {"monitoring_experiment", description_monitoring_experiment, true},
     {"promise_based_client_call", description_promise_based_client_call, false},
     {"free_large_allocator", description_free_large_allocator, false},
+    {"transport_supplies_client_latency",
+     description_transport_supplies_client_latency, false},
 };
 
 }  // namespace grpc_core

src/core/lib/experiments/experiments.h

@@ -40,6 +40,9 @@ inline bool IsEventEngineClientEnabled() { return IsExperimentEnabled(7); }
 inline bool IsMonitoringExperimentEnabled() { return IsExperimentEnabled(8); }
 inline bool IsPromiseBasedClientCallEnabled() { return IsExperimentEnabled(9); }
 inline bool IsFreeLargeAllocatorEnabled() { return IsExperimentEnabled(10); }
+inline bool IsTransportSuppliesClientLatencyEnabled() {
+  return IsExperimentEnabled(11);
+}
 
 struct ExperimentMetadata {
   const char* name;
@@ -47,7 +50,7 @@ struct ExperimentMetadata {
   bool default_value;
 };
 
-constexpr const size_t kNumExperiments = 11;
+constexpr const size_t kNumExperiments = 12;
 extern const ExperimentMetadata g_experiment_metadata[kNumExperiments];
 
 }  // namespace grpc_core

src/core/lib/experiments/experiments.yaml

@@ -119,4 +119,10 @@
   expiry: 2023/04/01
   owner: alishananda@google.com
   test_tags: [resource_quota_test]
-
+- name: transport_supplies_client_latency
+  description:
+    If set, use the transport represented value for client latency in opencensus
+  default: false
+  expiry: 2023/06/01
+  owner: ctiller@google.com
+  test_tags: [census_test]

src/cpp/ext/filters/census/client_filter.cc

@@ -48,6 +48,7 @@
 
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/context.h"
+#include "src/core/lib/experiments/experiments.h"
 #include "src/core/lib/gprpp/sync.h"
 #include "src/core/lib/resource_quota/arena.h"
 #include "src/core/lib/slice/slice.h"
@@ -213,6 +214,15 @@ void OpenCensusCallTracer::OpenCensusCallAttemptTracer::
          {RpcClientServerLatency(),
           ToDoubleMilliseconds(absl::Nanoseconds(elapsed_time))}},
         tags);
+    if (grpc_core::IsTransportSuppliesClientLatencyEnabled()) {
+      if (gpr_time_cmp(transport_stream_stats->latency,
+                       gpr_inf_future(GPR_TIMESPAN)) != 0) {
+        double latency_ms = absl::ToDoubleMilliseconds(absl::Microseconds(
+            gpr_timespec_to_micros(transport_stream_stats->latency)));
+        ::opencensus::stats::Record({{RpcClientTransportLatency(), latency_ms}},
+                                    tags);
+      }
+    }
   }
 }
 

src/cpp/ext/filters/census/grpc_plugin.cc

@@ -64,6 +64,7 @@ void RegisterOpenCensusPlugin() {
   RpcClientRetriesPerCall();
   RpcClientTransparentRetriesPerCall();
   RpcClientRetryDelayPerCall();
+  RpcClientTransportLatency();
 
   RpcServerSentBytesPerRpc();
   RpcServerReceivedBytesPerRpc();
@@ -146,6 +147,9 @@ ABSL_CONST_INIT const absl::string_view
 ABSL_CONST_INIT const absl::string_view kRpcClientRetryDelayPerCallMeasureName =
     "grpc.io/client/retry_delay_per_call";
 
+ABSL_CONST_INIT const absl::string_view kRpcClientTransportLatencyMeasureName =
+    "grpc.io/client/transport_latency";
+
 // Server
 ABSL_CONST_INIT const absl::string_view
     kRpcServerSentMessagesPerRpcMeasureName =

src/cpp/ext/filters/census/measures.cc

@@ -98,6 +98,15 @@ MeasureInt64 RpcClientStartedRpcs() {
   return measure;
 }
 
+MeasureDouble RpcClientTransportLatency() {
+  static const auto measure = MeasureDouble::Register(
+      experimental::kRpcClientTransportLatencyMeasureName,
+      "Time between first byte of request sent to last byte of response "
+      "received on the transport",
+      kUnitMilliseconds);
+  return measure;
+}
+
 // Client per-overall-client-call measures
 MeasureInt64 RpcClientRetriesPerCall() {
   static const auto measure =

src/cpp/ext/filters/census/measures.h

@@ -36,6 +36,7 @@ namespace grpc {
 ::opencensus::stats::MeasureInt64 RpcClientRetriesPerCall();
 ::opencensus::stats::MeasureInt64 RpcClientTransparentRetriesPerCall();
 ::opencensus::stats::MeasureDouble RpcClientRetryDelayPerCall();
+::opencensus::stats::MeasureDouble RpcClientTransportLatency();
 
 ::opencensus::stats::MeasureInt64 RpcServerSentMessagesPerRpc();
 ::opencensus::stats::MeasureDouble RpcServerSentBytesPerRpc();

src/cpp/ext/filters/census/views.cc

@@ -135,6 +135,16 @@ const ViewDescriptor& ClientRoundtripLatency() {
   return descriptor;
 }
 
+const ViewDescriptor& ClientTransportLatency() {
+  const static ViewDescriptor descriptor =
+      DefaultViewDescriptor()
+          .set_name("grpc.io/client/transport_latency")
+          .set_measure(kRpcClientTransportLatencyMeasureName)
+          .set_aggregation(MillisDistributionAggregation())
+          .add_column(ClientMethodTagKey());
+  return descriptor;
+}
+
 const ViewDescriptor& ClientSentCompressedMessageBytesPerRpc() {
   const static ViewDescriptor descriptor =
       DefaultViewDescriptor()

test/cpp/ext/filters/census/BUILD

@@ -56,7 +56,10 @@ grpc_cc_test(
     flaky = True,
     language = "C++",
     linkstatic = True,
-    tags = ["no_windows"],  # TODO(jtattermusch): fix test on windows
+    tags = [
+        "census_test",
+        "no_windows",
+    ],  # TODO(jtattermusch): fix test on windows
     deps = [
         "library",
         "//:grpc++",
@@ -82,7 +85,10 @@ grpc_cc_test(
     flaky = True,
     language = "C++",
     linkstatic = True,
-    tags = ["no_windows"],  # TODO(jtattermusch): fix test on windows
+    tags = [
+        "census_test",
+        "no_windows",
+    ],  # TODO(jtattermusch): fix test on windows
     deps = [
         "library",
         "//:grpc++",

test/cpp/ext/filters/census/stats_plugin_end2end_test.cc

@@ -216,6 +216,7 @@ TEST_F(StatsPluginEnd2EndTest, Latency) {
   View client_latency_view(ClientRoundtripLatencyCumulative());
   View client_server_latency_view(ClientServerLatencyCumulative());
   View server_server_latency_view(ServerServerLatencyCumulative());
+  View client_transport_latency_view(experimental::ClientTransportLatency());
 
   const absl::Time start_time = absl::Now();
   {
@@ -259,6 +260,25 @@ TEST_F(StatsPluginEnd2EndTest, Latency) {
               ::testing::Property(&Distribution::mean,
                                   ::testing::Lt(client_latency))))));
 
+  // Transport time is a subinterval of total latency.
+  if (grpc_core::IsTransportSuppliesClientLatencyEnabled()) {
+    const auto client_transport_latency =
+        client_transport_latency_view.GetData()
+            .distribution_data()
+            .find({client_method_name_})
+            ->second.mean();
+    EXPECT_THAT(
+        client_server_latency_view.GetData().distribution_data(),
+        ::testing::UnorderedElementsAre(::testing::Pair(
+            ::testing::ElementsAre(client_method_name_),
+            ::testing::AllOf(
+                ::testing::Property(&Distribution::count, 1),
+                ::testing::Property(&Distribution::mean, ::testing::Gt(0.0)),
+                ::testing::Property(
+                    &Distribution::mean,
+                    ::testing::Lt(client_transport_latency))))));
+  }
+
   // client server elapsed time should be the same value propagated to the
   // client.
   const auto client_elapsed_time = client_server_latency_view.GetData()