Skip to content

Harden delta sink counter tracking #167

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Areson merged 1 commit into from
Jul 15, 2026
Merged

Harden delta sink counter tracking #167

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 44 additions & 30 deletions sink/delta.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,9 @@ package sink
import (
"context"
"sort"
"strconv"
"strings"
"sync"

"github.com/cashapp/blip"
)
Expand All @@ -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{}
Expand All @@ -36,7 +46,7 @@ func NewDelta(sink blip.Sink) *Delta {

return &Delta{
sink: sink,
counters: make(map[string]float64),
counters: make(map[metricID]float64),
}
}

Expand All @@ -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

Expand All @@ -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 {
Expand Down Expand Up @@ -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)
Expand All @@ -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()}
}
152 changes: 152 additions & 0 deletions sink/delta_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@ package sink

import (
"context"
"fmt"
"sync"
"testing"
"time"

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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{
Expand Down
Loading