Generic event source connection retry logic broken in version 1.9.4

[grahamia]

Describe the bug
We have a generic event source deployed which streams off a gRPC port producing events.

  generic:
    our-event-source:
      config: |
        {}
      insecure: true
      url: our-event-producer:50051

On upgrading Argo Events > 1.9.3 we have realised that the event source fails to reconnect to the event source server in the event of the event source server restarting.

To Reproduce
Steps to reproduce the behaviour (argo events 1.9.4 or greater):

1. Deploy a simple event source server exposing events on a gRPC port
2. Deploy a generic event source as above
3. Here everything should work.
4. Restart the event source server
5. The event source sees the failed connect and logs the error
6. It retries 3 times and then stops doing anything, just in a hung state.

Expected behavior
The event source should retry forever (or be configurable) until it is able to connect to the source

Additional context
For Argo Events 1.9.4 or above

Logs showing the event source retrying 3 times after the initial failure then just hangs.

{"level":"info","ts":1747394260.585098,"logger":"argo-events.eventsource","caller":"generic/start.go:77","msg":"connected to eventsource server successfully, started event stream...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051"}
{"level":"error","ts":1747394311.7187333,"logger":"argo-events.eventsource","caller":"generic/start.go:81","msg":"failed to receive events from the event stream, reconnecting in 5 seconds...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051","error":"rpc error: code = Unavailable desc = error reading from server: EOF","stacktrace":"github.com/argoproj/argo-events/pkg/eventsources/sources/generic.(*EventListener).StartListening\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/sources/generic/start.go:81\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3.1\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:534\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry.func1\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:90\nk8s.io/apimachinery/pkg/util/wait.runConditionWithCrashProtection\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/wait.go:145\nk8s.io/apimachinery/pkg/util/wait.ExponentialBackoff\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/backoff.go:461\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:89\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:533"}
{"level":"info","ts":1747394311.7223485,"logger":"argo-events.eventsource","caller":"generic/start.go:71","msg":"dialing eventsource server...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051"}
{"level":"error","ts":1747394311.7637541,"logger":"argo-events.eventsource","caller":"generic/start.go:74","msg":"failed to connect eventsource server, reconnecting in 5 seconds...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051","error":"rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 34.118.224.26:50051: connect: connection refused\"","stacktrace":"github.com/argoproj/argo-events/pkg/eventsources/sources/generic.(*EventListener).StartListening\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/sources/generic/start.go:74\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3.1\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:534\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry.func1\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:90\nk8s.io/apimachinery/pkg/util/wait.runConditionWithCrashProtection\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/wait.go:145\nk8s.io/apimachinery/pkg/util/wait.ExponentialBackoff\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/backoff.go:461\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:89\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:533"}
{"level":"info","ts":1747394315.5372474,"logger":"argo-events.eventsource","caller":"generic/start.go:71","msg":"dialing eventsource server...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051"}
{"level":"error","ts":1747394315.5818846,"logger":"argo-events.eventsource","caller":"generic/start.go:74","msg":"failed to connect eventsource server, reconnecting in 5 seconds...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051","error":"rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 34.118.224.26:50051: connect: connection refused\"","stacktrace":"github.com/argoproj/argo-events/pkg/eventsources/sources/generic.(*EventListener).StartListening\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/sources/generic/start.go:74\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3.1\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:534\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry.func1\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:90\nk8s.io/apimachinery/pkg/util/wait.runConditionWithCrashProtection\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/wait.go:145\nk8s.io/apimachinery/pkg/util/wait.ExponentialBackoff\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/backoff.go:461\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:89\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:533"}
{"level":"info","ts":1747394320.5373504,"logger":"argo-events.eventsource","caller":"generic/start.go:71","msg":"dialing eventsource server...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051"}
{"level":"error","ts":1747394320.5788677,"logger":"argo-events.eventsource","caller":"generic/start.go:74","msg":"failed to connect eventsource server, reconnecting in 5 seconds...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051","error":"rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 34.118.224.26:50051: connect: connection refused\"","stacktrace":"github.com/argoproj/argo-events/pkg/eventsources/sources/generic.(*EventListener).StartListening\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/sources/generic/start.go:74\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3.1\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:534\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry.func1\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:90\nk8s.io/apimachinery/pkg/util/wait.runConditionWithCrashProtection\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/wait.go:145\nk8s.io/apimachinery/pkg/util/wait.ExponentialBackoff\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/backoff.go:461\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:89\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:533"}

In Argo Events 1.9.3 we were seeing the following output in the event source on a restart of the event source server:

{"level":"error","ts":1747394025.5009372,"logger":"argo-events.eventsource","caller":"generic/start.go:81","msg":"failed to receive events from the event stream, reconnecting in 5 seconds...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051","error":"rpc error: code = Unavailable desc = error reading from server: EOF","stacktrace":"github.com/argoproj/argo-events/pkg/eventsources/sources/generic.(*EventListener).StartListening\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/sources/generic/start.go:81\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3.1\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:534\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry.func1\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:90\nk8s.io/apimachinery/pkg/util/wait.runConditionWithCrashProtection\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/wait.go:145\nk8s.io/apimachinery/pkg/util/wait.ExponentialBackoff\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.29.2/pkg/util/wait/backoff.go:461\ngithub.com/argoproj/argo-events/pkg/shared/util.DoWithRetry\n\t/home/runner/work/argo-events/argo-events/pkg/shared/util/retry.go:89\ngithub.com/argoproj/argo-events/pkg/eventsources.(*EventSourceAdaptor).run.func3\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/eventing.go:533"}
{"level":"info","ts":1747394025.5013978,"logger":"argo-events.eventsource","caller":"generic/start.go:71","msg":"dialing eventsource server...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051"}
{"level":"info","ts":1747394041.4183476,"logger":"argo-events.eventsource","caller":"generic/start.go:77","msg":"connected to eventsource server successfully, started event stream...","eventSourceName":"our-event-source","eventSourceType":"generic","eventName":"our-event-source","url":"our-event-producer:50051"}

it dialled and waited till valid and then connected and all worked as expected.

Looking at the changes in the code between the two versions the grpc library was upgraded and the call made to create the client went from:

	opt = append(opt, grpc.WithBlock())
	if el.GenericEventSource.Insecure {
		opt = append(opt, grpc.WithTransportCredentials(insecure.NewCredentials()))
	}
	conn, err := grpc.DialContext(context.Background(), el.GenericEventSource.URL, opt...)

to:

	if el.GenericEventSource.Insecure {
		opt = append(opt, grpc.WithTransportCredentials(insecure.NewCredentials()))
	}
	conn, err := grpc.NewClient(el.GenericEventSource.URL, opt...)

Can with WithBlock is deprecated in the grpc library but then the calling code in here should handle the trying better. I am not sure at this point why it currently only does 3 retries, best guess is connection getState changes to some other value not in this if statement and then would just tick infinitely not doing anything.

---

Message from the maintainers:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.

[grahamia]

I have debugged a little further. In 1.9.3 the service blocked until the connection was active because of the opt = append(opt, grpc.WithBlock()).

So now in 1.9.4 the connection is initiated and set onto the EventListener conn attribute. The connection is then attempted to be used in the client.StartEventSource call, this fails and returns the error, the code then ticks over after 5 seconds and tries same logic again. After 3 attempts (or however many ticks until the server port is active) while the service is in its sleep state waiting for the next tick the connection state changes into a READY state, then when the next tick does happen it doesn't execute anything as the if statement evaluates to false and no else clause exists so therefore the service is now in a constant state of ticking every 5 seconds doing nothing.

[grahamia]

I have manually tested with the following code which seems to solve the issue

package generic

import (
	"context"
	"encoding/json"
	fmt "fmt"
	"time"

	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/connectivity"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"

	"github.com/argoproj/argo-events/pkg/apis/events/v1alpha1"
	eventsourcecommon "github.com/argoproj/argo-events/pkg/eventsources/common"
	"github.com/argoproj/argo-events/pkg/eventsources/events"
	"github.com/argoproj/argo-events/pkg/eventsources/sources"
	metrics "github.com/argoproj/argo-events/pkg/metrics"
	"github.com/argoproj/argo-events/pkg/shared/logging"
	sharedutil "github.com/argoproj/argo-events/pkg/shared/util"
)

// EventListener implements Eventing for generic event source
type EventListener struct {
	EventSourceName    string
	EventName          string
	GenericEventSource v1alpha1.GenericEventSource
	Metrics            *metrics.Metrics

	conn *grpc.ClientConn
}

// GetEventSourceName returns name of event source
func (el *EventListener) GetEventSourceName() string {
	return el.EventSourceName
}

// GetEventName returns name of event
func (el *EventListener) GetEventName() string {
	return el.EventName
}

// GetEventSourceType return type of event server
func (el *EventListener) GetEventSourceType() v1alpha1.EventSourceType {
	return v1alpha1.GenericEvent
}

// StartListening listens to generic events
func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte, ...eventsourcecommon.Option) error) error {
	logger := logging.FromContext(ctx).
		With(zap.String(logging.LabelEventSourceType, string(el.GetEventSourceType())),
			zap.String(logging.LabelEventName, el.GetEventName()),
			zap.String("url", el.GenericEventSource.URL))
	logger.Info("started processing the generic event source...")
	defer sources.Recover(el.GetEventName())

	logger.Info("connecting to eventsource server in 5 seconds...")
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			logger.Info("closing client connection and exiting eventsource...")
			if el.conn != nil && (el.conn.GetState() == connectivity.Ready || el.conn.GetState() == connectivity.Connecting) {
				el.conn.Close()
			}
			return nil
		case <-ticker.C:
			var eventStream Eventing_StartEventSourceClient
			if el.conn == nil || el.conn.GetState() == connectivity.Shutdown || el.conn.GetState() == connectivity.TransientFailure {
				logger.Info("dialing eventsource server...")
				var err error
				eventStream, err = el.connect(logger)
				if err != nil {
					logger.Errorw("failed to connect eventsource server, reconnecting in 5 seconds...", zap.Error(err))
					continue
				}
			}
			if el.conn.GetState() == connectivity.Ready || el.conn.GetState() == connectivity.Idle {
				if eventStream == nil {
					var err error
					eventStream, err = el.buildEventStream()
					if err != nil {
						logger.Errorw("failed to build stream, retrying in 5 seconds...", zap.Error(err), 
							zap.String("ess-connection-state", el.conn.GetState().String()))
						continue
					}
				}
				logger.Info("connected to eventsource server successfully, started event stream...")
				for {
					event, err := eventStream.Recv()
					if err != nil {
						logger.Errorw("failed to receive events from the event stream, reconnecting in 5 seconds...", zap.Error(err))
						// close the connection and retry in next cycle.
						el.conn.Close()
						break
					}
					if err := el.handleOne(event, dispatch, logger); err != nil {
						logger.Errorw("failed to process a Generics event", zap.Error(err))
						el.Metrics.EventProcessingFailed(el.GetEventSourceName(), el.GetEventName())
					}
				}
			} else {
				logger.Warnw("eventsource server connection in an incorrect state, retrying in 5 seconds...",
					zap.String("ess-connection-state", el.conn.GetState().String()))
			}
		}
	}
}

func (el *EventListener) handleOne(event *Event, dispatch func([]byte, ...eventsourcecommon.Option) error, logger *zap.SugaredLogger) error {
	defer func(start time.Time) {
		el.Metrics.EventProcessingDuration(el.GetEventSourceName(), el.GetEventName(), float64(time.Since(start)/time.Millisecond))
	}(time.Now())

	logger.Info("received an event from server")
	eventData := &events.GenericEventData{
		Metadata: el.GenericEventSource.Metadata,
	}
	if el.GenericEventSource.JSONBody {
		eventData.Body = (*json.RawMessage)(&event.Payload)
	} else {
		eventData.Body = event.Payload
	}
	eventBytes, err := json.Marshal(eventData)
	if err != nil {
		return fmt.Errorf("failed to marshal the event data, %w", err)
	}
	logger.Info("dispatching event...")
	if err := dispatch(eventBytes); err != nil {
		return fmt.Errorf("failed to dispatch a Generic event, %w", err)
	}
	return nil
}

func (el *EventListener) connect(logger *zap.SugaredLogger) (Eventing_StartEventSourceClient, error) {
	var opt []grpc.DialOption
	if el.GenericEventSource.Insecure {
		opt = append(opt, grpc.WithTransportCredentials(insecure.NewCredentials()))
	}
	conn, err := grpc.NewClient(el.GenericEventSource.URL, opt...)
	if err != nil {
		logger.Errorw("failed to inititalise gRPC stream to eventsource server", zap.Error(err))
		return nil, err
	}
	el.conn = conn
	return el.buildEventStream()
}

func (el *EventListener) buildEventStream() (Eventing_StartEventSourceClient, error) {
	client := NewEventingClient(el.conn)
	ctx := context.Background()
	if el.GenericEventSource.AuthSecret != nil {
		token, err := sharedutil.GetSecretFromVolume(el.GenericEventSource.AuthSecret)
		if err != nil {
			return nil, err
		}
		auth := fmt.Sprintf("Bearer %s", token)
		ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs("authorization", auth))
	}
	return client.StartEventSource(ctx, &EventSource{
		Name:   el.GetEventName(),
		Config: []byte(el.GenericEventSource.Config),
	})
}

[github-actions]

This issue has been automatically marked as stale because it has not had
any activity in the last 60 days. It will be closed if no further activity
occurs. Thank you for your contributions.