Fix DockerOperator xcom

[nullhack]

Solves the issue #9164

File changed: airflow/providers/docker/operators/docker.py

Problems

• Issue 1: When xcom_push=True is enabled (and xcom_push_all=False), the output sometimes is null OR captured in wrongly.
• Issue 2: When xcom_push_all=True a bytes string (b'...') is stored as xcom, It's harder to use the output on following operators and do not conform with other operators.
• Issue 3: Stderr and stdout are written to the same output xcom. In practice, we don't want warnings and errors messing up with the code to be parsed on following operators (But we need to capture the output on airflow logs). Sending stderr to xcom can lead to undefined/non-deterministic behavior.

Solutions:

• Issue 1: The issue here is that the attach starts before the wait. But seems there's an issue if we try to run wait before attach using stream=True that freezes and don't return any output (I did not dig deep, but probably will become an issue on docker python SDK later). A simple solution is to run logs as It's a wrapper for attach. I followed this way and refactored part of the code, instead of two calls I saved the output on a list after wait:

            self.cli.start(self.container["Id"])

            result = self.cli.wait(self.container["Id"])

            lines = [
                templ.decode("utf-8").strip() if hasattr(templ, "decode") else str(templ)
                for templ in self.cli.logs(
                    self.container["Id"], stream=True, stdout=True, stderr=False
                )
            ]

Why a list, not a generator?
Because I might use two times, a generator can be used only once per request:

            for line in lines:
                self.log.info(line)
   
            ...

            if self.do_xcom_push:
                return "\n".join(lines) if self.xcom_all else line
            else:
                return None

So I decided to just keep as a list to make the code less complex. And do not repeat the request as the original code.

• Issue 2: I just added:

...
templ.decode("utf-8").strip() if hasattr(templ, "decode") else str(templ)
...

This will ensure that every line will be decoded if possible, if not, then return the string representation of It (DockerOperator always return bytes string, but if in the future other objects are possible, this keep It consistent against errors).
I expect that this part can cause some discussion, because I'm changing the output from bytes to string (my personal opinion is that being bytes is an issue and should be changed to string as BashOperator). There's no documentation about what should be the value stored as xcom from DockerOperator, but If this is a big issue I suggest creating a flag (xcom_string=False), but personally I don't like this approach.

• Issue 3: I created a separated list that only stores stderr:

            warnings = [
                tempw.decode("utf-8").strip() if hasattr(tempw, "decode") else str(tempw)
                for tempw in self.cli.logs(
                    self.container["Id"], stream=True, stdout=False, stderr=True
                )
            ]

            for warning in warnings:
                self.log.warning(warning)

In the end, only logs are send to xcom and if any error is raised, the DAG run fail:

            if result["StatusCode"] != 0:
                raise AirflowException("docker container failed: " + repr(result))

            if self.do_xcom_push:
                return "\n".join(lines) if self.xcom_all else line
            else:
                return None

Sample Output

• Issue 1:

  • Captured correctly:
    
    We would expect only the last line '9'

• Issue 2:

  • Conform with similar operators
    

  • Easier to pull outputs and use on other operators
    

• Issue 3:

  • Stderr do not mess up with the output
    
    Note that new lines are not shown on UI, but if used on some operator the output will be correct
  • Deterministic behavior
    

---

• Description above provides context of the change
• Unit tests coverage for changes (not needed for documentation changes) unnecessary
• Target Github ISSUE in description if exists
• Commits follow "How to write a good git commit message"
• Relevant documentation is updated including usage instructions. unnecessary
• I will engage committers as explained in Contribution Workflow Example.

[mik-laj]

@retornam @CodingJonas @CaptainCuddleCube @nalepae @akki Can I ask for review? This change looks valuable, but requires expert knowledge about the Docker/Docker API.

[mik-laj]

@nullhack Can you add an example of how to use DockerOperator and XCOM?
https://github.com/apache/airflow/blob/master/airflow/providers/docker/example_dags/example_docker.py
If I understand correctly, this behavior is different from KubernetesPodOperator?
https://github.com/apache/airflow/blob/34d1441f0a0eae32a398d85ae95191f7a074e367/airflow/providers/google/cloud/example_dags/example_kubernetes_engine.py#L67-L77

[nullhack]

@mik-laj Thank you for the suggestions. I wrote some example on the issue Itself (#9164) if you need It to test now.

I'll write the example as you asked. But I would prefer to do It after some agreement is reached if the current code is acceptable or needs changes to avoid modifying the example every time a change is required.

About the KubernetesPodOperator, I haven't use It, my workflow is based on docker swarm. Can you describe the difference in behavior you mentioned earlier?

Apart from that, there's one check that were not successful. I tried to read It, but don't seems related to my development (at least directly). Can you (or any other member reading It) confirm if It?

[mik-laj]

About the KubernetesPodOperator, I haven't use It, my workflow is based on docker swarm. Can you describe the difference in behavior you mentioned earlier?

In kubernetes, you must create a file and this file will be saved to xcom.

mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json

In this way, the logs can still contain useful information for the developer.

Apart from that, there's one check that were not successful. I tried to read It, but don't seems related to my development (at least directly). Can you (or any other member reading It) confirm if It?

Quarantine tests are in quarantine because they are flaky. Please ignore it. We want to fix them soon.

[ashb]

@nullhack In cases like this (where you are opening a PR) creating the issue before hand is entierly optional -- you can just open a PR and describe the bug in the PR message; no issue needed.

[nullhack]

@nullhack In cases like this (where you are opening a PR) creating the issue before hand is entierly optional -- you can just open a PR and describe the bug in the PR message; no issue needed.

Noted. I didn't know about that, Thanks!

[nullhack]

In kubernetes, you must create a file and this file will be saved to xcom.

Yes, the current DockerOperator is different. It reads logs from self.client.logs instead. I think somewhere in between would be the ideal. The return is based on files, but the default file returned is stdout.

mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json

In this way, the logs can still contain useful information for the developer.

Yes, I like this approach. One problem I see is that /airflow/xcom/return.json is fixed. I think if the path could be changed (default stdout) would be more flexible.

[akki]

I think the self.cli.attach method with stream=True returns a generator which allows you to read the logs in a "streaming" fashion without using significant memory.

[akki]

So I decided to just keep as a list to make the code less complex. And do not repeat the request as the original code.

I do read this in the description. I am guessing the original code makes the repeated request because of the tiny memory footprint of this API.

[akki]

So this will load the full logs of the application into memory, right? So if the application log file is 1 GB in size Airflow will start using 1GB extra RAM?

[akki]

Please correct me if I am wrong but If we .wait() here, I understand that Airflow won't show up the application logs until the Docker container stops?

For long-running applications (which typically run for 1 or more hours), I think it is very crucial to see their logs in near-real-time but with this, I think we won't see any logs until the container process completes? Sorry if I am missing something here.

[akki]

I have a couple of questions here:

1. Why does Airflow explicitly need to stop the Docker container?
2. If I understand correctly the .wait() API (which we are already calling above) ensures that the container is stopped by blocking until the container gets stopped so why would the container be running at this point of time (hence why the need for this call?).

If there is some reason that justifies this call, I would suggest adding some (warn?) logging here telling the user that the container was forcefully stopped by Airflow, so that it is easy for them to debug.

[akki]

@mik-laj Thanks for drawing my attention to this.
@nullhack I have commented some questions/suggestions for these changes, hope they make sense to you and are useful.

[mik-laj]

@nullhack Could you do rebase instead of merging changes next time? This allows us to see all your changes in the "Commits" tab. It makes reviews easier.
Here is a guide about it: https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#id8
If you want, you can now undo your merge using git reflog and try to do rebase.

[nullhack]

Thank you @mik-laj I was trying to find a solution for this. Will use the reflog