diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go index 09189a8f..605390db 100644 --- a/pkg/controller/controller.go +++ b/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, diff --git a/pkg/controller/job.go b/pkg/controller/job.go new file mode 100644 index 00000000..64e6ce5b --- /dev/null +++ b/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() +} \ No newline at end of file diff --git a/pkg/controller/scheduler.go b/pkg/controller/scheduler.go index 6552898c..bb6907bd 100644 --- a/pkg/controller/scheduler.go +++ b/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) }