From 1f65d99912a570e20a634de0b024ef896fd60764 Mon Sep 17 00:00:00 2001 From: Etienne Perot Date: Mon, 6 May 2024 20:18:10 -0700 Subject: [PATCH] Profiling metrics: Support unprefixed data logging. This is part of a series of changes to add metric charts in performance benchmarks. In contexts where profiling metrics are not being intermixed with other logs, we do not have to prefix the data with a common prefix, nor do we need the checksum. To keep profiling metric overhead to a minimum in these cases, this distinguishes two types of writers: - Normal buffered writers, which do nothing other than buffer writes to the underlying writer. - Line-buffered "lossy" writers, which do the prefixing and checksumming that the code was previously doing. Both can reuse the same underlying write loop (thanks Go generics!). PiperOrigin-RevId: 631271157 --- pkg/metric/metric_test.go | 57 ++++++- pkg/metric/profiling_metric.go | 285 +++++++++++++++++++++++---------- runsc/cmd/boot.go | 13 +- runsc/sandbox/sandbox.go | 4 +- 4 files changed, 268 insertions(+), 91 deletions(-) diff --git a/pkg/metric/metric_test.go b/pkg/metric/metric_test.go index 6b40035a1..af8022829 100644 --- a/pkg/metric/metric_test.go +++ b/pkg/metric/metric_test.go @@ -18,6 +18,7 @@ import ( "bufio" "bytes" "fmt" + "hash/adler32" "math" "os" "reflect" @@ -1038,6 +1039,7 @@ func TestMetricProfiling(t *testing.T) { name string profilingMetricsFlag string metricNames []string + lossy bool incrementMetricBy []uint64 numIterations uint64 errOnStartProfiling bool @@ -1050,6 +1052,15 @@ func TestMetricProfiling(t *testing.T) { numIterations: 100, errOnStartProfiling: false, }, + { + name: "simple single metric, lossy writer", + profilingMetricsFlag: "/foo", + metricNames: []string{"/foo"}, + lossy: true, + incrementMetricBy: []uint64{50}, + numIterations: 100, + errOnStartProfiling: false, + }, { name: "single metric exceeds snapshot buffer", profilingMetricsFlag: "/foo", @@ -1094,12 +1105,16 @@ func TestMetricProfiling(t *testing.T) { t.Fatalf("failed to create file: '%v'", err) } fName := f.Name() - ProfilingMetricWriter = f if err := Initialize(); err != nil { t.Fatalf("Initialize error: got '%v' want nil", err) } - if err := StartProfilingMetrics(test.profilingMetricsFlag, profilingRate); err != nil { + if err := StartProfilingMetrics(ProfilingMetricsOptions[*os.File]{ + Sink: f, + Lossy: test.lossy, + Metrics: test.profilingMetricsFlag, + Rate: profilingRate, + }); err != nil { if test.errOnStartProfiling { return } @@ -1126,9 +1141,12 @@ func TestMetricProfiling(t *testing.T) { } // Check the header + h := adler32.New() lines := bufio.NewScanner(f) - expectedHeader := metricsPrefix + "Time (ns)\t" + strings.Join(test.metricNames, "\t") - lines.Scan() + expectedHeader := "Time (ns)\t" + strings.Join(test.metricNames, "\t") + if test.lossy { + expectedHeader = MetricsPrefix + expectedHeader + } var header string for header == "" && lines.Scan() { header = lines.Text() @@ -1136,6 +1154,7 @@ func TestMetricProfiling(t *testing.T) { if header != expectedHeader { t.Fatalf("header mismatch: got '%s' want '%s'", header, expectedHeader) } + h.Write([]byte(strings.TrimPrefix(header, MetricsPrefix) + "\n")) // Check that data looks sane: // - Each timestamp should always be bigger. @@ -1143,14 +1162,26 @@ func TestMetricProfiling(t *testing.T) { prevTS := uint64(0) prevValues := make([]uint64, numMetrics) numDatapoints := 0 + var hashLine string for lines.Scan() { - line := strings.TrimPrefix(lines.Text(), metricsPrefix) + line := lines.Text() if line == "" { continue } - if strings.HasPrefix(line, "ADLER32\t") { + if test.lossy { + if !strings.HasPrefix(line, MetricsPrefix) { + t.Fatalf("lossy writer output does not have expected prefix: got %q want %q", line, MetricsPrefix) + } + line = strings.TrimPrefix(line, MetricsPrefix) + } + if strings.HasPrefix(line, MetricsHashIndicator) { + if hashLine != "" { + t.Fatalf("got multiple hash lines") + } + hashLine = line continue } + h.Write([]byte(line + "\n")) numDatapoints++ items := strings.Split(line, "\t") if len(items) != (numMetrics + 1) { @@ -1189,6 +1220,20 @@ func TestMetricProfiling(t *testing.T) { t.Errorf("incorrect final metric value: got %d, want %d", prevValues[i], expected) } } + if test.lossy { + if hashLine == "" { + t.Fatal("lossy writer output does not have expected hash line") + } + wantHash := fmt.Sprintf("0x%x", h.Sum32()) + gotHash := strings.TrimPrefix(hashLine, MetricsHashIndicator) + if gotHash != wantHash { + t.Fatalf("lossy writer output does not have expected hash line: got %q want %q", gotHash, wantHash) + } + } else { + if hashLine != "" { + t.Fatalf("unexpectedly found hash line in non-lossy output: %q", hashLine) + } + } f.Close() }) } diff --git a/pkg/metric/profiling_metric.go b/pkg/metric/profiling_metric.go index d8423093a..24de96529 100644 --- a/pkg/metric/profiling_metric.go +++ b/pkg/metric/profiling_metric.go @@ -21,7 +21,6 @@ import ( "hash" "hash/adler32" "io" - "os" "strings" "time" @@ -33,14 +32,14 @@ import ( const ( snapshotBufferSize = 1000 snapshotRingbufferSize = 16 - // metricsPrefix is prepended before every metrics line. - metricsPrefix = "GVISOR_METRICS\t" + // MetricsPrefix is prepended before every metrics line. + MetricsPrefix = "GVISOR_METRICS\t" + // MetricsHashIndicator is prepended before the hash of the metrics + // data at the end of the metrics stream. + MetricsHashIndicator = "ADLER32\t" ) var ( - // ProfilingMetricWriter is the output destination to which - // ProfilingMetrics will be written to in TSV format. - ProfilingMetricWriter *os.File // profilingMetricsStarted indicates whether StartProfilingMetrics has // been called. profilingMetricsStarted atomicbitops.Bool @@ -56,7 +55,7 @@ var ( ) // snapshots is used to as temporary storage of metric data -// before it's written to the ProfilingMetricWriter. +// before it's written to the writer. type snapshots struct { numMetrics int // ringbuffer is used to store metric data. @@ -73,30 +72,53 @@ type writeReq struct { numLines int } +// ProfilingMetricsWriter is the interface for profiling metrics sinks. +type ProfilingMetricsWriter interface { + // WriteString from the io.StringWriter interface. + io.StringWriter + + // Close closes the writer. + Close() error +} + +// ProfilingMetricsOptions is the set of options to profile metrics. +type ProfilingMetricsOptions[T ProfilingMetricsWriter] struct { + // Sink is the sink to write the profiling metrics data to. + Sink T + + // Lossy specifies whether the sink is lossy, i.e. data may be dropped from + // too large logging volume. In this case, data integrity is desirable at the + // expense of extra CPU cost at data-writing time. The data will be prefixed + // with `MetricsPrefix` and the hash of the data will be appended at the end. + Lossy bool + + // Metrics is the comma-separated list of metrics to profile. + Metrics string + + // Rate is the rate at which the metrics are collected. + Rate time.Duration +} + // StartProfilingMetrics checks the ProfilingMetrics runsc flags and creates // goroutines responsible for outputting the profiling metric data. // // Preconditions: // - All metrics are registered. // - Initialize/Disable has been called. -func StartProfilingMetrics(profilingMetrics string, profilingRate time.Duration) error { +func StartProfilingMetrics[T ProfilingMetricsWriter](opts ProfilingMetricsOptions[T]) error { if !initialized.Load() { // Wait for initialization to complete to make sure that all // metrics are registered. return errors.New("metric initialization is not complete") } - if ProfilingMetricWriter == nil { - return errors.New("tried to initialize profiling metrics without log file") - } var values []func(fieldValues ...*FieldValue) uint64 var header strings.Builder - header.WriteString(metricsPrefix) header.WriteString("Time (ns)") numMetrics := 0 - if len(profilingMetrics) > 0 { - metrics := strings.Split(profilingMetrics, ",") + if len(opts.Metrics) > 0 { + metrics := strings.Split(opts.Metrics, ",") numMetrics = len(metrics) for _, name := range metrics { @@ -113,8 +135,6 @@ func StartProfilingMetrics(profilingMetrics string, profilingRate time.Duration) header.WriteString(name) values = append(values, m.value) } - - header.WriteRune('\n') } 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, ",")) @@ -140,8 +160,15 @@ func StartProfilingMetrics(profilingMetrics string, profilingRate time.Duration) stopProfilingMetrics = make(chan bool, 1) doneProfilingMetrics = make(chan bool, 1) writeCh := make(chan writeReq, snapshotRingbufferSize) - go collectProfilingMetrics(&s, values, profilingRate, writeCh) - go writeProfilingMetrics(&s, header.String(), writeCh) + + go collectProfilingMetrics(&s, values, opts.Rate, writeCh) + if opts.Lossy { + lossySink := newLossyBufferedWriter(opts.Sink) + go writeProfilingMetrics[*lossyBufferedWriter[T]](lossySink, &s, header.String(), writeCh) + } else { + bufferedSink := newBufferedWriter(opts.Sink) + go writeProfilingMetrics[*bufferedWriter[T]](bufferedSink, &s, header.String(), writeCh) + } log.Infof("Profiling metrics started.") return nil @@ -208,95 +235,189 @@ func collectProfilingMetrics(s *snapshots, values []func(fieldValues ...*FieldVa } } -const bufferedLines = 256 +// bufferedMetricsWriter is a ProfilingMetricsWriter that buffers data +// before writing it to some underlying writer. +type bufferedMetricsWriter interface { + // We inherit from the ProfilingMetricsWriter interface. + // Note however that calls to WriteString should *not* contain any + // newline character, unless called through NewLine. + ProfilingMetricsWriter -// 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 + // NewLine writes a newline character to the buffer. + // The writer may decide to flush the buffer at this point. + NewLine() + + // Flush flushes the buffer to the underlying writer. + Flush() } -// 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. +const ( + // Buffer size reasonable to use for a single line of metric data. + lineBufSize = 4 * 1024 // 4 KiB + + // Buffer size for a buffered write to an underlying sink. + bufSize = 984 * 1024 // 984 KiB + + // Number of lines to buffer before flushing to the underlying sink + // by a line-buffered writer. + bufferedLines = bufSize / lineBufSize +) + +// bufferedWriter is a buffered metrics writer that wraps an underlying +// ProfilingMetricsWriter. +// It implements `bufferedMetricsWriter`. +type bufferedWriter[T ProfilingMetricsWriter] struct { + buf bytes.Buffer + underlying T +} + +func newBufferedWriter[T ProfilingMetricsWriter](underlying T) *bufferedWriter[T] { + w := &bufferedWriter[T]{underlying: underlying} + w.buf.Grow(bufSize + lineBufSize) + return w +} + +// WriteString implements bufferedMetricsWriter.WriteString. +func (w *bufferedWriter[T]) WriteString(s string) (int, error) { return w.buf.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 - } -} - -// 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 { +// NewLine implements bufferedMetricsWriter.NewLine. +func (w *bufferedWriter[T]) NewLine() { + w.buf.WriteString("\n") + if w.buf.Len() >= bufSize { w.Flush() } } +// Flush implements bufferedMetricsWriter.Flush. +func (w *bufferedWriter[T]) Flush() { + w.underlying.WriteString(w.buf.String()) + w.buf.Reset() +} + +// Close implements bufferedMetricsWriter.Close. +func (w *bufferedWriter[T]) Close() error { + w.Flush() + return w.underlying.Close() +} + +// lossyBufferedWriter writes to an underlying ProfilingMetricsWriter +// and buffers data on a per-line basis. It adds a prefix to every line, +// and keeps track of the checksum of the data it has written (which is then +// also written to the underlying writer on `Close()`). +// 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. +// `lossyBufferedWriter` implements `bufferedMetricsWriter`. +type lossyBufferedWriter[T ProfilingMetricsWriter] struct { + lineBuf bytes.Buffer + flushBuf bytes.Buffer + hasher 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, + } + w.lineBuf.Grow(lineBufSize) + + // `lineBufSize + 1` to account for the newline at the end of each line. + // `+ 2` to account for the newline at the beginning and end of each flush. + w.flushBuf.Grow((lineBufSize+1)*bufferedLines + 2) + + w.flushBuf.WriteString("\n") + return w +} + +// WriteString implements bufferedMetricsWriter.WriteString. +func (w *lossyBufferedWriter[T]) WriteString(s string) (int, error) { + return w.lineBuf.WriteString(s) +} + +// Flush implements bufferedMetricsWriter.Flush. +func (w *lossyBufferedWriter[T]) Flush() { + if w.lines > 0 { + // 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. + w.flushBuf.WriteString("\n") + w.underlying.WriteString(w.flushBuf.String()) + w.flushBuf.Reset() + w.flushBuf.WriteString("\n") + w.lines = 0 + } +} + +// 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 { + w.flushBuf.Grow(growBy) + } + w.longestLine = lineLen + } + line := w.lineBuf.String() + w.lineBuf.Reset() + w.flushBuf.WriteString(MetricsPrefix) + w.flushBuf.WriteString(line) + // 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.lineBuf.Reset() + w.lines++ + if w.lines >= bufferedLines || w.flushBuf.Len() >= bufSize { + w.Flush() + } +} + +// Close implements bufferedMetricsWriter.Close. +// It writes the checksum of the data written to the underlying writer. +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.underlying.WriteString(w.flushBuf.String()) + w.hasher.Reset() + w.lineBuf.Reset() + w.flushBuf.Reset() + return w.underlying.Close() +} + // writeProfilingMetrics will write to the ProfilingMetricsWriter on every // request via writeReqs, until writeReqs is closed. -func writeProfilingMetrics(s *snapshots, header string, writeReqs <-chan writeReq) { +func writeProfilingMetrics[T bufferedMetricsWriter](sink T, s *snapshots, header string, writeReqs <-chan writeReq) { numEntries := s.numMetrics + 1 - - out := bufferedHashWriter[*os.File]{ - hasher: adler32.New(), - underlying: ProfilingMetricWriter, - } - out.buf.Grow(8 * 1024 * 1024) // 8 MiB - out.WriteString(header) - + sink.WriteString(header) + sink.NewLine() for req := range writeReqs { s.curWriterIndex.Store(int32(req.ringbufferIdx)) - for i := 0; i < req.numLines; i++ { - out.WriteString(metricsPrefix) base := i * numEntries // Write the time - prometheus.WriteInteger(&out, int64(s.ringbuffer[req.ringbufferIdx][base])) + prometheus.WriteInteger(sink, int64(s.ringbuffer[req.ringbufferIdx][base])) // Then everything else for j := 1; j < numEntries; j++ { - out.WriteString("\t") - prometheus.WriteInteger(&out, int64(s.ringbuffer[req.ringbufferIdx][base+j])) + sink.WriteString("\t") + prometheus.WriteInteger(sink, int64(s.ringbuffer[req.ringbufferIdx][base+j])) } - out.NewLine() + sink.NewLine() } } - - out.buf.WriteString(metricsPrefix) - out.buf.WriteString(fmt.Sprintf("ADLER32\t0x%x\n", out.hasher.Sum32())) - out.Flush() - ProfilingMetricWriter.Close() + sink.Close() doneProfilingMetrics <- true close(doneProfilingMetrics) - profilingMetricsStarted.Store(false) } diff --git a/runsc/cmd/boot.go b/runsc/cmd/boot.go index 665fec975..1262ef511 100644 --- a/runsc/cmd/boot.go +++ b/runsc/cmd/boot.go @@ -156,6 +156,10 @@ type Boot struct { // profilingMetricsFD is a file descriptor to write Sentry metrics data to. profilingMetricsFD int + // profilingMetricsLossy sets whether profilingMetricsFD is a lossy channel. + // If so, the format used to write to it will contain a checksum. + profilingMetricsLossy bool + // procMountSyncFD is a file descriptor that has to be closed when the // procfs mount isn't needed anymore. procMountSyncFD int @@ -219,6 +223,7 @@ func (b *Boot) SetFlags(f *flag.FlagSet) { // Profiling flags. b.profileFDs.SetFromFlags(f) f.IntVar(&b.profilingMetricsFD, "profiling-metrics-fd", -1, "file descriptor to write sentry profiling metrics.") + f.BoolVar(&b.profilingMetricsLossy, "profiling-metrics-fd-lossy", false, "if true, treat the sentry profiling metrics FD as lossy and write a checksum to it.") } // Execute implements subcommands.Command.Execute. It starts a sandbox in a @@ -483,8 +488,12 @@ func (b *Boot) Execute(_ context.Context, f *flag.FlagSet, args ...any) subcomma // registered metrics prior to sending the start signal. metric.Initialize() if b.profilingMetricsFD != -1 { - metric.ProfilingMetricWriter = os.NewFile(uintptr(b.profilingMetricsFD), "metrics file") - if err := metric.StartProfilingMetrics(conf.ProfilingMetrics, time.Duration(conf.ProfilingMetricsRate)*time.Microsecond); err != nil { + if err := metric.StartProfilingMetrics(metric.ProfilingMetricsOptions[*os.File]{ + Sink: os.NewFile(uintptr(b.profilingMetricsFD), "metrics file"), + Lossy: b.profilingMetricsLossy, + Metrics: conf.ProfilingMetrics, + Rate: time.Duration(conf.ProfilingMetricsRate) * time.Microsecond, + }); err != nil { l.Destroy() util.Fatalf("unable to start profiling metrics: %v", err) } diff --git a/runsc/sandbox/sandbox.go b/runsc/sandbox/sandbox.go index b5feb4228..86cdfbd5b 100644 --- a/runsc/sandbox/sandbox.go +++ b/runsc/sandbox/sandbox.go @@ -1074,10 +1074,12 @@ func (s *Sandbox) createSandboxProcess(conf *config.Config, args *Args, startSyn donations.Donate("stdio-fds", stdios[:]...) if conf.ProfilingMetricsLog == "-" { donations.Donate("profiling-metrics-fd", stdios[1]) - } else { + cmd.Args = append(cmd.Args, "--profiling-metrics-fd-lossy=true") + } else if conf.ProfilingMetricsLog != "" { if err := donations.DonateDebugLogFile("profiling-metrics-fd", conf.ProfilingMetricsLog, "metrics", test); err != nil { return err } + cmd.Args = append(cmd.Args, "--profiling-metrics-fd-lossy=false") } totalSysMem, err := totalSystemMemory()