diff --git a/pkg/metric/metric_test.go b/pkg/metric/metric_test.go index eb5c3846e..3fa88d223 100644 --- a/pkg/metric/metric_test.go +++ b/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") diff --git a/pkg/metric/profiling_metric.go b/pkg/metric/profiling_metric.go index 33878e9d2..6c2684fb7 100644 --- a/pkg/metric/profiling_metric.go +++ b/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() diff --git a/pkg/prometheus/prometheus.go b/pkg/prometheus/prometheus.go index 2f5ed72cc..0ce9b14c6 100644 --- a/pkg/prometheus/prometheus.go +++ b/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 { diff --git a/test/metricsviz/metricsviz.go b/test/metricsviz/metricsviz.go index 51516face..7bb9efd94 100644 --- a/test/metricsviz/metricsviz.go +++ b/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.