Add scheduled delete

azimuth_schedule_operator/models/v1alpha1/schedule.py

@@ -10,18 +10,18 @@ class ScheduleStatus(schema.BaseModel):
     ref_found: bool = False
     # updated when delete has been triggered
     delete_triggered: bool = False
+    updated_at: schema.Optional[datetime.datetime] = None
 
 
 class ScheduleRef(schema.BaseModel):
-    apiVersion: str
+    api_version: str
     kind: str
     name: str
 
 
 class ScheduleSpec(schema.BaseModel):
     ref: ScheduleRef
-    notBefore: datetime.datetime
-    notAfter: datetime.datetime
+    not_after: datetime.datetime
 
 
 class Schedule(
@@ -44,8 +44,6 @@ def get_fake_dict():
         metadata=dict(name="test1", uid="fakeuid1", namespace="ns1"),
         spec=dict(
             ref=dict(apiVersion="v1", kind="Pod", name="test1"),
-            notBefore=datetime.datetime.now(),
-            notAfter=datetime.datetime.now() + datetime.timedelta(days=1),
+            notAfter=datetime.datetime.now(datetime.timezone.utc),
         ),
-        status=dict(ref_found=True, delete_triggered=False),
     )

azimuth_schedule_operator/operator.py

@@ -1,4 +1,5 @@
 import asyncio
+import datetime
 import logging
 import os
 import sys
@@ -12,6 +13,8 @@
 LOG = logging.getLogger(__name__)
 K8S_CLIENT = None
 
+CHECK_INTERVAL_SECONDS = int(os.environ.get("AZIMUTH_SCHEDULE_CHECK_INTERVAL", "120"))
+
 
 @kopf.on.startup()
 async def startup(settings, **kwargs):
@@ -46,15 +49,73 @@ async def startup(settings, **kwargs):
 
 
 @kopf.on.cleanup()
-async def cleanup(**kwargs):
+async def cleanup(**_):
     if K8S_CLIENT:
         await K8S_CLIENT.aclose()
     LOG.info("Cleanup complete.")
 
 
-@kopf.on.create(registry.API_GROUP, "schedule")
-@kopf.on.update(registry.API_GROUP, "schedule")
-@kopf.on.resume(registry.API_GROUP, "schedule")
-async def schedule_changed(body, name, namespace, labels, **kwargs):
+async def get_reference(namespace: str, ref: schedule_crd.ScheduleRef):
+    resource = await K8S_CLIENT.api(ref.api_version).resource(ref.kind)
+    object = await resource.fetch(ref.name, namespace=namespace)
+    return object
+
+
+async def delete_reference(namespace: str, ref: schedule_crd.ScheduleRef):
+    resource = await K8S_CLIENT.api(ref.api_version).resource(ref.kind)
+    await resource.delete(ref.name, namespace=namespace)
+
+
+async def check_for_delete(namespace, schedule: schedule_crd.Schedule):
+    # TODO(johngarbutt): add some config in helm to set this?
+    max_delete_duration_int = int(
+        os.environ.get("AZIMUTH_SCHEDULE_MAX_DELETE_DURATION_MINUTES", "15")
+    ) + int(CHECK_INTERVAL_SECONDS / 60)
+    max_delete_duration = datetime.timedelta(minutes=max_delete_duration_int)
+    scheduled_at = schedule.spec.not_after - max_delete_duration
+    now = datetime.datetime.now(datetime.timezone.utc)
+
+    if now >= scheduled_at:
+        LOG.info(f"Attempting delete for {namespace} and {schedule.metadata.name}.")
+        await delete_reference(namespace, schedule.spec.ref)
+        await update_schedule(schedule.metadata.name, namespace, delete_triggered=True)
+    else:
+        LOG.info(f"No delete for {namespace} and {schedule.metadata.name}.")
+
+
+async def update_schedule(
+    name: str,
+    namespace: str,
+    ref_found: bool = None,
+    delete_triggered: bool = None,
+):
+    now = datetime.datetime.now(datetime.timezone.utc)
+    now_string = now.strftime("%Y-%m-%dT%H:%M:%SZ")
+    status_updates = dict(updated_at=now_string)
+
+    if ref_found is not None:
+        status_updates["refFound"] = ref_found
+    if delete_triggered is not None:
+        status_updates["deleteTriggered"] = delete_triggered
+
+    status_resource = await K8S_CLIENT.api(registry.API_VERSION).resource(
+        "schedules/status"
+    )
+    LOG.info(f"Updating status for {name} in {namespace} with: {status_updates}")
+    await status_resource.patch(
+        name,
+        dict(status=status_updates),
+        namespace=namespace,
+    )
+
+
+# check every two minutes
+@kopf.timer(registry.API_GROUP, "schedule", interval=CHECK_INTERVAL_SECONDS)
+async def schedule_check(body, namespace, **_):
     schedule = schedule_crd.Schedule(**body)
-    LOG.error(f"seen schedule changed {schedule.spec}")
+
+    if not schedule.status.ref_found:
+        await get_reference(namespace, schedule.spec.ref)
+        await update_schedule(schedule.metadata.name, namespace, ref_found=True)
+
+    await check_for_delete(namespace, schedule)

azimuth_schedule_operator/tests/models/test_crds.py

@@ -62,18 +62,13 @@ def test_cluster_type_crd_json(self):
                     ],
                     "type": "object"
                   },
-                  "notBefore": {
-                    "format": "date-time",
-                    "type": "string"
-                  },
                   "notAfter": {
                     "format": "date-time",
                     "type": "string"
                   }
                 },
                 "required": [
                   "ref",
-                  "notBefore",
                   "notAfter"
                 ],
                 "type": "object"
@@ -85,6 +80,11 @@ def test_cluster_type_crd_json(self):
                   },
                   "deleteTriggered": {
                     "type": "boolean"
+                  },
+                  "updatedAt": {
+                    "format": "date-time",
+                    "nullable": true,
+                    "type": "string"
                   }
                 },
                 "type": "object"

azimuth_schedule_operator/tests/test_operator.py

@@ -48,5 +48,34 @@ async def test_cleanup_calls_aclose(self, mock_client):
         await operator.cleanup()
         mock_client.aclose.assert_awaited_once_with()
 
-    async def test_cluster_type_create_success(self):
-        await operator.schedule_changed(schedule_crd.get_fake_dict(), "type1", "ns", {})
+    @mock.patch.object(operator, "update_schedule")
+    @mock.patch.object(operator, "check_for_delete")
+    @mock.patch.object(operator, "get_reference")
+    async def test_schedule_check(
+        self, mock_get_reference, mock_check_for_delete, mock_update_schedule
+    ):
+        body = schedule_crd.get_fake_dict()
+        fake = schedule_crd.Schedule(**body)
+        namespace = "ns1"
+
+        await operator.schedule_check(body, namespace)
+
+        # Assert the expected behavior
+        mock_get_reference.assert_awaited_once_with(namespace, fake.spec.ref)
+        mock_check_for_delete.assert_awaited_once_with(namespace, fake)
+        mock_update_schedule.assert_awaited_once_with(
+            fake.metadata.name, namespace, ref_found=True
+        )
+
+    @mock.patch.object(operator, "update_schedule")
+    @mock.patch.object(operator, "delete_reference")
+    async def test_check_for_delete(self, mock_delete_reference, mock_update_schedule):
+        namespace = "ns1"
+        schedule = schedule_crd.get_fake()
+
+        await operator.check_for_delete(namespace, schedule)
+
+        mock_delete_reference.assert_awaited_once_with(namespace, schedule.spec.ref)
+        mock_update_schedule.assert_awaited_once_with(
+            schedule.metadata.name, namespace, delete_triggered=True
+        )

charts/operator/templates/clusterrole-operator.yaml

@@ -18,8 +18,12 @@ rules:
   - apiGroups: ["scheduling.azimuth.stackhpc.com"]
     resources: ["*"]
     verbs: ["*"]
+  # allow these things to be deleted by the operator
   - apiGroups: ["caas.azimuth.stackhpc.com"]
-    resources: ["*"]
+    resources: ["clusters"]
+    verbs: ["*"]
+  - apiGroups: [""]
+    resources: ["configmaps"]
     verbs: ["*"]
   - apiGroups: ["rbac.authorization.k8s.io"]
     resources: ["clusterrolebindings"]

charts/operator/templates/prometheusrule.yaml

@@ -10,14 +10,14 @@ spec:
       rules:
         - alert: AzimuthCaasClusterNotReady
           expr: >-
-            sum(azimuth_schedule_clusters_phase{phase!="Ready"}) by(cluster_namespace, cluster_name) > 0
-          for: 1h
+            sum(azimuth_schedule_delete_triggered{refFound!="True"}) by(schedule_namespace, schedule_name) > 0
+          for: 15m
           annotations:
             description: >-
               Azimuth schedule
-              {{ "{{" }} $labels.cluster_namespace {{ "}}" }}/{{ "{{" }} $labels.cluster_name {{ "}}" }}
-              has been in a non-ready state for longer than one hour.
-            summary: Azimuth schedule has been in a non-ready state for more than one hour.
+              {{ "{{" }} $labels.schedule_namespace {{ "}}" }}/{{ "{{" }} $labels.schedule_name {{ "}}" }}
+              has been in a not found state for longer than 15 mins.
+            summary: Azimuth schedule has not found its ref for longer than 15 mins.
           labels:
             severity: warning
 {{- end }}

charts/operator/values.yaml

@@ -1,7 +1,3 @@
-# Config for the operator
-config:
-  placeholder: "asdf"
-
 # The operator image to use
 image:
   repository: ghcr.io/stackhpc/azimuth-schedule-operator
@@ -70,7 +66,7 @@ metrics:
     create: true
     extraRules:
       - apiGroups:
-          - schedule.azimuth.stackhpc.com
+          - scheduling.azimuth.stackhpc.com
         resources:
           - schedules
         verbs:
@@ -87,20 +83,21 @@ metrics:
       spec:
         resources:
           - groupVersionKind:
-              group: schedule.azimuth.stackhpc.com
+              group: scheduling.azimuth.stackhpc.com
               version: v1alpha1
               kind: Schedule
-            metricNamePrefix: azimuth_schedule_clusters
+            metricNamePrefix: azimuth_schedule
             labelsFromPath:
-              cluster_namespace: [metadata, namespace]
-              cluster_name: [metadata, name]
-              cluster_type_name: [spec, clusterTypeName]
-              cluster_type_version: [spec, clusterTypeVersion]
+              schedule_namespace: [metadata, namespace]
+              schedule_name: [metadata, name]
+              schedule_ref_kind: [spec, ref, kind]
+              schedule_ref_name: [spec, ref, name]
+              schedule_ref_found: [status, refFound]
             metrics:
-              - name: phase
+              - name: delete_triggered
                 help: "Schedule phase"
                 each:
                   type: Info
                   info:
                     labelsFromPath:
-                      phase: [status, phase]
+                      phase: [status, delete_triggered]

tools/functional_test.sh

@@ -19,9 +19,12 @@ until [ `kubectl get crds | grep schedules.scheduling.azimuth.stackhpc.com | wc
 kubectl get crds
 
 
-export BEFORE=$(date --date="-1 hour" +"%Y-%m-%dT%H:%M:%SZ")
-export AFTER=$(date --date="+2 hour" +"%Y-%m-%dT%H:%M:%SZ")
+export AFTER=$(date --date="-1 hour" +"%Y-%m-%dT%H:%M:%SZ")
 envsubst < $SCRIPT_DIR/test_schedule.yaml | kubectl apply -f -
 
-# until kubectl wait --for=jsonpath='{.status.phase}'=Available clustertype quick-test; do echo "wait for status to appear"; sleep 5; done
+until kubectl wait --for=jsonpath='{.status.refFound}'=true schedule caas-mycluster; do echo "wait for refFound"; sleep 5; done
+until kubectl wait --for=jsonpath='{.status.deleteTriggered}'=true schedule caas-mycluster; do echo "wait for deleteTriggered"; sleep 5; done
 kubectl get schedule caas-mycluster -o yaml
+
+# for debugging get the logs from the operator
+kubectl logs -n azimuth-schedule-operator deployment/azimuth-schedule-operator

tools/test_schedule.yaml

@@ -1,13 +1,20 @@
 ---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: caas-mycluster
+data:
+  key1: value1
+  key2: value2
+---
 apiVersion: scheduling.azimuth.stackhpc.com/v1alpha1
 kind: Schedule
 metadata:
   name: caas-mycluster
   ownerReferences: []
 spec:
   ref:
-    apiVersion: caas.azimuth.stackhpc.com/v1alpha1
-    kind: Cluster
-    name: mycluster
-  notBefore: $BEFORE
+    apiVersion: v1
+    kind: ConfigMap
+    name: caas-mycluster
   notAfter: $AFTER

tox.ini

@@ -67,7 +67,7 @@ commands =
 # E123, E125 skipped as they are invalid PEP-8.
 show-source = True
 # TODO add headers and remove H102
-ignore = E123,E125,H102
+ignore = E123,E125,H102,W503
 builtins = _
 exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,build
 # match black