diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py index c913d91c1e607..43e289a1cc49a 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py @@ -17,14 +17,16 @@ from __future__ import annotations from datetime import datetime -from typing import Annotated +from typing import Annotated, cast from pydantic import ( AliasPath, BeforeValidator, Field, + field_validator, ) +from airflow._shared.secrets_masker import redact from airflow.api_fastapi.core_api.base import BaseModel from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse from airflow.utils.state import TaskInstanceState @@ -62,6 +64,20 @@ class TaskInstanceHistoryResponse(BaseModel): executor: str | None executor_config: Annotated[str, BeforeValidator(str)] dag_version: DagVersionResponse | None + state_reason: str | None = Field( + default=None, + validation_alias="retry_reason", + description=( + "The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended." + ), + ) + + @field_validator("state_reason", mode="after") + @classmethod + def redact_state_reason(cls, v: str | None) -> str | None: + if v is None: + return None + return cast("str", redact(v)) class TaskInstanceHistoryCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py index d138629e73a00..e2d84e9bb6bb0 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py @@ -18,7 +18,7 @@ from collections.abc import Iterable from datetime import datetime -from typing import Annotated, Any +from typing import Annotated, Any, cast from uuid import UUID from pydantic import ( @@ -34,6 +34,7 @@ model_validator, ) +from airflow._shared.secrets_masker import redact from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse from airflow.api_fastapi.core_api.datamodels.job import JobResponse @@ -89,6 +90,23 @@ class TaskInstanceResponse(BaseModel): queued_by_job: JobResponse | None = Field(alias="triggerer_job") dag_version: DagVersionResponse | None team_name: str | None = None + state_reason: str | None = Field( + default=None, + validation_alias="retry_reason", + description=( + "The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended." + ), + ) + + @field_validator("state_reason", mode="after") + @classmethod + def redact_state_reason(cls, v: str | None) -> str | None: + # A retry policy composes this from the exception text, and a policy may opt out of the + # worker-side redaction, so the same string that would be masked in a task log can reach + # here unmasked. + if v is None: + return None + return cast("str", redact(v)) class TaskInstanceCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml index 24938a8f7658a..66032afd6091f 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml @@ -4837,6 +4837,15 @@ components: - type: string - type: 'null' title: Team Name + state_reason: + anyOf: + - type: string + - type: 'null' + title: State Reason + description: 'The reason the task instance reached its current state, as + recorded by a retry policy. May describe a previous attempt: it is cleared + only when the task next starts running, so a task waiting to be retried + or re-run can still carry the reason its last attempt ended.' type: object required: - id diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index 49694cd5d035c..6f515ee2eeaeb 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml @@ -16322,6 +16322,15 @@ components: anyOf: - $ref: '#/components/schemas/DagVersionResponse' - type: 'null' + state_reason: + anyOf: + - type: string + - type: 'null' + title: State Reason + description: 'The reason the task instance reached its current state, as + recorded by a retry policy. May describe a previous attempt: it is cleared + only when the task next starts running, so a task waiting to be retried + or re-run can still carry the reason its last attempt ended.' type: object required: - task_id @@ -16505,6 +16514,15 @@ components: - type: string - type: 'null' title: Team Name + state_reason: + anyOf: + - type: string + - type: 'null' + title: State Reason + description: 'The reason the task instance reached its current state, as + recorded by a retry policy. May describe a previous attempt: it is cleared + only when the task next starts running, so a task waiting to be retried + or re-run can still carry the reason its last attempt ended.' type: object required: - id diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index 71d21c125c53e..ab6d309bd5c20 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -7516,6 +7516,18 @@ export const $TaskInstanceHistoryResponse = { type: 'null' } ] + }, + state_reason: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'State Reason', + description: 'The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended.' } }, type: 'object', @@ -7816,6 +7828,18 @@ export const $TaskInstanceResponse = { } ], title: 'Team Name' + }, + state_reason: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'State Reason', + description: 'The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended.' } }, type: 'object', diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index 776009c9afc25..a2d395d29fda0 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts @@ -1995,6 +1995,10 @@ export type TaskInstanceHistoryResponse = { executor: string | null; executor_config: string; dag_version: DagVersionResponse | null; + /** + * The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended. + */ + state_reason?: string | null; }; /** @@ -2038,6 +2042,10 @@ export type TaskInstanceResponse = { triggerer_job: JobResponse | null; dag_version: DagVersionResponse | null; team_name?: string | null; + /** + * The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended. + */ + state_reason?: string | null; }; /** diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json index fe8b8b93d0d60..7a781ef45ed47 100644 --- a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json +++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json @@ -454,6 +454,11 @@ "queuedWhen": "Queued At", "renderedMapIndex": "Rendered Map Index", "scheduledWhen": "Scheduled At", + "stateReason": "Reason for state", + "stateReasonSummary": { + "failed": "Stopped on try {{tryNumber}} of {{totalTries}}", + "upForRetry": "Retrying after try {{tryNumber}} of {{totalTries}}" + }, "trigger": "Trigger", "triggerer": { "assigned": "Assigned triggerer", diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx new file mode 100644 index 0000000000000..cc7d063bbd744 --- /dev/null +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx @@ -0,0 +1,137 @@ +/*! + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +import "@testing-library/jest-dom"; +import { render, screen } from "@testing-library/react"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import type { TaskInstanceHistoryResponse, TaskInstanceResponse } from "openapi/requests/types.gen"; + +import i18n from "src/i18n/config"; +import { Wrapper } from "src/utils/Wrapper"; + +import commonLocale from "../../../public/i18n/locales/en/common.json"; +import { Details } from "./Details"; + +// Sibling panels each fetch their own data and are unrelated to the row under test. +vi.mock("./BlockingDeps", () => ({ BlockingDeps: () => undefined })); +vi.mock("./ExtraLinks", () => ({ ExtraLinks: () => undefined })); +vi.mock("./TriggererInfo", () => ({ TriggererInfo: () => undefined })); +vi.mock("src/components/DagVersionDetails", () => ({ DagVersionDetails: () => undefined })); +vi.mock("src/components/TaskTrySelect", () => ({ TaskTrySelect: () => undefined })); +vi.mock("src/components/TeamName", () => ({ TeamName: () => undefined })); +vi.mock("src/hooks/useShowTeam", () => ({ useShowTeam: () => false })); + +const mockTaskInstance = vi.fn<() => TaskInstanceResponse | undefined>(); +const mockTryInstance = vi.fn<() => TaskInstanceHistoryResponse | undefined>(); + +vi.mock("openapi/queries", async () => { + const actual = await vi.importActual("openapi/queries"); + + return { + ...actual, + useTaskInstanceServiceGetMappedTaskInstance: () => ({ data: mockTaskInstance() }), + useTaskInstanceServiceGetTaskInstanceTryDetails: () => ({ data: mockTryInstance() }), + }; +}); + +vi.mock("src/utils", async () => { + const actual = await vi.importActual("src/utils"); + + return { ...actual, useAutoRefresh: () => false }; +}); + +const buildTaskInstance = (overrides: Partial): TaskInstanceResponse => + ({ + dag_id: "test_dag", + dag_run_id: "run_1", + dag_version: null, + duration: null, + end_date: null, + id: "ti-id", + map_index: -1, + max_tries: 2, + note: null, + operator_name: "PythonOperator", + rendered_map_index: null, + start_date: null, + state: "failed", + state_reason: null, + task_display_name: "test_task", + task_id: "test_task", + trigger: null, + triggerer_job: null, + try_number: 3, + ...overrides, + }) as unknown as TaskInstanceResponse; + +const renderDetails = ( + taskInstance: TaskInstanceResponse, + tryInstance: Partial = {}, +) => { + mockTaskInstance.mockReturnValue(taskInstance); + mockTryInstance.mockReturnValue({ + ...taskInstance, + ...tryInstance, + }); + + return render(
, { wrapper: Wrapper }); +}; + +describe("Details state reason row", () => { + // Without the bundle i18n.t() echoes the key, so the label assertions below would pass blindly. + beforeEach(() => { + i18n.addResourceBundle("en", "common", commonLocale, true, true); + }); + + it("does not render the banner when there is no reason", () => { + renderDetails(buildTaskInstance({ state_reason: null })); + + expect(screen.queryByText(i18n.t("common:taskInstance.stateReason"))).not.toBeInTheDocument(); + }); + + // Cleared only once the task next reaches RUNNING, so these states still carry a stale reason. + it.each(["queued", "running", "success", null] as const)( + "renders neither the banner nor the row for a %s task that still carries a reason", + (state) => { + renderDetails(buildTaskInstance({ state, state_reason: "auth error, do not retry" })); + + expect(screen.queryByText(i18n.t("common:taskInstance.stateReason"))).not.toBeInTheDocument(); + expect(screen.queryByText("auth error, do not retry")).not.toBeInTheDocument(); + }, + ); + + it("keeps an earlier failed try's reason while the task is running again", () => { + renderDetails(buildTaskInstance({ state: "running", state_reason: null }), { + state: "failed", + state_reason: "try 1: auth error", + }); + + expect(screen.getByText("try 1: auth error")).toBeInTheDocument(); + }); + + it("shows the selected try's reason in the table", () => { + renderDetails(buildTaskInstance({ state_reason: "latest try: rate limit" }), { + state_reason: "older try: auth error", + }); + + expect(screen.getByText("older try: auth error")).toBeInTheDocument(); + expect(screen.queryByText("latest try: rate limit")).not.toBeInTheDocument(); + expect(screen.getByText(i18n.t("common:taskInstance.stateReason"))).toBeInTheDocument(); + }); +}); diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx index 017e24c54c7ae..c35a3462dd784 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx @@ -43,6 +43,7 @@ import { isStatePending, useAutoRefresh, useDurationFormat } from "src/utils"; import { BlockingDeps } from "./BlockingDeps"; import { ExtraLinks } from "./ExtraLinks"; import { TriggererInfo } from "./TriggererInfo"; +import { stateReasonDisplay } from "./stateReason"; export const Details = () => { const { t: translate } = useTranslation(); @@ -115,6 +116,15 @@ export const Details = () => { return translate("common:none", { defaultValue: "None" }); }; + // Keyed off the selected try's own state, so an earlier failed try keeps its reason while the + // current one is running again. + const tryStateReason = + tryInstance?.state_reason !== null && + tryInstance?.state_reason !== undefined && + stateReasonDisplay(tryInstance.state) !== undefined + ? tryInstance.state_reason + : undefined; + // omit kwargs from trigger const triggerWithoutKwargs = taskInstance?.trigger ? (({ kwargs, ...rest }) => rest)(taskInstance.trigger) @@ -162,6 +172,12 @@ export const Details = () => { + {tryStateReason === undefined ? undefined : ( + + {translate("taskInstance.stateReason")} + {tryStateReason} + + )} {translate("taskId")} diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.test.tsx index 34ab9a45cb939..d805f3ed0520b 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.test.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.test.tsx @@ -25,6 +25,7 @@ import type { TaskInstanceResponse } from "openapi/requests/types.gen"; import i18n from "src/i18n/config"; import { Wrapper } from "src/utils/Wrapper"; +import commonLocale from "../../../public/i18n/locales/en/common.json"; import { Header } from "./Header"; // Action buttons and note preview pull in mutation/permission wiring that is @@ -77,3 +78,56 @@ describe("Header", () => { expect(screen.queryByText(i18n.t("common:dagDetails.team"))).not.toBeInTheDocument(); }); }); + +const renderHeader = (overrides: Partial) => + render(
, { wrapper: Wrapper }); + +describe("Header state reason banner", () => { + // Without the bundle i18n.t() echoes the key, so the titles below would assert nothing. + beforeEach(() => { + i18n.addResourceBundle("en", "common", commonLocale, true, true); + }); + + it("does not render when there is no reason", () => { + renderHeader({ state: "failed", state_reason: null }); + + expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); + }); + + // Cleared only once the task next reaches RUNNING, so these states still carry a stale reason. + it.each(["queued", "running", "success", null] as const)( + "does not render for a %s task that still carries a reason", + (state) => { + renderHeader({ state, state_reason: "auth error, do not retry" }); + + expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); + }, + ); + + it.each([ + { maxTries: 2, state: "failed", titleKey: "failed", totalTries: 3, tryNumber: 3 }, + // Differing numbers are what make a swapped or off-by-one interpolation visible. + { maxTries: 3, state: "up_for_retry", titleKey: "upForRetry", totalTries: 4, tryNumber: 2 }, + ] as const)( + "titles the banner for a $state task", + ({ maxTries, state, titleKey, totalTries, tryNumber }) => { + renderHeader({ max_tries: maxTries, state, state_reason: "auth error", try_number: tryNumber }); + + expect(screen.getByTestId("state-reason-alert")).toHaveTextContent( + i18n.t(`common:taskInstance.stateReasonSummary.${titleKey}`, { totalTries, tryNumber }), + ); + expect(screen.getByTestId("state-reason-alert")).toHaveTextContent("auth error"); + }, + ); + + // Chakra puts `status` in a generated class, so "not identical" is all that can be asserted. + it("styles a failed banner differently from an up_for_retry one", () => { + const { unmount } = renderHeader({ state: "failed", state_reason: "auth error" }); + const failedClass = screen.getByTestId("state-reason-alert").className; + + unmount(); + renderHeader({ state: "up_for_retry", state_reason: "auth error" }); + + expect(screen.getByTestId("state-reason-alert").className).not.toBe(failedClass); + }); +}); diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx index 850c3b42e30fa..26d79acd9256a 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx @@ -24,6 +24,8 @@ import { MdOutlineTask } from "react-icons/md"; import type { TaskInstanceResponse } from "openapi/requests/types.gen"; +import { Alert } from "src/system-components"; + import { ClearTaskInstanceButton } from "src/components/Clear"; import ClearTaskInstanceDialog from "src/components/Clear/TaskInstance/ClearTaskInstanceDialog"; import { DagVersion } from "src/components/DagVersion"; @@ -37,6 +39,8 @@ import { useShowTeam } from "src/hooks/useShowTeam"; import { useTaskInstanceNote } from "src/queries/useTaskInstanceNote"; import { useDurationFormat } from "src/utils"; +import { stateReasonDisplay } from "./stateReason"; + export const Header = ({ taskInstance }: { readonly taskInstance: TaskInstanceResponse }) => { const { t: translate } = useTranslation(); const { formatElapsed, renderDuration } = useDurationFormat(); @@ -80,8 +84,24 @@ export const Header = ({ taskInstance }: { readonly taskInstance: TaskInstanceRe // Stable dialog state at header/page level const [clearOpen, setClearOpen] = useState(false); + // On the header, not the details tab, so it shows on every tab without duplicating the row. + const stateReasonDisplayed = stateReasonDisplay(taskInstance.state); + const stateReason = taskInstance.state_reason; + return ( + {stateReasonDisplayed === undefined || stateReason === null || stateReason === undefined ? undefined : ( + + {stateReason} + + )} diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/stateReason.ts b/airflow-core/src/airflow/ui/src/pages/TaskInstance/stateReason.ts new file mode 100644 index 0000000000000..4ed28ac8df364 --- /dev/null +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/stateReason.ts @@ -0,0 +1,30 @@ +/*! + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +// The header banner and the per-try row both look the state up here, so a state added to one +// surface cannot be forgotten on the other: it has to bring a title with it. +const STATE_REASON_DISPLAY = { + failed: { status: "error", titleKey: "failed" }, + up_for_retry: { status: "warning", titleKey: "upForRetry" }, +} as const satisfies Record; + +export const stateReasonDisplay = (state: string | null | undefined) => + state === null || state === undefined + ? undefined + : (STATE_REASON_DISPLAY as Record)[state]; diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py index dab5bef1a2f51..0df529b047a6e 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py @@ -268,6 +268,7 @@ def expected_sample_hitl_detail_dict(sample_ti: TaskInstance) -> dict[str, Any]: "task_display_name": "sample_task_hitl", "task_id": TASK_ID, "team_name": None, + "state_reason": None, "trigger": None, "triggerer_job": None, "try_number": 0, diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py index 7d14b4a7c0515..cf5123f20fb5e 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py @@ -31,6 +31,7 @@ from sqlalchemy import delete, func, select, update from sqlalchemy.orm import joinedload +from airflow._shared.secrets_masker import mask_secret from airflow._shared.state import TaskScope from airflow._shared.timezones.timezone import datetime from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser @@ -244,8 +245,40 @@ def test_should_respond_200(self, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } + def test_should_include_state_reason(self, test_client, session): + self.create_task_instances(session, task_instances=[{"retry_reason": "auth error, do not retry"}]) + response = test_client.get( + "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context" + ) + assert response.status_code == 200 + assert response.json()["state_reason"] == "auth error, do not retry" + + @pytest.fixture + def masked_secret(self): + """The masker is a cached process global, so drop the pattern again for the next test.""" + from airflow._shared.secrets_masker import _secrets_masker + + masker = _secrets_masker() + patterns, replacer = set(masker.patterns), masker.replacer + mask_secret("hunter2") + yield + masker.patterns, masker.replacer = patterns, replacer + + @pytest.mark.enable_redact + def test_should_redact_secrets_in_state_reason(self, test_client, session, masked_secret): + """A policy may compose the reason from an unredacted exception, so mask on the way out.""" + self.create_task_instances( + session, task_instances=[{"retry_reason": "auth: the token hunter2 expired"}] + ) + response = test_client.get( + "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context" + ) + assert response.status_code == 200 + assert response.json()["state_reason"] == "auth: the token *** expired" + @conf_vars({("core", "multi_team"): "True"}) def test_should_include_team_name(self, test_client, session): self.create_task_instances(session) @@ -329,6 +362,7 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, "dag_version": { "id": response_data["dag_version"]["id"], "version_number": expected_version_number, @@ -425,6 +459,7 @@ def test_should_respond_200_with_task_state_in_deferred(self, test_client, sessi "unixname": getuser(), }, "team_name": None, + "state_reason": None, } def test_should_respond_200_with_task_state_in_removed(self, test_client, session): @@ -479,6 +514,7 @@ def test_should_respond_200_with_task_state_in_removed(self, test_client, sessio "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } def test_should_respond_200_task_instance_with_rendered(self, test_client, session): @@ -536,6 +572,7 @@ def test_should_respond_200_task_instance_with_rendered(self, test_client, sessi "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } def test_raises_404_for_nonexistent_task_instance(self, test_client): @@ -657,6 +694,7 @@ def test_should_respond_200_mapped_task_instance_with_rtif(self, test_client, se "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } def test_should_respond_401(self, unauthenticated_test_client): @@ -2805,8 +2843,44 @@ def test_should_respond_200(self, test_client, session): "id": response_data["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, } + def test_should_include_state_reason_from_history(self, test_client, session): + self.create_task_instances( + session, + task_instances=[{"state": State.SUCCESS, "retry_reason": "auth error, do not retry"}], + with_ti_history=True, + ) + response = test_client.get( + "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context/tries/1" + ) + assert response.status_code == 200 + assert response.json()["state_reason"] == "auth error, do not retry" + + @pytest.fixture + def masked_secret(self): + from airflow._shared.secrets_masker import _secrets_masker + + masker = _secrets_masker() + patterns, replacer = set(masker.patterns), masker.replacer + mask_secret("hunter2") + yield + masker.patterns, masker.replacer = patterns, replacer + + @pytest.mark.enable_redact + def test_should_redact_secrets_in_state_reason_from_history(self, test_client, session, masked_secret): + self.create_task_instances( + session, + task_instances=[{"state": State.SUCCESS, "retry_reason": "auth: the token hunter2 expired"}], + with_ti_history=True, + ) + response = test_client.get( + "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context/tries/1" + ) + assert response.status_code == 200 + assert response.json()["state_reason"] == "auth: the token *** expired" + @pytest.mark.parametrize("try_number", [1, 2]) def test_should_respond_200_with_different_try_numbers(self, test_client, try_number, session): self.create_task_instances(session, task_instances=[{"state": State.SUCCESS}], with_ti_history=True) @@ -2851,6 +2925,7 @@ def test_should_respond_200_with_different_try_numbers(self, test_client, try_nu "id": response_data["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, } @pytest.mark.parametrize("try_number", [1, 2]) @@ -2928,6 +3003,7 @@ def test_should_respond_200_with_mapped_task_at_different_try_numbers( "id": response_data["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, } def test_should_respond_200_with_task_state_in_deferred(self, test_client, session): @@ -3000,6 +3076,7 @@ def test_should_respond_200_with_task_state_in_deferred(self, test_client, sessi "id": response_data["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, } def test_should_respond_200_with_task_state_in_removed(self, test_client, session): @@ -3047,6 +3124,7 @@ def test_should_respond_200_with_task_state_in_removed(self, test_client, sessio "id": response_data["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, } def test_should_respond_401(self, unauthenticated_test_client): @@ -3122,6 +3200,7 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve "created_at": mock.ANY, "dag_display_name": "dag_with_multiple_versions", }, + "state_reason": None, } def test_should_not_return_duplicate_runs(self, test_client, session): @@ -3859,6 +3938,7 @@ def test_should_respond_200_with_dag_run_id( "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, "try_number": 0, "unixname": getuser(), }, @@ -4418,6 +4498,7 @@ def test_should_respond_200(self, test_client, session): "id": response_data["task_instances"][0]["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, }, { "dag_id": "example_python_operator", @@ -4455,6 +4536,7 @@ def test_should_respond_200(self, test_client, session): "id": response_data["task_instances"][1]["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, }, ], "total_entries": 2, @@ -4526,6 +4608,7 @@ def test_ti_in_retry_state_not_returned(self, test_client, session): "id": response_data["task_instances"][0]["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, }, ], "total_entries": 1, @@ -4609,6 +4692,7 @@ def test_mapped_task_should_respond_200(self, test_client, session): "id": response_data["task_instances"][0]["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, }, { "dag_id": "example_python_operator", @@ -4646,6 +4730,7 @@ def test_mapped_task_should_respond_200(self, test_client, session): "id": response_data["task_instances"][1]["dag_version"]["id"], "version_number": 1, }, + "state_reason": None, }, ], "total_entries": 2, @@ -4713,6 +4798,7 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve "created_at": mock.ANY, "dag_display_name": "dag_with_multiple_versions", }, + "state_reason": None, } @@ -4838,6 +4924,7 @@ def test_should_call_mocked_api(self, mock_set_ti_state, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } ], "total_entries": 1, @@ -5116,6 +5203,7 @@ def test_should_raise_422_for_invalid_task_instance_state(self, payload, expecte "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } ], "total_entries": 1, @@ -5256,6 +5344,7 @@ def test_update_mask_set_note_should_respond_200( "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } ], "total_entries": 1, @@ -5321,6 +5410,7 @@ def test_set_note_should_respond_200(self, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } ], "total_entries": 1, @@ -5418,6 +5508,7 @@ def test_set_note_should_respond_200_mapped_task_with_rtif(self, test_client, se "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } ], "total_entries": 1, @@ -5503,6 +5594,7 @@ def test_set_note_should_respond_200_mapped_task_summary_with_rtif(self, test_cl "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } _check_task_instance_note( @@ -5699,6 +5791,7 @@ def test_should_call_mocked_api(self, mock_set_ti_state, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } ], "total_entries": 1, @@ -5989,6 +6082,7 @@ def test_should_raise_422_for_invalid_task_instance_state(self, payload, expecte "trigger": None, "triggerer_job": None, "team_name": None, + "state_reason": None, } ], "total_entries": 1, diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py index 8938ad8d0534b..1846803d774da 100644 --- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py +++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py @@ -2359,6 +2359,13 @@ class TaskInstanceHistoryResponse(BaseModel): executor: Annotated[str | None, Field(title="Executor")] executor_config: Annotated[str, Field(title="Executor Config")] dag_version: DagVersionResponse | None + state_reason: Annotated[ + str | None, + Field( + description="The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended.", + title="State Reason", + ), + ] = None class TaskInstanceResponse(BaseModel): @@ -2401,6 +2408,13 @@ class TaskInstanceResponse(BaseModel): triggerer_job: JobResponse | None dag_version: DagVersionResponse | None team_name: Annotated[str | None, Field(title="Team Name")] = None + state_reason: Annotated[ + str | None, + Field( + description="The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended.", + title="State Reason", + ), + ] = None class TaskResponse(BaseModel): diff --git a/providers/common/ai/docs/retry_policies.rst b/providers/common/ai/docs/retry_policies.rst index cd88962473f03..62e1c721d2517 100644 --- a/providers/common/ai/docs/retry_policies.rst +++ b/providers/common/ai/docs/retry_policies.rst @@ -133,7 +133,10 @@ When a task fails, either policy: ``retry_reason``, on a FAIL as well as a RETRY: ``: `` from ``LLMRetryPolicy``, or one line such as ``category=network confidence=0.91 threshold=0.60 action=retry delay=10s`` - from ``ClassifierRetryPolicy``. + from ``ClassifierRetryPolicy``. From Airflow 3.4 the REST API exposes it as + ``state_reason`` on a task instance and on each try, and the Task Instance + page shows it under **Reason for state** while the task is failed or up for + retry. This classification call is a separate model request, made by the policy itself rather than by an operator -- it is not subject to an operator's @@ -406,11 +409,12 @@ upper limit; zero or negative means no override, so the task's own ``retry_delay`` and backoff apply. ``category`` and ``reasoning`` become the ``retry_reason`` (truncated to 500 -characters), recorded on both outcomes. On a RETRY the value is cleared once the next attempt starts running; -a FAIL is terminal, so there is no next attempt to clear it and the reason stays -on the row. Only the model's own words are stored -- attempt counts are left to -whatever displays the reason. Recording on a FAIL requires Airflow 3.4.0; on -earlier versions only the RETRY outcome is recorded. +characters), recorded on both outcomes. On a RETRY the value is cleared once +the next attempt starts running; a FAIL is terminal, so there is no next +attempt to clear it and the reason stays on the row. Only the model's own words +are stored -- attempt counts are left to whatever displays the reason. +Recording on a FAIL requires Airflow 3.4.0; on earlier versions only the RETRY +outcome is recorded. Under ``ClassifierRetryPolicy`` it answers the category name and nothing else. It does not decide whether to retry, it does not choose the delay, and it does not explain diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py b/task-sdk/src/airflow/sdk/execution_time/task_runner.py index 46f2fdae41128..3ca875c651750 100644 --- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py +++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py @@ -30,7 +30,7 @@ from datetime import datetime, timedelta, timezone from itertools import product from pathlib import Path -from typing import TYPE_CHECKING, Annotated, Any, Literal +from typing import TYPE_CHECKING, Annotated, Any, Literal, cast from urllib.parse import quote import attrs @@ -1816,6 +1816,10 @@ def _evaluate_retry_policy( Returns ``None`` when no policy is configured so the caller falls through to the standard retry logic. """ + from dataclasses import replace + + from airflow.sdk._shared.secrets_masker import redact + policy = getattr(ti.task, "retry_policy", None) if policy is None: return None @@ -1828,6 +1832,9 @@ def _evaluate_retry_policy( context=context, ) if decision.reason: + # Mask here, where mask_secret() registered the value: the API server rendering this + # later has its own masker and does not know the worker's secrets. + decision = replace(decision, reason=cast("str", redact(decision.reason))) # Close the group so the retry policy decision is not hidden inside "Post Execute". log.info("::endgroup::") log.info("Retry policy decision", action=decision.action.value, reason=decision.reason) diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py index 198b6ec070b9c..b14e3e1bb696d 100644 --- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py +++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py @@ -64,6 +64,7 @@ timezone, ) from airflow.sdk._shared.observability.metrics.base_stats_logger import StatsLogger +from airflow.sdk._shared.secrets_masker import _secrets_masker from airflow.sdk._shared.state import AssetScope, TaskScope from airflow.sdk.api.datamodels._generated import ( AssetProfile, @@ -1274,6 +1275,29 @@ def execute(self, context): assert msg.retry_reason == "z" * 500 +@pytest.mark.enable_redact +def test_retry_policy_reason_is_redacted_in_the_worker(create_runtime_ti, mock_supervisor_comms): + """The reason is masked where mask_secret() registered the value, not in the API server.""" + _secrets_masker().add_mask("hunter2", None) + + class _AlwaysFails(BaseOperator): + def execute(self, context): + raise RuntimeError("403 Forbidden: token hunter2 expired") + + class _EchoPolicy(RetryPolicy): + def evaluate(self, exception, try_number, max_tries, context=None): + return RetryDecision(action=RetryAction.FAIL, reason=f"auth: {exception}") + + task = _AlwaysFails(task_id="redacted_reason", retries=2, retry_policy=_EchoPolicy()) + ti = create_runtime_ti(task=task) + + state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) + + assert state == TaskInstanceState.FAILED + assert isinstance(msg, TaskState) + assert msg.retry_reason == "auth: 403 Forbidden: token *** expired" + + def test_plain_retries_exhausted_has_no_reason(create_runtime_ti, mock_supervisor_comms): """Without a retry policy, exhausting the retry budget must not synthesize a reason."""