Add team_name tag to OpenLineage metrics#69109
Conversation
SameerMesiah97
force-pushed
the
68996-OpenLineage-Team-Name-Metrics
branch
4 times, most recently
from
588db94 to
e3f59ba
Compare
| "deadlines": lambda dagrun: DagRunInfo.deadlines(dagrun), | ||
| } | ||
|
|
||
| _team_name_cache: ClassVar[dict[str, str | None]] = {} |
There was a problem hiding this comment.
Is it fine that there is no TTL on this cached value? If a dag changes the bundle it wont be reflected as this cache never expires. The Core already caches this lookup here: DagModel.get_team_name(dag_id) uses a TTLCache.
This PR calls the uncached DagBundleModel.get_team_name(bundle_name) instead, so it had to add its own cache. I am not sure if they are interchangeable but something to think about.
There was a problem hiding this comment.
I think using a regular cache is acceptable for the following reasons:
- Bundles per team are expected to change very rarely and most certainly not during the DAG run.
- This is provider-layer code that runs in the worker in a short-lived process bounded by the lifetime of the task, so the cache will likely not remain stale for very long.
- Is it expected for OpenLineage events to pick up bundle-to-team relationships that change mid-flight? I don't believe so.
Also, if we were to switch to DagModel.get_team_name(dag_id), there would be a slight performance hit as we'd have to go through the DagModel table to get the DAG bundle, whereas with DagBundleModel.get_team_name(bundle_name), we can use the bundle name directly to fetch the team name.
But if you insist on a TTL cache, I think the best option would be to use DagModel.get_team_name(dag_id). However, I am not sure if it worth it.
There was a problem hiding this comment.
Good point about short lived processes as this code will run in the provider layer. I think we are good with keeping the cache.
There was a problem hiding this comment.
Do we really need the cache in that case honestly? If we're counting on this only executing in a temporary fork where only one or two calls will be cached, is it providing that much value? Or just more surface for something to go wrong in other code paths/code reuse? I'd vote to simplify where we can.
potiuk
added
the
ready for maintainer review
|
This is just a reminder for review. |
SameerMesiah97
force-pushed
the
68996-OpenLineage-Team-Name-Metrics
branch
from
e3f59ba to
240ab79
Compare
when multi-team is enabled on Airflow 3.1+. The team name is propagated through the Airflow run facet and included in ol.emit.attempts, ol.emit.failed, ol.extract, and ol.event.size. Tests are updated accordingly.
…nd reuse the existing team_name value when emitting metrics.
…ssociated unit tests by removing unnecessary database markers and simplifying the test cases.
SameerMesiah97
force-pushed
the
68996-OpenLineage-Team-Name-Metrics
branch
from
240ab79 to
3de74a8
Compare
kacpermuda
left a comment
kacpermuda
left a comment
There was a problem hiding this comment.
One small but important change needed.
| team_name = None | ||
|
|
||
| facets = event.run.facets or {} | ||
| airflow_facet = cast("AirflowRunFacet | None", facets.get("airflow")) |
There was a problem hiding this comment.
We need to fallback to "airflowDagRun" facet here, for DAG events. "airflow" is the facet name only used in task events. Let's also add a test for that case.
6 participants
Description
This change adds a
team_nametag to OpenLineage metrics to improve observability in multi-team deployments.The following metrics now include the
team_nametag when available:ol.emit.attemptsol.emit.failedol.extractol.event.sizeA new
DagRunInfo.team_name()helper has been added to resolve the team associated with aDagRunfrom its DAG bundle.For listener metrics (
ol.extractandol.event.size), theteam_nameis resolved directly from theDagRun. For adapter metrics (ol.emit.attemptsandol.emit.failed), the value is propagated through the Airflow run facet attached to the emitted OpenLineage event.Rationale
The OpenLineage integration emits metrics for lineage extraction and event publication, but these metrics could not previously be attributed to individual teams in multi-team deployments.
Adding the
team_nametag enables per-team dashboards, alerting, and troubleshooting while remaining backwards compatible for deployments that do not use multi-team scheduling.Tests
Added unit tests for the new
DagRunInfo.team_name()helper, including multi-team configuration handling.Updated the existing OpenLineage listener and adapter unit tests to verify metric emission both with and without the
team_nametag. Test cases covering the presence ofteam_nameare skipped when running against Airflow versions earlier than 3.3.0.Notes
A typographical error ('deg') in the docstring for
dag_version_infohas been corrected.Backwards Compatibility
This change is additive only. Existing metric names remain unchanged, and the
team_nametag is emitted only when running on Airflow 3.3+ with multi-team enabled. Existing dashboards and integrations that do not use the new tag continue to function unchanged.Related: #68996
Was generative AI tooling used to co-author this PR?
Generated-by: [GPT 5.5] following the guidelines