From b096b98d6fce8f76753e35eb1773706bcbcd95e2 Mon Sep 17 00:00:00 2001 From: Etienne Perot Date: Wed, 4 Jan 2023 12:28:08 -0800 Subject: [PATCH] gVisor Prometheus lib: Group same-name metrics when writing multiple snapshots This removes `prometheus.Snapshot.WriteTo` and replaces it with a `Write` function that handles writing multiple `Snapshot` objects to the same writer. This is necessary for following the OpenMetrics spec more closely, which e.g. requires same-name metrics to be grouped together in the output. When rendering data from multiple `Snapshot`s from multiple sandboxes, this grouping was not respected. This change is part of a series of changes to support Prometheus-style metrics in `runsc`. PiperOrigin-RevId: 499551401 --- pkg/metric/metric_test.go | 4 +- pkg/prometheus/prometheus.go | 123 ++++++++++++++++++++-------- pkg/prometheus/prometheus_test.go | 128 ++++++++++++++++++++++++------ runsc/cmd/metric_export.go | 17 ++-- 4 files changed, 205 insertions(+), 67 deletions(-) diff --git a/pkg/metric/metric_test.go b/pkg/metric/metric_test.go index bccac60b7..b356ef9f6 100644 --- a/pkg/metric/metric_test.go +++ b/pkg/metric/metric_test.go @@ -43,7 +43,9 @@ const ( func verifyPrometheusParsing(t *testing.T) { t.Helper() var buf bytes.Buffer - if _, err := GetSnapshot().WriteTo(&buf, prometheus.ExportOptions{}); err != nil { + if _, err := prometheus.Write(&buf, prometheus.ExportOptions{}, map[*prometheus.Snapshot]prometheus.SnapshotExportOptions{ + GetSnapshot(): prometheus.SnapshotExportOptions{}, + }); err != nil { t.Errorf("failed to get Prometheus snapshot: %v", err) } else if _, err := (&expfmt.TextParser{}).TextToMetricFamilies(&buf); err != nil { t.Errorf("failed to parse Prometheus output: %v", err) diff --git a/pkg/prometheus/prometheus.go b/pkg/prometheus/prometheus.go index 4df47d5b6..7ff46a28d 100644 --- a/pkg/prometheus/prometheus.go +++ b/pkg/prometheus/prometheus.go @@ -22,6 +22,7 @@ import ( "fmt" "io" "math" + "reflect" "sort" "strings" "time" @@ -54,7 +55,7 @@ type Metric struct { } // writeHeaderTo writes the metric comment header to the given writer. -func (m *Metric) writeHeaderTo(w io.Writer, options ExportOptions) error { +func (m *Metric) writeHeaderTo(w io.Writer, options SnapshotExportOptions) error { if m.Help != "" { // Prometheus metric description escape rules: Only backslashes and line breaks need escaping. if _, err := io.WriteString(w, fmt.Sprintf("# HELP %s%s %s\n", options.ExporterPrefix, m.Name, strings.ReplaceAll(strings.ReplaceAll(m.Help, "\\", "\\\\"), "\n", "\\n"))); err != nil { @@ -231,23 +232,29 @@ type ExportOptions struct { // CommentHeader is prepended as a comment before any metric data is exported. CommentHeader string - // ExporterPrefix is prepended to all metric names. - ExporterPrefix string - - // ExtraLabels is added as labels for all metric values. - ExtraLabels map[string]string - - // MetricsWritten memoizes written metric preambles (help/type comments) by metric name. + // MetricsWritten memoizes written metric preambles (help/type comments) + // by metric name. // If specified, this map can be used to avoid duplicate preambles across multiple snapshots. // Note that this map is modified in-place during the writing process. MetricsWritten map[string]bool } -// writeMetricPreambleTo writes the metric name to w. It may also write unwritten help and type -// comments of the metric, if they haven't been written to w yet. -func (d *Data) writeMetricPreambleTo(w io.Writer, options ExportOptions) error { +// SnapshotExportOptions contains options that control how metric data is exported for an +// individual Snapshot. +type SnapshotExportOptions struct { + // ExporterPrefix is prepended to all metric names. + ExporterPrefix string + + // ExtraLabels is added as labels for all metric values. + ExtraLabels map[string]string +} + +// writeMetricPreambleTo writes the metric name to the io.Writer. It may also +// write unwritten help and type comments of the metric if they haven't been +// written to the io.Writer yet. +func (d *Data) writeMetricPreambleTo(w io.Writer, options SnapshotExportOptions, metricsWritten map[string]bool) error { // Metric header, if we haven't printed it yet. - if !options.MetricsWritten[d.Metric.Name] { + if !metricsWritten[d.Metric.Name] { // Extra newline before each preamble for aesthetic reasons. if _, err := io.WriteString(w, "\n"); err != nil { return err @@ -255,7 +262,7 @@ func (d *Data) writeMetricPreambleTo(w io.Writer, options ExportOptions) error { if err := d.Metric.writeHeaderTo(w, options); err != nil { return err } - options.MetricsWritten[d.Metric.Name] = true + metricsWritten[d.Metric.Name] = true } // Metric name. @@ -344,8 +351,8 @@ func (d *Data) writeLabelsTo(w io.Writer, extraLabels map[string]string, leLabel } // writeMetricLine writes a single line with a single number (val) to w. -func (d *Data) writeMetricLine(w io.Writer, metricSuffix string, val *Number, when time.Time, options ExportOptions, leLabel *Number) error { - if err := d.writeMetricPreambleTo(w, options); err != nil { +func (d *Data) writeMetricLine(w io.Writer, metricSuffix string, val *Number, when time.Time, options SnapshotExportOptions, leLabel *Number, metricsWritten map[string]bool) error { + if err := d.writeMetricPreambleTo(w, options, metricsWritten); err != nil { return err } if metricSuffix != "" { @@ -369,10 +376,10 @@ func (d *Data) writeMetricLine(w io.Writer, metricSuffix string, val *Number, wh } // writeTo writes the Data to the given writer in Prometheus format. -func (d *Data) writeTo(w io.Writer, when time.Time, options ExportOptions) error { +func (d *Data) writeTo(w io.Writer, when time.Time, options SnapshotExportOptions, metricsWritten map[string]bool) error { switch d.Metric.Type { case TypeUntyped, TypeGauge, TypeCounter: - return d.writeMetricLine(w, "", d.Number, when, options, nil) + return d.writeMetricLine(w, "", d.Number, when, options, nil, metricsWritten) case TypeHistogram: // Write an empty line before and after histograms to easily distinguish them from // other metric lines. @@ -384,15 +391,15 @@ func (d *Data) writeTo(w io.Writer, when time.Time, options ExportOptions) error for _, bucket := range d.HistogramValue.Buckets { numSamples += bucket.Samples samples.Int = int64(numSamples) // Prometheus distribution bucket counts are cumulative. - if err := d.writeMetricLine(w, "_bucket", &samples, when, options, &bucket.UpperBound); err != nil { + if err := d.writeMetricLine(w, "_bucket", &samples, when, options, &bucket.UpperBound, metricsWritten); err != nil { return err } } - if err := d.writeMetricLine(w, "_sum", &d.HistogramValue.Total, when, options, nil); err != nil { + if err := d.writeMetricLine(w, "_sum", &d.HistogramValue.Total, when, options, nil, metricsWritten); err != nil { return err } samples.Int = int64(numSamples) - if err := d.writeMetricLine(w, "_count", &samples, when, options, nil); err != nil { + if err := d.writeMetricLine(w, "_count", &samples, when, options, nil, metricsWritten); err != nil { return err } // Empty line after the histogram. @@ -448,11 +455,30 @@ func (w *countingWriter) Written() int { return w.written - w.w.Buffered() } -// WriteTo writes the data to the given writer, in Prometheus format. +// writeSingleMetric writes the data to the given writer in Prometheus format. // It returns the number of bytes written. -func (s *Snapshot) WriteTo(w io.Writer, options ExportOptions) (int, error) { - // Add wrapping to the buffer. Note that bufio is smart enough to not wrap buffered writers within - // buffered writers, so if the caller passes in a buffered writer, this won't double-buffer. +func (s *Snapshot) writeSingleMetric(w io.Writer, options SnapshotExportOptions, metricName string, metricsWritten map[string]bool) error { + if !strings.HasPrefix(metricName, options.ExporterPrefix) { + return nil + } + wantMetricName := strings.TrimPrefix(metricName, options.ExporterPrefix) + for _, d := range s.Data { + if d.Metric.Name != wantMetricName { + continue + } + if err := d.writeTo(w, s.When, options, metricsWritten); err != nil { + return err + } + } + return nil +} + +// Write writes one or more snapshots to the writer. +// This ensures same-name metrics across different snapshots are printed together, per spec. +func Write(w io.Writer, options ExportOptions, snapshotsToOptions map[*Snapshot]SnapshotExportOptions) (int, error) { + if len(snapshotsToOptions) == 0 { + return 0, nil + } cw := &countingWriter{w: bufio.NewWriter(w)} if options.CommentHeader != "" { for _, commentLine := range strings.Split(options.CommentHeader, "\n") { @@ -467,18 +493,53 @@ func (s *Snapshot) WriteTo(w io.Writer, options ExportOptions) (int, error) { } } } + snapshots := make([]*Snapshot, 0, len(snapshotsToOptions)) + for snapshot := range snapshotsToOptions { + snapshots = append(snapshots, snapshot) + } + switch len(snapshots) { + case 1: // Single-snapshot case. + if _, err := io.WriteString(cw, fmt.Sprintf("# Writing data from snapshot containing %d data points taken at %v.\n", len(snapshots[0].Data), snapshots[0].When)); err != nil { + return cw.Written(), err + } + default: // Multi-snapshot case. + // Provide a consistent ordering of snapshots. + sort.Slice(snapshots, func(i, j int) bool { + return reflect.ValueOf(snapshots[i]).Pointer() < reflect.ValueOf(snapshots[j]).Pointer() + }) + if _, err := io.WriteString(cw, fmt.Sprintf("# Writing data from %d snapshots:\n", len(snapshots))); err != nil { + return cw.Written(), err + } + for _, snapshot := range snapshots { + if _, err := io.WriteString(cw, fmt.Sprintf("# - Snapshot with %d data points taken at %v: %v\n", len(snapshot.Data), snapshot.When, snapshotsToOptions[snapshot].ExtraLabels)); err != nil { + return cw.Written(), err + } + } + } + if _, err := io.WriteString(cw, "\n"); err != nil { + return cw.Written(), err + } if options.MetricsWritten == nil { options.MetricsWritten = make(map[string]bool) } - if _, err := io.WriteString(cw, fmt.Sprintf("# Metric snapshot containing %d data points taken at: %v\n", len(s.Data), s.When)); err != nil { - return cw.Written(), err - } - for _, d := range s.Data { - if err := d.writeTo(cw, s.When, options); err != nil { - return cw.Written(), err + metricNamesMap := make(map[string]bool, len(options.MetricsWritten)) + metricNames := make([]string, 0, len(options.MetricsWritten)) + for _, snapshot := range snapshots { + for _, data := range snapshot.Data { + metricName := snapshotsToOptions[snapshot].ExporterPrefix + data.Metric.Name + if !metricNamesMap[metricName] { + metricNamesMap[metricName] = true + metricNames = append(metricNames, metricName) + } } } - if _, err := io.WriteString(cw, fmt.Sprintf("\n# End of metric snapshot taken at: %v\n\n", s.When)); err != nil { + sort.Strings(metricNames) + for _, metricName := range metricNames { + for _, snapshot := range snapshots { + snapshot.writeSingleMetric(cw, snapshotsToOptions[snapshot], metricName, options.MetricsWritten) + } + } + if _, err := io.WriteString(cw, "\n# End of metric data.\n"); err != nil { return cw.Written(), err } if err := cw.w.Flush(); err != nil { diff --git a/pkg/prometheus/prometheus_test.go b/pkg/prometheus/prometheus_test.go index ebc855bc4..81b8c644e 100644 --- a/pkg/prometheus/prometheus_test.go +++ b/pkg/prometheus/prometheus_test.go @@ -18,11 +18,13 @@ import ( "bytes" "errors" "fmt" + "io" "math" "strings" "sync" "testing" "time" + "unicode" v1proto "github.com/golang/protobuf/proto" "github.com/google/go-cmp/cmp" @@ -973,9 +975,12 @@ func TestSnapshotToPrometheus(t *testing.T) { // Snapshot will be rendered as Prometheus and compared against WantData. Snapshot *Snapshot - // ExportOptions dictates the options used during Snapshot rendering. + // ExportOptions dictates the options used during overall rendering. ExportOptions ExportOptions + // SnapshotExportOptions dictates the options used during Snapshot rendering. + SnapshotExportOptions SnapshotExportOptions + // WantFail, if true, indicates that the test is expected to fail when // rendering or parsing the snapshot data. WantFail bool @@ -1063,7 +1068,9 @@ func TestSnapshotToPrometheus(t *testing.T) { Name: "simple integer with export options", Snapshot: newSnapshot().Add(fooInt.int(3)), ExportOptions: ExportOptions{ - CommentHeader: "Some header", + CommentHeader: "Some header", + }, + SnapshotExportOptions: SnapshotExportOptions{ ExporterPrefix: "some_prefix_", ExtraLabels: map[string]string{ "field3": "val3a", @@ -1081,7 +1088,7 @@ func TestSnapshotToPrometheus(t *testing.T) { fooInt.fieldVal(field1, "val1a").fieldVal(field2, "val2a").int(3), fooInt.fieldVal(field2, "val2b").fieldVal(field1, "val1b").int(7), ), - ExportOptions: ExportOptions{ + SnapshotExportOptions: SnapshotExportOptions{ ExtraLabels: map[string]string{ "field3": "val3a", }, @@ -1099,7 +1106,7 @@ func TestSnapshotToPrometheus(t *testing.T) { fooInt.fieldVal(field1, "val1a").fieldVal(field2, "val2a").int(3), fooInt.fieldVal(field2, "val2b").fieldVal(field1, "val1b").int(7), ), - ExportOptions: ExportOptions{ + SnapshotExportOptions: SnapshotExportOptions{ ExtraLabels: map[string]string{ "field2": "val2c", "field3": "val3a", @@ -1192,7 +1199,9 @@ func TestSnapshotToPrometheus(t *testing.T) { fooDist.fieldVal(field1, "val1b").dist(3, 5, 3), ), ExportOptions: ExportOptions{ - CommentHeader: "Some header", + CommentHeader: "Some header", + }, + SnapshotExportOptions: SnapshotExportOptions{ ExporterPrefix: "some_prefix_", ExtraLabels: map[string]string{"field2": "val2a"}, }, @@ -1221,7 +1230,8 @@ func TestSnapshotToPrometheus(t *testing.T) { t.Run(test.Name, func(t *testing.T) { // Render and parse snapshot data. var buf bytes.Buffer - if _, err := test.Snapshot.WriteTo(&buf, test.ExportOptions); err != nil { + snapshotToOptions := map[*Snapshot]SnapshotExportOptions{test.Snapshot: test.SnapshotExportOptions} + if _, err := Write(&buf, test.ExportOptions, snapshotToOptions); err != nil { if test.WantFail { return } @@ -1241,7 +1251,7 @@ func TestSnapshotToPrometheus(t *testing.T) { // Verify that the data is consistent (i.e. verify that it's not based on random map ordering) var buf2 bytes.Buffer - if _, err := test.Snapshot.WriteTo(&buf2, test.ExportOptions); err != nil { + if _, err := Write(&buf2, test.ExportOptions, snapshotToOptions); err != nil { if test.WantFail { return } @@ -1257,7 +1267,7 @@ func TestSnapshotToPrometheus(t *testing.T) { var shortWriter shortWriter for writeLength := 0; writeLength < len(gotMetricsRaw); writeLength++ { shortWriter.Reset(writeLength) - if _, err := test.Snapshot.WriteTo(&shortWriter, test.ExportOptions); err == nil { + if _, err := Write(&shortWriter, test.ExportOptions, snapshotToOptions); err == nil { t.Fatalf("snapshot data unexpectedly succeeded being written to short writer (length %d): %v", writeLength, shortWriter.String()) } if shortWriter.size != writeLength { @@ -1333,39 +1343,38 @@ func TestSnapshotToPrometheus(t *testing.T) { } } -func MultipleSnapshotsSameWriter(t *testing.T) { +func TestWriteMultipleSnapshots(t *testing.T) { testStart := time.Now() snapshot1 := newSnapshotAt(testStart).Add(fooInt.int(3)) snapshot2 := newSnapshotAt(testStart.Add(3 * time.Minute)).Add(fooInt.int(5)) - sharedOptions := ExportOptions{ - MetricsWritten: map[string]bool{}, - } var buf bytes.Buffer - snapshot1.WriteTo(&buf, sharedOptions) - snapshot2.WriteTo(&buf, sharedOptions) + Write(&buf, ExportOptions{CommentHeader: "A header\non two lines"}, map[*Snapshot]SnapshotExportOptions{ + snapshot1: {ExporterPrefix: "export_"}, + snapshot2: {ExporterPrefix: "export_"}, + }) gotData, err := (&expfmt.TextParser{}).TextToMetricFamilies(&buf) if err != nil { t.Fatalf("cannot parse data written from snapshots: %v", err) } - if len(gotData) != 1 || gotData[fooInt.PB.GetPrometheusName()] == nil { + if len(gotData) != 1 || gotData["export_"+fooInt.PB.GetPrometheusName()] == nil { t.Fatalf("unexpected data: %v", gotData) } - got := reflectProto(gotData[fooInt.PB.GetPrometheusName()]) + got := reflectProto(gotData["export_"+fooInt.PB.GetPrometheusName()]) var wantBuf bytes.Buffer - wantBuf.WriteString(fmt.Sprintf(` - # HELP foo_int An integer about foo - # TYPE foo_int gauge - foo_int 3 %d - foo_int 4 %d + io.WriteString(&wantBuf, fmt.Sprintf(` + # HELP export_foo_int An integer about foo + # TYPE export_foo_int gauge + export_foo_int 3 %d + export_foo_int 5 %d `, testStart.UnixMilli(), testStart.Add(3*time.Minute).UnixMilli())) wantData, err := (&expfmt.TextParser{}).TextToMetricFamilies(&wantBuf) if err != nil { - t.Fatalf("cannot parse refernce data: %v", err) + t.Fatalf("cannot parse reference data: %v", err) } - if len(wantData) != 1 || wantData[fooInt.PB.GetPrometheusName()] == nil { + if len(wantData) != 1 || wantData["export_"+fooInt.PB.GetPrometheusName()] == nil { t.Fatalf("unexpected reference data: %v", gotData) } - want := reflectProto(gotData[fooInt.PB.GetPrometheusName()]) + want := reflectProto(wantData["export_"+fooInt.PB.GetPrometheusName()]) if diff := cmp.Diff(want, got, protocmp.Transform()); diff != "" { multiLineFormatter := &prototext.MarshalOptions{Multiline: true, Indent: " ", EmitUnknown: true} wantText, err := multiLineFormatter.Marshal(want) @@ -1379,3 +1388,74 @@ func MultipleSnapshotsSameWriter(t *testing.T) { t.Errorf("Snapshot data did not produce the same data as the reference data.\n\nReference data:\n\n%v\n\nSnapshot data:\n\n%v\n\nDiff:\n\n%v\n\n", string(wantText), string(gotText), diff) } } + +func TestGroupSameNameMetrics(t *testing.T) { + snapshot1 := NewSnapshot().Add( + fooCounter.int(3), + fooInt.int(3), + fooDist.dist(0, 1), + ) + snapshot2 := NewSnapshot().Add( + fooDist.dist(1, 2), + fooCounter.int(2), + ) + snapshot3 := NewSnapshot().Add( + fooDist.dist(1, 2), + fooCounter.int(2), + ) + var buf bytes.Buffer + _, err := Write(&buf, ExportOptions{}, map[*Snapshot]SnapshotExportOptions{ + snapshot1: {ExporterPrefix: "my_little_prefix_", ExtraLabels: map[string]string{"snap": "1"}}, + snapshot2: {ExporterPrefix: "my_little_prefix_", ExtraLabels: map[string]string{"snap": "2"}}, + snapshot3: {ExporterPrefix: "not_the_same_prefix_", ExtraLabels: map[string]string{"snap": "1"}}, + }) + if err != nil { + t.Fatalf("Cannot write snapshot data: %v", err) + } + rawData := buf.String() // Capture the data written. + + // Make sure the data written does parse. + // We don't use this result here because the Prometheus library is more permissive than this test. + if _, err := (&expfmt.TextParser{}).TextToMetricFamilies(&buf); err != nil { + t.Fatalf("cannot parse data written from snapshots: %v", err) + } + + // Verify that we see all metrics, and that each time we see a new one, it's one we haven't seen + // before. + seenMetrics := map[string]bool{} + var lastMetric string + for lineNumber, line := range strings.Split(rawData, "\n") { + t.Logf("Line %d: %q", lineNumber+1, line) + if strings.TrimSpace(line) == "" || strings.HasPrefix(line, "#") { + continue + } + strippedMetricName := strings.TrimLeftFunc(line, func(r rune) bool { + return unicode.IsLetter(r) || unicode.IsDigit(r) || r == '_' + }) + if len(strippedMetricName) == 0 { + t.Fatalf("invalid line: %q", line) + } + if strippedMetricName[0] != '{' && strippedMetricName[0] != ' ' { + t.Fatalf("invalid line: %q", line) + } + metricName := line[:len(line)-len(strippedMetricName)] + for _, distribSuffix := range []string{"_sum", "_count", "_bucket"} { + metricName = strings.TrimSuffix(metricName, distribSuffix) + } + if lastMetric != "" && lastMetric != metricName && seenMetrics[metricName] { + t.Fatalf("line %q: got already-seen metric name %q yet it is not the last metric (%s)", line, metricName, lastMetric) + } + lastMetric = metricName + seenMetrics[metricName] = true + } + wantSeenMetrics := map[string]bool{ + fmt.Sprintf("my_little_prefix_%s", fooCounter.PB.GetPrometheusName()): true, + fmt.Sprintf("my_little_prefix_%s", fooInt.PB.GetPrometheusName()): true, + fmt.Sprintf("my_little_prefix_%s", fooDist.PB.GetPrometheusName()): true, + fmt.Sprintf("not_the_same_prefix_%s", fooCounter.PB.GetPrometheusName()): true, + fmt.Sprintf("not_the_same_prefix_%s", fooDist.PB.GetPrometheusName()): true, + } + if !cmp.Equal(seenMetrics, wantSeenMetrics) { + t.Errorf("Seen metrics: %v\nWant metrics: %v", seenMetrics, wantSeenMetrics) + } +} diff --git a/runsc/cmd/metric_export.go b/runsc/cmd/metric_export.go index bc704d259..a58cd8563 100644 --- a/runsc/cmd/metric_export.go +++ b/runsc/cmd/metric_export.go @@ -15,7 +15,6 @@ package cmd import ( - "bufio" "context" "fmt" "os" @@ -70,21 +69,17 @@ func (m *MetricExport) Execute(ctx context.Context, f *flag.FlagSet, args ...any if err != nil { util.Fatalf("ExportMetrics failed: %v", err) } - bufWriter := bufio.NewWriter(os.Stdout) - written, err := snapshot.WriteTo(bufWriter, prometheus.ExportOptions{ - CommentHeader: fmt.Sprintf("Command-line export for sandbox %s owning container %s", cont.Sandbox.ID, id), - ExporterPrefix: conf.MetricExporterPrefix, - ExtraLabels: map[string]string{ - "sandbox": cont.Sandbox.ID, - "container": cont.ID, + written, err := prometheus.Write(os.Stdout, prometheus.ExportOptions{ + CommentHeader: fmt.Sprintf("Command-line export for sandbox %s", cont.Sandbox.ID), + }, map[*prometheus.Snapshot]prometheus.SnapshotExportOptions{ + snapshot: { + ExporterPrefix: conf.MetricExporterPrefix, + ExtraLabels: map[string]string{"sandbox": cont.Sandbox.ID}, }, }) if err != nil { util.Fatalf("Cannot write metrics to stdout: %v", err) } - if err = bufWriter.Flush(); err != nil { - util.Fatalf("Cannot flush metrics to stdout: %v", err) - } util.Infof("Wrote %d bytes of Prometheus metric data to stdout", written) return subcommands.ExitSuccess