Merge branch 'use-generic-metrics-reporter' into 'main'

Metrics: Register metrics dynamically and add metric types

See merge request ecdc/ess-dmsc/forwarder!293

Author: danesss

changes.md

@@ -4,6 +4,8 @@
 
 * Remove support for deprecated ep00 schema
 * Adopt pip-tools to manage requirements.txt files
+* Refactor statistics reporter to support dynamically added metrics
+* Add latency and per-PV graphite metrics
 
 ## v2.1.0
 

forwarder/handle_config_change.py

@@ -8,6 +8,7 @@
 from forwarder.common import Channel, CommandType, ConfigUpdate
 from forwarder.configuration_store import ConfigurationStore, NullConfigurationStore
 from forwarder.kafka.kafka_producer import KafkaProducer
+from forwarder.metrics import Gauge
 from forwarder.status_reporter import StatusReporter
 from forwarder.update_handlers.create_update_handler import (
     UpdateHandler,
@@ -24,6 +25,7 @@ def _subscribe_to_pv(
     logger: Logger,
     fake_pv_period: int,
     pv_update_period: Optional[int],
+    pvs_subscribed_metric: Optional[Gauge] = None,
 ):
     if new_channel in update_handlers.keys():
         logger.warning(
@@ -46,12 +48,15 @@ def _subscribe_to_pv(
     logger.info(
         f"Subscribed to PV name='{new_channel.name}', schema='{new_channel.schema}', topic='{new_channel.output_topic}'"
     )
+    if pvs_subscribed_metric:
+        pvs_subscribed_metric.inc()
 
 
 def _unsubscribe_from_pv(
     remove_channel: Channel,
     update_handlers: Dict[Channel, UpdateHandler],
     logger: Logger,
+    pvs_subscribed_metric: Optional[Gauge] = None,
 ):
     def _match_channel_field(
         field_in_remove_request: Optional[str], field_in_existing_channel: Optional[str]
@@ -88,17 +93,23 @@ def _wildcard_match_channel_field(
     for channel in channels_to_remove:
         update_handlers[channel].stop()
         del update_handlers[channel]
+        if pvs_subscribed_metric:
+            pvs_subscribed_metric.dec()
 
     logger.info(
         f"Unsubscribed from PVs matching name='{remove_channel.name}', schema='{remove_channel.schema}', topic='{remove_channel.output_topic}'"
     )
 
 
 def _unsubscribe_from_all(
-    update_handlers: Dict[Channel, UpdateHandler], logger: Logger
+    update_handlers: Dict[Channel, UpdateHandler],
+    logger: Logger,
+    pvs_subscribed_metric: Optional[Gauge] = None,
 ):
     for update_handler in update_handlers.values():
         update_handler.stop()
+        if pvs_subscribed_metric:
+            pvs_subscribed_metric.dec()
     update_handlers.clear()
     logger.info("Unsubscribed from all PVs")
 
@@ -114,12 +125,15 @@ def handle_configuration_change(
     logger: Logger,
     status_reporter: StatusReporter,
     configuration_store: ConfigurationStore = NullConfigurationStore,
+    pvs_subscribed_metric: Optional[Gauge] = None,
 ):
     """
     Add or remove update handlers according to the requested change in configuration
     """
     if configuration_change.command_type == CommandType.REMOVE_ALL:
-        _unsubscribe_from_all(update_handlers, logger)
+        _unsubscribe_from_all(
+            update_handlers, logger, pvs_subscribed_metric=pvs_subscribed_metric
+        )
     elif configuration_change.command_type == CommandType.INVALID:
         return
     else:
@@ -135,8 +149,14 @@ def handle_configuration_change(
                         logger,
                         fake_pv_period,
                         pv_update_period,
+                        pvs_subscribed_metric=pvs_subscribed_metric,
                     )
                 elif configuration_change.command_type == CommandType.REMOVE:
-                    _unsubscribe_from_pv(channel, update_handlers, logger)
+                    _unsubscribe_from_pv(
+                        channel,
+                        update_handlers,
+                        logger,
+                        pvs_subscribed_metric=pvs_subscribed_metric,
+                    )
     status_reporter.report_status()
     configuration_store.save_configuration(update_handlers)

forwarder/kafka/kafka_helpers.py

@@ -4,7 +4,8 @@
 from confluent_kafka import Consumer, Producer
 from streaming_data_types.epics_connection_ep01 import ConnectionInfo, serialise_ep01
 
-from forwarder.utils import Counter
+from forwarder.metrics import Counter
+from forwarder.metrics.statistics_reporter import StatisticsReporter
 
 from .kafka_producer import KafkaProducer
 
@@ -59,9 +60,7 @@ def create_producer(
     username: Optional[str] = None,
     password: Optional[str] = None,
     ssl_ca_file: Optional[str] = None,
-    counter: Optional[Counter] = None,
-    buffer_err_counter: Optional[Counter] = None,
-    delivery_err_counter: Optional[Counter] = None,
+    statistics_reporter: Optional[StatisticsReporter] = None,
 ) -> KafkaProducer:
     producer_config = {
         "bootstrap.servers": broker_address,
@@ -74,12 +73,31 @@ def create_producer(
         if ssl_ca_file:
             producer_config["ssl.ca.location"] = ssl_ca_file
     producer = Producer(producer_config)
-    return KafkaProducer(
-        producer,
-        update_msg_counter=counter,
-        update_buffer_err_counter=buffer_err_counter,
-        update_delivery_err_counter=delivery_err_counter,
-    )
+    if not statistics_reporter:
+        return KafkaProducer(producer)
+    else:
+        update_message_counter = Counter(
+            "successful_sends_total", "Total number of updates sent to kafka"
+        )
+        buffer_err_counter = Counter(
+            "send_buffer_errors_total", "Kafka producer queue errors"
+        )
+        delivery_err_counter = Counter(
+            "send_delivery_errors_total", "Kafka delivery errors"
+        )
+        statistics_reporter.register_metric(
+            update_message_counter.name, update_message_counter
+        )
+        statistics_reporter.register_metric(buffer_err_counter.name, buffer_err_counter)
+        statistics_reporter.register_metric(
+            delivery_err_counter.name, delivery_err_counter
+        )
+        return KafkaProducer(
+            producer,
+            update_msg_counter=update_message_counter,
+            update_buffer_err_counter=buffer_err_counter,
+            update_delivery_err_counter=delivery_err_counter,
+        )
 
 
 def create_consumer(

forwarder/kafka/kafka_producer.py

@@ -4,7 +4,7 @@
 import confluent_kafka
 
 from forwarder.application_logger import get_logger
-from forwarder.utils import Counter
+from forwarder.metrics import Counter
 
 
 class KafkaProducer:
@@ -44,12 +44,12 @@ def ack(err, _):
             if err:
                 self.logger.error(f"Message failed delivery: {err}")
                 if self._update_delivery_err_counter:
-                    self._update_delivery_err_counter.increment()
+                    self._update_delivery_err_counter.inc()
             else:
                 # increment only for PVs related updates
                 # key is None when we send commands.
                 if self._update_msg_counter and key is not None:
-                    self._update_msg_counter.increment()
+                    self._update_msg_counter.inc()
 
         try:
             self._producer.produce(
@@ -59,5 +59,5 @@ def ack(err, _):
             # Producer message buffer is full.
             # Data loss occurred as messages are produced faster than are sent to the kafka broker.
             if self._update_buffer_err_counter:
-                self._update_buffer_err_counter.increment()
+                self._update_buffer_err_counter.inc()
         self._producer.poll(0)

forwarder/metrics/__init__.py

@@ -0,0 +1,45 @@
+"""
+Classes to model monitoring metrics based on the Prometheus client library.
+
+Since we push these metrics to Graphite instead of relying on Prometheus
+standard mechanisms, we provide adapter classes to prevent coupling other
+modules with certain Prometheus internals.
+"""
+
+from prometheus_client import Counter as PrometheusCounter
+from prometheus_client import Gauge as PrometheusGauge
+from prometheus_client import Summary as PrometheusSummary
+
+
+class Gauge(PrometheusGauge):
+    @property
+    def name(self):
+        return self._name
+
+    @property
+    def value(self):
+        return self._value.get()
+
+
+class Counter(PrometheusCounter):
+    @property
+    def name(self):
+        return self._name
+
+    @property
+    def value(self):
+        return self._value.get()
+
+
+class Summary(PrometheusSummary):
+    @property
+    def name(self):
+        return self._name
+
+    @property
+    def count(self):
+        return self._count.get()
+
+    @property
+    def sum(self):
+        return self._sum.get()

forwarder/metrics/statistics_reporter.py

@@ -0,0 +1,53 @@
+import time
+from logging import Logger
+from typing import Dict, Union, get_args
+
+import graphyte  # type: ignore
+
+from forwarder.metrics import Counter, Gauge, Summary
+from forwarder.repeat_timer import RepeatTimer
+
+MetricType = Union[Counter, Summary, Gauge]
+
+
+class StatisticsReporter:
+    def __init__(
+        self,
+        graphyte_server: str,
+        logger: Logger,
+        prefix: str = "forwarder",
+        update_interval_s: int = 10,
+    ):
+        self._graphyte_server = graphyte_server
+        self._logger = logger
+        self._sender = graphyte.Sender(self._graphyte_server, prefix=prefix)
+        self._repeating_timer = RepeatTimer(update_interval_s, self.send_statistics)
+        self._metrics: Dict[str, MetricType] = {}
+
+    def register_metric(self, name: str, metric: MetricType):
+        if not isinstance(metric, get_args(MetricType)):
+            raise TypeError(f"Unsupported metric type: {type(metric).__name__}")
+        self._metrics[name] = metric
+
+    def deregister_metric(self, name: str):
+        if name in self._metrics:
+            del self._metrics[name]
+
+    def start(self):
+        self._repeating_timer.start()
+
+    def send_statistics(self):
+        try:
+            for metric_name, metric in self._metrics.items():
+                timestamp = time.time()
+                if isinstance(metric, (Counter, Gauge)):
+                    self._sender.send(metric_name, metric.value, timestamp)
+                elif isinstance(metric, Summary):
+                    self._sender.send(f"{metric_name}_sum", metric.sum, timestamp)
+                    self._sender.send(f"{metric_name}_count", metric.count, timestamp)
+        except Exception as ex:
+            self._logger.error(f"Could not send statistics: {ex}")
+
+    def stop(self):
+        if self._repeating_timer:
+            self._repeating_timer.cancel()