Add Progressive failover mode to the failover connector

.chloggen/failoverconnector-multiple-mode.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: failoverconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Refactors failoverconnector to create failover mode interface.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [42657]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

connector/failoverconnector/config.go

@@ -26,6 +26,9 @@ type Config struct {
 	// level is considered unhealthy
 	PipelinePriority [][]pipeline.ID `mapstructure:"priority_levels"`
 
+	// FailoverMode determines the failover strategy to use. Options: "standard", "progressive"
+	FailoverMode FailoverMode `mapstructure:"failover_mode"`
+
 	// RetryInterval is the frequency at which the pipeline levels will attempt to recover by going over
 	// all levels below the current
 	RetryInterval time.Duration `mapstructure:"retry_interval"`

connector/failoverconnector/config_test.go

@@ -33,6 +33,7 @@ func TestLoadConfig(t *testing.T) {
 						pipeline.NewIDWithName(pipeline.SignalTraces, ""),
 					},
 				},
+				FailoverMode:  FailoverModeStandard,
 				RetryInterval: 10 * time.Minute,
 			},
 		},
@@ -55,6 +56,7 @@ func TestLoadConfig(t *testing.T) {
 						pipeline.NewIDWithName(pipeline.SignalTraces, "fourth"),
 					},
 				},
+				FailoverMode:  FailoverModeStandard,
 				RetryInterval: 5 * time.Minute,
 			},
 		},

connector/failoverconnector/factory.go

@@ -29,6 +29,7 @@ func NewFactory() connector.Factory {
 func createDefaultConfig() component.Config {
 	return &Config{
 		QueueSettings: exporterhelper.NewDefaultQueueConfig(),
+		FailoverMode:  FailoverModeStandard,
 		RetryInterval: 10 * time.Minute,
 		RetryGap:      0,
 		MaxRetries:    0,

connector/failoverconnector/failover.go

@@ -7,8 +7,6 @@ import (
 	"errors"
 
 	"go.opentelemetry.io/collector/pipeline"
-
-	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/failoverconnector/internal/state"
 )
 
 var (
@@ -21,51 +19,15 @@ type consumerProvider[C any] func(...pipeline.ID) (C, error)
 // baseFailoverRouter provides the common infrastructure for failover routing
 type baseFailoverRouter[C any] struct {
 	cfg       *Config
-	pS        *state.PipelineSelector
 	consumers []C
-
-	errTryLock  *state.TryLock
-	notifyRetry chan struct{}
-	done        chan struct{}
-}
-
-// getCurrentConsumer returns the consumer for the current healthy level
-func (f *baseFailoverRouter[C]) getCurrentConsumer() (C, int) {
-	var nilConsumer C
-	pl := f.pS.CurrentPipeline()
-	if pl >= len(f.cfg.PipelinePriority) {
-		return nilConsumer, pl
-	}
-	return f.consumers[pl], pl
 }
 
 // getConsumerAtIndex returns the consumer at a specific index
 func (f *baseFailoverRouter[C]) getConsumerAtIndex(idx int) C {
 	return f.consumers[idx]
 }
 
-// reportConsumerError ensures only one consumer is reporting an error at a time to avoid multiple failovers
-func (f *baseFailoverRouter[C]) reportConsumerError(idx int) {
-	f.errTryLock.TryExecute(f.pS.HandleError, idx)
-}
-
-func (f *baseFailoverRouter[C]) Shutdown() {
-	select {
-	case <-f.done:
-	default:
-		close(f.done)
-	}
-}
-
 func newBaseFailoverRouter[C any](provider consumerProvider[C], cfg *Config) (*baseFailoverRouter[C], error) {
-	done := make(chan struct{})
-	notifyRetry := make(chan struct{}, 1)
-	pSConstants := state.PSConstants{
-		RetryInterval: cfg.RetryInterval,
-		RetryGap:      cfg.RetryGap,
-		MaxRetries:    cfg.MaxRetries,
-	}
-
 	consumers := make([]C, 0)
 	for _, pipelines := range cfg.PipelinePriority {
 		baseConsumer, err := provider(pipelines...)
@@ -75,14 +37,9 @@ func newBaseFailoverRouter[C any](provider consumerProvider[C], cfg *Config) (*b
 		consumers = append(consumers, baseConsumer)
 	}
 
-	selector := state.NewPipelineSelector(notifyRetry, done, pSConstants)
 	return &baseFailoverRouter[C]{
-		consumers:   consumers,
-		cfg:         cfg,
-		pS:          selector,
-		errTryLock:  state.NewTryLock(),
-		done:        done,
-		notifyRetry: notifyRetry,
+		consumers: consumers,
+		cfg:       cfg,
 	}, nil
 }
 
@@ -91,14 +48,6 @@ func (f *baseFailoverRouter[C]) ModifyConsumerAtIndex(idx int, c C) {
 	f.consumers[idx] = c
 }
 
-func (f *baseFailoverRouter[C]) TestGetCurrentConsumerIndex() int {
-	return f.pS.CurrentPipeline()
-}
-
-func (f *baseFailoverRouter[C]) TestSetStableConsumerIndex(idx int) {
-	f.pS.TestSetCurrentPipeline(idx)
-}
-
 func (f *baseFailoverRouter[C]) TestGetConsumerAtIndex(idx int) C {
 	return f.consumers[idx]
 }

connector/failoverconnector/failover_test.go

@@ -43,6 +43,7 @@ func TestFailoverRecovery(t *testing.T) {
 
 	failoverConnector := conn.(*tracesFailover)
 	tRouter := failoverConnector.failover
+	strategy := tRouter.strategy.(*standardTracesStrategy)
 
 	tr := sampleTrace()
 
@@ -54,13 +55,13 @@ func TestFailoverRecovery(t *testing.T) {
 		defer func() {
 			resetConsumers(tRouter, &sinkFirst, &sinkSecond, &sinkThird, &sinkFourth)
 		}()
-		failoverConnector.failover.ModifyConsumerAtIndex(0, consumertest.NewErr(errTracesConsumer))
 
+		tRouter.ModifyConsumerAtIndex(0, consumertest.NewErr(errTracesConsumer))
 		require.NoError(t, conn.ConsumeTraces(t.Context(), tr))
-		idx := failoverConnector.failover.TestGetCurrentConsumerIndex()
+		idx := strategy.TestGetCurrentConsumerIndex()
 		require.Equal(t, 1, idx)
 
-		failoverConnector.failover.ModifyConsumerAtIndex(0, &sinkFirst)
+		tRouter.ModifyConsumerAtIndex(0, &sinkFirst)
 
 		require.Eventually(t, func() bool {
 			return consumeTracesAndCheckStable(tRouter, 0, tr)
@@ -71,21 +72,22 @@ func TestFailoverRecovery(t *testing.T) {
 		defer func() {
 			resetConsumers(tRouter, &sinkFirst, &sinkSecond, &sinkThird, &sinkFourth)
 		}()
-		failoverConnector.failover.ModifyConsumerAtIndex(0, consumertest.NewErr(errTracesConsumer))
-		failoverConnector.failover.ModifyConsumerAtIndex(1, consumertest.NewErr(errTracesConsumer))
+
+		tRouter.ModifyConsumerAtIndex(0, consumertest.NewErr(errTracesConsumer))
+		tRouter.ModifyConsumerAtIndex(1, consumertest.NewErr(errTracesConsumer))
 
 		require.Eventually(t, func() bool {
 			return consumeTracesAndCheckStable(tRouter, 2, tr)
 		}, 3*time.Second, 5*time.Millisecond)
 
 		// Simulate recovery of exporter
-		failoverConnector.failover.ModifyConsumerAtIndex(1, &sinkSecond)
+		tRouter.ModifyConsumerAtIndex(1, &sinkSecond)
 
 		require.Eventually(t, func() bool {
 			return consumeTracesAndCheckStable(tRouter, 1, tr)
 		}, 3*time.Second, 5*time.Millisecond)
 
-		failoverConnector.failover.ModifyConsumerAtIndex(0, &sinkFirst)
+		tRouter.ModifyConsumerAtIndex(0, &sinkFirst)
 
 		require.Eventually(t, func() bool {
 			return consumeTracesAndCheckStable(tRouter, 0, tr)
@@ -96,34 +98,35 @@ func TestFailoverRecovery(t *testing.T) {
 		defer func() {
 			resetConsumers(tRouter, &sinkFirst, &sinkSecond, &sinkThird, &sinkFourth)
 		}()
-		failoverConnector.failover.ModifyConsumerAtIndex(0, consumertest.NewErr(errTracesConsumer))
-		failoverConnector.failover.ModifyConsumerAtIndex(1, consumertest.NewErr(errTracesConsumer))
+
+		tRouter.ModifyConsumerAtIndex(0, consumertest.NewErr(errTracesConsumer))
+		tRouter.ModifyConsumerAtIndex(1, consumertest.NewErr(errTracesConsumer))
 
 		require.Eventually(t, func() bool {
 			return consumeTracesAndCheckStable(tRouter, 2, tr)
 		}, 3*time.Second, 5*time.Millisecond)
 
 		// Simulate recovery of exporter
-		failoverConnector.failover.ModifyConsumerAtIndex(1, &sinkSecond)
+		tRouter.ModifyConsumerAtIndex(1, &sinkSecond)
 
 		require.Eventually(t, func() bool {
 			return consumeTracesAndCheckStable(tRouter, 1, tr)
 		}, 3*time.Second, 5*time.Millisecond)
 
-		failoverConnector.failover.ModifyConsumerAtIndex(2, consumertest.NewErr(errTracesConsumer))
-		failoverConnector.failover.ModifyConsumerAtIndex(1, consumertest.NewErr(errTracesConsumer))
+		tRouter.ModifyConsumerAtIndex(2, consumertest.NewErr(errTracesConsumer))
+		tRouter.ModifyConsumerAtIndex(1, consumertest.NewErr(errTracesConsumer))
 
 		require.Eventually(t, func() bool {
 			return consumeTracesAndCheckStable(tRouter, 3, tr)
 		}, 3*time.Second, 5*time.Millisecond)
 
-		failoverConnector.failover.ModifyConsumerAtIndex(2, &sinkThird)
+		tRouter.ModifyConsumerAtIndex(2, &sinkThird)
 
 		require.Eventually(t, func() bool {
 			return consumeTracesAndCheckStable(tRouter, 2, tr)
 		}, 3*time.Second, 5*time.Millisecond)
 
-		failoverConnector.failover.ModifyConsumerAtIndex(0, &sinkThird)
+		tRouter.ModifyConsumerAtIndex(0, &sinkThird)
 
 		require.Eventually(t, func() bool {
 			return consumeTracesAndCheckStable(tRouter, 0, tr)
@@ -135,5 +138,6 @@ func resetConsumers(router *tracesRouter, consumers ...consumer.Traces) {
 	for i, sink := range consumers {
 		router.ModifyConsumerAtIndex(i, sink)
 	}
-	router.TestSetStableConsumerIndex(0)
+	strategy := router.strategy.(*standardTracesStrategy)
+	strategy.TestSetStableConsumerIndex(0)
 }

connector/failoverconnector/logs.go

@@ -15,58 +15,28 @@ import (
 
 type logsRouter struct {
 	*baseFailoverRouter[consumer.Logs]
+	strategy LogsFailoverStrategy
 }
 
 func newLogsRouter(provider consumerProvider[consumer.Logs], cfg *Config) (*logsRouter, error) {
 	failover, err := newBaseFailoverRouter(provider, cfg)
 	if err != nil {
 		return nil, err
 	}
-	return &logsRouter{baseFailoverRouter: failover}, nil
-}
-
-// Consume is the logs-specific consumption method
-func (f *logsRouter) Consume(ctx context.Context, ld plog.Logs) error {
-	select {
-	case <-f.notifyRetry:
-		if !f.sampleRetryConsumers(ctx, ld) {
-			return f.consumeByHealthyPipeline(ctx, ld)
-		}
-		return nil
-	default:
-		return f.consumeByHealthyPipeline(ctx, ld)
-	}
-}
 
-// consumeByHealthyPipeline will consume the logs by the current healthy level
-func (f *logsRouter) consumeByHealthyPipeline(ctx context.Context, ld plog.Logs) error {
-	for {
-		tc, idx := f.getCurrentConsumer()
-		if idx >= len(f.cfg.PipelinePriority) {
-			return errNoValidPipeline
-		}
+	// Create the appropriate strategy based on the failover mode
+	factory := GetFailoverStrategyFactory(cfg.FailoverMode)
+	strategy := factory.CreateLogsStrategy(failover)
 
-		if err := tc.ConsumeLogs(ctx, ld); err != nil {
-			f.reportConsumerError(idx)
-			continue
-		}
-
-		return nil
-	}
+	return &logsRouter{
+		baseFailoverRouter: failover,
+		strategy:           strategy,
+	}, nil
 }
 
-// sampleRetryConsumers iterates through all unhealthy consumers to re-establish a healthy connection
-func (f *logsRouter) sampleRetryConsumers(ctx context.Context, ld plog.Logs) bool {
-	stableIndex := f.pS.CurrentPipeline()
-	for i := 0; i < stableIndex; i++ {
-		consumer := f.getConsumerAtIndex(i)
-		err := consumer.ConsumeLogs(ctx, ld)
-		if err == nil {
-			f.pS.ResetHealthyPipeline(i)
-			return true
-		}
-	}
-	return false
+// Consume is the logs-specific consumption method
+func (f *logsRouter) Consume(ctx context.Context, ld plog.Logs) error {
+	return f.strategy.ConsumeLogs(ctx, ld)
 }
 
 type logsFailover struct {
@@ -88,8 +58,8 @@ func (f *logsFailover) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 }
 
 func (f *logsFailover) Shutdown(context.Context) error {
-	if f.failover != nil {
-		f.failover.Shutdown()
+	if f.failover != nil && f.failover.strategy != nil {
+		f.failover.strategy.Shutdown()
 	}
 	return nil
 }

connector/failoverconnector/logs_test.go

@@ -85,15 +85,18 @@ func TestLogsWithValidFailover(t *testing.T) {
 	require.NoError(t, err)
 
 	failoverConnector := conn.(*logsFailover)
-	failoverConnector.failover.ModifyConsumerAtIndex(0, consumertest.NewErr(errLogsConsumer))
+	lRouter := failoverConnector.failover
+	strategy := lRouter.strategy.(*standardLogsStrategy)
+
+	strategy.router.ModifyConsumerAtIndex(0, consumertest.NewErr(errLogsConsumer))
 	defer func() {
 		assert.NoError(t, failoverConnector.Shutdown(t.Context()))
 	}()
 
 	ld := sampleLog()
 
 	require.Eventually(t, func() bool {
-		return consumeLogsAndCheckStable(failoverConnector, 1, ld)
+		return consumeLogsAndCheckStable(lRouter, 1, ld)
 	}, 3*time.Second, 5*time.Millisecond)
 }
 
@@ -169,9 +172,10 @@ func TestLogsWithQueue(t *testing.T) {
 	assert.NoError(t, conn.ConsumeLogs(t.Context(), ld))
 }
 
-func consumeLogsAndCheckStable(conn *logsFailover, idx int, lr plog.Logs) bool {
-	_ = conn.ConsumeLogs(context.Background(), lr)
-	stableIndex := conn.failover.pS.CurrentPipeline()
+func consumeLogsAndCheckStable(router *logsRouter, idx int, lr plog.Logs) bool {
+	strategy := router.strategy.(*standardLogsStrategy)
+	_ = router.Consume(context.Background(), lr)
+	stableIndex := strategy.pS.CurrentPipeline()
 	return stableIndex == idx
 }