From 11aba400d58a02ae3320e9c660359704497a013e Mon Sep 17 00:00:00 2001 From: vincbeck Date: Wed, 16 Sep 2026 15:29:40 -0400 Subject: [PATCH] UI: Add team column and filter to the jobs list When multi-team mode is enabled, operators triaging scheduler, triggerer and Dag processor jobs need to see which team's workloads each job belongs to, and to scope the list to a single team when investigating that team's resource usage. The column and the filter render only when the `multi_team` configuration is enabled, so single-team deployments are unaffected. --- .../api_fastapi/common/parameters/__init__.py | 3 + .../api_fastapi/common/parameters/job.py | 50 ++++++++++++++++ .../openapi/v2-rest-api-generated.yaml | 8 +++ .../api_fastapi/core_api/routes/public/job.py | 3 + .../cli/commands/dag_processor_command.py | 26 ++++++++- .../airflow/ui/openapi-gen/queries/common.ts | 5 +- .../ui/openapi-gen/queries/ensureQueryData.ts | 6 +- .../ui/openapi-gen/queries/prefetch.ts | 6 +- .../airflow/ui/openapi-gen/queries/queries.ts | 6 +- .../ui/openapi-gen/queries/suspense.ts | 6 +- .../ui/openapi-gen/requests/services.gen.ts | 4 +- .../ui/openapi-gen/requests/types.gen.ts | 1 + .../src/airflow/ui/src/pages/Jobs.tsx | 22 ++++++- .../core_api/routes/public/test_job.py | 17 ++++++ .../commands/test_dag_processor_command.py | 58 +++++++++++++++++++ 15 files changed, 206 insertions(+), 15 deletions(-) create mode 100644 airflow-core/src/airflow/api_fastapi/common/parameters/job.py diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters/__init__.py b/airflow-core/src/airflow/api_fastapi/common/parameters/__init__.py index 39e748a297204..deae80a25acc9 100644 --- a/airflow-core/src/airflow/api_fastapi/common/parameters/__init__.py +++ b/airflow-core/src/airflow/api_fastapi/common/parameters/__init__.py @@ -86,6 +86,9 @@ FilterParam as FilterParam, filter_param_factory as filter_param_factory, ) +from airflow.api_fastapi.common.parameters.job import ( + QueryJobTeamsFilter as QueryJobTeamsFilter, +) from airflow.api_fastapi.common.parameters.misc import ( QueryConnectionIdPatternSearch as QueryConnectionIdPatternSearch, QueryConnectionIdPrefixPatternSearch as QueryConnectionIdPrefixPatternSearch, diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters/job.py b/airflow-core/src/airflow/api_fastapi/common/parameters/job.py new file mode 100644 index 0000000000000..6c418d2da48f7 --- /dev/null +++ b/airflow-core/src/airflow/api_fastapi/common/parameters/job.py @@ -0,0 +1,50 @@ +# 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. + +from __future__ import annotations + +from typing import TYPE_CHECKING, Annotated + +from fastapi import Depends, Query +from sqlalchemy import select as sql_select + +from airflow.api_fastapi.common.parameters.base import BaseParam +from airflow.jobs.job import Job +from airflow.models.team import JobTeam + +if TYPE_CHECKING: + from sqlalchemy.sql import Select + + +class _JobTeamsFilter(BaseParam[list[str]]): + """Filter jobs by the teams they serve (via the ``JobTeam`` association).""" + + def to_orm(self, select: Select) -> Select: + if self.skip_none is False: + raise ValueError(f"Cannot set 'skip_none' to False on a {type(self)}") + + if not self.value: + return select + + return select.where(Job.id.in_(sql_select(JobTeam.job_id).where(JobTeam.team_name.in_(self.value)))) + + @classmethod + def depends(cls, teams: list[str] = Query(default_factory=list)) -> _JobTeamsFilter: + return cls().set_value(teams) + + +QueryJobTeamsFilter = Annotated[_JobTeamsFilter, Depends(_JobTeamsFilter.depends)] 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..5466522fc471b 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 @@ -5462,6 +5462,14 @@ paths: - type: string - type: 'null' title: Executor Class + - name: teams + in: query + required: false + schema: + type: array + items: + type: string + title: Teams responses: '200': description: Successful Response diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py index 3e7e16bb69ad0..3760b461b233e 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py @@ -28,6 +28,7 @@ ) from airflow.api_fastapi.common.parameters import ( FilterParam, + QueryJobTeamsFilter, QueryLimit, QueryOffset, RangeFilter, @@ -101,6 +102,7 @@ def get_jobs( FilterParam[str | None], Depends(filter_param_factory(Job.executor_class, str | None, filter_name="executor_class")), ], + teams: QueryJobTeamsFilter, is_alive: bool | None = None, ) -> JobCollectionResponse: """Get all jobs.""" @@ -116,6 +118,7 @@ def get_jobs( job_type, hostname, executor_class, + teams, ], order_by=order_by, limit=limit, diff --git a/airflow-core/src/airflow/cli/commands/dag_processor_command.py b/airflow-core/src/airflow/cli/commands/dag_processor_command.py index 104065d6d78c5..5bf28fb0c2875 100644 --- a/airflow-core/src/airflow/cli/commands/dag_processor_command.py +++ b/airflow-core/src/airflow/cli/commands/dag_processor_command.py @@ -22,6 +22,8 @@ from typing import Any from airflow.cli.commands.daemon_utils import run_command_with_daemon_option +from airflow.configuration import conf +from airflow.dag_processing.bundles.manager import _get_configured_bundle_team_names from airflow.dag_processing.manager import DagFileProcessorManager from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner from airflow.jobs.job import Job, run_job @@ -33,12 +35,34 @@ log = logging.getLogger(__name__) +def _get_team_names(bundle_names: list[str] | None) -> list[str]: + """ + Return the teams this Dag processor serves, sorted and de-duplicated. + + Teams are resolved from the bundle configuration rather than the metadata DB: the job row is + written before ``sync_bundles()`` runs, so a DB lookup would see no rows on a fresh deployment + and stale rows right after a bundle is reassigned in config. Config is the source of truth, and + is what ``airflow_health.py`` resolves teams from too. + + A processor started without ``--bundle-name`` parses every configured bundle, so it serves + every configured team. A bundle mapped to no team contributes no team, so a processor parsing + only team-less (or unknown) bundles is not team-scoped and serves the empty list. Outside + multi-team mode team scoping is disabled entirely, matching ``DagFileProcessorManager``. + """ + if not conf.getboolean("core", "multi_team"): + return [] + + configured = _get_configured_bundle_team_names() + names = bundle_names or list(configured) + return sorted({team for name in names if (team := configured.get(name)) is not None}) + + def _create_dag_processor_job_runner(args: Any) -> DagProcessorJobRunner: """Create DagFileProcessorProcess instance.""" if args.bundle_name: cli_utils.validate_dag_bundle_arg(args.bundle_name) return DagProcessorJobRunner( - job=Job(bundle_names=args.bundle_name), + job=Job(bundle_names=args.bundle_name, team_names=_get_team_names(args.bundle_name)), processor=DagFileProcessorManager( max_runs=args.num_runs, bundle_names_to_parse=args.bundle_name, diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts index 05479e1f51d09..2440e7457ab28 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts @@ -786,7 +786,7 @@ export const UseImportErrorServiceGetImportErrorsKeyFn = ({ bundleName, filename export type JobServiceGetJobsDefaultResponse = Awaited>; export type JobServiceGetJobsQueryResult = UseQueryResult; export const useJobServiceGetJobsKey = "JobServiceGetJobs"; -export const UseJobServiceGetJobsKeyFn = ({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: { +export const UseJobServiceGetJobsKeyFn = ({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }: { dagId?: string; endDateGt?: string; endDateGte?: string; @@ -804,7 +804,8 @@ export const UseJobServiceGetJobsKeyFn = ({ dagId, endDateGt, endDateGte, endDat startDateGte?: string; startDateLt?: string; startDateLte?: string; -} = {}, queryKey?: Array) => [useJobServiceGetJobsKey, ...(queryKey ?? [{ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }])]; + teams?: string[]; +} = {}, queryKey?: Array) => [useJobServiceGetJobsKey, ...(queryKey ?? [{ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }])]; export type PluginServiceGetPluginsDefaultResponse = Awaited>; export type PluginServiceGetPluginsQueryResult = UseQueryResult; export const usePluginServiceGetPluginsKey = "PluginServiceGetPlugins"; diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts index e1c55f73cf52a..7533196f99ba7 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts @@ -1561,10 +1561,11 @@ export const ensureUseImportErrorServiceGetImportErrorsData = (queryClient: Quer * @param data.jobType * @param data.hostname * @param data.executorClass +* @param data.teams * @returns JobCollectionResponse Successful Response * @throws ApiError */ -export const ensureUseJobServiceGetJobsData = (queryClient: QueryClient, { dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: { +export const ensureUseJobServiceGetJobsData = (queryClient: QueryClient, { dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }: { dagId?: string; endDateGt?: string; endDateGte?: string; @@ -1582,7 +1583,8 @@ export const ensureUseJobServiceGetJobsData = (queryClient: QueryClient, { dagId startDateGte?: string; startDateLt?: string; startDateLte?: string; -} = {}) => queryClient.ensureQueryData({ queryKey: Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }), queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }) }); + teams?: string[]; +} = {}) => queryClient.ensureQueryData({ queryKey: Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }), queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }) }); /** * Get Plugins * @param data The data for the request. diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts index 613336b861bad..0b4134129c761 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts @@ -1561,10 +1561,11 @@ export const prefetchUseImportErrorServiceGetImportErrors = (queryClient: QueryC * @param data.jobType * @param data.hostname * @param data.executorClass +* @param data.teams * @returns JobCollectionResponse Successful Response * @throws ApiError */ -export const prefetchUseJobServiceGetJobs = (queryClient: QueryClient, { dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: { +export const prefetchUseJobServiceGetJobs = (queryClient: QueryClient, { dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }: { dagId?: string; endDateGt?: string; endDateGte?: string; @@ -1582,7 +1583,8 @@ export const prefetchUseJobServiceGetJobs = (queryClient: QueryClient, { dagId, startDateGte?: string; startDateLt?: string; startDateLte?: string; -} = {}) => queryClient.prefetchQuery({ queryKey: Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }), queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }) }); + teams?: string[]; +} = {}) => queryClient.prefetchQuery({ queryKey: Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }), queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }) }); /** * Get Plugins * @param data The data for the request. diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts index 2fd053cb65c04..66c02de525ff6 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts @@ -1561,10 +1561,11 @@ export const useImportErrorServiceGetImportErrors = = unknown[]>({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: { +export const useJobServiceGetJobs = = unknown[]>({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }: { dagId?: string; endDateGt?: string; endDateGte?: string; @@ -1582,7 +1583,8 @@ export const useJobServiceGetJobs = , "queryKey" | "queryFn">) => useQuery({ queryKey: Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }, queryKey), queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }) as TData, ...options }); + teams?: string[]; +} = {}, queryKey?: TQueryKey, options?: Omit, "queryKey" | "queryFn">) => useQuery({ queryKey: Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }, queryKey), queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }) as TData, ...options }); /** * Get Plugins * @param data The data for the request. diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts index 9add7eba93e19..c333cdf074129 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts @@ -1561,10 +1561,11 @@ export const useImportErrorServiceGetImportErrorsSuspense = = unknown[]>({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: { +export const useJobServiceGetJobsSuspense = = unknown[]>({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }: { dagId?: string; endDateGt?: string; endDateGte?: string; @@ -1582,7 +1583,8 @@ export const useJobServiceGetJobsSuspense = , "queryKey" | "queryFn">) => useSuspenseQuery({ queryKey: Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }, queryKey), queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }) as TData, ...options }); + teams?: string[]; +} = {}, queryKey?: TQueryKey, options?: Omit, "queryKey" | "queryFn">) => useSuspenseQuery({ queryKey: Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }, queryKey), queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte, teams }) as TData, ...options }); /** * Get Plugins * @param data The data for the request. diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts index 7fe097a0bba55..4a8fd03e90341 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts @@ -3619,6 +3619,7 @@ export class JobService { * @param data.jobType * @param data.hostname * @param data.executorClass + * @param data.teams * @returns JobCollectionResponse Successful Response * @throws ApiError */ @@ -3643,7 +3644,8 @@ export class JobService { dag_id: data.dagId, job_type: data.jobType, hostname: data.hostname, - executor_class: data.executorClass + executor_class: data.executorClass, + teams: data.teams }, errors: { 400: 'Bad Request', 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..d6e05925f3f35 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 @@ -4489,6 +4489,7 @@ export type GetJobsData = { startDateGte?: string | null; startDateLt?: string | null; startDateLte?: string | null; + teams?: Array<(string)>; }; export type GetJobsResponse = JobCollectionResponse; diff --git a/airflow-core/src/airflow/ui/src/pages/Jobs.tsx b/airflow-core/src/airflow/ui/src/pages/Jobs.tsx index 0db02b1e26c5e..7d66ab08feeb7 100644 --- a/airflow-core/src/airflow/ui/src/pages/Jobs.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Jobs.tsx @@ -33,9 +33,10 @@ import { StateBadge } from "src/components/StateBadge"; import Time from "src/components/Time"; import { SearchParamsKeys } from "src/constants/searchParams"; +import { useConfig } from "src/queries/useConfig"; import { useDocumentTitle, useFiltersHandler, type FilterableSearchParamsKeys } from "src/utils"; -const createColumns = (translate: TFunction): Array> => [ +const createColumns = (translate: TFunction, multiTeam: boolean): Array> => [ { accessorKey: "id", header: translate("jobs.columns.id"), @@ -44,6 +45,16 @@ const createColumns = (translate: TFunction): Array> => [ accessorKey: "job_type", header: translate("jobs.columns.jobType"), }, + ...(multiTeam + ? ([ + { + accessorKey: "team_names", + cell: ({ row: { original } }) => original.team_names?.join(", "), + enableSorting: false, + header: translate("common:dagDetails.team"), + }, + ] as Array>) + : []), { accessorKey: "state", cell: ({ @@ -97,15 +108,18 @@ const jobsFilterKeys: Array = [ export const Jobs = () => { const { t: translate } = useTranslation(["admin", "common"]); + const multiTeamEnabled = Boolean(useConfig("multi_team")); useDocumentTitle(translate("common:browse.jobs")); const { setTableURLState, tableURLState } = useTableURLState(); const [searchParams] = useSearchParams(); - const { filterConfigs, handleFiltersChange, initialValues } = useFiltersHandler(jobsFilterKeys); + const { filterConfigs, handleFiltersChange, initialValues } = useFiltersHandler( + multiTeamEnabled ? [...jobsFilterKeys, SearchParamsKeys.TEAMS] : jobsFilterKeys, + ); - const columns = createColumns(translate); + const columns = createColumns(translate, multiTeamEnabled); const { pagination, sorting } = tableURLState; const [sort] = sorting; @@ -119,6 +133,7 @@ export const Jobs = () => { const filteredStartDateLte = searchParams.get(SearchParamsKeys.START_DATE_LTE); const filteredEndDateGte = searchParams.get(SearchParamsKeys.END_DATE_GTE); const filteredEndDateLte = searchParams.get(SearchParamsKeys.END_DATE_LTE); + const teams = searchParams.getAll(SearchParamsKeys.TEAMS); const { data, error, isFetching, isLoading } = useJobServiceGetJobs({ endDateGte: filteredEndDateGte ?? undefined, @@ -132,6 +147,7 @@ export const Jobs = () => { orderBy, startDateGte: filteredStartDateGte ?? undefined, startDateLte: filteredStartDateLte ?? undefined, + teams: teams.length > 0 ? teams : undefined, }); return ( diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py index f969d48b30562..7c51ac5c6f708 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py @@ -222,6 +222,23 @@ def test_get_jobs_includes_team_names_and_bundle_names( assert response_json["jobs"][0]["team_names"] == sorted([testing_team.name, extra_team.name]) assert response_json["jobs"][0]["bundle_names"] == ["bundle-a", "bundle-b"] + def test_get_jobs_filters_by_teams(self, test_client, session: Session, testing_team): + clear_db_jobs() + session.add_all( + [ + Job(state=JobState.RUNNING, job_type="SchedulerJob", team_names=[testing_team.name]), + Job(state=JobState.RUNNING, job_type="SchedulerJob"), + ] + ) + session.commit() + + response = test_client.get("/jobs", params={"teams": [testing_team.name]}) + + assert response.status_code == 200 + response_json = response.json() + assert response_json["total_entries"] == 1 + assert response_json["jobs"][0]["team_names"] == [testing_team.name] + def test_should_raises_401_unauthenticated(self, unauthenticated_test_client): response = unauthenticated_test_client.get("/jobs") assert response.status_code == 401 diff --git a/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py b/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py index 1631001b59432..7f8d963e108b9 100644 --- a/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py +++ b/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py @@ -29,6 +29,26 @@ pytestmark = pytest.mark.db_test +# The Dag processor resolves teams from the configured bundle partition, not the metadata DB, so +# that a fresh or reassigned deployment gets the right team before ``sync_bundles()`` has run. +BUNDLE_TEAMS = { + "bundle_a": "team_a", + "other_bundle_a": "team_a", + "bundle_b": "team_b", + "global_bundle": None, +} + + +@pytest.fixture +def dag_bundles_with_teams(): + with ( + conf_vars({("core", "multi_team"): "True"}), + mock.patch.object( + dag_processor_command, "_get_configured_bundle_team_names", return_value=dict(BUNDLE_TEAMS) + ), + ): + yield + class TestDagProcessorCommand: """ @@ -58,6 +78,44 @@ def test_bundle_names_passed(self, mock_runner, configure_testing_dag_bundle): assert mock_runner.call_args.kwargs["processor"].bundle_names_to_parse == ["testing"] assert mock_runner.call_args.kwargs["job"].bundle_names == ["testing"] + @pytest.mark.usefixtures("dag_bundles_with_teams") + @pytest.mark.parametrize( + ("bundle_names", "expected_team_names"), + [ + pytest.param(None, ["team_a", "team_b"], id="every-configured-bundle"), + pytest.param(["bundle_a"], ["team_a"], id="one-bundle-of-a-team"), + pytest.param(["bundle_a", "other_bundle_a"], ["team_a"], id="several-bundles-of-one-team"), + pytest.param(["bundle_a", "bundle_b"], ["team_a", "team_b"], id="bundles-of-several-teams"), + pytest.param(["bundle_a", "global_bundle"], ["team_a"], id="bundles-of-a-team-and-of-no-team"), + pytest.param(["global_bundle"], [], id="bundle-of-no-team"), + pytest.param(["unknown_bundle"], [], id="unknown-bundle"), + ], + ) + def test_get_team_names(self, bundle_names, expected_team_names): + assert dag_processor_command._get_team_names(bundle_names) == expected_team_names + + @conf_vars({("core", "multi_team"): "False"}) + @mock.patch.object( + dag_processor_command, "_get_configured_bundle_team_names", return_value=dict(BUNDLE_TEAMS) + ) + def test_get_team_names_returns_empty_outside_multi_team(self, mock_configured): + assert dag_processor_command._get_team_names(["bundle_a"]) == [] + mock_configured.assert_not_called() + + @conf_vars({("core", "load_examples"): "False"}) + @mock.patch("airflow.cli.commands.dag_processor_command.DagProcessorJobRunner") + @mock.patch("airflow.utils.cli.validate_dag_bundle_arg") + @pytest.mark.usefixtures("dag_bundles_with_teams") + def test_job_records_the_teams_owning_the_parsed_bundles(self, _, mock_runner): + mock_runner.return_value.job_type = "DagProcessorJob" + args = self.parser.parse_args( + ["dag-processor", "--bundle-name", "bundle_a", "--bundle-name", "bundle_b"] + ) + + dag_processor_command.dag_processor(args) + + assert mock_runner.call_args.kwargs["job"].team_names == ["team_a", "team_b"] + @conf_vars({("core", "load_examples"): "False"}) @mock.patch("airflow.cli.commands.dag_processor_command.DagProcessorJobRunner") @mock.patch("airflow.utils.cli.DagBundlesManager", autospec=True)