Add DaskJob resource

ci/pre-commit-crd.py

@@ -82,6 +82,7 @@ def main(version, *args):
             # TODO: Not sure how to do this other than hard-code the template names
             run_action("daskcluster.yaml", temp_dir, crd_path, output_paths)
             run_action("daskworkergroup.yaml", temp_dir, crd_path, output_paths)
+            run_action("daskjob.yaml", temp_dir, crd_path, output_paths)
 
         else:
             run_action(changed_file, temp_dir, crd_path, output_paths)

dask_kubernetes/conftest.py

@@ -65,16 +65,12 @@ def customresources(k8s_cluster):
     temp_dir = tempfile.TemporaryDirectory()
     crd_path = os.path.join(DIR, "operator", "customresources")
 
-    run_generate(
-        os.path.join(crd_path, "daskcluster.yaml"),
-        os.path.join(crd_path, "daskcluster.patch.yaml"),
-        os.path.join(temp_dir.name, "daskcluster.yaml"),
-    )
-    run_generate(
-        os.path.join(crd_path, "daskworkergroup.yaml"),
-        os.path.join(crd_path, "daskworkergroup.patch.yaml"),
-        os.path.join(temp_dir.name, "daskworkergroup.yaml"),
-    )
+    for crd in ["daskcluster", "daskworkergroup", "daskjob"]:
+        run_generate(
+            os.path.join(crd_path, f"{crd}.yaml"),
+            os.path.join(crd_path, f"{crd}.patch.yaml"),
+            os.path.join(temp_dir.name, f"{crd}.yaml"),
+        )
 
     k8s_cluster.kubectl("apply", "-f", temp_dir.name)
     yield

dask_kubernetes/operator/customresources/daskcluster.yaml

@@ -18,22 +18,9 @@ spec:
       storage: true
       schema:
         openAPIV3Schema:
-          type: object
-          properties:
-            status:
-              type: object
-            spec:
-              type: object
-              required:
-              - scheduler
-              - worker
-              properties:
-                scheduler:
-                  $ref: 'python://dask_kubernetes/operator/customresources/templates.yaml#/definitions/dask.k8s.api.v1.DaskScheduler'
-                worker:
-                  $ref: 'python://dask_kubernetes/operator/customresources/templates.yaml#/definitions/dask.k8s.api.v1.DaskWorker' 
-
+          $ref: 'python://dask_kubernetes/operator/customresources/templates.yaml#/definitions/dask.k8s.api.v1.DaskCluster'
+
       subresources:
         scale:
           specReplicasPath: .spec.worker.replicas
-          statusReplicasPath: .status.replicas
+          statusReplicasPath: .status.replicas

dask_kubernetes/operator/customresources/daskjob.patch.yaml

@@ -0,0 +1,10 @@
+[
+  {"op": "add", "path": "/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties/job/properties/spec/properties/containers/items/properties/ports/items/properties/protocol/default", "value": "TCP"},
+  {"op": "add", "path": "/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties/job/properties/spec/properties/initContainers/items/properties/ports/items/properties/protocol/default", "value": "TCP"},
+  {"op": "add", "path": "/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties/job/properties/spec/properties/restartPolicy/default", "value": "OnFailure"},
+  {"op": "add", "path": "/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties/cluster/properties/spec/properties/worker/properties/spec/properties/initContainers/items/properties/ports/items/properties/protocol/default", "value": "TCP"},
+  {"op": "add", "path": "/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties/cluster/properties/spec/properties/worker/properties/spec/properties/containers/items/properties/ports/items/properties/protocol/default", "value": "TCP"},
+  {"op": "add", "path": "/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties/cluster/properties/spec/properties/scheduler/properties/service/properties/ports/items/properties/protocol/default", "value": "TCP"},
+  {"op": "add", "path": "/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties/cluster/properties/spec/properties/scheduler/properties/spec/properties/containers/items/properties/ports/items/properties/protocol/default", "value": "TCP"},
+  {"op": "add", "path": "/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties/cluster/properties/spec/properties/scheduler/properties/spec/properties/initContainers/items/properties/ports/items/properties/protocol/default", "value": "TCP"}
+]

dask_kubernetes/operator/customresources/daskjob.yaml

@@ -0,0 +1,36 @@
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: daskjobs.kubernetes.dask.org
+spec:
+  scope: Namespaced
+  group: kubernetes.dask.org
+  names:
+    kind: DaskJob
+    plural: daskjobs
+    singular: daskjob
+    shortNames:
+      - djb
+  versions:
+    - name: v1
+      served: true
+      storage: true
+      schema:
+        openAPIV3Schema:
+          type: object
+          properties:
+            status:
+              type: object
+            spec:
+              type: object
+              required:
+              - cluster
+              - job
+              properties:
+                job:
+                  type: object
+                  properties:
+                    spec:
+                      $ref: 'python://k8s_crd_resolver/schemata/k8s-1.21.1.json#/definitions/io.k8s.api.core.v1.PodSpec'
+                cluster:
+                  $ref: 'python://dask_kubernetes/operator/customresources/templates.yaml#/definitions/dask.k8s.api.v1.DaskCluster'

dask_kubernetes/operator/customresources/templates.yaml

@@ -12,7 +12,7 @@ definitions:
         description: Number of workers to start
       spec:
         $ref: 'python://k8s_crd_resolver/schemata/k8s-1.21.1.json#/definitions/io.k8s.api.core.v1.PodSpec'
- 
+
   dask.k8s.api.v1.DaskScheduler:
     type: object
     description: Dask scheduler configuration
@@ -23,4 +23,23 @@ definitions:
       spec:
         $ref: 'python://k8s_crd_resolver/schemata/k8s-1.21.1.json#/definitions/io.k8s.api.core.v1.PodSpec'
       service:
-        $ref: 'python://k8s_crd_resolver/schemata/k8s-1.21.1.json#/definitions/io.k8s.api.core.v1.ServiceSpec'
+        $ref: 'python://k8s_crd_resolver/schemata/k8s-1.21.1.json#/definitions/io.k8s.api.core.v1.ServiceSpec'
+
+  dask.k8s.api.v1.DaskCluster:
+    type: object
+    description: Dask scheduler configuration
+    required:
+    - spec
+    properties:
+      status:
+        type: object
+      spec:
+        type: object
+        required:
+        - scheduler
+        - worker
+        properties:
+          scheduler:
+            $ref: 'python://dask_kubernetes/operator/customresources/templates.yaml#/definitions/dask.k8s.api.v1.DaskScheduler'
+          worker:
+            $ref: 'python://dask_kubernetes/operator/customresources/templates.yaml#/definitions/dask.k8s.api.v1.DaskWorker'

dask_kubernetes/operator/deployment/helm/dask-kubernetes-operator/crds/daskcluster.yaml

@@ -5895,6 +5895,8 @@ spec:
             type: object
           status:
             type: object
+        required:
+        - spec
         type: object
     served: true
     storage: true