tracing,execinfrapb: include child metadata in trace agg metas

Release note: none.
Epic: none.

Author: dt

pkg/backup/backup_job.go

@@ -332,7 +332,7 @@ func backup(
 			// Update the running aggregate of the component with the latest received
 			// aggregate.
 			resumer.mu.Lock()
-			resumer.mu.perNodeAggregatorStats[componentID] = agg.Events
+			resumer.mu.perNodeAggregatorStats[componentID] = *agg
 			resumer.mu.Unlock()
 		}
 		return nil

pkg/backup/restore_job.go

@@ -498,7 +498,7 @@ func restore(
 			// Update the running aggregate of the component with the latest received
 			// aggregate.
 			resumer.mu.Lock()
-			resumer.mu.perNodeAggregatorStats[componentID] = agg.Events
+			resumer.mu.perNodeAggregatorStats[componentID] = *agg
 			resumer.mu.Unlock()
 		}
 		return nil

pkg/crosscluster/logical/logical_replication_job.go

@@ -751,7 +751,7 @@ func (rh *rowHandler) handleTraceAgg(agg *execinfrapb.TracingAggregatorEvents) {
 	// aggregate.
 	rh.r.mu.Lock()
 	defer rh.r.mu.Unlock()
-	rh.r.mu.perNodeAggregatorStats[componentID] = agg.Events
+	rh.r.mu.perNodeAggregatorStats[componentID] = *agg
 }
 
 func (rh *rowHandler) handleMeta(ctx context.Context, meta *execinfrapb.ProducerMetadata) error {

pkg/crosscluster/physical/stream_ingestion_dist.go

@@ -172,7 +172,7 @@ func startDistIngestion(
 			// Update the running aggregate of the component with the latest received
 			// aggregate.
 			resumer.mu.Lock()
-			resumer.mu.perNodeAggregatorStats[componentID] = agg.Events
+			resumer.mu.perNodeAggregatorStats[componentID] = *agg
 			resumer.mu.Unlock()
 		}
 		return nil

pkg/sql/execinfrapb/data.proto

@@ -379,4 +379,6 @@ message TracingAggregatorEvents {
     (gogoproto.customname) = "FlowID",
     (gogoproto.customtype) = "FlowID"];
   map<string, bytes> events = 3;
+
+  map<string, util.tracing.tracingpb.OperationMetadata> span_totals = 4 [(gogoproto.nullable) = false];
 }

pkg/util/bulk/BUILD.bazel

@@ -18,5 +18,6 @@ go_library(
         "//pkg/util/log",
         "//pkg/util/timeutil",
         "//pkg/util/tracing",
+        "//pkg/util/tracing/tracingpb",
     ],
 )

pkg/util/bulk/aggregator_stats.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb"
 )
 
 // ConstructTracingAggregatorProducerMeta constructs a ProducerMetadata that
@@ -45,25 +46,37 @@ func ConstructTracingAggregatorProducerMeta(
 		}
 	})
 
+	sp := tracing.SpanFromContext(ctx)
+	if sp != nil {
+		recType := sp.RecordingType()
+		if recType != tracingpb.RecordingOff {
+			aggEvents.SpanTotals = sp.GetFullTraceRecording(recType).Root.ChildrenMetadata
+		}
+	}
 	return &execinfrapb.ProducerMetadata{AggregatorEvents: aggEvents}
 }
 
 // ComponentAggregatorStats is a mapping from a component to all the Aggregator
 // Stats collected for that component.
-type ComponentAggregatorStats map[execinfrapb.ComponentID]map[string][]byte
+type ComponentAggregatorStats map[execinfrapb.ComponentID]execinfrapb.TracingAggregatorEvents
 
 // DeepCopy takes a deep copy of the component aggregator stats map.
 func (c ComponentAggregatorStats) DeepCopy() ComponentAggregatorStats {
 	mapCopy := make(ComponentAggregatorStats, len(c))
 	for k, v := range c {
-		innerMap := make(map[string][]byte, len(v))
-		for k2, v2 := range v {
+		copied := v
+		copied.Events = make(map[string][]byte, len(v.Events))
+		copied.SpanTotals = make(map[string]tracingpb.OperationMetadata, len(v.SpanTotals))
+		for k2, v2 := range v.Events {
 			// Create a copy of the byte slice to avoid modifying the original data.
 			dataCopy := make([]byte, len(v2))
 			copy(dataCopy, v2)
-			innerMap[k2] = dataCopy
+			copied.Events[k2] = dataCopy
 		}
-		mapCopy[k] = innerMap
+		for k2, v2 := range v.SpanTotals {
+			copied.SpanTotals[k2] = v2
+		}
+		mapCopy[k] = copied
 	}
 	return mapCopy
 }
@@ -84,13 +97,17 @@ func FlushTracingAggregatorStats(
 ) error {
 	return db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 		clusterWideAggregatorStats := make(map[string]tracing.AggregatorEvent)
+		clusterWideOpMetadata := make(map[string]tracingpb.OperationMetadata)
 		asOf := timeutil.Now().Format("20060102_150405.00")
 
 		var clusterWideSummary bytes.Buffer
 		for component, nameToEvent := range perNodeAggregatorStats {
-			clusterWideSummary.WriteString(fmt.Sprintf("## SQL Instance ID: %s; Flow ID: %s\n\n",
+			clusterWideSummary.WriteString(fmt.Sprintf("## SQL Instance ID: %s; Flow ID: %s\n",
 				component.SQLInstanceID.String(), component.FlowID.String()))
-			for name, event := range nameToEvent {
+
+			clusterWideSummary.WriteString("### aggregated events\n\n")
+
+			for name, event := range nameToEvent.Events {
 				// Write a proto file per tag. This machine-readable file can be consumed
 				// by other places we want to display this information egs: annotated
 				// DistSQL diagrams, DBConsole etc.
@@ -122,6 +139,13 @@ func FlushTracingAggregatorStats(
 					clusterWideAggregatorStats[name] = aggEvent
 				}
 			}
+
+			clusterWideSummary.WriteString("### span metadata\n\n")
+
+			for name, metadata := range nameToEvent.SpanTotals {
+				fmt.Fprintf(&clusterWideSummary, " - %s (%d): %s\n", name, metadata.Count, metadata.Duration)
+				clusterWideOpMetadata[name] = clusterWideOpMetadata[name].Combine(metadata)
+			}
 		}
 
 		for tag, event := range clusterWideAggregatorStats {

pkg/util/tracing/span.go

@@ -315,6 +315,16 @@ func (sp *Span) GetRecording(recType tracingpb.RecordingType) tracingpb.Recordin
 	return sp.i.GetRecording(recType, false /* finishing */)
 }
 
+func (sp *Span) GetFullTraceRecording(recType tracingpb.RecordingType) Trace {
+	if sp.detectUseAfterFinish() {
+		return Trace{}
+	}
+	if sp.RecordingType() == tracingpb.RecordingOff {
+		return Trace{}
+	}
+	return sp.i.GetFullTraceRecording(recType)
+}
+
 // GetConfiguredRecording is like GetRecording, except the type of recording it
 // returns is the one that the span has been previously configured with.
 //

pkg/util/tracing/span_inner.go

@@ -65,6 +65,12 @@ func (s *spanInner) GetTraceRecording(recType tracingpb.RecordingType, finishing
 	return s.crdb.GetRecording(recType, finishing)
 }
 
+// GetFullTraceRecording returns the span's full recording, including detached
+// children, as a Trace. See GetTraceRecording and WithDetachedRecording.
+func (s *spanInner) GetFullTraceRecording(recType tracingpb.RecordingType) Trace {
+	return s.crdb.GetFullRecording(recType)
+}
+
 // GetRecording returns the span's recording.
 //
 // finishing indicates whether s is in the process of finishing. If it isn't,