From 8a1514cca8f35b9d76defa4ddbff8dfd81b7f00c Mon Sep 17 00:00:00 2001 From: Etienne Perot Date: Mon, 6 May 2024 16:02:40 -0700 Subject: [PATCH] Profiling metrics: Buffer metric data writes on a per-line basis. This is part of a series of changes to add metric charts in performance benchmarks. Because profiling metric data may be injected within container logs, it can find itself surrounded by output from the container logs which are not profiling metric data. When using regular byte-size-based buffering, this can lead to lines being merged together with unrelated output, resulting in corrupted data. The new writer implementation instead buffers write on a per-line-number basis. The buffer is flushed when a certain number of lines has been reached. The contents are also prefixed and suffixed by `\n` to ensure the data does not share a line with the rest of the logs. PiperOrigin-RevId: 631213716 --- pkg/metric/BUILD | 1 + pkg/metric/metric_test.go | 8 +++- pkg/metric/profiling_metric.go | 75 ++++++++++++++++++++++------------ 3 files changed, 58 insertions(+), 26 deletions(-) diff --git a/pkg/metric/BUILD b/pkg/metric/BUILD index cb4ef6ae0..f7e35c12d 100644 --- a/pkg/metric/BUILD +++ b/pkg/metric/BUILD @@ -15,6 +15,7 @@ go_library( "sentry_profiling.go", "sentry_profiling_fake.go", ], + stateify = False, visibility = ["//:sandbox"], deps = [ ":metric_go_proto", diff --git a/pkg/metric/metric_test.go b/pkg/metric/metric_test.go index c02a0647f..6b40035a1 100644 --- a/pkg/metric/metric_test.go +++ b/pkg/metric/metric_test.go @@ -1129,7 +1129,10 @@ func TestMetricProfiling(t *testing.T) { lines := bufio.NewScanner(f) expectedHeader := metricsPrefix + "Time (ns)\t" + strings.Join(test.metricNames, "\t") lines.Scan() - header := lines.Text() + var header string + for header == "" && lines.Scan() { + header = lines.Text() + } if header != expectedHeader { t.Fatalf("header mismatch: got '%s' want '%s'", header, expectedHeader) } @@ -1142,6 +1145,9 @@ func TestMetricProfiling(t *testing.T) { numDatapoints := 0 for lines.Scan() { line := strings.TrimPrefix(lines.Text(), metricsPrefix) + if line == "" { + continue + } if strings.HasPrefix(line, "ADLER32\t") { continue } diff --git a/pkg/metric/profiling_metric.go b/pkg/metric/profiling_metric.go index 815ad4c80..d8423093a 100644 --- a/pkg/metric/profiling_metric.go +++ b/pkg/metric/profiling_metric.go @@ -15,11 +15,12 @@ package metric import ( - "bufio" + "bytes" "errors" "fmt" "hash" "hash/adler32" + "io" "os" "strings" "time" @@ -207,34 +208,57 @@ func collectProfilingMetrics(s *snapshots, values []func(fieldValues ...*FieldVa } } -// hashWriter is a wrapper around a bufio.Writer that also updates a hasher. -// The goal here is speed, and correctness is enforced by the reader of this -// data checking the hash at the end. -type hashWriter struct { - realWriter *bufio.Writer +const bufferedLines = 256 + +// bufferedHashWriter is a buffered writer that also keeps track of +// a hash of the data written. It flushes every `bufferedLines` lines. +type bufferedHashWriter[T io.StringWriter] struct { + buf bytes.Buffer + lines int + underlying T hasher hash.Hash32 } -// Write writes to the realWriter and also updates the hasher. -func (w *hashWriter) Write(p []byte) (int, error) { - w.hasher.Write(p) +// WriteString writes a string to the buffer and updates the hasher. +// This should *not* contain any newline character, unless this function +// is being called through NewLine. +func (w *bufferedHashWriter[T]) WriteString(s string) (int, error) { + w.hasher.Write([]byte(s)) // We ignore the effects of partial writes 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. - return w.realWriter.Write(p) + return w.buf.WriteString(s) } -// WriteString writes a string to the realWriter and also updates the hasher. -func (w *hashWriter) WriteString(s string) (int, error) { - w.hasher.Write([]byte(s)) - return w.realWriter.WriteString(s) +// Flush flushes the buffer to the underlying writer. +func (w *bufferedHashWriter[T]) Flush() { + if w.buf.Len() > 0 { + data := w.buf.String() + // Ensure that we write a complete line atomically, as this + // may get parsed while being mixed with other logs that may not + // have clean line endings a the time we print this. + if !strings.HasPrefix(data, "\n") { + data = "\n" + data + } + if !strings.HasSuffix(data, "\n") { + data = data + "\n" + } + w.underlying.WriteString(data) + w.buf.Reset() + w.lines = 0 + } } -// WriteRune writes a rune to the realWriter and also updates the hasher. -func (w *hashWriter) WriteRune(r rune) { - w.hasher.Write([]byte(string(r))) - w.realWriter.WriteRune(r) +// NewLine writes a newline character to the buffer, updates the hasher, +// and flushes the buffer if there are at least `bufferedLines` lines in +// the buffer. +func (w *bufferedHashWriter[T]) NewLine() { + w.WriteString("\n") + w.lines++ + if w.lines >= bufferedLines { + w.Flush() + } } // writeProfilingMetrics will write to the ProfilingMetricsWriter on every @@ -242,10 +266,11 @@ func (w *hashWriter) WriteRune(r rune) { func writeProfilingMetrics(s *snapshots, header string, writeReqs <-chan writeReq) { numEntries := s.numMetrics + 1 - out := hashWriter{ - realWriter: bufio.NewWriter(ProfilingMetricWriter), + out := bufferedHashWriter[*os.File]{ hasher: adler32.New(), + underlying: ProfilingMetricWriter, } + out.buf.Grow(8 * 1024 * 1024) // 8 MiB out.WriteString(header) for req := range writeReqs { @@ -258,16 +283,16 @@ func writeProfilingMetrics(s *snapshots, header string, writeReqs <-chan writeRe prometheus.WriteInteger(&out, int64(s.ringbuffer[req.ringbufferIdx][base])) // Then everything else for j := 1; j < numEntries; j++ { - out.WriteRune('\t') + out.WriteString("\t") prometheus.WriteInteger(&out, int64(s.ringbuffer[req.ringbufferIdx][base+j])) } - out.WriteRune('\n') + out.NewLine() } } - out.realWriter.WriteString(metricsPrefix) - out.realWriter.WriteString(fmt.Sprintf("ADLER32\t0x%x\n", out.hasher.Sum32())) - out.realWriter.Flush() + out.buf.WriteString(metricsPrefix) + out.buf.WriteString(fmt.Sprintf("ADLER32\t0x%x\n", out.hasher.Sum32())) + out.Flush() ProfilingMetricWriter.Close() doneProfilingMetrics <- true close(doneProfilingMetrics)