From ce24ae6346dc9bf7b8ce77382000198294045555 Mon Sep 17 00:00:00 2001 From: Mike Langmayr Date: Thu, 5 Nov 2026 13:04:24 -0800 Subject: [PATCH 1/2] Write keygrabber samples to several named sinks, and add an OTLP sink for Prometheus --- docs/source/keygrabber.md | 80 +++++--- libby/keygrabber/channel.py | 159 ++++++++++++++++ libby/keygrabber/config.py | 73 +++++-- libby/keygrabber/daemon.py | 285 +++++++++++----------------- libby/keygrabber/otlp.py | 103 ++++++++++ tests/test_keygrabber_collection.py | 1 + tests/test_keygrabber_config.py | 44 ++++- tests/test_keygrabber_daemon.py | 64 +++++-- tests/test_otlp_sink.py | 201 ++++++++++++++++++++ 9 files changed, 770 insertions(+), 240 deletions(-) create mode 100644 libby/keygrabber/channel.py create mode 100644 libby/keygrabber/otlp.py create mode 100644 tests/test_otlp_sink.py diff --git a/docs/source/keygrabber.md b/docs/source/keygrabber.md index 4790ec5..4a11730 100644 --- a/docs/source/keygrabber.md +++ b/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,37 @@ 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. + +**`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,19 +145,24 @@ 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 | | `.enabled` | bool | R/W | Collect this one collection | | `.interval` | float | R/W | Cadence in seconds | | `.lastsample` | string | R | UTC time of the last successful tick | | `.lag` | float | R | Seconds the last tick ran past its due time | +| `.isconnected` | bool | R/W | Last write to this sink succeeded; write true to reconnect it | +| `.pointswritten` | int | R | Samples this sink has stored since start | +| `.writeerrors` | int | R | Writes to this sink that failed | +| `.queuedepth` | int | R | Batches waiting in this sink's retry queue | +| `.droppedbatches` | int | R | Batches discarded because this sink's queue was full | ```bash libby show hispec.keygrabber.% @@ -138,19 +170,21 @@ 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: `.isconnected` +goes false, `.queuedepth` climbs as batches wait, and +`.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 +195,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 diff --git a/libby/keygrabber/channel.py b/libby/keygrabber/channel.py new file mode 100644 index 0000000..a75b808 --- /dev/null +++ b/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) diff --git a/libby/keygrabber/config.py b/libby/keygrabber/config.py index 6d93cc1..5a2fba4 100644 --- a/libby/keygrabber/config.py +++ b/libby/keygrabber/config.py @@ -1,8 +1,8 @@ """Parse and validate the keygrabber's own configuration sections. A keygrabber config file is an ordinary ``LibbyDaemon`` config (``peer_id``, -``group_id``, ``transport``, ...) plus three sections this module owns: -``sink``, ``defaults`` and ``collections``. +``group_id``, ``transport``, ...) plus the sections this module owns: +``sinks``, ``defaults`` and ``collections``. """ from __future__ import annotations @@ -32,6 +32,9 @@ _COLLECTION_NAME = re.compile(r"^[a-z0-9_]+$") +# Standard OpenTelemetry variables an otlp sink can export to; otel.env sets the first +_OTLP_ENDPOINT_VARIABLES = ("OTEL_EXPORTER_OTLP_ENDPOINT", "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") + # A config record, so the field count is the schema rather than complexity @dataclass(frozen=True) @@ -64,13 +67,18 @@ class KeygrabberConfig: """The keygrabber's parsed configuration.""" collections: Tuple[CollectionConfig, ...] - sink: Mapping[str, Any] + # Sink name to its settings; every sample goes to each of them + sinks: Mapping[str, Mapping[str, Any]] workers: int retry: RetryPolicy def parse_config(config: Mapping[str, Any]) -> KeygrabberConfig: """Validate a daemon config's keygrabber sections into typed objects.""" + if "sink" in config: + raise ConfigError( + "'sink' is now 'sinks', a map of named sinks, e.g. " + "sinks: {influxdb: {type: influxdb, ...}}") defaults = _mapping(config, "defaults") collections = _mapping(config, "collections") @@ -80,44 +88,60 @@ def parse_config(config: Mapping[str, Any]) -> KeygrabberConfig: ) return KeygrabberConfig( collections=parsed, - sink=_mapping(config, "sink"), + sinks=_parse_sinks(_mapping(config, "sinks")), workers=_positive_int(config.get("workers", DEFAULT_WORKERS), "workers"), retry=_parse_retry(_mapping(config, "retry")), ) -def build_sink(sink_config: Mapping[str, Any]) -> Sink: - """Build the configured sink, importing its backend client on demand.""" +def _parse_sinks(raw: Mapping[str, Any]) -> Mapping[str, Mapping[str, Any]]: + if not raw: + raise ConfigError("sinks needs at least one sink") + for name, settings in raw.items(): + if not _COLLECTION_NAME.match(str(name)): + raise ConfigError( + f"sink name {name!r} must be lowercase letters, digits and underscores") + if not isinstance(settings, Mapping): + raise ConfigError(f"sinks.{name} must be a mapping") + return dict(raw) + + +def build_sink(name: str, sink_config: Mapping[str, Any]) -> Sink: + """Build the sink named ``name``, importing its backend client on demand.""" + label = f"sinks.{name}" kind = str(sink_config.get("type", "")).strip().lower() if not kind: - raise ConfigError("sink.type is required") + raise ConfigError(f"{label}.type is required") + if kind == "otlp": + return _build_otlp_sink(label, sink_config) if kind != "influxdb": - raise ConfigError(f"unsupported sink type {kind!r}; expected 'influxdb'") + raise ConfigError( + f"unsupported {label}.type {kind!r}; expected 'influxdb' or 'otlp'") if "token" in sink_config: raise ConfigError( - "sink.token must not appear in the config; name an environment " - "variable with sink.token_env instead" + f"{label}.token must not appear in the config; name an environment " + f"variable with {label}.token_env instead" ) - token_env = _required(sink_config, "token_env", "sink") + token_env = _required(sink_config, "token_env", label) token = os.environ.get(str(token_env)) if not token: raise ConfigError( - f"environment variable {token_env} is unset or empty, so the " - "sink has no token" + f"environment variable {token_env} is unset or empty, so {label} " + "has no token" ) # Validate every field before importing the backend, so a malformed config # is reported the same way whether or not the extra is installed settings = { - "url": str(_required(sink_config, "url", "sink")), - "org": str(_required(sink_config, "org", "sink")), - "bucket": str(_required(sink_config, "bucket", "sink")), + "url": str(_required(sink_config, "url", label)), + "org": str(_required(sink_config, "org", label)), + "bucket": str(_required(sink_config, "bucket", label)), "token": token, } if "timeout_ms" in sink_config: settings["timeout_ms"] = _positive_int( - sink_config["timeout_ms"], "sink.timeout_ms") + sink_config["timeout_ms"], f"{label}.timeout_ms") # Deferred: influxdb-client is an optional extra, so a deployment using a # different backend never has to install it @@ -127,6 +151,21 @@ def build_sink(sink_config: Mapping[str, Any]) -> Sink: return InfluxSink(InfluxConfig(**settings)) +def _build_otlp_sink(label: str, sink_config: Mapping[str, Any]) -> Sink: + if not any(os.environ.get(variable) for variable in _OTLP_ENDPOINT_VARIABLES): + raise ConfigError( + f"{label} exports to OTEL_EXPORTER_OTLP_ENDPOINT, which is unset; " + "set it in otel.env") + timeout_s = None + if "timeout_s" in sink_config: + timeout_s = _positive_float(sink_config["timeout_s"], f"{label}.timeout_s") + # Deferred: the OpenTelemetry SDK is the optional otel extra + from .otlp import ( # pylint: disable=import-outside-toplevel + DEFAULT_TIMEOUT_S as OTLP_TIMEOUT_S, OtlpSink, + ) + return OtlpSink(timeout_s=timeout_s or OTLP_TIMEOUT_S) + + def _parse_collection( name: str, raw: Any, diff --git a/libby/keygrabber/daemon.py b/libby/keygrabber/daemon.py index 6a1eb5e..f83c87b 100644 --- a/libby/keygrabber/daemon.py +++ b/libby/keygrabber/daemon.py @@ -2,39 +2,30 @@ from __future__ import annotations import dataclasses -import queue import threading -import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone -from typing import Callable, List, Optional, Sequence, Tuple +from typing import Any, Callable, List, Mapping, Optional, Sequence from ..client import Client from ..config import ConfigError, DaemonConfigLoader, with_env_overrides from ..daemon import LibbyDaemon from ..errors import LibbyError +from .channel import DRAIN_DEADLINE_S, WRITER_POLL_S, SinkChannel from .collection import FAILURES_BEFORE_BACKOFF, Collection from .config import TIMEOUT_HEADROOM, KeygrabberConfig, build_sink, parse_config from .scheduler import Scheduler -from .sink import RetryingWriter, Sample, Sink - -# How long the writer waits for a batch before checking the retry queue -WRITER_POLL_S = 0.25 +from .sink import Sample, Sink # Scheduler wake interval, so a stop request is noticed promptly even when the # next tick is far off SCHEDULER_TICK_S = 0.25 -# Batches held between the readers and the writer. Bounded so a stalled writer -# cannot grow memory; the retry queue inside RetryingWriter handles a stalled -# backend. +# Batches held between the readers and each sink's writer. Bounded so a +# stalled writer cannot grow memory; the retry queue inside RetryingWriter +# handles a stalled backend. QUEUE_DEPTH_PER_WORKER = 4 -# 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 - # How long ``on_stop`` waits for a thread to finish. Outlasts the drain # deadline, so a writer that is draining gets to finish rather than being # abandoned mid-batch. @@ -51,20 +42,18 @@ class _ConfigSource: class Counters: # pylint: disable=too-few-public-methods - """Tallies the daemon reports, guarded for cross-thread increments. + """Read-side tallies the daemon reports, guarded for cross-thread increments. The lock is only ever held around an increment, never across a read or a sink write, because these are served on the transport's receive thread and - blocking it would time out every read already in flight. + blocking it would time out every read already in flight. Write-side tallies + are each sink's, on its :class:`SinkChannel`. """ def __init__(self) -> None: self._lock = threading.Lock() - self.points_written = 0 self.read_errors = 0 - self.write_errors = 0 self.skipped_ticks = 0 - self.dropped_batches = 0 def add(self, **deltas: int) -> None: """Add to one or more counters.""" @@ -82,11 +71,11 @@ def set(self, **values: int) -> None: # Coordinates config, client, sink, collections and three kinds of thread; # the attribute count is collaborators, not state class KeygrabberDaemon(LibbyDaemon): # pylint: disable=too-many-instance-attributes - """Polls keywords from other libby peers and writes them to a sink. + """Polls keywords from other libby peers and writes them to every sink. One process serves the whole fleet: cadence lives in this daemon's config - rather than in each hardware daemon, and the database credential lives in - one place. + rather than in each hardware daemon, and the database credentials live in + one place. Each keyword is read once per tick, whatever the number of sinks. """ transport = "rabbitmq" @@ -98,45 +87,47 @@ def __init__(self) -> None: self.enabled = True self._settings: Optional[KeygrabberConfig] = None self._client: Optional[Client] = None - self._writer: Optional[RetryingWriter] = None + self._channels: List[SinkChannel] = [] self._collections: List[Collection] = [] - self._queue: "queue.Queue[Tuple[Sample, ...]]" = queue.Queue() self._pool: Optional[ThreadPoolExecutor] = None self._threads: List[threading.Thread] = [] - self._writer_thread: Optional[threading.Thread] = None self._halt = threading.Event() self._scheduler = Scheduler() self._source: Optional[_ConfigSource] = None - self._sink_healthy = True - self._reconnect_wanted = False - self._clock: Callable[[], float] = time.monotonic def on_start(self) -> None: - """Parse config, open the sink, and start the reader and writer threads.""" + """Parse config, open the sinks, and start the reader and writer threads.""" self._settings = parse_config(self._config) self._client = Client(self.libby) - self._writer = RetryingWriter(self.make_sink(), - policy=self._settings.retry) - self._writer.connect() + queue_size = max(1, self._settings.workers * QUEUE_DEPTH_PER_WORKER) + self._channels = [ + SinkChannel(name, self.make_sink(name, sink_config), + policy=self._settings.retry, queue_size=queue_size, + logger=self.logger) + for name, sink_config in self._settings.sinks.items() + ] + for channel in self._channels: + channel.connect() self._collections = [Collection(config) for config in self._settings.collections] - self._queue = queue.Queue( - maxsize=max(1, self._settings.workers * QUEUE_DEPTH_PER_WORKER)) self._pool = ThreadPoolExecutor(max_workers=self._settings.workers, thread_name_prefix="keygrabber-read") self._scheduler.replace(self._collections) self._register_control_keywords() self._halt.clear() - self._writer_thread = self._spawn(self._write_loop, "keygrabber-write") + for channel in self._channels: + channel.thread = self._spawn(lambda c=channel: c.run(self._halt), + f"keygrabber-write-{channel.name}") self._spawn(self._schedule_loop, "keygrabber-schedule") - self.logger.info("collecting %d collections with %d workers", - len(self._collections), self._settings.workers) + self.logger.info("collecting %d collections with %d workers into %s", + len(self._collections), self._settings.workers, + ", ".join(channel.name for channel in self._channels)) def on_stop(self) -> None: - """Stop the threads and close the sink once the writer has drained it. + """Stop the threads and close each sink once its writer has drained it. - The draining happens on the writer thread rather than here, so only one - thread is ever inside the sink. This method just waits for it. + The draining happens on each sink's writer thread rather than here, so + only one thread is ever inside a sink. This method just waits for them. """ self._halt.set() if self._pool is not None: @@ -146,17 +137,16 @@ def on_stop(self) -> None: thread.join(timeout=STOP_JOIN_TIMEOUT_S) self._threads = [] - writer, self._writer = self._writer, None - if writer is None: - return - if self._writer_thread is not None and self._writer_thread.is_alive(): - # Closing while the writer is still inside a sink call would put - # two threads in one sink; leave it to process exit instead - self.logger.error( - "writer thread did not finish within %.1fs; leaving the sink open", - STOP_JOIN_TIMEOUT_S) - return - writer.close() + channels, self._channels = self._channels, [] + for channel in channels: + if channel.thread is not None and channel.thread.is_alive(): + # Closing while the writer is still inside a sink call would put + # two threads in one sink; leave it to process exit instead + self.logger.error( + "sink %s: writer did not finish within %.1fs; leaving it open", + channel.name, STOP_JOIN_TIMEOUT_S) + continue + channel.close() @classmethod def from_config_file(cls, path, daemon_id=None, *, env_prefix=None): @@ -165,15 +155,28 @@ def from_config_file(cls, path, daemon_id=None, *, env_prefix=None): daemon._source = _ConfigSource(str(path), daemon_id, env_prefix) return daemon - def make_sink(self) -> Sink: - """Build the configured sink. + def make_sink(self, name: str, sink_config: Mapping[str, Any]) -> Sink: + """Build one configured sink. Override to supply a sink the config cannot describe, or to inject one in a test without installing a database client. """ - if self._settings is None: - raise LibbyError("keygrabber config has not been parsed yet") - return build_sink(self._settings.sink) + return build_sink(name, sink_config) + + @property + def points_written(self) -> int: + """Samples stored, summed over every sink.""" + return sum(channel.points_written for channel in self._channels) + + @property + def write_errors(self) -> int: + """Failed sink writes, summed over every sink.""" + return sum(channel.write_errors for channel in self._channels) + + @property + def dropped_batches(self) -> int: + """Batches dropped on a full queue, summed over every sink.""" + return sum(channel.dropped_batches for channel in self._channels) ### Control surface @@ -182,8 +185,9 @@ def _register_control_keywords(self) -> None: Every getter here is answered on the transport's receive thread, so none of them may block: the counters are plain reads, ``isconnected`` - reports a flag the writer thread maintains rather than pinging the - database, and writing it only requests a reconnect. + reports flags the writer threads maintain rather than pinging a + database, and writing it only requests a reconnect. The unprefixed + sink keywords cover every sink; ``.*`` report one. """ registry = self.keyword_registry registry.bool("enabled", @@ -192,30 +196,53 @@ def _register_control_keywords(self) -> None: description="Collect on the configured cadences; " "write false to pause without exiting.") registry.bool("isconnected", - getter=lambda: self._sink_healthy, - setter=self._set_connected, - description="Last sink write succeeded; write true to " - "request a reconnect.") - registry.int("pointswritten", getter=lambda: self.counters.points_written, - description="Samples the sink has stored since start.") + getter=lambda: all(c.healthy for c in self._channels), + setter=lambda value: self._set_connected(self._channels, value), + description="Last write to every sink succeeded; write true " + "to request a reconnect of each.") + registry.int("pointswritten", getter=lambda: self.points_written, + description="Samples stored since start, summed over sinks.") registry.int("readerrors", getter=lambda: self.counters.read_errors, description="Keyword reads that failed.") - registry.int("writeerrors", getter=lambda: self.counters.write_errors, - description="Sink writes that failed.") - registry.int("queuedepth", getter=self._queue_depth, - description="Batches waiting in the retry queue.") + registry.int("writeerrors", getter=lambda: self.write_errors, + description="Sink writes that failed, summed over sinks.") + registry.int("queuedepth", + getter=lambda: sum(c.queue_depth for c in self._channels), + description="Batches waiting in retry queues, summed over sinks.") registry.int("skippedticks", getter=lambda: self.counters.skipped_ticks, description="Ticks skipped because the previous read " "was still in flight.") - registry.int("droppedbatches", getter=lambda: self.counters.dropped_batches, - description="Batches discarded because a queue was full.") + registry.int("droppedbatches", getter=lambda: self.dropped_batches, + description="Batches discarded because a sink's queue was " + "full, summed over sinks.") registry.trigger("reload", action=self._reload, description="Re-read the config file and apply it.") registry.trigger("shutdown", action=self.request_stop, description="Gracefully stop this daemon.") + for channel in self._channels: + self._register_sink_keywords(channel) for collection in self._collections: self._register_collection_keywords(collection) + def _register_sink_keywords(self, channel: SinkChannel) -> None: + """Expose one sink's health, so a failing one can be told apart.""" + name = channel.name + registry = self.keyword_registry + registry.bool(f"{name}.isconnected", + getter=lambda c=channel: c.healthy, + setter=lambda value, c=channel: self._set_connected([c], value), + description=f"Last write to {name} succeeded; write true " + "to request a reconnect.") + registry.int(f"{name}.pointswritten", getter=lambda c=channel: c.points_written, + description=f"Samples {name} has stored since start.") + registry.int(f"{name}.writeerrors", getter=lambda c=channel: c.write_errors, + description=f"Writes to {name} that failed.") + registry.int(f"{name}.queuedepth", getter=lambda c=channel: c.queue_depth, + description=f"Batches waiting in {name}'s retry queue.") + registry.int(f"{name}.droppedbatches", + getter=lambda c=channel: c.dropped_batches, + description=f"Batches discarded because {name}'s queue was full.") + def _register_collection_keywords(self, collection: Collection) -> None: """Expose one collection's cadence and health.""" name = collection.name @@ -244,16 +271,18 @@ def _set_enabled(self, value: bool) -> None: self.enabled = value self.logger.info("collection %s", "enabled" if value else "paused") - def _set_connected(self, value: bool) -> None: - """Request a reconnect; refuse a write of false. + @staticmethod + def _set_connected(channels: Sequence[SinkChannel], value: bool) -> None: + """Request a reconnect of ``channels``; refuse a write of false. - The reconnect happens on the writer thread rather than here, because a + The reconnect happens on each writer thread rather than here, because a dead database would otherwise hold the receive thread for the client's whole connect timeout. Poll ``isconnected`` for the outcome. """ if not value: raise ValueError("write true to reconnect; there is no manual disconnect") - self._reconnect_wanted = True + for channel in channels: + channel.request_reconnect() def _set_interval(self, name: str, value: float) -> None: """Change one collection's cadence, keeping the timeout headroom rule.""" @@ -273,10 +302,6 @@ def _collection(self, name: str) -> Collection: return collection raise ValueError(f"unknown collection {name!r}") - def _queue_depth(self) -> int: - writer = self._writer - return writer.queue_depth if writer is not None else 0 - def _reload(self) -> None: """Re-read the config file and apply it to the running collections. @@ -405,105 +430,9 @@ def _note_success(self, collection: Collection) -> None: collection.note_success() def _enqueue(self, samples: Sequence[Sample]) -> None: - try: - self._queue.put_nowait(tuple(samples)) - except queue.Full: - self.counters.add(dropped_batches=1) - self.logger.error("sample queue full; dropped %d samples", - len(samples)) - - ### Writing - - def _write_loop(self) -> None: - """Own every sink call, so no reader thread ever touches the backend.""" - while not self._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() - # Halted: drain here, on the one thread allowed to touch the sink - self._drain() - - def _reconnect_if_requested(self) -> None: - """Honour an isconnected write, on this thread rather than the caller's.""" - if not self._reconnect_wanted: - return - self._reconnect_wanted = False - writer = self._writer - if writer is None: - return - try: - writer.connect() - self._sink_healthy = True - self.logger.info("reconnected to the sink on request") - except LibbyError as exc: - self._sink_healthy = False - self.logger.error("sink reconnect failed: %s", exc) - - def _write(self, batch: Tuple[Sample, ...]) -> None: - writer = self._writer - if writer is None: - return - try: - self.counters.add(points_written=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 write raised: %s", exc) - self._track_sink_health(writer) - - def _flush(self) -> None: - writer = self._writer - if writer is None: - return - try: - self.counters.add(points_written=writer.flush_due()) - except LibbyError as exc: - self.logger.error("sink retry raised: %s", exc) - self._track_sink_health(writer) - - def _track_sink_health(self, writer: RetryingWriter) -> 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. It also means a - failure cannot be noticed from an exception here, so health and the - error count are read back from the writer instead. - """ - self.counters.set(write_errors=writer.failed_attempts) - healthy = writer.healthy - if healthy == self._sink_healthy: - return - self._sink_healthy = healthy - if healthy: - self.logger.info("sink writes are succeeding again") - else: - self.logger.error("sink writes are failing, %d batch(es) queued", - writer.queue_depth) - - def _drain(self) -> None: - """Write whatever is still queued, under a deadline.""" - deadline = self._clock() + DRAIN_DEADLINE_S - while self._clock() < deadline: - try: - self._write(self._queue.get_nowait()) - continue - except queue.Empty: - pass - writer = self._writer - if writer is None or writer.queue_depth == 0: - return - self._flush() - if writer.queue_depth: - time.sleep(WRITER_POLL_S) - if self._writer is not None and self._writer.queue_depth: - self.logger.error("gave up draining %d batches after %.1fs", - self._writer.queue_depth, DRAIN_DEADLINE_S) + batch = tuple(samples) + for channel in self._channels: + channel.offer(batch) def _require_client(self) -> Client: if self._client is None: diff --git a/libby/keygrabber/otlp.py b/libby/keygrabber/otlp.py new file mode 100644 index 0000000..d215bd2 --- /dev/null +++ b/libby/keygrabber/otlp.py @@ -0,0 +1,103 @@ +"""OTLP sink: keyword samples as gauges, for Prometheus behind an OpenTelemetry Collector. + +Every numeric and bool keyword is one gauge, ``libby_keyword``, labelled by +``group``, ``peer``, ``keyword`` and ``units``, so a PromQL query is one +metric name plus label matchers. Each point carries its keyword's read time +rather than an export time. String and null values are not stored. + +Exports to the endpoint named by the standard ``OTEL_EXPORTER_OTLP_*`` +variables. Needs the optional dependency: ``pip install libby[otel]``. +""" +from __future__ import annotations + +import socket +from typing import Any, List, Optional, Sequence + +from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter +from opentelemetry.sdk.metrics.export import (Gauge, Metric, MetricExportResult, MetricsData, + NumberDataPoint, ResourceMetrics, ScopeMetrics) +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.util.instrumentation import InstrumentationScope + +from .sink import Sample, SinkWriteError + +METRIC_NAME = "libby_keyword" +SERVICE_NAME = "libby-keygrabber" +DEFAULT_TIMEOUT_S = 10.0 +_SCOPE = InstrumentationScope("libby.keygrabber") +_NANOS_PER_SECOND = 1_000_000_000 + + +def to_point(sample: Sample) -> Optional[NumberDataPoint]: + """Map a sample to a gauge point, or ``None`` if it is not a number. + + A bool becomes 0 or 1, and every number a float, matching what Prometheus + stores. Units are a label only when the keyword declares them. + """ + value = sample.value + if value is None or isinstance(value, str): + return None + attributes = {"group": sample.group, "peer": sample.peer, "keyword": sample.keyword} + if sample.units: + attributes["units"] = sample.units + stamp = sample.timestamp + nanos = (int(stamp.timestamp()) * _NANOS_PER_SECOND + + stamp.microsecond * 1_000) + return NumberDataPoint(attributes=attributes, start_time_unix_nano=0, + time_unix_nano=nanos, value=float(value)) + + +class OtlpSink: + """Exports samples as OTLP gauges, one request per batch.""" + + def __init__(self, timeout_s: float = DEFAULT_TIMEOUT_S, + exporter: Optional[Any] = None) -> None: + self._timeout_s = timeout_s + self._exporter = exporter + self._owns_exporter = exporter is None + # Built by hand: Resource.create would add a random service.instance.id, + # which Prometheus labels, starting new series on every restart + self._resource = Resource({"service.name": SERVICE_NAME, + "host.name": socket.gethostname()}) + + def connect(self) -> None: + """Create the exporter, reading the endpoint from the environment.""" + if self._owns_exporter: + self.close() + self._exporter = OTLPMetricExporter(timeout=self._timeout_s) + + def is_connected(self) -> bool: + """Return whether an exporter exists; OTLP has nothing to ping.""" + return self._exporter is not None + + def write(self, samples: Sequence[Sample]) -> int: + """Export a batch as one request, skipping samples that are not numbers.""" + if self._exporter is None: + raise SinkWriteError("otlp sink is not connected") + points: List[NumberDataPoint] = [ + point for point in map(to_point, samples) if point is not None] + if not points: + return 0 + data = MetricsData(resource_metrics=[ResourceMetrics( + resource=self._resource, schema_url="", + scope_metrics=[ScopeMetrics(scope=_SCOPE, schema_url="", metrics=[Metric( + name=METRIC_NAME, description="libby keyword value", + unit=None, data=Gauge(data_points=points))])], + )]) + try: + result = self._exporter.export(data, timeout_millis=self._timeout_s * 1000) + # The exporter surfaces HTTP and socket errors with no common base, and + # the caller's contract is a single retryable error + except Exception as exc: # pylint: disable=broad-exception-caught + raise SinkWriteError(f"otlp export failed: {exc}") from exc + if result is not MetricExportResult.SUCCESS: + raise SinkWriteError("otlp export was rejected or did not arrive") + return len(points) + + def close(self) -> None: + """Shut the exporter down; safe to call when never connected.""" + exporter = self._exporter + if self._owns_exporter: + self._exporter = None + if exporter is not None and self._owns_exporter: + exporter.shutdown() diff --git a/tests/test_keygrabber_collection.py b/tests/test_keygrabber_collection.py index 41a3305..c9795e5 100644 --- a/tests/test_keygrabber_collection.py +++ b/tests/test_keygrabber_collection.py @@ -13,6 +13,7 @@ def _collection_config(**entry: Any): config = { + "sinks": {"influxdb": {"type": "influxdb"}}, "collections": { "adc": {"peer": "hsfei.adc", "keywords": ["%"], **entry}, }, diff --git a/tests/test_keygrabber_config.py b/tests/test_keygrabber_config.py index 005ee3e..fd77b5a 100644 --- a/tests/test_keygrabber_config.py +++ b/tests/test_keygrabber_config.py @@ -21,6 +21,7 @@ def _config(**overrides: Any) -> Dict[str, Any]: "peer_id": "keygrabber", "group_id": "hispec", "defaults": {"interval_s": 10.0, "timeout_s": 2.0, "refresh_s": 300.0}, + "sinks": {"influxdb": {"type": "influxdb"}}, "collections": { "adc": {"peer": "hsfei.adc", "keywords": ["positionvalue%"]}, }, @@ -183,13 +184,44 @@ def test_no_matches_selects_nothing(self): self.assertEqual(self._selected(["nosuch%"]), ()) +class SinksTests(unittest.TestCase): + """The named sinks section.""" + + def test_every_named_sink_is_kept(self): + """Keep each sink's settings under its name.""" + sinks = {"influxdb": {"type": "influxdb"}, "prometheus": {"type": "otlp"}} + self.assertEqual(parse_config(_config(sinks=sinks)).sinks, sinks) + + def test_the_old_single_sink_key_is_refused(self): + """Point an old config at the new shape rather than ignoring its sink.""" + config = _config(sink={"type": "influxdb"}) + del config["sinks"] + with self.assertRaisesRegex(ConfigError, "'sink' is now 'sinks'"): + parse_config(config) + + def test_at_least_one_sink_is_required(self): + """Refuse a keygrabber that would read keywords and store none.""" + with self.assertRaisesRegex(ConfigError, "at least one sink"): + parse_config(_config(sinks={})) + + def test_sink_names_become_keyword_prefixes(self): + """Refuse a name that could not prefix the sink's status keywords.""" + with self.assertRaisesRegex(ConfigError, "Influx-DB"): + parse_config(_config(sinks={"Influx-DB": {"type": "influxdb"}})) + + class BuildSinkTests(unittest.TestCase): - """Sink construction from the config's sink section.""" + """Sink construction from one named sink's settings.""" + + def test_errors_name_the_sink(self): + """Say which sink is wrong when there are several.""" + with self.assertRaisesRegex(ConfigError, "sinks.archive.type"): + build_sink("archive", {}) def test_token_in_the_config_is_refused(self): """Refuse an inline credential so it cannot reach a config file.""" with self.assertRaises(ConfigError) as caught: - build_sink({"type": "influxdb", "url": "http://x", "org": "o", + build_sink("influxdb", {"type": "influxdb", "url": "http://x", "org": "o", "bucket": "b", "token": "secret"}) self.assertIn("token_env", str(caught.exception)) @@ -197,26 +229,26 @@ def test_unset_token_env_is_reported(self): """Name the missing variable rather than failing at write time.""" with mock.patch.dict(os.environ, {}, clear=True): with self.assertRaises(ConfigError) as caught: - build_sink({"type": "influxdb", "url": "http://x", "org": "o", + build_sink("influxdb", {"type": "influxdb", "url": "http://x", "org": "o", "bucket": "b", "token_env": "NOT_SET_ANYWHERE"}) self.assertIn("NOT_SET_ANYWHERE", str(caught.exception)) def test_missing_type_is_reported(self): """Require an explicit sink type.""" with self.assertRaises(ConfigError): - build_sink({}) + build_sink("influxdb", {}) def test_unknown_type_is_reported(self): """Name the unsupported backend.""" with self.assertRaises(ConfigError) as caught: - build_sink({"type": "sqlite"}) + build_sink("influxdb", {"type": "sqlite"}) self.assertIn("sqlite", str(caught.exception)) def test_missing_connection_fields_are_reported(self): """Require url, org and bucket before reaching the client.""" with mock.patch.dict(os.environ, {"TOKEN": "t"}, clear=True): with self.assertRaises(ConfigError): - build_sink({"type": "influxdb", "token_env": "TOKEN"}) + build_sink("influxdb", {"type": "influxdb", "token_env": "TOKEN"}) if __name__ == "__main__": diff --git a/tests/test_keygrabber_daemon.py b/tests/test_keygrabber_daemon.py index b7f9ed7..14c9f8b 100644 --- a/tests/test_keygrabber_daemon.py +++ b/tests/test_keygrabber_daemon.py @@ -13,7 +13,7 @@ import threading import time import unittest -from typing import List, Sequence +from typing import Any, Dict, List, Mapping, Sequence from libby import Client, KeywordError from libby.daemon import LibbyDaemon @@ -111,17 +111,17 @@ def write(self, samples: Sequence[Sample]) -> int: class _TestKeygrabber(KeygrabberDaemon): - """Keygrabber writing to a recording sink instead of InfluxDB. + """Keygrabber writing to recording sinks instead of InfluxDB. - ``from_config`` constructs a daemon with no arguments, so the sink is - assigned after construction and before ``start``. + ``from_config`` constructs a daemon with no arguments, so the sinks are + assigned after construction and before ``start``, by name. """ - sink: _RecordingSink + sinks: Dict[str, _RecordingSink] - def make_sink(self) -> _RecordingSink: - """Return the injected sink.""" - return self.sink + def make_sink(self, name: str, sink_config: Mapping[str, Any]) -> _RecordingSink: + """Return the injected sink of that name.""" + return self.sinks[name] # Nested inside a plain class so unittest's loader, which collects every @@ -145,7 +145,7 @@ def grabber_config(self) -> dict: def setUp(self) -> None: # pylint: disable=invalid-name self.sink = _RecordingSink() self.grabber = _TestKeygrabber.from_config(self.grabber_config()) - self.grabber.sink = self.sink + self.grabber.sinks = {"recording": self.sink} def tearDown(self) -> None: # pylint: disable=invalid-name self.grabber.stop() @@ -190,8 +190,8 @@ def test_points_written_is_counted(self): """Report what the sink actually stored.""" self.grabber.start() self._await_samples() - self.assertGreater(self.grabber.counters.points_written, 0) - self.assertEqual(self.grabber.counters.write_errors, 0) + self.assertGreater(self.grabber.points_written, 0) + self.assertEqual(self.grabber.write_errors, 0) def test_submit_to_a_shut_down_pool_releases_the_claim(self): """Survive the pool shutting down between a claim and its submit. @@ -305,11 +305,11 @@ def test_a_dead_sink_is_reported_not_hidden(self): Getting this wrong showed a live daemon with queued batches, nothing written, and ``isconnected`` still true. """ - self.grabber.sink = _FailingSink() + self.grabber.sinks = {"recording": _FailingSink()} self.grabber.start() deadline = time.monotonic() + SETTLE_TIMEOUT_S while time.monotonic() < deadline: - if self.grabber.counters.write_errors: + if self.grabber.write_errors: break time.sleep(0.05) @@ -337,7 +337,39 @@ def test_stop_is_clean_with_nothing_left_queued(self): self.grabber.start() self._await_samples() self.grabber.stop() - self.assertEqual(self.grabber.counters.dropped_batches, 0) + self.assertEqual(self.grabber.dropped_batches, 0) + + def test_every_sink_gets_every_sample(self): + """Hand each batch to each sink, from one read.""" + second = _RecordingSink() + self.grabber.sinks = {"recording": self.sink, "second": second} + self.grabber._config["sinks"] = { # pylint: disable=protected-access + "recording": {"type": "recording"}, "second": {"type": "recording"}} + self.grabber.start() + self._await_samples() + self.grabber.stop() + self.assertEqual(sorted(map(repr, second.snapshot())), + sorted(map(repr, self.sink.snapshot()))) + + def test_a_failing_sink_leaves_the_other_writing(self): + """Keep writing to a healthy sink while another one fails. + + The failing sink is listed first, so a shared writer would stall + in front of the healthy one. + """ + self.grabber.sinks = {"failing": _FailingSink(), "recording": self.sink} + self.grabber._config["sinks"] = { # pylint: disable=protected-access + "failing": {"type": "recording"}, "recording": {"type": "recording"}} + self.grabber.start() + self._await_samples() + with self._grabber_client() as client: + self.assertFalse(client.get(self._keyword("isconnected"))) + self.assertFalse(client.get(self._keyword("failing.isconnected"))) + self.assertTrue(client.get(self._keyword("recording.isconnected"))) + self.assertEqual(client.get(self._keyword("failing.pointswritten")), 0) + self.assertGreater(client.get(self._keyword("recording.pointswritten")), 0) + self.assertGreater(client.get(self._keyword("failing.writeerrors")), 0) + self.assertEqual(client.get(self._keyword("recording.writeerrors")), 0) def _collections(peer: str) -> dict: @@ -386,7 +418,7 @@ def grabber_config(self) -> dict: "bind": self.grabber_endpoint, "address_book": {self.target_peer: self.endpoint}, "discovery_enabled": False, - "sink": {"type": "recording"}, + "sinks": {"recording": {"type": "recording"}}, "workers": 2, "collections": _collections(self.target_peer), } @@ -417,7 +449,7 @@ def grabber_config(self) -> dict: "transport": "rabbitmq", "rabbitmq_url": RABBITMQ_URL, "discovery_enabled": False, - "sink": {"type": "recording"}, + "sinks": {"recording": {"type": "recording"}}, "workers": 2, "collections": _collections(self.target_peer), } diff --git a/tests/test_otlp_sink.py b/tests/test_otlp_sink.py new file mode 100644 index 0000000..cb81e9d --- /dev/null +++ b/tests/test_otlp_sink.py @@ -0,0 +1,201 @@ +"""Tests for the keygrabber's OTLP sink: sample mapping, export, and config. + +Needs the optional ``otel`` extra; tox installs it. The end-to-end case also +needs a Collector feeding a Prometheus, and runs only when +``LIBBY_TEST_PROMETHEUS_URL`` names that Prometheus and +``OTEL_EXPORTER_OTLP_ENDPOINT`` the Collector. +""" +from __future__ import annotations + +import importlib.util +import json +import os +import time +import unittest +import urllib.parse +import urllib.request +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any, List, Optional +from unittest import mock + +from libby.config import ConfigError +from libby.keygrabber import Sample, SinkWriteError +from libby.keygrabber.config import build_sink + +# find_spec on the submodule raises when the opentelemetry package itself is absent +HAS_OTEL = (importlib.util.find_spec("opentelemetry") is not None + and importlib.util.find_spec("opentelemetry.sdk") is not None) +PROMETHEUS_URL = os.environ.get("LIBBY_TEST_PROMETHEUS_URL") +ARRIVAL_TIMEOUT_S = 30.0 +READ_AT = datetime(2026, 11, 4, 12, 0, 0, 123456, tzinfo=timezone.utc) + +# Every class below is skipUnless-guarded on HAS_OTEL, which pylint cannot +# see through +# pylint: disable=possibly-used-before-assignment + +if HAS_OTEL: + # pylint: disable=ungrouped-imports + from opentelemetry.sdk.metrics.export import MetricExportResult + + from libby.keygrabber.otlp import METRIC_NAME, OtlpSink, to_point + + +def _sample(keyword: str = "positionvalue", value: Any = 7.5, + units: Optional[str] = "mm", timestamp: datetime = READ_AT) -> Sample: + return Sample(keyword=keyword, group="hsfei", peer="adc", value=value, + units=units, timestamp=timestamp) + + +class _RecordingExporter: + """Stands in for OTLPMetricExporter, keeping what it was asked to export.""" + + def __init__(self, result: Any = None, error: Optional[Exception] = None) -> None: + self.exports: List[Any] = [] + self._result = result + self._error = error + + def export(self, data: Any, timeout_millis: float = 0) -> Any: + """Record one export and answer as configured.""" + del timeout_millis + if self._error is not None: + raise self._error + self.exports.append(data) + return self._result if self._result is not None else MetricExportResult.SUCCESS + + def shutdown(self) -> None: + """Nothing to release.""" + + +@unittest.skipUnless(HAS_OTEL, "needs the otel extra") +class ToPointTests(unittest.TestCase): + """Mapping one sample to one gauge point.""" + + def test_a_float_keeps_its_labels_and_read_time(self): + """Label by group, peer, keyword and units, at the read time to the microsecond.""" + point = to_point(_sample()) + self.assertEqual(point.value, 7.5) + self.assertEqual(dict(point.attributes), { + "group": "hsfei", "peer": "adc", "keyword": "positionvalue", "units": "mm"}) + self.assertEqual(point.time_unix_nano, + int(READ_AT.timestamp()) * 1_000_000_000 + 123_456_000) + + def test_a_bool_becomes_zero_or_one(self): + """Store a flag as a number, as Prometheus can only hold numbers.""" + self.assertEqual(to_point(_sample(value=True)).value, 1.0) + self.assertEqual(to_point(_sample(value=False)).value, 0.0) + + def test_no_units_means_no_units_label(self): + """Leave the label off rather than storing an empty one.""" + self.assertNotIn("units", to_point(_sample(units=None)).attributes) + + def test_strings_and_nulls_are_skipped(self): + """Skip what a gauge cannot hold.""" + self.assertIsNone(to_point(_sample(value="moving"))) + self.assertIsNone(to_point(_sample(value=None))) + + +@unittest.skipUnless(HAS_OTEL, "needs the otel extra") +class WriteTests(unittest.TestCase): + """Exporting a batch.""" + + def test_a_batch_is_one_gauge_with_a_point_per_number(self): + """Export one libby_keyword metric and report the numbers stored.""" + exporter = _RecordingExporter() + sink = OtlpSink(exporter=exporter) + written = sink.write([_sample(), _sample(keyword="status", value="ok"), + _sample(keyword="ismoving", value=False)]) + self.assertEqual(written, 2) + self.assertEqual(len(exporter.exports), 1) + metric = exporter.exports[0].resource_metrics[0].scope_metrics[0].metrics[0] + self.assertEqual(metric.name, METRIC_NAME) + self.assertEqual(len(metric.data.data_points), 2) + + def test_the_resource_has_no_random_instance_id(self): + """Keep series continuous across restarts; an instance id would be a label.""" + exporter = _RecordingExporter() + OtlpSink(exporter=exporter).write([_sample()]) + attributes = exporter.exports[0].resource_metrics[0].resource.attributes + self.assertNotIn("service.instance.id", attributes) + self.assertEqual(attributes["service.name"], "libby-keygrabber") + + def test_a_batch_with_no_numbers_exports_nothing(self): + """Do not send an empty request.""" + exporter = _RecordingExporter() + self.assertEqual(OtlpSink(exporter=exporter).write([_sample(value="ok")]), 0) + self.assertEqual(exporter.exports, []) + + def test_a_rejected_export_is_retryable(self): + """Raise the error RetryingWriter retries on.""" + sink = OtlpSink(exporter=_RecordingExporter(result=MetricExportResult.FAILURE)) + with self.assertRaises(SinkWriteError): + sink.write([_sample()]) + + def test_a_raising_exporter_is_retryable(self): + """Turn a transport error into the same retryable error.""" + sink = OtlpSink(exporter=_RecordingExporter(error=ConnectionError("refused"))) + with self.assertRaisesRegex(SinkWriteError, "refused"): + sink.write([_sample()]) + + +@unittest.skipUnless(HAS_OTEL, "needs the otel extra") +class BuildTests(unittest.TestCase): + """Building an otlp sink from its config.""" + + def test_an_endpoint_is_required(self): + """Refuse to start a sink that would export nowhere.""" + with mock.patch.dict(os.environ, {}, clear=True): + with self.assertRaisesRegex(ConfigError, "OTEL_EXPORTER_OTLP_ENDPOINT"): + build_sink("prometheus", {"type": "otlp"}) + + def test_the_endpoint_from_otel_env_is_enough(self): + """Build from the variable otel.env already sets.""" + with mock.patch.dict(os.environ, {"OTEL_EXPORTER_OTLP_ENDPOINT": "http://b:4318"}, + clear=True): + self.assertIsInstance(build_sink("prometheus", {"type": "otlp"}), OtlpSink) + + def test_a_bad_timeout_names_the_sink(self): + """Say which sink's timeout is wrong.""" + with mock.patch.dict(os.environ, {"OTEL_EXPORTER_OTLP_ENDPOINT": "http://b:4318"}, + clear=True): + with self.assertRaisesRegex(ConfigError, "sinks.prometheus.timeout_s"): + build_sink("prometheus", {"type": "otlp", "timeout_s": 0}) + + +def _query(expr: str) -> List[Any]: + url = f"{PROMETHEUS_URL}/api/v1/query?" + urllib.parse.urlencode({"query": expr}) + with urllib.request.urlopen(url, timeout=10) as response: + return json.load(response)["data"]["result"] + + +@unittest.skipUnless(HAS_OTEL and PROMETHEUS_URL, "needs LIBBY_TEST_PROMETHEUS_URL") +class PrometheusTests(unittest.TestCase): + """A real export through a Collector into Prometheus.""" + + def test_samples_arrive_as_libby_keyword_series(self): + """Query back a fresh sample, and one read a while ago.""" + keyword = f"otlptest{uuid.uuid4().hex[:8]}" + late = datetime.now(timezone.utc) - timedelta(minutes=30) + sink = build_sink("prometheus", {"type": "otlp"}) + sink.connect() + try: + sink.write([_sample(keyword=keyword, value=1.25, + timestamp=datetime.now(timezone.utc))]) + sink.write([_sample(keyword=f"{keyword}late", value=2.5, timestamp=late)]) + finally: + sink.close() + deadline = time.monotonic() + ARRIVAL_TIMEOUT_S + while time.monotonic() < deadline: + fresh = _query(f'{METRIC_NAME}{{keyword="{keyword}"}}') + late_points = _query(f'last_over_time({METRIC_NAME}' + f'{{keyword="{keyword}late"}}[1h])') + if fresh and late_points: + break + time.sleep(1) + self.assertEqual(fresh[0]["value"][1], "1.25") + self.assertEqual(fresh[0]["metric"]["units"], "mm") + self.assertEqual(late_points[0]["value"][1], "2.5") + + +if __name__ == "__main__": + unittest.main() From 8ac7f388038e9bda1c0b23174960001c4d61adda Mon Sep 17 00:00:00 2001 From: Mike Langmayr Date: Fri, 6 Nov 2026 11:31:36 -0800 Subject: [PATCH 2/2] Reject unknown keys in a keygrabber sink and drop the unused Counters.set --- docs/source/keygrabber.md | 3 ++- libby/keygrabber/config.py | 20 +++++++++++++++----- libby/keygrabber/daemon.py | 6 ------ tests/test_keygrabber_config.py | 5 +++++ tests/test_otlp_sink.py | 7 +++++++ 5 files changed, 29 insertions(+), 12 deletions(-) diff --git a/docs/source/keygrabber.md b/docs/source/keygrabber.md index 4a11730..3a39fe4 100644 --- a/docs/source/keygrabber.md +++ b/docs/source/keygrabber.md @@ -78,7 +78,8 @@ 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. +`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 diff --git a/libby/keygrabber/config.py b/libby/keygrabber/config.py index 5a2fba4..1230289 100644 --- a/libby/keygrabber/config.py +++ b/libby/keygrabber/config.py @@ -32,6 +32,12 @@ _COLLECTION_NAME = re.compile(r"^[a-z0-9_]+$") +# Settings each sink type takes; any other key is most likely a typo +_SINK_KEYS = { + "influxdb": frozenset({"type", "url", "org", "bucket", "token_env", "timeout_ms"}), + "otlp": frozenset({"type", "timeout_s"}), +} + # Standard OpenTelemetry variables an otlp sink can export to; otel.env sets the first _OTLP_ENDPOINT_VARIABLES = ("OTEL_EXPORTER_OTLP_ENDPOINT", "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") @@ -112,17 +118,21 @@ def build_sink(name: str, sink_config: Mapping[str, Any]) -> Sink: kind = str(sink_config.get("type", "")).strip().lower() if not kind: raise ConfigError(f"{label}.type is required") - if kind == "otlp": - return _build_otlp_sink(label, sink_config) - if kind != "influxdb": + if kind not in _SINK_KEYS: raise ConfigError( f"unsupported {label}.type {kind!r}; expected 'influxdb' or 'otlp'") - - if "token" in sink_config: + if kind == "influxdb" and "token" in sink_config: raise ConfigError( f"{label}.token must not appear in the config; name an environment " f"variable with {label}.token_env instead" ) + unknown = sorted(set(sink_config) - _SINK_KEYS[kind]) + if unknown: + raise ConfigError(f"{label} has unknown key(s) {', '.join(unknown)}; " + f"{kind} takes {', '.join(sorted(_SINK_KEYS[kind]))}") + if kind == "otlp": + return _build_otlp_sink(label, sink_config) + token_env = _required(sink_config, "token_env", label) token = os.environ.get(str(token_env)) if not token: diff --git a/libby/keygrabber/daemon.py b/libby/keygrabber/daemon.py index f83c87b..a723db8 100644 --- a/libby/keygrabber/daemon.py +++ b/libby/keygrabber/daemon.py @@ -61,12 +61,6 @@ def add(self, **deltas: int) -> None: for name, delta in deltas.items(): setattr(self, name, getattr(self, name) + delta) - def set(self, **values: int) -> None: - """Set one or more counters to an absolute value.""" - with self._lock: - for name, value in values.items(): - setattr(self, name, value) - # Coordinates config, client, sink, collections and three kinds of thread; # the attribute count is collaborators, not state diff --git a/tests/test_keygrabber_config.py b/tests/test_keygrabber_config.py index fd77b5a..bb48b8d 100644 --- a/tests/test_keygrabber_config.py +++ b/tests/test_keygrabber_config.py @@ -213,6 +213,11 @@ def test_sink_names_become_keyword_prefixes(self): class BuildSinkTests(unittest.TestCase): """Sink construction from one named sink's settings.""" + def test_an_unknown_key_is_named(self): + """Refuse a misspelt setting rather than running on its default.""" + with self.assertRaisesRegex(ConfigError, "sinks.archive has unknown key.*bukcet"): + build_sink("archive", {"type": "influxdb", "bukcet": "b"}) + def test_errors_name_the_sink(self): """Say which sink is wrong when there are several.""" with self.assertRaisesRegex(ConfigError, "sinks.archive.type"): diff --git a/tests/test_otlp_sink.py b/tests/test_otlp_sink.py index cb81e9d..46729bd 100644 --- a/tests/test_otlp_sink.py +++ b/tests/test_otlp_sink.py @@ -154,6 +154,13 @@ def test_the_endpoint_from_otel_env_is_enough(self): clear=True): self.assertIsInstance(build_sink("prometheus", {"type": "otlp"}), OtlpSink) + def test_an_unknown_key_is_named(self): + """Refuse a misspelt timeout rather than exporting with the default.""" + with mock.patch.dict(os.environ, {"OTEL_EXPORTER_OTLP_ENDPOINT": "http://b:4318"}, + clear=True): + with self.assertRaisesRegex(ConfigError, "unknown key.*timout_s"): + build_sink("prometheus", {"type": "otlp", "timout_s": 30}) + def test_a_bad_timeout_names_the_sink(self): """Say which sink's timeout is wrong.""" with mock.patch.dict(os.environ, {"OTEL_EXPORTER_OTLP_ENDPOINT": "http://b:4318"},