Copilot commented on code in PR #1206: URL: https://github.com/apache/skywalking-banyandb/pull/1206#discussion_r3535210540
########## banyand/trace/plugin_telemetry.go: ########## @@ -0,0 +1,712 @@ +// 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 trace + +import ( + "fmt" + "slices" + "strings" + "sync" + "time" + "unicode" + + "github.com/rs/zerolog" + + "github.com/apache/skywalking-banyandb/banyand/observability" + "github.com/apache/skywalking-banyandb/pkg/logger" + "github.com/apache/skywalking-banyandb/pkg/meter" + "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk" +) + +const ( + // maxSeriesPerInstrument is the cardinality cap for custom label-value tuples + // per instrument instance. + maxSeriesPerInstrument = 100 + // maxMetricNameLen is the maximum sanitized metric name length. + maxMetricNameLen = 64 + // logRatePerSec is the sustained token-bucket refill rate (tokens/second). + logRatePerSec = 50.0 + // logBurstSize is the maximum burst capacity of the log token bucket. + logBurstSize = 100.0 + // maxMsgLen is the maximum log message length in bytes. + maxMsgLen = 1024 + // maxLogFields is the maximum number of structured key/value pairs emitted. + maxLogFields = 16 + // suppressWarnInterval is the minimum interval between rate-limiter summary WARNs. + suppressWarnInterval = 5 * time.Second + // overflowSentinel is the fixed label value substituted when cardinality is exceeded. + overflowSentinel = "__overflow__" +) + +// reservedLabelSet contains label names plugins may not declare; the host +// always prepends group and plugin_name itself. +var reservedLabelSet = map[string]struct{}{ + "group": {}, + "plugin_name": {}, +} + +// noopCounter is a zero-allocation sdk.Counter for the bypass path. +type noopCounter struct{} + +func (noopCounter) Inc(_ float64, _ ...string) {} + +// noopGauge is a zero-allocation sdk.Gauge for the bypass path. +type noopGauge struct{} + +func (noopGauge) Set(_ float64, _ ...string) {} +func (noopGauge) Add(_ float64, _ ...string) {} + +// noopHistogram is a zero-allocation sdk.Histogram for the bypass path. +type noopHistogram struct{} + +func (noopHistogram) Observe(_ float64, _ ...string) {} + +// noopMeter is a zero-allocation sdk.Meter for the bypass path. +type noopMeter struct{} + +func (noopMeter) Counter(_ string, _ ...string) sdk.Counter { return noopCounter{} } +func (noopMeter) Gauge(_ string, _ ...string) sdk.Gauge { return noopGauge{} } +func (noopMeter) Histogram(_ string, _ []float64, _ ...string) sdk.Histogram { return noopHistogram{} } + +// instrumentKey identifies a registered instrument in the cache by its full +// metric name and joined label names. +type instrumentKey struct { + fullName string + labelNames string // "\x00"-joined +} + +// boundInstrumentKind distinguishes the three instrument kinds. +type boundInstrumentKind uint8 + +const ( + kindCounter boundInstrumentKind = iota + kindGauge boundInstrumentKind = iota + kindHistogram boundInstrumentKind = iota +) Review Comment: `kindGauge` and `kindHistogram` are both assigned `iota` explicitly, so all three constants evaluate to the same value. This breaks kind discrimination (e.g., teardown’s switch will treat every cached instrument as the same kind). Define only the first constant with `= iota` and let subsequent constants auto-increment, or assign distinct explicit values. ########## banyand/trace/metadata.go: ########## @@ -333,30 +340,79 @@ func (sr *schemaRepo) reconcilePipeline(group string, cfg *commonv1.TracePipelin if cfg == nil || !cfg.GetEnabled() || !mergeEventEnabled(cfg) { hadSamplers := len(lookupSamplers(group)) > 0 removeSamplersForGroup(group) + teardownGroupTelemetry(group) setMergeGraceForGroup(group, 0) sr.samplerMeter.setActiveCount(group, 0) if hadSamplers { sr.samplerMeter.incRemoveTotal(group) } return } + // Idempotent skip: if the desired identity list (name+configHash in order) equals + // the currently registered one, this is a redundant watch-replay re-apply. Return + // early with no reload, no UseHost, no teardown, no swap, and no metrics churn. + // This is defense-in-depth for the same-group re-reconcile race on top of the + // once-only-UseHost invariant enforced inside loadSamplerPlugin. + desired := make([]nameHash, 0, len(cfg.GetPlugins())) + for _, p := range cfg.GetPlugins() { + sp := p.GetSampler() + if sp == nil { + continue + } + desired = append(desired, nameHash{name: p.GetName(), configHash: computeConfigHash(sp)}) + } + current := currentSamplerIdentity(group) + if len(desired) == len(current) { + identical := true + for idx := range desired { + if desired[idx] != current[idx] { + identical = false + break + } + } + if identical { + return + } + } Review Comment: The idempotent-skip identity only compares `(plugin name, configHash)` and ignores `SamplerPlugin.Path`, `SamplerPlugin.Symbol`, and `AbiVersion`. If any of those change while config JSON stays the same, the reconcile can incorrectly skip and keep using the previously loaded plugin. Include path/symbol (and ABI version) in the identity used for the skip check (and store them in the registry alongside `configHash`) so changes in the actual plugin reference trigger a rebuild. ########## test/plugins/telemetry_smoke_test.go: ########## @@ -0,0 +1,171 @@ +// 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. + +//go:build trace_pipeline Review Comment: This file’s header comment states it “must NOT be run with -race”, but the build constraint doesn’t enforce that. Consider updating the build tag to exclude race builds (e.g., `trace_pipeline && !race`) so an accidental `go test -tags trace_pipeline -race ./...` doesn’t attempt to run it and fail in confusing ways. ########## banyand/trace/plugin_telemetry_test.go: ########## @@ -0,0 +1,369 @@ +// 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 trace + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/apache/skywalking-banyandb/pkg/logger" + "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk" +) + +// newTestSamplerMetrics builds a samplerMetrics backed by the fakeMetricFactory. +func newTestSamplerMetrics() (*samplerMetrics, *fakeMetricFactory) { + f := newFakeMetricFactory() + sm := newSamplerMetrics(f) + return sm, f +} + +// TestPluginTelemetry_MetricNamePrefix verifies that plugin metric names are +// always prefixed with "plugin_" regardless of what the plugin requests. +func TestPluginTelemetry_MetricNamePrefix(t *testing.T) { + sm, f := newTestSamplerMetrics() + baseLogger := logger.GetLogger("trace") + pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug") + m := pt.Meter() + require.NotNil(t, m) + + _ = m.Counter("my_counter") + _ = m.Gauge("my_gauge") + + // The factory should have received names prefixed with "plugin_". + f.mu.Lock() + _, hasCounter := f.counters["plugin_my_counter"] + _, hasGauge := f.gauges["plugin_my_gauge"] + f.mu.Unlock() + + assert.True(t, hasCounter, "counter name must be prefixed with plugin_") + assert.True(t, hasGauge, "gauge name must be prefixed with plugin_") +} + +// TestPluginTelemetry_MetricNameSanitization verifies that mixed-case and +// special-character names are lowercased and sanitized. +func TestPluginTelemetry_MetricNameSanitization(t *testing.T) { + sm, f := newTestSamplerMetrics() + baseLogger := logger.GetLogger("trace") + pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug") + m := pt.Meter() + + _ = m.Counter("MyCounter-V2") + + f.mu.Lock() + _, ok := f.counters["plugin_mycounter_v2"] + f.mu.Unlock() + assert.True(t, ok, "sanitized counter name must be plugin_mycounter_v2") +} + +// TestPluginTelemetry_ReservedLabelRejected verifies that reserved label names +// {group, plugin_name} are silently removed and series_rejected incremented. +func TestPluginTelemetry_ReservedLabelRejected(t *testing.T) { + sm, f := newTestSamplerMetrics() + baseLogger := logger.GetLogger("trace") + pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug") + m := pt.Meter() + + // Both "group" and "plugin_name" are reserved — they should be stripped. + c := m.Counter("hits", "group", "plugin_name", "env") + require.NotNil(t, c) + + rejectedCounter := f.counter("plugin_telemetry_series_rejected_total") + require.NotNil(t, rejectedCounter) + // Two reserved labels → two increments. + assert.GreaterOrEqual(t, len(rejectedCounter.calls), 2, + "two reserved label names must each increment series_rejected") +} + +// TestPluginTelemetry_CardinalityCap verifies that emitting more than +// maxSeriesPerInstrument distinct custom-value tuples routes extras to the +// overflow sentinel and increments series_rejected for each excess tuple. +func TestPluginTelemetry_CardinalityCap(t *testing.T) { + sm, f := newTestSamplerMetrics() + baseLogger := logger.GetLogger("trace") + pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug") + m := pt.Meter() + + c := m.Counter("reqs", "svc") + require.NotNil(t, c) + + // Emit exactly maxSeriesPerInstrument distinct tuples. + for i := range maxSeriesPerInstrument { + c.Inc(1, strings.Repeat("x", i+1)) + } + + rejBefore := len(f.counter("plugin_telemetry_series_rejected_total").calls) + + // Emit K extra distinct tuples — each should be rejected. + const extraK = 5 + for i := range extraK { + c.Inc(1, "extra_"+strings.Repeat("y", i+1)) + } + + rejCounter := f.counter("plugin_telemetry_series_rejected_total") + rejAfter := len(rejCounter.calls) + assert.Equal(t, extraK, rejAfter-rejBefore, + "each excess distinct tuple must increment series_rejected exactly once") + + // Verify the underlying counter received the overflow sentinel label values. + underlying := f.counter("plugin_reqs") + require.NotNil(t, underlying) + overflowCount := underlying.callsWithLabels("grp", "plug", overflowSentinel) + assert.Equal(t, extraK, overflowCount, "overflow tuples must use the sentinel value") +} + +// TestPluginTelemetry_BypassZeroAllocs verifies that Inc/Set/Observe on a +// bypass-backed adapter perform zero heap allocations for no-label invocations. +// The BenchmarkPluginTelemetry_BypassInc benchmark covers the labeled case; +// the Go compiler is unable to avoid a []string backing allocation when a +// variadic call crosses an interface boundary in a test context, but the +// adapter itself builds no internal slices. +func TestPluginTelemetry_BypassZeroAllocs(t *testing.T) { + sm, _ := newTestSamplerMetrics() + baseLogger := logger.GetLogger("trace") + // nil factory → bypass path. + pt := newPluginTelemetry(nil, baseLogger, sm, "grp", "plug") + m := pt.Meter() + + c := m.Counter("hits") + g := m.Gauge("lat") + h := m.Histogram("size", []float64{1, 5, 10}) + + allocs := testing.AllocsPerRun(100, func() { + c.Inc(1) + g.Set(3.14) + h.Observe(7.0) + }) + assert.Equal(t, float64(0), allocs, "bypass path must have 0 allocs/op on no-label calls") +} + +// TestPluginTelemetry_LoggerInfoWarnError verifies that Info/Warn/Error calls +// go through the rate limiter without panicking. +func TestPluginTelemetry_LoggerInfoWarnError(t *testing.T) { + sm, _ := newTestSamplerMetrics() + baseLogger := logger.GetLogger("trace") + pt := newPluginTelemetry(nil, baseLogger, sm, "grp", "plug") + l := pt.Logger() + + require.NotPanics(t, func() { + l.Info("hello", "k1", "v1") + l.Warn("warn msg", "k2", 42) + l.Error("err msg") + }) +} + +// TestPluginTelemetry_LoggerDebugDropped verifies that Debug calls are silently +// dropped under the INFO level cap. +func TestPluginTelemetry_LoggerDebugDropped(t *testing.T) { + sm, _ := newTestSamplerMetrics() + baseLogger := logger.GetLogger("trace") + pt := newPluginTelemetry(nil, baseLogger, sm, "grp", "plug") + l := pt.Logger() + + // This must not panic and must not count against the rate limiter. + require.NotPanics(t, func() { + for range 200 { + l.Debug("debug msg", "k", "v") + } + }) + // Rate limiter tokens must be unaffected by Debug calls. + pl := pt.loggerAdapter + pl.mu.Lock() + tok := pl.tokens + pl.mu.Unlock() + assert.Equal(t, logBurstSize, tok, "Debug must not consume rate-limiter tokens") +} + +// TestPluginTelemetry_LogRateLimit verifies that emitting many lines quickly +// drops some and increments plugin_log_dropped_total. +func TestPluginTelemetry_LogRateLimit(t *testing.T) { + sm, f := newTestSamplerMetrics() + baseLogger := logger.GetLogger("trace") + pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug") + l := pt.Logger() + + const burst = 200 + for range burst { + l.Info("rapid fire") + } + + droppedCounter := f.counter("plugin_log_dropped_total") + require.NotNil(t, droppedCounter) + // We emitted 200 lines but burst capacity is 100 — at least some must drop. + assert.Greater(t, len(droppedCounter.calls), 0, + "emitting beyond burst capacity must increment plugin_log_dropped_total") +} + +// TestPluginTelemetry_LogMsgTruncated verifies that messages longer than +// maxMsgLen are truncated before logging. +func TestPluginTelemetry_LogMsgTruncated(t *testing.T) { + // Build a pluginLogger directly and test emitEvent via a no-op event. + longMsg := strings.Repeat("A", maxMsgLen+100) + // Truncation happens inside emitEvent; verify the helper directly. + truncated := longMsg + if len(truncated) > maxMsgLen { + truncated = truncated[:maxMsgLen] + } + assert.Equal(t, maxMsgLen, len(truncated), "truncated message must be exactly maxMsgLen bytes") +} Review Comment: This test currently truncates the string locally and never calls the production truncation path (`emitEvent` / logger adapter), so it doesn’t validate the behavior it describes. Update it to exercise the real implementation (e.g., by calling `emitEvent` with a real `zerolog.Event` wired to a buffer and asserting the emitted message length/content). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
