vincbeck commented on code in PR #73222:
URL: https://github.com/apache/airflow/pull/73222#discussion_r4038633677
##########
airflow-core/src/airflow/api/common/airflow_health.py:
##########
@@ -78,14 +95,65 @@ def _dag_processor_instance_health(job: Job) -> dict[str,
Any]:
}
-def _aggregate_detailed_status(jobs: list[Job]) -> str:
- """detailed_status: healthy (all alive), degraded (some alive), down (none
alive)."""
- alive_count = sum(1 for job in jobs if job.is_alive())
- if alive_count == 0:
- return DOWN
- if alive_count == len(jobs):
+# ``detailed_status`` answers "is every part of this component's work being
done", which needs a
+# denominator. Counting job rows cannot supply one: ``end_date`` is only
written by a cooperative
+# shutdown, so a replica lost to SIGKILL, an OOM kill, or a node eviction
leaves an unfinished row
+# behind forever and a restarted replica adds a second one. The denominator is
therefore taken from
+# the declared work partition instead, which is unaffected by how replicas
come and go:
+#
+# * Dag processor -- the bundles in ``[dag_processor] dag_bundle_config_list``.
+# * Triggerer -- the team scopes those bundles declare, since a triggerer only
picks up triggers for
+# its own team (see ``Trigger.ids_for_triggerer``).
+# * Scheduler -- schedulers are symmetric, so there is no partition and no
partial state to report.
+
+
+def _configured_bundle_teams() -> dict[str, str | None]:
+ """Map every configured Dag bundle to the team owning it, empty when the
config is unreadable."""
+ try:
+ return get_configured_bundle_team_names()
+ except Exception:
+ # A health probe must not fail on malformed bundle config; callers
fall back to liveness only.
+ log.warning("Could not read the Dag bundle configuration",
exc_info=True)
+ return {}
+
+
+def _liveness_status(jobs: list[Job]) -> str:
+ """Status for a component with no declared work partition: one live
replica covers everything."""
+ return HEALTHY if any(job.is_alive() for job in jobs) else DOWN
+
+
+def _coverage_status(expected: set[Any], covered: set[Any]) -> str:
+ """Status from how much of a component's declared work partition its live
replicas cover."""
+ if not expected - covered:
return HEALTHY
- return DEGRADED
+ if expected & covered:
+ return DEGRADED
+ return DOWN
+
+
+def _dag_processor_detailed_status(jobs: list[Job]) -> str:
+ expected = set(_configured_bundle_teams())
Review Comment:
I dont think we should. Having multiple dag processor can happen in a non
multi team environment. I agree, it is likely the environment is running in
multi-team mode but not 100%. You might want to have 2 dag processors, one per
dag bundle, and have multi-team feature flag turned off
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
