client channel: record call completion even if recv_trailing_metadata was not started (grpc#29198)

markdroth · web-flow · commit eded3b6bc326 · 2022-03-24T14:22:16.000-07:00
* client channel: record call completion even if recv_trailing_metadata was not started

* add test

* add test for opencensus filter

* remove unnecessary code
diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc
@@ -2427,6 +2427,7 @@ class ClientChannel::LoadBalancedCall::Metadata
   explicit Metadata(grpc_metadata_batch* batch) : batch_(batch) {}
 
   void Add(absl::string_view key, absl::string_view value) override {
+    if (batch_ == nullptr) return;
     // Gross, egregious hack to support legacy grpclb behavior.
     // TODO(ctiller): Use a promise context for this once that plumbing is done.
     if (key == GrpcLbClientStatsMetadata::key()) {
@@ -2447,13 +2448,15 @@ class ClientChannel::LoadBalancedCall::Metadata
 
   std::vector<std::pair<std::string, std::string>> TestOnlyCopyToVector()
       override {
+    if (batch_ == nullptr) return {};
     Encoder encoder;
     batch_->Encode(&encoder);
     return encoder.Take();
   }
 
   absl::optional<absl::string_view> Lookup(absl::string_view key,
                                            std::string* buffer) const override {
+    if (batch_ == nullptr) return absl::nullopt;
     return batch_->GetStringValue(key, buffer);
   }
 
@@ -2524,7 +2527,8 @@ class ClientChannel::LoadBalancedCall::BackendMetricAccessor
       : lb_call_(lb_call) {}
 
   const BackendMetricData* GetBackendMetricData() override {
-    if (lb_call_->backend_metric_data_ == nullptr) {
+    if (lb_call_->backend_metric_data_ == nullptr &&
+        lb_call_->recv_trailing_metadata_ != nullptr) {
       if (const auto* md = lb_call_->recv_trailing_metadata_->get_pointer(
               XEndpointLoadMetricsBinMetadata())) {
         lb_call_->backend_metric_data_ =
@@ -2594,6 +2598,12 @@ ClientChannel::LoadBalancedCall::~LoadBalancedCall() {
 }
 
 void ClientChannel::LoadBalancedCall::Orphan() {
+  // If the recv_trailing_metadata op was never started, then notify
+  // about call completion here, as best we can.  We assume status
+  // CANCELLED in this case.
+  if (recv_trailing_metadata_ == nullptr) {
+    RecordCallCompletion(absl::CancelledError("call cancelled"));
+  }
   // Compute latency and report it to the tracer.
   if (call_attempt_tracer_ != nullptr) {
     gpr_timespec latency =
@@ -2905,22 +2915,7 @@ void ClientChannel::LoadBalancedCall::RecvTrailingMetadataReady(
         status = absl::Status(static_cast<absl::StatusCode>(code), message);
       }
     }
-    // If we have a tracer, notify it.
-    if (self->call_attempt_tracer_ != nullptr) {
-      self->call_attempt_tracer_->RecordReceivedTrailingMetadata(
-          status, self->recv_trailing_metadata_,
-          *self->transport_stream_stats_);
-    }
-    // If the LB policy requested a callback for trailing metadata, invoke
-    // the callback.
-    if (self->lb_subchannel_call_tracker_ != nullptr) {
-      Metadata trailing_metadata(self->recv_trailing_metadata_);
-      BackendMetricAccessor backend_metric_accessor(self);
-      LoadBalancingPolicy::SubchannelCallTrackerInterface::FinishArgs args = {
-          status, &trailing_metadata, &backend_metric_accessor};
-      self->lb_subchannel_call_tracker_->Finish(args);
-      self->lb_subchannel_call_tracker_.reset();
-    }
+    self->RecordCallCompletion(status);
   }
   // Chain to original callback.
   if (self->failure_error_ != GRPC_ERROR_NONE) {
@@ -2933,6 +2928,25 @@ void ClientChannel::LoadBalancedCall::RecvTrailingMetadataReady(
                error);
 }
 
+void ClientChannel::LoadBalancedCall::RecordCallCompletion(
+    absl::Status status) {
+  // If we have a tracer, notify it.
+  if (call_attempt_tracer_ != nullptr) {
+    call_attempt_tracer_->RecordReceivedTrailingMetadata(
+        status, recv_trailing_metadata_, transport_stream_stats_);
+  }
+  // If the LB policy requested a callback for trailing metadata, invoke
+  // the callback.
+  if (lb_subchannel_call_tracker_ != nullptr) {
+    Metadata trailing_metadata(recv_trailing_metadata_);
+    BackendMetricAccessor backend_metric_accessor(this);
+    LoadBalancingPolicy::SubchannelCallTrackerInterface::FinishArgs args = {
+        status, &trailing_metadata, &backend_metric_accessor};
+    lb_subchannel_call_tracker_->Finish(args);
+    lb_subchannel_call_tracker_.reset();
+  }
+}
+
 void ClientChannel::LoadBalancedCall::CreateSubchannelCall() {
   SubchannelCall::Args call_args = {
       std::move(connected_subchannel_), pollent_, path_.Ref(), /*start_time=*/0,
diff --git a/src/core/ext/filters/client_channel/client_channel.h b/src/core/ext/filters/client_channel/client_channel.h
@@ -433,6 +433,8 @@ class ClientChannel::LoadBalancedCall
   static void RecvMessageReady(void* arg, grpc_error_handle error);
   static void RecvTrailingMetadataReady(void* arg, grpc_error_handle error);
 
+  void RecordCallCompletion(absl::Status status);
+
   void CreateSubchannelCall();
   // Invoked when a pick is completed, on both success or failure.
   static void PickDone(void* arg, grpc_error_handle error);
diff --git a/src/core/lib/channel/call_tracer.h b/src/core/lib/channel/call_tracer.h
@@ -59,9 +59,12 @@ class CallTracer {
     virtual void RecordReceivedInitialMetadata(
         grpc_metadata_batch* recv_initial_metadata, uint32_t flags) = 0;
     virtual void RecordReceivedMessage(const ByteStream& recv_message) = 0;
+    // If the call was cancelled before the recv_trailing_metadata op
+    // was started, recv_trailing_metadata and transport_stream_stats
+    // will be null.
     virtual void RecordReceivedTrailingMetadata(
         absl::Status status, grpc_metadata_batch* recv_trailing_metadata,
-        const grpc_transport_stream_stats& transport_stream_stats) = 0;
+        const grpc_transport_stream_stats* transport_stream_stats) = 0;
     virtual void RecordCancel(grpc_error_handle cancel_error) = 0;
     // Should be the last API call to the object. Once invoked, the tracer
     // library is free to destroy the object.
diff --git a/src/cpp/ext/filters/census/client_filter.cc b/src/cpp/ext/filters/census/client_filter.cc
@@ -95,7 +95,7 @@ OpenCensusCallTracer::OpenCensusCallAttemptTracer::OpenCensusCallAttemptTracer(
 
 void OpenCensusCallTracer::OpenCensusCallAttemptTracer::
     RecordSendInitialMetadata(grpc_metadata_batch* send_initial_metadata,
-                              uint32_t /* flags */) {
+                              uint32_t /*flags*/) {
   char tracing_buf[kMaxTraceContextLen];
   size_t tracing_len = TraceContextSerialize(context_.Context(), tracing_buf,
                                              kMaxTraceContextLen);
@@ -114,12 +114,12 @@ void OpenCensusCallTracer::OpenCensusCallAttemptTracer::
 }
 
 void OpenCensusCallTracer::OpenCensusCallAttemptTracer::RecordSendMessage(
-    const grpc_core::ByteStream& /* send_message */) {
+    const grpc_core::ByteStream& /*send_message*/) {
   ++sent_message_count_;
 }
 
 void OpenCensusCallTracer::OpenCensusCallAttemptTracer::RecordReceivedMessage(
-    const grpc_core::ByteStream& /* recv_message */) {
+    const grpc_core::ByteStream& /*recv_message*/) {
   ++recv_message_count_;
 }
 
@@ -140,21 +140,25 @@ void FilterTrailingMetadata(grpc_metadata_batch* b, uint64_t* elapsed_time) {
 void OpenCensusCallTracer::OpenCensusCallAttemptTracer::
     RecordReceivedTrailingMetadata(
         absl::Status status, grpc_metadata_batch* recv_trailing_metadata,
-        const grpc_transport_stream_stats& transport_stream_stats) {
-  FilterTrailingMetadata(recv_trailing_metadata, &elapsed_time_);
-  const uint64_t request_size = transport_stream_stats.outgoing.data_bytes;
-  const uint64_t response_size = transport_stream_stats.incoming.data_bytes;
+        const grpc_transport_stream_stats* transport_stream_stats) {
+  status_code_ = status.code();
+  if (recv_trailing_metadata == nullptr || transport_stream_stats == nullptr) {
+    return;
+  }
+  uint64_t elapsed_time = 0;
+  FilterTrailingMetadata(recv_trailing_metadata, &elapsed_time);
   std::vector<std::pair<opencensus::tags::TagKey, std::string>> tags =
       context_.tags().tags();
   tags.emplace_back(ClientMethodTagKey(), std::string(parent_->method_));
-  status_code_ = status.code();
   std::string final_status = absl::StatusCodeToString(status_code_);
   tags.emplace_back(ClientStatusTagKey(), final_status);
   ::opencensus::stats::Record(
-      {{RpcClientSentBytesPerRpc(), static_cast<double>(request_size)},
-       {RpcClientReceivedBytesPerRpc(), static_cast<double>(response_size)},
+      {{RpcClientSentBytesPerRpc(),
+        static_cast<double>(transport_stream_stats->outgoing.data_bytes)},
+       {RpcClientReceivedBytesPerRpc(),
+        static_cast<double>(transport_stream_stats->incoming.data_bytes)},
        {RpcClientServerLatency(),
-        ToDoubleMilliseconds(absl::Nanoseconds(elapsed_time_))}},
+        ToDoubleMilliseconds(absl::Nanoseconds(elapsed_time))}},
       tags);
 }
 
@@ -165,7 +169,7 @@ void OpenCensusCallTracer::OpenCensusCallAttemptTracer::RecordCancel(
 }
 
 void OpenCensusCallTracer::OpenCensusCallAttemptTracer::RecordEnd(
-    const gpr_timespec& /* latency */) {
+    const gpr_timespec& /*latency*/) {
   double latency_ms = absl::ToDoubleMilliseconds(absl::Now() - start_time_);
   std::vector<std::pair<opencensus::tags::TagKey, std::string>> tags =
       context_.tags().tags();
diff --git a/src/cpp/ext/filters/census/open_census_call_tracer.h b/src/cpp/ext/filters/census/open_census_call_tracer.h
@@ -33,25 +33,23 @@ class OpenCensusCallTracer : public grpc_core::CallTracer {
     OpenCensusCallAttemptTracer(OpenCensusCallTracer* parent,
                                 uint64_t attempt_num, bool is_transparent_retry,
                                 bool arena_allocated);
-    void RecordSendInitialMetadata(
-        grpc_metadata_batch* /* send_initial_metadata */,
-        uint32_t /* flags */) override;
-    void RecordOnDoneSendInitialMetadata(gpr_atm* /* peer_string */) override {}
+    void RecordSendInitialMetadata(grpc_metadata_batch* send_initial_metadata,
+                                   uint32_t /*flags*/) override;
+    void RecordOnDoneSendInitialMetadata(gpr_atm* /*peer_string*/) override {}
     void RecordSendTrailingMetadata(
-        grpc_metadata_batch* /* send_trailing_metadata */) override {}
+        grpc_metadata_batch* /*send_trailing_metadata*/) override {}
     void RecordSendMessage(
-        const grpc_core::ByteStream& /* send_message */) override;
+        const grpc_core::ByteStream& /*send_message*/) override;
     void RecordReceivedInitialMetadata(
-        grpc_metadata_batch* /* recv_initial_metadata */,
-        uint32_t /* flags */) override {}
+        grpc_metadata_batch* /*recv_initial_metadata*/,
+        uint32_t /*flags*/) override {}
     void RecordReceivedMessage(
-        const grpc_core::ByteStream& /* recv_message */) override;
+        const grpc_core::ByteStream& /*recv_message*/) override;
     void RecordReceivedTrailingMetadata(
-        absl::Status /* status */, grpc_metadata_batch* recv_trailing_metadata,
-        const grpc_transport_stream_stats& /* transport_stream_stats */)
-        override;
+        absl::Status status, grpc_metadata_batch* recv_trailing_metadata,
+        const grpc_transport_stream_stats* transport_stream_stats) override;
     void RecordCancel(grpc_error_handle cancel_error) override;
-    void RecordEnd(const gpr_timespec& /* latency */) override;
+    void RecordEnd(const gpr_timespec& /*latency*/) override;
 
     CensusContext* context() { return &context_; }
 
@@ -65,8 +63,6 @@ class OpenCensusCallTracer : public grpc_core::CallTracer {
     CensusContext context_;
     // Start time (for measuring latency).
     absl::Time start_time_;
-    // Server elapsed time in nanoseconds.
-    uint64_t elapsed_time_ = 0;
     // Number of messages in this RPC.
     uint64_t recv_message_count_ = 0;
     uint64_t sent_message_count_ = 0;
diff --git a/test/core/util/test_lb_policies.cc b/test/core/util/test_lb_policies.cc
@@ -281,6 +281,7 @@ class InterceptRecvTrailingMetadataLoadBalancingPolicy
 
     void Finish(FinishArgs args) override {
       TrailingMetadataArgsSeen args_seen;
+      args_seen.status = args.status;
       args_seen.backend_metric_data =
           args.backend_metric_accessor->GetBackendMetricData();
       args_seen.metadata = args.trailing_metadata->TestOnlyCopyToVector();
diff --git a/test/core/util/test_lb_policies.h b/test/core/util/test_lb_policies.h
@@ -36,6 +36,7 @@ void RegisterTestPickArgsLoadBalancingPolicy(
     TestPickArgsCallback cb, const char* delegate_policy_name = "pick_first");
 
 struct TrailingMetadataArgsSeen {
+  absl::Status status;
   const LoadBalancingPolicy::BackendMetricAccessor::BackendMetricData*
       backend_metric_data;
   MetadataVector metadata;
diff --git a/test/cpp/end2end/client_lb_end2end_test.cc b/test/cpp/end2end/client_lb_end2end_test.cc
diff --git a/test/cpp/ext/filters/census/BUILD b/test/cpp/ext/filters/census/BUILD
diff --git a/test/cpp/ext/filters/census/stats_plugin_end2end_test.cc b/test/cpp/ext/filters/census/stats_plugin_end2end_test.cc
