Prevent Dag processor crashes with AWS CloudWatch when its log directory is outside base_log_folder - #74929

Review

@shahar1, this is a really nice piece of work. A few things stand out:

• The diagnosis is sharp. Path.relative_to() raising ValueError on a file outside base_log_folder is exactly the kind of edge case that's easy to miss, and you connected it to the actual user-visible symptom (Dag processor dying, no Dags parsed) without me having to read the code to follow.
• The fix mirrors the existing pattern in upload_to_remote (sdk/log.py), so it's consistent with the codebase rather than inventing a new shape.
• Keeping the change small and the symptom well-scoped — one fix for one provider, with the Google PR tracked separately — is the right way to do this.
• The unit test is tight: it asserts the right thing (no CloudWatch handle call when the path is outside base_log_folder) and the new test name reads well.
• The PR description is excellent: it states the failure mode, the root cause, the parity with upload_to_remote, and how it was verified. Thanks for the AI disclosure too.

I have a few notes below — one is a real bug in the new test, the rest are small polish items.

1. The new test's mock.patch is not actually wired up to the wrapper (test bug)

cloudwatch_task_handler.py does:

from airflow.sdk.log import relative_path_from_logger as _relative_path_from_logger

That as rebinding severs the link. The wrapper then closes over _relative_path_from_logger directly:

def relative_path_from_logger(logger) -> Path | None:
    try:
        return _relative_path_from_logger(logger)
    except ValueError:
        return None

So @mock.patch("airflow.sdk.log.relative_path_from_logger") in the new test mutates the original module attribute, but the wrapper still calls the unmocked _relative_path_from_logger. Today the test happens to pass because mock.MagicMock() (the logger argument) auto-creates _file, fh.fileno() is a MagicMock that compares unequal to 1, and you eventually hit a TypeError/ValueError somewhere down the real code path — so the assertion is satisfied for the wrong reason. If someone later tightens the real function to not raise, this test will go silently green.

Two clean fixes; I'd take the second:

(a) Test through a real file that lives outside base_log_folder. That's the strongest test because it actually exercises both the wrapper and the real relative_path_from_logger, which is what we want regression coverage on. Concretely:

def test_processors_skip_log_file_outside_base_log_folder(self, tmp_path):
    # File intentionally lives outside base_log_folder so relative_to() raises ValueError.
    outside = tmp_path / "elsewhere" / "x.log"
    outside.parent.mkdir(parents=True)
    outside.touch()
    # … pass a logger whose _file is `outside` and assert no upload happened

(b) Or, refactor the import so the wrapper calls the module attribute at call time, which makes the existing mock.patch effective:

from airflow.sdk import log as airflow_sdk_log
…
def relative_path_from_logger(logger) -> Path | None:
    try:
        return airflow_sdk_log.relative_path_from_logger(logger)
    except ValueError:
        return None

I'd lean toward (a) because the existing test in the Google PR (#74926) went with mock.patch and reviewer @SameerMesiah97 already flagged this same pattern there — the underlying test machinery is the same in both PRs, so it's worth getting right in at least one of them.

2. Add a debug log when skipping

The Google PR #74926 already got the "Perhaps you could add a debug log here?" comment from @SameerMesiah97. Same applies here — silently dropping the log line is the right behavior, but a one-line self.log.debug("Skipping CloudWatch upload: log file %s is outside base_log_folder", …) makes it possible to diagnose the case from the Dag processor log itself. That matches the level used in upload():

self.log.warning(
    "Skipping deletion: path %s is outside base_log_folder %s",
    local_path, base,
)

upload() uses warning because deletion is destructive; for the processor path, debug is the right level (this is the expected, non-fatal branch, not a misconfiguration that the user needs to know about).

3. Catching Exception vs ValueError

upload_to_remote catches a bare Exception. Your wrapper catches only ValueError. The current relative_path_from_logger only raises ValueError from Path.relative_to, so ValueError is correct and more precise. I'd just call this out in a comment so a future reader doesn't "broaden" it by analogy with upload_to_remote:

# relative_to() raises ValueError when the file is not a subpath of
# base_log_folder. We intentionally catch only ValueError here; a bare
# Exception would also hide real bugs (e.g. config not loaded).

4. Newsfragment

Per the PR template, this is a user-facing bug fix (Dag processor crash with a non-default log directory). It should ship with a newsfragment in providers/amazon/newsfragments/, e.g.:

providers/amazon/newsfragments/74929.bugfix.rst

with content like:

Fix ``CloudWatchRemoteLogIO`` crashing the Dag processor when
``[logging] dag_processor_child_process_log_directory`` is configured
outside ``[logging] base_log_folder``. The processor now skips
log files outside ``base_log_folder`` instead of raising ``ValueError``.

Same ask for the Google PR.

5. Minor: name shadowing is fine but worth a short comment

The wrapper is named relative_path_from_logger (same as the upstream) and shadows the import renamed to _relative_path_from_logger. This is a normal pattern but a one-line comment that the inner name is the safe wrapper and the leading-underscore name is the un-wrapped upstream would help a future reader who lands in this file cold.

Overall: solid PR, the fix is correct, the test setup is the only thing I'd ask you to revisit before merging. Happy to re-review if you push an update.