Add Job support

Author: pabloxio

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

@@ -6,6 +6,7 @@
 
 
 @when("the CronJob called {cronjob_name} is created")
+@given("the CronJob called {cronjob_name} exists")
 def step_impl(context, cronjob_name):
     cronjob_namespace = "default"
 
@@ -31,29 +32,13 @@ def step_impl(context, cronjob_name):
     cronjob.delete(context.k8s_v1_batch_client, cronjob_name)
 
 
-@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
-    )
-
-
 @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 hello are returned")
-def step_impl(context):
+@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"
 
 

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/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/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