Source code for multistorageclient.telemetry.metrics.readers.diperiodic_exporting
1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2# SPDX-License-Identifier: Apache-2.0
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import logging
17import math
18import os
19import threading
20import time
21import weakref
22
23import opentelemetry.sdk.environment_variables as sdk_environment_variables
24import opentelemetry.sdk.metrics as sdk_metrics
25import opentelemetry.sdk.metrics.export as sdk_metrics_export
26
27# Not OTel spec. Use 1 second to keep the data volume per export interval reasonably small.
28DEFAULT_COLLECT_INTERVAL_MILLIS: float = 1000
29# Not OTel spec. Use the default on :py:meth:`sdk_metrics_export.MetricReader.collect`.
30DEFAULT_COLLECT_TIMEOUT_MILLIS: float = 10000
31# OTel spec.
32DEFAULT_EXPORT_INTERVAL_MILLIS: float = 60000
33# OTel spec.
34DEFAULT_EXPORT_TIMEOUT_MILLIS: float = 30000
35
36logger = logging.getLogger(__name__)
37
38
[docs]
39class DiperiodicExportingMetricReader(sdk_metrics_export.MetricReader):
40 """
41 :py:class:`opentelemetry.sdk.metrics.export.MetricReader` that collects + exports metrics on separate user-configurable time intervals.
42 This is in contrast with :py:class:`opentelemetry.sdk.metrics.export.PeriodicExportingMetricReader` which couples them with a 1 minute default.
43
44 The metrics collection interval limits the temporal resolution. Most metric backends have 1 millisecond or finer temporal resolution.
45 """
46
47 #: Collect buffer.
48 _collect_metrics_data: sdk_metrics_export.MetricsData | None
49 _collect_metrics_data_lock: threading.Lock
50 #: Export buffer.
51 _export_metrics_data: sdk_metrics_export.MetricsData | None
52 _export_metrics_data_lock: threading.Lock
53
54 _exporter: sdk_metrics_export.MetricExporter
55 _collect_interval_millis: float
56 _collect_timeout_millis: float
57 _export_interval_millis: float
58 _export_timeout_millis: float
59
60 _shutdown_event: threading.Event
61 _shutdown_event_lock: threading.Lock
62 _collect_daemon: threading.Thread | None
63 _export_daemon: threading.Thread | None
64
65 def __init__(
66 self,
67 exporter: sdk_metrics_export.MetricExporter,
68 collect_interval_millis: float | None = None,
69 collect_timeout_millis: float | None = None,
70 export_interval_millis: float | None = None,
71 export_timeout_millis: float | None = None,
72 ):
73 """
74 :param exporter: Metrics exporter.
75 :param collect_interval_millis: Collect interval in milliseconds.
76 :param collect_timeout_millis: Collect timeout in milliseconds.
77 :param export_interval_millis: Export interval in milliseconds.
78 :param export_timeout_millis: Export timeout in milliseconds.
79 """
80
81 # Defer to the exporter for aggregation and temporality configurations.
82 super().__init__(
83 preferred_aggregation=exporter._preferred_aggregation, preferred_temporality=exporter._preferred_temporality
84 )
85
86 self._collect_metrics_data = None
87 self._collect_metrics_data_lock = threading.Lock()
88 self._export_metrics_data = None
89 self._export_metrics_data_lock = threading.Lock()
90
91 self._exporter = exporter
92 if collect_interval_millis is None:
93 # OTEL_METRIC_COLLECT_INTERVAL isn't an official OTel SDK environment variable (yet).
94 collect_interval_millis = DEFAULT_COLLECT_INTERVAL_MILLIS
95 if collect_timeout_millis is None:
96 # OTEL_METRIC_COLLECT_TIMEOUT isn't an official OTel SDK environment variable (yet).
97 collect_timeout_millis = DEFAULT_COLLECT_TIMEOUT_MILLIS
98 if export_interval_millis is None:
99 try:
100 export_interval_millis = float(
101 os.environ.get(
102 sdk_environment_variables.OTEL_METRIC_EXPORT_INTERVAL, DEFAULT_EXPORT_INTERVAL_MILLIS
103 )
104 )
105 except ValueError:
106 logger.warning(
107 f"Found invalid value for export interval. Using default of {DEFAULT_EXPORT_INTERVAL_MILLIS}."
108 )
109 export_interval_millis = DEFAULT_EXPORT_INTERVAL_MILLIS
110 if export_timeout_millis is None:
111 try:
112 export_timeout_millis = float(
113 os.environ.get(sdk_environment_variables.OTEL_METRIC_EXPORT_TIMEOUT, DEFAULT_EXPORT_TIMEOUT_MILLIS)
114 )
115 except ValueError:
116 logger.warning(
117 f"Found invalid value for export timeout. Using default of {DEFAULT_EXPORT_TIMEOUT_MILLIS}."
118 )
119 export_timeout_millis = DEFAULT_EXPORT_TIMEOUT_MILLIS
120 self._collect_interval_millis = collect_interval_millis
121 self._collect_timeout_millis = collect_timeout_millis
122 self._export_interval_millis = export_interval_millis
123 self._export_timeout_millis = export_timeout_millis
124
125 self._shutdown_event = threading.Event()
126 self._shutdown_event_lock = threading.Lock()
127 self._collect_daemon = None
128 self._export_daemon = None
129 if (
130 self._collect_interval_millis > 0
131 and self._collect_interval_millis < math.inf
132 and self._export_interval_millis > 0
133 and self._export_interval_millis < math.inf
134 ):
135 self._init_daemons()
136 if hasattr(os, "register_at_fork"):
137 os.register_at_fork(after_in_child=weakref.WeakMethod(self._init_daemons)())
138 else:
139 raise ValueError("Collect and export intervals must be in (0, infinity).")
140
141 def _init_daemons(self) -> None:
142 # Empty the buffers. Prevents duplicate metrics when forking.
143 with self._collect_metrics_data_lock, self._export_metrics_data_lock:
144 self._collect_metrics_data, self._export_metrics_data = None, None
145
146 # Create the collect daemon.
147 self._collect_daemon = threading.Thread(
148 name="OtelDiperiodicExportingMetricReader._collect_daemon", target=self._collect_daemon_target, daemon=True
149 )
150 self._collect_daemon.start()
151
152 # Create the export daemon.
153 self._export_daemon = threading.Thread(
154 name="OtelDiperiodicExportingMetricReader._export_daemon", target=self._export_daemon_target, daemon=True
155 )
156 self._export_daemon.start()
157
158 def _collect_daemon_target(self) -> None:
159 while not self._shutdown_event.wait(timeout=self._collect_interval_millis / 10**3):
160 self._collect_iteration()
161
162 def _export_daemon_target(self) -> None:
163 while not self._shutdown_event.wait(timeout=self._export_interval_millis / 10**3):
164 self._export_iteration()
165 # Final collect + export.
166 self._collect_iteration()
167 self._export_iteration()
168
169 # :py:meth:`sdk_metrics_export.MetricReader._collect` is reserved. Using another name.
170 def _collect_iteration(self, timeout_millis: float | None = None) -> None:
171 try:
172 # Only set when registered on a :py:class:`sdk_metrics.MeterProvider` which calls
173 # :py:meth:`sdk_metrics_export.MetricReader._set_collect_callback`.
174 if self._collect is not None:
175 # Inherited from :py:class:``sdk_metrics_export.MetricReader``.
176 self.collect(timeout_millis=timeout_millis or self._collect_timeout_millis)
177 except sdk_metrics.MetricsTimeoutError:
178 logger.warning("Metrics collection timed out.", exc_info=True)
179 except Exception:
180 logger.exception("Exception while collecting metrics.")
181
182 # Called by :py:meth:`sdk_metrics_export.MetricReader.collect`.
183 def _receive_metrics(
184 self, metrics_data: sdk_metrics_export.MetricsData, timeout_millis: float = 0, **kwargs
185 ) -> None:
186 with self._collect_metrics_data_lock:
187 self._collect_metrics_data = sdk_metrics_export.MetricsData(
188 resource_metrics=(
189 *(() if self._collect_metrics_data is None else self._collect_metrics_data.resource_metrics),
190 *metrics_data.resource_metrics,
191 )
192 )
193
194 def _export_iteration(self, timeout_millis: float | None = None) -> None:
195 with self._export_metrics_data_lock:
196 with self._collect_metrics_data_lock:
197 # Rotate the collect + export buffers.
198 #
199 # We don't merge the collect buffer into the export buffer to prevent infinite accumulation.
200 self._collect_metrics_data, self._export_metrics_data = None, self._collect_metrics_data
201
202 if self._export_metrics_data is not None:
203 try:
204 # Export.
205 self._exporter.export(
206 metrics_data=self._export_metrics_data,
207 timeout_millis=timeout_millis or self._export_timeout_millis,
208 )
209 except sdk_metrics.MetricsTimeoutError:
210 logger.warning(
211 f"Metrics export timed out. {sum(len(rm.scope_metrics) for rm in self._export_metrics_data.resource_metrics)} data points lost.",
212 exc_info=True,
213 )
214 except Exception:
215 logger.exception(
216 f"Exception while exporting metrics. {sum(len(rm.scope_metrics) for rm in self._export_metrics_data.resource_metrics)} data points lost."
217 )
218 finally:
219 # Immediately empty the export buffer for garbage collection.
220 self._export_metrics_data = None
221
[docs]
222 def force_flush(
223 self, timeout_millis: float = DEFAULT_COLLECT_TIMEOUT_MILLIS + DEFAULT_EXPORT_TIMEOUT_MILLIS
224 ) -> bool:
225 deadline_ns = time.time_ns() + (timeout_millis * 10**6)
226
227 # Calls :py:meth:`sdk_metrics_export.MetricReader.collect`.
228 super().force_flush(timeout_millis=(deadline_ns - time.time_ns()) / 10**6)
229 self._export_iteration(timeout_millis=(deadline_ns - time.time_ns()) / 10**6)
230 self._exporter.force_flush(timeout_millis=(deadline_ns - time.time_ns()) / 10**6)
231 return True
232
[docs]
233 def shutdown(
234 self, timeout_millis: float = DEFAULT_COLLECT_TIMEOUT_MILLIS + DEFAULT_EXPORT_TIMEOUT_MILLIS, **kwargs
235 ) -> None:
236 deadline_ns = time.time_ns() + (timeout_millis * 10**6)
237
238 with self._shutdown_event_lock:
239 if not self._shutdown_event.is_set():
240 # Signal the collect + export daemons to stop first. Their loops only
241 # exit once this event is set, and the export daemon performs a final
242 # collect + export on the way out. Joining before setting the event
243 # would block for the full timeout and then run that final export
244 # against an already-shut-down exporter.
245 self._shutdown_event.set()
246 if self._collect_daemon is not None:
247 self._collect_daemon.join(timeout=(deadline_ns - time.time_ns()) / 10**9)
248 if self._export_daemon is not None:
249 self._export_daemon.join(timeout=(deadline_ns - time.time_ns()) / 10**9)
250 self._exporter.shutdown(timeout_millis=(deadline_ns - time.time_ns()) / 10**6)