Write keygrabber samples to several named sinks, and add an OTLP sink for Prometheus

docs/source/keygrabber.md

@@ -39,12 +39,15 @@ group_id: hispec
 transport: rabbitmq
 rabbitmq_url: amqp://localhost
 
-sink:
-  type: influxdb
-  url: http://influx.hispec:8086
-  org: hispec
-  bucket: telemetry
-  token_env: HISPEC_INFLUX_TOKEN
+sinks:
+  influxdb:
+    type: influxdb
+    url: http://influx.hispec:8086
+    org: hispec
+    bucket: telemetry
+    token_env: HISPEC_INFLUX_TOKEN
+  prometheus:
+    type: otlp
 
 workers: 4
 
@@ -65,13 +68,38 @@ collections:
     exclude: ["units_code"]
 ```
 
-### sink
+### sinks
 
-`type` selects the backend; `influxdb` is the only one implemented.
+A map of named sinks, at least one. Every sample goes to each of them, from a
+single read: adding a sink adds no load on the peers. A sink name must match
+`[a-z0-9_]+`, because it becomes the prefix of that sink's status keywords.
 
-The token is **never** written in the config. `token_env` names an environment
-variable to read it from, and a `token` key in the file is rejected outright.
-`timeout_ms` is optional.
+Each sink has its own writer thread, queue and retry backoff, so one that is
+down or slow fills and drops only its own queue. The others, and the reads,
+carry on.
+
+`type` selects the backend, and each type takes only the settings below; any
+other key is refused, so a misspelt one cannot pass unnoticed.
+
+**`influxdb`** writes one measurement per keyword, tagged by `group`, `peer` and
+`units`. Its token is **never** written in the config: `token_env` names an
+environment variable to read it from, and a `token` key in the file is rejected
+outright. `timeout_ms` is optional. Needs the `influxdb` extra.
+
+**`otlp`** exports OpenTelemetry gauges to the endpoint in
+`OTEL_EXPORTER_OTLP_ENDPOINT`, for Prometheus behind an OpenTelemetry Collector
+(see {doc}`telemetry`). Every numeric and bool keyword is one metric,
+`libby_keyword`, labelled `group`, `peer`, `keyword` and `units`, and each point
+carries the keyword's read time:
+
+```promql
+libby_keyword{group="hsfei", peer="adc", keyword="positionvalue1"}
+```
+
+Bools are stored as 0 and 1; string and null values are not stored. A batch
+written late, after an outage, still lands at its read time as long as
+Prometheus accepts it, which the backend does for two hours. `timeout_s` is
+optional. Needs the `otel` extra.
 
 ### collections
 
@@ -118,39 +146,46 @@ the ordinary `libby` verbs, and it can feed the same dashboards it fills.
 | Keyword | Type | Access | Meaning |
 |---|---|---|---|
 | `enabled` | bool | R/W | Collect on the configured cadences; false pauses without exiting |
-| `isconnected` | bool | R/W | Last sink write succeeded; write true to request a reconnect |
-| `pointswritten` | int | R | Samples the sink has stored since start |
+| `isconnected` | bool | R/W | Last write to every sink succeeded; write true to request a reconnect of each |
+| `pointswritten` | int | R | Samples stored since start, summed over sinks |
 | `readerrors` | int | R | Keyword reads that failed |
-| `writeerrors` | int | R | Sink writes that failed |
-| `queuedepth` | int | R | Batches waiting in the retry queue |
+| `writeerrors` | int | R | Sink writes that failed, summed over sinks |
+| `queuedepth` | int | R | Batches waiting in retry queues, summed over sinks |
 | `skippedticks` | int | R | Ticks skipped because the previous read was still in flight |
-| `droppedbatches` | int | R | Batches discarded because a queue was full |
+| `droppedbatches` | int | R | Batches discarded because a sink's queue was full, summed over sinks |
 | `reload` | trigger | W | Re-read the config file and apply it |
 | `shutdown` | trigger | W | Gracefully stop the daemon |
 | `<collection>.enabled` | bool | R/W | Collect this one collection |
 | `<collection>.interval` | float | R/W | Cadence in seconds |
 | `<collection>.lastsample` | string | R | UTC time of the last successful tick |
 | `<collection>.lag` | float | R | Seconds the last tick ran past its due time |
+| `<sink>.isconnected` | bool | R/W | Last write to this sink succeeded; write true to reconnect it |
+| `<sink>.pointswritten` | int | R | Samples this sink has stored since start |
+| `<sink>.writeerrors` | int | R | Writes to this sink that failed |
+| `<sink>.queuedepth` | int | R | Batches waiting in this sink's retry queue |
+| `<sink>.droppedbatches` | int | R | Batches discarded because this sink's queue was full |
 
 ```bash
 libby show   hispec.keygrabber.%
 libby modify hispec.keygrabber.adc.interval=30
 libby modify hispec.keygrabber.enabled=false
 ```
 
-Per-collection keywords contain a dot, and `%` matches within a single segment
+Per-collection and per-sink keywords contain a dot, and `%` matches within a single segment
 only, so `libby list hispec.keygrabber.%` will not show `adc.enabled`. Use
 `hispec.keygrabber.%.%` for those.
 
-`isconnected` reports whether the last write to the sink succeeded; it does not
-ping the database, because these getters are answered on the transport's
+`isconnected` reports whether the last write to each sink succeeded; it does not
+ping a database, because these getters are answered on the transport's
 receive thread and a blocking one would time out every read in flight. Writing
 `true` asks the writer thread to reconnect and returns immediately, so poll the
 keyword for the outcome. There is no manual disconnect.
 
-A failing sink is visible from three keywords together: `isconnected` goes
-false, `queuedepth` climbs as batches wait, and `pointswritten` stops moving.
-`writeerrors` counts every failed attempt including retries.
+A failing sink is visible from three keywords together: `<sink>.isconnected`
+goes false, `<sink>.queuedepth` climbs as batches wait, and
+`<sink>.pointswritten` stops moving, while the other sinks' keep moving. The
+unprefixed `isconnected` goes false when any sink fails. `writeerrors` counts
+every failed attempt including retries.
 
 ### reload
 
@@ -161,7 +196,7 @@ to the caller and on `lasterror`.
 
 It will not add or remove collections, and says so rather than half-applying:
 libby has no way to withdraw a keyword, so a new collection's control keywords
-could not appear without a restart. Changing `sink` or `workers` also needs a
+could not appear without a restart. Changing `sinks` or `workers` also needs a
 restart.
 
 ## How it reads

libby/keygrabber/channel.py

@@ -0,0 +1,159 @@
+"""One sink's write path: its retry writer, queue, thread and health."""
+from __future__ import annotations
+
+import logging
+import queue
+import threading
+import time
+from typing import Optional, Tuple
+
+from ..errors import LibbyError
+from .sink import RetryingWriter, RetryPolicy, Sample, Sink
+
+# How long the writer waits for a batch before checking the retry queue
+WRITER_POLL_S = 0.25
+
+# Time allowed for the retry queue to drain during shutdown. Must stay well
+# under the systemd unit's TimeoutStopSec so a wedged sink cannot turn a stop
+# into a SIGKILL.
+DRAIN_DEADLINE_S = 5.0
+
+
+# The attributes are the sink's writer, queue, thread and the health it
+# reports, each one value rather than hidden state
+class SinkChannel:  # pylint: disable=too-many-instance-attributes
+    """Writes batches to one sink, independently of every other sink.
+
+    Each sink has its own bounded queue and writer thread, so one that stalls
+    or fails fills and drops only its own queue, and never holds up the
+    readers or another sink. Only this channel's thread calls into its sink.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        sink: Sink,
+        *,
+        policy: RetryPolicy,
+        queue_size: int,
+        logger: logging.Logger,
+    ) -> None:
+        self.name = name
+        self.healthy = True
+        self.points_written = 0
+        self.dropped_batches = 0
+        self.thread: Optional[threading.Thread] = None
+        self._writer = RetryingWriter(sink, policy=policy)
+        self._queue: "queue.Queue[Tuple[Sample, ...]]" = queue.Queue(maxsize=queue_size)
+        self._logger = logger
+        self._reconnect_wanted = False
+        # Readers on several threads count drops; only ever held for the increment
+        self._lock = threading.Lock()
+
+    @property
+    def write_errors(self) -> int:
+        """Attempts on the sink that failed, including retries."""
+        return self._writer.failed_attempts
+
+    @property
+    def queue_depth(self) -> int:
+        """Batches waiting in the retry queue."""
+        return self._writer.queue_depth
+
+    def connect(self) -> None:
+        """Open the sink."""
+        self._writer.connect()
+
+    def offer(self, batch: Tuple[Sample, ...]) -> None:
+        """Queue a batch for this sink, dropping it if the queue is full."""
+        try:
+            self._queue.put_nowait(batch)
+        except queue.Full:
+            with self._lock:
+                self.dropped_batches += 1
+            self._logger.error("sink %s: sample queue full; dropped %d samples",
+                               self.name, len(batch))
+
+    def request_reconnect(self) -> None:
+        """Ask the writer thread to reconnect, rather than blocking the caller."""
+        self._reconnect_wanted = True
+
+    def run(self, halt: threading.Event) -> None:
+        """Write batches until ``halt`` is set, then drain under a deadline."""
+        while not halt.is_set():
+            self._reconnect_if_requested()
+            try:
+                batch = self._queue.get(timeout=WRITER_POLL_S)
+            except queue.Empty:
+                self._flush()
+                continue
+            self._write(batch)
+            self._flush()
+        self._drain()
+
+    def close(self) -> None:
+        """Release the sink."""
+        self._writer.close()
+
+    def _reconnect_if_requested(self) -> None:
+        if not self._reconnect_wanted:
+            return
+        self._reconnect_wanted = False
+        try:
+            self._writer.connect()
+            self.healthy = True
+            self._logger.info("sink %s: reconnected on request", self.name)
+        except LibbyError as exc:
+            self.healthy = False
+            self._logger.error("sink %s: reconnect failed: %s", self.name, exc)
+
+    def _write(self, batch: Tuple[Sample, ...]) -> None:
+        try:
+            self.points_written += self._writer.write(batch)
+        except LibbyError as exc:
+            # A sink raising something other than SinkWriteError is a bug in
+            # that sink; the retry writer turns the expected failure into a
+            # queued batch instead
+            self._logger.error("sink %s: write raised: %s", self.name, exc)
+        self._track_health()
+
+    def _flush(self) -> None:
+        try:
+            self.points_written += self._writer.flush_due()
+        except LibbyError as exc:
+            self._logger.error("sink %s: retry raised: %s", self.name, exc)
+        self._track_health()
+
+    def _track_health(self) -> None:
+        """Mirror the writer's view, and log only when it changes.
+
+        ``RetryingWriter.write`` queues a failed batch and returns 0 rather
+        than raising, which is what makes retrying possible, so health is read
+        back from the writer rather than from an exception.
+        """
+        healthy = self._writer.healthy
+        if healthy == self.healthy:
+            return
+        self.healthy = healthy
+        if healthy:
+            self._logger.info("sink %s: writes are succeeding again", self.name)
+        else:
+            self._logger.error("sink %s: writes are failing, %d batch(es) queued",
+                               self.name, self._writer.queue_depth)
+
+    def _drain(self) -> None:
+        deadline = time.monotonic() + DRAIN_DEADLINE_S
+        while time.monotonic() < deadline:
+            try:
+                self._write(self._queue.get_nowait())
+                continue
+            except queue.Empty:
+                pass
+            if self._writer.queue_depth == 0:
+                return
+            self._flush()
+            if self._writer.queue_depth:
+                time.sleep(WRITER_POLL_S)
+        if self._writer.queue_depth:
+            self._logger.error("sink %s: gave up draining %d batches after %.1fs",
+                               self.name, self._writer.queue_depth, DRAIN_DEADLINE_S)