Profiling metrics: Add per-line checksum to output.

This allows detecting data corruption more finely.

A future change will make these errors skipped over, allowing data to
still be visualized even if partially corrupt.

PiperOrigin-RevId: 634038986

pkg/metric/metric_test.go

@@ -1151,6 +1151,9 @@ func TestMetricProfiling(t *testing.T) {
 			h := adler32.New()
 			lines := bufio.NewScanner(f)
 			expectedHeader := TimeColumn + "\t" + strings.Join(test.metricNames, "\t")
+			if test.lossy {
+				expectedHeader += "\tChecksum"
+			}
 			prevTS := uint64(0)
 			prevValues := make([]uint64, numMetrics)
 			numDatapoints := 0
@@ -1177,6 +1180,17 @@ func TestMetricProfiling(t *testing.T) {
 					continue
 				}
 				h.Write([]byte(line + "\n"))
+				// Check line checksum
+				if test.lossy {
+					tabSplit := strings.Split(line, "\t")
+					gotLineChecksum := tabSplit[len(tabSplit)-1]
+					wantLineChecksum := fmt.Sprintf("0x%x", adler32.Checksum([]byte(strings.Join(tabSplit[:len(tabSplit)-1], "\t"))))
+					if gotLineChecksum != wantLineChecksum {
+						t.Errorf("got line checksum %q, want %q", gotLineChecksum, wantLineChecksum)
+						continue
+					}
+					line = strings.TrimSuffix(line, "\t"+gotLineChecksum)
+				}
 				if strings.HasPrefix(line, MetricsMetaIndicator) {
 					line = strings.TrimPrefix(line, MetricsMetaIndicator)
 					components := strings.Split(line, "\t")

pkg/metric/profiling_metric.go

@@ -159,6 +159,9 @@ func StartProfilingMetrics[T ProfilingMetricsWriter](opts ProfilingMetricsOption
 			columnHeaders.WriteString(name)
 			values = append(values, m.value)
 		}
+		if opts.Lossy {
+			columnHeaders.WriteString("\tChecksum")
+		}
 	} else {
 		if len(definedProfilingMetrics) > 0 {
 			return fmt.Errorf("a value for --profiling-metrics was not specified; consider using a subset of '--profiling-metrics=%s'", strings.Join(definedProfilingMetrics, ","))
@@ -337,22 +340,27 @@ func (w *bufferedWriter[T]) Close() error {
 // The checksum covers all of the per-line data written after the line prefix,
 // including the newline character of these lines, with the exception of
 // the checksum data line itself.
+// All lines are also checksummed individually, with the checksum covering
+// the contents of the line after the line prefix but before the tab and
+// line checksum itself at the end of the line.
 // `lossyBufferedWriter` implements `bufferedMetricsWriter`.
 type lossyBufferedWriter[T ProfilingMetricsWriter] struct {
-	lineBuf     bytes.Buffer
-	flushBuf    bytes.Buffer
-	hasher      hash.Hash32
-	lines       int
-	longestLine int
-	underlying  T
+	lineBuf       bytes.Buffer
+	flushBuf      bytes.Buffer
+	lineHasher    hash.Hash32
+	overallHasher hash.Hash32
+	lines         int
+	longestLine   int
+	underlying    T
 }
 
 // newLossyBufferedWriter creates a new lossyBufferedWriter.
 func newLossyBufferedWriter[T ProfilingMetricsWriter](underlying T) *lossyBufferedWriter[T] {
 	w := &lossyBufferedWriter[T]{
-		underlying:  underlying,
-		hasher:      adler32.New(),
-		longestLine: lineBufSize,
+		underlying:    underlying,
+		lineHasher:    adler32.New(),
+		overallHasher: adler32.New(),
+		longestLine:   lineBufSize,
 	}
 	w.lineBuf.Grow(lineBufSize)
 
@@ -389,7 +397,6 @@ func (w *lossyBufferedWriter[T]) Flush() {
 
 // NewLine implements bufferedMetricsWriter.NewLine.
 func (w *lossyBufferedWriter[T]) NewLine() {
-	w.lineBuf.WriteString("\n")
 	if lineLen := w.lineBuf.Len(); lineLen > w.longestLine {
 		wantTotalSize := (lineLen+1)*bufferedLines + 2
 		if growBy := wantTotalSize - w.flushBuf.Len(); growBy > 0 {
@@ -398,15 +405,23 @@ func (w *lossyBufferedWriter[T]) NewLine() {
 		w.longestLine = lineLen
 	}
 	line := w.lineBuf.String()
+	w.lineHasher.Reset()
+	w.lineHasher.Write([]byte(line))
+	lineHash := w.lineHasher.Sum32()
 	w.lineBuf.Reset()
 	w.flushBuf.WriteString(MetricsPrefix)
+	beforeLineIndex := w.flushBuf.Len()
 	w.flushBuf.WriteString(line)
+	w.flushBuf.WriteString("\t0x")
+	prometheus.WriteHex(&w.flushBuf, uint64(lineHash))
+	w.flushBuf.WriteString("\n")
+	afterLineIndex := w.flushBuf.Len()
 	// We ignore the effects that partial writes on the underlying writer
 	// would have on the hash computation here.
 	// This is OK because the goal of this writer is speed over correctness,
 	// and correctness is enforced by the reader of this data checking the
 	// hash at the end.
-	w.hasher.Write([]byte(line))
+	w.overallHasher.Write(w.flushBuf.Bytes()[beforeLineIndex:afterLineIndex])
 	w.lineBuf.Reset()
 	w.lines++
 	if w.lines >= bufferedLines || w.flushBuf.Len() >= bufSize {
@@ -419,9 +434,12 @@ func (w *lossyBufferedWriter[T]) NewLine() {
 func (w *lossyBufferedWriter[T]) Close() error {
 	w.Flush()
 	w.flushBuf.WriteString(MetricsPrefix)
-	w.flushBuf.WriteString(fmt.Sprintf("%s0x%x\n", MetricsHashIndicator, w.hasher.Sum32()))
+	w.flushBuf.WriteString(MetricsHashIndicator)
+	w.flushBuf.WriteString("0x")
+	prometheus.WriteHex(&w.flushBuf, uint64(w.overallHasher.Sum32()))
+	w.flushBuf.WriteString("\n")
 	w.underlying.WriteString(w.flushBuf.String())
-	w.hasher.Reset()
+	w.overallHasher.Reset()
 	w.lineBuf.Reset()
 	w.flushBuf.Reset()
 	return w.underlying.Close()

pkg/prometheus/prometheus.go

@@ -270,6 +270,30 @@ func WriteInteger[T io.StringWriter](w T, val int64) (int, error) {
 	return written, nil
 }
 
+// WriteHex writes the given integer as hex to a writer
+// without allocating strings.
+//
+//go:nosplit
+func WriteHex[T io.StringWriter](w T, val uint64) (int, error) {
+	const hexDigits = "0123456789abcdef"
+	if val == 0 {
+		return w.WriteString(hexDigits[0:1])
+	}
+	var written int
+	hex := uint64(16)
+	for ; val/hex != 0; hex <<= 4 {
+	}
+	for hex >>= 4; hex > 0; hex >>= 4 {
+		digit := (val / hex) % 16
+		n, err := w.WriteString(hexDigits[digit : digit+1])
+		written += n
+		if err != nil {
+			return written, err
+		}
+	}
+	return written, nil
+}
+
 // writeNumberTo writes the number to the given writer.
 // This only causes heap allocations when the number is a non-zero, non-special float.
 func writeNumberTo[T io.StringWriter](w T, n *Number) error {

test/metricsviz/metricsviz.go

@@ -432,7 +432,8 @@ func Parse(logs string, hasPrefix bool) (*Data, error) {
 	data := &Data{data: make(map[MetricAndFields]*TimeSeries)}
 	var header []MetricAndFields
 	metricsMeta := make(map[MetricName]*Metric)
-	h := adler32.New()
+	lineChecksum := adler32.New()
+	overallChecksum := adler32.New()
 	checkedHash := false
 	metricsLineFound := false
 	for _, line := range strings.Split(logs, "\n") {
@@ -453,7 +454,7 @@ func Parse(logs string, hasPrefix bool) (*Data, error) {
 				return nil, fmt.Errorf("invalid hash line: %q: %w", line, err)
 			}
 			wantHash := uint32(wantHashInt64)
-			if gotHash := h.Sum32(); gotHash != wantHash {
+			if gotHash := overallChecksum.Sum32(); gotHash != wantHash {
 				return nil, fmt.Errorf("checksum mismatch: computed 0x%x, logs said it should be 0x%x. This is likely due to a log buffer overrun or similar issue causing some lines to be omitted; please configure the container or the runtime to allow higher logging volume", gotHash, wantHash)
 			}
 			checkedHash = true
@@ -462,8 +463,23 @@ func Parse(logs string, hasPrefix bool) (*Data, error) {
 
 		// If it's not a hash line, add it to the hash regardless of which other
 		// type of line it is.
-		h.Write([]byte(lineData))
-		h.Write([]byte("\n"))
+		overallChecksum.Write([]byte(lineData))
+		overallChecksum.Write([]byte("\n"))
+
+		if hasPrefix {
+			// There should be a per-line checksum at the end of each line.
+			tabSplit := strings.Split(lineData, "\t")
+			if len(tabSplit) < 2 {
+				return nil, fmt.Errorf("invalid line: %q (no tab separator found)", line)
+			}
+			lineChecksum.Reset()
+			lineChecksum.Write([]byte(strings.Join(tabSplit[:len(tabSplit)-1], "\t")))
+			wantLineChecksum := fmt.Sprintf("0x%x", lineChecksum.Sum32())
+			if gotLineChecksum := tabSplit[len(tabSplit)-1]; gotLineChecksum != wantLineChecksum {
+				return nil, fmt.Errorf("per-line checksum mismatch: computed 0x%x, line said it should be 0x%x. This is likely due to a log buffer overrun or similar issue causing some lines to be omitted; please configure the container or the runtime to allow higher logging volume", gotLineChecksum, wantLineChecksum)
+			}
+			lineData = strings.Join(tabSplit[:len(tabSplit)-1], "\t")
+		}
 
 		if strings.HasPrefix(lineData, metric.MetricsMetaIndicator) {
 			lineMetadata := strings.TrimPrefix(lineData, metric.MetricsMetaIndicator)
@@ -500,7 +516,11 @@ func Parse(logs string, hasPrefix bool) (*Data, error) {
 			if headerCells[0] != metric.TimeColumn {
 				return nil, fmt.Errorf("invalid header line: %q", line)
 			}
-			for _, cell := range headerCells[1:] {
+			for i, cell := range headerCells[1:] {
+				if hasPrefix && i == len(headerCells)-2 && cell == "Checksum" {
+					// Ignore this column name; it is the column indicator for the per-line checksum.
+					continue
+				}
 				// If metric fields were to be implemented, they would be part of
 				// the header cell here. For now we just assume that the header
 				// cells are just metric names.