fix(notifications): Remove deprecated retryingSink with backoffSink and add backward compatibility (ba2dc330) · Commits · GitLab.org / container-registry

docs/configuration.md

+29 −5

Original line number	Diff line number	Diff line
		@@ -290,7 +290,7 @@ notifications:
		url: https://my.listener.com/event
		headers: <http.Header>
		timeout: 1s
		threshold: 10
		threshold: 10 # DEPRECATED: will be transparently translated into maxretries, use maxretries for full control
		maxretries: 5
		backoff: 1s
		ignoredmediatypes:
		@@ -1272,7 +1272,8 @@ notifications:
		url: https://my.listener.com/event
		headers: <http.Header>
		timeout: 1s
		threshold: 10
		threshold: 10 # DEPRECATED: will be transparently translated into maxretries, use maxretries for full control
		maxretries: 5
		backoff: 1s
		ignoredmediatypes:
		- application/octet-stream
		@@ -1302,14 +1303,37 @@ accept event notifications.
		\| `url` \| yes \| The URL to which events should be published. \|
		\| `headers` \| yes \| A list of static headers to add to each request. Each header's name is a key beneath `headers`, and each value is a list of payloads for that header name. Values must always be lists. \|
		\| `timeout` \| yes \| A value for the HTTP timeout. A positive integer and an optional suffix indicating the unit of time, which may be `ns`, `us`, `ms`, `s`, `m`, or `h`. If you omit the unit of time, `ns` is used. \|
		\| `threshold` \| yes \| DEPRECATED: use maxretries instead, more details [here](https://gitlab.com/gitlab-org/container-registry/-/issues/1243). An integer specifying how long to wait before backing off a failure. \|
		\| `maxretries` \| no \| An integer specifying the maximum number of times to retry sending a failed event. `threshold` is ignored when defining this field. \|
		\| `backoff` \| yes \| How long the system backs off before retrying after a failure. A positive integer and an optional suffix indicating the unit of time, which may be `ns`, `us`, `ms`, `s`, `m`, or `h`. If you omit the unit of time, `ns` is used. \|
		\| `threshold` \| no \| DEPRECATED: This parameter is deprecated in favor of `maxretries`. When `maxretries` is not set, `threshold` will be automatically translated to an equivalent `maxretries` value based on the configured `backoff` time. The translation uses a time window calculation to determine the appropriate number of retries. See [here](#migration-from-threshold-to-maxretries) for migration details. \|
		\| `maxretries` \| no \| An integer specifying the maximum number of times to retry sending a failed event before dropping it. When this field is defined, it takes precedence over `threshold`. If neither `threshold` nor `maxretries` is specified, defaults to 10. \|
		\| `backoff` \| yes \| The base backoff duration between retry attempts. Used as the initial interval in an exponential backoff strategy. A positive integer and an optional suffix indicating the unit of time, which may be `ns`, `us`, `ms`, `s`, `m`, or `h`. If you omit the unit of time, `ns` is used. \|
		\| `ignoredmediatypes`\|no\| A list of target media types to ignore. Events with these target media types are not published to the endpoint. \|
		\| `ignore` \|no\| Events with these mediatypes or actions are not published to the endpoint. \|
		\| `queuepurgetimeout` \| no \| The maximum amount of time registry tries to sent unsent notifications in the buffer after it received SIGINT. A positive integer and an optional suffix indicating the unit of time, which may be `ns`, `us`, `ms`, `s`, `m`, or `h`. If you omit the unit of time, `ns` is used. The default is 5 seconds. The zero value is always defaulted to 5 seconds. User may set a very low value (e.g. 1ns) to simulate no-wait if desired. \|
		\| `queuesizelimit` \| no \| The maximum size of the notifications queue with events pending for sending. Once the queue gets full, the events are dropped. The default is 3000. \|

		#### Migration from `threshold` to `maxretries`

		The `threshold` parameter has been deprecated in favor of `maxretries` to better align with exponential backoff retry strategies.
		The registry automatically translates `threshold` values to equivalent `maxretries` values when `maxretries` is not explicitly set.

		Translation behavior:

		- If only `maxretries` is set: Uses the specified value directly
		- If only `threshold` is set: Automatically translates to `maxretries` based on the `backoff` duration
		- If neither is set: Defaults to `maxretries: 10`

		Translation formula:
		The translation calculates how many retries can fit within a time window of `120 * backoff_duration` seconds, using exponential backoff with a multiplier of 1.5.
		The resulting `maxretries` will be at least equal to the original `threshold` value to maintain backward compatibility.

		Examples:

		- `threshold: 10, backoff: 1s` → `maxretries: 10`
		- `threshold: 5, backoff: 2s` → `maxretries: 10`
		- `threshold: 15, backoff: 1s` → `maxretries: 15` (maintains minimum threshold)

		Recommendation: Update your configuration to use `maxretries` directly to avoid automatic translation and have explicit control over retry behavior.

		#### `ignore`

		\| Parameter \| Required \| Description \|

notifications/endpoint.go

+79 −18

Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ import (
		"sync/atomic"
		"time"

		"github.com/cenkalti/backoff/v4"
		log "github.com/sirupsen/logrus"

		"github.com/docker/distribution/configuration"
		@@ -48,20 +49,91 @@ type EndpointConfig struct {
		QueueSizeLimit int
		}

		// translateBackoffParams translates old backoff parameters (threshold, backoff)
		// into new parameters (maxretries, backoff) based on a time window calculation.
		//
		// The time window scale factor (120s for 1s backoff) was arbitrarily chosen based on:
		// - GitLab's production configuration for their large SaaS installation (1s backoff, 10 max retries)
		// - The necessity to limit retry attempts to avoid head-of-line blocking
		// - The need to prevent infinite retries which would ultimately lead to dropped notifications
		// anyway, as the registry will start dropping events to protect itself from RAM exhaustion
		//
		// Parameters:
		// - threshold: number of immediate retries in the old system (used as minimum for maxretries)
		// - backoff: base backoff duration in the old system, becomes InitialInterval in the new system
		//
		// Returns:
		// - maxretries: calculated number of maximum retries for the new system (at least threshold)
		func translateBackoffParams(threshold int, backoffTime time.Duration) int {
		// Calculate time window: 120s for 1s backoff, scales linearly
		// timeWindow = 120 * backoff_in_seconds
		timeWindow := 120 * backoffTime

		// Cap time window at DefaultMaxElapsedTime
		if timeWindow > backoff.DefaultMaxElapsedTime {
		timeWindow = backoff.DefaultMaxElapsedTime
		}

		// Calculate maxretries using reverse calculation of backoff algorithm
		// Assumes RandomizationFactor = 0 for predictable calculations
		currentInterval := backoffTime
		cumulativeTime := time.Duration(0)
		retryCount := 0

		// Simulate the backoff algorithm until we exceed the time window
		for cumulativeTime < timeWindow {
		// Add the delay to cumulative time
		cumulativeTime += currentInterval

		// If we've exceeded the time window, don't count this retry
		if cumulativeTime > timeWindow {
		break
		}

		retryCount++

		// Calculate next interval using the same logic as the library
		currentInterval = time.Duration(float64(currentInterval) * backoff.DefaultMultiplier)
		if currentInterval > backoff.DefaultMaxInterval {
		currentInterval = backoff.DefaultMaxInterval
		}
		}

		if retryCount < threshold {
		return threshold
		}
		return retryCount
		}

		// defaults set any zero-valued fields to a reasonable default.
		func (ec *EndpointConfig) defaults() {
		if ec.Timeout <= 0 {
		ec.Timeout = time.Second
		}

		if ec.Threshold <= 0 {
		ec.Threshold = 10
		}

		if ec.Backoff <= 0 {
		ec.Backoff = time.Second
		}

		if ec.MaxRetries == 0 {
		// NOTE(prozlach): in order to keep old behavior intact if possible,
		// if maxRetries is not defined, we check if threshold is. If not - we
		// assume defaults for maxRetries. If yes - we translate threshold to
		// maxRetries.
		if ec.Threshold <= 0 {
		log.Info("defaulting maxRetries parameter to 10")
		ec.MaxRetries = 10
		} else {
		ec.MaxRetries = translateBackoffParams(ec.Threshold, ec.Backoff)
		log.Warnf(
		"notifications `threshold` is deprecated, please use `maxretries` instead. "+
		"Value `threshold` of %d has been converted to `maxretries` of %d. "+
		"See https://gitlab.com/gitlab-org/container-registry/-/issues/1243 for more details.",
		ec.Threshold, ec.MaxRetries,
		)
		}
		}

		if ec.QueuePurgeTimeout <= 0 {
		ec.QueuePurgeTimeout = DefaultQueuePurgeTimeout
		}
		@@ -99,23 +171,12 @@ func NewEndpoint(name, url string, config EndpointConfig) *Endpoint {

		// Configures the inmemory queue, retry, http pipeline.
		endpoint.Sink = newHTTPSink(
		endpoint.url, endpoint.Timeout, endpoint.Headers,
		endpoint.Transport, endpoint.metrics.httpStatusListener())
		endpoint.url, endpoint.Timeout, endpoint.Headers, endpoint.Transport, endpoint.metrics.httpStatusListener(),
		)

		// TODO: threshold has been deprecated and we should use MaxRetries with backoffSink instead.
		// Remove this check along with https://gitlab.com/gitlab-org/container-registry/-/issues/1244.
		if endpoint.MaxRetries != 0 {
		endpoint.Sink = newBackoffSink(
		endpoint.Sink, endpoint.Backoff, endpoint.MaxRetries,
		endpoint.metrics.deliveryListener(),
		endpoint.Sink, endpoint.Backoff, endpoint.MaxRetries, endpoint.metrics.deliveryListener(),
		)
		} else {
		log.Warn("notifications `threshold` is deprecated, use maxretries instead. See https://gitlab.com/gitlab-org/container-registry/-/issues/1243.")
		endpoint.Sink = newRetryingSink(
		endpoint.Sink, endpoint.Threshold, endpoint.Backoff,
		endpoint.metrics.deliveryListener(),
		)
		}

		endpoint.Sink = newEventQueue(
		endpoint.Sink,

notifications/endpoint_test.go

0 → 100644

+87 −0

Original line number	Diff line number	Diff line
		package notifications

		import (
		"testing"
		"time"

		"github.com/stretchr/testify/require"
		)

		func TestTranslateBackoffParams(t *testing.T) {
		tests := []struct {
		name string
		threshold int
		backoffTime time.Duration
		expectedRetries int
		}{
		{
		name: "Zero threshold",
		threshold: 0,
		backoffTime: 1 * time.Second,
		expectedRetries: 10,
		},
		{
		name: "1s backoff with low threshold",
		threshold: 3,
		backoffTime: 1 * time.Second,
		expectedRetries: 10,
		},
		{
		name: "1s backoff with high threshold",
		threshold: 15,
		backoffTime: 1 * time.Second,
		expectedRetries: 15,
		},
		{
		name: "Very small backoff",
		threshold: 5,
		backoffTime: 10 * time.Millisecond,
		expectedRetries: 10,
		},
		{
		name: "100ms backoff",
		threshold: 3,
		backoffTime: 100 * time.Millisecond,
		expectedRetries: 10,
		},
		{
		name: "500ms backoff",
		threshold: 5,
		backoffTime: 500 * time.Millisecond,
		expectedRetries: 10,
		},
		{
		name: "2s backoff",
		threshold: 5,
		backoffTime: 2 * time.Second,
		expectedRetries: 10,
		},
		{
		name: "3s backoff",
		threshold: 5,
		backoffTime: 3 * time.Second,
		expectedRetries: 11,
		},
		{
		name: "10s backoff",
		threshold: 2,
		backoffTime: 10 * time.Second,
		expectedRetries: 17,
		},
		{
		name: "30s backoff - hits MaxElapsedTime cap",
		threshold: 1,
		backoffTime: 30 * time.Second,
		expectedRetries: 15,
		},
		}

		for _, tc := range tests {
		t.Run(tc.name, func(tt *testing.T) {
		gotRetries := translateBackoffParams(tc.threshold, tc.backoffTime)

		require.Equal(tt, tc.expectedRetries, gotRetries)
		require.GreaterOrEqual(tt, gotRetries, tc.threshold)
		})
		}
		}

notifications/sinks.go

+0 −181

Original line number	Diff line number	Diff line
		@@ -2,7 +2,6 @@ package notifications

		import (
		"container/list"
		"errors"
		"fmt"
		"sync"
		"time"
		@@ -458,186 +457,6 @@ type deliveryListener interface {
		eventLost(retriesCount int64)
		}

		// retryingSink retries the write until success or an ErrSinkClosed is
		// returned. Underlying sink must have p > 0 of succeeding or the sink will
		// block. Internally, it is a circuit breaker retries to manage reset.
		// Concurrent calls to a retrying sink are serialized through the sink,
		// meaning that if one is in-flight, another will not proceed.
		type retryingSink struct {
		sink Sink

		doneCh chan struct{}
		eventsCh chan *Event
		errCh chan error

		wg *sync.WaitGroup

		// circuit breaker heuristics
		failures struct {
		threshold int
		backoff time.Duration // time after which we retry after failure.
		}

		listeners []deliveryListener
		}

		// newRetryingSink returns a sink that will retry writes to a sink, backing
		// off on failure. Parameters threshold and backoff adjust the behavior of the
		// circuit breaker.
		func newRetryingSink(
		sink Sink,
		threshold int,
		backoff time.Duration,
		listeners ...deliveryListener,
		) *retryingSink {
		rs := &retryingSink{
		sink: sink,

		doneCh: make(chan struct{}),
		eventsCh: make(chan *Event),
		errCh: make(chan error),

		wg: new(sync.WaitGroup),

		listeners: listeners,
		}
		rs.failures.threshold = threshold
		rs.failures.backoff = backoff

		rs.wg.Add(1)
		go rs.run()

		return rs
		}

		func (rs *retryingSink) run() {
		defer rs.wg.Done()

		main:
		for {
		// nolint: revive // max-control-nesting
		select {
		case <-rs.doneCh:
		return
		case event := <-rs.eventsCh:
		var retriesCount int64 = 0

		for failuresCount := 0; failuresCount < rs.failures.threshold; failuresCount++ {
		select {
		case <-rs.doneCh:
		rs.errCh <- ErrSinkClosed
		return
		default:
		}

		err := rs.sink.Write(event)

		// Event sent successfully, fetch next event from channel:
		if err == nil {
		for _, listener := range rs.listeners {
		listener.eventDelivered(retriesCount)
		}

		rs.errCh <- nil
		continue main
		}

		// Underlying sink is closed, let's wrap up:
		if errors.Is(err, ErrSinkClosed) {
		rs.errCh <- ErrSinkClosed
		return
		}

		log.WithError(err).
		WithField("failure_count", failuresCount).
		Error("retryingsink: error writing event, retrying")
		retriesCount++
		}

		log.WithField("sink", rs.sink).
		Warnf("encountered too many errors when writing to sink, enabling backoff")

		for {
		// NOTE(prozlach): We can't use Ticker here as the write()
		// operation may take longer than the backoff period and this
		// would result in triggering new write imediatelly after the
		// previous one.
		timer := time.NewTimer(rs.failures.backoff)

		select {
		case <-rs.doneCh:
		rs.errCh <- ErrSinkClosed
		timer.Stop()
		return
		case lastFailureTime := <-timer.C:
		err := rs.sink.Write(event)

		// Event sent successfully, fetch next event from channel:
		if err == nil {
		for _, listener := range rs.listeners {
		listener.eventDelivered(retriesCount)
		}

		rs.errCh <- nil
		continue main
		}

		// Underlying sink is closed, let's wrap up:
		if errors.Is(err, ErrSinkClosed) {
		rs.errCh <- ErrSinkClosed
		return
		}

		log.WithError(err).
		WithField("next_retry_time", lastFailureTime.Add(rs.failures.backoff).String()).
		Error("retryingsink: error writing event, backing off")
		retriesCount++
		}
		}
		}
		}
		}

		// Write attempts to flush the event to the downstream sink until it succeeds
		// or the sink is closed.
		func (rs retryingSink) Write(event Event) error {
		// NOTE(prozlach): avoid a racy situation when both channels are "ready",
		// and make sure that closing the Sink takes priority:
		select {
		case <-rs.doneCh:
		return ErrSinkClosed
		default:
		select {
		case rs.eventsCh <- event:
		return <-rs.errCh
		case <-rs.doneCh:
		return ErrSinkClosed
		}
		}
		}

		// Close closes the sink and the underlying sink.
		func (rs *retryingSink) Close() error {
		log.Infof("retryingSink: closing")
		select {
		case <-rs.doneCh:
		return fmt.Errorf("retryingSink: already closed")
		default:
		close(rs.doneCh)
		}

		rs.wg.Wait()
		err := rs.sink.Close()

		// NOTE(prozlach): not stricly necessary, just a basic hygiene
		close(rs.eventsCh)
		close(rs.errCh)

		log.Debugf("retryingSink: closed")

		return err
		}

		// backoffSink attempts to write an event to the given sink.
		// It will retry up to a number of maxretries as defined in the configuration
		// and will drop the event after it reaches the number of retries.

notifications/sinks_test.go

+0 −165

Original line number	Diff line number	Diff line
		@@ -96,102 +96,6 @@ func TestEventQueue(t *testing.T) {
		require.Zero(t, smetrics.pending.Load(), "unexpected egress count")
		}

		func TestRetryingSinkWithDeliveryListener(t *testing.T) {
		t.Run("successful delivery on first attempt", func(tt *testing.T) {
		metrics := newSafeMetrics(tt.Name())
		deliveryListener := metrics.deliveryListener()
		ts := &testSink{}

		s := newRetryingSink(ts, 3, 10*time.Millisecond, deliveryListener)
		defer s.Close()

		event := createTestEvent("push", "blob")
		require.NoError(tt, s.Write(&event))

		assert.EqualValues(tt, 1, metrics.delivered.Load())
		assert.Zero(tt, metrics.retries.Load())

		ts.mu.Lock()
		assert.Len(tt, ts.events, 1, "event should be in test sink")
		ts.mu.Unlock()
		})

		t.Run("successful delivery after retries", func(tt *testing.T) {
		metrics := newSafeMetrics(tt.Name())
		deliveryListener := metrics.deliveryListener()

		failing := &failingSink{
		failBelowCount: 2,
		Sink: &testSink{},
		}

		s := newRetryingSink(failing, 3, 10*time.Millisecond, deliveryListener)
		defer s.Close()

		event := createTestEvent("push", "blob")
		require.NoError(tt, s.Write(&event))

		assert.EqualValues(tt, 1, metrics.delivered.Load())
		assert.Positive(tt, metrics.retries.Load())
		})

		t.Run("delivery with backoff period", func(tt *testing.T) {
		metrics := newSafeMetrics(tt.Name())
		deliveryListener := metrics.deliveryListener()

		failing := &failingSink{
		failBelowCount: 5,
		Sink: &testSink{},
		}

		s := newRetryingSink(failing, 3, 50*time.Millisecond, deliveryListener)
		defer s.Close()

		event := createTestEvent("push", "blob")
		start := time.Now()
		require.NoError(tt, s.Write(&event))
		elapsed := time.Since(start)

		assert.Greater(tt, elapsed, 50*time.Millisecond)
		assert.EqualValues(tt, 1, metrics.delivered.Load())
		assert.Positive(tt, metrics.retries.Load())
		})

		t.Run("concurrent writes", func(tt *testing.T) {
		metrics := newSafeMetrics(tt.Name())
		deliveryListener := metrics.deliveryListener()

		// Flaky sink that fails 30% of the time
		flaky := &flakySink{
		rate: 0.3,
		Sink: &testSink{},
		}

		s := newRetryingSink(flaky, 5, 10*time.Millisecond, deliveryListener)
		defer s.Close()

		const nEvents = 1000
		var wg sync.WaitGroup

		for i := 0; i < nEvents; i++ {
		wg.Add(1)
		go func(i int) {
		defer wg.Done()
		event := createTestEvent("push", fmt.Sprintf("blob-%d", i))
		assert.NoError(tt, s.Write(&event))
		}(i)
		}

		wg.Wait()

		// All events should be delivered (retryingSink retries indefinitely)
		assert.EqualValues(tt, nEvents, metrics.delivered.Load())

		// Should have some retries due to 30% failure rate
		assert.Positive(tt, metrics.retries.Load())
		})
		}

		func TestIgnoredSink(t *testing.T) {
		blob := createTestEvent("push", "blob")
		manifest := createTestEvent("pull", "manifest")
		@@ -229,36 +133,6 @@ func TestIgnoredSink(t *testing.T) {
		}
		}

		func TestRetryingSink(t *testing.T) {
		// Make a sync that fails most of the time, ensuring that all the events
		// make it through.
		var ts testSink
		flaky := &flakySink{
		rate: 0.9, // 90% failure rate
		Sink: &ts,
		}
		s := newRetryingSink(flaky, 3, 10*time.Millisecond)

		event := createTestEvent("push", "blob")
		errCh := make(chan error, 10)
		for i := 1; i <= 10; i++ {
		go func() {
		errCh <- s.Write(&event)
		}()
		}

		for i := 1; i <= 10; i++ {
		require.NoErrorf(t, <-errCh, "error writing event %d", i)
		}

		checkClose(t, s)

		ts.mu.Lock()
		defer ts.mu.Unlock()

		require.Len(t, ts.events, 10, "events not propagated")
		}

		func TestBackoffSink(t *testing.T) {
		tcs := map[string]struct {
		maxRetries int
		@@ -435,45 +309,6 @@ func TestBackoffSinkWithDeliveryListener(t *testing.T) {
		}

		func TestConcurrentDeliveryReporting(t *testing.T) {
		t.Run("retryingSink concurrent writes", func(t *testing.T) {
		metrics := newSafeMetrics(t.Name())
		deliveryListener := metrics.deliveryListener()

		// Create a flaky sink that fails 30% of the time
		flaky := &flakySink{
		rate: 0.3,
		Sink: &testSink{},
		}

		s := newRetryingSink(flaky, 5, 5*time.Millisecond, deliveryListener)
		defer s.Close()

		const nGoroutines = 10
		const nEventsPerGoroutine = 10

		var wg sync.WaitGroup
		for i := 0; i < nGoroutines; i++ {
		wg.Add(1)
		go func(goroutineID int) {
		defer wg.Done()
		for j := 0; j < nEventsPerGoroutine; j++ {
		event := createTestEvent("push", fmt.Sprintf("blob-%d-%d", goroutineID, j))
		assert.NoError(t, s.Write(&event))
		}
		}(i)
		}

		wg.Wait()

		totalEvents := nGoroutines * nEventsPerGoroutine
		// All events should be delivered (retryingSink retries indefinitely)
		require.EqualValues(t, totalEvents, metrics.delivered.Load())
		require.Zero(t, metrics.lost.Load())

		// Should have some retries due to 30% failure rate
		require.Positive(t, metrics.retries.Load())
		})

		t.Run("backoffSink concurrent writes", func(t *testing.T) {
		metrics := newSafeMetrics(t.Name())
		deliveryListener := metrics.deliveryListener()