Create a publisher per cron timer (#3218)

* Create a publisher per cron timer

We want to be able to run multiple timers at the same time.

* Remove error

pkg/timer/main.go

@@ -23,7 +23,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/fission/fission/pkg/crd"
-	"github.com/fission/fission/pkg/publisher"
 	"github.com/fission/fission/pkg/utils/manager"
 )
 
@@ -38,8 +37,7 @@ func Start(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger *
 		return fmt.Errorf("error waiting for CRDs: %w", err)
 	}
 
-	poster := publisher.MakeWebhookPublisher(logger, routerUrl)
+	timerSync, err := MakeTimerSync(ctx, logger, fissionClient, MakeTimer(logger, routerUrl))
-	timerSync, err := MakeTimerSync(ctx, logger, fissionClient, MakeTimer(logger, poster))
 	if err != nil {
 		return fmt.Errorf("error making timer sync: %w", err)
 	}

pkg/timer/timer.go

@@ -38,7 +38,7 @@ type (
 	Timer struct {
 		logger    *zap.Logger
 		triggers  map[types.UID]*timerTriggerWithCron
-		publisher *publisher.Publisher
+		routerUrl string
 	}
 
 	timerTriggerWithCron struct {
@@ -47,17 +47,21 @@ type (
 	}
 )
 
-func MakeTimer(logger *zap.Logger, publisher publisher.Publisher) *Timer {
+func MakeTimer(logger *zap.Logger, routerUrl string) *Timer {
 	timer := &Timer{
 		logger:    logger.Named("timer"),
 		triggers:  make(map[types.UID]*timerTriggerWithCron),
-		publisher: &publisher,
+		routerUrl: routerUrl,
 	}
 	return timer
 }
 
-func (timer *Timer) newCron(t fv1.TimeTrigger) *cron.Cron {
+func (timer *Timer) newCron(t fv1.TimeTrigger, routerUrl string) *cron.Cron {
 	target := utils.UrlForFunction(t.Spec.Name, t.Namespace) + t.Spec.Subpath
+
+	// create one publisher per-cron timer
+	timerPublisher := publisher.MakeWebhookPublisher(timer.logger, routerUrl)
+
 	c := cron.New(
 		cron.WithParser(
 			cron.NewParser(
@@ -70,7 +74,7 @@ func (timer *Timer) newCron(t fv1.TimeTrigger) *cron.Cron {
 		// 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(context.Background(), "", headers, t.Spec.Method, target)
+		(timerPublisher).Publish(context.Background(), "", headers, t.Spec.Method, target)
 	})
 	c.Start()
 	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))

pkg/timer/timerSync.go

@@ -67,12 +67,12 @@ func (ws *TimerSync) AddUpdateTimeTrigger(timeTrigger *fv1.TimeTrigger) {
 			item.cron.Stop()
 		}
 		item.trigger = *timeTrigger
-		item.cron = ws.timer.newCron(*timeTrigger)
+		item.cron = ws.timer.newCron(*timeTrigger, ws.timer.routerUrl)
 		logger.Debug("cron updated")
 	} else {
 		ws.timer.triggers[crd.CacheKeyUIDFromMeta(&timeTrigger.ObjectMeta)] = &timerTriggerWithCron{
 			trigger: *timeTrigger,
-			cron:    ws.timer.newCron(*timeTrigger),
+			cron:    ws.timer.newCron(*timeTrigger, ws.timer.routerUrl),
 		}
 		logger.Debug("cron added")
 	}