Schedule canary analysis based on interval

pkg/controller/controller.go

@@ -40,6 +40,7 @@ type Controller struct {
 	eventRecorder record.EventRecorder
 	logger        *zap.SugaredLogger
 	canaries      *sync.Map
+	jobs          map[string]CanaryJob
 	deployer      CanaryDeployer
 	router        CanaryRouter
 	observer      CanaryObserver
@@ -98,6 +99,7 @@ func NewController(
 		eventRecorder: eventRecorder,
 		logger:        logger,
 		canaries:      new(sync.Map),
+		jobs:          map[string]CanaryJob{},
 		flaggerWindow: flaggerWindow,
 		deployer:      deployer,
 		router:        router,

pkg/controller/job.go

@@ -0,0 +1,34 @@
+package controller
+
+import "time"
+
+// CanaryJob holds the reference to a canary deployment schedule
+type CanaryJob struct {
+	Name      string
+	Namespace string
+	function  func(name string, namespace string)
+	done      chan bool
+	ticker    *time.Ticker
+}
+
+// Start runs the canary analysis on a schedule
+func (j CanaryJob) Start() {
+	go func() {
+		// run the infra bootstrap on job creation
+		j.function(j.Name, j.Namespace)
+		for {
+			select {
+			case <-j.ticker.C:
+				j.function(j.Name, j.Namespace)
+			case <-j.done:
+				return
+			}
+		}
+	}()
+}
+
+// Stop closes the job channel and stops the ticker
+func (j CanaryJob) Stop() {
+	close(j.done)
+	j.ticker.Stop()
+}

pkg/controller/scheduler.go

@@ -8,22 +8,51 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+// scheduleCanaries synchronises the canary map with the jobs map,
+// for new canaries new jobs are created and started
+// for the removed canaries the jobs are stopped and deleted
 func (c *Controller) scheduleCanaries() {
+	current := make(map[string]bool)
 	stats := make(map[string]int)
+
 	c.canaries.Range(func(key interface{}, value interface{}) bool {
-		r := value.(*flaggerv1.Canary)
-		if r.Spec.TargetRef.Kind == "Deployment" {
-			go c.advanceCanary(r.Name, r.Namespace)
+		canary := value.(*flaggerv1.Canary)
+		name := key.(string)
+		current[name] = true
+
+		// schedule new jobs
+		if _, exists := c.jobs[name]; !exists {
+			job := CanaryJob{
+				Name:      canary.Name,
+				Namespace: canary.Namespace,
+				function:  c.advanceCanary,
+				done:      make(chan bool),
+				ticker:    time.NewTicker(canary.GetAnalysisInterval()),
+			}
+
+			c.jobs[name] = job
+			job.Start()
 		}
 
-		t, ok := stats[r.Namespace]
+		// compute canaries per namespace total
+		t, ok := stats[canary.Namespace]
 		if !ok {
-			stats[r.Namespace] = 1
+			stats[canary.Namespace] = 1
 		} else {
-			stats[r.Namespace] = t + 1
+			stats[canary.Namespace] = t + 1
 		}
 		return true
 	})
+
+	// cleanup deleted jobs
+	for job := range c.jobs {
+		if _, exists := current[job]; !exists {
+			c.jobs[job].Stop()
+			delete(c.jobs, job)
+		}
+	}
+
+	// set total canaries per namespace metric
 	for k, v := range stats {
 		c.recorder.SetTotal(k, v)
 	}