fix(eventbus): NATS eventbus auto reconnect and code refactory

eventbus/driver/driver.go

@@ -14,11 +14,12 @@ type Driver interface {
 	// SubscribeEventSources is used to subscribe multiple event source dependencies
 	// Parameter - ctx, context
 	// Parameter - conn, eventbus connection
+	// Parameter - closeCh, channel to indicate to close the subscription
 	// Parameter - dependencyExpr, example: "(dep1 || dep2) && dep3"
 	// Parameter - dependencies, array of dependencies information
 	// Parameter - filter, a function used to filter the message
 	// Parameter - action, a function to be triggered after all conditions meet
-	SubscribeEventSources(ctx context.Context, conn Connection, dependencyExpr string, dependencies []Dependency, filter func(string, cloudevents.Event) bool, action func(map[string]cloudevents.Event)) error
+	SubscribeEventSources(ctx context.Context, conn Connection, closeCh <-chan struct{}, dependencyExpr string, dependencies []Dependency, filter func(string, cloudevents.Event) bool, action func(map[string]cloudevents.Event)) error
 
 	// Publish a message
 	Publish(conn Connection, message []byte) error

eventbus/driver/nats.go

@@ -3,8 +3,6 @@ package driver
 import (
 	"context"
 	"encoding/json"
-	"os"
-	"os/signal"
 	"strings"
 	"time"
 
@@ -73,7 +71,16 @@ func NewNATSStreaming(url, clusterID, subject, clientID string, auth *Auth, logg
 
 func (n *natsStreaming) Connect() (Connection, error) {
 	log := n.logger.WithField("clientID", n.clientID)
-	opts := []nats.Option{}
+	opts := []nats.Option{
+		nats.MaxReconnects(-1),
+		nats.ReconnectWait(3 * time.Second),
+		nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
+			log.Errorf("NATS connection lost, reason: %v", err)
+		}),
+		nats.ReconnectHandler(func(nnc *nats.Conn) {
+			log.Info("Reconnected to NATS server")
+		}),
+	}
 	switch n.auth.Strategy {
 	case eventbusv1alpha1.AuthStrategyToken:
 		log.Info("NATS auth strategy: Token")
@@ -89,9 +96,10 @@ func (n *natsStreaming) Connect() (Connection, error) {
 		return nil, err
 	}
 	log.Info("Connected to NATS server.")
-	sc, err := stan.Connect(n.clusterID, n.clientID, stan.NatsConn(nc),
+
+	sc, err := stan.Connect(n.clusterID, n.clientID, stan.NatsConn(nc), stan.Pings(5, 60),
 		stan.SetConnectionLostHandler(func(_ stan.Conn, reason error) {
-			log.Fatalf("Connection lost, reason: %v", reason)
+			log.Errorf("NATS streaming connection lost, reason: %v", reason)
 		}))
 	if err != nil {
 		log.Errorf("Failed to connect to NATS streaming server, %v", err)
@@ -114,7 +122,7 @@ func (n *natsStreaming) Publish(conn Connection, message []byte) error {
 // Parameter - dependencies, array of dependencies information
 // Parameter - filter, a function used to filter the message
 // Parameter - action, a function to be triggered after all conditions meet
-func (n *natsStreaming) SubscribeEventSources(ctx context.Context, conn Connection, dependencyExpr string, dependencies []Dependency, filter func(string, cloudevents.Event) bool, action func(map[string]cloudevents.Event)) error {
+func (n *natsStreaming) SubscribeEventSources(ctx context.Context, conn Connection, closeCh <-chan struct{}, dependencyExpr string, dependencies []Dependency, filter func(string, cloudevents.Event) bool, action func(map[string]cloudevents.Event)) error {
 	log := n.logger.WithField("clientID", n.clientID)
 	msgHolder, err := newEventSourceMessageHolder(dependencyExpr, dependencies)
 	if err != nil {
@@ -138,20 +146,18 @@ func (n *natsStreaming) SubscribeEventSources(ctx context.Context, conn Connecti
 		return err
 	}
 	log.Infof("Subscribed to subject %s ...", n.subject)
-
-	signalChan := make(chan os.Signal, 1)
-	cleanupDone := make(chan bool)
-	signal.Notify(signalChan, os.Interrupt)
-	go func() {
-		for range signalChan {
-			log.Info("Received an interrupt, unsubscribing and closing connection...")
+	for {
+		select {
+		case <-ctx.Done():
+			log.Info("existing, unsubscribing and closing connection...")
+			_ = sub.Close()
+			log.Infof("subscription on subject %s closed", n.subject)
+		case <-closeCh:
+			log.Info("closing subscription...")
 			_ = sub.Close()
 			log.Infof("subscription on subject %s closed", n.subject)
-			cleanupDone <- true
 		}
-	}()
-	<-cleanupDone
-	return nil
+	}
 }
 
 func (n *natsStreaming) processEventSourceMsg(m *stan.Msg, msgHolder *eventSourceMessageHolder, filter func(dependencyName string, event cloudevents.Event) bool, action func(map[string]cloudevents.Event), log *logrus.Entry) {

eventbus/eventbus.go

@@ -47,6 +47,7 @@ func GetDriver(ctx context.Context, eventBusConfig eventbusv1alpha1.BusConfig, s
 		}
 		v.WatchConfig()
 		v.OnConfigChange(func(e fsnotify.Event) {
+			logger.Info("eventbus auth config file changed.")
 			err = v.Unmarshal(cred)
 			if err != nil {
 				logger.Errorf("failed to unmarshal auth.yaml after reloading, err: %v", err)

eventsources/eventing.go

@@ -53,7 +53,7 @@ type EventingServer interface {
 	GetEventSourceType() apicommon.EventSourceType
 
 	// Function to start listening events.
-	StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error
+	StartListening(ctx context.Context, dispatch func([]byte) error) error
 }
 
 // GetEventingServers returns the mapping of event source type and list of eventing servers
@@ -253,6 +253,31 @@ func (e *EventSourceAdaptor) Start(ctx context.Context, stopCh <-chan struct{})
 		logger.WithError(err).Errorln("failed to connect to eventbus")
 		return err
 	}
+	defer e.eventBusConn.Close()
+
+	// Daemon to reconnect
+	go func(ctx context.Context) {
+		logger.Info("starting eventbus connection daemon...")
+		for {
+			select {
+			case <-ctx.Done():
+				logger.Info("exiting eventbus connection daemon...")
+				return
+			default:
+				time.Sleep(3 * time.Second)
+			}
+			if e.eventBusConn == nil || e.eventBusConn.IsClosed() {
+				logger.Info("NATS connection lost, reconnecting...")
+				e.eventBusConn, err = driver.Connect()
+				if err != nil {
+					logger.WithError(err).Errorln("failed to reconnect to eventbus")
+					continue
+				}
+				logger.Info("reconnected the NATS streaming server...")
+			}
+		}
+	}(cctx)
+
 	for _, ss := range servers {
 		for _, server := range ss {
 			err := server.ValidateEventSource(cctx)
@@ -261,7 +286,7 @@ func (e *EventSourceAdaptor) Start(ctx context.Context, stopCh <-chan struct{})
 				return err
 			}
 			go func(s EventingServer) {
-				err := s.StartListening(cctx, stopCh, func(data []byte) error {
+				err := s.StartListening(cctx, func(data []byte) error {
 					event := cloudevents.NewEvent()
 					event.SetID(fmt.Sprintf("%x", uuid.New()))
 					event.SetType(string(s.GetEventSourceType()))
@@ -280,7 +305,6 @@ func (e *EventSourceAdaptor) Start(ctx context.Context, stopCh <-chan struct{})
 				})
 				logger.WithField(logging.LabelEventSourceName, s.GetEventSourceName()).
 					WithField(logging.LabelEventName, s.GetEventName()).WithError(err).Errorln("failed to start service.")
-				// TODO: Need retry.
 			}(server)
 		}
 	}

eventsources/sources/amqp/start.go

@@ -54,7 +54,7 @@ func (el *EventListener) GetEventSourceType() apicommon.EventSourceType {
 }
 
 // StartListening starts listening events
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),
 		logging.LabelEventSourceName: el.GetEventSourceName(),
@@ -141,7 +141,7 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 			if err != nil {
 				log.WithError(err).Errorln("failed to dispatch event")
 			}
-		case <-stopCh:
+		case <-ctx.Done():
 			err = conn.Close()
 			if err != nil {
 				log.WithError(err).Info("failed to close connection")

eventsources/sources/awssns/start.go

@@ -212,7 +212,7 @@ func (el *EventListener) GetEventSourceType() apicommon.EventSourceType {
 }
 
 // StartListening starts an SNS event source
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	logger := logging.FromContext(ctx)
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),

eventsources/sources/awssqs/start.go

@@ -58,7 +58,7 @@ func (el *EventListener) GetEventSourceType() apicommon.EventSourceType {
 }
 
 // StartListening starts listening events
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),
 		logging.LabelEventSourceName: el.GetEventSourceName(),
@@ -96,7 +96,7 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 	log.Infoln("listening for messages on the queue...")
 	for {
 		select {
-		case <-stopCh:
+		case <-ctx.Done():
 			log.Info("exiting SQS event listener...")
 			return nil
 		default:

eventsources/sources/azureeventshub/start.go

@@ -55,7 +55,7 @@ func (el *EventListener) GetEventSourceType() apicommon.EventSourceType {
 }
 
 // StartListening starts listening events
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),
 		logging.LabelEventSourceName: el.GetEventSourceName(),
@@ -123,7 +123,7 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 		listenerHandles = append(listenerHandles, listenerHandle)
 	}
 
-	<-stopCh
+	<-ctx.Done()
 	log.Infoln("stopping listener handlers")
 
 	for _, handler := range listenerHandles {

eventsources/sources/calendar/start.go

@@ -54,7 +54,7 @@ func (el *EventListener) GetEventSourceType() apicommon.EventSourceType {
 }
 
 // StartListening starts listening events
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),
 		logging.LabelEventSourceName: el.GetEventSourceName(),
@@ -126,7 +126,7 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 			if err != nil {
 				log.WithError(err).Errorln("failed to dispatch event")
 			}
-		case <-stopCh:
+		case <-ctx.Done():
 			log.Info("exiting calendar event listener...")
 			return nil
 		}

eventsources/sources/emitter/start.go

@@ -53,7 +53,7 @@ func (el *EventListener) GetEventSourceType() apicommon.EventSourceType {
 }
 
 // StartListening starts listening events
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),
 		logging.LabelEventSourceName: el.GetEventSourceName(),
@@ -130,7 +130,7 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 		return errors.Wrapf(err, "failed to subscribe to channel %s", emitterEventSource.ChannelName)
 	}
 
-	<-stopCh
+	<-ctx.Done()
 
 	log.WithField("channel-name", emitterEventSource.ChannelName).Infoln("event source stopped, unsubscribe the channel")
 

eventsources/sources/file/start.go

@@ -58,7 +58,7 @@ func (el *EventListener) GetEventSourceType() apicommon.EventSourceType {
 }
 
 // StartListening starts listening events
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),
 		logging.LabelEventSourceName: el.GetEventSourceName(),
@@ -68,12 +68,12 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 
 	fileEventSource := &el.FileEventSource
 	if fileEventSource.Polling {
-		if err := el.listenEventsPolling(ctx, stopCh, dispatch, log); err != nil {
+		if err := el.listenEventsPolling(ctx, dispatch, log); err != nil {
 			log.WithError(err).Errorln("failed to listen to events")
 			return err
 		}
 	} else {
-		if err := el.listenEvents(ctx, stopCh, dispatch, log); err != nil {
+		if err := el.listenEvents(ctx, dispatch, log); err != nil {
 			log.WithError(err).Errorln("failed to listen to events")
 			return err
 		}
@@ -82,7 +82,7 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 }
 
 // listenEvents listen to file related events.
-func (el *EventListener) listenEvents(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error, log *logrus.Entry) error {
+func (el *EventListener) listenEvents(ctx context.Context, dispatch func([]byte) error, log *logrus.Entry) error {
 	fileEventSource := &el.FileEventSource
 
 	// create new fs watcher
@@ -154,15 +154,15 @@ func (el *EventListener) listenEvents(ctx context.Context, stopCh <-chan struct{
 			}
 		case err := <-watcher.Errors:
 			return errors.Wrapf(err, "failed to process %s", el.GetEventName())
-		case <-stopCh:
+		case <-ctx.Done():
 			log.Infoln("event source has been stopped")
 			return nil
 		}
 	}
 }
 
 // listenEvents listen to file related events using polling.
-func (el *EventListener) listenEventsPolling(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error, log *logrus.Entry) error {
+func (el *EventListener) listenEventsPolling(ctx context.Context, dispatch func([]byte) error, log *logrus.Entry) error {
 	fileEventSource := &el.FileEventSource
 
 	// create new fs watcher
@@ -234,7 +234,7 @@ func (el *EventListener) listenEventsPolling(ctx context.Context, stopCh <-chan
 			case err := <-watcher.Error:
 				log.WithError(err).Errorf("failed to process %s", el.GetEventName())
 				return
-			case <-stopCh:
+			case <-ctx.Done():
 				log.Infoln("event source has been stopped")
 				return
 			}

eventsources/sources/gcppubsub/start.go

@@ -55,7 +55,7 @@ func (el *EventListener) GetEventSourceType() apicommon.EventSourceType {
 }
 
 // StartListening listens to GCP PubSub events
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	// In order to listen events from GCP PubSub,
 	// 1. Parse the event source that contains configuration to connect to GCP PubSub
 	// 2. Create a new PubSub client
@@ -169,7 +169,7 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 		return errors.Wrapf(err, "failed to receive the messages for subscription %s and topic %s for %s", subscriptionName, pubsubEventSource.Topic, el.GetEventName())
 	}
 
-	<-stopCh
+	<-ctx.Done()
 
 	log.Infoln("event source has been stopped")
 

eventsources/sources/github/start.go

@@ -274,8 +274,8 @@ func (router *Router) PostInactivate() error {
 	return nil
 }
 
-// StartListening starts an SNS event source
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+// StartListening starts an event source
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	logger := logging.FromContext(ctx)
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),

eventsources/sources/gitlab/start.go

@@ -262,8 +262,8 @@ func (router *Router) PostInactivate() error {
 	return nil
 }
 
-// StartListening starts an SNS event source
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+// StartListening starts an event source
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	logger := logging.FromContext(ctx)
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),

eventsources/sources/hdfs/start.go

@@ -61,7 +61,7 @@ func (w *WatchableHDFS) GetFileID(fi os.FileInfo) interface{} {
 }
 
 // StartListening starts listening events
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),
 		logging.LabelEventSourceName: el.GetEventSourceName(),
@@ -164,7 +164,7 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 			}
 		case err := <-watcher.Errors:
 			return errors.Wrapf(err, "failed to watch events for %s", el.GetEventName())
-		case <-stopCh:
+		case <-ctx.Done():
 			return nil
 		}
 	}

eventsources/sources/kafka/start.go

@@ -63,7 +63,7 @@ func verifyPartitionAvailable(part int32, partitions []int32) bool {
 }
 
 // StartListening starts listening events
-func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struct{}, dispatch func([]byte) error) error {
+func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byte) error) error {
 	log := logging.FromContext(ctx).WithFields(map[string]interface{}{
 		logging.LabelEventSourceType: el.GetEventSourceType(),
 		logging.LabelEventSourceName: el.GetEventSourceName(),
@@ -157,7 +157,7 @@ func (el *EventListener) StartListening(ctx context.Context, stopCh <-chan struc
 		case err := <-partitionConsumer.Errors():
 			return errors.Wrapf(err, "failed to consume messages for event source %s", el.GetEventName())
 
-		case <-stopCh:
+		case <-ctx.Done():
 			log.Infoln("event source is stopped, closing partition consumer")
 			err = partitionConsumer.Close()
 			if err != nil {