From afde7aea6bb128fe890d526ab2a6d18d3be16031 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Wed, 15 Jul 2026 09:52:08 -0700 Subject: [PATCH] Harden delta sink counter tracking Preserve metric batch metadata, disambiguate counter series, and protect counter state during concurrent sends without changing Retry queueing behavior. Co-authored-by: Codex Ai-assisted: true --- sink/delta.go | 74 +++++++++++++--------- sink/delta_test.go | 152 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 196 insertions(+), 30 deletions(-) diff --git a/sink/delta.go b/sink/delta.go index 32469459..2df49d3a 100644 --- a/sink/delta.go +++ b/sink/delta.go @@ -5,7 +5,9 @@ package sink import ( "context" "sort" + "strconv" "strings" + "sync" "github.com/cashapp/blip" ) @@ -20,8 +22,16 @@ import ( // can cause incorrect metric values to be submitted then delta calculations are performed. // The Delta sink should never be wrapped inside of a Retry sink to prevent this. type Delta struct { - sink blip.Sink - counters map[string]float64 // holds last value of the counter so deltas can be calculated + sink blip.Sink + mu sync.Mutex + + counters map[metricID]float64 // holds last value of the counter so deltas can be calculated +} + +type metricID struct { + domain string + name string + group string } var _ blip.Sink = &Delta{} @@ -36,7 +46,7 @@ func NewDelta(sink blip.Sink) *Delta { return &Delta{ sink: sink, - counters: make(map[string]float64), + counters: make(map[metricID]float64), } } @@ -51,6 +61,16 @@ func (d *Delta) Name() string { // // This is safe to call from multiple goroutines. func (d *Delta) Send(ctx context.Context, metrics *blip.Metrics) error { + return d.sink.Send(ctx, d.transform(metrics)) +} + +func (d *Delta) transform(metrics *blip.Metrics) *blip.Metrics { + // Protect counter state and transformation, but do not serialize the + // wrapped sink: Retry depends on accepting a new batch while an earlier + // batch is still in flight. + d.mu.Lock() + defer d.mu.Unlock() + newValues := make(map[string][]blip.MetricValue) hasNewValues := false @@ -64,18 +84,18 @@ func (d *Delta) Send(ctx context.Context, metrics *blip.Metrics) error { case blip.CUMULATIVE_COUNTER: hasDelta = true metricValue := value.Value - metricId := d.metricID(value.Name, value.Group) + id := d.metricID(key, value.Name, value.Group) - val, ok := d.counters[metricId] + val, ok := d.counters[id] if !ok { // If we don't have a prior data point then we should // not calculate a delta and just remove the point. - d.counters[metricId] = value.Value + d.counters[id] = value.Value continue } delta := value.Value - val - d.counters[metricId] = value.Value + d.counters[id] = value.Value if delta >= 0 { metricValue = delta } else { @@ -104,25 +124,18 @@ func (d *Delta) Send(ctx context.Context, metrics *blip.Metrics) error { if !hasNewValues { // If we didn't have to calculate any deltas then we should - // just submit the original metrics - return d.sink.Send(ctx, metrics) + // just return the original metrics. + return metrics } - return d.sink.Send(ctx, &blip.Metrics{ - Begin: metrics.Begin, - End: metrics.End, - MonitorId: metrics.MonitorId, - Plan: metrics.Plan, - Level: metrics.Level, - State: metrics.State, - Values: newValues, - }) + transformed := *metrics + transformed.Values = newValues + return &transformed } -// metricID returns the metric name concatenated with sorted group keys. -// For example, if the metric name is "foo" and the group keys are "a" and "b", -// it returns "fooab". This is used to calculate delta counter values in Send. -func (s *Delta) metricID(name string, groups map[string]string) string { +// metricID identifies one cumulative counter series. Group entries are sorted +// and length-prefixed so keys, values, and their boundaries are unambiguous. +func (d *Delta) metricID(domain, name string, groups map[string]string) metricID { keys := make([]string, 0, len(groups)) for k := range groups { keys = append(keys, k) @@ -131,15 +144,16 @@ func (s *Delta) metricID(name string, groups map[string]string) string { // sort by keys sort.Strings(keys) - var values []string - // collect values by sorted keys + var group strings.Builder for _, k := range keys { - values = append(values, groups[k]) + value := groups[k] + group.WriteString(strconv.Itoa(len(k))) + group.WriteByte(':') + group.WriteString(k) + group.WriteString(strconv.Itoa(len(value))) + group.WriteByte(':') + group.WriteString(value) } - var key string - key += name - key += strings.Join(values, "") - - return key + return metricID{domain: domain, name: name, group: group.String()} } diff --git a/sink/delta_test.go b/sink/delta_test.go index 3316f890..bfb50022 100644 --- a/sink/delta_test.go +++ b/sink/delta_test.go @@ -2,6 +2,8 @@ package sink import ( "context" + "fmt" + "sync" "testing" "time" @@ -67,6 +69,7 @@ func TestDeltaSink(t *testing.T) { MonitorId: "testmonitor", Plan: "testplan", Level: "testlevel", + Interval: 42, State: "teststate", Values: map[string][]blip.MetricValue{ "nochanges": noChangesValues, @@ -95,6 +98,7 @@ func TestDeltaSink(t *testing.T) { MonitorId: "testmonitor", Plan: "testplan", Level: "testlevel", + Interval: metrics.Interval, State: "teststate", Values: map[string][]blip.MetricValue{ "nochanges": noChangesValues, @@ -125,6 +129,7 @@ func TestDeltaSink(t *testing.T) { MonitorId: "testmonitor", Plan: "testplan", Level: "testlevel", + Interval: 43, State: "teststate", Values: map[string][]blip.MetricValue{ "nochanges": noChangesValues, @@ -148,6 +153,7 @@ func TestDeltaSink(t *testing.T) { MonitorId: "testmonitor", Plan: "testplan", Level: "testlevel", + Interval: metrics.Interval, State: "teststate", Values: map[string][]blip.MetricValue{ "nochanges": noChangesValues, @@ -182,6 +188,152 @@ func TestDeltaSink(t *testing.T) { } } +func TestDeltaSinkSeriesIdentity(t *testing.T) { + tests := []struct { + name string + values map[string][]blip.MetricValue + }{ + { + name: "domain", + values: map[string][]blip.MetricValue{ + "domain-1": {{Name: "counter", Value: 10, Type: blip.CUMULATIVE_COUNTER}}, + "domain-2": {{Name: "counter", Value: 20, Type: blip.CUMULATIVE_COUNTER}}, + }, + }, + { + name: "group keys", + values: map[string][]blip.MetricValue{ + "domain": { + {Name: "counter", Value: 10, Type: blip.CUMULATIVE_COUNTER, Group: map[string]string{"a": "1"}}, + {Name: "counter", Value: 20, Type: blip.CUMULATIVE_COUNTER, Group: map[string]string{"b": "1"}}, + }, + }, + }, + { + name: "group value boundaries", + values: map[string][]blip.MetricValue{ + "domain": { + {Name: "counter", Value: 10, Type: blip.CUMULATIVE_COUNTER, Group: map[string]string{"a": "12", "b": "3"}}, + {Name: "counter", Value: 20, Type: blip.CUMULATIVE_COUNTER, Group: map[string]string{"a": "1", "b": "23"}}, + }, + }, + }, + { + name: "metric name boundary", + values: map[string][]blip.MetricValue{ + "domain": { + {Name: "counter-1", Value: 10, Type: blip.CUMULATIVE_COUNTER, Group: map[string]string{"a": "23"}}, + {Name: "counter-12", Value: 20, Type: blip.CUMULATIVE_COUNTER, Group: map[string]string{"a": "3"}}, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var got *blip.Metrics + deltaSink := NewDelta(&mock.Sink{SendFunc: func(_ context.Context, m *blip.Metrics) error { + got = m + return nil + }}) + + if err := deltaSink.Send(context.Background(), &blip.Metrics{Values: tt.values}); err != nil { + t.Fatal(err) + } + + for domain, values := range got.Values { + if len(values) != 0 { + t.Errorf("first sample for %s emitted %d values; want none: %+v", domain, len(values), values) + } + } + }) + } +} + +func TestDeltaSinkConcurrentSend(t *testing.T) { + const sends = 100 + + var ( + mu sync.Mutex + received = make(map[string]float64) + ) + deltaSink := NewDelta(&mock.Sink{SendFunc: func(_ context.Context, m *blip.Metrics) error { + mu.Lock() + defer mu.Unlock() + for _, values := range m.Values { + for _, value := range values { + received[value.Group["id"]] = value.Value + } + } + return nil + }}) + + sendAll := func(value float64) { + t.Helper() + var wg sync.WaitGroup + for i := 0; i < sends; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + id := fmt.Sprintf("counter-%d", i) + err := deltaSink.Send(context.Background(), &blip.Metrics{Values: map[string][]blip.MetricValue{ + "domain": {{ + Name: "counter", + Value: value, + Type: blip.CUMULATIVE_COUNTER, + Group: map[string]string{"id": id}, + }}, + }}) + if err != nil { + t.Errorf("send %s: %v", id, err) + } + }(i) + } + wg.Wait() + } + + sendAll(10) + sendAll(11) + + mu.Lock() + defer mu.Unlock() + if len(received) != sends { + t.Fatalf("received %d deltas; want %d", len(received), sends) + } + for id, value := range received { + if value != 1 { + t.Errorf("%s delta = %v; want 1", id, value) + } + } +} + +func TestDeltaSinkCounterReset(t *testing.T) { + var got *blip.Metrics + deltaSink := NewDelta(&mock.Sink{SendFunc: func(_ context.Context, m *blip.Metrics) error { + got = m + return nil + }}) + + send := func(value float64) { + t.Helper() + if err := deltaSink.Send(context.Background(), &blip.Metrics{Values: map[string][]blip.MetricValue{ + "domain": {{Name: "counter", Value: value, Type: blip.CUMULATIVE_COUNTER}}, + }}); err != nil { + t.Fatal(err) + } + } + + send(100) + send(5) + if value := got.Values["domain"][0].Value; value != 5 { + t.Fatalf("reset delta = %v; want current partial value 5", value) + } + send(8) + if value := got.Values["domain"][0].Value; value != 3 { + t.Fatalf("post-reset delta = %v; want 3", value) + } +} + func TestDeltaSink_Passthrough(t *testing.T) { var returnedResults *blip.Metrics mockSink := &mock.Sink{