Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add task.trigger rule to grid_data #30130

Merged
merged 2 commits into from
Mar 16, 2023
Merged

Add task.trigger rule to grid_data #30130

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
be7030b
add trigger rule to grid_data
bbovenzi Mar 15, 2023
7639f89
fix instance check
bbovenzi Mar 15, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion airflow/www/static/js/dag/InstanceTooltip.test.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,12 +40,18 @@ describe("Test Task InstanceTooltip", () => {
test("Displays a normal task", () => {
const { getByText, queryByText } = render(
<InstanceTooltip
group={{ id: "task", label: "task", instances: [] }}
group={{
id: "task",
label: "task",
instances: [],
triggerRule: "all_failed",
}}
Comment on lines +43 to +48
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This feels wrong. A "group" doesn't have a trigger rule. Should this be "node" or "item", not "group"? Definitely a problem for another PR, but we should probably rename this.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I already started a separate branch to make our names and types consistent.

instance={instance}
/>,
{ wrapper: Wrapper }
);

expect(getByText("Trigger Rule: all_failed")).toBeDefined();
expect(getByText("Status: success")).toBeDefined();
expect(queryByText("Contains a note")).toBeNull();
expect(getByText("Duration: 00:00:00")).toBeDefined();
Expand Down
1 change: 1 addition & 0 deletions airflow/www/static/js/dag/InstanceTooltip.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,7 @@ const InstanceTooltip = ({
</Text>
</>
)}
{group.triggerRule && <Text>Trigger Rule: {group.triggerRule}</Text>}
{note && <Text>Contains a note</Text>}
</Box>
);
Expand Down
8 changes: 7 additions & 1 deletion airflow/www/static/js/dag/details/taskInstance/Details.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ const Details = ({ instance, group, dagId }: Props) => {
const { taskId, runId, startDate, endDate, state, mappedStates, mapIndex } =
instance;

const { isMapped, tooltip, operator, hasOutletDatasets } = group;
const { isMapped, tooltip, operator, hasOutletDatasets, triggerRule } = group;

const { data: apiTI } = useTaskInstance({
dagId,
Expand Down Expand Up @@ -168,6 +168,12 @@ const Details = ({ instance, group, dagId }: Props) => {
<Td>{operator}</Td>
</Tr>
)}
{triggerRule && (
<Tr>
<Td>Trigger Rule</Td>
<Td>{triggerRule}</Td>
</Tr>
)}
{startDate && (
<Tr>
<Td>
Expand Down
3 changes: 2 additions & 1 deletion airflow/www/static/js/types/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
* under the License.
*/

import * as API from "./api-generated";
import type * as API from "./api-generated";

type RunState = "success" | "running" | "queued" | "failed";

Expand Down Expand Up @@ -96,6 +96,7 @@ interface Task {
isMapped?: boolean;
operator?: string;
hasOutletDatasets?: boolean;
triggerRule?: API.TriggerRule;
}

type RunOrdering = (
Expand Down
5 changes: 3 additions & 2 deletions airflow/www/views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@
from airflow.utils.session import NEW_SESSION, create_session, provide_session
from airflow.utils.state import State, TaskInstanceState
from airflow.utils.strings import to_boolean
from airflow.utils.task_group import MappedTaskGroup, task_group_to_dict
from airflow.utils.task_group import MappedTaskGroup, TaskGroup, task_group_to_dict
from airflow.utils.timezone import td_format, utcnow
from airflow.version import version
from airflow.www import auth, utils as wwwutils
Expand Down Expand Up @@ -297,7 +297,7 @@ def dag_to_grid(dag: DagModel, dag_runs: Sequence[DagRun], session: Session):
grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}

def task_group_to_grid(item, grouped_tis, *, is_parent_mapped: bool):
if isinstance(item, AbstractOperator):
if not isinstance(item, TaskGroup):
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

trigger_rule doesn't exist on AbstractOperator. Open to ideas on a better way to check if this is a task or a task group.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems okay to me.


def _get_summary(task_instance):
return {
Expand Down Expand Up @@ -363,6 +363,7 @@ def set_overall_state(record):
"is_mapped": isinstance(item, MappedOperator) or is_parent_mapped,
"has_outlet_datasets": any(isinstance(i, Dataset) for i in (item.outlets or [])),
"operator": item.operator_name,
"trigger_rule": item.trigger_rule,
}

# Task Group
Expand Down
7 changes: 7 additions & 0 deletions tests/www/views/test_views_grid.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,7 @@ def test_no_runs(admin_client, dag_without_runs):
"is_mapped": False,
"label": "task1",
"operator": "EmptyOperator",
"trigger_rule": "all_success",
},
{
"children": [
Expand All @@ -119,6 +120,7 @@ def test_no_runs(admin_client, dag_without_runs):
"is_mapped": True,
"label": "subtask2",
"operator": "MockOperator",
"trigger_rule": "all_success",
}
],
"is_mapped": True,
Expand All @@ -137,6 +139,7 @@ def test_no_runs(admin_client, dag_without_runs):
"is_mapped": True,
"label": "mapped",
"operator": "MockOperator",
"trigger_rule": "all_success",
}
],
"id": "group",
Expand Down Expand Up @@ -255,6 +258,7 @@ def test_one_run(admin_client, dag_with_runs: list[DagRun], session):
"is_mapped": False,
"label": "task1",
"operator": "EmptyOperator",
"trigger_rule": "all_success",
},
{
"children": [
Expand Down Expand Up @@ -283,6 +287,7 @@ def test_one_run(admin_client, dag_with_runs: list[DagRun], session):
"is_mapped": True,
"label": "subtask2",
"operator": "MockOperator",
"trigger_rule": "all_success",
}
],
"is_mapped": True,
Expand Down Expand Up @@ -335,6 +340,7 @@ def test_one_run(admin_client, dag_with_runs: list[DagRun], session):
"is_mapped": True,
"label": "mapped",
"operator": "MockOperator",
"trigger_rule": "all_success",
},
],
"id": "group",
Expand Down Expand Up @@ -398,6 +404,7 @@ def _expected_task_details(task_id, has_outlet_datasets):
"is_mapped": False,
"label": task_id,
"operator": "EmptyOperator",
"trigger_rule": "all_success",
}

assert resp.status_code == 200, resp.json
Expand Down