fix: patchStrategicMerge now works for kind:Job

When a job is invoked, we now replace the job in k8s/jobs.yml
instead of rewriting jobs.yml to only contain the relevant
job. This allows patchStrategicMerge to work for jobs.

changelog.d/20230214_105510_keith_fix_jobs_merge.md

@@ -0,0 +1,12 @@
+<!--
+Create a changelog entry for every new user-facing change. Please respect the following instructions:
+- Indicate breaking changes by prepending an explosion 💥 character.
+- Prefix your changes with either [Bugfix], [Improvement], [Feature], [Security], [Deprecation].
+- You may optionally append "(by @<author>)" at the end of the line, where "<author>" is either one (just one)
+  of your GitHub username, real name or affiliated organization. These affiliations will be displayed in
+  the release notes for every release.
+-->
+
+<!-- - 💥[Feature] Foobarize the blorginator. This breaks plugins by renaming the `FOO_DO` filter to `BAR_DO`. (by @regisb) -->
+<!-- - [Improvement] This is a non-breaking change. Life is good. (by @billgates) -->
+[Bugfix] `patchStrategicMerge` can now be applied to jobs (by @keithgg)

tutor/commands/k8s.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 from time import sleep
-from typing import Any, List, Optional, Type
+from typing import Any, List, Optional, Type, Iterable
 
 import click
 
@@ -64,11 +64,12 @@ class K8sTaskRunner(BaseTaskRunner):
     """
 
     def run_task(self, service: str, command: str) -> int:
-        job_name = f"{service}-job"
+        canonical_job_name = f"{service}-job"
-        job = self.load_job(job_name)
+        all_jobs = list(self._load_jobs())
+        job = self._find_job(canonical_job_name, all_jobs)
         # Create a unique job name to make it deduplicate jobs and make it easier to
         # find later. Logs of older jobs will remain available for some time.
-        job_name += "-" + datetime.now().strftime("%Y%m%d%H%M%S")
+        job_name = canonical_job_name + "-" + datetime.now().strftime("%Y%m%d%H%M%S")
 
         # Wait until all other jobs are completed
         while True:
@@ -98,11 +99,12 @@ class K8sTaskRunner(BaseTaskRunner):
         job["spec"]["template"]["spec"]["containers"][0]["args"] = container_args
         job["spec"]["backoffLimit"] = 1
         job["spec"]["ttlSecondsAfterFinished"] = 3600
-        # Save patched job to "jobs.yml" file
+
         with open(
             tutor_env.pathjoin(self.root, "k8s", "jobs.yml"), "w", encoding="utf-8"
         ) as job_file:
-            serialize.dump(job, job_file)
+            serialize.dump_all(all_jobs, job_file)
+
         # We cannot use the k8s API to create the job: configMap and volume names need
         # to be found with the right suffixes.
         kubectl_apply(
@@ -143,8 +145,15 @@ class K8sTaskRunner(BaseTaskRunner):
         """
         Find a given job definition in the rendered k8s/jobs.yml template.
         """
-        all_jobs = self.render("k8s", "jobs.yml")
+        return self._find_job(name, self._load_jobs())
-        for job in serialize.load_all(all_jobs):
+
+    def _find_job(self, name: str, all_jobs: Iterable[Any]) -> Any:
+        """
+        Find the matching job definition in the in the list of jobs provided.
+
+        Returns the found job's manifest.
+        """
+        for job in all_jobs:
             job_name = job["metadata"]["name"]
             if not isinstance(job_name, str):
                 raise exceptions.TutorError(
@@ -154,6 +163,12 @@ class K8sTaskRunner(BaseTaskRunner):
                 return job
         raise exceptions.TutorError(f"Could not find job '{name}'")
 
+    def _load_jobs(self) -> Iterable[Any]:
+        manifests = self.render("k8s", "jobs.yml")
+        for manifest in serialize.load_all(manifests):
+            if manifest["kind"] == "Job":
+                yield manifest
+
     def active_job_names(self) -> List[str]:
         """
         Return a list of active job names

tutor/serialize.py

@@ -17,6 +17,10 @@ def load_all(stream: str) -> t.Iterator[t.Any]:
     return yaml.load_all(stream, Loader=yaml.SafeLoader)
 
 
+def dump_all(documents: t.Sequence[t.Any], fileobj: TextIOWrapper) -> None:
+    yaml.safe_dump_all(documents, stream=fileobj, default_flow_style=False)
+
+
 def dump(content: t.Any, fileobj: TextIOWrapper) -> None:
     yaml.dump(content, stream=fileobj, default_flow_style=False)