xBis7 commented on code in PR #63932:
URL: https://github.com/apache/airflow/pull/63932#discussion_r3139637275
##########
shared/observability/src/airflow_shared/observability/metrics/stats.py:
##########
@@ -62,72 +72,239 @@ def normalize_name_for_stats(name: str, log_warning: bool
= True) -> str:
return normalized
-class _Stats(type):
- factory: Callable[[], StatsLogger | NoStatsLogger] | None = None
- instance: StatsLogger | NoStatsLogger | None = None
- _instance_pid: int | None = None
-
- def __getattr__(cls, name: str) -> str:
- factory = type.__getattribute__(cls, "factory")
- instance = type.__getattribute__(cls, "instance")
-
- # When using OpenTelemetry, some subprocesses are short-lived and
- # often exit before flushing any metrics.
- #
- # The solution is to register a hook that performs a force flush at
exit.
- # The atexit hook is registered when initializing the instance.
- #
- # The instance gets initialized once per process. In case a process is
forked, then
- # the new subprocess, will inherit the already initialized instance of
the parent process.
- #
- # Store the instance pid so that it can be compared with the current
pid
- # to decide whether to initialize the instance again or not.
- #
- # So far, all forks are resetting their state to remove anything
inherited by the parent.
- # But in the future that might not always be true.
- current_pid = os.getpid()
- if cls.instance and cls._instance_pid != current_pid:
- log.info(
- "Stats instance was created in PID %s but accessed in PID %s.
Re-initializing.",
- cls._instance_pid,
- current_pid,
+def initialize(
+ *,
+ factory: Callable[[], StatsLogger | NoStatsLogger],
+ export_legacy_names: bool,
+) -> None:
+ """Initialize the stats module with a backend factory and legacy name
configuration."""
+ global _factory, _backend, _instance_pid, _export_legacy_names
+ _factory = factory
+ _backend = None
+ _instance_pid = None
+ _export_legacy_names = export_legacy_names
+
+
+def _get_backend() -> StatsLogger | NoStatsLogger:
+ """Return the current backend, re-initializing if the process has been
forked."""
+ global _backend, _instance_pid
+ current_pid = os.getpid()
+
+ if _backend is not None and _instance_pid != current_pid:
+ log.info(
+ "Stats backend was created in PID %s but accessed in PID %s.
Re-initializing.",
+ _instance_pid,
+ current_pid,
+ )
+ _backend = None
+ _instance_pid = None
+
+ if _backend is None:
+ factory = _factory if _factory is not None else NoStatsLogger
+ try:
+ _backend = factory()
+ _instance_pid = current_pid
+ except (socket.gaierror, ImportError) as e:
+ log.error("Could not configure StatsClient: %s, using
NoStatsLogger instead.", e)
+ _backend = NoStatsLogger()
+ _instance_pid = current_pid
+
+ return _backend
Review Comment:
Good point! I made the changes.
--
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]
