Use informer for time trigger handling with multiple namespace support (#2593)

* changes to add informer in timer for time trigger
* Refactor timer trigger handlers and remove unwanted code

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
Co-authored-by: Sanket Sudake <sanketsudake@gmail.com>

pkg/timer/main.go

@@ -38,7 +38,8 @@ func Start(ctx context.Context, logger *zap.Logger, routerUrl string) error {
 	}
 
 	poster := publisher.MakeWebhookPublisher(logger, routerUrl)
-	MakeTimerSync(ctx, logger, fissionClient, MakeTimer(logger, poster))
+	timerSync := MakeTimerSync(ctx, logger, fissionClient, MakeTimer(logger, poster))
+	timerSync.Run(ctx)
 
 	return nil
 }

pkg/timer/timer.go

@@ -21,7 +21,6 @@ import (
 	"go.uber.org/zap"
 
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
-	"github.com/fission/fission/pkg/crd"
 	"github.com/fission/fission/pkg/publisher"
 	"github.com/fission/fission/pkg/utils"
 )
@@ -34,20 +33,11 @@ const (
 
 type (
 	Timer struct {
-		logger         *zap.Logger
+		logger    *zap.Logger
-		triggers       map[string]*timerTriggerWithCron
+		triggers  map[string]*timerTriggerWithCron
-		requestChannel chan *timerRequest
+		publisher *publisher.Publisher
-		publisher      *publisher.Publisher
 	}
 
-	timerRequest struct {
-		requestType
-		triggers        []fv1.TimeTrigger
-		responseChannel chan *timerResponse
-	}
-	timerResponse struct {
-		error
-	}
 	timerTriggerWithCron struct {
 		trigger fv1.TimeTrigger
 		cron    *cron.Cron
@@ -56,88 +46,26 @@ type (
 
 func MakeTimer(logger *zap.Logger, publisher publisher.Publisher) *Timer {
 	timer := &Timer{
-		logger:         logger.Named("timer"),
+		logger:    logger.Named("timer"),
-		triggers:       make(map[string]*timerTriggerWithCron),
+		triggers:  make(map[string]*timerTriggerWithCron),
-		requestChannel: make(chan *timerRequest),
+		publisher: &publisher,
-		publisher:      &publisher,
 	}
-	go timer.svc()
 	return timer
 }
 
-func (timer *Timer) Sync(triggers []fv1.TimeTrigger) error {
-	req := &timerRequest{
-		requestType:     SYNC,
-		triggers:        triggers,
-		responseChannel: make(chan *timerResponse),
-	}
-	timer.requestChannel <- req
-	resp := <-req.responseChannel
-	return resp.error
-}
-
-func (timer *Timer) svc() {
-	for {
-		req := <-timer.requestChannel
-		switch req.requestType {
-		case SYNC:
-			err := timer.syncCron(req.triggers)
-			req.responseChannel <- &timerResponse{error: err}
-		}
-	}
-}
-
-func (timer *Timer) syncCron(triggers []fv1.TimeTrigger) error {
-	// add new triggers or update existing ones
-	triggerMap := make(map[string]bool)
-	for _, t := range triggers {
-		triggerMap[crd.CacheKey(&t.ObjectMeta)] = true
-		if item, ok := timer.triggers[crd.CacheKey(&t.ObjectMeta)]; ok {
-			// update cron if the cron spec changed
-			if item.trigger.Spec.Cron != t.Spec.Cron {
-				// if there is an cron running, stop it
-				if item.cron != nil {
-					item.cron.Stop()
-				}
-				item.cron = timer.newCron(t)
-			}
-
-			item.trigger = t
-		} else {
-			timer.triggers[crd.CacheKey(&t.ObjectMeta)] = &timerTriggerWithCron{
-				trigger: t,
-				cron:    timer.newCron(t),
-			}
-		}
-	}
-
-	// process removed triggers
-	for k, v := range timer.triggers {
-		if _, found := triggerMap[k]; !found {
-			if v.cron != nil {
-				v.cron.Stop()
-				timer.logger.Info("cron for time trigger stopped", zap.String("trigger", v.trigger.ObjectMeta.Name))
-			}
-			delete(timer.triggers, k)
-		}
-	}
-
-	return nil
-}
-
 func (timer *Timer) newCron(t fv1.TimeTrigger) *cron.Cron {
 	c := cron.New()
 	c.AddFunc(t.Spec.Cron, func() { //nolint: errCheck
 		headers := map[string]string{
-			"X-Fission-Timer-Name": t.ObjectMeta.Name,
+			"X-Fission-Timer-Name": t.Name,
 		}
 
 		// with the addition of multi-tenancy, the users can create functions in any namespace. however,
 		// the triggers can only be created in the same namespace as the function.
 		// so essentially, function namespace = trigger namespace.
-		(*timer.publisher).Publish("", headers, utils.UrlForFunction(t.Spec.FunctionReference.Name, t.ObjectMeta.Namespace))
+		(*timer.publisher).Publish("", headers, utils.UrlForFunction(t.Spec.FunctionReference.Name, t.Namespace))
 	})
 	c.Start()
-	timer.logger.Info("added new cron for time trigger", zap.String("trigger", t.ObjectMeta.Name))
+	timer.logger.Info("started cron for time trigger", zap.String("trigger_name", t.Name), zap.String("trigger_namespace", t.Namespace), zap.String("cron", t.Spec.Cron))
 	return c
 }

pkg/timer/timerSync.go

@@ -21,17 +21,20 @@ import (
 	"time"
 
 	"go.uber.org/zap"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8sCache "k8s.io/client-go/tools/cache"
 
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+	"github.com/fission/fission/pkg/crd"
 	"github.com/fission/fission/pkg/generated/clientset/versioned"
 	"github.com/fission/fission/pkg/utils"
 )
 
 type (
 	TimerSync struct {
-		logger        *zap.Logger
+		logger              *zap.Logger
-		fissionClient versioned.Interface
+		fissionClient       versioned.Interface
-		timer         *Timer
+		timer               *Timer
+		timeTriggerInformer map[string]k8sCache.SharedIndexInformer
 	}
 )
 
@@ -41,24 +44,68 @@ func MakeTimerSync(ctx context.Context, logger *zap.Logger, fissionClient versio
 		fissionClient: fissionClient,
 		timer:         timer,
 	}
-	go ws.syncSvc(ctx)
+	ws.timeTriggerInformer = utils.GetInformersForNamespaces(fissionClient, time.Minute*30, fv1.TimeTriggerResource)
+	ws.TimeTriggerEventHandlers(ctx)
 	return ws
 }
 
-func (ws *TimerSync) syncSvc(ctx context.Context) {
+func (ws *TimerSync) Run(ctx context.Context) {
-	for {
+	for _, informer := range ws.timeTriggerInformer {
-		triggers, err := ws.fissionClient.CoreV1().TimeTriggers(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		go informer.Run(ctx.Done())
-		if err != nil {
+	}
-			if utils.IsNetworkError(err) {
+}
-				ws.logger.Info("encountered a network error - will retry", zap.Error(err))
+
-				time.Sleep(5 * time.Second)
+func (ws *TimerSync) AddUpdateTimeTrigger(timeTrigger *fv1.TimeTrigger) {
-				continue
+	logger := ws.logger.With(zap.String("trigger_name", timeTrigger.Name), zap.String("trigger_namespace", timeTrigger.Namespace))
-			}
+
-			ws.logger.Fatal("failed to get time trigger list", zap.Error(err))
+	ws.logger.Debug("cron event")
-		}
+
-		ws.timer.Sync(triggers.Items) //nolint: errCheck
+	if item, ok := ws.timer.triggers[crd.CacheKeyUID(&timeTrigger.ObjectMeta)]; ok {
-
+		if item.trigger.Spec.Cron != timeTrigger.Spec.Cron {
-		// TODO switch to watches
+			if item.cron != nil {
-		time.Sleep(3 * time.Second)
+				item.cron.Stop()
+			}
+			item.trigger = *timeTrigger
+			item.cron = ws.timer.newCron(*timeTrigger)
+			logger.Debug("cron updated")
+		}
+	} else {
+		ws.timer.triggers[crd.CacheKeyUID(&timeTrigger.ObjectMeta)] = &timerTriggerWithCron{
+			trigger: *timeTrigger,
+			cron:    ws.timer.newCron(*timeTrigger),
+		}
+		logger.Debug("cron added")
+	}
+}
+
+func (ws *TimerSync) DeleteTimeTrigger(timeTrigger *fv1.TimeTrigger) {
+	logger := ws.logger.With(zap.String("trigger_name", timeTrigger.Name), zap.String("trigger_namespace", timeTrigger.Namespace))
+
+	if item, ok := ws.timer.triggers[crd.CacheKeyUID(&timeTrigger.ObjectMeta)]; ok {
+		if item.cron != nil {
+			item.cron.Stop()
+			logger.Info("cron for time trigger stopped")
+		}
+		delete(ws.timer.triggers, crd.CacheKeyUID(&timeTrigger.ObjectMeta))
+		logger.Debug("cron deleted")
+	}
+}
+
+func (ws *TimerSync) TimeTriggerEventHandlers(ctx context.Context) {
+	for _, informer := range ws.timeTriggerInformer {
+		informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				timeTrigger := obj.(*fv1.TimeTrigger)
+				ws.AddUpdateTimeTrigger(timeTrigger)
+			},
+			UpdateFunc: func(_ interface{}, obj interface{}) {
+				timeTrigger := obj.(*fv1.TimeTrigger)
+				ws.AddUpdateTimeTrigger(timeTrigger)
+			},
+			DeleteFunc: func(obj interface{}) {
+				timeTrigger := obj.(*fv1.TimeTrigger)
+				ws.DeleteTimeTrigger(timeTrigger)
+			},
+		})
 	}
 }