Ensure a WatchedSubprocess redacts only its values and not parent processes'

[amoghrajesh]

Problem

If you have noticed some of the recent logs by running tasks from the latest main, branch they somewhat look like:

[2025-03-26, 17:24:08] ERROR - Task failed with exception: source="task"
KeyError: 'logical_date'
File "/opt/***/task-sdk/src/***/sdk/execution_time/task_runner.py", line 646 in run

File "/opt/***/task-sdk/src/***/sdk/execution_time/task_runner.py", line 892 in _execute_task

File "/opt/***/task-sdk/src/***/sdk/definitions/baseoperator.py", line 380 in wrapper

File "/opt/***/***-core/src/***/decorators/base.py", line 252 in execute

File "/opt/***/task-sdk/src/***/sdk/definitions/baseoperator.py", line 380 in wrapper

File "/opt/***/providers/standard/src/***/providers/standard/operators/python.py", line 212 in execute

File "/opt/***/providers/standard/src/***/providers/standard/operators/python.py", line 235 in execute_callable

File "/opt/***/task-sdk/src/***/sdk/execution_time/callback_runner.py", line 81 in run

File "/files/dags/context_test.py", line 18 in access_context_task

This problem is because we mask the "password" of SQLA when running in breeze (which is airflow), here: https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/settings.py#L372

Although this might seem like a trivial problem for breeze, it isn't. There is a bigger problem here because we SHOULDNT be redacting the SQLA stuff at all for the "subprocess" that task sdk supervisor launches, and the reason for that is the subprocess has not access to the DB at ALL!!

Solution

• Solving this by introducing a new "reset" method on secrets masker that will reset the current patterns and replacer in it

• What that does basically is it treats the masker as a new instance, which is what i want: https://github.com/apache/airflow/blob/main/task-sdk/src/airflow/sdk/execution_time/secrets_masker.py#L330-L352

• Adding a condition while adding the mask_logs log processor to the logger. We control that using the sending_to_supervisor flag now. If we are sending to supervisor, then DO NOT add the mask_logs as it will be already masked by the supervisor and no need to remask it.

• This will not impact any existing behaviour as you might think because all calls will still go through: https://github.com/apache/airflow/blob/main/task-sdk/src/airflow/sdk/execution_time/secrets_masker.py#L83-L94

• This same functionality needs to be extended to the dag processor as well as triggerer which use the task sdk supervisor now. The subprocess shan't have access to the parent process' masking.

Testing

Scenario 1: Retrieving a conneciton inside a task and trying to print its password, which is "password"

Connection:

DAG:

import structlog

from airflow import DAG

from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import Connection


def print_sensitive():
    log = structlog.get_logger()

    c = Connection.get("teradata_default")
    log.info("Retrieved connection", password=c.password)


with DAG(
    dag_id="print_sensitive_data",
    schedule=None,
) as dag:

    print_sensitive_task = PythonOperator(
        task_id="print_sensitive_task",
        python_callable=print_sensitive,
    )

Scenario 2: Trying to do the same with conneciton but at global level

DAG:

import structlog

from airflow import DAG

from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import Connection

log = structlog.get_logger()
c = Connection.get("teradata_default")
log.info("At the top level is", password=c.password)

def print_sensitive():
    log.info("Retrieved connection and printing in task", password=c.password)

with DAG(
    dag_id="print_sensitive_data_top_level",
    schedule=None,
) as dag:

    print_sensitive_task = PythonOperator(
        task_id="print_sensitive_task",
        python_callable=print_sensitive,
    )

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[amoghrajesh]

Guess i have to write some tests, but majority of the work is complete here.

[kaxil]

Overall lgtm -- added few comments

[amoghrajesh]

Test failure isnt related