From a9d063f30547e0640ac3e1e4dc892dc11c51c5bb Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 22 Oct 2026 13:09:09 -0700 Subject: [PATCH 1/4] Accept retry options on built-in sinks sink.Make parsed buffer-size, send-timeout, and send-retry-wait into RetryArgs but then passed the full options map to the specific sink constructor, which rejects unknown options. Any monitor that set a retry option failed to build its sink. Pass the specific sink, and the HTTP client factory, a copy of the options without the retry keys. Defaults are unchanged. Also fix the retry sink docs, which showed a separate "retry" sink section that Make rejects as unregistered. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 1 + docs/content/config/quick-ref.md | 6 +- docs/content/sinks/retry.md | 37 +++++++++- sink/factory.go | 23 ++++-- sink/factory_test.go | 108 ++++++++++++++++++++++++++++ 5 files changed, 164 insertions(+), 11 deletions(-) create mode 100644 sink/factory_test.go diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 0a7a56ef..e6a06eff 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -57,6 +57,7 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * Fixed the `wait.io.table` collector after a failed read. With `truncate-table=no`, or on a read error not caused by the CMR, it returned partial metrics and no error; it now returns the error and no metrics. With `truncate-table=yes` (the default), it doesn't truncate the table after a failed read, and the truncate error policy's metric setting (`drop` or `zero`) now applies to the next sample, which covers more than one interval. Previously, a failed query or row scan was followed by one sample reported as a one-interval delta, about twice too high. A failed read is also no longer subject to the truncate error policy's `ignore` and `stop` settings: previously, a CMR timeout mid-read also failed the truncate, so `ignore` hid the error and `stop` stopped the collector. * Fixed the `repl.lag` collector's heartbeat reader, which kept running after its monitor stopped when `repl.lag` was collected at more than one level. It now stops when its monitor stops. * Changed the engine to call a collector's `Prepare` cleanup func when `Prepare` returns an error with a non-nil cleanup func, and to call the cleanup funcs of the collectors already prepared when a later collector in the plan fails. Previously they were not called, so resources started in `Prepare` leaked. +* Fixed built-in sinks rejecting the retry options `buffer-size`, `send-timeout`, and `send-retry-wait`. ### v2.1.0 (10 Aug 2026) diff --git a/docs/content/config/quick-ref.md b/docs/content/config/quick-ref.md index 73e35182..928dec65 100644 --- a/docs/content/config/quick-ref.md +++ b/docs/content/config/quick-ref.md @@ -103,12 +103,10 @@ sinks: # No options noop: # No options - retry: - buffer-size: 60 - send-timeout: 5s - send-retry-wait: 200ms signalfx: # See Sinks > signalfx + # Built-in sinks except log and noop also accept + # retry options. See Sinks > retry tags: env: ${ENVIRONMENT:-dev} diff --git a/docs/content/sinks/retry.md b/docs/content/sinks/retry.md index c9dcf693..d2834978 100644 --- a/docs/content/sinks/retry.md +++ b/docs/content/sinks/retry.md @@ -8,12 +8,47 @@ The built-in sinks, except [`log`]({{< ref "log" >}}), use a retry sink to handl The retry sink uses a LIFO queue (a stack) to prioritize sending the latest metrics. During a long outage of the real sink, the retry sink drops the oldest metrics and keeps the latest metrics, up to its buffer size, which is configurable. +Retry is not configured as a separate sink. +Set its options in the options of each built-in sink that uses it, alongside that sink's own options. + ## Quick Reference ```yaml sinks: - retry: + datadog: + # datadog options buffer-size: 60 send-timeout: 5s send-retry-wait: 200ms ``` + +## Options + +### `buffer-size` + +| | | +|-|-| +|**Valid values**|Integer greater than zero| +|**Default value**|60| + +Maximum number of metrics batches buffered while the real sink is failing. +When the buffer is full, the oldest batch is dropped. + +### `send-timeout` + +| | | +|-|-| +|**Valid values**|[Go duration string](https://pkg.go.dev/time#ParseDuration)| +|**Default value**|5s| + +Maximum time for each send cycle, in which buffered metrics batches are sent to the real sink, newest first. +Batches not sent in time stay in the buffer for the next cycle. + +### `send-retry-wait` + +| | | +|-|-| +|**Valid values**|[Go duration string](https://pkg.go.dev/time#ParseDuration)| +|**Default value**|200ms| + +Wait time between sends to the real sink within one send cycle. diff --git a/sink/factory.go b/sink/factory.go index 5ad4135e..ba5865da 100644 --- a/sink/factory.go +++ b/sink/factory.go @@ -138,31 +138,42 @@ func (f *factory) Make(args blip.SinkFactoryArgs) (blip.Sink, error) { retryArgs.SendRetryWait = d } + // Retry options are consumed here, so remove them before passing + // options to the specific sink, which rejects unknown options + sinkOpts := make(map[string]string, len(args.Options)) + for k, v := range args.Options { + switch k { + case "buffer-size", "send-timeout", "send-retry-wait": + continue + } + sinkOpts[k] = v + } + // Make specific built-in sink var err error switch args.SinkName { case "chronosphere": - retryArgs.Sink, err = NewChronosphere(args.MonitorId, args.Options, args.Tags) + retryArgs.Sink, err = NewChronosphere(args.MonitorId, sinkOpts, args.Tags) case "signalfx": - httpClient, err := f.HTTPClient.MakeForSink("signalfx", args.MonitorId, args.Options, args.Tags) + httpClient, err := f.HTTPClient.MakeForSink("signalfx", args.MonitorId, sinkOpts, args.Tags) if err != nil { return nil, err } - retryArgs.Sink, err = NewSignalFx(args.MonitorId, args.Options, args.Tags, httpClient) + retryArgs.Sink, err = NewSignalFx(args.MonitorId, sinkOpts, args.Tags, httpClient) if err != nil { return nil, err } case "datadog": - httpClient, err := f.HTTPClient.MakeForSink("datadog", args.MonitorId, args.Options, args.Tags) + httpClient, err := f.HTTPClient.MakeForSink("datadog", args.MonitorId, sinkOpts, args.Tags) if err != nil { return nil, err } - retryArgs.Sink, err = NewDatadog(args.MonitorId, args.Options, args.Tags, httpClient) + retryArgs.Sink, err = NewDatadog(args.MonitorId, sinkOpts, args.Tags, httpClient) if err != nil { return nil, err } case "prom-pushgateway": - retryArgs.Sink, err = NewPromPushgateway(args.MonitorId, args.Options, args.Tags) + retryArgs.Sink, err = NewPromPushgateway(args.MonitorId, sinkOpts, args.Tags) default: return nil, fmt.Errorf("sink %s not registered", args.SinkName) } diff --git a/sink/factory_test.go b/sink/factory_test.go new file mode 100644 index 00000000..9894823c --- /dev/null +++ b/sink/factory_test.go @@ -0,0 +1,108 @@ +// Copyright 2026 Block, Inc. + +package sink + +import ( + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/cashapp/blip/v2" + "github.com/cashapp/blip/v2/status" +) + +type testHTTPClientFactory struct { + opts map[string]map[string]string // sink name => options +} + +func (f *testHTTPClientFactory) MakeForSink(sinkName, monitorId string, opts, tags map[string]string) (*http.Client, error) { + f.opts[sinkName] = opts + return &http.Client{}, nil +} + +func TestFactoryMake_RetryOptions(t *testing.T) { + httpFactory := &testHTTPClientFactory{opts: map[string]map[string]string{}} + defer func(prev blip.HTTPClientFactory) { f.HTTPClient = prev }(f.HTTPClient) + f.HTTPClient = httpFactory + + retryOpts := map[string]string{ + "buffer-size": "3", + "send-timeout": "7s", + "send-retry-wait": "300ms", + } + + tests := []struct { + sinkName string + opts map[string]string // sink-specific options + http bool // uses HTTPClient.MakeForSink + }{ + {sinkName: "chronosphere", opts: map[string]string{"url": "http://127.0.0.1:3030"}}, + {sinkName: "signalfx", opts: map[string]string{"auth-token": "token"}, http: true}, + {sinkName: "datadog", opts: map[string]string{"api-key-auth": "key", "app-key-auth": "key"}, http: true}, + {sinkName: "prom-pushgateway", opts: map[string]string{"addr": "http://127.0.0.1:9091"}}, + } + for _, tt := range tests { + t.Run(tt.sinkName, func(t *testing.T) { + opts := map[string]string{} + for k, v := range tt.opts { + opts[k] = v + } + for k, v := range retryOpts { + opts[k] = v + } + + monitorId := "factory-retry-" + tt.sinkName + defer status.RemoveMonitor(monitorId) + + s, err := Make(blip.SinkFactoryArgs{ + SinkName: tt.sinkName, + MonitorId: monitorId, + Options: opts, + }) + require.NoError(t, err) + + // Monitor registers the retry buffer counter through the returned sink + cs, ok := s.(CounterSink) + require.True(t, ok, "sink is %T, expected CounterSink", s) + cs.RegisterCounters() + counter := status.SINK_RETRY_DROPPED + ":" + tt.sinkName + _, ok = status.ReportCounters(monitorId)[monitorId][counter] + assert.True(t, ok, "%s not reported", counter) + + if d, ok := s.(*Delta); ok { + s = d.sink + } + retry, ok := s.(*Retry) + require.True(t, ok, "sink is %T, expected *Retry", s) + assert.Equal(t, 3, retry.max+1) + assert.Equal(t, 3, len(retry.stack)) + assert.Equal(t, 7*time.Second, retry.sendTimeout) + assert.Equal(t, 300*time.Millisecond, retry.retryWait) + assert.Equal(t, tt.sinkName, retry.Name()) + + // Caller's options are not modified + assert.Len(t, opts, len(tt.opts)+len(retryOpts)) + + if tt.http { + assert.Equal(t, tt.opts, httpFactory.opts[tt.sinkName]) + } + }) + } +} + +func TestFactoryMake_RetryDefaults(t *testing.T) { + s, err := Make(blip.SinkFactoryArgs{ + SinkName: "chronosphere", + MonitorId: "m1", + }) + require.NoError(t, err) + + retry, ok := s.(*Retry) + require.True(t, ok, "sink is %T, expected *Retry", s) + assert.Equal(t, DEFAULT_RETRY_BUFFER_SIZE, retry.max+1) + assert.Equal(t, 5*time.Second, retry.sendTimeout) + assert.Equal(t, 200*time.Millisecond, retry.retryWait) +} From db29086d72c27fbc18715ce95d57786c6ac490ba Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Wed, 28 Oct 2026 13:49:35 -0700 Subject: [PATCH 2/4] Validate retry options and document them on each sink Reject buffer-size outside 1..10000 (MAX_RETRY_BUFFER_SIZE) and send-timeout or send-retry-wait that isn't greater than zero. A negative send-timeout made Retry buffer without ever sending, a zero silently meant the default, and a huge buffer-size could allocate a huge buffer slice. Parse and range errors now name the sink and option. Clone the options and delete each retry key where it's parsed, so a new retry option can't be missed when stripping. Document the valid values, link the retry options from each built-in sink page, and note the removed sinks.retry section in the release notes. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 3 +- docs/content/sinks/chronosphere.md | 2 ++ docs/content/sinks/datadog.md | 2 ++ docs/content/sinks/prom-pushgateway.md | 2 ++ docs/content/sinks/retry.md | 13 ++++---- docs/content/sinks/signalfx.md | 2 ++ sink/factory.go | 46 +++++++++++++++----------- sink/factory_test.go | 37 +++++++++++++++++++++ sink/retry.go | 3 ++ 9 files changed, 83 insertions(+), 27 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index e6a06eff..194f08ae 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -57,7 +57,8 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * Fixed the `wait.io.table` collector after a failed read. With `truncate-table=no`, or on a read error not caused by the CMR, it returned partial metrics and no error; it now returns the error and no metrics. With `truncate-table=yes` (the default), it doesn't truncate the table after a failed read, and the truncate error policy's metric setting (`drop` or `zero`) now applies to the next sample, which covers more than one interval. Previously, a failed query or row scan was followed by one sample reported as a one-interval delta, about twice too high. A failed read is also no longer subject to the truncate error policy's `ignore` and `stop` settings: previously, a CMR timeout mid-read also failed the truncate, so `ignore` hid the error and `stop` stopped the collector. * Fixed the `repl.lag` collector's heartbeat reader, which kept running after its monitor stopped when `repl.lag` was collected at more than one level. It now stops when its monitor stops. * Changed the engine to call a collector's `Prepare` cleanup func when `Prepare` returns an error with a non-nil cleanup func, and to call the cleanup funcs of the collectors already prepared when a later collector in the plan fails. Previously they were not called, so resources started in `Prepare` leaked. -* Fixed built-in sinks rejecting the retry options `buffer-size`, `send-timeout`, and `send-retry-wait`. +* Fixed built-in sinks rejecting the retry options `buffer-size`, `send-timeout`, and `send-retry-wait`. Set these options in each built-in sink's options. The docs previously showed a separate `sinks.retry` section, which fails with "sink retry not registered". +* Added validation of the retry options: `buffer-size` must be from 1 to 10000, and `send-timeout` and `send-retry-wait` must be greater than zero. Other values prevent the monitor from loading. Previously a built-in sink rejected any retry option, so no config that loaded before is affected. ### v2.1.0 (10 Aug 2026) diff --git a/docs/content/sinks/chronosphere.md b/docs/content/sinks/chronosphere.md index 9f54b8a4..1edd8d9f 100644 --- a/docs/content/sinks/chronosphere.md +++ b/docs/content/sinks/chronosphere.md @@ -19,6 +19,8 @@ If you want a Chronosphere collector to scrape metrics, use [Prometheus emulatio This sink reports Prometheus-style metric names: `mysql_status_threads_running` instead of `status.global.threads_running`. It reports all [tags]({{< ref "/config/config-file#tags" >}}) as Prometheus labels. +This sink also accepts the [retry]({{< ref "retry" >}}) options `buffer-size`, `send-timeout`, and `send-retry-wait`. + ## Quick Reference ```yaml diff --git a/docs/content/sinks/datadog.md b/docs/content/sinks/datadog.md index ca1dea62..0832e3b3 100644 --- a/docs/content/sinks/datadog.md +++ b/docs/content/sinks/datadog.md @@ -28,6 +28,8 @@ Starting from v1.1, this sink now sends delta values to Datadog for counter metr If you wish to track the difference in counter values since v1.1, you can tag metrics with the blip version. +This sink also accepts the [retry]({{< ref "retry" >}}) options `buffer-size`, `send-timeout`, and `send-retry-wait`. + ## Quick Reference ```yaml diff --git a/docs/content/sinks/prom-pushgateway.md b/docs/content/sinks/prom-pushgateway.md index f822ea2d..4413b6cb 100644 --- a/docs/content/sinks/prom-pushgateway.md +++ b/docs/content/sinks/prom-pushgateway.md @@ -13,6 +13,8 @@ It uses Prometheus domain translation, so it only works with [compatible domains Currently, this sink is intended to work with [Vector by Datadog](https://vector.dev/). Therefore, it uses only `POST` and the text (exposition) protocol; it does not use RPC with protos. +This sink also accepts the [retry]({{< ref "retry" >}}) options `buffer-size`, `send-timeout`, and `send-retry-wait`. + ## Quick Reference ```yaml diff --git a/docs/content/sinks/retry.md b/docs/content/sinks/retry.md index d2834978..2247e17a 100644 --- a/docs/content/sinks/retry.md +++ b/docs/content/sinks/retry.md @@ -3,13 +3,14 @@ title: retry --- The retry sink is a pseudo-sink that provides buffering, serialization, and retry for real sinks. -The built-in sinks, except [`log`]({{< ref "log" >}}), use a retry sink to handle those three complexities. +The built-in sinks, except [`log`]({{< ref "log" >}}) and `noop`, use a retry sink to handle those three complexities. The retry sink uses a LIFO queue (a stack) to prioritize sending the latest metrics. During a long outage of the real sink, the retry sink drops the oldest metrics and keeps the latest metrics, up to its buffer size, which is configurable. -Retry is not configured as a separate sink. -Set its options in the options of each built-in sink that uses it, alongside that sink's own options. +Retry is not configured as a separate sink: a `sinks.retry` section fails with "sink retry not registered". +Set the retry options in the options of each built-in sink that uses it, alongside that sink's own options. +An invalid value prevents the monitor from loading. ## Quick Reference @@ -28,7 +29,7 @@ sinks: | | | |-|-| -|**Valid values**|Integer greater than zero| +|**Valid values**|Integer from 1 to 10000| |**Default value**|60| Maximum number of metrics batches buffered while the real sink is failing. @@ -38,7 +39,7 @@ When the buffer is full, the oldest batch is dropped. | | | |-|-| -|**Valid values**|[Go duration string](https://pkg.go.dev/time#ParseDuration)| +|**Valid values**|[Go duration string](https://pkg.go.dev/time#ParseDuration) greater than zero| |**Default value**|5s| Maximum time for each send cycle, in which buffered metrics batches are sent to the real sink, newest first. @@ -48,7 +49,7 @@ Batches not sent in time stay in the buffer for the next cycle. | | | |-|-| -|**Valid values**|[Go duration string](https://pkg.go.dev/time#ParseDuration)| +|**Valid values**|[Go duration string](https://pkg.go.dev/time#ParseDuration) greater than zero| |**Default value**|200ms| Wait time between sends to the real sink within one send cycle. diff --git a/docs/content/sinks/signalfx.md b/docs/content/sinks/signalfx.md index b050b1b9..e36cba33 100644 --- a/docs/content/sinks/signalfx.md +++ b/docs/content/sinks/signalfx.md @@ -10,6 +10,8 @@ The signalfx sink sends metrics to [Splunk](https://www.splunk.com/), which [acq It reports all [tags]({{< ref "/config/config-file#tags" >}}) as dimensions. +This sink also accepts the [retry]({{< ref "retry" >}}) options `buffer-size`, `send-timeout`, and `send-retry-wait`. + ## Quick Reference ```yaml diff --git a/sink/factory.go b/sink/factory.go index ba5865da..46808570 100644 --- a/sink/factory.go +++ b/sink/factory.go @@ -5,6 +5,7 @@ package sink import ( "context" "fmt" + "maps" "strconv" "sync" "time" @@ -109,46 +110,40 @@ func (f *factory) Make(args blip.SinkFactoryArgs) (blip.Sink, error) { // Built-in sinks use Retry to serialize access and retry on Send error. // First build the specific sink, then return it wrapped in Retry. - // Parse retry options + // Parse retry options. Retry options are consumed here, so delete them + // from the options passed to the specific sink, which rejects unknown options. + sinkOpts := maps.Clone(args.Options) retryArgs := RetryArgs{ MonitorId: args.MonitorId, } - if v, ok := args.Options["buffer-size"]; ok { + if v, ok := sinkOpts["buffer-size"]; ok { + delete(sinkOpts, "buffer-size") n, err := strconv.Atoi(v) if err != nil { - return nil, err + return nil, fmt.Errorf("invalid %s sink option buffer-size: %w", args.SinkName, err) } - if n < 0 { - return nil, fmt.Errorf("invalid retry buffer-size: %d: must be greater than zero", n) + if n <= 0 || n > MAX_RETRY_BUFFER_SIZE { + return nil, fmt.Errorf("invalid %s sink option buffer-size: %d: must be greater than zero and at most %d", args.SinkName, n, MAX_RETRY_BUFFER_SIZE) } retryArgs.BufferSize = uint(n) } - if v, ok := args.Options["send-timeout"]; ok { - d, err := time.ParseDuration(v) + if v, ok := sinkOpts["send-timeout"]; ok { + delete(sinkOpts, "send-timeout") + d, err := parseRetryDuration(args.SinkName, "send-timeout", v) if err != nil { return nil, err } retryArgs.SendTimeout = d } - if v, ok := args.Options["send-retry-wait"]; ok { - d, err := time.ParseDuration(v) + if v, ok := sinkOpts["send-retry-wait"]; ok { + delete(sinkOpts, "send-retry-wait") + d, err := parseRetryDuration(args.SinkName, "send-retry-wait", v) if err != nil { return nil, err } retryArgs.SendRetryWait = d } - // Retry options are consumed here, so remove them before passing - // options to the specific sink, which rejects unknown options - sinkOpts := make(map[string]string, len(args.Options)) - for k, v := range args.Options { - switch k { - case "buffer-size", "send-timeout", "send-retry-wait": - continue - } - sinkOpts[k] = v - } - // Make specific built-in sink var err error switch args.SinkName { @@ -191,3 +186,14 @@ func (f *factory) Make(args blip.SinkFactoryArgs) (blip.Sink, error) { return NewRetry(retryArgs), nil } } + +func parseRetryDuration(sinkName, option, v string) (time.Duration, error) { + d, err := time.ParseDuration(v) + if err != nil { + return 0, fmt.Errorf("invalid %s sink option %s: %w", sinkName, option, err) + } + if d <= 0 { + return 0, fmt.Errorf("invalid %s sink option %s: %s: must be greater than zero", sinkName, option, v) + } + return d, nil +} diff --git a/sink/factory_test.go b/sink/factory_test.go index 9894823c..d5d5498a 100644 --- a/sink/factory_test.go +++ b/sink/factory_test.go @@ -106,3 +106,40 @@ func TestFactoryMake_RetryDefaults(t *testing.T) { assert.Equal(t, 5*time.Second, retry.sendTimeout) assert.Equal(t, 200*time.Millisecond, retry.retryWait) } + +func TestFactoryMake_InvalidRetryOptions(t *testing.T) { + tests := []struct { + option string + value string + }{ + {"buffer-size", "ten"}, + {"buffer-size", "-1"}, + {"buffer-size", "0"}, + {"buffer-size", "10001"}, + {"send-timeout", "5"}, + {"send-timeout", "-1s"}, + {"send-timeout", "0s"}, + {"send-retry-wait", "fast"}, + {"send-retry-wait", "-200ms"}, + {"send-retry-wait", "0"}, + } + for _, tt := range tests { + t.Run(tt.option+"="+tt.value, func(t *testing.T) { + _, err := Make(blip.SinkFactoryArgs{ + SinkName: "chronosphere", + MonitorId: "m1", + Options: map[string]string{tt.option: tt.value}, + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "invalid chronosphere sink option "+tt.option+": ") + }) + } + + // Max buffer-size is valid + _, err := Make(blip.SinkFactoryArgs{ + SinkName: "chronosphere", + MonitorId: "m1", + Options: map[string]string{"buffer-size": "10000"}, + }) + require.NoError(t, err) +} diff --git a/sink/retry.go b/sink/retry.go index 31dccdc1..249f1ad9 100644 --- a/sink/retry.go +++ b/sink/retry.go @@ -17,6 +17,9 @@ const ( DEFAULT_RETRY_BUFFER_SIZE = 60 DEFAULT_RETRY_SEND_TIMEOUT = "5s" DEFAULT_RETRY_SEND_RETRY_WAIT = "200ms" + + // MAX_RETRY_BUFFER_SIZE is the maximum buffer-size sink option. + MAX_RETRY_BUFFER_SIZE = 10000 ) // Retry is a pseudo-sink that provides buffering, serialization, and retry for From 841e8ce59ff4133a473bc763faa17caf1a263561 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Wed, 28 Oct 2026 14:03:57 -0700 Subject: [PATCH 3/4] Bound retry send time and validate RetryArgs in NewRetry Cap send-timeout at 1m (MAX_RETRY_SEND_TIMEOUT). The level collector sends to a monitor's sinks one at a time without its own timeout, so a long send-timeout stalled the monitor's other sinks in an outage. Stop the wait between sends when send-timeout expires. time.Sleep ignored the timeout, so a long send-retry-wait overran it. Move the range checks into RetryArgs.Validate, which the factory calls and NewRetry enforces with a panic like its other argument checks, so custom sinks that build RetryArgs get the same bounds. Parse the retry options in one loop that deletes each key and wraps each error in one place, and derive test boundaries from the constants. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 4 +- docs/content/sinks/retry.md | 5 ++- sink/factory.go | 60 ++++++++++++++++------------- sink/factory_test.go | 11 ++++-- sink/retry.go | 40 +++++++++++++++++-- sink/retry_test.go | 55 ++++++++++++++++++++++++++ 6 files changed, 140 insertions(+), 35 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 194f08ae..bdf74a43 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -58,7 +58,9 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * Fixed the `repl.lag` collector's heartbeat reader, which kept running after its monitor stopped when `repl.lag` was collected at more than one level. It now stops when its monitor stops. * Changed the engine to call a collector's `Prepare` cleanup func when `Prepare` returns an error with a non-nil cleanup func, and to call the cleanup funcs of the collectors already prepared when a later collector in the plan fails. Previously they were not called, so resources started in `Prepare` leaked. * Fixed built-in sinks rejecting the retry options `buffer-size`, `send-timeout`, and `send-retry-wait`. Set these options in each built-in sink's options. The docs previously showed a separate `sinks.retry` section, which fails with "sink retry not registered". -* Added validation of the retry options: `buffer-size` must be from 1 to 10000, and `send-timeout` and `send-retry-wait` must be greater than zero. Other values prevent the monitor from loading. Previously a built-in sink rejected any retry option, so no config that loaded before is affected. +* Added validation of the retry options: `buffer-size` must be from 1 to 10000, `send-timeout` must be greater than zero and at most 1m, and `send-retry-wait` must be greater than zero. Other values prevent the monitor from loading. Previously a built-in sink rejected any retry option, so no config that loaded before is affected. +* Changed the retry sink to stop waiting `send-retry-wait` between sends when `send-timeout` expires. Previously a send cycle could run past `send-timeout` by up to one `send-retry-wait`. +* Added `sink.RetryArgs.Validate` and the `sink.MAX_RETRY_BUFFER_SIZE` and `sink.MAX_RETRY_SEND_TIMEOUT` constants. **Behavior change:** `sink.NewRetry` now panics when `Validate` returns an error: a `BufferSize` above 10000, a `SendTimeout` above 1m, or a negative `SendTimeout` or `SendRetryWait`. Custom sinks that build `RetryArgs` from user input should call `Validate` first. ### v2.1.0 (10 Aug 2026) diff --git a/docs/content/sinks/retry.md b/docs/content/sinks/retry.md index 2247e17a..019601fd 100644 --- a/docs/content/sinks/retry.md +++ b/docs/content/sinks/retry.md @@ -39,11 +39,13 @@ When the buffer is full, the oldest batch is dropped. | | | |-|-| -|**Valid values**|[Go duration string](https://pkg.go.dev/time#ParseDuration) greater than zero| +|**Valid values**|[Go duration string](https://pkg.go.dev/time#ParseDuration) greater than zero and at most 1m| |**Default value**|5s| Maximum time for each send cycle, in which buffered metrics batches are sent to the real sink, newest first. +When the timeout expires, the send in progress is cancelled and the cycle ends; a real sink that ignores cancellation can run past it. Batches not sent in time stay in the buffer for the next cycle. +A monitor finishes each send cycle before it sends metrics to its next sink, so a long timeout delays the monitor's other sinks during an outage. ### `send-retry-wait` @@ -53,3 +55,4 @@ Batches not sent in time stay in the buffer for the next cycle. |**Default value**|200ms| Wait time between sends to the real sink within one send cycle. +The wait ends early when `send-timeout` expires, which ends the cycle. diff --git a/sink/factory.go b/sink/factory.go index 46808570..7eddacd4 100644 --- a/sink/factory.go +++ b/sink/factory.go @@ -112,36 +112,29 @@ func (f *factory) Make(args blip.SinkFactoryArgs) (blip.Sink, error) { // Parse retry options. Retry options are consumed here, so delete them // from the options passed to the specific sink, which rejects unknown options. - sinkOpts := maps.Clone(args.Options) retryArgs := RetryArgs{ MonitorId: args.MonitorId, } - if v, ok := sinkOpts["buffer-size"]; ok { - delete(sinkOpts, "buffer-size") - n, err := strconv.Atoi(v) - if err != nil { - return nil, fmt.Errorf("invalid %s sink option buffer-size: %w", args.SinkName, err) - } - if n <= 0 || n > MAX_RETRY_BUFFER_SIZE { - return nil, fmt.Errorf("invalid %s sink option buffer-size: %d: must be greater than zero and at most %d", args.SinkName, n, MAX_RETRY_BUFFER_SIZE) + sinkOpts := maps.Clone(args.Options) + for k, v := range args.Options { + var err error + switch k { + case "buffer-size": + retryArgs.BufferSize, err = parseRetryBufferSize(v) + case "send-timeout": + retryArgs.SendTimeout, err = parseRetryDuration(v) + case "send-retry-wait": + retryArgs.SendRetryWait, err = parseRetryDuration(v) + default: + continue // specific sink option } - retryArgs.BufferSize = uint(n) - } - if v, ok := sinkOpts["send-timeout"]; ok { - delete(sinkOpts, "send-timeout") - d, err := parseRetryDuration(args.SinkName, "send-timeout", v) if err != nil { - return nil, err + return nil, fmt.Errorf("invalid %s sink option %s: %w", args.SinkName, k, err) } - retryArgs.SendTimeout = d + delete(sinkOpts, k) } - if v, ok := sinkOpts["send-retry-wait"]; ok { - delete(sinkOpts, "send-retry-wait") - d, err := parseRetryDuration(args.SinkName, "send-retry-wait", v) - if err != nil { - return nil, err - } - retryArgs.SendRetryWait = d + if err := retryArgs.Validate(); err != nil { + return nil, fmt.Errorf("invalid %s sink option %w", args.SinkName, err) } // Make specific built-in sink @@ -187,13 +180,28 @@ func (f *factory) Make(args blip.SinkFactoryArgs) (blip.Sink, error) { } } -func parseRetryDuration(sinkName, option, v string) (time.Duration, error) { +// parseRetryBufferSize parses a buffer-size option, which must be greater than +// zero. RetryArgs.Validate checks the maximum. +func parseRetryBufferSize(v string) (uint, error) { + n, err := strconv.Atoi(v) + if err != nil { + return 0, err + } + if n <= 0 { + return 0, fmt.Errorf("%d: must be greater than zero", n) + } + return uint(n), nil +} + +// parseRetryDuration parses a send-timeout or send-retry-wait option, which must +// be greater than zero. RetryArgs.Validate checks the maximum. +func parseRetryDuration(v string) (time.Duration, error) { d, err := time.ParseDuration(v) if err != nil { - return 0, fmt.Errorf("invalid %s sink option %s: %w", sinkName, option, err) + return 0, err } if d <= 0 { - return 0, fmt.Errorf("invalid %s sink option %s: %s: must be greater than zero", sinkName, option, v) + return 0, fmt.Errorf("%s: must be greater than zero", v) } return d, nil } diff --git a/sink/factory_test.go b/sink/factory_test.go index d5d5498a..e27252a1 100644 --- a/sink/factory_test.go +++ b/sink/factory_test.go @@ -4,6 +4,7 @@ package sink import ( "net/http" + "strconv" "testing" "time" @@ -115,10 +116,11 @@ func TestFactoryMake_InvalidRetryOptions(t *testing.T) { {"buffer-size", "ten"}, {"buffer-size", "-1"}, {"buffer-size", "0"}, - {"buffer-size", "10001"}, + {"buffer-size", strconv.Itoa(MAX_RETRY_BUFFER_SIZE + 1)}, {"send-timeout", "5"}, {"send-timeout", "-1s"}, {"send-timeout", "0s"}, + {"send-timeout", (MAX_RETRY_SEND_TIMEOUT + time.Second).String()}, {"send-retry-wait", "fast"}, {"send-retry-wait", "-200ms"}, {"send-retry-wait", "0"}, @@ -135,11 +137,14 @@ func TestFactoryMake_InvalidRetryOptions(t *testing.T) { }) } - // Max buffer-size is valid + // Maximum values are valid _, err := Make(blip.SinkFactoryArgs{ SinkName: "chronosphere", MonitorId: "m1", - Options: map[string]string{"buffer-size": "10000"}, + Options: map[string]string{ + "buffer-size": strconv.Itoa(MAX_RETRY_BUFFER_SIZE), + "send-timeout": MAX_RETRY_SEND_TIMEOUT.String(), + }, }) require.NoError(t, err) } diff --git a/sink/retry.go b/sink/retry.go index 249f1ad9..e3634eb8 100644 --- a/sink/retry.go +++ b/sink/retry.go @@ -4,6 +4,7 @@ package sink import ( "context" + "fmt" "sync" "sync/atomic" "time" @@ -18,8 +19,10 @@ const ( DEFAULT_RETRY_SEND_TIMEOUT = "5s" DEFAULT_RETRY_SEND_RETRY_WAIT = "200ms" - // MAX_RETRY_BUFFER_SIZE is the maximum buffer-size sink option. - MAX_RETRY_BUFFER_SIZE = 10000 + // Maximum RetryArgs values. MAX_RETRY_SEND_TIMEOUT bounds how long Send + // can block its caller. + MAX_RETRY_BUFFER_SIZE = 10000 + MAX_RETRY_SEND_TIMEOUT = time.Minute ) // Retry is a pseudo-sink that provides buffering, serialization, and retry for @@ -87,6 +90,25 @@ type RetryArgs struct { SendRetryWait time.Duration // optional; DEFAULT_RETRY_SEND_RETRY_WAIT } +// Validate returns an error if an optional value is out of range. Zero values +// are valid and mean the default. Errors are prefixed with the sink option +// name of the invalid value. NewRetry panics if Validate returns an error. +func (a RetryArgs) Validate() error { + if a.BufferSize > MAX_RETRY_BUFFER_SIZE { + return fmt.Errorf("buffer-size: %d: must be at most %d", a.BufferSize, MAX_RETRY_BUFFER_SIZE) + } + if a.SendTimeout < 0 { + return fmt.Errorf("send-timeout: %s: must not be negative", a.SendTimeout) + } + if a.SendTimeout > MAX_RETRY_SEND_TIMEOUT { + return fmt.Errorf("send-timeout: %s: must be at most %s", a.SendTimeout, MAX_RETRY_SEND_TIMEOUT) + } + if a.SendRetryWait < 0 { + return fmt.Errorf("send-retry-wait: %s: must not be negative", a.SendRetryWait) + } + return nil +} + func NewRetry(args RetryArgs) *Retry { // Panic if caller doesn't provide required args if args.MonitorId == "" { @@ -99,6 +121,9 @@ func NewRetry(args RetryArgs) *Retry { if _, ok := args.Sink.(*Delta); ok { panic("RetryArgs.Sink cannot be a Delta sink.") } + if err := args.Validate(); err != nil { + panic("invalid RetryArgs: " + err.Error()) + } // Set defaults if args.BufferSize == 0 { @@ -182,9 +207,16 @@ func (rb *Retry) Send(ctx context.Context, m *blip.Metrics) error { default: } - // Throttle between send, except on first send + // Throttle between send, except on first send. Stop waiting when + // either context is cancelled, so Send doesn't overrun sendTimeout. if n > 0 { - time.Sleep(rb.retryWait) + t := time.NewTimer(rb.retryWait) + select { + case <-ctx2.Done(): + t.Stop() + return nil + case <-t.C: + } } n += 1 diff --git a/sink/retry_test.go b/sink/retry_test.go index f2edb8a7..b3d0ca97 100644 --- a/sink/retry_test.go +++ b/sink/retry_test.go @@ -156,6 +156,61 @@ func TestRetry_SendTimeoutBoundsInFlightSend(t *testing.T) { } } +func TestRetry_SendTimeoutBoundsRetryWait(t *testing.T) { + // Real sink always fails, so Send waits send-retry-wait before retrying + failingSink := mock.Sink{ + SendFunc: func(ctx context.Context, m *blip.Metrics) error { + return fmt.Errorf("send failed") + }, + } + rb := NewRetry(RetryArgs{ + MonitorId: "m1", + Sink: failingSink, + BufferSize: 3, + SendTimeout: 200 * time.Millisecond, + SendRetryWait: time.Minute, + }) + + done := make(chan struct{}) + go func() { + rb.Send(context.Background(), &blip.Metrics{Level: "1"}) + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Retry.Send stalled: SendTimeout did not bound the retry wait") + } + assert.Equal(t, []string{"1", "", ""}, stack(rb)) +} + +func TestRetryArgsValidate(t *testing.T) { + sink := mock.Sink{} + valid := []RetryArgs{ + {}, // defaults + {BufferSize: MAX_RETRY_BUFFER_SIZE, SendTimeout: MAX_RETRY_SEND_TIMEOUT, SendRetryWait: time.Hour}, + } + for _, args := range valid { + assert.NoError(t, args.Validate(), "%+v", args) + } + + invalid := []RetryArgs{ + {BufferSize: MAX_RETRY_BUFFER_SIZE + 1}, + {SendTimeout: -time.Second}, + {SendTimeout: MAX_RETRY_SEND_TIMEOUT + time.Second}, + {SendRetryWait: -time.Second}, + } + for _, args := range invalid { + assert.Error(t, args.Validate(), "%+v", args) + + // NewRetry panics on invalid args + args.MonitorId = "m1" + args.Sink = sink + assert.Panics(t, func() { NewRetry(args) }, "%+v", args) + } +} + func TestRetryPopMiddle(t *testing.T) { // Test that popping values from the middle of the stack works. This happens // when, in this test for example, while sending metrics1, two more metrics From 7c8d3bebf7876f0105d51d60a031bb39942bbd93 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Wed, 28 Oct 2026 14:36:30 -0700 Subject: [PATCH 4/4] Keep NewRetry unchanged and parse retry options in fixed order Remove the Validate panic from NewRetry. NewRetry is exported, so panicking on values that v2.1.0 accepted would break Go callers in a minor release. NewRetry handles every value as before; the factory still validates sink options, and RetryArgs and NewRetry now say that callers building RetryArgs from user input should call Validate. Parse the retry options in a fixed order instead of map order, so the same invalid options always return the same error. Fix the Retry comment: "noop" doesn't use Retry either. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 2 +- sink/factory.go | 10 +++++++--- sink/factory_test.go | 16 ++++++++++++++++ sink/retry.go | 14 ++++++++------ sink/retry_test.go | 4 ++-- 5 files changed, 34 insertions(+), 12 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index bdf74a43..9b9b89d1 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -60,7 +60,7 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * Fixed built-in sinks rejecting the retry options `buffer-size`, `send-timeout`, and `send-retry-wait`. Set these options in each built-in sink's options. The docs previously showed a separate `sinks.retry` section, which fails with "sink retry not registered". * Added validation of the retry options: `buffer-size` must be from 1 to 10000, `send-timeout` must be greater than zero and at most 1m, and `send-retry-wait` must be greater than zero. Other values prevent the monitor from loading. Previously a built-in sink rejected any retry option, so no config that loaded before is affected. * Changed the retry sink to stop waiting `send-retry-wait` between sends when `send-timeout` expires. Previously a send cycle could run past `send-timeout` by up to one `send-retry-wait`. -* Added `sink.RetryArgs.Validate` and the `sink.MAX_RETRY_BUFFER_SIZE` and `sink.MAX_RETRY_SEND_TIMEOUT` constants. **Behavior change:** `sink.NewRetry` now panics when `Validate` returns an error: a `BufferSize` above 10000, a `SendTimeout` above 1m, or a negative `SendTimeout` or `SendRetryWait`. Custom sinks that build `RetryArgs` from user input should call `Validate` first. +* Added `sink.RetryArgs.Validate` and the `sink.MAX_RETRY_BUFFER_SIZE` and `sink.MAX_RETRY_SEND_TIMEOUT` constants. `sink.NewRetry` doesn't call `Validate`, so custom sinks that build `RetryArgs` from user input should call it first. ### v2.1.0 (10 Aug 2026) diff --git a/sink/factory.go b/sink/factory.go index 7eddacd4..937a1e95 100644 --- a/sink/factory.go +++ b/sink/factory.go @@ -115,8 +115,14 @@ func (f *factory) Make(args blip.SinkFactoryArgs) (blip.Sink, error) { retryArgs := RetryArgs{ MonitorId: args.MonitorId, } + // Parse in a fixed order, not map order, so the same invalid options always + // return the same error. sinkOpts := maps.Clone(args.Options) - for k, v := range args.Options { + for _, k := range []string{"buffer-size", "send-timeout", "send-retry-wait"} { + v, ok := sinkOpts[k] + if !ok { + continue + } var err error switch k { case "buffer-size": @@ -125,8 +131,6 @@ func (f *factory) Make(args blip.SinkFactoryArgs) (blip.Sink, error) { retryArgs.SendTimeout, err = parseRetryDuration(v) case "send-retry-wait": retryArgs.SendRetryWait, err = parseRetryDuration(v) - default: - continue // specific sink option } if err != nil { return nil, fmt.Errorf("invalid %s sink option %s: %w", args.SinkName, k, err) diff --git a/sink/factory_test.go b/sink/factory_test.go index e27252a1..5fa0b057 100644 --- a/sink/factory_test.go +++ b/sink/factory_test.go @@ -137,6 +137,22 @@ func TestFactoryMake_InvalidRetryOptions(t *testing.T) { }) } + // With more than one invalid option, the error is always for the first + // option in parse order: buffer-size, send-timeout, send-retry-wait + for i := 0; i < 20; i++ { + _, err := Make(blip.SinkFactoryArgs{ + SinkName: "chronosphere", + MonitorId: "m1", + Options: map[string]string{ + "send-retry-wait": "fast", + "send-timeout": "5", + "buffer-size": "ten", + }, + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "invalid chronosphere sink option buffer-size: ") + } + // Maximum values are valid _, err := Make(blip.SinkFactoryArgs{ SinkName: "chronosphere", diff --git a/sink/retry.go b/sink/retry.go index e3634eb8..4f3f64c6 100644 --- a/sink/retry.go +++ b/sink/retry.go @@ -26,8 +26,8 @@ const ( ) // Retry is a pseudo-sink that provides buffering, serialization, and retry for -// a real sink. The built-in sinks, except "log", use Retry to handle those three -// complexities. +// a real sink. The built-in sinks, except "log" and "noop", use Retry to handle +// those three complexities. // // Retry uses a LIFO queue (a stack) to prioritize sending the latest metrics. // This means that, during a long outage of the real sink, Retry drops the oldest @@ -82,6 +82,9 @@ type retryItem struct { checkpoint any } +// RetryArgs are the arguments for NewRetry. NewRetry doesn't check the ranges +// of the optional values, so callers that build RetryArgs from user input, like +// sink options, should call Validate first. type RetryArgs struct { MonitorId string // required Sink blip.Sink // required @@ -92,7 +95,7 @@ type RetryArgs struct { // Validate returns an error if an optional value is out of range. Zero values // are valid and mean the default. Errors are prefixed with the sink option -// name of the invalid value. NewRetry panics if Validate returns an error. +// name of the invalid value. func (a RetryArgs) Validate() error { if a.BufferSize > MAX_RETRY_BUFFER_SIZE { return fmt.Errorf("buffer-size: %d: must be at most %d", a.BufferSize, MAX_RETRY_BUFFER_SIZE) @@ -109,6 +112,8 @@ func (a RetryArgs) Validate() error { return nil } +// NewRetry returns a Retry sink. It panics if a required arg is missing or +// invalid. It doesn't check optional values; see RetryArgs.Validate. func NewRetry(args RetryArgs) *Retry { // Panic if caller doesn't provide required args if args.MonitorId == "" { @@ -121,9 +126,6 @@ func NewRetry(args RetryArgs) *Retry { if _, ok := args.Sink.(*Delta); ok { panic("RetryArgs.Sink cannot be a Delta sink.") } - if err := args.Validate(); err != nil { - panic("invalid RetryArgs: " + err.Error()) - } // Set defaults if args.BufferSize == 0 { diff --git a/sink/retry_test.go b/sink/retry_test.go index b3d0ca97..5a933e75 100644 --- a/sink/retry_test.go +++ b/sink/retry_test.go @@ -204,10 +204,10 @@ func TestRetryArgsValidate(t *testing.T) { for _, args := range invalid { assert.Error(t, args.Validate(), "%+v", args) - // NewRetry panics on invalid args + // NewRetry doesn't validate, so existing callers keep working args.MonitorId = "m1" args.Sink = sink - assert.Panics(t, func() { NewRetry(args) }, "%+v", args) + assert.NotPanics(t, func() { NewRetry(args) }, "%+v", args) } }