You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by li...@apache.org on 2026/09/27 13:15:53 UTC
(skywalking-banyandb) branch fodc-proxy-metrics-oom created (now d6f0c5a87)
This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a change to branch fodc-proxy-metrics-oom
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at d6f0c5a87 Fix FODC proxy OOM kills on `/metrics` scrapes
This branch includes the following new commits:
new d6f0c5a87 Fix FODC proxy OOM kills on `/metrics` scrapes
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-banyandb) 01/01: Fix FODC proxy OOM kills on `/metrics` scrapes
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a commit to branch fodc-proxy-metrics-oom
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit d6f0c5a879fc466acd82bb39b6633288e27de020
Author: mrproliu <74...@qq.com>
AuthorDate: Sun Sep 27 21:15:41 2026 +0800
Fix FODC proxy OOM kills on `/metrics` scrapes
---
CHANGES.md | 1 +
fodc/proxy/cmd/proxy/main.go | 10 ++
fodc/proxy/internal/api/format_test.go | 170 +++++++++++++++++++
fodc/proxy/internal/api/server.go | 149 ++++++++++++-----
fodc/proxy/internal/api/server_test.go | 11 ++
fodc/proxy/internal/metrics/aggregator.go | 100 +++++++++++-
.../internal/metrics/aggregator_concurrent_test.go | 5 +
.../internal/metrics/aggregator_labels_test.go | 92 +++++++++++
.../metrics/aggregator_singleflight_test.go | 180 +++++++++++++++++++++
9 files changed, 671 insertions(+), 47 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 89ad16dd1..ab8639f94 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -43,6 +43,7 @@ Release Notes.
- Reject a measure or stream whose tag name appears in more than one tag family, or twice within one family, when it is created or updated. The query layer resolves a bare tag name against a flat map keyed by name alone, so such a schema previously registered successfully and then resolved that name differently depending on which query path read it. Already-persisted schemas keep loading unchanged; only the registry's create and update paths enforce the rule.
- Fix a silently wrong result from the vectorized measure liaison's aggregation reduce when a single per-node partial exceeds 65,536 rows: past that boundary the consuming row index wrapped, folding early rows into a group a second time and never visiting the rest, with no error. It is reachable today on `Top`-bearing queries, whose per-node limit is unbounded. The reduce now consumes an oversized partial in bounded chunks, and `BatchSize` is capped at `MaxUint16` so it cannot reopen the [...]
- Reject a `--trace-vectorized-batch-size` above 65,536 at startup instead of accepting it. The trace query path indexes a batch's active rows through a `uint16` selection, so a batch size above that bound silently wrapped -- reading early rows a second time and never reading the rest. The stream path already enforced this bound; measure now does too. A node configured above the bound will fail to start rather than return wrong results, and must lower the flag.
+- Fix FODC proxy OOM kills on `/metrics`: stream the Prometheus text to the response, stop copying label maps, share one agent collection between concurrent scrapes, and install the same panic diagnostics as the agent.
### Document
diff --git a/fodc/proxy/cmd/proxy/main.go b/fodc/proxy/cmd/proxy/main.go
index 067f3af84..25acd5329 100644
--- a/fodc/proxy/cmd/proxy/main.go
+++ b/fodc/proxy/cmd/proxy/main.go
@@ -37,6 +37,7 @@ import (
"github.com/apache/skywalking-banyandb/fodc/proxy/internal/registry"
"github.com/apache/skywalking-banyandb/pkg/config"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/panicdiag"
"github.com/apache/skywalking-banyandb/pkg/version"
)
@@ -63,6 +64,7 @@ var (
httpWriteTimeout time.Duration
heartbeatInterval time.Duration
logging logger.Logging
+ crashOutputCfg = panicdiag.NewCrashOutputConfig()
rootCmd = &cobra.Command{
Use: "fodc-proxy",
@@ -94,6 +96,7 @@ func init() {
rootCmd.Flags().DurationVar(&heartbeatInterval, "heartbeat-interval", defaultHeartbeatInterval,
"Default heartbeat interval for agents")
logger.RegisterFlags(rootCmd.Flags(), &logging)
+ crashOutputCfg.RegisterFlags(rootCmd.Flags())
}
func main() {
@@ -104,6 +107,13 @@ func main() {
}
func runProxy(cmd *cobra.Command, _ []string) error {
+ // Same crash-diagnostics setup as the agent and the BanyanDB nodes. Besides the panic
+ // artifacts it sets GOMEMLIMIT from the cgroup memory limit, so the GC starts working
+ // harder before the kernel's OOM killer does; the proxy's /metrics scrapes are short
+ // bursts of allocation that the default pacer otherwise lets run into the limit.
+ if installErr := crashOutputCfg.InstallGlobalCrashOutput(); installErr != nil {
+ return fmt.Errorf("failed to install crash output: %w", installErr)
+ }
if loadErr := config.Load("logging", cmd.Flags()); loadErr != nil {
return fmt.Errorf("failed to load logging config: %w", loadErr)
}
diff --git a/fodc/proxy/internal/api/format_test.go b/fodc/proxy/internal/api/format_test.go
new file mode 100644
index 000000000..3457228ee
--- /dev/null
+++ b/fodc/proxy/internal/api/format_test.go
@@ -0,0 +1,170 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package api
+
+import (
+ "bufio"
+ "bytes"
+ "fmt"
+ "sort"
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/fodc/proxy/internal/metrics"
+)
+
+func renderLine(t *testing.T, m *metrics.AggregatedMetric) string {
+ t.Helper()
+ var buf bytes.Buffer
+ bw := bufio.NewWriter(&buf)
+ var lw metricLineWriter
+ lw.write(bw, m)
+ require.NoError(t, bw.Flush())
+ return buf.String()
+}
+
+func TestMetricLineWriter_SortsByLabelKeyAndEscapes(t *testing.T) {
+ m := &metrics.AggregatedMetric{
+ Name: "banyandb_test",
+ Value: 1.5,
+ Labels: map[string]string{
+ "ab": "second",
+ "a": `q"uote\back` + "\nline",
+ "pod_name": "hot-0",
+ },
+ }
+ // The legacy formatter sorted the rendered `k="v"` pairs; sorting by key gives the
+ // same order because '=' sorts before every legal label-name character.
+ require.Equal(t,
+ "banyandb_test{a=\"q\\\"uote\\\\back\\nline\",ab=\"second\",pod_name=\"hot-0\"} 1.5\n",
+ renderLine(t, m))
+}
+
+func TestMetricLineWriter_NoLabels(t *testing.T) {
+ m := &metrics.AggregatedMetric{Name: "up", Value: 1}
+ require.Equal(t, "up 1\n", renderLine(t, m))
+ m.Labels = map[string]string{}
+ require.Equal(t, "up 1\n", renderLine(t, m))
+}
+
+func TestMetricLineWriter_ValueFormatting(t *testing.T) {
+ for _, tc := range []struct {
+ want string
+ value float64
+ }{
+ {"0", 0}, {"1", 1}, {"0.0001", 0.0001}, {"123456789", 123456789}, {"0.00000025", 2.5e-7},
+ } {
+ m := &metrics.AggregatedMetric{Name: "v", Value: tc.value}
+ require.Equal(t, "v "+tc.want+"\n", renderLine(t, m), "value %v", tc.value)
+ }
+}
+
+func TestMetricLineWriter_AllocatesAtMostOncePerLine(t *testing.T) {
+ m := &metrics.AggregatedMetric{Name: "banyandb_queue_sub_total_latency_bucket", Value: 42}
+ m.Labels = map[string]string{
+ "container_name": "data", "group": "sw_trace", "le": "0.005", "node_role": "data",
+ "node_type": "hot", "operation": "write", "pod_name": "demo-banyandb-data-hot-0",
+ "remote_node": "demo-banyandb-liaison-0.demo-banyandb-liaison-headless.skywalking-showcase:17912",
+ "remote_role": "liaison", "remote_tier": "hot",
+ }
+ var lw metricLineWriter
+ bw := bufio.NewWriterSize(&bytes.Buffer{}, 1<<20)
+ lw.write(bw, m) // warm the reusable buffers
+ allocs := testing.AllocsPerRun(1000, func() { lw.write(bw, m) })
+ require.LessOrEqual(t, allocs, 1.0, "one line must not allocate per label")
+}
+
+func TestWritePrometheusText_MatchesLegacyOnTypedFamilies(t *testing.T) {
+ list := []*metrics.AggregatedMetric{
+ {Name: "banyandb_up", Type: "gauge", Description: "Up.", Value: 1, Labels: map[string]string{"pod_name": "a"}},
+ {Name: "banyandb_lat_bucket", Type: "histogram", Description: "Lat.", Value: 3, Labels: map[string]string{"le": "0.1", "pod_name": "a"}},
+ {Name: "banyandb_lat_bucket", Type: "histogram", Description: "Lat.", Value: 5, Labels: map[string]string{"le": "+Inf", "pod_name": "a"}},
+ {Name: "banyandb_lat_sum", Type: "histogram", Description: "Lat.", Value: 0.4, Labels: map[string]string{"pod_name": "a"}},
+ {Name: "banyandb_lat_count", Type: "histogram", Description: "Lat.", Value: 5, Labels: map[string]string{"pod_name": "a"}},
+ {Name: "banyandb_errors_total", Type: "counter", Value: 7, Labels: map[string]string{"pod_name": "b"}},
+ {Name: "banyandb_lat_count", Value: 9, Labels: map[string]string{"pod_name": "legacy"}}, // untyped, absorbed by the typed family
+ }
+ s := &Server{}
+ require.Equal(t, legacyFormatPrometheusText(list), s.formatPrometheusText(list))
+}
+
+func TestWritePrometheusText_UnbufferedWriterIsFlushed(t *testing.T) {
+ list := []*metrics.AggregatedMetric{{Name: "banyandb_up", Type: "gauge", Value: 1}}
+ var buf bytes.Buffer
+ s := &Server{}
+ require.NoError(t, s.writePrometheusText(&buf, list))
+ require.Equal(t, "# TYPE banyandb_up gauge\nbanyandb_up 1\n", buf.String())
+}
+
+// legacyFormatPrometheusText is the pre-streaming formatter, kept verbatim as the golden
+// reference for typed families. Its untyped path iterated maps and was not deterministic,
+// so untyped coverage stays with the behavioral tests in server_test.go.
+func legacyFormatPrometheusText(aggregatedMetrics []*metrics.AggregatedMetric) string {
+ if len(aggregatedMetrics) == 0 {
+ return ""
+ }
+ var typed, untyped []*metrics.AggregatedMetric
+ for _, m := range aggregatedMetrics {
+ if m.Type != "" {
+ typed = append(typed, m)
+ } else {
+ untyped = append(untyped, m)
+ }
+ }
+ var builder strings.Builder
+ typedFamilyOrder := make([]string, 0)
+ typedFamilies := make(map[string]*metricGroup)
+ for _, m := range typed {
+ base := typedFamilyBase(m.Name, m.Type)
+ grp, exists := typedFamilies[base]
+ if !exists {
+ grp = &metricGroup{name: base, description: m.Description, metricType: m.Type}
+ typedFamilies[base] = grp
+ typedFamilyOrder = append(typedFamilyOrder, base)
+ }
+ grp.metrics = append(grp.metrics, m)
+ }
+ for _, m := range untyped {
+ if grp := matchTypedFamily(typedFamilies, m.Name); grp != nil {
+ grp.metrics = append(grp.metrics, m)
+ }
+ }
+ sort.Strings(typedFamilyOrder)
+ for _, base := range typedFamilyOrder {
+ grp := typedFamilies[base]
+ if grp.description != "" {
+ builder.WriteString(fmt.Sprintf("# HELP %s %s\n", base, grp.description))
+ }
+ builder.WriteString(fmt.Sprintf("# TYPE %s %s\n", base, grp.metricType))
+ for _, m := range grp.metrics {
+ labelParts := make([]string, 0, len(m.Labels))
+ for k, v := range m.Labels {
+ labelParts = append(labelParts, fmt.Sprintf(`%s="%s"`, k, labelValueEscaper.Replace(v)))
+ }
+ sort.Strings(labelParts)
+ labelStr := ""
+ if len(labelParts) > 0 {
+ labelStr = "{" + strings.Join(labelParts, ",") + "}"
+ }
+ builder.WriteString(fmt.Sprintf("%s%s %s\n", m.Name, labelStr, formatFloat(m.Value)))
+ }
+ }
+ return builder.String()
+}
diff --git a/fodc/proxy/internal/api/server.go b/fodc/proxy/internal/api/server.go
index 24f2823fa..e12168a6a 100644
--- a/fodc/proxy/internal/api/server.go
+++ b/fodc/proxy/internal/api/server.go
@@ -19,9 +19,11 @@
package api
import (
+ "bufio"
"context"
"encoding/json"
"fmt"
+ "io"
"net/http"
"sort"
"strconv"
@@ -175,14 +177,19 @@ func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) {
} else {
// Schema consistency is a cluster-wide, per-group verdict, not a node-scoped
// metric, so it only rides on the unfiltered scrape Grafana uses.
+ // aggregatedMetrics may be shared with a concurrent scrape; it comes back with
+ // cap == len so this append always reallocates instead of writing into it.
aggregatedMetrics = append(aggregatedMetrics, s.schemaConsistencyMetrics()...)
}
- prometheusText := s.formatPrometheusText(aggregatedMetrics)
-
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
w.WriteHeader(http.StatusOK)
- _, _ = w.Write([]byte(prometheusText))
+ bw := bufio.NewWriterSize(w, prometheusWriteBufferSize)
+ if writeErr := s.writePrometheusText(bw, aggregatedMetrics); writeErr != nil {
+ // The status line is already out; a mid-body failure means the scraper went
+ // away, which it observes as a truncated response.
+ s.logger.Debug().Err(writeErr).Msg("Failed to stream metrics response")
+ }
}
// schemaConsistencyMetrics turns the cached per-group schema consistency verdict
@@ -321,9 +328,17 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
// Metrics with a known Type (non-empty) are emitted using the real type. Metrics with
// Type=="" (pre-upgrade agents) fall back to the legacy suffix-heuristic path so behavior
// is unchanged for those metrics.
-func (s *Server) formatPrometheusText(aggregatedMetrics []*metrics.AggregatedMetric) string {
+//
+// The text is streamed to w and never materialized as a whole: grouping works on
+// pointer slices and every line is rendered into one reusable buffer, so a scrape
+// costs a few MB of allocation instead of ~12x its output size.
+func (s *Server) writePrometheusText(w io.Writer, aggregatedMetrics []*metrics.AggregatedMetric) error {
if len(aggregatedMetrics) == 0 {
- return ""
+ return nil
+ }
+ bw, ok := w.(*bufio.Writer)
+ if !ok {
+ bw = bufio.NewWriterSize(w, prometheusWriteBufferSize)
}
// Partition into typed (type known from agent) and untyped (legacy/unknown).
@@ -337,10 +352,31 @@ func (s *Server) formatPrometheusText(aggregatedMetrics []*metrics.AggregatedMet
}
}
- var builder strings.Builder
- remainingUntyped := writeTypedFamilies(&builder, typed, untyped)
- writeUntypedFamilies(&builder, remainingUntyped)
- return builder.String()
+ var lw metricLineWriter
+ remainingUntyped := writeTypedFamilies(bw, &lw, typed, untyped)
+ writeUntypedFamilies(bw, &lw, remainingUntyped)
+ return bw.Flush()
+}
+
+// prometheusWriteBufferSize is the response buffer for /metrics; 64KB keeps syscalls
+// per 9MB scrape in the low hundreds without pinning a large buffer per request.
+const prometheusWriteBufferSize = 64 << 10
+
+// writeFamilyHeader emits the "# HELP" (when a description exists) and "# TYPE" lines
+// that open one metric family.
+func writeFamilyHeader(w *bufio.Writer, name, description, metricType string) {
+ if description != "" {
+ _, _ = w.WriteString("# HELP ")
+ _, _ = w.WriteString(name)
+ _ = w.WriteByte(' ')
+ _, _ = w.WriteString(description)
+ _ = w.WriteByte('\n')
+ }
+ _, _ = w.WriteString("# TYPE ")
+ _, _ = w.WriteString(name)
+ _ = w.WriteByte(' ')
+ _, _ = w.WriteString(metricType)
+ _ = w.WriteByte('\n')
}
// writeTypedFamilies groups typed metrics by family base and emits one "# TYPE" line per
@@ -350,9 +386,9 @@ func (s *Server) formatPrometheusText(aggregatedMetrics []*metrics.AggregatedMet
// collides with a typed family are absorbed under the authoritative typed line — this
// prevents a mixed-version rollout from emitting two conflicting "# TYPE" lines for one
// name. The untyped metrics with no typed-family collision are returned for the legacy path.
-func writeTypedFamilies(builder *strings.Builder, typed, untyped []*metrics.AggregatedMetric) []*metrics.AggregatedMetric {
- typedFamilyOrder := make([]string, 0)
- typedFamilies := make(map[string]*metricGroup) // base → group
+func writeTypedFamilies(w *bufio.Writer, lw *metricLineWriter, typed, untyped []*metrics.AggregatedMetric) []*metrics.AggregatedMetric {
+ typedFamilyOrder := make([]string, 0, 256)
+ typedFamilies := make(map[string]*metricGroup, 256) // base → group
for _, m := range typed {
base := typedFamilyBase(m.Name, m.Type)
grp, exists := typedFamilies[base]
@@ -385,12 +421,9 @@ func writeTypedFamilies(builder *strings.Builder, typed, untyped []*metrics.Aggr
sort.Strings(typedFamilyOrder)
for _, base := range typedFamilyOrder {
grp := typedFamilies[base]
- if grp.description != "" {
- builder.WriteString(fmt.Sprintf("# HELP %s %s\n", base, grp.description))
- }
- builder.WriteString(fmt.Sprintf("# TYPE %s %s\n", base, grp.metricType))
+ writeFamilyHeader(w, base, grp.description, grp.metricType)
for _, m := range grp.metrics {
- builder.WriteString(formatMetricLine(m))
+ lw.write(w, m)
}
}
return remainingUntyped
@@ -399,7 +432,7 @@ func writeTypedFamilies(builder *strings.Builder, typed, untyped []*metrics.Aggr
// writeUntypedFamilies emits metrics with no known type using the legacy suffix heuristic:
// a base name with a _bucket/_sum/_count sibling is treated as a histogram, everything else
// as a gauge. This preserves pre-typed behavior for pre-upgrade agents.
-func writeUntypedFamilies(builder *strings.Builder, untyped []*metrics.AggregatedMetric) {
+func writeUntypedFamilies(w *bufio.Writer, lw *metricLineWriter, untyped []*metrics.AggregatedMetric) {
if len(untyped) == 0 {
return
}
@@ -442,12 +475,9 @@ func writeUntypedFamilies(builder *strings.Builder, untyped []*metrics.Aggregate
if len(allM) == 0 {
continue
}
- if description := allM[0].Description; description != "" {
- builder.WriteString(fmt.Sprintf("# HELP %s %s\n", baseName, description))
- }
- builder.WriteString(fmt.Sprintf("# TYPE %s histogram\n", baseName))
+ writeFamilyHeader(w, baseName, allM[0].Description, "histogram")
for _, m := range allM {
- builder.WriteString(formatMetricLine(m))
+ lw.write(w, m)
}
}
@@ -458,12 +488,9 @@ func writeUntypedFamilies(builder *strings.Builder, untyped []*metrics.Aggregate
sort.Strings(regularNames)
for _, name := range regularNames {
grp := regularMetrics[name]
- if grp.description != "" {
- builder.WriteString(fmt.Sprintf("# HELP %s %s\n", grp.name, grp.description))
- }
- builder.WriteString(fmt.Sprintf("# TYPE %s gauge\n", grp.name))
+ writeFamilyHeader(w, grp.name, grp.description, "gauge")
for _, m := range grp.metrics {
- builder.WriteString(formatMetricLine(m))
+ lw.write(w, m)
}
}
}
@@ -514,18 +541,62 @@ func typedFamilyBase(name, metricType string) string {
// applied in a single left-to-right pass, so an escaped backslash is not re-escaped.
var labelValueEscaper = strings.NewReplacer(`\`, `\\`, `"`, `\"`, "\n", `\n`)
-// formatMetricLine renders a single metric sample as a Prometheus text line.
-func formatMetricLine(m *metrics.AggregatedMetric) string {
- labelParts := make([]string, 0, len(m.Labels))
- for k, v := range m.Labels {
- labelParts = append(labelParts, fmt.Sprintf(`%s="%s"`, k, labelValueEscaper.Replace(v)))
- }
- sort.Strings(labelParts)
- labelStr := ""
- if len(labelParts) > 0 {
- labelStr = "{" + strings.Join(labelParts, ",") + "}"
+// metricLineWriter renders one Prometheus sample line per call, reusing its scratch
+// buffers so a 40k-series scrape does not allocate per line. It is not safe for
+// concurrent use; each response owns one instance.
+type metricLineWriter struct {
+ buf []byte
+ keys []string
+}
+
+// write emits `name{k="v",...} value\n`. Labels are sorted by key; that yields the
+// same order as sorting the rendered `k="v"` pairs because '=' (0x3D) sorts before
+// every legal label-name character.
+func (lw *metricLineWriter) write(w *bufio.Writer, m *metrics.AggregatedMetric) {
+ buf := lw.buf[:0]
+ buf = append(buf, m.Name...)
+ if len(m.Labels) > 0 {
+ keys := lw.keys[:0]
+ for k := range m.Labels {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ buf = append(buf, '{')
+ for i, k := range keys {
+ if i > 0 {
+ buf = append(buf, ',')
+ }
+ buf = append(buf, k...)
+ buf = append(buf, '=', '"')
+ buf = appendEscapedLabelValue(buf, m.Labels[k])
+ buf = append(buf, '"')
+ }
+ buf = append(buf, '}')
+ lw.keys = keys
+ }
+ buf = append(buf, ' ')
+ buf = strconv.AppendFloat(buf, m.Value, 'f', -1, 64)
+ buf = append(buf, '\n')
+ lw.buf = buf
+ _, _ = w.Write(buf)
+}
+
+// appendEscapedLabelValue applies the Prometheus text-format escapes for a label value
+// (backslash, double quote, line feed) in one pass without allocating.
+func appendEscapedLabelValue(buf []byte, v string) []byte {
+ for i := 0; i < len(v); i++ {
+ switch v[i] {
+ case '\\':
+ buf = append(buf, '\\', '\\')
+ case '"':
+ buf = append(buf, '\\', '"')
+ case '\n':
+ buf = append(buf, '\\', 'n')
+ default:
+ buf = append(buf, v[i])
+ }
}
- return fmt.Sprintf("%s%s %s\n", m.Name, labelStr, formatFloat(m.Value))
+ return buf
}
// formatMetricsWindowJSON formats aggregated metrics as JSON for metrics-windows endpoint.
diff --git a/fodc/proxy/internal/api/server_test.go b/fodc/proxy/internal/api/server_test.go
index f7f7843bf..692f04b1f 100644
--- a/fodc/proxy/internal/api/server_test.go
+++ b/fodc/proxy/internal/api/server_test.go
@@ -18,6 +18,7 @@
package api
import (
+ "bytes"
"context"
"encoding/json"
"fmt"
@@ -1198,3 +1199,13 @@ func TestHandleClusterLifecycle_StatusReportsUseProtoJSON(t *testing.T) {
assert.True(t, hasReportJSON, "empty report_json must still be emitted under EmitUnpopulated")
assert.Equal(t, "", second["report_json"])
}
+
+// formatPrometheusText renders through the streaming writer so the existing
+// string-comparison tests keep exercising the production code path.
+func (s *Server) formatPrometheusText(list []*metrics.AggregatedMetric) string {
+ var buf bytes.Buffer
+ if err := s.writePrometheusText(&buf, list); err != nil {
+ panic(err)
+ }
+ return buf.String()
+}
diff --git a/fodc/proxy/internal/metrics/aggregator.go b/fodc/proxy/internal/metrics/aggregator.go
index 3be6b863d..87cdc049c 100644
--- a/fodc/proxy/internal/metrics/aggregator.go
+++ b/fodc/proxy/internal/metrics/aggregator.go
@@ -20,7 +20,8 @@ package metrics
import (
"context"
- "maps"
+ "sort"
+ "strconv"
"strings"
"sync"
"time"
@@ -77,9 +78,20 @@ type Aggregator struct {
logger *logger.Logger
grpcService RequestSender
collecting map[string]map[uint64]chan []*AggregatedMetric
+ inflight map[string]*collection
nextSubID uint64
mu sync.RWMutex
collectingMu sync.RWMutex
+ inflightMu sync.Mutex
+}
+
+// collection is one in-flight round of asking every filtered agent for metrics.
+// Requests with the same filter that arrive while it runs wait on done and share
+// metrics, so overlapping scrapes cost one round of agent traffic and one copy.
+type collection struct {
+ err error
+ done chan struct{}
+ metrics []*AggregatedMetric
}
// RequestSender is an interface for sending metrics requests to agents.
@@ -94,6 +106,7 @@ func NewAggregator(registry *registry.AgentRegistry, grpcService RequestSender,
grpcService: grpcService,
logger: logger,
collecting: make(map[string]map[uint64]chan []*AggregatedMetric),
+ inflight: make(map[string]*collection),
}
}
@@ -128,21 +141,28 @@ func (ma *Aggregator) ProcessMetricsFromAgent(ctx context.Context, agentID strin
aggregatedMetrics := make([]*AggregatedMetric, 0, len(req.Metrics))
for _, metric := range req.Metrics {
- labels := make(map[string]string, len(metric.Labels))
- maps.Copy(labels, metric.Labels)
-
+ // req was decoded by this proxy's gRPC stream and is owned by it from here on, so
+ // the node-label overlay mutates metric.Labels in place instead of copying every
+ // map. Everything downstream only reads labels (see the read-only note on the
+ // fan-out below).
+ //
// Overlay the agent's node labels under a "node_" prefix so they can never collide
// with a metric-intrinsic label of the same name (e.g. the merge "type"). pod_name and
// container_name are already first-class labels and are skipped. A namespaced label
// already present (stamped per metric by the agent) is left untouched.
+ labels := metric.Labels
for key, value := range agentInfo.Labels {
if value == "" || key == podNameLabelName || key == containerNameLabelName {
continue
}
prefixed := nodeLabelPrefix + key
- if _, exists := labels[prefixed]; !exists {
- labels[prefixed] = value
+ if _, exists := labels[prefixed]; exists {
+ continue
+ }
+ if labels == nil {
+ labels = make(map[string]string, len(agentInfo.Labels))
}
+ labels[prefixed] = value
}
var timestamp time.Time
@@ -239,8 +259,72 @@ func (ma *Aggregator) unsubscribe(agentID string, subID uint64) {
}
}
-// CollectMetricsFromAgents requests metrics from all agents (or filtered agents) when external client queries.
+// collectionKey identifies the set of agents and time window a Filter selects; two
+// requests with equal keys are interchangeable and share one in-flight collection.
+func collectionKey(filter *Filter) string {
+ if filter == nil {
+ return ""
+ }
+ var b strings.Builder
+ b.WriteString(filter.Role)
+ b.WriteByte(0)
+ b.WriteString(filter.PodName)
+ b.WriteByte(0)
+ if len(filter.AgentIDs) > 0 {
+ ids := append([]string(nil), filter.AgentIDs...)
+ sort.Strings(ids)
+ b.WriteString(strings.Join(ids, ","))
+ }
+ b.WriteByte(0)
+ if filter.StartTime != nil {
+ b.WriteString(strconv.FormatInt(filter.StartTime.UnixNano(), 10))
+ }
+ b.WriteByte(0)
+ if filter.EndTime != nil {
+ b.WriteString(strconv.FormatInt(filter.EndTime.UnixNano(), 10))
+ }
+ return b.String()
+}
+
+// CollectMetricsFromAgents returns one round of metrics from the agents filter selects.
+// Concurrent callers with the same filter share a single round: the first starts it and
+// the rest wait for it, so two scrapers landing together cost one round of agent traffic
+// and hold one copy of the data. The round runs detached from the first caller's context
+// (each agent already has its own timeout), so a scraper that disconnects mid-round
+// cannot fail the others; the caller that started the round still receives its result,
+// while a caller waiting on someone else's round returns as soon as its context ends.
func (ma *Aggregator) CollectMetricsFromAgents(ctx context.Context, filter *Filter) ([]*AggregatedMetric, error) {
+ key := collectionKey(filter)
+
+ ma.inflightMu.Lock()
+ if current, ok := ma.inflight[key]; ok {
+ ma.inflightMu.Unlock()
+ select {
+ case <-current.done:
+ return current.metrics, current.err
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ }
+ }
+ current := &collection{done: make(chan struct{})}
+ ma.inflight[key] = current
+ ma.inflightMu.Unlock()
+
+ current.metrics, current.err = ma.collectOnce(context.WithoutCancel(ctx), filter)
+
+ // Unregister before signaling so a request arriving after completion starts a
+ // fresh round instead of reading this one.
+ ma.inflightMu.Lock()
+ delete(ma.inflight, key)
+ ma.inflightMu.Unlock()
+ close(current.done)
+ return current.metrics, current.err
+}
+
+// collectOnce asks every filtered agent for metrics and gathers the replies. The returned
+// slice has no spare capacity: callers append onto their own copy, never into memory
+// another waiter of the same round is reading.
+func (ma *Aggregator) collectOnce(ctx context.Context, filter *Filter) ([]*AggregatedMetric, error) {
agents := ma.getFilteredAgents(filter)
if len(agents) == 0 {
return []*AggregatedMetric{}, nil
@@ -312,7 +396,7 @@ func (ma *Aggregator) CollectMetricsFromAgents(ctx context.Context, filter *Filt
}
wg.Wait()
- return allMetrics, nil
+ return allMetrics[:len(allMetrics):len(allMetrics)], nil
}
// ActiveCollections returns the number of agents currently being collected.
diff --git a/fodc/proxy/internal/metrics/aggregator_concurrent_test.go b/fodc/proxy/internal/metrics/aggregator_concurrent_test.go
index 8070ed131..65348efb9 100644
--- a/fodc/proxy/internal/metrics/aggregator_concurrent_test.go
+++ b/fodc/proxy/internal/metrics/aggregator_concurrent_test.go
@@ -20,6 +20,7 @@ package metrics
import (
"context"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -36,9 +37,11 @@ type replyingSender struct {
reg *registry.AgentRegistry
delay time.Duration
wg sync.WaitGroup
+ requests atomic.Int64
}
func (s *replyingSender) RequestMetrics(agentID string, _ *time.Time, _ *time.Time) error {
+ s.requests.Add(1)
s.wg.Add(1)
//panicdiag:allow-rawgo test-only agent stub; a panic here must fail the test loudly rather than be recovered and hidden
go func() {
@@ -58,6 +61,8 @@ func (s *replyingSender) RequestMetrics(agentID string, _ *time.Time, _ *time.Ti
// TestCollectMetricsFromAgents_ConcurrentScrapes reproduces the production failure:
// the OTel collector scrapes /metrics every 10s and Prometheus every 30s, so their
// collections regularly overlap. Every scrape must get every agent's metrics.
+// Overlapping scrapes now share one collection (see aggregator_singleflight_test.go), so
+// this holds by construction; the test stays as the end-to-end guard.
func TestCollectMetricsFromAgents_ConcurrentScrapes(t *testing.T) {
initTestLogger(t)
aggregator, testRegistry, _ := newTestAggregator(t)
diff --git a/fodc/proxy/internal/metrics/aggregator_labels_test.go b/fodc/proxy/internal/metrics/aggregator_labels_test.go
new file mode 100644
index 000000000..218d5082b
--- /dev/null
+++ b/fodc/proxy/internal/metrics/aggregator_labels_test.go
@@ -0,0 +1,92 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package metrics
+
+import (
+ "context"
+ "reflect"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ fodcv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/fodc/v1"
+)
+
+// collectOne runs one collection for agentID in the background, pushes req through the
+// real ProcessMetricsFromAgent once the collector has subscribed, and returns the result.
+func collectOne(t *testing.T, aggregator *Aggregator, agentID string, req *fodcv1.StreamMetricsRequest) []*AggregatedMetric {
+ t.Helper()
+ agentInfo, err := aggregator.registry.GetAgentByID(agentID)
+ require.NoError(t, err)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ var got []*AggregatedMetric
+ var collectErr error
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ got, collectErr = aggregator.CollectMetricsFromAgents(ctx, &Filter{AgentIDs: []string{agentID}})
+ }()
+ require.Eventually(t, func() bool { return aggregator.ActiveCollections() == 1 }, time.Second, 5*time.Millisecond)
+ require.NoError(t, aggregator.ProcessMetricsFromAgent(ctx, agentID, agentInfo, req))
+ <-done
+ require.NoError(t, collectErr)
+ require.Len(t, got, len(req.Metrics))
+ return got
+}
+
+func TestProcessMetricsFromAgent_ReusesProtoLabelMap(t *testing.T) {
+ aggregator, testRegistry, _ := newTestAggregator(t)
+ agentID := createTestAgent(t, testRegistry, "hot-0", "data", map[string]string{"type": "hot", "pod_name": "ignored"})
+
+ now := time.Now()
+ req := createTestStreamMetricsRequest("banyandb_up", 1, map[string]string{"group": "sw_trace"}, &now)
+ protoLabels := req.Metrics[0].Labels
+
+ got := collectOne(t, aggregator, agentID, req)
+
+ require.Equal(t, reflect.ValueOf(protoLabels).Pointer(), reflect.ValueOf(got[0].Labels).Pointer(),
+ "the aggregated metric must reference the decoded proto's label map, not a copy")
+ require.Equal(t, "hot", got[0].Labels["node_type"], "node labels are overlaid in place")
+ require.NotContains(t, got[0].Labels, "node_pod_name", "pod_name is a first-class label and is never re-prefixed")
+ require.Equal(t, "sw_trace", got[0].Labels["group"])
+}
+
+func TestProcessMetricsFromAgent_NilLabelsGetsNodeLabels(t *testing.T) {
+ aggregator, testRegistry, _ := newTestAggregator(t)
+ agentID := createTestAgent(t, testRegistry, "hot-0", "data", map[string]string{"type": "hot"})
+
+ req := &fodcv1.StreamMetricsRequest{Metrics: []*fodcv1.Metric{{Name: "banyandb_up", Value: 1}}} // Labels == nil
+
+ got := collectOne(t, aggregator, agentID, req)
+
+ require.Equal(t, map[string]string{"node_type": "hot"}, got[0].Labels)
+}
+
+func TestProcessMetricsFromAgent_NilLabelsWithNothingToOverlayStaysNil(t *testing.T) {
+ aggregator, testRegistry, _ := newTestAggregator(t)
+ agentID := createTestAgent(t, testRegistry, "hot-0", "data", nil)
+
+ req := &fodcv1.StreamMetricsRequest{Metrics: []*fodcv1.Metric{{Name: "banyandb_up", Value: 1}}}
+
+ got := collectOne(t, aggregator, agentID, req)
+
+ require.Nil(t, got[0].Labels, "no map is allocated when there is nothing to overlay")
+}
diff --git a/fodc/proxy/internal/metrics/aggregator_singleflight_test.go b/fodc/proxy/internal/metrics/aggregator_singleflight_test.go
new file mode 100644
index 000000000..be7f0fc14
--- /dev/null
+++ b/fodc/proxy/internal/metrics/aggregator_singleflight_test.go
@@ -0,0 +1,180 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package metrics
+
+import (
+ "context"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func newReplyingSetup(t *testing.T, agents int, delay time.Duration) (*Aggregator, *replyingSender) {
+ t.Helper()
+ aggregator, testRegistry, _ := newTestAggregator(t)
+ sender := &replyingSender{aggregator: aggregator, reg: testRegistry, delay: delay}
+ aggregator.SetGRPCService(sender)
+ for i := 0; i < agents; i++ {
+ createTestAgent(t, testRegistry, "pod", "datanode-warm", nil)
+ }
+ return aggregator, sender
+}
+
+// In production two scrapers (the OTel collector every 10s and Prometheus every 30s) hit
+// /metrics within tens of milliseconds of each other every 30s. One collection must serve
+// both: every agent is asked once and both scrapers read the same result.
+func TestCollectMetricsFromAgents_OverlappingScrapesShareOneCollection(t *testing.T) {
+ const agents = 3
+ aggregator, sender := newReplyingSetup(t, agents, 200*time.Millisecond)
+
+ results := make([][]*AggregatedMetric, 2)
+ errs := make([]error, 2)
+ var wg sync.WaitGroup
+ for i := 0; i < 2; i++ {
+ wg.Add(1)
+ //panicdiag:allow-rawgo test-only scrape driver; a panic here must fail the test loudly rather than be recovered and hidden
+ go func(idx int) {
+ defer wg.Done()
+ time.Sleep(time.Duration(idx) * 50 * time.Millisecond)
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
+ results[idx], errs[idx] = aggregator.GetLatestMetrics(ctx, nil)
+ }(i)
+ }
+ wg.Wait()
+ sender.wg.Wait()
+
+ for i := 0; i < 2; i++ {
+ require.NoError(t, errs[i])
+ require.Len(t, results[i], agents)
+ }
+ require.EqualValues(t, agents, sender.requests.Load(), "each agent is asked once, not once per scraper")
+ require.Same(t, results[0][0], results[1][0], "both scrapers read the same collection")
+ require.Equal(t, len(results[0]), cap(results[0]), "a shared result must not carry spare capacity an append could write into")
+ require.Equal(t, 0, aggregator.ActiveCollections())
+}
+
+func TestCollectMetricsFromAgents_SequentialScrapesCollectSeparately(t *testing.T) {
+ const agents = 2
+ aggregator, sender := newReplyingSetup(t, agents, 10*time.Millisecond)
+ for i := 0; i < 2; i++ {
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ got, err := aggregator.GetLatestMetrics(ctx, nil)
+ cancel()
+ require.NoError(t, err)
+ require.Len(t, got, agents)
+ }
+ sender.wg.Wait()
+ require.EqualValues(t, 2*agents, sender.requests.Load(), "a scrape that starts after the previous one finished collects afresh")
+}
+
+func TestCollectMetricsFromAgents_DifferentFiltersAreNotCoalesced(t *testing.T) {
+ aggregator, testRegistry, _ := newTestAggregator(t)
+ sender := &replyingSender{aggregator: aggregator, reg: testRegistry, delay: 200 * time.Millisecond}
+ aggregator.SetGRPCService(sender)
+ createTestAgent(t, testRegistry, "hot-0", "data", nil)
+ createTestAgent(t, testRegistry, "liaison-0", "liaison", nil)
+
+ var wg sync.WaitGroup
+ got := make([][]*AggregatedMetric, 2)
+ for i, role := range []string{"data", "liaison"} {
+ wg.Add(1)
+ //panicdiag:allow-rawgo test-only scrape driver; a panic here must fail the test loudly rather than be recovered and hidden
+ go func(idx int, role string) {
+ defer wg.Done()
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
+ got[idx], _ = aggregator.GetLatestMetrics(ctx, &Filter{Role: role})
+ }(i, role)
+ }
+ wg.Wait()
+ sender.wg.Wait()
+ require.Len(t, got[0], 1)
+ require.Len(t, got[1], 1)
+ require.EqualValues(t, 2, sender.requests.Load(), "one collection per distinct filter")
+}
+
+func TestCollectMetricsFromAgents_FollowerCancelDoesNotAffectLeader(t *testing.T) {
+ const agents = 2
+ aggregator, sender := newReplyingSetup(t, agents, 300*time.Millisecond)
+
+ leaderCtx, leaderCancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer leaderCancel()
+ var leaderGot []*AggregatedMetric
+ var leaderErr error
+ var wg sync.WaitGroup
+ wg.Add(1)
+ //panicdiag:allow-rawgo test-only scrape driver; a panic here must fail the test loudly rather than be recovered and hidden
+ go func() {
+ defer wg.Done()
+ leaderGot, leaderErr = aggregator.GetLatestMetrics(leaderCtx, nil)
+ }()
+ require.Eventually(t, func() bool { return aggregator.ActiveCollections() == agents }, time.Second, 5*time.Millisecond)
+
+ followerCtx, followerCancel := context.WithCancel(context.Background())
+ time.AfterFunc(50*time.Millisecond, followerCancel)
+ start := time.Now()
+ _, followerErr := aggregator.GetLatestMetrics(followerCtx, nil)
+ require.ErrorIs(t, followerErr, context.Canceled)
+ require.Less(t, time.Since(start), 250*time.Millisecond, "a canceled follower returns at once instead of waiting for the round")
+
+ wg.Wait()
+ sender.wg.Wait()
+ require.NoError(t, leaderErr)
+ require.Len(t, leaderGot, agents)
+}
+
+func TestCollectMetricsFromAgents_LeaderCancelStillServesFollower(t *testing.T) {
+ const agents = 2
+ aggregator, sender := newReplyingSetup(t, agents, 300*time.Millisecond)
+
+ leaderCtx, leaderCancel := context.WithCancel(context.Background())
+ var leaderGot []*AggregatedMetric
+ var leaderErr error
+ var wg sync.WaitGroup
+ wg.Add(1)
+ //panicdiag:allow-rawgo test-only scrape driver; a panic here must fail the test loudly rather than be recovered and hidden
+ go func() {
+ defer wg.Done()
+ leaderGot, leaderErr = aggregator.GetLatestMetrics(leaderCtx, nil)
+ }()
+ require.Eventually(t, func() bool { return aggregator.ActiveCollections() == agents }, time.Second, 5*time.Millisecond)
+
+ followerCtx, followerCancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer followerCancel()
+ var followerGot []*AggregatedMetric
+ var followerErr error
+ wg.Add(1)
+ //panicdiag:allow-rawgo test-only scrape driver; a panic here must fail the test loudly rather than be recovered and hidden
+ go func() {
+ defer wg.Done()
+ followerGot, followerErr = aggregator.GetLatestMetrics(followerCtx, nil)
+ }()
+ time.Sleep(50 * time.Millisecond)
+ leaderCancel() // the scraper that started the round goes away mid-collection
+
+ wg.Wait()
+ sender.wg.Wait()
+ require.NoError(t, leaderErr, "the round is not aborted by its starter's cancellation")
+ require.Len(t, leaderGot, agents)
+ require.NoError(t, followerErr, "the round the leader started must still complete for everyone else")
+ require.Len(t, followerGot, agents)
+ require.EqualValues(t, agents, sender.requests.Load())
+}