From 45d6132ffd95a56f51c4e82ef3e90885403b6102 Mon Sep 17 00:00:00 2001 From: markretallack <92091464+markretallack@users.noreply.github.com> Date: Mon, 19 May 2025 13:36:12 +0100 Subject: [PATCH] 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 | 4 +--- pkg/timer/timer.go | 14 +++++++++----- pkg/timer/timerSync.go | 4 ++-- 3 files changed, 12 insertions(+), 10 deletions(-) diff --git a/pkg/timer/main.go b/pkg/timer/main.go index e588e73b..d34810ce 100644 --- a/pkg/timer/main.go +++ b/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, poster)) + timerSync, err := MakeTimerSync(ctx, logger, fissionClient, MakeTimer(logger, routerUrl)) if err != nil { return fmt.Errorf("error making timer sync: %w", err) } diff --git a/pkg/timer/timer.go b/pkg/timer/timer.go index 9e8dd624..75100d4c 100644 --- a/pkg/timer/timer.go +++ b/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)) diff --git a/pkg/timer/timerSync.go b/pkg/timer/timerSync.go index 539f7fc8..e6c0786f 100644 --- a/pkg/timer/timerSync.go +++ b/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") }