kubernetes contrib: labels are applied to pod (#3007)

Author: dav009

luigi/contrib/kubernetes.py

@@ -363,7 +363,8 @@ def run(self):
                 "backoffLimit": self.backoff_limit,
                 "template": {
                     "metadata": {
-                        "name": self.uu_name
+                        "name": self.uu_name,
+                        "labels": {}
                     },
                     "spec": self.spec_schema
                 }
@@ -376,6 +377,8 @@ def run(self):
                 self.active_deadline_seconds
         # Update user labels
         job_json['metadata']['labels'].update(self.labels)
+        job_json['spec']['template']['metadata']['labels'].update(self.labels)
+
         # Add default restartPolicy if not specified
         if "restartPolicy" not in self.spec_schema:
             job_json["spec"]["template"]["spec"]["restartPolicy"] = "Never"

test/contrib/kubernetes_test.py

@@ -71,6 +71,10 @@ class FailJob(KubernetesJobTask):
         }]
     }
 
+    @property
+    def labels(self):
+        return {"dummy_label": "dummy_value"}
+
 
 @attr('contrib')
 class TestK8STask(unittest.TestCase):
@@ -90,6 +94,7 @@ def test_fail_job(self):
         job = Job(kube_api, jobs.response["items"][0])
         self.assertTrue("failed" in job.obj["status"])
         self.assertTrue(job.obj["status"]["failed"] > fail.max_retrials)
+        self.assertTrue(job.obj['spec']['template']['metadata']['labels'] == fail.labels())
 
     @mock.patch.object(KubernetesJobTask, "_KubernetesJobTask__get_job_status")
     @mock.patch.object(KubernetesJobTask, "signal_complete")