kaxil commented on code in PR #69834:
URL: https://github.com/apache/airflow/pull/69834#discussion_r3574106477
##########
providers/ssh/src/airflow/providers/ssh/utils/remote_job.py:
##########
@@ -30,18 +30,25 @@
WINDOWS_DEFAULT_BASE_DIR = "$env:TEMP\\airflow-ssh-jobs"
-def _validate_job_dir(job_dir: str, remote_os: Literal["posix", "windows"]) ->
None:
+def _validate_job_dir(
+ job_dir: str,
+ remote_os: Literal["posix", "windows"],
+ expected_base: str | None = None,
+) -> None:
"""
Validate that job_dir is under the expected base directory.
:param job_dir: The job directory path to validate
:param remote_os: Operating system type
+ :param expected_base: Expected base directory for the job
:raises ValueError: If job_dir doesn't start with the expected base path
"""
if remote_os == "posix":
- expected_prefix = POSIX_DEFAULT_BASE_DIR + "/"
+ expected_base = expected_base or POSIX_DEFAULT_BASE_DIR
Review Comment:
`expected_base or POSIX_DEFAULT_BASE_DIR` treats a falsy value as unset, but
`RemoteJobPaths.__post_init__` only substitutes the default when `base_dir is
None`. Since `remote_base_dir` is templated and only validated pre-render, a
value that renders empty builds `job_dir` directly under `/` at submission,
then cleanup validates it against the hardcoded default here and raises.
Checking `expected_base is None` keeps the two consistent.
##########
providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py:
##########
@@ -451,10 +451,11 @@ def _cleanup_remote_job(self, job_dir: str, remote_os:
str) -> None:
leave the directory rather than failing the (already finished) task.
"""
self.log.info("Cleaning up remote job directory: %s", job_dir)
+ expected_base = self._paths.base_dir if self._paths else
self.remote_base_dir
if remote_os == "posix":
- cleanup_cmd = build_posix_cleanup_command(job_dir)
+ cleanup_cmd = build_posix_cleanup_command(job_dir,
expected_base=expected_base)
Review Comment:
If validation fails here, the ValueError escapes `_cleanup_remote_job`
before the retry loop and fails a task whose remote job already finished, which
is the failure mode from #69813 and contradicts the log-and-leave contract in
the docstring above. That's still reachable after this fix: `remote_base_dir`
is templated, and templates are re-rendered when the operator resumes after
deferral, so a value like `{{ var.value.ssh_jobs_base }}` re-renders against
resume-time state while `event["job_dir"]` keeps the submission-time base the
trigger was constructed with. If the Variable changes while the job runs,
cleanup rejects a valid path and the task fails.
The cheap guard is to catch the ValueError and route it through the same
warn-and-leave path used when retries are exhausted. A fuller fix would thread
the submission-time base (`self._paths.base_dir`) through the trigger event
alongside `job_dir` and validate against that, which also makes the
`self._paths` fallback on the line above unnecessary, since cleanup always runs
on a fresh instance where `_paths` is `None`. Happy for either to land as a
follow-up if you'd rather keep this PR minimal.
##########
providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py:
##########
@@ -451,10 +451,11 @@ def _cleanup_remote_job(self, job_dir: str, remote_os:
str) -> None:
leave the directory rather than failing the (already finished) task.
"""
self.log.info("Cleaning up remote job directory: %s", job_dir)
+ expected_base = self._paths.base_dir if self._paths else
self.remote_base_dir
Review Comment:
The new tests pin the builders, but this line is the wiring #69813 actually
hit: if the operator change were reverted and only the utils change kept, every
test in the PR would still pass. `test_execute_complete_with_cleanup` in
`operators/test_ssh_remote_job.py` is easy to clone into a variant with
`remote_base_dir="/tmp-data/airflow-ssh-jobs"` and a matching event `job_dir`,
which would cover the reported scenario end to end.
##########
providers/ssh/src/airflow/providers/ssh/utils/remote_job.py:
##########
@@ -452,27 +459,29 @@ def build_windows_kill_command(pid_file: str) -> str:
return f"powershell.exe -NoProfile -NonInteractive -EncodedCommand
{encoded_script}"
-def build_posix_cleanup_command(job_dir: str) -> str:
+def build_posix_cleanup_command(job_dir: str, expected_base: str | None =
None) -> str:
"""
Build a POSIX command to clean up the job directory.
:param job_dir: Path to the job directory
+ :param expected_base: Expected base directory for the job
:return: Shell command to remove the directory
:raises ValueError: If job_dir is not under the expected base directory
"""
- _validate_job_dir(job_dir, "posix")
+ _validate_job_dir(job_dir, "posix", expected_base)
Review Comment:
On the suggestion to drop this validation from the cleanup builders instead:
I'd keep it. By the time cleanup runs, `job_dir` is whatever came back in the
trigger event (`execute_complete` passes `event["job_dir"]` straight through),
and it ends up in `rm -rf`. This prefix check is the only thing pinning an
event-supplied path under the configured base.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
