Skip to content

External python operator logging #27012

Closed

Description

@juraa14

Apache Airflow version

2.4.1

What happened

When using the external python operator for running tasks inside a different environment, logs do not appear for the task instance.
The log files are always empty if the task is successful.
If the task fails, one gets very rudimentary logs about the process command failing.

Both environments use the same Python version (3.9.10) and Airflow uses Celery executor

What you think should happen instead

The External Operator is great for cases where you might have multiple environments that require other packages and versions, so they don't mix up with the airflow environment and so other teams could potentially use their own environments inside DAGs. Unfortunately, the External Python Operator does not provide any meaningful logs when the task fails or succeeds.

An example of a successful task without logs:
image

In case a task fails, the following log will appear (regardless of the actual error):

[2022-10-12, 11:23:05 CEST] {taskinstance.py:1851} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/airflow_env/lib/python3.9/site-packages/airflow/decorators/base.py", line 188, in execute
    return_value = super().execute(context)
  File "/usr/local/airflow_env/lib/python3.9/site-packages/airflow/operators/python.py", line 370, in execute
    return super().execute(context=serializable_context)
  File "/usr/local/airflow_env/lib/python3.9/site-packages/airflow/operators/python.py", line 175, in execute
    return_value = self.execute_callable()
  File "/usr/local/airflow_env/lib/python3.9/site-packages/airflow/operators/python.py", line 678, in execute_callable
    return self._execute_python_callable_in_subprocess(python_path, tmp_path)
  File "/usr/local/airflow_env/lib/python3.9/site-packages/airflow/operators/python.py", line 426, in _execute_python_callable_in_subprocess
    execute_in_subprocess(
  File "/usr/local/airflow_env/lib/python3.9/site-packages/airflow/utils/process_utils.py", line 168, in execute_in_subprocess
    execute_in_subprocess_with_kwargs(cmd, cwd=cwd)
  File "/usr/local/airflow_env/lib/python3.9/site-packages/airflow/utils/process_utils.py", line 191, in execute_in_subprocess_with_kwargs
    raise subprocess.CalledProcessError(exit_code, cmd)
subprocess.CalledProcessError: Command '['/usr/local/aiops_env/bin/python', '/tmp/tmdclodljc4/script.py', '/tmp/tmdclodljc4/script.in', '/tmp/tmdclodljc4/script.out', '/tmp/tmdclodljc4/string_args.txt']' returned non-zero exit status 1.
[2022-10-12, 11:23:05 CEST] {standard_task_runner.py:102} ERROR - Failed to execute job 1976894 for task external_python_pipeline (Command '['/usr/local/aiops_env/bin/python', '/tmp/tmdclodljc4/script.py', '/tmp/tmdclodljc4/script.in', '/tmp/tmdclodljc4/script.out', '/tmp/tmdclodljc4/string_args.txt']' returned non-zero exit status 1.; 23728)

How to reproduce

Using celery executor, so the task is ran on another machine. Simply run a DAG that uses the External Python Operator. A simple hello world will reproduce the issue

Operating System

CentOS 7

Versions of Apache Airflow Providers

apache-airflow-providers-celery==3.0.0
apache-airflow-providers-common-sql==1.2.0
apache-airflow-providers-ftp==3.1.0
apache-airflow-providers-http==4.0.0
apache-airflow-providers-imap==3.0.0
apache-airflow-providers-postgres==5.2.2

Deployment

Virtualenv installation

Deployment details

No response

Anything else

I have not yet gone into the details, but am willing to contribute!

Are you willing to submit PR?

  • Yes I am willing to submit a PR!

Code of Conduct

Activity

  1. boring-cyborg commented on Oct 12, 2022

    @boring-cyborg

    Thanks for opening your first issue here! Be sure to follow the issue template!

  2. pierrejeambrun commented on Oct 15, 2022

    @pierrejeambrun
    Member

    @juraa14 Thank you for being willing to help on this one.

    I just assigned you :)

  3. potiuk commented on Oct 24, 2022

    @potiuk
    Member

    And that should also fix the same case for Python Virtualenv Operator (which should have the same problem).

  4. juraa14 commented on Dec 21, 2022

    @juraa14
    Author

    Been a while
    I've looked into the issue a little bit to find out what is happening.
    The task is run using python's subprocess module. It uses Popen which executes a child program in a new process. Now, we can capture STDOUT and STDERR from such a subprocess and output that once the program is finished. So I'll explain what is happening first inside the operator:

    with subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=0, close_fds=True, **kwargs
        ) as proc:
            log.info("Output:")
            if proc.stdout:
                with proc.stdout:
                    for line in iter(proc.stdout.readline, b""):
                        log.info("%s", line.decode().rstrip())
    
            exit_code = proc.wait()
        if exit_code != 0:
            raise subprocess.CalledProcessError(exit_code, cmd)
    

    The external task is run using subprocess.Popen and all the output is logged at INFO level. If your Airflow logging level is set to WARNING or higher, you will not see the output of the subprocess. So using print() and return statements will output to Airflow logging at INFO level.
    You can also output other logs by using handler = logging.StreamHandler(sys.stdout) and then all logs are forwarded to stdout.

    Here is a sample DAG that output using print, logging and return:

    from datetime import datetime
    
    from airflow.decorators import dag, task
    from airflow.operators.empty import EmptyOperator
    
    
    @dag(dag_id="test_external_op_py_logging",
         schedule=None,
         description="Testing external python logging",
         catchup=False,
         start_date=datetime(2021, 1, 1),
         max_active_runs=1,
         tags=["TEST"])
    def test():
        start = EmptyOperator(
            task_id="start"
        )
    
        @task.external_python(task_id="external_op", python="/path/to/python/bin/python")
        def test_logging():
            import pandas as pd
            import logging
            import sys
    
            root_logger = logging.getLogger()
            handler = logging.StreamHandler(sys.stdout)
    
            root_logger.addHandler(handler)
    
            print("PRNT:starting task")
            logging.info("LOG:starting task")
            ret_str = "RET:starting task\n"
    
            data = {"id": [1, 2],
                    "name": ["Arnold", "Schwarzenegger"]
                    }
    
            df = pd.DataFrame(data)
    
            print("PRNT: " + df.to_string())
            logging.debug("LOG: " + df.to_string())
            logging.error("LOG_ERROR: " + df.to_string())
            ret_str += "RET_ " + df.to_string() + "\n"
    
            return ret_str
    
        end = EmptyOperator(
            task_id="end"
        )
    
        start >> test_logging() >> end
    
    
    dag = test()
    
    

    This DAG outputs the following:

    [2022-12-14, 14:50:29 UTC] {python.py:711} WARNING - When checking for Airflow installed in venv got The version of Airflow installed for the /path/bin/python(2.4.1) is different than the runtime Airflow version: 2.5.0. Make sure your environment has the same Airflow version installed as the Airflow runtime.
    [2022-12-14, 14:50:29 UTC] {python.py:712} WARNING - This means that Airflow is not properly installed by  /path/bin/python. Airflow context keys will not be available. Please Install Airflow 2.5.0 in your environment to access them.
    [2022-12-14, 14:50:29 UTC] {process_utils.py:179} INFO - Executing cmd: /path/bin/python /tmp/tmdyokq8x9g/script.py /tmp/tmdyokq8x9g/script.in /tmp/tmdyokq8x9g/script.out /tmp/tmdyokq8x9g/string_args.txt
    [2022-12-14, 14:50:29 UTC] {process_utils.py:183} INFO - Output:
    [2022-12-14, 14:50:30 UTC] {process_utils.py:187} INFO - [�[34m2022-12-14 15:50:30,333�[0m] {�[34mstats.py:�[0m349} ERROR�[0m - Could not configure StatsClient: No module named 'statsd', using DummyStatsLogger instead.�[0m
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - PRNT:starting task
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - [�[34m2022-12-14 15:50:31,363�[0m] {�[34mscript.py:�[0m36} INFO�[0m - LOG:starting task�[0m
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - LOG:starting task
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - PRNT:    id            name
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - 0   1          Arnold
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - 1   2  Schwarzenegger
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - [�[34m2022-12-14 15:50:31,370�[0m] {�[34mscript.py:�[0m47} ERROR�[0m - LOG_ERROR:    id            name
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - 0   1          Arnold
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - 1   2  Schwarzenegger�[0m
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - LOG_ERROR:    id            name
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - 0   1          Arnold
    [2022-12-14, 14:50:31 UTC] {process_utils.py:187} INFO - 1   2  Schwarzenegger
    [2022-12-14, 14:50:31 UTC] {python.py:177} INFO - Done. Returned value was: RET:starting task
    RET_    id            name
    0   1          Arnold
    1   2  Schwarzenegger
    

    The logs are pretty ugly and there are some encoding issues using logger, but it seems to be sort of a hack around this. Changing this would require changes in the subprocess module, which I guess wouldn't be possible

    Here's a prettier example of logs where I printed a spark dataframe using show():

    [2022-12-21, 00:01:44 UTC] {python.py:711} WARNING - When checking for Airflow installed in venv got Command '['/random/bin/python', '-c', 'from airflow import version; print(version.version)']' returned non-zero exit status 1.
    [2022-12-21, 00:01:44 UTC] {python.py:712} WARNING - This means that Airflow is not properly installed by  /random/bin/python. Airflow context keys will not be available. Please Install Airflow 2.5.0 in your environment to access them.
    [2022-12-21, 00:01:44 UTC] {process_utils.py:179} INFO - Executing cmd: /random/bin/python /tmp/tmdjh10pukv/script.py /tmp/tmdjh10pukv/script.in /tmp/tmdjh10pukv/script.out /tmp/tmdjh10pukv/string_args.txt
    [2022-12-21, 00:01:44 UTC] {process_utils.py:183} INFO - Output:
    [2022-12-21, 00:01:50 UTC] {process_utils.py:187} INFO - Setting default log level to "WARN".
    [2022-12-21, 00:01:50 UTC] {process_utils.py:187} INFO - To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - 
    [Stage 0:>                                                          (0 + 1) / 1]
    
                                                                                    
    
    [Stage 1:>                                                          (0 + 1) / 1]
    
    [Stage 3:>                                                          (0 + 8) / 8]
    
    [Stage 3:==============>                                            (2 + 6) / 8]
    
    [Stage 3:======================>                                    (3 + 5) / 8]
    
    [Stage 3:============================================>              (6 + 2) / 8]
    
                                                                                    
    Reading from DB
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - +------------------------------------+----+----+------------------------------+--------+----------------+
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - |id                                  |User|Card|Merchant Name                 |Amount  |date_transaction|
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - +------------------------------------+----+----+------------------------------+--------+----------------+
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - |ef259c63-cd92-4e2a-8713-6083dff03166|0   |0   |Terry, Jackson and Williams   |1.0     |2022-12-20      |
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - |72142628-5573-4166-bdd2-6543512dc8d6|1   |1   |Wilson and Sons               |11.0    |2022-12-20      |
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - |8879bcb5-8df7-40ee-933e-9e0cce408e1f|2   |2   |Collins-Rodriguez             |11359.95|2022-12-20      |
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - |c95ac77e-c627-4bd1-9cf5-f5604b8b5baf|3   |3   |Wyatt-Holden                  |31.0    |2022-12-20      |
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - |c9af6312-6440-4b9c-bd29-44675b920586|4   |4   |Woodward, Perez and Mccullough|41.0    |2022-12-20      |
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - +------------------------------------+----+----+------------------------------+--------+----------------+
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - only showing top 5 rows
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - 
    [2022-12-21, 00:02:16 UTC] {process_utils.py:187} INFO - None
    [2022-12-21, 00:02:17 UTC] {python.py:177} INFO - Done. Returned value was: None
    [2022-12-21, 00:02:17 UTC] {taskinstance.py:1322} INFO - Marking task as SUCCESS. dag_id=spark, task_id=aggregations, execution_date=20221220T000000, start_date=20221221T000144, end_date=20221221T000217
    [2022-12-21, 00:02:17 UTC] {local_task_job.py:159} INFO - Task exited with return code 0
    [2022-12-21, 00:02:17 UTC] {taskinstance.py:2582} INFO - 1 downstream tasks scheduled from follow-on schedule check
    

    One thing I nearly forgot to mention, getting the traceback from python is possible again if you put logging level to INFO :)

  5. potiuk commented on Jan 16, 2023

    @potiuk
    Member

    Are there any actions form that?

  6. juraa14 commented on Jan 17, 2023

    @juraa14
    Author

    No, we can only capture the STDOUT and STDERR output from a subprocess and the comment above shows that, but we cannot control the way subprocesses work in Python.

  7. potiuk commented on Jan 17, 2023

    @potiuk
    Member

    So should we close it ? Is that the suggestion?

  8. potiuk commented on Jan 17, 2023

    @potiuk
    Member

    (asking because you are the author - I would simply close it).

  9. Docdez commented on Jul 18, 2025

    @Docdez

    I'm surprised that there is no common handler to pretty parse the logs of the subprocess output at this time.

  10. potiuk commented on Jul 19, 2025

    @potiuk
    Member

    I'm surprised that there is no common handler to pretty parse the logs of the subprocess output at this time.

    You can always contribute one. This is how things work here - there are more than 3300 contributors here - many of them contributed something they needed. I am actually quite surprised you have not done it yet @Docdez

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions