Fix DockerOperator xcom

[nullhack]

This PR solves the issue #9164
This PR continues the discussion of #9165

Files changed:
airflow/providers/docker/operators/docker.py
airflow/providers/docker/example_dags/example_docker_xcom.py

Problems

• Issue 1: When xcom_push=True is enabled (and xcom_push_all=False), the output sometimes is null OR captured 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 is not compliant 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: Refactored the generator, using logs

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

            for line in gen_output(stdout=True, stderr=True):
                self.log.info(line)

• Issue 2: I just added:

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

• Issue 3: I store stdout into xcom:

           return_value = None
            if self.do_xcom_push:
                lines = gen_output(stdout=True)
                if self.xcom_all:
                    return_value = "\n".join(lines)
                else:
                    line_deque = deque(lines, maxlen=1)
                    return_value = line_deque.pop()

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

            result = self.cli.wait(self.container["Id"])
            if result["StatusCode"] != 0:
                raise AirflowException("docker container failed: " + repr(result))

Sample Output

• Issue 1:
  • Captured correctly:
    

We would expect only the last line '9'

• Issue 2:

  • Easier to pull outputs and use on other operators
    

• Issue 3:

  • Stderr do not mess up with the output and has 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.

[nullhack]

@mik-laj @akki I closed accidentally the other PR. Then I cleaned up things and put on this new.

Can you review It again?

About your questions:

I added the example on airflow/providers/docker/example_dags/example_docker_xcom.py

The reviews about wait and memory usage are correct. I did not considered huge containers running and capturing outputs on last PR. This one I used generators and deque to read last line efficiently.

I decided to keep the warnings together with stdout on airflow logs, but save only stdout on xcom to solve the issue 3.

The stop was an issue due to running wait before the functions. I shifted wait to It's right place and removed the stop. Thanks @akki for spotting that.

[stale]

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

[nullhack]

I'm not having enough time to update It, but I still want to merge this PR, on the following month I'll try to resolve conflicts and ask for reviews again. Please don't close this PR yet.

[turbaszek]

Hey @nullhack can you rebase, please? In the meantime we introduced black formatter in providers packages :)

[nullhack]

Hey @nullhack can you rebase, please? In the meantime we introduced black formatter in providers packages :)

Yes, I'll do soon

[turbaszek]

Hey @nullhack the test are failing, can you take a look?

[nullhack]

@turbaszek I couldn't find failing tests related to this PR, I just assumed they are long time failing tests like others I did. For example CI Build / Core:Pg9.6,Py3.7 points to a mysql_operator error, which is not touched here.

[potiuk]

@nullhack I think in all cases TestDockerOperator.test_execute_xcom_behavior is failing

[turbaszek]

@nullhack I think in all cases TestDockerOperator.test_execute_xcom_behavior is failing

Yes:

________________ TestDockerOperator.test_execute_xcom_behavior _________________
1052

1053
self = <tests.providers.docker.operators.test_docker.TestDockerOperator testMethod=test_execute_xcom_behavior>
1054

1055
    def test_execute_xcom_behavior(self):
1056
        self.client_mock.pull.return_value = [b'{"status":"pull log"}']
1057
    
1058
        kwargs = {
1059
            'api_version': '1.19',
1060
            'command': 'env',
1061
            'environment': {'UNIT': 'TEST'},
1062
            'private_environment': {'PRIVATE': 'MESSAGE'},
1063
            'image': 'ubuntu:latest',
1064
            'network_mode': 'bridge',
1065
            'owner': 'unittest',
1066
            'task_id': 'unittest',
1067
            'volumes': ['/host/path:/container/path'],
1068
            'working_dir': '/container/path',
1069
            'shm_size': 1000,
1070
            'host_tmp_dir': '/host/airflow',
1071
            'container_name': 'test_container',
1072
            'tty': True,
1073
        }
1074
    
1075
        xcom_push_operator = DockerOperator(**kwargs, do_xcom_push=True)
1076
        no_xcom_push_operator = DockerOperator(**kwargs, do_xcom_push=False)
1077
    
1078
        xcom_push_result = xcom_push_operator.execute(None)
1079
        no_xcom_push_result = no_xcom_push_operator.execute(None)
1080
    
1081
>       self.assertEqual(xcom_push_result, b'container log')
1082
E       AssertionError: 'container log' != b'container log'
1083

1084
tests/providers/docker/operators/test_docker.py:248: AssertionError

[nullhack]

Thank you, yes you're right. They're failing because the test assume a binary string b'container log'

One of the modifications of this PR is changing this to a decoded string instead (to have same output type as bash operator and make It easier to use the xcom results from other operators).

At this point I have three options

1. Change the behavior of the tests to the new one (can introduce breaking changes, as the current behavior is binary)
2. Introduce a new flag argument to the docker operator with default as binary, but modifiable to include the new behavior
3. Remove the new behavior

Which one should I go for?

[turbaszek]

I would be in favor of 2. + deprecation warning. I think we should return strings. @feluelle @mik-laj any opinions?

[nullhack]

Ok, I'll change the code then to include deprecation warning and the new flag with default value as binary

[akki]

Please have a look at the generator part. Check if using yield makes more sense to you as well.

[akki]

Hi

I don't think using a generator instead of a list here solves the memory issue. The data will in the end be kept in memory - no matter you use a list or a generator.
I did a small test to verify this; I ran the following code in a Python3 terminal:

>>> with open('yes.log', 'r') as file_:
...   x = (linee for linee in file_.read())
...

where yes.log was a 650 MB file.
As soon as the execution of this code-block completed, I saw the memory used by this process increase by 650 MB. It makes sense as well because where else would generators be storing these logs if not in memory.

I am thinking that you might be able to achieve what you're trying to do by streaming the logs and using yield.

[nullhack]

Thank you for the feedback Akki. I'll make some tests during this week and see if I can solve this using yield and streaming.

[Vasi-Shche]

I have the same problem, but I don't understand how the proposed changes can fix this case:

It will just print it in a column.

[nullhack]

Sorry, I'm not having time to work on this and probably will not be able to do for a while. I'll close the PR to give the chance of other people to work on this. Anyone feel free to fork my repo and continue