Accept retry options on built-in sinks

docs/content/about/release-notes.md

@@ -57,6 +57,10 @@ 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`. 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. `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)
 

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}

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

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

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

docs/content/sinks/retry.md

@@ -3,17 +3,56 @@ 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: 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
 
 ```yaml
 sinks:
-  retry:
+  datadog:
+    # datadog options
     buffer-size: 60
     send-timeout: 5s
     send-retry-wait: 200ms
 ```
+
+## Options
+
+### `buffer-size`
+
+| | |
+|-|-|
+|**Valid values**|Integer from 1 to 10000|
+|**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) 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`
+
+| | |
+|-|-|
+|**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.
+The wait ends early when `send-timeout` expires, which ends the cycle.

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

sink/factory.go

@@ -5,6 +5,7 @@ package sink
 import (
 	"context"
 	"fmt"
+	"maps"
 	"strconv"
 	"sync"
 	"time"
@@ -109,60 +110,62 @@ 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.
 	retryArgs := RetryArgs{
 		MonitorId: args.MonitorId,
 	}
-	if v, ok := args.Options["buffer-size"]; ok {
-		n, err := strconv.Atoi(v)
-		if err != nil {
-			return nil, err
+	// 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 := range []string{"buffer-size", "send-timeout", "send-retry-wait"} {
+		v, ok := sinkOpts[k]
+		if !ok {
+			continue
 		}
-		if n < 0 {
-			return nil, fmt.Errorf("invalid retry buffer-size: %d: must be greater than zero", n)
+		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)
 		}
-		retryArgs.BufferSize = uint(n)
-	}
-	if v, ok := args.Options["send-timeout"]; ok {
-		d, err := time.ParseDuration(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 := args.Options["send-retry-wait"]; ok {
-		d, err := time.ParseDuration(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
 	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)
 	}
@@ -180,3 +183,29 @@ func (f *factory) Make(args blip.SinkFactoryArgs) (blip.Sink, error) {
 		return NewRetry(retryArgs), nil
 	}
 }
+
+// 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, err
+	}
+	if d <= 0 {
+		return 0, fmt.Errorf("%s: must be greater than zero", v)
+	}
+	return d, nil
+}