Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Introduce new config variable to control whether DAG processor outputs to stdout #37439

Merged
17 changes: 17 additions & 0 deletions airflow/config_templates/airflow_local_settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,10 @@
"logging", "DAG_PROCESSOR_MANAGER_LOG_LOCATION"
)

DAG_PROCESSOR_MANAGER_LOG_STDOUT: str = conf.get_mandatory_value(
"logging", "DAG_PROCESSOR_MANAGER_LOG_STDOUT"
)

# FILENAME_TEMPLATE only uses in Remote Logging Handlers since Airflow 2.3.3
# All of these handlers inherited from FileTaskHandler and providing any value rather than None
# would raise deprecation warning.
Expand Down Expand Up @@ -171,6 +175,19 @@
},
}

if DAG_PROCESSOR_MANAGER_LOG_STDOUT == "True":
DEFAULT_DAG_PARSING_LOGGING_CONFIG["handlers"].update(
{
"console": {
"class": "airflow.utils.log.logging_mixin.RedirectStdHandler",
"formatter": "airflow",
"stream": "sys.stdout",
"filters": ["mask_secrets"],
}
}
)
DEFAULT_DAG_PARSING_LOGGING_CONFIG["loggers"]["airflow.processor_manager"]["handlers"].append("console")

# Only update the handlers and loggers when CONFIG_PROCESSOR_MANAGER_LOGGER is set.
# This is to avoid exceptions when initializing RotatingFileHandler multiple times
# in multiple processes.
Expand Down
7 changes: 7 additions & 0 deletions airflow/config_templates/config.yml
Original file line number Diff line number Diff line change
Expand Up @@ -857,6 +857,13 @@ logging:
type: string
example: ~
default: "{AIRFLOW_HOME}/logs/dag_processor_manager/dag_processor_manager.log"
dag_processor_manager_log_stdout:
description: |
Whether DAG processor manager will write logs to stdout
version_added: 2.8.1
potiuk marked this conversation as resolved.
Show resolved Hide resolved
type: boolean
example: ~
default: "False"
task_log_reader:
description: |
Name of handler to read task instance logs.
Expand Down
17 changes: 17 additions & 0 deletions tests/dag_processing/test_job_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -1460,3 +1460,20 @@ def test_launch_process(self):
processor_agent._process.join()

assert os.path.isfile(log_file_loc)

@conf_vars({("logging", "dag_processor_manager_log_stdout"): "True"})
potiuk marked this conversation as resolved.
Show resolved Hide resolved
def test_log_to_stdout(self, capfd):
test_dag_path = TEST_DAG_FOLDER / "test_scheduler_dags.py"
async_mode = "sqlite" not in conf.get("database", "sql_alchemy_conn")

# Starting dag processing with 0 max_runs to avoid redundant operations.
processor_agent = DagFileProcessorAgent(test_dag_path, 0, timedelta(days=365), [], False, async_mode)
processor_agent.start()
if not async_mode:
processor_agent.run_single_parsing_loop()

processor_agent._process.join()

# Capture the stdout and stderr
out, _ = capfd.readouterr()
assert "DAG File Processing Stats" in out
Loading