Repository: spotify/luigi
Pull request: Add pod_creation_wait_interal
<!--- This template is optional. Please use it as a starting point to help guide PRs -->

<!--- Provide a general summary of your changes in the Title above -->

## Description
<!--- Describe your changes -->
Add pod_creation_wait_interal property to give Kubernetes cluster some time to create pods for Job.

## Motivation and Context
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here. -->
Fix https://github.com/spotify/luigi/issues/2570, also aimed to replace abandoned https://github.com/spotify/luigi/pull/2664

## Have you tested this? If so, how?
<!--- Valid responses are "I have included unit tests." or --> 
<!--- "I ran my jobs with this code and it works for me." -->
I ran my jobs with this code and it works for me.
<!---
for more information on how to submit valuable contributions,
see https://opensource.guide/how-to-contribute/#how-to-submit-a-contribution
-->


Problem statement:
KubernetesJobTask "No pod scheduled" error
I often see the following error when using `KubernetesJobTask`:
```
[28:luigi-interface:224@18:33] [ERROR] [pid 28] Worker Worker(salt=476402868, workers=20, host=something-host-1541289600-zntd7, username=root, pid=1) failed    Something(local_execution=Falser)
Traceback (most recent call last):
  File "/opt/conda/lib/python3.6/site-packages/luigi/worker.py", line 205, in run
    new_deps = self._run_get_new_deps()
  File "/opt/conda/lib/python3.6/site-packages/luigi/worker.py", line 142, in _run_get_new_deps
    task_gen = self.task.run()
  File "/usr/src/app/phpipeline/luigi_util.py", line 231, in run
    super(PHKubernetesJobTask, self).run()
  File "/opt/conda/lib/python3.6/site-packages/luigi/contrib/kubernetes.py", line 355, in run
    self.__track_job()
  File "/opt/conda/lib/python3.6/site-packages/luigi/contrib/kubernetes.py", line 199, in __track_job
    while not self.__verify_job_has_started():
  File "/opt/conda/lib/python3.6/site-packages/luigi/contrib/kubernetes.py", line 261, in __verify_job_has_started
    assert len(pods) > 0, "No pod scheduled by " + self.uu_name
AssertionError: No pod scheduled by something-20181104183346-b51d371a5bfd4197
[1:luigi-interface:570@18:33] [INFO] Informed scheduler that task   Something_42b6a6d55a   has status   FAILED
```
It is hard to reproduce, but seems like sometimes pod needs a bit more time to be created, but task does not wait for it to be created and ends up raising the error. Task gets FAILED status, but a pod is still created and is run without Luigi control.

When the task is restarted and not failed, a pod is run under Luigi control, but there's still that uncontrolled pod, so we end up having two pods making the same thing.

I've managed to fix it the most naive way. When getting pods to verify that they are started, instead of:
```python
pods = self.__get_pods()
```
I do the following:
```python
for _ in range(3):
    pods = self._KubernetesJobTask__get_pods()

    if pods:
        break

    # If pod was not returned, sleep to wait for pod to be created.
    sleep(15)
```
This is not a beautiful way of fixing the issue, but it works (though it will for sure fail if pod needs >45 seconds to be started).

Hints from issue discussion:
I have same issues.
@keisuke-umezawa I've managed to fix it with the code above. Just create a custom `KubernetesJobTask` class with an overridden `__verify_job_has_started` method.
@StasDeep Thanks! Do you have example how to override `__verify_job_has_started`?
Yep. Something like that:

```
class MyKubernetesJobTask(KubernetesJobTask):

    def _KubernetesJobTask__verify_job_has_started(self):
        """
        Does the same thing as the base method, but tries to get pods three times instead of one.
        This way, it's less likely to get empty list of pods and get an exception.
        Sleeps between attempts for 15 seconds.
        """
        # Verify that the job started
        self._KubernetesJobTask__get_job()

        # Verify that the pod started
        for _ in range(3):
            pods = self._KubernetesJobTask__get_pods()

            if pods:
                break

            # If pod was not returned, sleep to wait for pod to be created.
            sleep(15)
```

Note that `__verify_job_has_started`, `__get_job` etc. methods have two underscores in the beginning, so don't forget to use `_KubernetesJobTask` prefix when invoking them.
It seems reasonable to me to allow there to be some amount of wait time and attempts for a POD to become available. Of course, it should be configurable. Happy to review a PR!
This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. If closed, you may revisit when your time allows and reopen! Thank you for your contributions.

Commit message: Add pod_creation_wait_interal

Patch to review:
diff --git a/luigi/contrib/kubernetes.py b/luigi/contrib/kubernetes.py
index c1c7afce65..13db2dbe94 100644
--- a/luigi/contrib/kubernetes.py
+++ b/luigi/contrib/kubernetes.py
@@ -213,10 +213,15 @@ def poll_interval(self):
         """How often to poll Kubernetes for job status, in seconds."""
         return self.__DEFAULT_POLL_INTERVAL
 
+    @property
+    def pod_creation_wait_interal(self):
+        """Delay for initial pod creation for just submitted job"""
+        return 5
+
     def __track_job(self):
         """Poll job status while active"""
         while not self.__verify_job_has_started():
-            time.sleep(self.__DEFAULT_POLL_INTERVAL)
+            time.sleep(self.poll_interval)
             self.__logger.debug("Waiting for Kubernetes job " + self.uu_name + " to start")
         self.__print_kubectl_hints()
 
@@ -276,6 +281,11 @@ def __verify_job_has_started(self):
 
         # Verify that the pod started
         pods = self.__get_pods()
+        if not pods:
+            self.__logger.debug(
+                'No pods found for {}, waiting for cluster state to match the job definition', self.uu_name
+            )
+            time.sleep(self.pod_creation_wait_interal)
 
         assert len(pods) > 0, "No pod scheduled by " + self.uu_name
         for pod in pods:


Code review comments:
diff_hunk: @@ -213,10 +213,15 @@ def poll_interval(self):
         """How often to poll Kubernetes for job status, in seconds."""
         return self.__DEFAULT_POLL_INTERVAL
 
+    @property
+    def pod_creation_wait_interal(self):
+        """Delay for initial pod creation for just submitted job"""
path: luigi/contrib/kubernetes.py
text: @user1:
Please use a constant similar as `__DEFAULT_POLL_INTERVAL`, also in docstring please describe the unit so it would be easier to follow.