jobs: reformat dumped trace stats

Previously we had one file that had cluster-wide and per-node stats.
It dumped per-node first, and cluster-wide after, making it harder to find. It also dumped the nodes in random (map) order.

Instead, we now dump the cluster-wide stats to their own file, making
these easy to find/consume, and the per-node stats are dumped in node
order, making them easier to browse as well. The formatting is also now
tweaked to make the sections clearer.

Finally the files are renamed to be timestamp-then-name, so that each time one
dumps stats, all the dumped files for that request sort togeher, vs sorting by type of file first.

Release note: none.
Epic: none.

Author: dt

pkg/util/bulk/aggregator_stats.go

@@ -9,6 +9,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"sort"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
@@ -95,69 +96,76 @@ func FlushTracingAggregatorStats(
 	db isql.DB,
 	perNodeAggregatorStats ComponentAggregatorStats,
 ) 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")
+	clusterWideSpanStats := make(map[string]tracingpb.OperationMetadata)
+	clusterWideAggregatorStats := make(map[string]tracing.AggregatorEvent)
+	ids := make([]execinfrapb.ComponentID, 0, len(perNodeAggregatorStats))
 
-		var clusterWideSummary bytes.Buffer
-		for component, nameToEvent := range perNodeAggregatorStats {
-			clusterWideSummary.WriteString(fmt.Sprintf("## SQL Instance ID: %s; Flow ID: %s\n",
-				component.SQLInstanceID.String(), component.FlowID.String()))
-
-			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.
-				filename := fmt.Sprintf("%s/%s",
-					component.SQLInstanceID.String(), asOf)
-				msg, err := protoreflect.DecodeMessage(name, event)
-				if err != nil {
-					clusterWideSummary.WriteString(fmt.Sprintf("invalid protocol message: %v", err))
-					// If we failed to decode the event write the error to the file and
-					// carry on.
-					continue
-				}
-
-				if err := jobs.WriteProtobinExecutionDetailFile(ctx, filename, msg, txn, jobID); err != nil {
-					return err
-				}
-
-				// Construct a single text file that contains information on a per-node
-				// basis as well as a cluster-wide aggregate.
-				clusterWideSummary.WriteString(fmt.Sprintf("# %s\n", name))
-
-				aggEvent := msg.(tracing.AggregatorEvent)
-				clusterWideSummary.WriteString(aggEvent.String())
-				clusterWideSummary.WriteString("\n")
-
-				if _, ok := clusterWideAggregatorStats[name]; ok {
-					clusterWideAggregatorStats[name].Combine(aggEvent)
-				} else {
-					clusterWideAggregatorStats[name] = aggEvent
-				}
-			}
+	for component := range perNodeAggregatorStats {
+		ids = append(ids, component)
+	}
+	sort.Slice(ids, func(i, j int) bool { return ids[i].SQLInstanceID < ids[j].SQLInstanceID })
+
+	// Write a summary for each per-node to a buffer. While doing so, accumulate a
+	// cluster-wide summary as well to be written to a second buffer below.
+	var perNode bytes.Buffer
+	fmt.Fprintf(&perNode, "# Per-componant Details (%d)\n", len(perNodeAggregatorStats))
+	for _, component := range ids {
+		nodeStats := perNodeAggregatorStats[component]
+		fmt.Fprintf(&perNode, "# SQL Instance ID: %s (%s); Flow/proc ID: %s/%d\n\n",
+			component.SQLInstanceID, component.Region, component.FlowID, component.ID)
+
+		// Print span stats.
+		perNode.WriteString("## Span Totals\n\n")
+		for name, stats := range nodeStats.SpanTotals {
+			fmt.Fprintf(&perNode, "- %-40s (%d):\t%s\n", name, stats.Count, stats.Duration)
+		}
+		perNode.WriteString("\n")
 
-			clusterWideSummary.WriteString("### span metadata\n\n")
+		// Add span stats to the cluster-wide span stats.
+		for spanName, totals := range nodeStats.SpanTotals {
+			clusterWideSpanStats[spanName] = clusterWideSpanStats[spanName].Combine(totals)
+		}
 
-			for name, metadata := range nameToEvent.SpanTotals {
-				fmt.Fprintf(&clusterWideSummary, " - %s (%d): %s\n", name, metadata.Count, metadata.Duration)
-				clusterWideOpMetadata[name] = clusterWideOpMetadata[name].Combine(metadata)
+		perNode.WriteString("## Aggregate Stats\n\n")
+		for name, event := range nodeStats.Events {
+			msg, err := protoreflect.DecodeMessage(name, event)
+			if err != nil {
+				continue
+			}
+			aggEvent := msg.(tracing.AggregatorEvent)
+			fmt.Fprintf(&perNode, "- %s:\n%s\n\n", name, aggEvent)
+
+			// Populate the cluster-wide aggregator stats.
+			if _, ok := clusterWideAggregatorStats[name]; ok {
+				clusterWideAggregatorStats[name].Combine(aggEvent)
+			} else {
+				clusterWideAggregatorStats[name] = aggEvent
 			}
 		}
+		perNode.WriteString("\n")
+	}
 
-		for tag, event := range clusterWideAggregatorStats {
-			clusterWideSummary.WriteString("## Cluster-wide\n\n")
-			clusterWideSummary.WriteString(fmt.Sprintf("# %s\n", tag))
-			clusterWideSummary.WriteString(event.String())
-		}
+	// Write the cluster-wide summary.
+	var combined bytes.Buffer
+	combined.WriteString("# Cluster-wide\n\n")
+	combined.WriteString("## Span Totals\n\n")
+	for name, stats := range clusterWideSpanStats {
+		fmt.Fprintf(&combined, " - %-40s (%d):\t%s\n", name, stats.Count, stats.Duration)
+	}
+	combined.WriteString("\n")
+	combined.WriteString("## Aggregate Stats\n\n")
+	for name, ev := range clusterWideAggregatorStats {
+		fmt.Fprintf(&combined, " - %s:\n%s\n", name, ev)
+	}
+	combined.WriteString("\n")
 
-		// Ensure the file always has a trailing newline, regardless of whether or
-		// not the loops above wrote anything.
-		clusterWideSummary.WriteString("\n")
-		filename := fmt.Sprintf("aggregatorstats.%s.txt", asOf)
-		return jobs.WriteExecutionDetailFile(ctx, filename, clusterWideSummary.Bytes(), txn, jobID)
+	return db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		asOf := timeutil.Now().Format("20060102_150405.00")
+		combinedFilename := fmt.Sprintf("%s/trace-stats-cluster-wide.txt", asOf)
+		perNodeFilename := fmt.Sprintf("%s/trace-stats-by-node.txt", asOf)
+		if err := jobs.WriteExecutionDetailFile(ctx, combinedFilename, combined.Bytes(), txn, jobID); err != nil {
+			return err
+		}
+		return jobs.WriteExecutionDetailFile(ctx, perNodeFilename, perNode.Bytes(), txn, jobID)
 	})
 }