[GitHub] [airflow] tanelk commented on issue #20982: Triggered tasks from web-ui are killed by scheduler because they are orphaned task instances

[GitBox <gi...@apache.org>]

tanelk commented on issue #20982:
URL: https://github.com/apache/airflow/issues/20982#issuecomment-1092089810

   I have never worked with kubernetes, but based on the triggering code:
   https://github.com/apache/airflow/blob/1f29d844411522e9a2cd4a14c3573e903b4e5074/airflow/www/views.py#L1786-L1794
   
   I believe this diff could get you in the right direction.
   ```diff --git a/airflow/executors/kubernetes_executor.py b/airflow/executors/kubernetes_executor.py
   index bdac0889f..fe25d25e8 100644
   --- a/airflow/executors/kubernetes_executor.py
   +++ b/airflow/executors/kubernetes_executor.py
   @@ -685,9 +685,14 @@ class KubernetesExecutor(BaseExecutor):
            self.event_buffer[key] = state, None
   
        def try_adopt_task_instances(self, tis: List[TaskInstance]) -> List[TaskInstance]:
   -        tis_to_flush = [ti for ti in tis if not ti.queued_by_job_id]
   -        scheduler_job_ids = {ti.queued_by_job_id for ti in tis}
   -        pod_ids = {ti.key: ti for ti in tis if ti.queued_by_job_id}
   +        scheduler_job_ids = {ti.queued_by_job_id for ti in tis if ti.queued_by_job_id}
   +
   +        # Tasks triggered through API will have no ti.queued_by_job_id
   +        # and their pod will have label 'airflow-worker=manual'
   +        if any(ti for ti in tis if not ti.queued_by_job_id):
   +            scheduler_job_ids.add('manual')
   +
   +        pod_ids = {ti.key: ti for ti in tis}
            kube_client: client.CoreV1Api = self.kube_client
            for scheduler_job_id in scheduler_job_ids:
                scheduler_job_id = pod_generator.make_safe_label_value(str(scheduler_job_id))
   @@ -696,8 +701,8 @@ class KubernetesExecutor(BaseExecutor):
                for pod in pod_list.items:
                    self.adopt_launched_task(kube_client, pod, pod_ids)
            self._adopt_completed_pods(kube_client)
   -        tis_to_flush.extend(pod_ids.values())
   -        return tis_to_flush
   +
   +        return list(pod_ids.values())
   
        def adopt_launched_task(
            self, kube_client: client.CoreV1Api, pod: k8s.V1Pod, pod_ids: Dict[TaskInstanceKey, k8s.V1Pod]
   ```
   
   It would be nice if someone could try this out. 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org