feat: show disk/network throughput with bottleneck indicator

Replace generic "source/effective" labels with operation-aware labels
(push: disk read / network write, pull: network read / disk write).
Derive sink throughput from transferTime - sourceReadTime. Mark the
slower side as bottleneck. Add overall effective throughput to output.

Signed-off-by: Zhao Chen <winters.zc@antgroup.com>

Author: aftersnow

pkg/iometrics/tracker.go

@@ -66,6 +66,22 @@ func (t *Tracker) TrackTransfer(fn func() error) error {
 	return err
 }
 
+// sourceLabel and sinkLabel return human-readable labels for the source
+// and sink sides based on the operation type.
+func (t *Tracker) sourceLabel() string {
+	if t.operation == "push" {
+		return "disk read"
+	}
+	return "network read"
+}
+
+func (t *Tracker) sinkLabel() string {
+	if t.operation == "push" {
+		return "network write"
+	}
+	return "disk write"
+}
+
 // Summary outputs a throughput summary to both the log file (logrus)
 // and the terminal (stderr). Call this after all goroutines have
 // completed (after g.Wait()) — the happens-before from errgroup
@@ -80,30 +96,43 @@ func (t *Tracker) Summary() {
 		return
 	}
 
-	sourceReadTime := time.Duration(sourceNanos)
+	sourceDuration := time.Duration(sourceNanos)
+	sinkNanos := transferNanos - sourceNanos
+	sinkDuration := time.Duration(max(sinkNanos, 0))
 
-	var readFraction float64
-	if transferNanos > 0 {
-		readFraction = float64(sourceNanos) / float64(transferNanos)
+	sourceThroughput := formatThroughput(totalBytes, sourceDuration)
+	sinkThroughput := formatThroughput(totalBytes, sinkDuration)
+
+	// Identify the bottleneck by comparing cumulative durations.
+	bottleneck := t.sinkLabel()
+	if sourceNanos > sinkNanos {
+		bottleneck = t.sourceLabel()
 	}
 
 	// Log structured fields to log file.
 	logrus.WithFields(logrus.Fields{
 		"operation":            t.operation,
 		"totalBytes":           formatBytes(uint64(totalBytes)),
 		"wallClock":            wallClock.Round(time.Millisecond).String(),
-		"sourceReadTime":       sourceReadTime.Round(time.Millisecond).String(),
-		"sourceReadThroughput": formatThroughput(totalBytes, sourceReadTime),
 		"effectiveThroughput":  formatThroughput(totalBytes, wallClock),
-		"readFraction":         fmt.Sprintf("%.2f", readFraction),
+		t.sourceLabel():        sourceThroughput,
+		t.sinkLabel():          sinkThroughput,
+		"bottleneck":           bottleneck,
 	}).Info("io throughput summary")
 
 	// Print concise summary to terminal.
-	fmt.Fprintf(os.Stderr, "IO summary: %s in %s, effective %s, source %s, read ratio %.2f\n",
+	srcArrow := ""
+	snkArrow := ""
+	if bottleneck == t.sourceLabel() {
+		srcArrow = " ← bottleneck"
+	} else {
+		snkArrow = " ← bottleneck"
+	}
+	fmt.Fprintf(os.Stderr, "IO summary: %s in %s, %s | %s: %s%s | %s: %s%s\n",
 		formatBytes(uint64(totalBytes)),
 		wallClock.Round(time.Millisecond),
 		formatThroughput(totalBytes, wallClock),
-		formatThroughput(totalBytes, sourceReadTime),
-		readFraction,
+		t.sourceLabel(), sourceThroughput, srcArrow,
+		t.sinkLabel(), sinkThroughput, snkArrow,
 	)
 }

pkg/iometrics/tracker_test.go

@@ -210,44 +210,60 @@ func TestSummaryOutput(t *testing.T) {
 		"io throughput summary",
 		"operation=push",
 		"totalBytes=",
-		"readFraction=",
+		"bottleneck=",
+		"disk read",
+		"network write",
 	} {
 		if !strings.Contains(output, expected) {
 			t.Errorf("log output missing %q, got:\n%s", expected, output)
 		}
 	}
 }
 
-func TestReadFractionAccuracy(t *testing.T) {
-	tracker := NewTracker("test")
+func TestBottleneckDetection(t *testing.T) {
+	// Simulate source-bottleneck: source read takes 80ms, sink takes 20ms.
+	t.Run("source bottleneck", func(t *testing.T) {
+		tracker := NewTracker("pull")
+		sr := &slowReader{data: make([]byte, 64), delay: 80 * time.Millisecond}
+		wrapped := tracker.WrapReader(sr)
+		tracker.TrackTransfer(func() error {
+			_, err := io.ReadAll(wrapped)
+			if err != nil {
+				return err
+			}
+			time.Sleep(20 * time.Millisecond)
+			return nil
+		})
 
-	// Simulate: source read takes 80ms, total transfer takes 100ms.
-	// readFraction should be ~0.8.
-	sr := &slowReader{data: make([]byte, 64), delay: 80 * time.Millisecond}
-	wrapped := tracker.WrapReader(sr)
-	tracker.TrackTransfer(func() error {
-		_, err := io.ReadAll(wrapped)
-		if err != nil {
-			return err
+		sourceNanos := tracker.sourceNanos.Load()
+		sinkNanos := tracker.transferNanos.Load() - sourceNanos
+		if sourceNanos <= sinkNanos {
+			t.Errorf("expected source > sink for source bottleneck: source=%d, sink=%d",
+				sourceNanos, sinkNanos)
 		}
-		// Simulate 20ms of sink write time.
-		time.Sleep(20 * time.Millisecond)
-		return nil
 	})
 
-	sourceNanos := tracker.sourceNanos.Load()
-	transferNanos := tracker.transferNanos.Load()
-
-	if transferNanos == 0 {
-		t.Fatal("transferNanos should be > 0")
-	}
+	// Simulate sink-bottleneck: source read takes 10ms, sink takes 80ms.
+	t.Run("sink bottleneck", func(t *testing.T) {
+		tracker := NewTracker("push")
+		sr := &slowReader{data: make([]byte, 64), delay: 10 * time.Millisecond}
+		wrapped := tracker.WrapReader(sr)
+		tracker.TrackTransfer(func() error {
+			_, err := io.ReadAll(wrapped)
+			if err != nil {
+				return err
+			}
+			time.Sleep(80 * time.Millisecond)
+			return nil
+		})
 
-	readFraction := float64(sourceNanos) / float64(transferNanos)
-	// Expect readFraction to be roughly 0.8 (with tolerance for scheduling jitter).
-	if readFraction < 0.5 || readFraction > 0.95 {
-		t.Errorf("readFraction = %.2f, expected ~0.8 (sourceNanos=%d, transferNanos=%d)",
-			readFraction, sourceNanos, transferNanos)
-	}
+		sourceNanos := tracker.sourceNanos.Load()
+		sinkNanos := tracker.transferNanos.Load() - sourceNanos
+		if sinkNanos <= sourceNanos {
+			t.Errorf("expected sink > source for sink bottleneck: source=%d, sink=%d",
+				sourceNanos, sinkNanos)
+		}
+	})
 }
 
 func TestFormatBytes(t *testing.T) {