pierrejeambrun commented on code in PR #73222:
URL: https://github.com/apache/airflow/pull/73222#discussion_r4038262129
##########
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:
Should we gate with the same `if not conf.getboolean("core",
"multi_team"):`?
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py:
##########
Review Comment:
Unrelated but this shouldn't be a 'str'. An enum would give more
information. Same for the legagcy 'status'.
##########
airflow-core/docs/administration-and-deployment/logging-monitoring/check-health.rst:
##########
@@ -100,11 +91,28 @@ including per-instance details when multiple schedulers,
triggerers, or Dag proc
* ``status`` (legacy aggregate): ``"healthy"`` if **any** running instance
is alive, otherwise ``"unhealthy"``
(including when no running jobs exist for that component).
- * ``detailed_status``: reflects the full set of running instances:
-
- * ``"healthy"`` — every running instance is alive
- * ``"degraded"`` — some instances are alive and some are not
- * ``"down"`` — no running instance is alive (including when no jobs exist)
+ * ``detailed_status``: whether every part of that component's work is being
covered by a live instance.
+ What counts as "every part" differs per component, because only some of
them divide their work up:
+
+ * **Dag processor** — the parts are the Dag bundles in ``[dag_processor]
dag_bundle_config_list``.
+ A processor started without ``--bundle-name`` covers every configured
bundle;
+ one started with it covers only the bundles it was given.
+ ``"healthy"`` when every configured bundle has a live processor,
``"degraded"`` when only some do,
+ ``"down"`` when none do.
+ * **Triggerer** — with ``[core] multi_team`` enabled, the parts are the
teams those bundles are scoped to
+ (plus the unscoped bundles), because a triggerer only picks up triggers
for its own team.
+ ``"healthy"`` when every team scope has a live triggerer, ``"degraded"``
when only some do,
+ ``"down"`` when none do. With multi-team disabled, no team filtering
applies, so any live triggerer
+ covers everything: ``"healthy"`` if one is alive, ``"down"`` if none is.
+ * **Scheduler** — schedulers are symmetric and share no partitioned work,
so there is nothing partial
+ to report: ``"healthy"`` if at least one is alive, ``"down"`` if none
is. ``"degraded"`` is never
+ returned for the scheduler. Use ``instances`` to see how many replicas
are up, and your orchestrator
+ or the ``scheduler_heartbeat`` metric to alert on reduced scheduling
throughput.
Review Comment:
Here we can't make the difference when we have 1 scheduler job up, and 1
scheduler job down:
- I expect 1 scheduler only, it's caused by a restart -> healthy
- I expect 2 scheduler up, 1 is down -> degraded
So this is why we count `at leaset one is alive` as healthy, and we can't
detect degrated ?
##########
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()
Review Comment:
`_configured_bundle_teams` private calling public
`get_configured_bundle_team_names` which is only used here.
Maybe mark `get_configured_bundle_team_names` private too.
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -130,6 +130,32 @@ def _parse_bundle_config(config_list) ->
list[_ExternalBundleConfig]:
return list(bundles.values())
+def _read_bundle_config_list() -> list[_ExternalBundleConfig]:
+ config_list = conf.getjson("dag_processor", "dag_bundle_config_list")
+ if not config_list:
+ return []
+ if not isinstance(config_list, list):
+ raise AirflowConfigException(
+ "Section `dag_processor` key `dag_bundle_config_list` "
+ f"must be list but got {config_list.__class__}"
+ )
+ return _parse_bundle_config(config_list)
+
+
+def get_configured_bundle_team_names() -> dict[str, str | None]:
+ """
+ Get the team owning each explicitly configured Dag bundle.
+
+ This reads the config rather than going through ``DagBundlesManager`` so
that callers who only
+ need the declared bundle partition neither import every bundle class nor
see the example-Dag
+ bundles that ``DagBundlesManager.parse_config`` injects when ``[core]
load_examples`` is set --
+ those are added by Airflow, not declared by the deployment.
+
+ :return: mapping of bundle name to team name, ``None`` for bundles that
are not team scoped.
+ """
+ return {cfg.name: cfg.team_name for cfg in _read_bundle_config_list()}
Review Comment:
Should we skip when `team_name` is None? Basically this can give fully
`bundle: None` mapping, so basically 0 team if the multi team isn't enable.
A short circuit somewhere here or checking the config value could help make
things consistent and not do work when there is no team.
--
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]
