[chaotic-good] Add fathom metrics to ztrace (grpc#39328)

Ensure any collected fathom metrics show up in the ztrace so we can use them for debugging.

Closes grpc#39328

COPYBARA_INTEGRATE_REVIEW=grpc#39328 from ctiller:trace-fathomable ce593b2
PiperOrigin-RevId: 750304444

Author: ctiller

src/core/ext/transport/chaotic_good/data_endpoints.cc

@@ -283,7 +283,8 @@ void InputQueue::Cancel(uint64_t payload_tag) {
 
 auto Endpoint::WriteLoop(uint32_t id,
                          RefCountedPtr<OutputBuffers> output_buffers,
-                         std::shared_ptr<PromiseEndpoint> endpoint) {
+                         std::shared_ptr<PromiseEndpoint> endpoint,
+                         std::shared_ptr<TcpZTraceCollector> ztrace_collector) {
   output_buffers->AddEndpoint(id);
   std::vector<size_t> requested_metrics;
   std::optional<size_t> data_rate_metric =
@@ -294,13 +295,13 @@ auto Endpoint::WriteLoop(uint32_t id,
   return Loop([id, endpoint = std::move(endpoint),
                output_buffers = std::move(output_buffers),
                requested_metrics = std::move(requested_metrics),
-               data_rate_metric]() {
+               data_rate_metric, ztrace_collector]() {
     return TrySeq(
         output_buffers->Next(id),
         [endpoint, id,
          requested_metrics = absl::Span<const size_t>(requested_metrics),
-         data_rate_metric,
-         output_buffers](data_endpoints_detail::NextWrite next_write) {
+         data_rate_metric, output_buffers,
+         ztrace_collector](data_endpoints_detail::NextWrite next_write) {
           GRPC_TRACE_LOG(chaotic_good, INFO)
               << "CHAOTIC_GOOD: Write " << next_write.bytes.Length()
               << "b to data endpoint #" << id;
@@ -310,9 +311,25 @@ auto Endpoint::WriteLoop(uint32_t id,
             write_args.set_metrics_sink(EventEngine::Endpoint::WriteEventSink(
                 requested_metrics,
                 {EventEngine::Endpoint::WriteEvent::kSendMsg},
-                [data_rate_metric, id, output_buffers](
-                    EventEngine::Endpoint::WriteEvent event, absl::Time,
+                [data_rate_metric, id, output_buffers, ztrace_collector,
+                 endpoint](
+                    EventEngine::Endpoint::WriteEvent event,
+                    absl::Time timestamp,
                     std::vector<EventEngine::Endpoint::WriteMetric> metrics) {
+                  ztrace_collector->Append([event, timestamp, &metrics,
+                                            endpoint = endpoint.get()]() {
+                    EndpointWriteMetricsTrace trace{timestamp, event, {}};
+                    trace.metrics.reserve(metrics.size());
+                    for (const auto [id, value] : metrics) {
+                      if (auto name =
+                              endpoint->GetEventEngineEndpoint()->GetMetricName(
+                                  id);
+                          name.has_value()) {
+                        trace.metrics.push_back({*name, value});
+                      }
+                    }
+                    return trace;
+                  });
                   if (event != EventEngine::Endpoint::WriteEvent::kSendMsg) {
                     return;
                   }
@@ -405,13 +422,14 @@ Endpoint::Endpoint(uint32_t id, RefCountedPtr<OutputBuffers> output_buffers,
               read_party->Spawn(
                   "read",
                   [id, input_queues = std::move(input_queues), endpoint,
-                   ztrace_collector = std::move(ztrace_collector)]() {
+                   ztrace_collector]() {
                     return ReadLoop(id, input_queues, endpoint,
                                     ztrace_collector);
                   },
                   [](absl::Status) {});
               return Map(
-                  WriteLoop(id, std::move(output_buffers), std::move(endpoint)),
+                  WriteLoop(id, std::move(output_buffers), std::move(endpoint),
+                            std::move(ztrace_collector)),
                   [read_party](auto x) { return x; });
             });
       },

src/core/ext/transport/chaotic_good/data_endpoints.h

@@ -221,7 +221,8 @@ class Endpoint final {
  private:
   static auto WriteLoop(uint32_t id,
                         RefCountedPtr<OutputBuffers> output_buffers,
-                        std::shared_ptr<PromiseEndpoint> endpoint);
+                        std::shared_ptr<PromiseEndpoint> endpoint,
+                        std::shared_ptr<TcpZTraceCollector> ztrace_collector);
   static auto ReadLoop(uint32_t id, RefCountedPtr<InputQueue> input_queues,
                        std::shared_ptr<PromiseEndpoint> endpoint,
                        std::shared_ptr<TcpZTraceCollector> ztrace_collector);

src/core/ext/transport/chaotic_good/tcp_ztrace_collector.h

@@ -76,6 +76,35 @@ struct WriteFrameHeaderTrace {
   }
 };
 
+struct EndpointWriteMetricsTrace {
+  absl::Time timestamp;
+  grpc_event_engine::experimental::EventEngine::Endpoint::WriteEvent
+      write_event;
+  std::vector<std::pair<absl::string_view, size_t>> metrics;
+
+  size_t MemoryUsage() const {
+    return sizeof(*this) + sizeof(metrics[0]) * metrics.size();
+  }
+
+  void RenderJson(Json::Object& object) const {
+    std::string name;
+    switch (write_event) {
+      case grpc_event_engine::experimental::EventEngine::Endpoint::WriteEvent::
+          kSendMsg:
+        name = "SEND_MSG_METRICS";
+        break;
+      default:
+        name = absl::StrCat("ENDPOINT_WRITE_METRICS_TYPE_",
+                            static_cast<int>(write_event));
+    }
+    object["metadata_type"] = Json::FromString(name);
+    object["fathom_timestamp"] = Json::FromString(absl::StrCat(timestamp));
+    for (const auto& [name, value] : metrics) {
+      object.emplace(name, Json::FromNumber(value));
+    }
+  }
+};
+
 struct WriteLargeFrameHeaderTrace {
   TcpDataFrameHeader data_header;
   std::vector<double> lb_decisions;
@@ -96,9 +125,11 @@ struct WriteLargeFrameHeaderTrace {
   }
 };
 
-using TcpZTraceCollector = channelz::ZTraceCollector<
-    tcp_ztrace_collector_detail::Config, ReadFrameHeaderTrace,
-    ReadDataHeaderTrace, WriteFrameHeaderTrace, WriteLargeFrameHeaderTrace>;
+using TcpZTraceCollector =
+    channelz::ZTraceCollector<tcp_ztrace_collector_detail::Config,
+                              ReadFrameHeaderTrace, ReadDataHeaderTrace,
+                              WriteFrameHeaderTrace, WriteLargeFrameHeaderTrace,
+                              EndpointWriteMetricsTrace>;
 
 }  // namespace grpc_core::chaotic_good