updated connector structure

.chloggen/failover-PR4.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: new_component
+
+# 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: PR updates connector structure to avoid exposing indexes outside of pipeline_selector
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [20766]
+
+# (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/failover.go

@@ -4,9 +4,7 @@
 package failoverconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/failoverconnector"
 
 import (
-	"context"
 	"errors"
-	"time"
 
 	"go.opentelemetry.io/collector/component"
 
@@ -20,7 +18,6 @@ type failoverRouter[C any] struct {
 	cfg              *Config
 	pS               *state.PipelineSelector
 	consumers        []C
-	rS               *state.RetryState
 }
 
 var (
@@ -29,22 +26,25 @@ var (
 )
 
 func newFailoverRouter[C any](provider consumerProvider[C], cfg *Config) *failoverRouter[C] {
+	pSConstants := state.PSConstants{
+		RetryInterval: cfg.RetryInterval,
+		RetryGap:      cfg.RetryGap,
+		MaxRetries:    cfg.MaxRetries,
+	}
 	return &failoverRouter[C]{
 		consumerProvider: provider,
 		cfg:              cfg,
-		pS:               state.NewPipelineSelector(len(cfg.PipelinePriority), cfg.MaxRetries),
-		rS:               &state.RetryState{},
+		pS:               state.NewPipelineSelector(len(cfg.PipelinePriority), pSConstants),
 	}
 }
 
-func (f *failoverRouter[C]) getCurrentConsumer() (C, int, bool) {
-	// if currentIndex incremented passed bounds of pipeline list
+func (f *failoverRouter[C]) getCurrentConsumer() (C, chan bool, bool) {
 	var nilConsumer C
-	idx := f.pS.CurrentIndex()
-	if idx >= len(f.cfg.PipelinePriority) {
-		return nilConsumer, -1, false
+	pl, ch := f.pS.SelectedPipeline()
+	if pl >= len(f.cfg.PipelinePriority) {
+		return nilConsumer, nil, false
 	}
-	return f.consumers[idx], idx, true
+	return f.consumers[pl], ch, true
 }
 
 func (f *failoverRouter[C]) registerConsumers() error {
@@ -60,77 +60,8 @@ func (f *failoverRouter[C]) registerConsumers() error {
 	return nil
 }
 
-func (f *failoverRouter[C]) handlePipelineError(idx int) {
-	// avoids race condition in case of consumeSIGNAL invocations
-	// where index was updated during execution
-	if idx != f.pS.CurrentIndex() {
-		return
-	}
-	doRetry := f.pS.IndexIsStable(idx)
-	// UpdatePipelineIndex either increments the pipeline to the next priority
-	// or returns it to the stable
-	f.pS.UpdatePipelineIndex(idx)
-	// if the currentIndex is not the stableIndex, that means the currentIndex is a higher
-	// priority index that was set during a retry, in which case we don't want to start a
-	// new retry goroutine
-	if !doRetry {
-		return
-	}
-	// kill existing retry goroutine if error is from a stable pipeline that failed for the first time
-	ctx, cancel := context.WithCancel(context.Background())
-	f.rS.InvokeCancel()
-	f.rS.UpdateCancelFunc(cancel)
-	f.enableRetry(ctx)
-}
-
-func (f *failoverRouter[C]) enableRetry(ctx context.Context) {
-	go func() {
-		ticker := time.NewTicker(f.cfg.RetryInterval)
-		defer ticker.Stop()
-
-		stableIndex := f.pS.StableIndex()
-		var cancelFunc context.CancelFunc
-		// checkContinueRetry checks that any higher priority levels have retries remaining
-		// (have not exceeded their maxRetries)
-		for f.checkContinueRetry(stableIndex) {
-			select {
-			case <-ticker.C:
-				// When the nextRetry interval starts we kill the existing iteration through
-				// the higher priority pipelines if still in progress
-				if cancelFunc != nil {
-					cancelFunc()
-				}
-				cancelFunc = f.handleRetry(ctx, stableIndex)
-			case <-ctx.Done():
-				return
-			}
-		}
-		f.rS.InvokeCancel()
-	}()
-}
-
-// handleRetry is responsible for launching goroutine and returning cancelFunc for context to be called if new
-// interval starts in the middle of the execution
-func (f *failoverRouter[C]) handleRetry(parentCtx context.Context, stableIndex int) context.CancelFunc {
-	retryCtx, cancelFunc := context.WithCancel(parentCtx)
-	go f.pS.RetryHighPriorityPipelines(retryCtx, stableIndex, f.cfg.RetryGap)
-	return cancelFunc
-}
-
-// checkStopRetry checks if retry should be suspended if all higher priority levels have exceeded their max retries
-func (f *failoverRouter[C]) checkContinueRetry(index int) bool {
-	for i := 0; i < index; i++ {
-		if f.pS.IndexRetryCount(i) < f.cfg.MaxRetries {
-			return true
-		}
-	}
-	return false
-}
-
-// reportStable reports back to the failoverRouter that the current priority level that was called by Consume.SIGNAL was
-// stable
-func (f *failoverRouter[C]) reportStable(idx int) {
-	f.pS.ReportStable(idx)
+func (f *failoverRouter[C]) Shutdown() {
+	f.pS.RS.InvokeCancel()
 }
 
 // For Testing