Work in progress

Author: Jayclifford345

docs/examples/metrics/reader/README.rst

@@ -6,6 +6,7 @@ These examples show how to customize the metrics that are output by the SDK usin
 * preferred_aggregation.py: Shows how to configure the preferred aggregation for metric instrument types.
 * preferred_temporality.py: Shows how to configure the preferred temporality for metric instrument types.
 * preferred_exemplarfilter.py: Shows how to configure the exemplar filter.
+* sychronous_reader.py: Shows how to configure the reader to collect and export metrics synchronously.
 
 The source files of these examples are available :scm_web:`here <docs/examples/metrics/reader/>`.
 

docs/examples/metrics/reader/synchronous_read.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import random
+import time
+
+from opentelemetry import metrics
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import (
+ ConsoleMetricExporter,
+ SynchronousExportingMetricReader,
+)
+from opentelemetry.sdk.resources import Resource
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+# Create a resource that identifies our service
+resource = Resource.create({"service.name": "batch-metrics-auto-export-demo"})
+# Create a console exporter for visible output
+exporter = ConsoleMetricExporter()
+# Create a batch exporting reader with a small batch size for demonstration
+reader = SynchronousExportingMetricReader(
+ exporter,
+ max_export_batch_size=5, # Export after every 5 metrics
+ max_queue_size=1000, # Queue up to 100 metrics
+)
+# Create a meter provider with our reader
+meter_provider = MeterProvider(resource=resource, metric_readers=[reader])
+# Set the global meter provider
+metrics.set_meter_provider(meter_provider)
+# Get a meter
+meter = metrics.get_meter("batch-demo-auto-export")
+# Create instruments
+request_counter = meter.create_counter(
+ name="request_count",
+ description="Number of requests processed",
+ unit="1",
+)
+active_requests_gauge = meter.create_gauge(
+ name="active_requests",
+ description="Number of active requests",
+ unit="1",
+)
+processing_time_histogram = meter.create_histogram(
+ name="request_processing_time",
+ description="Time taken to process requests",
+ unit="ms",
+)
+
+
+def simulate_request():
+ """Simulate a request and record metrics."""
+ # Simulate active requests fluctuating
+ active_requests = random.randint(1, 20)
+ active_requests_gauge.set(active_requests)
+ # Simulate processing time
+ processing_time = random.uniform(10, 500) # Between 10-500ms
+ processing_time_histogram.record(processing_time)
+ # Increment the request counter
+ request_counter.add(1)
+ logger.info("Processed a request")
+ reader.collect()
+ # Each request generates 3 metrics (counter, gauge, histogram)
+ # With batch size of 5, we should see exports after every ~2 requests
+
+
+def main():
+ """Main application function to demonstrate reading and exporting metrics synchronously."""
+ logger.info("Starting to process requests...")
+ logger.info(
+ "With batch size of 5, exports should happen automatically after every ~2 requests"
+ )
+ logger.info("Watch for the metric exports in the console...")
+ # Process 25 requests with pauses to observe collection and export mechanism
+ for i in range(1, 25):
+ simulate_request()
+ # Add a pause between requests to make it easier to observe the exports
+ time.sleep(0.5)
+ logger.info("\nAll requests completed.")
+ # We still need to shut down cleanly
+ logger.info("Shutting down...")
+ meter_provider.shutdown()
+ logger.info(
+ "Done. Any metrics left in the queue were exported during shutdown."
+ )
+
+
+if __name__ == "__main__":
+ main()

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 from __future__ import annotations
 
+import collections
 import math
 import os
 import weakref
@@ -21,7 +22,7 @@
 from logging import getLogger
 from os import environ, linesep
 from sys import stdout
-from threading import Event, Lock, RLock, Thread
+from threading import Condition, Event, Lock, RLock, Thread
 from time import time_ns
 from typing import IO, Callable, Iterable, Optional
 
@@ -574,3 +575,198 @@ def force_flush(self, timeout_millis: float = 10_000) -> bool:
  super().force_flush(timeout_millis=timeout_millis)
  self._exporter.force_flush(timeout_millis=timeout_millis)
  return True
+
+
+class SynchronousExportingMetricReader(MetricReader):
+ """Implementation of `MetricReader` that exports metrics in batches.
+
+ Metrics are collected when `collect()` is called, then added to a queue until
+ the configured batch size is reached or until `force_flush` is called. The
+ metrics are then exported in batches to the configured exporter.
+
+ Unlike the `PeriodicExportingMetricReader`, this reader doesn't automatically
+ collect metrics on a schedule - collection must be triggered by calling
+ `collect()` or systems built on top of this reader.
+
+ `SynchronousExportingMetricReader` is configurable with the following parameters:
+
+ - `max_queue_size`: Maximum number of metric batches to store in memory
+ - `max_export_batch_size`: Maximum number of metric batches to export at once
+ - `export_timeout_millis`: Timeout for export operations in milliseconds
+ """
+
+ def __init__(
+ self,
+ exporter: MetricExporter,
+ max_export_batch_size: int = 512,
+ export_timeout_millis: float = 30000,
+ max_queue_size: int = 2048,
+ ) -> None:
+ # BatchExportingMetricReader defers to exporter for configuration
+ super().__init__(
+ preferred_temporality=exporter._preferred_temporality,
+ preferred_aggregation=exporter._preferred_aggregation,
+ )
+
+ self._validate_arguments(max_queue_size, max_export_batch_size)
+
+ self._exporter = exporter
+ self._max_queue_size = max_queue_size
+ self._max_export_batch_size = max_export_batch_size
+ self._export_timeout_millis = export_timeout_millis
+
+ # This lock is held whenever calling self._exporter.export() to prevent concurrent
+ # execution of MetricExporter.export()
+ self._export_lock = Lock()
+
+ # Queue to store metrics
+ self._queue = collections.deque([], max_queue_size)
+
+ # Thread handling and synchronization
+ self._condition = Condition(Lock())
+ self._shutdown = False
+ self._flush_event = Event()
+ self._shutdown_once = Once()
+
+ # Process fork handling
+ if hasattr(os, "register_at_fork"):
+ weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+ os.register_at_fork(
+ after_in_child=lambda: weak_reinit()() # pylint: disable=unnecessary-lambda
+ )
+ self._pid = os.getpid()
+
+ def _at_fork_reinit(self):
+ """Reinitialize the reader after fork."""
+ self._condition = Condition(Lock())
+ self._queue.clear()
+ self._pid = os.getpid()
+
+ def _export_batch(self) -> int:
+ """Exports at most max_export_batch_size metrics and returns the number of exported metrics."""
+ idx = 0
+ pending_metrics = []
+
+ with self._condition:
+ while idx < self._max_export_batch_size and self._queue:
+ metrics_data = self._queue.pop()
+ pending_metrics.append(metrics_data)
+ idx += 1
+
+ if pending_metrics:
+ token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
+ try:
+ with self._export_lock:
+ for metrics_data in pending_metrics:
+ self._exporter.export(
+ metrics_data,
+ timeout_millis=self._export_timeout_millis,
+ )
+ except Exception: # pylint: disable=broad-exception-caught
+ _logger.exception("Exception while exporting metrics.")
+ finally:
+ detach(token)
+
+ return idx
+
+ def _drain_queue(self):
+ """Export all elements until queue is empty."""
+ while self._queue:
+ self._export_batch()
+
+ def _receive_metrics(
+ self,
+ metrics_data: MetricsData,
+ timeout_millis: float = 10_000,
+ **kwargs,
+ ) -> None:
+ """Add metrics to the queue for batched export."""
+ if self._shutdown:
+ return
+
+ # Handle fork
+ if self._pid != os.getpid():
+ self._at_fork_reinit()
+
+ # Add metrics to queue
+ with self._condition:
+ self._queue.appendleft(metrics_data)
+ if len(self._queue) >= self._max_export_batch_size:
+ self._condition.notify()
+ self._flush_event.set()
+
+ # If queue has reached batch size, export immediately
+ if len(self._queue) >= self._max_export_batch_size:
+ self._export_batch()
+
+ def force_flush(self, timeout_millis: float = 10_000) -> bool:
+ """Forces flush of metrics to the exporter
+
+ Args:
+ timeout_millis: The maximum amount of time to wait for the flush
+ to complete, in milliseconds.
+
+ Returns:
+ True if the flush was successful, False otherwise.
+ """
+ if timeout_millis is None:
+ timeout_millis = self._export_timeout_millis
+
+ if self._shutdown:
+ return True
+
+ # Collect any pending metrics first (this will trigger _receive_metrics)
+ super().force_flush(timeout_millis=timeout_millis)
+
+ # Export all batches in queue
+ try:
+ with self._condition:
+ if not self._queue:
+ return True
+
+ self._drain_queue()
+ return self._exporter.force_flush(timeout_millis=timeout_millis)
+ except Exception: # pylint: disable=broad-exception-caught
+ _logger.exception("Exception during force_flush")
+ return False
+
+ def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
+ """Shuts down the metric reader and exporter.
+
+ Args:
+ timeout_millis: The maximum amount of time to wait for the exporter
+ to shutdown, in milliseconds.
+ """
+ deadline_ns = time_ns() + timeout_millis * 10**6
+
+ def _shutdown():
+ self._shutdown = True
+
+ did_set = self._shutdown_once.do_once(_shutdown)
+ if not did_set:
+ _logger.warning("Can't shutdown multiple times")
+ return
+
+ # Export any metrics still in the queue
+ self._drain_queue()
+
+ # Shutdown the exporter
+ self._exporter.shutdown(
+ timeout=(deadline_ns - time_ns()) / 10**6, **kwargs
+ )
+
+ @staticmethod
+ def _validate_arguments(max_queue_size, max_export_batch_size):
+ """Validate constructor arguments."""
+ if max_queue_size <= 0:
+ raise ValueError("max_queue_size must be a positive integer.")
+
+ if max_export_batch_size <= 0:
+ raise ValueError(
+ "max_export_batch_size must be a positive integer."
+ )
+
+ if max_export_batch_size > max_queue_size:
+ raise ValueError(
+ "max_export_batch_size must be less than or equal to max_queue_size."
+ )

opentelemetry-sdk/src/opentelemetry/sdk/metrics/export/__init__.py

@@ -21,6 +21,7 @@
  MetricExportResult,
  MetricReader,
  PeriodicExportingMetricReader,
+ SynchronousExportingMetricReader,
 )
 
 # The point module is not in the export directory to avoid a circular import.
@@ -63,4 +64,5 @@
  "ResourceMetrics",
  "ScopeMetrics",
  "Sum",
+ "SynchronousExportingMetricReader",
 ]