airflow.providers.ssh.utils.remote_job

Utilities for SSH remote job execution.

Attributes

POSIX_DEFAULT_BASE_DIR

WINDOWS_DEFAULT_BASE_DIR

Classes

RemoteJobPaths

Paths for remote job artifacts on the target system.

Functions

generate_job_id(dag_id, task_id, run_id, try_number[, ...])

Generate a unique job ID for remote execution.

build_posix_wrapper_command(command, paths[, environment])

Build a POSIX shell wrapper that runs the command detached.

build_windows_wrapper_command(command, paths[, ...])

Build a PowerShell wrapper that runs the command detached via Win32_Process.Create.

build_posix_log_tail_command(log_file, offset, max_bytes)

Build a POSIX command to read log bytes from offset.

build_windows_log_tail_command(log_file, offset, max_bytes)

Build a PowerShell command to read log bytes from offset.

build_posix_file_size_command(file_path)

Build a POSIX command to get file size in bytes.

build_windows_file_size_command(file_path)

Build a PowerShell command to get file size in bytes.

build_posix_completion_check_command(exit_code_file)

Build a POSIX command to check if job completed and get exit code.

build_windows_completion_check_command(exit_code_file)

Build a PowerShell command to check if job completed and get exit code.

build_posix_kill_command(pid_file)

Build a POSIX command to kill the remote process.

build_windows_kill_command(pid_file)

Build a PowerShell command to kill the remote process.

build_posix_cleanup_command(job_dir[, base_dir])

Build a POSIX command to clean up the job directory.

build_windows_cleanup_command(job_dir[, base_dir])

Build a PowerShell command to clean up the job directory.

build_posix_os_detection_command()

Build a command to detect if the remote system is POSIX-compliant.

build_windows_os_detection_command()

Build a command to detect if the remote system is Windows.

Module Contents

airflow.providers.ssh.utils.remote_job.POSIX_DEFAULT_BASE_DIR = '/tmp/airflow-ssh-jobs'[source]
airflow.providers.ssh.utils.remote_job.WINDOWS_DEFAULT_BASE_DIR = '$env:TEMP\\airflow-ssh-jobs'[source]
airflow.providers.ssh.utils.remote_job.generate_job_id(dag_id, task_id, run_id, try_number, suffix_length=8)[source]

Generate a unique job ID for remote execution.

Creates a deterministic identifier from the task context with a random suffix to ensure uniqueness across retries and potential race conditions.

Parameters:

  • dag_id (str) – The DAG identifier

  • task_id (str) – The task identifier

  • run_id (str) – The run identifier

  • try_number (int) – The attempt number

  • suffix_length (int) – Length of random suffix (default 8)

Returns:

Sanitized job ID string

Return type:

str

class airflow.providers.ssh.utils.remote_job.RemoteJobPaths[source]

Paths for remote job artifacts on the target system.

job_id: str[source]
remote_os: Literal['posix', 'windows'][source]
base_dir: str | None = None[source]
__post_init__()[source]
property sep: str[source]

Path separator for the remote OS.

property job_dir: str[source]

Directory containing all job artifacts.

property log_file: str[source]

Path to stdout/stderr log file.

property exit_code_file: str[source]

Path to exit code file (written on completion).

property exit_code_tmp_file: str[source]

Temporary exit code file (for atomic write).

property pid_file: str[source]

Path to PID file for the background process.

property status_file: str[source]

Path to optional status file for progress updates.

airflow.providers.ssh.utils.remote_job.build_posix_wrapper_command(command, paths, environment=None)[source]

Build a POSIX shell wrapper that runs the command detached.

The wrapper:

  • Creates the job directory

  • Starts the command detached in its own session via setsid (falling back to nohup when setsid is unavailable)

  • Redirects stdout/stderr to the log file

  • Writes the exit code atomically on completion

  • Writes the PID for potential cancellation

Parameters:

  • command (str) – The command to execute

  • paths (RemoteJobPaths) – RemoteJobPaths instance with all paths

  • environment (dict[str, str] | None) – Optional environment variables to set

Returns:

Shell command string to submit via SSH

Return type:

str

airflow.providers.ssh.utils.remote_job.build_windows_wrapper_command(command, paths, environment=None)[source]

Build a PowerShell wrapper that runs the command detached via Win32_Process.Create.

The wrapper:

  • Creates the job directory and writes the command into it as cmd.ps1, next to the job script that runs it

  • Starts the job script as a separate process that outlives the SSH session, in the SSH session’s working directory

  • Appends everything the command writes, on every stream, to the log file as UTF-8

  • Writes the exit code on completion, including when the command calls exit, throws, or fails to parse

  • Writes the PID for potential cancellation, and fails the submission if the job directory cannot be set up or the process cannot be started

The command runs as a script, so the host’s execution policy applies to it with -ExecutionPolicy Bypass for the process; a Group Policy that only allows signed scripts fails the job with the policy error in its log.

Parameters:

  • command (str) – The command to execute (PowerShell script path or command)

  • paths (RemoteJobPaths) – RemoteJobPaths instance with all paths

  • environment (dict[str, str] | None) – Optional environment variables to set

Returns:

PowerShell command string to submit via SSH

Return type:

str

airflow.providers.ssh.utils.remote_job.build_posix_log_tail_command(log_file, offset, max_bytes)[source]

Build a POSIX command to read log bytes from offset.

Parameters:

  • log_file (str) – Path to the log file

  • offset (int) – Byte offset to start reading from

  • max_bytes (int) – Maximum bytes to read

Returns:

Shell command that outputs the log chunk

Return type:

str

airflow.providers.ssh.utils.remote_job.build_windows_log_tail_command(log_file, offset, max_bytes)[source]

Build a PowerShell command to read log bytes from offset.

The chunk is written to stdout as UTF-8 bytes. Printing it as a PowerShell string would re-encode it in the SSH session’s console code page, which the trigger’s strict UTF-8 decoding rejects for any non-ASCII character.

Parameters:

  • log_file (str) – Path to the log file

  • offset (int) – Byte offset to start reading from

  • max_bytes (int) – Maximum bytes to read

Returns:

PowerShell command that outputs the log chunk

Return type:

str

airflow.providers.ssh.utils.remote_job.build_posix_file_size_command(file_path)[source]

Build a POSIX command to get file size in bytes.

Parameters:

file_path (str) – Path to the file

Returns:

Shell command that outputs the file size

Return type:

str

airflow.providers.ssh.utils.remote_job.build_windows_file_size_command(file_path)[source]

Build a PowerShell command to get file size in bytes.

Parameters:

file_path (str) – Path to the file

Returns:

PowerShell command that outputs the file size

Return type:

str

airflow.providers.ssh.utils.remote_job.build_posix_completion_check_command(exit_code_file)[source]

Build a POSIX command to check if job completed and get exit code.

Parameters:

exit_code_file (str) – Path to the exit code file

Returns:

Shell command that outputs exit code if done, empty otherwise

Return type:

str

airflow.providers.ssh.utils.remote_job.build_windows_completion_check_command(exit_code_file)[source]

Build a PowerShell command to check if job completed and get exit code.

Parameters:

exit_code_file (str) – Path to the exit code file

Returns:

PowerShell command that outputs exit code if done, empty otherwise

Return type:

str

airflow.providers.ssh.utils.remote_job.build_posix_kill_command(pid_file)[source]

Build a POSIX command to kill the remote process.

Signals the whole process group first (the negative-PID form kill -<pgid>) so the user command and anything it spawned are terminated together, not just the wrapper. The recorded PID is the job’s session/group leader when it was launched under setsid (see build_posix_wrapper_command()); if the job is not a group leader (host without setsid), the group signal is a no-op and we fall back to killing the single PID, matching the previous behaviour.

The pid value is validated as an integer > 1 before being negated: a corrupt or partial pid of 0/1 would otherwise turn kill -<pid> into a broadcast to every process the SSH account can signal. Best-effort: the command never fails the SSH call.

Parameters:

pid_file (str) – Path to the PID file

Returns:

Shell command to kill the process

Return type:

str

airflow.providers.ssh.utils.remote_job.build_windows_kill_command(pid_file)[source]

Build a PowerShell command to kill the remote process.

Uses taskkill /T to terminate the recorded process and its child processes (the Windows equivalent of a process-group kill), so the user command launched by the detached wrapper is not left orphaned. $procId is used instead of $pid because $PID is a read-only automatic variable in PowerShell.

Parameters:

pid_file (str) – Path to the PID file

Returns:

PowerShell command to kill the process

Return type:

str

airflow.providers.ssh.utils.remote_job.build_posix_cleanup_command(job_dir, base_dir=None)[source]

Build a POSIX command to clean up the job directory.

Parameters:

  • job_dir (str) – Path to the job directory

  • base_dir (str | None) – Base directory the job was created under. Defaults to the OS-specific default base directory.

Returns:

Shell command to remove the directory

Raises:

ValueError – If job_dir is not under the expected base directory

Return type:

str

airflow.providers.ssh.utils.remote_job.build_windows_cleanup_command(job_dir, base_dir=None)[source]

Build a PowerShell command to clean up the job directory.

Parameters:

  • job_dir (str) – Path to the job directory

  • base_dir (str | None) – Base directory the job was created under. Defaults to the OS-specific default base directory.

Returns:

PowerShell command to remove the directory

Raises:

ValueError – If job_dir is not under the expected base directory

Return type:

str

airflow.providers.ssh.utils.remote_job.build_posix_os_detection_command()[source]

Build a command to detect if the remote system is POSIX-compliant.

Returns the OS name (Linux, Darwin, FreeBSD, etc.) or UNKNOWN.

airflow.providers.ssh.utils.remote_job.build_windows_os_detection_command()[source]

Build a command to detect if the remote system is Windows.

Was this entry helpful?