diff --git a/docs/source/keygrabber.md b/docs/source/keygrabber.md index 4790ec5..3a39fe4 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,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,19 +146,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 +171,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 +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 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..1230289 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,15 @@ _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") + # A config record, so the field count is the schema rather than complexity @dataclass(frozen=True) @@ -64,13 +73,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 +94,64 @@ 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") - if kind != "influxdb": - raise ConfigError(f"unsupported sink type {kind!r}; expected 'influxdb'") - - if "token" in sink_config: + raise ConfigError(f"{label}.type is required") + if kind not in _SINK_KEYS: raise ConfigError( - "sink.token must not appear in the config; name an environment " - "variable with sink.token_env instead" + f"unsupported {label}.type {kind!r}; expected 'influxdb' or 'otlp'") + 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" ) - token_env = _required(sink_config, "token_env", "sink") + 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: 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 +161,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..a723db8 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.""" @@ -72,21 +61,15 @@ 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 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 +81,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 +131,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 +149,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 +179,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 +190,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 +265,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 +296,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 +424,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..bb48b8d 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,49 @@ 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_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"): + 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 +234,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..46729bd --- /dev/null +++ b/tests/test_otlp_sink.py @@ -0,0 +1,208 @@ +"""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_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"}, + 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()