Kubernetes Pod Operator support running with multiple containers

[dgdelahera]

Motivation of the PR

This PR allows Kubernetes Pod Operator to run with sidecars containers.

When running the Kubernetes Pod Operator, the task waits until the pod is in a Terminate state. In the case that we are running the Pod Operator with multiple containers, the Pod will not be in a Terminate state until all the containers are in a Terminate state. This is a problem when running the main container among sidecars containers, because in many case this containers doesn't receive the signal to exit. An example would be an envoy container that handles the pod network.

Example of a Pod with one container in a terminated state. Note that the Pod status is Running.

Name:        ********
Namespace:   ********
Priority:     0
Node:        ********
Start Time:   Thu, 28 Apr 2022 15:50:25 +0200
Labels:       airflow_version=2.1.4
              dag_id=accounts-refresh
              execution_date=2022-04-28T135015.7751980000-8035c9679
              kubernetes_pod_operator=True
              task_id=********
              try_number=1
Annotations:  [kubernetes.io/psp](http://kubernetes.io/psp): eks.privileged
Status:       Running
IP:           ********
IPs:
  IP:  ********
Containers:
  base:
    Container ID:   ********
    Image:          ********
    Image ID:       ********
    Port:           <none>
    Host Port:      <none>
    State:          Terminated
      Reason:       Error
      Exit Code:    1
      Started:      Thu, 28 Apr 2022 15:50:26 +0200
      Finished:     Thu, 28 Apr 2022 15:50:30 +0200
    Ready:          False
    Restart Count:  0
    Environment:
      ********
    Mounts:
      /var/run/secrets/kubernetes.io/serviceaccount from kube-api-access-xnh99 (ro)
  envoy:
    Container ID:   ********
    Image:         ********
    Image ID:       ********
    Port:           <none>
    Host Port:      <none>
    State:          Running
      Started:      Thu, 28 Apr 2022 15:50:26 +0200

The other part that won't work is the cleanup. At the moment, the task will be marked as failed if the pod is not in the SUCCEEDED phase. When running with sidecars, if the sidecars hasn't stop running, the pod won't be in that state.

Changes

As we can see in the follow diagram of the run workflow, we don't need to wait until the pod have finished, so I have removed that part. Our base container (the one that is executing the workload) will always be finished when we reach that point. If we are running with a xcom container, now we will wait for that container.

For the cleanup, now we check if the base container is completed or not. In the case that the container is not in that state, it means that the task has failed.

The changes have been tested more than a week in an AWS EKS.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[jedcunningham]

I'm wondering if instead we should just check the base container state, not just wait until any container finishes.

We also need to be mindful of how this will play with delete_worker_pods being false. K8s sidecars can get so messy.

[dgdelahera]

My first approach was to check the base container. I think it is the simplest one, but I wasn't sure it that approach was generic enough (sidecars finishing before the workload). But yeah, we can specify to the users to use one workload per pod (which is a good practice) and if they have any sidecar that finish before the workload they should use an init container instead.

[dgdelahera]

Hi @jedcunningham. Following your suggestion to watch the base container, I have updated the PR. Now I think we are keeping it simple.

In addition, I think that we can just check if the container is not running, because that implies that the container is in a terminated state. According to the documentation, there are three different states, and if the container is in a running state it can only move to the terminated state.

[potiuk]

Hey @jedcunningham - not sure if you are available :) .but if you do, I am preparing release of providers now and I think this one might be a good one to be included.

[potiuk]

Seems we have some realated failures - it will have to wait for the next round of providers.

[dstandish]

this changes the meaning of this method from "await pod completion" to "await container completion"

[jedcunningham]

Probably worth having a separate method for doing it based on container, if we go that way, so we dont introduce a breaking change here.

[dstandish]

if i were gonna add this param, i would not call it 'base_container'. this method doesn't know that the caller is providing the "base" container, so better to just be generic with container_name or something. BUT i am not sure we should change this method anyway, since the purpose is to wait for pod termination.

[dstandish]

i'm not seeing the await all containers option?

[jedcunningham]

I still have concerns with how this will interact with delete_worker_pods being false, seems like a good way to get orphaned still-running pods in your cluster.

[dgdelahera]

Hi @dstandish and @jedcunningham. Thank you for your review, and sorry for the delay, it's been a busy week. Going back to this PR, I have made a diagram to clarify it because I think that we can delete this function method call https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py#L418. That point can only be reached when the base container has finished, so we don't need to wait until the pod has finished. This will continue working with single containers, and it will fix the behavior when running with sidecars.

Regarding the delete_worker_pods interaction, I think that the best way to do the clean up is to inspect the base container state (the one that it is executing the workload) instead of the pod state.

Let me know what do you think about this :)

[dgdelahera]

I have found other behavior that it isn't working as expected with multiple containers. During the cleanup(), if the pod isn't in the succeeded phase, it will fail. But as you can see in the following pod, the base container is terminated. I will make some changes and ask for a new review.

            'container_statuses': [{'container_id': '...',
                                    'image': '...',
                                    'image_id': '...',
                                    'last_state': {'running': None,
                                                   'terminated': None,
                                                   'waiting': None},
                                    'name': 'base',
                                    'ready': False,
                                    'restart_count': 0,
                                    'started': False,
                                    'state': {'running': None,
                                              'terminated': {'container_id': '...',
                                                             'exit_code': 0,
                                                             'finished_at': datetime.datetime(2022, 6, 14, 9, 35, 17, tzinfo=tzlocal()),
                                                             'message': None,
                                                             'reason': 'Completed',
                                                             'signal': None,
                                                             'started_at': datetime.datetime(2022, 6, 14, 9, 35, 2, tzinfo=tzlocal())},
                                              'waiting': None}},
                                   {'container_id': '...',
                                    'image': '...',
                                    'image_id': '...',
                                    'last_state': {'running': None,
                                                   'terminated': None,
                                                   'waiting': None},
                                    'name': 'envoy',
                                    'ready': True,
                                    'restart_count': 0,
                                    'started': True,
                                    'state': {'running': {'started_at': datetime.datetime(2022, 6, 14, 9, 35, 2, tzinfo=tzlocal())},
                                              'terminated': None,
                                              'waiting': None}}],
            'ephemeral_container_statuses': None,
            'host_ip': '...',
            'init_container_statuses': None,
            'message': None,
            'nominated_node_name': None,
            'phase': 'Running',
       ```

[dstandish]

@dgdelahera re usage of remote pod there's a related PR in flight #22092

also, when we patch the pod, which is sometimes done in cleanup, we need to use the latest remotely read pod (i.e. remote_pod) otherwise we might get error because of patching with outdated pod.

[dgdelahera]

@dstandish My PR doesn't modify that workflow. https://github.com/apache/airflow/pull/23450/files#diff-947f9f98b4aedccee6dd81558568c350c738f5da050aad1224d8b48ec43677cdR418 There we are still reading the latest pod state before calling the cleanup

[dstandish]

Hi @dstandish and @jedcunningham. Thank you for your review, and sorry for the delay, it's been a busy week. Going back to this PR, I have made a diagram to clarify it because I think that we can delete this function method call https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py#L418. That point can only be reached when the base container has finished, so we don't need to wait until the pod has finished. This will continue working with single containers, and it will fix the behavior when running with sidecars.

So, when there is an xcom sidecar, it is terminated when we call self.extract_xcom(pod=self.pod). So after we terminate xcom sidecar, we need to wait for the pod to complete before proceeding on to cleanup.

Separately it seems either your description needs updating or the PR because it's talking about adding an "await all containers" option but not seeing that in the PR.

It might be helpful, in evaluating your approach, to have an example too, demonstrating how your change would help with multiple containers / sidecars.

[dgdelahera]

@dstandish Oh, I didn't take into account the xcom sidecar. I will think it again, and I will also update the description. Thank you so much for the feedback

[dgdelahera]

@dstandish @jedcunningham I have updated the description and also added a check to the xcom container. If you are agree with this approach, I will add the unit tests for the new methods. Thanks 😄

[dgdelahera]

Hi @jedcunningham @dstandish, would you mind taking a new review? Thanks

[potiuk]

Hi @jedcunningham @dstandish, would you mind taking a new review? Thanks

I think it would be good to make the tests pass first.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.