Merge pull request #8 from LF-Engineering/feature/cronjob

Add support for CronJob/Job

Author: pabloxio

features/cronjob.feature

@@ -0,0 +1,18 @@
+Feature: Manage CronJobs
+
+  Scenario: Try retrieve a CronJob that does not exist
+    Given a CronJob called hello does not exist
+      When the user attempts to retrieve the CronJob hello
+        Then None is returned instead of the CronJob
+
+
+  Scenario: Create a CronJob
+    Given a CronJob called salamander does not exist
+      When the CronJob called salamander is created
+        Then a valid CronJob called salamander can be found
+
+
+  Scenario: Retrieve a CronJob that exists
+    Given the CronJob called hello exists
+      When the user attempts to retrieve the CronJob hello
+        Then Results for the CronJob hello are returned

features/environment.py

@@ -6,3 +6,4 @@ def before_all(context):
     config.load_kube_config(context="minikube")
     context.k8s_v1_core_client = client.CoreV1Api()
     context.k8s_v1_apps_client = client.AppsV1Api()
+    context.k8s_v1_batch_client = client.BatchV1Api()

features/job.feature

@@ -0,0 +1,18 @@
+Feature: Manage Jobs
+
+  Scenario: Try retrieve a Job that does not exist
+    Given a Job called sleepy-bohr does not exist
+      When the user attempts to retrieve the Job sleepy-bohr
+        Then None is returned instead of the Job
+
+
+  Scenario: Create a Job
+    Given a Job called strange-bhaskara does not exist
+      When the Job called strange-bhaskara is created
+        Then a valid Job called strange-bhaskara can be found
+
+
+  Scenario: Retrieve a Job that exists
+    Given the Job called lucid-lovelace exists
+      When the user attempts to retrieve the Job lucid-lovelace
+        Then Results for the Job lucid-lovelace are returned

features/steps/cronjob.py

@@ -0,0 +1,47 @@
+from behave import *
+
+from kelpy import cronjob
+from kubernetes import config, client
+from jinja2 import Environment, FileSystemLoader
+
+
+@when("the CronJob called {cronjob_name} is created")
+@given("the CronJob called {cronjob_name} exists")
+def step_impl(context, cronjob_name):
+    cronjob_namespace = "default"
+
+    env = Environment(
+        loader=FileSystemLoader("./templates"), trim_blocks=True, lstrip_blocks=True
+    )
+    cronjob_tmpl = env.get_template("cronjob.j2")
+    cronjob_spec_file = cronjob_tmpl.render(
+        cronjob_name=cronjob_name, cronjob_namespace=cronjob_namespace
+    )
+    context.create_resp = cronjob.create(
+        context.k8s_v1_batch_client, cronjob_spec_file, cronjob_namespace
+    )
+
+
+@when("the user attempts to retrieve the CronJob {cronjob_name}")
+def step_impl(context, cronjob_name):
+    context.get_resp = cronjob.get(context.k8s_v1_batch_client, cronjob_name)
+
+
+@given("a CronJob called {cronjob_name} does not exist")
+def step_impl(context, cronjob_name):
+    cronjob.delete(context.k8s_v1_batch_client, cronjob_name)
+
+
+@then("None is returned instead of the CronJob")
+def step_impl(context):
+    assert context.get_resp is None, "Did not return None"
+
+
+@then("Results for the CronJob {cronjob_name} are returned")
+def step_impl(context, cronjob_name):
+    assert context.get_resp, "Should've returned a valid response"
+
+
+@then("a valid CronJob called {cronjob_name} can be found")
+def step_impl(context, cronjob_name):
+    assert context.create_resp, f"Should've found {cronjob_name} CronJob"

features/steps/job.py

@@ -0,0 +1,45 @@
+from behave import *
+
+from kelpy import job
+from kubernetes import config, client
+from jinja2 import Environment, FileSystemLoader
+
+
+@given("a Job called {job_name} does not exist")
+def step_impl(context, job_name):
+    job.delete(context.k8s_v1_batch_client, job_name)
+
+
+@given("the Job called {job_name} exists")
+@when("the Job called {job_name} is created")
+def step_impl(context, job_name):
+    job_namespace = "default"
+
+    env = Environment(
+        loader=FileSystemLoader("./templates"), trim_blocks=True, lstrip_blocks=True
+    )
+    job_tmpl = env.get_template("job.j2")
+    job_spec_file = job_tmpl.render(job_name=job_name, job_namespace=job_namespace)
+    context.create_resp = job.create(
+        context.k8s_v1_batch_client, job_spec_file, job_namespace
+    )
+
+
+@when("the user attempts to retrieve the Job {job_name}")
+def step_impl(context, job_name):
+    context.get_resp = job.get(context.k8s_v1_batch_client, job_name)
+
+
+@then("None is returned instead of the Job")
+def step_impl(context):
+    assert context.get_resp is None, "Did not return None"
+
+
+@then("a valid Job called {job_name} can be found")
+def step_impl(context, job_name):
+    assert context.create_resp, f"Should've found {job_name} Job"
+
+
+@then("Results for the Job {job_name} are returned")
+def step_impl(context, job_name):
+    assert context.get_resp, "Should've returned a valid response"

kelpy/cronjob.py

@@ -0,0 +1,91 @@
+import yaml
+from kubernetes.client.exceptions import ApiException
+from kubernetes import watch
+
+
+def create(client, spec: str, namespace: str = "default", timeout=100):
+    """Create a CronJob.
+
+    :batch_v1_api: The Batch V1 API object.
+    :spec: A valid CronJob YAML manifest.
+    :namespace: The namespace of the CronJob.
+    :timeout: Timeout in seconds to wait for object creation/modification
+
+    :returns: True on creation, False if it already exists.
+    """
+    body = yaml.safe_load(spec)
+
+    try:
+        response = client.create_namespaced_cron_job(namespace, body)
+    except ApiException as e:
+        # If the object already exists, return False.
+        if e.reason == "Conflict":
+            return False
+        raise e
+
+    name = body["metadata"]["name"]
+    if get(client, name, namespace) is None:
+        w = watch.Watch()
+        for event in w.stream(
+            client.list_cron_job_for_all_namespaces, timeout_seconds=timeout
+        ):
+            if (
+                (event["type"] == "ADDED" or event["type"] == "MODIFIED")
+                and event["object"].kind == "CronJob"
+                and event["object"].metadata.name == response.metadata.name
+                and event["object"].metadata.namespace == response.metadata.namespace
+            ):
+                break
+
+    return response
+
+
+def get(client, name: str, namespace: str = "default"):
+    """Get a CronJob if it does exist, if it doesn't, return None.
+
+    :batch_v1_api: The Batch V1 API object.
+    :name: The name of the CronJob.
+    :namespace: The namespace of the CronJob.
+
+    :returns: CronJob if it does exist, None if it doesn't exist.
+    """
+    field_selector = "metadata.name={0}".format(name)
+
+    response = client.list_namespaced_cron_job(namespace, field_selector=field_selector)
+
+    if len(response.items) == 1:
+        return response
+
+    return None
+
+
+def delete(client, name: str, namespace: str = "default", timeout=100):
+    """Delete a CronJob if it does exist, if it doesn't, return False.
+
+    :batch_v1_api: The Batch V1 API object.
+    :name: The name of the CronJob.
+    :namespace: The namespace of the CronJob.
+    :timeout: Timeout in seconds to wait for object deletion
+
+    :returns: Batch API Response if it does exist, False if it doesn't exist.
+    """
+    try:
+        response = client.delete_namespaced_cron_job(name, namespace)
+    except ApiException as e:
+        if e.reason == "Not Found":
+            return False
+        raise e
+
+    if get(client, name, namespace) is not None:
+        w = watch.Watch()
+        for event in w.stream(
+            client.list_cron_job_for_all_namespaces, timeout_seconds=timeout
+        ):
+            if (
+                event["type"] == "DELETED"
+                and event["object"].metadata.name == name
+                and event["object"].metadata.namespace == namespace
+            ):
+                break
+
+    return response

kelpy/job.py

@@ -0,0 +1,91 @@
+import yaml
+from kubernetes.client.exceptions import ApiException
+from kubernetes import watch
+
+
+def create(client, spec: str, namespace: str = "default", timeout=100):
+    """Create a Job.
+
+    :batch_v1_api: The Batch V1 API object.
+    :spec: A valid Job YAML manifest.
+    :namespace: The namespace of the Job.
+    :timeout: Timeout in seconds to wait for object creation/modification
+
+    :returns: True on creation, False if it already exists.
+    """
+    body = yaml.safe_load(spec)
+
+    try:
+        response = client.create_namespaced_job(namespace, body)
+    except ApiException as e:
+        # If the object already exists, return False.
+        if e.reason == "Conflict":
+            return False
+        raise e
+
+    name = body["metadata"]["name"]
+    if get(client, name, namespace) is None:
+        w = watch.Watch()
+        for event in w.stream(
+            client.list_job_for_all_namespaces, timeout_seconds=timeout
+        ):
+            if (
+                (event["type"] == "ADDED" or event["type"] == "MODIFIED")
+                and event["object"].kind == "Job"
+                and event["object"].metadata.name == response.metadata.name
+                and event["object"].metadata.namespace == response.metadata.namespace
+            ):
+                break
+
+    return response
+
+
+def get(client, name: str, namespace: str = "default"):
+    """Get a Job if it does exist, if it doesn't, return None.
+
+    :batch_v1_api: The Batch V1 API object.
+    :name: The name of the Job.
+    :namespace: The namespace of the Job.
+
+    :returns: Job if it does exist, None if it doesn't exist.
+    """
+    field_selector = "metadata.name={0}".format(name)
+
+    response = client.list_namespaced_job(namespace, field_selector=field_selector)
+
+    if len(response.items) == 1:
+        return response
+
+    return None
+
+
+def delete(client, name: str, namespace: str = "default", timeout=100):
+    """Delete a Job if it does exist, if it doesn't, return False.
+
+    :batch_v1_api: The Batch V1 API object.
+    :name: The name of the Job.
+    :namespace: The namespace of the Job.
+    :timeout: Timeout in seconds to wait for object deletion
+
+    :returns: Batch API Response if it does exist, False if it doesn't exist.
+    """
+    try:
+        response = client.delete_namespaced_job(name, namespace)
+    except ApiException as e:
+        if e.reason == "Not Found":
+            return False
+        raise e
+
+    if get(client, name, namespace) is not None:
+        w = watch.Watch()
+        for event in w.stream(
+            client.list_job_for_all_namespaces, timeout_seconds=timeout
+        ):
+            if (
+                event["type"] == "DELETED"
+                and event["object"].metadata.name == name
+                and event["object"].metadata.namespace == namespace
+            ):
+                break
+
+    return response

templates/cronjob.j2

@@ -0,0 +1,20 @@
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: {{ cronjob_name }}
+  namespace: {{ cronjob_namespace }}
+spec:
+  schedule: "*/1 * * * *"
+  jobTemplate:
+    spec:
+      template:
+        spec:
+          containers:
+          - name: hello
+            image: busybox
+            imagePullPolicy: IfNotPresent
+            command:
+            - /bin/sh
+            - -c
+            - date; echo Hello from the Kubernetes cluster
+          restartPolicy: OnFailure

templates/job.j2

@@ -0,0 +1,14 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: {{ job_name }}
+  namespace: {{ job_namespace }}
+spec:
+  template:
+    spec:
+      containers:
+      - name: pi
+        image: perl
+        command: ["perl",  "-Mbignum=bpi", "-wle", "print bpi(2000)"]
+      restartPolicy: Never
+  backoffLimit: 4