Import the Repeater's stored history
On first start the plugin now imports everything the Repeater has stored (initial_backfill_hours, default "all"): - packets from /api/bulk_packets, paged by timestamp with no row cap - every advert received, noise-floor samples and CRC error events, which then keep flowing every snapshot interval - RRD aggregates for up to a year, fetched per archive tier Each source resumes from its own cursor and advances it only after a whole window has shipped. Re-imports are idempotent and can be started with --backfill [HOURS|all] or by changing backfill_request_id in the plugin settings. RRD import is retried after transient errors, and its resolution tag comes from the step the Repeater actually returned. Also adds a History row to the dashboard, stored history to the dev fake Repeater, and tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
richandClaude Opus 5.5 committed
1 parent
ebd5e6d149
commit
2da26f899b
15 files changed
+1339
-70
No files matched your search
@@ -9,6 +9,7 @@ openHop Repeater ──REST /api/*────────► metrics (every
|
||||
──WS /ws/packets───► packets (real time, gap backfill)
|
||||
──SSE /api/logs_stream► logs (real time, resumable)
|
||||
──REST /api/*────────► snapshots (every 5 min)
|
||||
──REST history APIs──► history (full import, then every 5 min)
|
||||
│
|
||||
batch + gzip + retry + disk spool
|
||||
▼
|
||||
@@ -32,6 +33,10 @@ several repeaters can share one bucket.
|
||||
| `openhop_advert` | `/api/stats` `neighbors` | `pubkey`, `contact_type`, `is_repeater`, `zero_hop` | `node_name`, `rssi`, `snr`, `advert_count`, `first_seen`, `last_seen`, `last_seen_age_seconds`, `latitude`, `longitude` |
|
||||
| `openhop_gps` | `/api/gps` | `state` | fix, satellites, DOP, speed, `latitude`, `longitude`, `altitude_m` |
|
||||
| `openhop_identities`, `openhop_acl`, `openhop_rooms`, `openhop_db`, `openhop_mqtt`, `openhop_crc_errors`, `openhop_advert_rate_limit` | matching `/api/*` endpoints | `item` for list payloads | every numeric value, flattened (`a_b_c`) |
|
||||
| `openhop_advert_received` | `/api/adverts_by_contact_type` (every advert the Repeater stored) | `pubkey`, `contact_type`, `is_repeater`, `zero_hop`, `new_neighbor` | `node_name`, `rssi`, `snr`, `route_type`, `advert_count`, `first_seen`, `last_seen`, `latitude`, `longitude` |
|
||||
| `openhop_noise_floor` | `/api/noise_floor_history` | — | `noise_floor_dbm` (each stored sample) |
|
||||
| `openhop_crc_error` | `/api/crc_error_history` | — | `count` (each stored CRC error event) |
|
||||
| `openhop_rrd` | `/api/rrd_data` (imported once) | `resolution` (`1m`, `5m`, `1h`) | `rx_count_per_second`, `tx_count_per_second`, `drop_count_per_second`, `type_N_per_second`, `avg_rssi`, `avg_snr`, `avg_length`, `avg_score`, `neighbor_count` |
|
||||
| `openhop_exporter` | the plugin itself | `stream` | lines written, errors, spool size, connection state |
|
||||
|
||||
The metric families come from the official
|
||||
@@ -140,11 +145,13 @@ plugin manager runs in the same container.
|
||||
| `influx_username` / `influx_password` | `""` | v1 basic auth |
|
||||
| `influx_create_database` | `true` | v1 only: run `CREATE DATABASE` if the database is missing (needs admin) |
|
||||
| `influx_verify_tls` | `true` | Verify InfluxDB TLS certificates |
|
||||
| `stream_metrics` / `stream_packets` / `stream_logs` / `stream_snapshots` | `true` | Enable each stream |
|
||||
| `stream_metrics` / `stream_packets` / `stream_logs` / `stream_snapshots` / `stream_history` | `true` | Enable each stream |
|
||||
| `metrics_interval_seconds` | `30` | Metrics poll interval (5–3600) |
|
||||
| `snapshot_interval_seconds` | `300` | Snapshot poll interval (30–86400) |
|
||||
| `request_timeout_seconds` | `10` | HTTP timeout for Repeater and InfluxDB calls |
|
||||
| `initial_backfill_hours` | `24` | On first start, also ship stored packets from this many hours back (0 disables) |
|
||||
| `initial_backfill_hours` | `"all"` | On first start, import the Repeater's stored history: `"all"`, a number of hours, or `0` for none |
|
||||
| `backfill_hours` | `"all"` | Window for a manual re-import (`--backfill` default, `backfill_request_id`) |
|
||||
| `backfill_request_id` | `""` | Set to any new value and save with restart to re-import `backfill_hours` of history once |
|
||||
| `log_min_level` | `INFO` | Minimum Repeater log level to ship |
|
||||
| `include_payload` / `include_raw_packet` | `false` | Ship packet payload hex / raw packet hex |
|
||||
| `include_locations` | `true` | Ship GPS and advert coordinates |
|
||||
@@ -154,6 +161,40 @@ plugin manager runs in the same container.
|
||||
|
||||
Restart the plugin after changing settings (the Settings dialog can do this).
|
||||
|
||||
## Importing history
|
||||
|
||||
On its **first start** the plugin imports everything the Repeater has stored
|
||||
(`initial_backfill_hours`, default `"all"`), alongside the live streams:
|
||||
|
||||
| Stored on the Repeater | How far back | Imported as |
|
||||
| --- | --- | --- |
|
||||
| Every packet (SQLite) | `storage.retention.sqlite_cleanup_days` (default 31 days) | `openhop_packet` |
|
||||
| Every advert received | same retention | `openhop_advert_received` |
|
||||
| Noise-floor samples | same retention | `openhop_noise_floor` |
|
||||
| CRC error events | same retention | `openhop_crc_error` |
|
||||
| Aggregated traffic/RF (RRD) | 1-min steps for 7 days, 5-min for 30 days, hourly for **1 year** | `openhop_rrd` |
|
||||
| Logs | only the in-memory buffer | `openhop_log` (sent on connect) |
|
||||
|
||||
Counters such as total RX/forwarded have no history on the Repeater; their series
|
||||
start when the plugin starts (the RRD import covers the past year at lower resolution).
|
||||
|
||||
After the import, adverts, noise floor and CRC errors keep flowing every
|
||||
`snapshot_interval_seconds`, and packets keep flowing live. Each source resumes
|
||||
from its own cursor in `state-influxdb.json` and only advances it after a whole
|
||||
window has shipped, so an interrupted import is redone rather than left with a gap.
|
||||
Packets are paged by timestamp, so even a month of history on a Pi imports with
|
||||
bounded memory.
|
||||
|
||||
**Re-importing** is safe at any time: points keep their original timestamps and
|
||||
tags, so InfluxDB overwrites them instead of duplicating. Two ways:
|
||||
|
||||
- From the Repeater UI (no shell needed): in the plugin's **Settings**, set
|
||||
`backfill_request_id` to any new value (for example today's date), optionally set
|
||||
`backfill_hours`, and save with restart. The import runs once in the background;
|
||||
`GET /api/plugins/runtime?id=community.influxdb` shows `backfill` → `done`.
|
||||
- From a shell: `openhop-influxdb --backfill` (everything) or `--backfill 72`
|
||||
(last 72 hours). It flushes and exits; the running service's cursors are untouched.
|
||||
|
||||
## Reliability
|
||||
|
||||
- **Independent streams.** Each stream runs in its own thread with exponential
|
||||
@@ -179,6 +220,7 @@ Restart the plugin after changing settings (the Settings dialog can do this).
|
||||
openhop-influxdb --check-config # validate config.json, print it redacted
|
||||
openhop-influxdb --once --dry-run # one collection cycle, print line protocol
|
||||
openhop-influxdb --once # one collection cycle, write to InfluxDB
|
||||
openhop-influxdb --backfill [HOURS|all] # import stored Repeater history, then exit
|
||||
openhop-influxdb # run as a service (what the plugin manager does)
|
||||
```
|
||||
|
||||
@@ -221,8 +263,11 @@ WebSocket and the `/api/logs_stream` SSE feed with live synthetic data:
|
||||
# then use "repeater_url": "http://127.0.0.1:8765", "repeater_api_token": "dev-token"
|
||||
```
|
||||
|
||||
Note: a `--dry-run` never saves the packet resume point, so a real run afterwards
|
||||
still performs the full `initial_backfill_hours` backfill.
|
||||
The fake keeps 35 days of packets, 30 days of adverts, 31 days of noise samples and
|
||||
a year of RRD data, so the history import can be exercised end to end.
|
||||
|
||||
Note: `--dry-run` and `--backfill` never save resume points, so a later service run
|
||||
still performs its own first-start import.
|
||||
|
||||
## Licence
|
||||
|
||||
|
||||
+4
-1
@@ -18,10 +18,13 @@
|
||||
"stream_packets": true,
|
||||
"stream_logs": true,
|
||||
"stream_snapshots": true,
|
||||
"stream_history": true,
|
||||
"metrics_interval_seconds": 30,
|
||||
"snapshot_interval_seconds": 300,
|
||||
"request_timeout_seconds": 10,
|
||||
"initial_backfill_hours": 24,
|
||||
"initial_backfill_hours": "all",
|
||||
"backfill_hours": "all",
|
||||
"backfill_request_id": "",
|
||||
"log_min_level": "INFO",
|
||||
"include_payload": false,
|
||||
"include_raw_packet": false,
|
||||
|
||||
@@ -1080,7 +1080,7 @@
|
||||
},
|
||||
{
|
||||
"type": "row",
|
||||
"title": "Exporter health",
|
||||
"title": "History (imported from the Repeater, continued live)",
|
||||
"id": 30,
|
||||
"collapsed": false,
|
||||
"gridPos": {
|
||||
@@ -1093,17 +1093,207 @@
|
||||
},
|
||||
{
|
||||
"type": "timeseries",
|
||||
"title": "Lines written per minute",
|
||||
"title": "Long-term traffic (RRD, packets/min)",
|
||||
"id": 31,
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"gridPos": {
|
||||
"h": 8,
|
||||
"w": 12,
|
||||
"x": 0,
|
||||
"y": 91
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"query": "SELECT mean(\"rx_count_per_second\") * 60 AS \"received\", mean(\"tx_count_per_second\") * 60 AS \"transmitted\", mean(\"drop_count_per_second\") * 60 AS \"dropped\" FROM \"openhop_rrd\" WHERE \"repeater\" =~ /^$repeater$/ AND $timeFilter GROUP BY time($__interval) fill(none)",
|
||||
"rawQuery": true,
|
||||
"refId": "A",
|
||||
"resultFormat": "time_series"
|
||||
}
|
||||
],
|
||||
"fieldConfig": {
|
||||
"defaults": {},
|
||||
"overrides": []
|
||||
},
|
||||
"options": {}
|
||||
},
|
||||
{
|
||||
"type": "timeseries",
|
||||
"title": "Long-term RF (RRD)",
|
||||
"id": 32,
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"gridPos": {
|
||||
"h": 8,
|
||||
"w": 12,
|
||||
"x": 12,
|
||||
"y": 91
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"query": "SELECT mean(\"avg_rssi\") AS \"avg RSSI (dBm)\", mean(\"avg_snr\") AS \"avg SNR (dB)\" FROM \"openhop_rrd\" WHERE \"repeater\" =~ /^$repeater$/ AND $timeFilter GROUP BY time($__interval) fill(none)",
|
||||
"rawQuery": true,
|
||||
"refId": "A",
|
||||
"resultFormat": "time_series"
|
||||
}
|
||||
],
|
||||
"fieldConfig": {
|
||||
"defaults": {},
|
||||
"overrides": []
|
||||
},
|
||||
"options": {}
|
||||
},
|
||||
{
|
||||
"type": "timeseries",
|
||||
"title": "Noise floor samples",
|
||||
"id": 33,
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"gridPos": {
|
||||
"h": 8,
|
||||
"w": 8,
|
||||
"x": 0,
|
||||
"y": 91
|
||||
"y": 99
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"query": "SELECT mean(\"noise_floor_dbm\") AS \"noise floor\" FROM \"openhop_noise_floor\" WHERE \"repeater\" =~ /^$repeater$/ AND $timeFilter GROUP BY time($__interval) fill(none)",
|
||||
"rawQuery": true,
|
||||
"refId": "A",
|
||||
"resultFormat": "time_series"
|
||||
}
|
||||
],
|
||||
"fieldConfig": {
|
||||
"defaults": {
|
||||
"unit": "dBm"
|
||||
},
|
||||
"overrides": []
|
||||
},
|
||||
"options": {}
|
||||
},
|
||||
{
|
||||
"type": "timeseries",
|
||||
"title": "Adverts received by contact type",
|
||||
"id": 34,
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"gridPos": {
|
||||
"h": 8,
|
||||
"w": 8,
|
||||
"x": 8,
|
||||
"y": 99
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"query": "SELECT count(\"last_seen\") FROM \"openhop_advert_received\" WHERE \"repeater\" =~ /^$repeater$/ AND $timeFilter GROUP BY time($__interval), \"contact_type\" fill(0)",
|
||||
"rawQuery": true,
|
||||
"refId": "A",
|
||||
"resultFormat": "time_series",
|
||||
"alias": "$tag_contact_type"
|
||||
}
|
||||
],
|
||||
"fieldConfig": {
|
||||
"defaults": {
|
||||
"custom": {
|
||||
"drawStyle": "bars",
|
||||
"stacking": {
|
||||
"mode": "normal"
|
||||
},
|
||||
"fillOpacity": 80
|
||||
}
|
||||
},
|
||||
"overrides": []
|
||||
},
|
||||
"options": {}
|
||||
},
|
||||
{
|
||||
"type": "timeseries",
|
||||
"title": "CRC errors",
|
||||
"id": 35,
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"gridPos": {
|
||||
"h": 8,
|
||||
"w": 8,
|
||||
"x": 16,
|
||||
"y": 99
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"query": "SELECT sum(\"count\") FROM \"openhop_crc_error\" WHERE \"repeater\" =~ /^$repeater$/ AND $timeFilter GROUP BY time($__interval) fill(0)",
|
||||
"rawQuery": true,
|
||||
"refId": "A",
|
||||
"resultFormat": "time_series",
|
||||
"alias": "CRC errors"
|
||||
}
|
||||
],
|
||||
"fieldConfig": {
|
||||
"defaults": {
|
||||
"custom": {
|
||||
"drawStyle": "bars",
|
||||
"fillOpacity": 80
|
||||
}
|
||||
},
|
||||
"overrides": []
|
||||
},
|
||||
"options": {}
|
||||
},
|
||||
{
|
||||
"type": "row",
|
||||
"title": "Exporter health",
|
||||
"id": 36,
|
||||
"collapsed": false,
|
||||
"gridPos": {
|
||||
"h": 1,
|
||||
"w": 24,
|
||||
"x": 0,
|
||||
"y": 107
|
||||
},
|
||||
"panels": []
|
||||
},
|
||||
{
|
||||
"type": "timeseries",
|
||||
"title": "Lines written per minute",
|
||||
"id": 37,
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
},
|
||||
"gridPos": {
|
||||
"h": 8,
|
||||
"w": 8,
|
||||
"x": 0,
|
||||
"y": 108
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
@@ -1127,7 +1317,7 @@
|
||||
{
|
||||
"type": "timeseries",
|
||||
"title": "Spool backlog",
|
||||
"id": 32,
|
||||
"id": 38,
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
@@ -1136,7 +1326,7 @@
|
||||
"h": 8,
|
||||
"w": 8,
|
||||
"x": 8,
|
||||
"y": 91
|
||||
"y": 108
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
@@ -1162,7 +1352,7 @@
|
||||
{
|
||||
"type": "timeseries",
|
||||
"title": "Stream connections",
|
||||
"id": 33,
|
||||
"id": 39,
|
||||
"datasource": {
|
||||
"type": "influxdb",
|
||||
"uid": "${datasource}"
|
||||
@@ -1171,7 +1361,7 @@
|
||||
"h": 8,
|
||||
"w": 8,
|
||||
"x": 16,
|
||||
"y": 91
|
||||
"y": 108
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
|
||||
+151
-9
@@ -66,6 +66,10 @@ class FakeRepeater:
|
||||
self.log_waiters: list[asyncio.Queue] = []
|
||||
self.packet_id = 0
|
||||
self.log_id = 0
|
||||
self.max_packets = 50_000
|
||||
self.advert_rows: list[dict] = []
|
||||
self.noise_rows: list[dict] = []
|
||||
self.crc_rows: list[dict] = []
|
||||
|
||||
# -- auth ------------------------------------------------------------------
|
||||
|
||||
@@ -137,7 +141,29 @@ class FakeRepeater:
|
||||
"lbt_channel_busy": random.random() < 0.1,
|
||||
}
|
||||
self.packets.append(packet)
|
||||
del self.packets[:-5000]
|
||||
if ptype == 4 and self.advert_rows:
|
||||
pubkey = random.choice(list(NODES))
|
||||
prev = next(a for a in reversed(self.advert_rows) if a["pubkey"] == pubkey)
|
||||
self.advert_rows.append(
|
||||
dict(
|
||||
prev,
|
||||
id=len(self.advert_rows) + 1,
|
||||
timestamp=packet["timestamp"],
|
||||
last_seen=packet["timestamp"],
|
||||
rssi=rssi,
|
||||
snr=snr,
|
||||
advert_count=prev["advert_count"] + 1,
|
||||
is_new_neighbor=False,
|
||||
)
|
||||
)
|
||||
if self.noise_rows and random.random() < 0.2:
|
||||
self.noise_rows.append(
|
||||
{
|
||||
"timestamp": packet["timestamp"],
|
||||
"noise_floor_dbm": round(random.gauss(-112, 1.5), 1),
|
||||
}
|
||||
)
|
||||
del self.packets[: -self.max_packets]
|
||||
return packet
|
||||
|
||||
def make_log(self, message: str | None = None, level: str | None = None) -> dict:
|
||||
@@ -147,7 +173,8 @@ class FakeRepeater:
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"level": level or random.choice(["INFO"] * 6 + ["DEBUG", "WARNING", "ERROR"]),
|
||||
"logger": random.choice(["Engine", "PacketRouter", "HTTPServer", "Radio"]),
|
||||
"message": message or random.choice(
|
||||
"message": message
|
||||
or random.choice(
|
||||
[
|
||||
"Forwarded flood packet",
|
||||
"Duplicate packet suppressed",
|
||||
@@ -161,11 +188,50 @@ class FakeRepeater:
|
||||
del self.logs[:-1000]
|
||||
return entry
|
||||
|
||||
async def producer(self) -> None:
|
||||
for _ in range(300): # history for backfill
|
||||
p = self.make_packet()
|
||||
p["timestamp"] -= random.uniform(60, 3600)
|
||||
def seed_history(self, packets: int = 20_000, days: int = 35) -> None:
|
||||
"""Stored history like a Repeater that has been running for a while."""
|
||||
now = time.time()
|
||||
for _ in range(packets):
|
||||
packet = self.make_packet()
|
||||
packet["timestamp"] = now - random.uniform(60, days * 86400)
|
||||
packet["payload"] = packet["raw_packet"] = None # keep memory small
|
||||
self.packets.sort(key=lambda p: p["timestamp"])
|
||||
for i in range(len(self.packets)):
|
||||
self.packets[i]["id"] = i + 1
|
||||
for pubkey, node in NODES.items():
|
||||
first = now - 30 * 86400
|
||||
ts, count = first, 0
|
||||
while ts < now:
|
||||
count += 1
|
||||
self.advert_rows.append(
|
||||
{
|
||||
"id": len(self.advert_rows) + 1,
|
||||
"timestamp": ts,
|
||||
"pubkey": pubkey,
|
||||
**node,
|
||||
"route_type": 1,
|
||||
"first_seen": first,
|
||||
"last_seen": ts,
|
||||
"rssi": round(random.gauss(-95, 6)),
|
||||
"snr": round(random.gauss(5, 3), 2),
|
||||
"advert_count": count,
|
||||
"is_new_neighbor": count == 1,
|
||||
"zero_hop": node["is_repeater"],
|
||||
}
|
||||
)
|
||||
ts += random.uniform(1, 3) * 3600
|
||||
ts = now - 31 * 86400
|
||||
while ts < now:
|
||||
self.noise_rows.append(
|
||||
{"timestamp": ts, "noise_floor_dbm": round(random.gauss(-112, 1.5), 1)}
|
||||
)
|
||||
ts += 300
|
||||
self.crc_rows = sorted(
|
||||
({"timestamp": now - random.uniform(0, 31 * 86400), "count": 1} for _ in range(60)),
|
||||
key=lambda c: c["timestamp"],
|
||||
)
|
||||
|
||||
async def producer(self) -> None:
|
||||
while True:
|
||||
await asyncio.sleep(random.uniform(0.3, 1.5))
|
||||
packet = self.make_packet()
|
||||
@@ -311,9 +377,12 @@ class FakeRepeater:
|
||||
limit = min(int(q.get("limit", 1000)), 10000)
|
||||
offset = int(q.get("offset", 0))
|
||||
start = float(q.get("start_timestamp") or 0)
|
||||
rows = [p for p in reversed(self.packets) if p["timestamp"] >= start]
|
||||
out = [{k: v for k, v in p.items() if k not in {"raw_packet", "airtime_ms"}}
|
||||
for p in rows[offset : offset + limit]]
|
||||
end = float(q.get("end_timestamp") or 1e12)
|
||||
rows = [p for p in reversed(self.packets) if start <= p["timestamp"] <= end]
|
||||
out = [
|
||||
{k: v for k, v in p.items() if k not in {"raw_packet", "airtime_ms"}}
|
||||
for p in rows[offset : offset + limit]
|
||||
]
|
||||
for row in out: # the DB returns sqlite ints and JSON-encoded paths
|
||||
row["transmitted"] = int(row["transmitted"])
|
||||
row["is_duplicate"] = int(row["is_duplicate"])
|
||||
@@ -415,6 +484,74 @@ class FakeRepeater:
|
||||
return web.json_response({"error": "not found"}, status=404)
|
||||
return self.ok(data)
|
||||
|
||||
@staticmethod
|
||||
def _cutoff(request) -> float:
|
||||
return time.time() - int(request.query.get("hours", 24)) * 3600
|
||||
|
||||
async def adverts_by_contact_type(self, request):
|
||||
q = request.query
|
||||
if not q.get("contact_type"):
|
||||
return web.json_response(
|
||||
{"success": False, "error": "contact_type parameter is required"}
|
||||
)
|
||||
cutoff = self._cutoff(request)
|
||||
rows = [
|
||||
a
|
||||
for a in reversed(self.advert_rows)
|
||||
if a["contact_type"] == q["contact_type"] and a["timestamp"] > cutoff
|
||||
]
|
||||
offset, limit = int(q.get("offset", 0)), int(q.get("limit", 500))
|
||||
return self.ok(rows[offset : offset + limit], count=len(rows))
|
||||
|
||||
async def noise_floor_history(self, request):
|
||||
q = request.query
|
||||
cutoff = self._cutoff(request)
|
||||
rows = [n for n in reversed(self.noise_rows) if n["timestamp"] > cutoff]
|
||||
offset, limit = int(q.get("offset", 0)), int(q.get("limit", 2000))
|
||||
page = rows[offset : offset + limit][::-1]
|
||||
return self.ok({"history": page, "count": len(page), "offset": offset, "limit": limit})
|
||||
|
||||
async def crc_error_history(self, request):
|
||||
cutoff = self._cutoff(request)
|
||||
limit = int(request.query.get("limit") or 1000)
|
||||
rows = [c for c in self.crc_rows if c["timestamp"] > cutoff][-limit:]
|
||||
return self.ok({"history": rows, "count": len(rows)})
|
||||
|
||||
async def rrd_data(self, request):
|
||||
"""rrdtool fetch semantics: the finest archive covering the requested range."""
|
||||
now = int(time.time())
|
||||
start = int(request.query.get("start_time") or now - 86400)
|
||||
end = int(request.query.get("end_time") or now)
|
||||
age = now - start
|
||||
step = 60 if age <= 7 * 86400 else 300 if age <= 30 * 86400 else 3600
|
||||
stamps = list(range(start - start % step + step, end + 1, step))
|
||||
rx = [round(random.uniform(0.05, 0.4), 4) for _ in stamps]
|
||||
metrics = {
|
||||
"rx_count": rx,
|
||||
"tx_count": [round(v * 0.6, 4) for v in rx],
|
||||
"drop_count": [round(v * 0.1, 4) for v in rx],
|
||||
"avg_rssi": [round(random.gauss(-97, 3), 2) for _ in stamps],
|
||||
"avg_snr": [round(random.gauss(4, 2), 2) for _ in stamps],
|
||||
"avg_length": [round(random.uniform(40, 120), 1) for _ in stamps],
|
||||
"avg_score": [round(random.random(), 3) for _ in stamps],
|
||||
"neighbor_count": [6.0 for _ in stamps],
|
||||
}
|
||||
types = {f"type_{t}": [round(v / 8, 4) for v in rx] for t in PACKET_TYPES}
|
||||
if stamps:
|
||||
for values in [*metrics.values(), *types.values()]:
|
||||
values[-1] = None # the newest RRD row is usually still unknown
|
||||
return self.ok(
|
||||
{
|
||||
"start_time": start,
|
||||
"end_time": end,
|
||||
"step": step,
|
||||
"timestamps": stamps,
|
||||
"metrics": metrics,
|
||||
"packet_types": types,
|
||||
"data_source": "rrd",
|
||||
}
|
||||
)
|
||||
|
||||
async def ws_packets(self, request):
|
||||
ws = web.WebSocketResponse()
|
||||
await ws.prepare(request)
|
||||
@@ -467,6 +604,7 @@ class FakeRepeater:
|
||||
|
||||
async def main(port: int, token: str) -> None:
|
||||
fake = FakeRepeater(token)
|
||||
fake.seed_history()
|
||||
for _ in range(40):
|
||||
fake.make_log()
|
||||
app = web.Application(middlewares=[fake.auth_middleware])
|
||||
@@ -480,6 +618,10 @@ async def main(port: int, token: str) -> None:
|
||||
web.get("/api/noise_floor_stats", fake.noise),
|
||||
web.get("/api/neighbor_links", fake.neighbor_links),
|
||||
web.get("/api/logs_stream", fake.logs_stream),
|
||||
web.get("/api/adverts_by_contact_type", fake.adverts_by_contact_type),
|
||||
web.get("/api/noise_floor_history", fake.noise_floor_history),
|
||||
web.get("/api/crc_error_history", fake.crc_error_history),
|
||||
web.get("/api/rrd_data", fake.rrd_data),
|
||||
web.get("/api/{name}", fake.generic),
|
||||
web.get("/ws/packets", fake.ws_packets),
|
||||
]
|
||||
|
||||
+4
-1
@@ -29,10 +29,13 @@
|
||||
"stream_packets": true,
|
||||
"stream_logs": true,
|
||||
"stream_snapshots": true,
|
||||
"stream_history": true,
|
||||
"metrics_interval_seconds": 30,
|
||||
"snapshot_interval_seconds": 300,
|
||||
"request_timeout_seconds": 10,
|
||||
"initial_backfill_hours": 24,
|
||||
"initial_backfill_hours": "all",
|
||||
"backfill_hours": "all",
|
||||
"backfill_request_id": "",
|
||||
"log_min_level": "INFO",
|
||||
"include_payload": false,
|
||||
"include_raw_packet": false,
|
||||
|
||||
@@ -197,6 +197,32 @@ panel("table", "Repeater log", [target(
|
||||
fmt="table")], 0, 24, 12)
|
||||
|
||||
# -- Exporter ------------------------------------------------------------------
|
||||
row("History (imported from the Repeater, continued live)")
|
||||
panel("timeseries", "Long-term traffic (RRD, packets/min)", [target(
|
||||
'SELECT mean("rx_count_per_second") * 60 AS "received", '
|
||||
'mean("tx_count_per_second") * 60 AS "transmitted", '
|
||||
'mean("drop_count_per_second") * 60 AS "dropped" '
|
||||
f'FROM "openhop_rrd" WHERE {R} AND $timeFilter GROUP BY time($__interval) fill(none)')],
|
||||
0, 12)
|
||||
panel("timeseries", "Long-term RF (RRD)", [target(
|
||||
'SELECT mean("avg_rssi") AS "avg RSSI (dBm)", mean("avg_snr") AS "avg SNR (dB)" '
|
||||
f'FROM "openhop_rrd" WHERE {R} AND $timeFilter GROUP BY time($__interval) fill(none)')],
|
||||
12, 12)
|
||||
panel("timeseries", "Noise floor samples", [target(
|
||||
'SELECT mean("noise_floor_dbm") AS "noise floor" FROM "openhop_noise_floor" '
|
||||
f"WHERE {R} AND $timeFilter GROUP BY time($__interval) fill(none)")], 0, 8, unit="dBm")
|
||||
panel("timeseries", "Adverts received by contact type", [target(
|
||||
f'SELECT count("last_seen") FROM "openhop_advert_received" WHERE {R} AND $timeFilter '
|
||||
'GROUP BY time($__interval), "contact_type" fill(0)', alias="$tag_contact_type")],
|
||||
8, 8, extra={"fieldConfig": {"defaults": {"custom": {
|
||||
"drawStyle": "bars", "stacking": {"mode": "normal"}, "fillOpacity": 80}},
|
||||
"overrides": []}})
|
||||
panel("timeseries", "CRC errors", [target(
|
||||
f'SELECT sum("count") FROM "openhop_crc_error" WHERE {R} AND $timeFilter '
|
||||
"GROUP BY time($__interval) fill(0)", alias="CRC errors")], 16, 8,
|
||||
extra={"fieldConfig": {"defaults": {"custom": {"drawStyle": "bars", "fillOpacity": 80}},
|
||||
"overrides": []}})
|
||||
|
||||
row("Exporter health")
|
||||
panel("timeseries", "Lines written per minute", [target(
|
||||
'SELECT non_negative_derivative(last("lines_written"), 1m) FROM "openhop_exporter" '
|
||||
|
||||
@@ -26,10 +26,13 @@ DEFAULT_CONFIG: dict[str, Any] = {
|
||||
"stream_packets": True,
|
||||
"stream_logs": True,
|
||||
"stream_snapshots": True,
|
||||
"stream_history": True,
|
||||
"metrics_interval_seconds": 30,
|
||||
"snapshot_interval_seconds": 300,
|
||||
"request_timeout_seconds": 10,
|
||||
"initial_backfill_hours": 24,
|
||||
"initial_backfill_hours": "all",
|
||||
"backfill_hours": "all",
|
||||
"backfill_request_id": "",
|
||||
"log_min_level": "INFO",
|
||||
"include_payload": False,
|
||||
"include_raw_packet": False,
|
||||
@@ -71,11 +74,14 @@ class PluginConfig:
|
||||
stream_packets: bool
|
||||
stream_logs: bool
|
||||
stream_snapshots: bool
|
||||
stream_history: bool
|
||||
|
||||
metrics_interval_seconds: int
|
||||
snapshot_interval_seconds: int
|
||||
request_timeout_seconds: float
|
||||
initial_backfill_hours: int
|
||||
initial_backfill_hours: int | None # None = everything the Repeater still has
|
||||
backfill_hours: int | None
|
||||
backfill_request_id: str
|
||||
log_min_level: str
|
||||
|
||||
include_payload: bool
|
||||
@@ -144,6 +150,21 @@ def _float(value: Any, name: str, low: float, high: float) -> float:
|
||||
return number
|
||||
|
||||
|
||||
MAX_BACKFILL_HOURS = 24 * 3650
|
||||
|
||||
|
||||
def parse_hours(value: Any, name: str = "hours") -> int | None:
|
||||
"""Hours of history to import: an integer, or "all" (returned as None)."""
|
||||
if isinstance(value, str) and value.strip().lower() == "all":
|
||||
return None
|
||||
if isinstance(value, bool):
|
||||
raise ConfigError(f'{name} must be a number of hours or "all"')
|
||||
try:
|
||||
return _int(value, name, 0, MAX_BACKFILL_HOURS)
|
||||
except ConfigError as exc:
|
||||
raise ConfigError(f'{name} must be 0-{MAX_BACKFILL_HOURS} hours or "all"') from exc
|
||||
|
||||
|
||||
def _str(value: Any) -> str:
|
||||
return str(value if value is not None else "").strip()
|
||||
|
||||
@@ -186,6 +207,7 @@ def parse_config(raw: dict[str, Any]) -> PluginConfig:
|
||||
stream_packets=_bool(cfg["stream_packets"], "stream_packets"),
|
||||
stream_logs=_bool(cfg["stream_logs"], "stream_logs"),
|
||||
stream_snapshots=_bool(cfg["stream_snapshots"], "stream_snapshots"),
|
||||
stream_history=_bool(cfg["stream_history"], "stream_history"),
|
||||
metrics_interval_seconds=_int(
|
||||
cfg["metrics_interval_seconds"], "metrics_interval_seconds", 5, 3600
|
||||
),
|
||||
@@ -195,9 +217,9 @@ def parse_config(raw: dict[str, Any]) -> PluginConfig:
|
||||
request_timeout_seconds=_float(
|
||||
cfg["request_timeout_seconds"], "request_timeout_seconds", 1, 120
|
||||
),
|
||||
initial_backfill_hours=_int(
|
||||
cfg["initial_backfill_hours"], "initial_backfill_hours", 0, 24 * 90
|
||||
),
|
||||
initial_backfill_hours=parse_hours(cfg["initial_backfill_hours"], "initial_backfill_hours"),
|
||||
backfill_hours=parse_hours(cfg["backfill_hours"], "backfill_hours"),
|
||||
backfill_request_id=_str(cfg["backfill_request_id"]),
|
||||
log_min_level=level,
|
||||
include_payload=_bool(cfg["include_payload"], "include_payload"),
|
||||
include_raw_packet=_bool(cfg["include_raw_packet"], "include_raw_packet"),
|
||||
|
||||
@@ -11,13 +11,14 @@ import time
|
||||
from pathlib import Path
|
||||
|
||||
from . import __version__
|
||||
from .config import ConfigError, PluginConfig, load_config
|
||||
from .config import ConfigError, PluginConfig, load_config, parse_hours
|
||||
from .lineprotocol import Point, now_ns
|
||||
from .repeater_api import RepeaterAPI, RepeaterAPIError, resolve_repeater_tag
|
||||
from .spool import Spool
|
||||
from .state import PersistentState
|
||||
from .status import Status, write_runtime_json
|
||||
from .streams.base import StreamContext, StreamWorker
|
||||
from .streams.history import BackfillRequestWorker, HistoryStream, run_backfill
|
||||
from .streams.logs import LogStream
|
||||
from .streams.metrics import MetricsStream
|
||||
from .streams.packets import PacketStream
|
||||
@@ -50,6 +51,13 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
|
||||
action="store_true",
|
||||
help="collect metrics + snapshots once (and backfill packets), flush, and exit",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--backfill",
|
||||
nargs="?",
|
||||
const="all",
|
||||
metavar="HOURS|all",
|
||||
help="import all stored Repeater history (or the last HOURS), flush, and exit",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
@@ -134,6 +142,8 @@ def build_streams(ctx: StreamContext, *, once: bool) -> list[StreamWorker]:
|
||||
streams.append(MetricsStream(ctx))
|
||||
if cfg.stream_snapshots:
|
||||
streams.append(SnapshotStream(ctx))
|
||||
if cfg.stream_history:
|
||||
streams.append(HistoryStream(ctx))
|
||||
if not once:
|
||||
if cfg.stream_packets:
|
||||
streams.append(PacketStream(ctx))
|
||||
@@ -158,6 +168,12 @@ def run_once(ctx: StreamContext) -> int:
|
||||
except Exception as exc:
|
||||
failures += 1
|
||||
logger.error("snapshots: %s", exc)
|
||||
if cfg.stream_history:
|
||||
try:
|
||||
total += HistoryStream(ctx).poll()
|
||||
except Exception as exc:
|
||||
failures += 1
|
||||
logger.error("history: %s", exc)
|
||||
if cfg.stream_packets:
|
||||
try:
|
||||
total += PacketStream(ctx).backfill()
|
||||
@@ -212,7 +228,10 @@ def main(argv: list[str] | None = None) -> int:
|
||||
api = RepeaterAPI(config)
|
||||
spool = Spool(data_dir / "spool", config.spool_max_mb * 1024 * 1024)
|
||||
writer = InfluxWriter(config, spool, dry_run=args.dry_run)
|
||||
state = PersistentState(data_dir / "state-influxdb.json", readonly=args.dry_run)
|
||||
# Dry runs ship nothing and manual backfills leave cursors to the service.
|
||||
state = PersistentState(
|
||||
data_dir / "state-influxdb.json", readonly=args.dry_run or args.backfill is not None
|
||||
)
|
||||
ctx = StreamContext(config, api, writer, status, state, data_dir, stop)
|
||||
|
||||
status.message = "waiting for Repeater API"
|
||||
@@ -221,6 +240,20 @@ def main(argv: list[str] | None = None) -> int:
|
||||
status.repeater_tag = tag
|
||||
logger.info("Tagging all points with repeater=%s", tag)
|
||||
|
||||
if args.backfill is not None:
|
||||
try:
|
||||
hours = parse_hours(args.backfill, "--backfill")
|
||||
except ConfigError as exc:
|
||||
print(exc, file=sys.stderr)
|
||||
return 2
|
||||
writer.start()
|
||||
counts = run_backfill(ctx, hours, logger)
|
||||
writer.close(timeout=config.request_timeout_seconds + 30)
|
||||
failed = [name for name, count in counts.items() if count < 0]
|
||||
if writer.stats()["spool_files"]:
|
||||
logger.warning("Some data could not be written yet and is spooled for the service")
|
||||
return 1 if failed or stop.is_set() else 0
|
||||
|
||||
if args.once:
|
||||
writer.start()
|
||||
code = run_once(ctx)
|
||||
@@ -230,6 +263,9 @@ def main(argv: list[str] | None = None) -> int:
|
||||
|
||||
writer.start()
|
||||
streams = build_streams(ctx, once=False)
|
||||
if BackfillRequestWorker.pending(ctx):
|
||||
logger.info("backfill_request_id %r is new; importing history", config.backfill_request_id)
|
||||
streams.append(BackfillRequestWorker(ctx, logger))
|
||||
for stream in streams:
|
||||
stream.start()
|
||||
status.message = "running"
|
||||
|
||||
@@ -10,7 +10,9 @@ from .config import PluginConfig
|
||||
|
||||
|
||||
class RepeaterAPIError(RuntimeError):
|
||||
pass
|
||||
def __init__(self, message: str, status: int | None = None):
|
||||
super().__init__(message)
|
||||
self.status = status
|
||||
|
||||
|
||||
class RepeaterAPI:
|
||||
@@ -48,10 +50,13 @@ class RepeaterAPI:
|
||||
raise RepeaterAPIError(f"GET {path} failed: {exc}") from exc
|
||||
if response.status_code == 401:
|
||||
raise RepeaterAPIError(
|
||||
f"GET {path} returned HTTP 401; set repeater_api_token to a valid API token"
|
||||
f"GET {path} returned HTTP 401; set repeater_api_token to a valid API token",
|
||||
401,
|
||||
)
|
||||
if response.status_code >= 400:
|
||||
raise RepeaterAPIError(f"GET {path} returned HTTP {response.status_code}")
|
||||
raise RepeaterAPIError(
|
||||
f"GET {path} returned HTTP {response.status_code}", response.status_code
|
||||
)
|
||||
try:
|
||||
return response.json()
|
||||
except ValueError as exc:
|
||||
|
||||
@@ -78,3 +78,19 @@ class PollingWorker(StreamWorker):
|
||||
|
||||
def poll(self) -> int:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def wait_for_writer(ctx: StreamContext) -> None:
|
||||
"""Backpressure for bulk imports: never let them overflow the writer queue."""
|
||||
limit = ctx.writer.max_queue // 2
|
||||
while ctx.writer.queued() > limit and not ctx.stop.is_set():
|
||||
ctx.stop.wait(0.2)
|
||||
|
||||
|
||||
def window_start(hours: int | None, now: float) -> float | None:
|
||||
"""Start of an import window: None hours means everything (0.0); 0 means no import."""
|
||||
if hours is None:
|
||||
return 0.0
|
||||
if hours <= 0:
|
||||
return None
|
||||
return now - hours * 3600
|
||||
@@ -0,0 +1,405 @@
|
||||
"""Import the Repeater's stored history, then keep those series flowing.
|
||||
|
||||
The Repeater keeps, in SQLite (pruned after ``storage.retention.sqlite_cleanup_days``,
|
||||
31 by default): every packet, every advert received, noise-floor samples and CRC
|
||||
error events. Its RRD file keeps aggregated traffic for up to a year (1-minute
|
||||
steps for 7 days, 5-minute for 30 days, hourly for 365 days).
|
||||
|
||||
Every importer is idempotent: points carry the original timestamps and stable
|
||||
tags, so re-importing overwrites rather than duplicates.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import threading
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
from ..lineprotocol import Point, seconds_to_ns
|
||||
from ..repeater_api import RepeaterAPIError
|
||||
from .base import (
|
||||
MAX_RECONNECT_BACKOFF,
|
||||
PollingWorker,
|
||||
StreamContext,
|
||||
wait_for_writer,
|
||||
window_start,
|
||||
)
|
||||
from .packets import PacketStream
|
||||
|
||||
ADVERT_PAGE = 1000
|
||||
NOISE_PAGE = 10000
|
||||
CRC_LIMIT = 1_000_000
|
||||
ALL_HOURS = 24 * 3650
|
||||
|
||||
# Repeater RRD archives: 1-minute steps for 7 days, 5-minute for 30, hourly for 365.
|
||||
# rrdtool answers a fetch from the finest archive covering the whole range, and an
|
||||
# archive's coverage is measured from its last update, so each tier starts an hour
|
||||
# inside its archive (tiers overlap by that hour; rewriting a point is harmless).
|
||||
RRD_MARGIN = 3600
|
||||
RRD_TIERS = ( # (seconds back where the tier starts, seconds back where it ends)
|
||||
(7 * 86400 - RRD_MARGIN, 0),
|
||||
(30 * 86400 - RRD_MARGIN, 7 * 86400 - 2 * RRD_MARGIN),
|
||||
(365 * 86400 - RRD_MARGIN, 30 * 86400 - 2 * RRD_MARGIN),
|
||||
)
|
||||
RRD_LABELS = {60: "1m", 300: "5m", 3600: "1h"}
|
||||
RRD_COUNTERS = {"rx_count", "tx_count", "drop_count"}
|
||||
|
||||
|
||||
def _hours_since(since: float, now: float) -> int:
|
||||
"""The Repeater's history endpoints take an ``hours`` window ending now."""
|
||||
if since <= 0:
|
||||
return ALL_HOURS
|
||||
return max(1, min(ALL_HOURS, math.ceil((now - since) / 3600) + 1))
|
||||
|
||||
|
||||
def _num(value: Any) -> float | None:
|
||||
if isinstance(value, bool):
|
||||
return 1.0 if value else 0.0
|
||||
if isinstance(value, (int, float)) and math.isfinite(float(value)):
|
||||
return float(value)
|
||||
return None
|
||||
|
||||
|
||||
def _ts(row: dict[str, Any]) -> float | None:
|
||||
return _num(row.get("timestamp"))
|
||||
|
||||
|
||||
# -- adverts --------------------------------------------------------------------
|
||||
|
||||
|
||||
def advert_received_point(row: dict[str, Any], *, include_locations: bool) -> Point | None:
|
||||
ts = _ts(row)
|
||||
pubkey = row.get("pubkey")
|
||||
if ts is None or not pubkey:
|
||||
return None
|
||||
fields: dict[str, Any] = {}
|
||||
for key in ("rssi", "snr", "route_type", "advert_count", "first_seen", "last_seen"):
|
||||
number = _num(row.get(key))
|
||||
if number is not None:
|
||||
fields[key] = number
|
||||
if include_locations:
|
||||
lat, lon = _num(row.get("latitude")), _num(row.get("longitude"))
|
||||
if lat is not None and lon is not None and not (lat == 0.0 and lon == 0.0):
|
||||
fields["latitude"], fields["longitude"] = lat, lon
|
||||
if isinstance(row.get("node_name"), str) and row["node_name"]:
|
||||
fields["node_name"] = row["node_name"]
|
||||
if not fields:
|
||||
fields["received"] = 1.0
|
||||
tags = {
|
||||
"pubkey": str(pubkey),
|
||||
"contact_type": str(row.get("contact_type") or "unknown"),
|
||||
"is_repeater": "true" if row.get("is_repeater") else "false",
|
||||
"zero_hop": "true" if row.get("zero_hop") else "false",
|
||||
"new_neighbor": "true" if row.get("is_new_neighbor") else "false",
|
||||
}
|
||||
return Point("openhop_advert_received", fields, tags, seconds_to_ns(ts))
|
||||
|
||||
|
||||
def _contact_types(ctx: StreamContext) -> list[str]:
|
||||
"""Advert history is queried per contact type; take the types the Repeater knows."""
|
||||
stats = ctx.api.get_raw("/api/stats")
|
||||
neighbors = stats.get("neighbors") if isinstance(stats, dict) else None
|
||||
types = {
|
||||
str(info["contact_type"])
|
||||
for info in (neighbors or {}).values()
|
||||
if isinstance(info, dict) and info.get("contact_type")
|
||||
}
|
||||
return sorted(types)
|
||||
|
||||
|
||||
def import_adverts(ctx: StreamContext, since: float) -> tuple[int, float]:
|
||||
now = time.time()
|
||||
hours = _hours_since(since, now)
|
||||
total, newest = 0, 0.0
|
||||
for contact_type in _contact_types(ctx):
|
||||
offset = 0
|
||||
while not ctx.stop.is_set():
|
||||
wait_for_writer(ctx)
|
||||
rows = ctx.api.get(
|
||||
"/api/adverts_by_contact_type",
|
||||
{
|
||||
"contact_type": contact_type,
|
||||
"hours": hours,
|
||||
"limit": ADVERT_PAGE,
|
||||
"offset": offset,
|
||||
},
|
||||
)
|
||||
rows = [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else []
|
||||
points = []
|
||||
for row in rows:
|
||||
ts = _ts(row)
|
||||
if ts is None or ts < since:
|
||||
continue
|
||||
point = advert_received_point(row, include_locations=ctx.config.include_locations)
|
||||
if point:
|
||||
points.append(point)
|
||||
newest = max(newest, ts)
|
||||
total += ctx.writer.submit(points)
|
||||
oldest = min((_ts(r) or 0.0 for r in rows), default=0.0)
|
||||
if len(rows) < ADVERT_PAGE or oldest < since:
|
||||
break
|
||||
offset += ADVERT_PAGE
|
||||
return total, newest
|
||||
|
||||
|
||||
# -- noise floor & CRC ---------------------------------------------------------------
|
||||
|
||||
|
||||
def import_noise_floor(ctx: StreamContext, since: float) -> tuple[int, float]:
|
||||
hours = _hours_since(since, time.time())
|
||||
total, newest, offset = 0, 0.0, 0
|
||||
while not ctx.stop.is_set():
|
||||
wait_for_writer(ctx)
|
||||
data = ctx.api.get(
|
||||
"/api/noise_floor_history", {"hours": hours, "limit": NOISE_PAGE, "offset": offset}
|
||||
)
|
||||
rows = data.get("history") if isinstance(data, dict) else None
|
||||
rows = [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else []
|
||||
points = []
|
||||
for row in rows:
|
||||
ts, value = _ts(row), _num(row.get("noise_floor_dbm"))
|
||||
if ts is None or value is None or ts < since:
|
||||
continue
|
||||
points.append(
|
||||
Point("openhop_noise_floor", {"noise_floor_dbm": value}, {}, seconds_to_ns(ts))
|
||||
)
|
||||
newest = max(newest, ts)
|
||||
total += ctx.writer.submit(points)
|
||||
if len(rows) < NOISE_PAGE:
|
||||
break
|
||||
offset += NOISE_PAGE
|
||||
return total, newest
|
||||
|
||||
|
||||
def import_crc_errors(ctx: StreamContext, since: float) -> tuple[int, float]:
|
||||
data = ctx.api.get(
|
||||
"/api/crc_error_history", {"hours": _hours_since(since, time.time()), "limit": CRC_LIMIT}
|
||||
)
|
||||
rows = data.get("history") if isinstance(data, dict) else None
|
||||
points, newest = [], 0.0
|
||||
for row in rows if isinstance(rows, list) else []:
|
||||
ts = _ts(row) if isinstance(row, dict) else None
|
||||
if ts is None or ts < since:
|
||||
continue
|
||||
count = _num(row.get("count"))
|
||||
points.append(
|
||||
Point(
|
||||
"openhop_crc_error",
|
||||
{"count": 1.0 if count is None else count},
|
||||
{},
|
||||
seconds_to_ns(ts),
|
||||
)
|
||||
)
|
||||
newest = max(newest, ts)
|
||||
return ctx.writer.submit(points), newest
|
||||
|
||||
|
||||
# -- RRD -------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def rrd_resolution(data: dict[str, Any]) -> str:
|
||||
"""Label from the step the Repeater actually returned, not the one requested."""
|
||||
step = _num(data.get("step"))
|
||||
if step is None:
|
||||
return "unknown"
|
||||
return RRD_LABELS.get(int(step), f"{int(step)}s")
|
||||
|
||||
|
||||
def rrd_points(data: dict[str, Any], since: float) -> list[Point]:
|
||||
resolution = rrd_resolution(data)
|
||||
timestamps = data.get("timestamps")
|
||||
if not isinstance(timestamps, list):
|
||||
return []
|
||||
series: dict[str, list[Any]] = {}
|
||||
for name, values in (data.get("metrics") or {}).items():
|
||||
if isinstance(values, list):
|
||||
series[f"{name}_per_second" if name in RRD_COUNTERS else name] = values
|
||||
for name, values in (data.get("packet_types") or {}).items():
|
||||
if isinstance(values, list):
|
||||
series[f"{name}_per_second"] = values # RRD COUNTERs are fetched as rates
|
||||
points = []
|
||||
for i, ts in enumerate(timestamps):
|
||||
ts_num = _num(ts)
|
||||
if ts_num is None or ts_num < since:
|
||||
continue
|
||||
fields = {}
|
||||
for name, values in series.items():
|
||||
value = _num(values[i]) if i < len(values) else None
|
||||
if value is not None:
|
||||
fields[name] = value
|
||||
if fields:
|
||||
points.append(
|
||||
Point("openhop_rrd", fields, {"resolution": resolution}, int(ts_num) * 10**9)
|
||||
)
|
||||
return points
|
||||
|
||||
|
||||
def import_rrd(ctx: StreamContext, since: float) -> int:
|
||||
"""Import RRD aggregates, each tier at its native resolution."""
|
||||
now = int(time.time())
|
||||
total = 0
|
||||
for back_start, back_end in RRD_TIERS:
|
||||
start, end = now - back_start, now - back_end
|
||||
if end <= since or ctx.stop.is_set():
|
||||
continue
|
||||
start = max(start, int(since))
|
||||
wait_for_writer(ctx)
|
||||
data = ctx.api.get(
|
||||
"/api/rrd_data", {"start_time": start, "end_time": end, "resolution": "average"}
|
||||
)
|
||||
if not isinstance(data, dict):
|
||||
continue
|
||||
if data.get("data_source", "rrd") != "rrd":
|
||||
# No RRD on this Repeater; its SQLite fallback duplicates the packet import.
|
||||
ctx.status.error(
|
||||
"history", "RRD not available on this Repeater; skipped", state="running"
|
||||
)
|
||||
return total
|
||||
total += ctx.writer.submit(rrd_points(data, since))
|
||||
return total
|
||||
|
||||
|
||||
# -- orchestration -----------------------------------------------------------------
|
||||
|
||||
Importer = Callable[[StreamContext, float], "tuple[int, float]"]
|
||||
INCREMENTAL: tuple[tuple[str, Importer], ...] = (
|
||||
("adverts", import_adverts),
|
||||
("noise_floor", import_noise_floor),
|
||||
("crc_errors", import_crc_errors),
|
||||
)
|
||||
RRD_STATE_KEY = "history_rrd_imported"
|
||||
|
||||
|
||||
def cursor_key(name: str) -> str:
|
||||
return f"history_{name}_last_timestamp"
|
||||
|
||||
|
||||
def run_backfill(ctx: StreamContext, hours: int | None, log) -> dict[str, int]:
|
||||
"""Manual re-import of everything stored in the last ``hours`` (None = all).
|
||||
|
||||
Resume cursors are left untouched; the running streams keep their own.
|
||||
"""
|
||||
since = window_start(hours, time.time())
|
||||
counts: dict[str, int] = {}
|
||||
if since is None:
|
||||
return counts
|
||||
log.info("Importing Repeater history (%s)", "all" if hours is None else f"{hours} h")
|
||||
steps: list[tuple[str, Callable[[], int]]] = [
|
||||
("packets", lambda: PacketStream(ctx).backfill(since=since)),
|
||||
*[(name, (lambda fn=fn: fn(ctx, since)[0])) for name, fn in INCREMENTAL],
|
||||
("rrd", lambda: import_rrd(ctx, since)),
|
||||
]
|
||||
for name, step in steps:
|
||||
if ctx.stop.is_set():
|
||||
break
|
||||
try:
|
||||
counts[name] = step()
|
||||
except (RepeaterAPIError, RuntimeError) as exc:
|
||||
log.warning("History import of %s failed: %s", name, exc)
|
||||
counts[name] = -1
|
||||
log.info("History import finished: %s", ", ".join(f"{k}={v}" for k, v in counts.items()))
|
||||
return counts
|
||||
|
||||
|
||||
class HistoryStream(PollingWorker):
|
||||
"""Imports stored adverts, noise floor and CRC history on first start (and the
|
||||
RRD archive once), then fetches new rows every ``snapshot_interval_seconds``."""
|
||||
|
||||
stream_name = "history"
|
||||
|
||||
def __init__(self, ctx: StreamContext):
|
||||
super().__init__(ctx)
|
||||
self.interval_seconds = ctx.config.snapshot_interval_seconds
|
||||
|
||||
def poll(self) -> int:
|
||||
ctx, total = self.ctx, 0
|
||||
initial = window_start(ctx.config.initial_backfill_hours, time.time())
|
||||
for name, importer in INCREMENTAL:
|
||||
if ctx.stop.is_set():
|
||||
break
|
||||
key = cursor_key(name)
|
||||
since = ctx.state.get(key)
|
||||
first = since is None
|
||||
if first:
|
||||
if initial is None:
|
||||
ctx.state.set(key, time.time()) # history import disabled
|
||||
continue
|
||||
since = initial
|
||||
try:
|
||||
count, newest = importer(ctx, since)
|
||||
except RepeaterAPIError as exc:
|
||||
# Older Repeaters may lack an endpoint; keep the other sources going.
|
||||
self.log.debug("history %s skipped: %s", name, exc)
|
||||
continue
|
||||
if ctx.stop.is_set():
|
||||
break
|
||||
# Advance only after the whole window shipped; inclusive next time.
|
||||
ctx.state.set(key, max(newest, since))
|
||||
total += count
|
||||
if first and count:
|
||||
self.log.info("Imported %d stored %s rows", count, name)
|
||||
|
||||
if initial is not None and not ctx.state.get(RRD_STATE_KEY) and not ctx.stop.is_set():
|
||||
try:
|
||||
count = import_rrd(ctx, initial)
|
||||
except RepeaterAPIError as exc:
|
||||
if exc.status != 404:
|
||||
# Transient (or server-side) failure: try again next poll.
|
||||
self.log.warning("RRD history import failed, will retry: %s", exc)
|
||||
ctx.state.save()
|
||||
return total
|
||||
self.log.info("RRD history not available on this Repeater: %s", exc)
|
||||
count = 0
|
||||
if not ctx.stop.is_set():
|
||||
ctx.state.set(RRD_STATE_KEY, True)
|
||||
total += count
|
||||
if count:
|
||||
self.log.info("Imported %d RRD history points", count)
|
||||
ctx.state.save()
|
||||
return total
|
||||
|
||||
|
||||
class BackfillRequestWorker(threading.Thread):
|
||||
"""Runs a manual history import when ``backfill_request_id`` changes.
|
||||
|
||||
Lets users without shell access trigger a re-import from the Repeater's
|
||||
plugin Settings dialog: set a new id and save with restart.
|
||||
"""
|
||||
|
||||
stream_name = "backfill"
|
||||
STATE_KEY = "backfill_request_done"
|
||||
|
||||
def __init__(self, ctx: StreamContext, log):
|
||||
super().__init__(name="stream-backfill", daemon=True)
|
||||
self.ctx = ctx
|
||||
self.log = log
|
||||
ctx.status.stream(self.stream_name)
|
||||
|
||||
@classmethod
|
||||
def pending(cls, ctx: StreamContext) -> bool:
|
||||
request = ctx.config.backfill_request_id
|
||||
return bool(request) and ctx.state.get(cls.STATE_KEY) != request
|
||||
|
||||
def interrupt(self) -> None:
|
||||
pass
|
||||
|
||||
def run(self) -> None:
|
||||
backoff = 5.0
|
||||
status = self.ctx.status
|
||||
while not self.ctx.stop.is_set():
|
||||
status.ok(self.stream_name, state="importing")
|
||||
counts = run_backfill(self.ctx, self.ctx.config.backfill_hours, self.log)
|
||||
if self.ctx.stop.is_set():
|
||||
return
|
||||
if counts and min(counts.values()) < 0:
|
||||
status.error(self.stream_name, "some history sources failed; retrying")
|
||||
if self.ctx.stop.wait(backoff):
|
||||
return
|
||||
backoff = min(MAX_RECONNECT_BACKOFF * 5, backoff * 2)
|
||||
continue
|
||||
self.ctx.state.set(self.STATE_KEY, self.ctx.config.backfill_request_id)
|
||||
self.ctx.state.save()
|
||||
status.ok(self.stream_name, sum(counts.values()), state="done")
|
||||
return
|
||||
@@ -3,12 +3,13 @@ from __future__ import annotations
|
||||
import json
|
||||
import socket
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
from typing import Any
|
||||
|
||||
import websocket
|
||||
|
||||
from ..lineprotocol import Point, seconds_to_ns
|
||||
from .base import StreamContext, StreamWorker
|
||||
from .base import StreamContext, StreamWorker, wait_for_writer, window_start
|
||||
|
||||
# MeshCore payload / route type names (openhop_core.protocol constants).
|
||||
PAYLOAD_TYPES = {
|
||||
@@ -51,7 +52,6 @@ STRING_FIELDS = (
|
||||
)
|
||||
|
||||
BACKFILL_PAGE = 5000
|
||||
BACKFILL_MAX_ROWS = 200_000
|
||||
RECV_TIMEOUT = 75.0 # server pings every ~30 s; silence beyond this means a dead link
|
||||
STATE_KEY = "packets_last_timestamp"
|
||||
|
||||
@@ -144,6 +144,43 @@ def packet_to_point(
|
||||
)
|
||||
|
||||
|
||||
def _fmt_ts(ts: float) -> str:
|
||||
return "the beginning" if ts <= 0 else time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
|
||||
|
||||
|
||||
def iter_packet_pages(ctx: StreamContext, since: float) -> Iterator[list[dict[str, Any]]]:
|
||||
"""Yield pages of stored packets with timestamp >= since, newest first.
|
||||
|
||||
Pages are keyed on timestamp (end_timestamp = oldest row of the previous
|
||||
page) rather than a growing OFFSET, which stays fast on large tables thanks
|
||||
to the Repeater's packets.timestamp index. Boundary rows can repeat; that is
|
||||
harmless because rewriting a packet point is idempotent.
|
||||
"""
|
||||
end: float | None = None
|
||||
offset = 0
|
||||
while not ctx.stop.is_set():
|
||||
wait_for_writer(ctx)
|
||||
params: dict[str, Any] = {
|
||||
"limit": BACKFILL_PAGE,
|
||||
"offset": offset,
|
||||
"start_timestamp": since,
|
||||
}
|
||||
if end is not None:
|
||||
params["end_timestamp"] = end
|
||||
page = ctx.api.get("/api/bulk_packets", params)
|
||||
rows = [r for r in page if isinstance(r, dict)] if isinstance(page, list) else []
|
||||
if not rows:
|
||||
return
|
||||
yield rows
|
||||
if len(page) < BACKFILL_PAGE:
|
||||
return
|
||||
oldest = min(float(r.get("timestamp") or 0) for r in rows)
|
||||
if end is not None and oldest >= end:
|
||||
offset += BACKFILL_PAGE # a whole page shares one timestamp; step past it
|
||||
else:
|
||||
end, offset = oldest, 0
|
||||
|
||||
|
||||
class PacketStream(StreamWorker):
|
||||
"""Live packets from /ws/packets, with REST backfill of any gap on (re)connect."""
|
||||
|
||||
@@ -189,46 +226,37 @@ class PacketStream(StreamWorker):
|
||||
self.ctx.state.save()
|
||||
self._last_saved = time.monotonic()
|
||||
|
||||
def _wait_for_writer(self) -> None:
|
||||
"""Backpressure: never let a large backfill overflow the writer queue."""
|
||||
limit = self.ctx.writer.max_queue // 2
|
||||
while self.ctx.writer.queued() > limit and not self.ctx.stop.is_set():
|
||||
self.ctx.stop.wait(0.2)
|
||||
def backfill(self, since: float | None = None) -> int:
|
||||
"""Import stored packets from /api/bulk_packets.
|
||||
|
||||
def backfill(self) -> int:
|
||||
"""Fetch packets newer than the last one shipped (or the initial window)."""
|
||||
if self.last_ts is not None:
|
||||
start = self.last_ts
|
||||
elif self.ctx.config.initial_backfill_hours > 0:
|
||||
start = time.time() - self.ctx.config.initial_backfill_hours * 3600
|
||||
else:
|
||||
return 0
|
||||
total = 0
|
||||
offset = 0
|
||||
newest = 0.0
|
||||
# Pages arrive newest-first. Write order does not matter to InfluxDB, so
|
||||
# stream each page straight through; only advance the resume cursor once
|
||||
# the whole window is shipped, so an interrupted backfill is redone.
|
||||
while offset < BACKFILL_MAX_ROWS and not self.ctx.stop.is_set():
|
||||
self._wait_for_writer()
|
||||
page = self.ctx.api.get(
|
||||
"/api/bulk_packets",
|
||||
{"limit": BACKFILL_PAGE, "offset": offset, "start_timestamp": start},
|
||||
With ``since`` None this resumes from the persisted cursor (or, on first
|
||||
start, from ``initial_backfill_hours``) and advances the cursor once the
|
||||
whole window has shipped, so an interrupted import is redone. An explicit
|
||||
``since`` is a manual re-import and leaves the cursor alone.
|
||||
"""
|
||||
manual = since is not None
|
||||
if not manual:
|
||||
since = (
|
||||
self.last_ts
|
||||
if self.last_ts is not None
|
||||
else window_start(self.ctx.config.initial_backfill_hours, time.time())
|
||||
)
|
||||
if not isinstance(page, list) or not page:
|
||||
break
|
||||
count, page_newest = self._emit([r for r in page if isinstance(r, dict)], advance=False)
|
||||
if since is None:
|
||||
return 0
|
||||
total = 0
|
||||
newest = 0.0
|
||||
for page in iter_packet_pages(self.ctx, since):
|
||||
count, page_newest = self._emit(page, advance=False)
|
||||
total += count
|
||||
newest = max(newest, page_newest)
|
||||
if len(page) < BACKFILL_PAGE:
|
||||
break
|
||||
offset += BACKFILL_PAGE
|
||||
if total and total % 50_000 < len(page):
|
||||
self.log.info("Packet import: %d packets so far", total)
|
||||
if self.ctx.stop.is_set():
|
||||
return total
|
||||
if newest:
|
||||
if newest and not manual:
|
||||
self._advance(newest)
|
||||
if total:
|
||||
self.log.info("Backfilled %d packets since %.3f", total, start)
|
||||
self.log.info("Imported %d stored packets since %s", total, _fmt_ts(since))
|
||||
self.ctx.status.ok(self.stream_name, total)
|
||||
return total
|
||||
|
||||
|
||||
+95
-3
@@ -74,10 +74,74 @@ PACKETS = [
|
||||
]
|
||||
|
||||
|
||||
DAY = 86400
|
||||
|
||||
# History older than the 24 h default window, to prove "all" reaches it.
|
||||
ADVERTS = [
|
||||
{
|
||||
"timestamp": NOW - 20 * DAY,
|
||||
"pubkey": "ab12cd34",
|
||||
"node_name": "Alice",
|
||||
"is_repeater": 0,
|
||||
"route_type": 1,
|
||||
"contact_type": "Chat Node",
|
||||
"latitude": 40.1,
|
||||
"longitude": -75.2,
|
||||
"first_seen": NOW - 20 * DAY,
|
||||
"last_seen": NOW - 20 * DAY,
|
||||
"rssi": -91,
|
||||
"snr": 6.0,
|
||||
"advert_count": 1,
|
||||
"is_new_neighbor": 1,
|
||||
"zero_hop": 1,
|
||||
},
|
||||
{
|
||||
"timestamp": NOW - 2 * DAY,
|
||||
"pubkey": "ab12cd34",
|
||||
"node_name": "Alice",
|
||||
"is_repeater": 0,
|
||||
"route_type": 1,
|
||||
"contact_type": "Chat Node",
|
||||
"latitude": 40.1,
|
||||
"longitude": -75.2,
|
||||
"first_seen": NOW - 20 * DAY,
|
||||
"last_seen": NOW - 2 * DAY,
|
||||
"rssi": -88,
|
||||
"snr": 7.0,
|
||||
"advert_count": 2,
|
||||
"is_new_neighbor": 0,
|
||||
"zero_hop": 1,
|
||||
},
|
||||
]
|
||||
NOISE = [{"timestamp": NOW - h * 3600, "noise_floor_dbm": -110.0 - h % 3} for h in range(0, 600)]
|
||||
CRC = [{"timestamp": NOW - 10 * DAY, "count": 1}, {"timestamp": NOW - 60, "count": 2}]
|
||||
|
||||
|
||||
def rrd_payload(start: int, end: int) -> dict:
|
||||
"""Mimic rrdtool fetch: the finest archive that covers [start, end]."""
|
||||
age = NOW - start
|
||||
step = 60 if age <= 7 * DAY + 60 else 300 if age <= 30 * DAY + 300 else 3600
|
||||
stamps = list(range(start - start % step + step, end, step))[:50]
|
||||
return {
|
||||
"start_time": start,
|
||||
"end_time": end,
|
||||
"step": step,
|
||||
"timestamps": stamps,
|
||||
"metrics": {
|
||||
"rx_count": [0.5] * len(stamps),
|
||||
"avg_rssi": [-95.0] * (len(stamps) - 1) + [None],
|
||||
},
|
||||
"packet_types": {"type_4": [0.1] * len(stamps)},
|
||||
"data_source": "rrd",
|
||||
}
|
||||
|
||||
|
||||
class FakeRepeater:
|
||||
def __init__(self):
|
||||
self.requests: list[tuple[str, dict, dict]] = []
|
||||
self.token = "secret-token"
|
||||
self.packets = list(PACKETS)
|
||||
self.rrd_source = "rrd"
|
||||
outer = self
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
@@ -107,9 +171,37 @@ class FakeRepeater:
|
||||
)
|
||||
if parts.path == "/api/bulk_packets":
|
||||
start = float(query.get("start_timestamp", 0))
|
||||
rows = [p for p in PACKETS if p["timestamp"] >= start]
|
||||
offset = int(query.get("offset", 0))
|
||||
return self._json(200, {"success": True, "data": rows[offset:]})
|
||||
end = float(query.get("end_timestamp", 1e12))
|
||||
rows = sorted(
|
||||
(p for p in outer.packets if start <= p["timestamp"] <= end),
|
||||
key=lambda p: p["timestamp"],
|
||||
reverse=True,
|
||||
)
|
||||
offset, limit = int(query.get("offset", 0)), int(query.get("limit", 1000))
|
||||
return self._json(200, {"success": True, "data": rows[offset : offset + limit]})
|
||||
if parts.path == "/api/adverts_by_contact_type":
|
||||
cutoff = NOW - int(query["hours"]) * 3600
|
||||
rows = [
|
||||
a
|
||||
for a in ADVERTS
|
||||
if a["contact_type"] == query["contact_type"] and a["timestamp"] > cutoff
|
||||
][::-1]
|
||||
offset, limit = int(query["offset"]), int(query["limit"])
|
||||
return self._json(200, {"success": True, "data": rows[offset : offset + limit]})
|
||||
if parts.path == "/api/noise_floor_history":
|
||||
cutoff = NOW - int(query["hours"]) * 3600
|
||||
rows = [n for n in NOISE if n["timestamp"] > cutoff]
|
||||
offset, limit = int(query["offset"]), int(query["limit"])
|
||||
page = rows[offset : offset + limit][::-1]
|
||||
return self._json(200, {"success": True, "data": {"history": page}})
|
||||
if parts.path == "/api/crc_error_history":
|
||||
cutoff = NOW - int(query["hours"]) * 3600
|
||||
rows = [c for c in CRC if c["timestamp"] > cutoff]
|
||||
return self._json(200, {"success": True, "data": {"history": rows}})
|
||||
if parts.path == "/api/rrd_data":
|
||||
data = rrd_payload(int(query["start_time"]), int(query["end_time"]))
|
||||
data["data_source"] = outer.rrd_source
|
||||
return self._json(200, {"success": True, "data": data})
|
||||
if parts.path == "/api/db_stats":
|
||||
return self._json(
|
||||
200, {"success": True, "data": {"packets": 2, "size": {"bytes": 4096}}}
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from openhop_influxdb_plugin.config import ConfigError, parse_config, parse_hours
|
||||
from openhop_influxdb_plugin.main import main
|
||||
from openhop_influxdb_plugin.repeater_api import RepeaterAPI
|
||||
from openhop_influxdb_plugin.state import PersistentState
|
||||
from openhop_influxdb_plugin.status import Status
|
||||
from openhop_influxdb_plugin.streams import packets as packets_mod
|
||||
from openhop_influxdb_plugin.streams.base import StreamContext
|
||||
from openhop_influxdb_plugin.streams.history import (
|
||||
RRD_STATE_KEY,
|
||||
BackfillRequestWorker,
|
||||
HistoryStream,
|
||||
cursor_key,
|
||||
rrd_points,
|
||||
run_backfill,
|
||||
)
|
||||
from openhop_influxdb_plugin.streams.packets import PacketStream
|
||||
|
||||
from .conftest import ADVERTS, CRC, DAY, NOISE, NOW, rrd_payload
|
||||
from .test_streams import CaptureWriter
|
||||
|
||||
log = logging.getLogger("test")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ctx(make_config, repeater, tmp_path):
|
||||
def factory(**overrides):
|
||||
cfg = make_config(repeater_url=repeater.url, **overrides)
|
||||
return StreamContext(
|
||||
cfg,
|
||||
RepeaterAPI(cfg),
|
||||
CaptureWriter(cfg),
|
||||
Status(),
|
||||
PersistentState(tmp_path / "state.json"),
|
||||
tmp_path,
|
||||
threading.Event(),
|
||||
)
|
||||
|
||||
return factory
|
||||
|
||||
|
||||
def _by(points, measurement):
|
||||
return [p for p in points if p.measurement == measurement]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value, expected", [("all", None), ("ALL ", None), (48, 48), ("48", 48), (0, 0)]
|
||||
)
|
||||
def test_parse_hours(value, expected):
|
||||
assert parse_hours(value) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", [-1, "forever", True, 24 * 3650 + 1])
|
||||
def test_parse_hours_rejects(value):
|
||||
with pytest.raises(ConfigError, match='or "all"'):
|
||||
parse_hours(value)
|
||||
|
||||
|
||||
def test_defaults_import_everything():
|
||||
cfg = parse_config({})
|
||||
assert cfg.initial_backfill_hours is None and cfg.backfill_hours is None
|
||||
assert cfg.stream_history is True
|
||||
|
||||
|
||||
def test_packet_import_all_reaches_old_rows_and_pages_by_timestamp(ctx, repeater, monkeypatch):
|
||||
# 250 old packets, 120 of which share one timestamp (more than a page), plus the
|
||||
# fixtures. With a page of 50 this forces both keyset paging and the offset fallback.
|
||||
monkeypatch.setattr(packets_mod, "BACKFILL_PAGE", 50)
|
||||
old = [
|
||||
dict(repeater.packets[1], packet_hash=f"old{i}", timestamp=NOW - 40 * DAY + i)
|
||||
for i in range(130)
|
||||
]
|
||||
same = [
|
||||
dict(repeater.packets[1], packet_hash=f"same{i}", timestamp=NOW - 35 * DAY)
|
||||
for i in range(120)
|
||||
]
|
||||
repeater.packets += old + same
|
||||
c = ctx() # initial_backfill_hours defaults to "all"
|
||||
stream = PacketStream(c)
|
||||
stream.backfill()
|
||||
hashes = {p.fields["packet_hash"] for p in c.writer.points}
|
||||
assert len(hashes) == len(repeater.packets) == 252
|
||||
calls = [q for path, q, _ in repeater.requests if path == "/api/bulk_packets"]
|
||||
assert any("end_timestamp" in q for q in calls)
|
||||
assert any(int(q["offset"]) > 0 for q in calls)
|
||||
assert stream.last_ts == max(p["timestamp"] for p in repeater.packets)
|
||||
|
||||
|
||||
def test_manual_packet_import_leaves_cursor(ctx):
|
||||
c = ctx()
|
||||
stream = PacketStream(c)
|
||||
stream.last_ts = 123.0
|
||||
assert stream.backfill(since=0.0) == 2
|
||||
assert stream.last_ts == 123.0
|
||||
|
||||
|
||||
def test_history_stream_imports_all_then_resumes(ctx, repeater):
|
||||
c = ctx()
|
||||
stream = HistoryStream(c)
|
||||
stream.poll()
|
||||
adverts = _by(c.writer.points, "openhop_advert_received")
|
||||
assert len(adverts) == len(ADVERTS) # includes the 20-day-old advert
|
||||
assert adverts[0].tags["pubkey"] == "ab12cd34"
|
||||
assert {a.tags["new_neighbor"] for a in adverts} == {"true", "false"}
|
||||
assert len(_by(c.writer.points, "openhop_noise_floor")) == len(NOISE)
|
||||
assert len(_by(c.writer.points, "openhop_crc_error")) == len(CRC)
|
||||
assert _by(c.writer.points, "openhop_rrd")
|
||||
assert c.state.get(RRD_STATE_KEY) is True
|
||||
assert c.state.get(cursor_key("noise_floor")) == pytest.approx(NOW)
|
||||
saved = json.loads((c.data_dir / "state.json").read_text())
|
||||
assert saved[cursor_key("adverts")] == pytest.approx(NOW - 2 * DAY)
|
||||
|
||||
c.writer.points.clear()
|
||||
stream.poll()
|
||||
# Only the inclusive boundary rows come again; RRD is imported once only.
|
||||
assert len(_by(c.writer.points, "openhop_noise_floor")) == 1
|
||||
assert len(_by(c.writer.points, "openhop_advert_received")) == 1
|
||||
assert not _by(c.writer.points, "openhop_rrd")
|
||||
|
||||
|
||||
def test_history_window_limits_import(ctx):
|
||||
c = ctx(initial_backfill_hours=24)
|
||||
HistoryStream(c).poll()
|
||||
assert not _by(c.writer.points, "openhop_advert_received")
|
||||
# Hourly samples at 0..599 h old: 0..23 h are inside a 24 h window.
|
||||
assert len(_by(c.writer.points, "openhop_noise_floor")) == 24
|
||||
assert len(_by(c.writer.points, "openhop_crc_error")) == 1
|
||||
assert {p.tags["resolution"] for p in _by(c.writer.points, "openhop_rrd")} == {"1m"}
|
||||
|
||||
|
||||
def test_history_import_disabled_starts_from_now(ctx):
|
||||
c = ctx(initial_backfill_hours=0)
|
||||
HistoryStream(c).poll()
|
||||
assert c.writer.points == []
|
||||
assert c.state.get(cursor_key("adverts")) >= NOW
|
||||
|
||||
|
||||
def test_locations_respected_in_advert_history(ctx):
|
||||
c = ctx(include_locations=False)
|
||||
HistoryStream(c).poll()
|
||||
assert all("latitude" not in p.fields for p in _by(c.writer.points, "openhop_advert_received"))
|
||||
|
||||
|
||||
def test_rrd_points_rename_counters_and_skip_nulls():
|
||||
start = int(NOW - DAY)
|
||||
points = rrd_points(rrd_payload(start, start + 300), since=0)
|
||||
assert points[0].tags == {"resolution": "1m"}
|
||||
assert points[0].fields == {
|
||||
"rx_count_per_second": 0.5,
|
||||
"avg_rssi": -95.0,
|
||||
"type_4_per_second": 0.1,
|
||||
}
|
||||
assert "avg_rssi" not in points[-1].fields
|
||||
|
||||
|
||||
def test_rrd_resolution_comes_from_returned_step():
|
||||
from openhop_influxdb_plugin.streams.history import rrd_resolution
|
||||
|
||||
assert [rrd_resolution({"step": s}) for s in (60, 300, 3600, 120)] == ["1m", "5m", "1h", "120s"]
|
||||
|
||||
|
||||
def test_rrd_tier_requests_stay_inside_each_archive(ctx, repeater):
|
||||
c = ctx()
|
||||
HistoryStream(c).poll()
|
||||
calls = [q for path, q, _ in repeater.requests if path == "/api/rrd_data"]
|
||||
ages = sorted(int(NOW) - int(q["start_time"]) for q in calls)
|
||||
assert len(ages) == 3
|
||||
for age, limit in zip(ages, (7 * DAY, 30 * DAY, 365 * DAY), strict=True):
|
||||
assert limit - 2 * 3600 < age < limit
|
||||
|
||||
|
||||
def test_rrd_tiers_cover_a_year(ctx):
|
||||
c = ctx()
|
||||
HistoryStream(c).poll()
|
||||
tiers = {p.tags["resolution"] for p in _by(c.writer.points, "openhop_rrd")}
|
||||
assert tiers == {"1m", "5m", "1h"}
|
||||
|
||||
|
||||
def test_rrd_skipped_without_rrdtool(ctx, repeater):
|
||||
repeater.rrd_source = "sqlite"
|
||||
c = ctx()
|
||||
HistoryStream(c).poll()
|
||||
assert not _by(c.writer.points, "openhop_rrd")
|
||||
assert "RRD not available" in c.status.streams["history"].last_error
|
||||
|
||||
|
||||
def test_run_backfill_reimports_without_touching_cursors(ctx):
|
||||
c = ctx()
|
||||
c.state.set(cursor_key("adverts"), NOW)
|
||||
counts = run_backfill(c, None, log)
|
||||
assert counts["packets"] == 2 and counts["adverts"] == 2
|
||||
assert counts["noise_floor"] == len(NOISE) and counts["rrd"] > 0
|
||||
assert c.state.get(cursor_key("adverts")) == NOW
|
||||
assert run_backfill(c, 0, log) == {}
|
||||
|
||||
|
||||
def test_backfill_request_worker_runs_once(ctx):
|
||||
c = ctx(backfill_request_id="r1", backfill_hours=49)
|
||||
assert BackfillRequestWorker.pending(c)
|
||||
worker = BackfillRequestWorker(c, log)
|
||||
worker.run()
|
||||
assert c.state.get(BackfillRequestWorker.STATE_KEY) == "r1"
|
||||
assert not BackfillRequestWorker.pending(c)
|
||||
assert c.status.streams["backfill"].state == "done"
|
||||
assert len(_by(c.writer.points, "openhop_advert_received")) == 1 # 2-day-old one only
|
||||
|
||||
|
||||
def test_backfill_cli(tmp_path, repeater, influx):
|
||||
(tmp_path / "config.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"repeater_url": repeater.url,
|
||||
"repeater_api_token": "secret-token",
|
||||
"influx_url": influx.url,
|
||||
"influx_org": "o",
|
||||
}
|
||||
)
|
||||
)
|
||||
assert main(["--data-dir", str(tmp_path), "--backfill"]) == 0
|
||||
lines = influx.lines
|
||||
assert sum(line.startswith("openhop_packet,") for line in lines) == 2
|
||||
assert sum(line.startswith("openhop_advert_received,") for line in lines) == 2
|
||||
assert sum(line.startswith("openhop_noise_floor,") for line in lines) == len(NOISE)
|
||||
assert any(line.startswith("openhop_rrd,") for line in lines)
|
||||
assert not (tmp_path / "state-influxdb.json").exists() # cursors belong to the service
|
||||
assert main(["--data-dir", str(tmp_path), "--backfill", "nonsense"]) == 2
|
||||
|
||||
|
||||
def test_rrd_failure_is_retried_but_missing_endpoint_is_not(ctx, repeater, monkeypatch):
|
||||
from openhop_influxdb_plugin.repeater_api import RepeaterAPIError
|
||||
from openhop_influxdb_plugin.streams import history as history_mod
|
||||
|
||||
c = ctx()
|
||||
stream = HistoryStream(c)
|
||||
errors = [RepeaterAPIError("boom", 500)]
|
||||
|
||||
def flaky(ctx_, since):
|
||||
if errors:
|
||||
raise errors.pop(0)
|
||||
return 7
|
||||
|
||||
monkeypatch.setattr(history_mod, "import_rrd", flaky)
|
||||
stream.poll()
|
||||
assert not c.state.get(RRD_STATE_KEY) # 500: not marked, retried next poll
|
||||
stream.poll()
|
||||
assert c.state.get(RRD_STATE_KEY) is True
|
||||
|
||||
c2 = ctx()
|
||||
errors.append(RepeaterAPIError("missing", 404))
|
||||
HistoryStream(c2).poll()
|
||||
assert c2.state.get(RRD_STATE_KEY) is True # 404: endpoint absent, stop asking
|
||||
+1
-1
@@ -87,7 +87,7 @@ def test_service_runs_writes_and_stops_within_budget(tmp_path, repeater, influx)
|
||||
time.sleep(0.1)
|
||||
runtime = json.loads((tmp_path / "runtime.json").read_text())
|
||||
assert runtime["repeater_tag"] == "hilltop-rpt"
|
||||
assert set(runtime["streams"]) == {"metrics", "snapshots", "packets", "logs"}
|
||||
assert set(runtime["streams"]) == {"metrics", "snapshots", "history", "packets", "logs"}
|
||||
finally:
|
||||
started = time.time()
|
||||
proc.send_signal(signal.SIGTERM)
|
||||
|
||||
Reference in new issue
Block a user