diff --git a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py index 6ac690fc9d5..fef33b01081 100644 --- a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py +++ b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py @@ -113,10 +113,12 @@ TEST_COMMANDS = [ 'tasks states-for-dag-run example_bash_operator --logical-date "{date_param}"', 'tasks state example_bash_operator runme_0 "manual__{date_param}"', 'tasks state example_bash_operator runme_0 --logical-date "{date_param}"', + 'tasks logs example_bash_operator "manual__{date_param}" runme_0 --try-number 1', 'tasks clear example_bash_operator --dag-run-id "manual__{date_param}" --task-ids runme_0 -o json', # Task Instances commands 'taskinstances get example_bash_operator "manual__{date_param}" runme_0', 'taskinstances get-dependencies example_bash_operator "manual__{date_param}" runme_0', + 'taskinstances logs example_bash_operator "manual__{date_param}" runme_0 1', 'taskinstances list example_bash_operator "manual__{date_param}"', 'taskinstances state-store set example_bash_operator "manual__{date_param}" runme_0 test_key test_value', 'taskinstances state-store get example_bash_operator "manual__{date_param}" runme_0 test_key', diff --git a/airflow-ctl/docs/images/command_hashes.txt b/airflow-ctl/docs/images/command_hashes.txt index 7b30c1068bf..6b663dd3048 100644 --- a/airflow-ctl/docs/images/command_hashes.txt +++ b/airflow-ctl/docs/images/command_hashes.txt @@ -9,8 +9,8 @@ dagrun:07035226eaaff0a557d3ad9aab4b41b5 jobs:d4af478f28dae48ee18d43b998edc345 pools:19efe105b9515ab1926ebcaf0e028d71 providers:826e68c9de8254293ba810a5c67fa8c7 -taskinstances:af95eda24a2bc4c6bd87c9f1324dbc3b -tasks:56fa16e067fea389a11727b21985ec2f +taskinstances:ef73172ce155dcad1f4ee38c1050db8f +tasks:79059c3c8de9114e73895d4b8e86ed00 variables:f8fc76d3d398b2780f4e97f7cd816646 version:31f4efdf8de0dbaaa4fac71ff7efecc3 plugins:4864fd8f356704bd2b3cd1aec3567e35 diff --git a/airflow-ctl/docs/images/output_taskinstances.svg b/airflow-ctl/docs/images/output_taskinstances.svg index 7c02563d8f2..f3d6be8f3e7 100644 --- a/airflow-ctl/docs/images/output_taskinstances.svg +++ b/airflow-ctl/docs/images/output_taskinstances.svg @@ -1,4 +1,4 @@ - + - - + + - + - + - + - + - + - + - + - + - + - + - + - + - + - + + + + - + - + - - Usage:airflowctl taskinstances [-h] COMMAND... - -Perform TaskInstances operations - -Positional Arguments: -COMMAND -getGet a task instance for a given Dag run -get-dependencies -Get unmet scheduler dependencies for a task instance -listList all task instances for a given Dag run -state-storeManage the state store of a task instance - -Options: --h, --helpshow this help message and exit - + + Usage:airflowctl taskinstances [-h] COMMAND... + +Perform TaskInstances operations + +Positional Arguments: +COMMAND +getGet a task instance for a given Dag run +get-dependencies +Get unmet scheduler dependencies for a task instance +listList all task instances for a given Dag run +logsGet the logs of a task instance for a given try number +state-storeManage the state store of a task instance + +Options: +-h, --helpshow this help message and exit + diff --git a/airflow-ctl/docs/images/output_tasks.svg b/airflow-ctl/docs/images/output_tasks.svg index b1b5bf1014b..5af11113e74 100644 --- a/airflow-ctl/docs/images/output_tasks.svg +++ b/airflow-ctl/docs/images/output_tasks.svg @@ -1,4 +1,4 @@ - + - - + + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + + + + - + - + - - Usage:airflowctl tasks [-h] COMMAND... - -Perform Tasks operations - -Positional Arguments: -COMMAND -clearClear task instances of a Dag by its ID -failed-depsReturns the unmet dependencies for a task instance -getRetrieve a task for a Dag by its ID and task ID -listList tasks for a Dag -stateGet the state of a task instance -states-for-dag-run -Get the status of all task instances in a Dag run - -Options: --h, --helpshow this help message and exit - + + Usage:airflowctl tasks [-h] COMMAND... + +Perform Tasks operations + +Positional Arguments: +COMMAND +clearClear task instances of a Dag by its ID +failed-depsReturns the unmet dependencies for a task instance +getRetrieve a task for a Dag by its ID and task ID +listList tasks for a Dag +logsGet the logs of a task instance +stateGet the state of a task instance +states-for-dag-run +Get the status of all task instances in a Dag run + +Options: +-h, --helpshow this help message and exit + diff --git a/airflow-ctl/src/airflowctl/api/operations.py b/airflow-ctl/src/airflowctl/api/operations.py index ac934fe6f94..802f663abb4 100644 --- a/airflow-ctl/src/airflowctl/api/operations.py +++ b/airflow-ctl/src/airflowctl/api/operations.py @@ -83,6 +83,7 @@ from airflowctl.api.datamodels.generated import ( TaskDependencyCollectionResponse, TaskInstanceCollectionResponse, TaskInstanceResponse, + TaskInstancesLogResponse, TaskResponse, TaskStateStoreBody, TaskStateStoreCollectionResponse, @@ -879,6 +880,32 @@ class TaskInstancesOperations(BaseOperations): ) return validate_response(self.response.content, TaskDependencyCollectionResponse) + def logs( + self, + dag_id: str, + dag_run_id: str, + task_id: str, + try_number: int, + map_index: int | None = None, + *, + suppress_error_log: bool = False, + ) -> TaskInstancesLogResponse | ServerResponseError: + """Get the logs of a task instance for a given try number.""" + path = _build_task_instance_path( + dag_id=dag_id, dag_run_id=dag_run_id, task_id=task_id, map_index=None + ) + # Without full_content the endpoint returns one chunk and a continuation token, so a running or + # remotely stored log would print only partially. + params: dict[str, Any] = {"full_content": True} + if map_index is not None and map_index >= 0: + params["map_index"] = map_index + self.response = self.client.get( + f"{path}/logs/{try_number}", + params=params, + extensions={"airflowctl_suppress_error_log": suppress_error_log}, + ) + return TaskInstancesLogResponse.model_validate_json(self.response.content) + def list(self, dag_id: str, dag_run_id: str) -> TaskInstanceCollectionResponse | ServerResponseError: """List task instances for a Dag run.""" return super().execute_list( diff --git a/airflow-ctl/src/airflowctl/ctl/cli_config.py b/airflow-ctl/src/airflowctl/ctl/cli_config.py index 25d5d6c870f..b7702e68d07 100755 --- a/airflow-ctl/src/airflowctl/ctl/cli_config.py +++ b/airflow-ctl/src/airflowctl/ctl/cli_config.py @@ -415,6 +415,20 @@ ARG_MAP_INDEX = Arg( help="Mapped task index", ) +# Task logs command args. Required primitive parameters stay positional per the +# airflowctl parameter style consensus (#66768). +ARG_TASKS_LOGS_DAG_RUN_ID = Arg( + flags=("dag_run_id",), + type=str, + help="The run ID of the Dag run", +) +ARG_TRY_NUMBER = Arg( + flags=("--try-number",), + type=int, + required=True, + help="The try number of the task instance logs to fetch", +) + ARG_ACTION_ON_EXISTING_KEY = Arg( flags=("-a", "--action-on-existing-key"), type=str, @@ -1430,6 +1444,19 @@ TASK_COMMANDS = ( ARG_OUTPUT, ), ), + ActionCommand( + name="logs", + help="Get the logs of a task instance", + description="Get the logs of a task instance for a given try number.", + func=lazy_load_command("airflowctl.ctl.commands.task_command.logs"), + args=( + ARG_DAG_ID, + ARG_TASKS_LOGS_DAG_RUN_ID, + ARG_TASK_ID, + ARG_MAP_INDEX, + ARG_TRY_NUMBER, + ), + ), ) VARIABLE_COMMANDS = ( diff --git a/airflow-ctl/src/airflowctl/ctl/commands/task_command.py b/airflow-ctl/src/airflowctl/ctl/commands/task_command.py index f03a83e5103..dc756bd8056 100644 --- a/airflow-ctl/src/airflowctl/ctl/commands/task_command.py +++ b/airflow-ctl/src/airflowctl/ctl/commands/task_command.py @@ -23,7 +23,7 @@ from typing import TYPE_CHECKING import rich from airflowctl.api.client import NEW_API_CLIENT, ClientKind, ServerResponseError, provide_api_client -from airflowctl.api.datamodels.generated import TaskInstanceState +from airflowctl.api.datamodels.generated import StructuredLogMessage, TaskInstanceState from airflowctl.ctl.console_formatting import AirflowConsole from airflowctl.ctl.utils.dag_run import resolve_dag_run_id @@ -158,3 +158,27 @@ def state(args, api_client=NEW_API_CLIENT) -> None: # Unset states print as None to stay drop-in compatible with the deprecated # `airflow tasks state`, which prints its nullable `ti.state` column directly. print(task_instance.state.value if task_instance.state else None) + + +@provide_api_client(kind=ClientKind.CLI) +def logs(args, api_client=NEW_API_CLIENT) -> None: + """Get the logs of a task instance for a given try.""" + try: + response = api_client.task_instances.logs( + dag_id=args.dag_id, + dag_run_id=args.dag_run_id, + task_id=args.task_id, + try_number=args.try_number, + map_index=args.map_index, + suppress_error_log=True, + ) + except ServerResponseError as e: + if e.response.status_code == 404: + rich.print( + f"[red]{_task_instance_not_found_message(args.dag_id, args.dag_run_id, args.task_id, args.map_index, e)}[/red]" + ) + sys.exit(1) + raise + + for entry in response.content: + print(entry.event if isinstance(entry, StructuredLogMessage) else entry) diff --git a/airflow-ctl/src/airflowctl/ctl/help_texts.yaml b/airflow-ctl/src/airflowctl/ctl/help_texts.yaml index 26e6074f16d..ef0e0ddc7bb 100644 --- a/airflow-ctl/src/airflowctl/ctl/help_texts.yaml +++ b/airflow-ctl/src/airflowctl/ctl/help_texts.yaml @@ -96,6 +96,7 @@ taskinstances: list: "List all task instances for a given Dag run" get: "Get a task instance for a given Dag run" get-dependencies: "Get unmet scheduler dependencies for a task instance" + logs: "Get the logs of a task instance for a given try number" # state-store is the nested subcommand group. The *-state-store keys stay because # help is still looked up by the operation method name, not the CLI path. state-store: "Manage the state store of a task instance" diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py b/airflow-ctl/tests/airflow_ctl/api/test_operations.py index 400feaa9423..3642ed7473b 100644 --- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py +++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py @@ -105,6 +105,7 @@ from airflowctl.api.datamodels.generated import ( TaskInletAssetReference, TaskInstanceCollectionResponse, TaskInstanceResponse, + TaskInstancesLogResponse, TaskInstanceState, TaskOutletAssetReference, TaskResponse, @@ -2450,6 +2451,27 @@ class TestTaskInstancesOperations: ) assert response == self.task_dependency_collection_response + @pytest.mark.parametrize( + ("map_index", "expected_params"), + [ + pytest.param(None, {"full_content": "true"}, id="unmapped"), + pytest.param(2, {"full_content": "true", "map_index": "2"}, id="mapped"), + ], + ) + def test_logs_requests_full_content(self, map_index, expected_params): + log_response = TaskInstancesLogResponse(content=["line"], continuation_token=None) + + def handle_request(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/api/v2/dags/dag_id/dagRuns/dag_run_id/taskInstances/task_id/logs/1" + assert dict(request.url.params) == expected_params + return httpx.Response(200, json=json.loads(log_response.model_dump_json())) + + client = make_api_client(transport=httpx.MockTransport(handle_request)) + response = client.task_instances.logs( + dag_id="dag_id", dag_run_id="dag_run_id", task_id="task_id", try_number=1, map_index=map_index + ) + assert response == log_response + def test_get_dependencies_with_map_index(self): client = self._make_client_asserting_path( "/api/v2/dags/dag_id/dagRuns/dag_run_id/taskInstances/task_id/3/dependencies", diff --git a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py index d086f37abc5..c803a880221 100644 --- a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py +++ b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py @@ -24,10 +24,12 @@ import httpx import pytest from airflowctl.api.datamodels.generated import ( + StructuredLogMessage, TaskDependencyCollectionResponse, TaskDependencyResponse, TaskInstanceCollectionResponse, TaskInstanceResponse, + TaskInstancesLogResponse, TaskInstanceState, ) from airflowctl.api.operations import ServerResponseError @@ -909,3 +911,122 @@ class TestState: self.parser.parse_args(["tasks", "state", self.dag_id, self.task_id, self.run_id]), api_client=api_client, ) + + +class TestLogs: + parser = cli_parser.get_parser() + dag_id = "test_dag" + run_id = "test_run" + task_id = "test_task" + + def _make_log_response(self, content: list[StructuredLogMessage] | list[str]) -> TaskInstancesLogResponse: + return TaskInstancesLogResponse(content=content, continuation_token=None) + + def _make_structured_content(self) -> list[StructuredLogMessage]: + return [ + StructuredLogMessage( + timestamp=datetime.datetime(2025, 1, 1, tzinfo=datetime.UTC), + event="INFO - hello", + ), + StructuredLogMessage( + timestamp=datetime.datetime(2025, 1, 1, tzinfo=datetime.UTC), + event="plain line", + ), + ] + + def test_logs_with_try_number(self, capsys): + api_client = mock.MagicMock() + api_client.task_instances.logs.return_value = self._make_log_response(self._make_structured_content()) + + task_command.logs( + self.parser.parse_args( + ["tasks", "logs", self.dag_id, self.run_id, self.task_id, "--try-number", "1"] + ), + api_client=api_client, + ) + + api_client.task_instances.logs.assert_called_once_with( + dag_id=self.dag_id, + dag_run_id=self.run_id, + task_id=self.task_id, + try_number=1, + map_index=-1, + suppress_error_log=True, + ) + assert capsys.readouterr().out == "INFO - hello\nplain line\n" + + def test_logs_requires_try_number(self, capsys): + with pytest.raises(SystemExit, match="2"): + self.parser.parse_args(["tasks", "logs", self.dag_id, self.run_id, self.task_id]) + assert "--try-number" in capsys.readouterr().err + + def test_logs_prints_plain_string_entries(self, capsys): + api_client = mock.MagicMock() + api_client.task_instances.logs.return_value = self._make_log_response(["plain line"]) + + task_command.logs( + self.parser.parse_args( + ["tasks", "logs", self.dag_id, self.run_id, self.task_id, "--try-number", "1"] + ), + api_client=api_client, + ) + + assert capsys.readouterr().out == "plain line\n" + + def test_logs_with_explicit_try_number_and_map_index(self, capsys): + api_client = mock.MagicMock() + api_client.task_instances.logs.return_value = self._make_log_response(self._make_structured_content()) + + task_command.logs( + self.parser.parse_args( + [ + "tasks", + "logs", + self.dag_id, + self.run_id, + self.task_id, + "--try-number", + "2", + "--map-index", + "3", + ] + ), + api_client=api_client, + ) + + api_client.task_instances.logs.assert_called_once_with( + dag_id=self.dag_id, + dag_run_id=self.run_id, + task_id=self.task_id, + try_number=2, + map_index=3, + suppress_error_log=True, + ) + + def test_logs_task_instance_not_found(self, capsys): + api_client = mock.MagicMock() + api_client.task_instances.logs.side_effect = _make_server_error(404) + + with pytest.raises(SystemExit, match="1"): + task_command.logs( + self.parser.parse_args( + ["tasks", "logs", self.dag_id, self.run_id, self.task_id, "--try-number", "1"] + ), + api_client=api_client, + ) + + assert _normalize_rich_output(capsys.readouterr().out) == ( + "Task instance for task 'test_task' in Dag run 'test_run' of Dag 'test_dag' not found" + ) + + def test_logs_reraises_non_404_error(self): + api_client = mock.MagicMock() + api_client.task_instances.logs.side_effect = _make_server_error(500) + + with pytest.raises(ServerResponseError): + task_command.logs( + self.parser.parse_args( + ["tasks", "logs", self.dag_id, self.run_id, self.task_id, "--try-number", "1"] + ), + api_client=api_client, + )