This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f04f123b2cc Prevent Dag processor crashes with AWS CloudWatch when its
log directory is outside `base_log_folder` (#74929)
f04f123b2cc is described below
commit f04f123b2cc2f47fe1bcca87f85105dd6c48ea2d
Author: Shahar Epstein <[email protected]>
AuthorDate: Mon Oct 19 22:17:42 2026 +0300
Prevent Dag processor crashes with AWS CloudWatch when its log directory is
outside `base_log_folder` (#74929)
The CloudWatch log processor resolved every log file's path relative to
base_log_folder without handling files that live elsewhere. With
dag_processor_child_process_log_directory set outside base_log_folder,
the first log line from a Dag file processor raised ValueError and took
down the whole Dag processor, so no Dag was parsed. Remote log upload
already skips such files.
---
.../providers/amazon/aws/log/cloudwatch_task_handler.py | 10 +++++++++-
.../unit/amazon/aws/log/test_cloudwatch_task_handler.py | 15 +++++++++++++++
2 files changed, 24 insertions(+), 1 deletion(-)
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
b/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
index 192d8d10566..5a71dc4ab70 100644
---
a/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
+++
b/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
@@ -182,7 +182,15 @@ class CloudWatchRemoteLogIO(LoggingMixin): # noqa: D101
# The handler MUST be initted here, before the processor is actually
used to log anything.
# Otherwise, logging that occurs during the creation of the handler
can create infinite loops.
_ = self.handler
- from airflow.sdk.log import relative_path_from_logger
+ from airflow.sdk.log import relative_path_from_logger as
_relative_path_from_logger
+
+ def relative_path_from_logger(logger) -> Path | None:
+ try:
+ return _relative_path_from_logger(logger)
+ except ValueError:
+ # Skip log files outside ``base_log_folder``, such as a Dag
processor log directory
+ # configured elsewhere, as ``upload_to_remote`` does.
+ return None
def proc(logger: structlog.typing.WrappedLogger, method_name: str,
event: structlog.typing.EventDict):
if not logger or not (stream_name :=
relative_path_from_logger(logger)):
diff --git
a/providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py
b/providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py
index c924658ea2e..3775e9dd208 100644
--- a/providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py
+++ b/providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py
@@ -316,6 +316,21 @@ class TestCloudRemoteLogIO:
dag_dir = self.local_log_location / "dag_id=a"
assert dag_dir.exists()
+ @mock.patch("airflow.sdk.log.relative_path_from_logger")
+ @mock.patch.object(CloudWatchRemoteLogIO, "handler",
new_callable=mock.PropertyMock)
+ def test_processors_skip_log_file_outside_base_log_folder(self,
mock_handler, mock_relative_path):
+ mock_relative_path.side_effect = ValueError("'/elsewhere/x.log' is not
in the subpath of '/logs'")
+ subject = CloudWatchRemoteLogIO(
+ base_log_folder=self.local_log_location,
+
log_group_arn=f"arn:aws:logs:{self.region_name}:11111111:log-group:{self.remote_log_group}",
+ )
+ event = {"event": "Hi"}
+
+ result = subject.processors[0](mock.MagicMock(), "info", event)
+
+ assert result is event
+ mock_handler.return_value.handle.assert_not_called()
+
@time_machine.travel(timezone.datetime(2025, 3, 27, 21, 58, 1, 2345),
tick=False)
def test_log_message(self):
# Use a context instead of a decorator on the test method because we
need access to self to
