Update monitors on schedule changes (#27)

* update monitor on schedule changes

* Add test that schedule updates work properly

* typo in test

* fix

* try waiting for update

* always update cronjobs regardless, see what happens

* add dev environment reamde

* try again with logging

* need to clear the cronitor wrapper cache

* Put back conditional updates

* update chart version

Co-authored-by: Michael Owings <mowings@turbosquid.com>

CONTRIBUTE.md

@@ -1,6 +1,16 @@
 Contributing
 ======
 
+To set up a development environment:
+1. Ensure Go is installed, maybe update to the latest
+2. Install [Kind](https://kind.sigs.k8s.io/)
+3. Install [Skaffold](https://skaffold.dev/)
+4. Run `kind create cluster` to create a new Kubernetes cluster locally
+5. Add a `Secret` object to the cluster containing your Cronitor API key according to the [instructions](./README.md#instructions)
+6. In the main directory of this repository, run `skaffold dev`. Skaffold will build the Docker container and agent and push it to your local Kubernetes cluster. Anytime you make code changes it will rebuild and re-push.
+7. If you'd like to add some sample `CronJob`s easily, you can run `kubectl apply -f e2e-test/resources/main/`, which will add a bunch of pre-written test resources.
+8. When you're done, stop skaffold and clean up with `kind delete cluster`!
+
 
 To make a new release:
 1. Update the `version` number in the chart's `Chart.yaml`

charts/cronitor-kubernetes/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v1
 name: cronitor-kubernetes
 description: |
   Helm chart to deploy the Cronitor Kubernetes agent to automatically monitor your CronJobs
-version: "0.4.3"
-appVersion: "0.4.3"
+version: "0.4.4"
+appVersion: "0.4.4"

e2e-test/api/cronitor_wrapper.py

@@ -22,6 +22,9 @@ def delete_all_monitors(self):
         for monitor in monitors:
             self.delete_monitor_by_key(monitor['key'])
 
+    def bust_monitor_cache(self):
+        self.get_all_monitors.cache_clear()
+
     @cache
     def get_all_monitors(self, *, page: int = 1):
         PAGE_SIZE = 50

e2e-test/api/kubernetes_wrapper.py

@@ -14,11 +14,33 @@
 from typing import Optional, Dict
 import json
 
-
 class CronJobNotFound(BaseException):
     pass
 
 
+def patch_cronjob_by_name(name: str, namespace: Optional[str] = None, patch: Optional[Dict] = None):
+    if not patch:
+        return
+
+    try:
+        response = subprocess.check_output([
+            'kubectl', 'patch',
+            *(['-n', namespace] if namespace else []),
+            'cronjob', name,
+            '--patch', json.dumps(patch),
+            '-o', 'json'
+        ])
+
+    except subprocess.CalledProcessError as err:
+        # Generally if we get an error here with kubectl exiting 1,
+        # it's because the CronJob we're fetching doesn't exist.
+        if b'not found' in err.output:
+            raise CronJobNotFound(err.output)
+        else:
+            raise
+    return json.loads(response)
+
+
 def get_cronjob_by_name(name: str, namespace: Optional[str] = None) -> Dict:
     try:
         response = subprocess.check_output([

e2e-test/api/tests/test_monitor_types.py

@@ -3,7 +3,8 @@
 from typing import Optional
 import os
 import uuid
-from ..kubernetes_wrapper import get_cronjob_by_name
+import time
+from ..kubernetes_wrapper import get_cronjob_by_name, patch_cronjob_by_name
 from ..cronitor_wrapper import cronitor_wrapper_from_environment
 
 cronitor_wrapper = cronitor_wrapper_from_environment()
@@ -67,6 +68,21 @@ def test_monitor_created_with_new_id():
     assert monitor is not None, f"no monitor with key {monitor_key} exists but one should"
 
 
+def test_monitor_schedule_gets_updated():
+    random_id = os.getenv("RANDOM_ID")
+    monitor_key = "test-schedule-change-{RANDOM_ID}".format(RANDOM_ID=random_id)
+    monitor = cronitor_wrapper.get_ci_monitor_by_key(monitor_key)
+    assert monitor is not None, f"no monitor with key {monitor_key} exists"
+    assert monitor['schedule'] == "*/5 */10 * * *", f"expected monitor schedule '*/5 */10 * * *', got '{monitor['schedule']}'"
+
+    new_schedule = "*/10 */50 * * *"
+    patch_cronjob_by_name("test-schedule-change", None, {"spec": {"schedule": new_schedule}})
+    cronitor_wrapper.bust_monitor_cache()
+    time.sleep(3)
+    monitor = cronitor_wrapper.get_ci_monitor_by_key(monitor_key)
+    assert monitor['schedule'] == new_schedule, f"expected monitor schedule '{new_schedule}', got '{monitor['schedule']}'"
+
+
 def test_monitor_created_with_specified_name():
     random_id = os.getenv("RANDOM_ID")
     monitor_name = "annotation-test-name-{RANDOM_ID}".format(RANDOM_ID=random_id)

e2e-test/kustomize/base/kustomization-sub.yaml

@@ -22,6 +22,14 @@ patches:
         path: /metadata/annotations/k8s.cronitor.io~1cronitor-name
         value: "annotation-test-name-${RANDOM_ID}"
 
+  - target:
+      kind: CronJob
+      name: test-schedule-change
+    patch: |-
+      - op: replace
+        path: /metadata/annotations/k8s.cronitor.io~1cronitor-id
+        value: "test-schedule-change-${RANDOM_ID}"
+
   - target:
       kind: CronJob
       name: test-cronjob-namespace

e2e-test/resources/kustomization.yaml

@@ -8,4 +8,5 @@ resources:
   - main/cronjob-env-home.yaml
   - main/cronjob-exclude.yaml
   - main/cronjob-fail.yaml
-  - main/scoped-serviceaccount.yaml
+  - main/scoped-serviceaccount.yaml
+  - main/cronjob-schedule-change.yaml

e2e-test/resources/main/cronjob-schedule-change.yaml

@@ -0,0 +1,19 @@
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: test-schedule-change
+  annotations:
+    k8s.cronitor.io/cronitor-id: ''
+spec:
+  schedule: "*/5 */10 * * *"
+  jobTemplate:
+    spec:
+      backoffLimit: 3
+      template:
+        spec:
+          containers:
+            - name: hello
+              image: busybox
+              args: [/bin/sh, -c, date ; sleep 5 ; echo Hello from k8s]
+          restartPolicy: OnFailure
+  concurrencyPolicy: Forbid

pkg/annotations.go

@@ -2,10 +2,11 @@ package pkg
 
 import (
 	"fmt"
-	v1 "k8s.io/api/batch/v1"
 	"os"
 	"strconv"
 	"strings"
+
+	v1 "k8s.io/api/batch/v1"
 )
 
 type defaultBehaviorValue string
@@ -74,6 +75,10 @@ func (cronitorParser CronitorConfigParser) GetEnvironment() string {
 	return ""
 }
 
+func (cronitorParser CronitorConfigParser) GetSchedule() string {
+	return cronitorParser.cronjob.Spec.Schedule
+}
+
 func (cronitorParser CronitorConfigParser) GetTags() []string {
 	var tagList []string
 

pkg/annotations_test.go

@@ -26,3 +26,21 @@ func TestCronJobInclusion(t *testing.T) {
 		t.Errorf("cronjob.IsCronJobIncluded() = %s; wanted true", strconv.FormatBool(got))
 	}
 }
+
+func TestGetSchedule(t *testing.T) {
+	var jsonBlob v1.CronJobList
+	err := json.Unmarshal([]byte(`{"metadata":{"selfLink":"/apis/batch/v1beta1/cronjobs","resourceVersion":"41530"},"items":[{"metadata":{"name":"eventrouter-test-croonjob","namespace":"cronitor","selfLink":"/apis/batch/v1beta1/namespaces/cronitor/cronjobs/eventrouter-test-croonjob","uid":"a4892036-090f-4019-8bd1-98bfe0a9034c","resourceVersion":"41467","creationTimestamp":"2020-11-13T06:06:44Z","annotations":{"k8s.cronitor.io/env": "test-env"},"labels":{"app.kubernetes.io/managed-by":"skaffold","skaffold.dev/run-id":"a592b4e3-dd8e-4b25-a69f-7abe35e264f0"},"managedFields":[{"manager":"Go-http-client","operation":"Update","ApiVersion":"batch/v1beta1","time":"2020-11-13T06:06:44Z","fieldsType":"FieldsV1","fieldsV1":{"f:spec":{"f:concurrencyPolicy":{},"f:failedJobsHistoryLimit":{},"f:jobTemplate":{"f:spec":{"f:template":{"f:spec":{"f:containers":{"k:{\"name\":\"hello\"}":{".":{},"f:args":{},"f:image":{},"f:imagePullPolicy":{},"f:name":{},"f:resources":{},"f:terminationMessagePath":{},"f:terminationMessagePolicy":{}}},"f:dnsPolicy":{},"f:restartPolicy":{},"f:schedulerName":{},"f:securityContext":{},"f:terminationGracePeriodSeconds":{}}}}},"f:schedule":{},"f:successfulJobsHistoryLimit":{},"f:suspend":{}}}},{"manager":"skaffold","operation":"Update","ApiVersion":"batch/v1beta1","time":"2020-11-13T06:06:45Z","fieldsType":"FieldsV1","fieldsV1":{"f:metadata":{"f:labels":{".":{},"f:app.kubernetes.io/managed-by":{},"f:skaffold.dev/run-id":{}}}}},{"manager":"kube-controller-manager","operation":"Update","ApiVersion":"batch/v1beta1","time":"2020-11-13T07:57:06Z","fieldsType":"FieldsV1","fieldsV1":{"f:status":{"f:active":{},"f:lastScheduleTime":{}}}}]},"spec":{"schedule":"*/1 * * * *","concurrencyPolicy":"Forbid","suspend":false,"jobTemplate":{"metadata":{"creationTimestamp":null},"spec":{"template":{"metadata":{"creationTimestamp":null},"spec":{"containers":[{"name":"hello","image":"busybox","args":["/bin/sh","-c","date ; echo Hello from k8s"],"resources":{},"terminationMessagePath":"/dev/termination-log","terminationMessagePolicy":"File","imagePullPolicy":"Always"}],"restartPolicy":"OnFailure","terminationGracePeriodSeconds":30,"dnsPolicy":"ClusterFirst","securityContext":{},"schedulerName":"default-scheduler"}}}},"successfulJobsHistoryLimit":3,"failedJobsHistoryLimit":1},"status":{"active":[{"kind":"Job","namespace":"cronitor","name":"eventrouter-test-croonjob-1605254220","uid":"697df5f5-6366-42fe-a20e-19ec2fefd826","ApiVersion":"batch/v1","resourceVersion":"41465"}],"lastScheduleTime":"2020-11-13T07:57:00Z"}}]}`), &jsonBlob)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	got := len(jsonBlob.Items)
+	if got != 1 {
+		t.Errorf("len(CronJobs) = %d; want 1", got)
+	}
+
+	parser := NewCronitorConfigParser(&jsonBlob.Items[0])
+	if result := parser.GetSchedule(); result != "*/1 * * * *" {
+		t.Errorf("expected schedule \"*/1 * * * *\", got %s", result)
+	}
+}

pkg/collector/cronjob_watcher.go

@@ -2,6 +2,7 @@ package collector
 
 import (
 	"fmt"
+
 	"github.com/cronitorio/cronitor-kubernetes/pkg"
 	"github.com/cronitorio/cronitor-kubernetes/pkg/normalizer"
 	log "github.com/sirupsen/logrus"
@@ -43,9 +44,20 @@ func onUpdate(coll CronJobCollection, cronjobOld *v1.CronJob, cronjobNew *v1.Cro
 	} else if wasIncluded && !nowIncluded {
 		onDelete(coll, cronjobOld)
 	} else if wasIncluded && nowIncluded {
-		// Otherwise, if we're keeping it around, check if there are any changes to
-		// configurable annotations and handle accordingly.
-		// Right now we don't actually have any logic to put here, but we might down the line.
+		log.WithFields(log.Fields{
+			"namespace":   cronjobNew.Namespace,
+			"name":        cronjobNew.Name,
+			"UID":         cronjobNew.UID,
+			"oldSchedule": cronjobOld.Spec.Schedule,
+			"newSchedule": cronjobNew.Spec.Schedule,
+			"configOld":   configParserOld.GetSchedule(),
+			"configNew":   configParserNew.GetSchedule(),
+		}).Info("cronjob updated")
+
+		// If the schedule is updated
+		if configParserOld.GetSchedule() != configParserNew.GetSchedule() {
+		  onAdd(coll, cronjobNew)
+		}
 	}
 }