diff --git a/openhop-plugin.json b/openhop-plugin.json index 1fe9471..7161015 100644 --- a/openhop-plugin.json +++ b/openhop-plugin.json @@ -2,7 +2,7 @@ "schema": 1, "id": "community.influxdb", "name": "InfluxDB Exporter", - "version": "0.1.5", + "version": "0.1.6", "description": "Pushes Repeater metrics, per-packet events, logs and node/GPS snapshots to a remote InfluxDB (v1, v2 or v3) for Grafana dashboards.", "runtime": { "type": "python", diff --git a/pyproject.toml b/pyproject.toml index b73771c..153bfd4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "openhop-influxdb-plugin" -version = "0.1.5" +version = "0.1.6" description = "openHop Repeater plugin that pushes metrics, packets, logs and snapshots to InfluxDB" readme = "README.md" requires-python = ">=3.10" diff --git a/src/openhop_influxdb_plugin/__init__.py b/src/openhop_influxdb_plugin/__init__.py index 1276d02..0a8da88 100644 --- a/src/openhop_influxdb_plugin/__init__.py +++ b/src/openhop_influxdb_plugin/__init__.py @@ -1 +1 @@ -__version__ = "0.1.5" +__version__ = "0.1.6" diff --git a/src/openhop_influxdb_plugin/spool.py b/src/openhop_influxdb_plugin/spool.py index e792b21..9b60ec9 100644 --- a/src/openhop_influxdb_plugin/spool.py +++ b/src/openhop_influxdb_plugin/spool.py @@ -1,5 +1,6 @@ from __future__ import annotations +import bisect import gzip import logging import os @@ -15,6 +16,9 @@ class Spool: Batches that cannot be delivered are written here and replayed oldest first. When the size cap is exceeded the oldest batches are discarded. + + The directory is scanned once, at startup; after that the file names and sizes + are tracked in memory, so a large backlog costs nothing to drain or measure. """ def __init__(self, directory: str | Path, max_bytes: int): @@ -23,17 +27,39 @@ class Spool: self._lock = threading.Lock() self._seq = 0 self.dropped_batches = 0 + self._names: list[str] = [] # sorted, so oldest first + self._sizes: dict[str, int] = {} + self._total = 0 if self.enabled: self.directory.mkdir(parents=True, exist_ok=True) + self._load() @property def enabled(self) -> bool: return self.max_bytes > 0 - def _files(self) -> list[Path]: - if not self.directory.exists(): - return [] - return sorted(self.directory.glob("*.lp.gz")) + def _load(self) -> None: + for path in self.directory.glob("*.lp.gz"): + try: + size = path.stat().st_size + except OSError: + continue + self._sizes[path.name] = size + self._total += size + self._names = sorted(self._sizes) + + def _forget(self, name: str) -> None: + size = self._sizes.pop(name, None) + if size is None: + return + self._total -= size + i = bisect.bisect_left(self._names, name) + if i < len(self._names) and self._names[i] == name: + del self._names[i] + + def _discard(self, name: str) -> None: + self._forget(name) + (self.directory / name).unlink(missing_ok=True) def put(self, body: bytes) -> bool: if not self.enabled: @@ -42,37 +68,39 @@ class Spool: with self._lock: self._seq += 1 name = f"{time.time_ns():020d}-{self._seq:06d}.lp.gz" + data = gzip.compress(body) tmp = self.directory / (name + ".tmp") - tmp.write_bytes(gzip.compress(body)) + tmp.write_bytes(data) os.replace(tmp, self.directory / name) + self._sizes[name] = len(data) + self._total += len(data) + bisect.insort(self._names, name) self._enforce_cap() return True def _enforce_cap(self) -> None: - files = self._files() - total = sum(f.stat().st_size for f in files) - while files and total > self.max_bytes: - oldest = files.pop(0) - total -= oldest.stat().st_size - oldest.unlink(missing_ok=True) + while self._names and self._total > self.max_bytes: + oldest = self._names[0] + self._discard(oldest) self.dropped_batches += 1 - logger.warning( - "Spool over %d bytes; discarded oldest batch %s", self.max_bytes, oldest.name - ) + logger.warning("Spool over %d bytes; discarded oldest batch %s", self.max_bytes, oldest) def oldest(self) -> tuple[Path, bytes] | None: with self._lock: - for path in self._files(): + while self._names: + name = self._names[0] + path = self.directory / name try: return path, gzip.decompress(path.read_bytes()) except (OSError, EOFError) as exc: - logger.warning("Discarding unreadable spool file %s: %s", path.name, exc) - path.unlink(missing_ok=True) + logger.warning("Discarding unreadable spool file %s: %s", name, exc) + self._discard(name) return None def remove(self, path: Path) -> None: - path.unlink(missing_ok=True) + with self._lock: + self._discard(path.name) def stats(self) -> tuple[int, int]: - files = self._files() - return len(files), sum(f.stat().st_size for f in files) + with self._lock: + return len(self._names), self._total diff --git a/tests/test_spool.py b/tests/test_spool.py new file mode 100644 index 0000000..6f68dda --- /dev/null +++ b/tests/test_spool.py @@ -0,0 +1,73 @@ +from pathlib import Path + +from openhop_influxdb_plugin.spool import Spool + + +def _drain(spool): + bodies = [] + while (item := spool.oldest()) is not None: + path, body = item + bodies.append(body) + spool.remove(path) + return bodies + + +def test_drains_oldest_first(tmp_path): + spool = Spool(tmp_path, max_bytes=1024 * 1024) + for i in range(5): + spool.put(f"m v={i}".encode()) + assert _drain(spool) == [f"m v={i}".encode() for i in range(5)] + assert spool.stats() == (0, 0) + assert list(tmp_path.glob("*.lp.gz")) == [] + + +def test_reloads_backlog_left_by_previous_run(tmp_path): + first = Spool(tmp_path, max_bytes=1024 * 1024) + for i in range(3): + first.put(f"m v={i}".encode()) + files, size = first.stats() + + second = Spool(tmp_path, max_bytes=1024 * 1024) + assert second.stats() == (files, size) + second.put(b"m v=3") + assert _drain(second) == [f"m v={i}".encode() for i in range(4)] + + +def test_stats_track_puts_and_removes(tmp_path): + spool = Spool(tmp_path, max_bytes=1024 * 1024) + for i in range(4): + spool.put(f"m v={i}".encode()) + on_disk = sum(p.stat().st_size for p in tmp_path.glob("*.lp.gz")) + assert spool.stats() == (4, on_disk) + path, _ = spool.oldest() + spool.remove(path) + on_disk = sum(p.stat().st_size for p in tmp_path.glob("*.lp.gz")) + assert spool.stats() == (3, on_disk) + + +def test_skips_unreadable_and_vanished_files(tmp_path): + spool = Spool(tmp_path, max_bytes=1024 * 1024) + for i in range(3): + spool.put(f"m v={i}".encode()) + names = sorted(p.name for p in tmp_path.glob("*.lp.gz")) + (tmp_path / names[0]).write_bytes(b"not gzip") + (tmp_path / names[1]).unlink() + path, body = spool.oldest() + assert body == b"m v=2" + assert spool.stats()[0] == 1 + assert not (tmp_path / names[0]).exists() + + +def test_large_backlog_is_not_rescanned(tmp_path, monkeypatch): + spool = Spool(tmp_path, max_bytes=100 * 1024 * 1024) + for i in range(500): + spool.put(f"m v={i}".encode()) + + def no_scan(*args, **kwargs): + raise AssertionError("spool directory rescanned") + + monkeypatch.setattr(Path, "glob", no_scan) + monkeypatch.setattr(Path, "iterdir", no_scan) + spool.put(b"m v=500") + assert spool.stats()[0] == 501 + assert len(_drain(spool)) == 501