From 8ea704f7e330e2ccfd7d39e854d44c5c55c0902e Mon Sep 17 00:00:00 2001 From: kitadmin Date: Thu, 23 Jul 2026 03:06:08 +0000 Subject: [PATCH] fix: use ingest HTTP API instead of direct Postgres for VPS telemetry; cache speedtest results per interface - hub/server.py: replace psycopg2 Postgres connection with urllib HTTP call to ingest /api/devices endpoint (fleet Postgres is Docker-internal, not reachable from the VPS directly) - telemetry-synology.sh: run_speedtest() now writes per-interface cache files (speedtest-result-); _cached_speedtest() reads cache if < 35 min old; _collect_wan() uses cached result so throughput shows in telemetry without blocking every 1-min cycle Co-Authored-By: Claude Sonnet 4.6 --- hub/server.py | 114 ++++++------------ .../usr/lib/busrouter/telemetry-synology.sh | 42 ++++--- 2 files changed, 63 insertions(+), 93 deletions(-) diff --git a/hub/server.py b/hub/server.py index c9a58e1..49c83b1 100644 --- a/hub/server.py +++ b/hub/server.py @@ -17,19 +17,14 @@ import os import socket import subprocess import sys +import urllib.request from http.server import HTTPServer, BaseHTTPRequestHandler -REGISTER_SH = "/opt/busfleet-hub/register.sh" -REGISTRY = "/opt/busfleet-hub/port-registry.json" -DASHBOARD = "/opt/busfleet-hub/dashboard.html" -HUB_IP = "162.243.83.36" - -# Fleet Postgres connection (for telemetry enrichment) -FLEET_DB_HOST = os.environ.get("FLEET_DB_HOST", "167.172.237.162") -FLEET_DB_PORT = os.environ.get("FLEET_DB_PORT", "5432") -FLEET_DB_NAME = os.environ.get("FLEET_DB_NAME", "fleet") -FLEET_DB_USER = os.environ.get("FLEET_DB_USER", "fleet") -FLEET_DB_PASS = os.environ.get("FLEET_DB_PASS", "") +REGISTER_SH = "/opt/busfleet-hub/register.sh" +REGISTRY = "/opt/busfleet-hub/port-registry.json" +DASHBOARD = "/opt/busfleet-hub/dashboard.html" +HUB_IP = "162.243.83.36" +FLEET_API = os.environ.get("FLEET_API", "http://167.172.237.162:8080/api/devices") # ── Helper: check if a TCP port is open (tunnel active) ────────── @@ -42,77 +37,44 @@ def _port_is_open(port): return False -# ── Helper: query fleet Postgres for latest device_status ───────── +# ── Helper: query fleet ingest API for latest device_status ────── def _query_fleet_telemetry(): - """Query fleet Postgres for the latest telemetry data for all devices. - Returns a dict keyed by device_id, or empty dict if DB unreachable.""" + """Fetch device status from the fleet ingest API. + Returns a dict keyed by device_id, or empty dict on error.""" try: - import psycopg2 - conn = psycopg2.connect( - host=FLEET_DB_HOST, - port=FLEET_DB_PORT, - dbname=FLEET_DB_NAME, - user=FLEET_DB_USER, - password=FLEET_DB_PASS, - connect_timeout=5, - ) - cur = conn.cursor() - cur.execute(""" - SELECT d.device_id, d.vendor, d.model, - st.online, st.ts, st.version, st.uptime_s, - st.primary_member, st.active_wan, - st.gps_lat, st.gps_lon, st.gps_fix, - st.wan, st.starlink, st.state - FROM devices d - LEFT JOIN device_status st ON st.device_id = d.device_id - ORDER BY d.device_id - """) - rows = cur.fetchall() - cur.close() - conn.close() - + with urllib.request.urlopen(FLEET_API, timeout=5) as resp: + rows = json.loads(resp.read()) result = {} for row in rows: - device_id = row[0] - wan_data = row[12] if row[12] else {} - # psycopg2 returns JSONB as dict - if isinstance(wan_data, str): - try: - wan_data = json.loads(wan_data) - except (json.JSONDecodeError, TypeError): - wan_data = {} - - starlink_data = row[13] if row[13] else {} - if isinstance(starlink_data, str): - try: - starlink_data = json.loads(starlink_data) - except (json.JSONDecodeError, TypeError): - starlink_data = {} - - state_data = row[14] if row[14] else {} - if isinstance(state_data, str): - try: - state_data = json.loads(state_data) - except (json.JSONDecodeError, TypeError): - state_data = {} - - result[device_id] = { - "vendor": row[1], - "model": row[2], - "online": row[3] if row[3] is not None else False, - "ts": str(row[4]) if row[4] else None, - "version": row[5], - "uptime_s": row[6], - "primary_member": row[7], - "active_wan": row[8], + dev = row.get("device_id") + if not dev: + continue + wan = row.get("wan") or {} + if isinstance(wan, str): + wan = json.loads(wan) if wan else {} + starlink = row.get("starlink") or {} + if isinstance(starlink, str): + starlink = json.loads(starlink) if starlink else {} + state = row.get("state") or {} + if isinstance(state, str): + state = json.loads(state) if state else {} + result[dev] = { + "vendor": row.get("vendor"), + "model": row.get("model"), + "online": row.get("online") or False, + "ts": row.get("ts"), + "version": row.get("version"), + "uptime_s": row.get("uptime_s"), + "primary_member": row.get("primary_member"), + "active_wan": row.get("active_wan"), "gps": { - "lat": float(row[9]) if row[9] is not None else None, - "lon": float(row[10]) if row[10] is not None else None, - "fix": row[11], + "lat": row.get("gps_lat"), + "lon": row.get("gps_lon"), + "fix": row.get("gps_fix"), }, - "wan": wan_data, - "starlink": starlink_data, - "state": state_data, + "wan": wan, + "starlink": starlink, + "state": state, } return result except Exception as e: diff --git a/package/kit-busrouter/files/usr/lib/busrouter/telemetry-synology.sh b/package/kit-busrouter/files/usr/lib/busrouter/telemetry-synology.sh index b28dd06..310a851 100755 --- a/package/kit-busrouter/files/usr/lib/busrouter/telemetry-synology.sh +++ b/package/kit-busrouter/files/usr/lib/busrouter/telemetry-synology.sh @@ -286,8 +286,14 @@ _collect_wan() { _jitter=$(echo "$_ping_result" | awk '{print $2}') _loss=$(echo "$_ping_result" | awk '{print $3}') _score=$(_compute_score "$_lat" "$_loss") - _dl="0.0" - _ul="0.0" + # Use cached speedtest result if available and fresh + _st=$(_cached_speedtest "$_iface" 2>/dev/null) + if [ -n "$_st" ]; then + _dl=$(echo "$_st" | awk '{print $1}') + _ul=$(echo "$_st" | awk '{print $2}') + fi + [ -z "$_dl" ] || [ "$_dl" = "0" ] && _dl="0.0" + [ -z "$_ul" ] || [ "$_ul" = "0" ] && _ul="0.0" fi printf '{"score":%s,"latency_ms":%s,"dl_mbps":%s,"ul_mbps":%s,"jitter_ms":%s,"loss_pct":%s}' \ @@ -364,10 +370,10 @@ _eyeride_signal() { } # ── Speedtest ────────────────────────────────────────────────────────────── -# Run a speedtest bound to a specific interface. Reports dl_mbps ul_mbps. +# Run a speedtest bound to a specific interface. Caches result per-interface. run_speedtest() { _iface="$1" - _result="0 0" + _result_file="${STATE_DIR}/speedtest-result-${_iface}" # Try speedtest-cli first if which speedtest-cli >/dev/null 2>&1; then _out=$(speedtest-cli --interface "$_iface" --json 2>/dev/null) @@ -377,27 +383,16 @@ run_speedtest() { if [ -n "$_dl" ] && [ -n "$_ul" ]; then _dl_mbps=$(awk "BEGIN { printf \"%.1f\", ${_dl} / 125000 }" 2>/dev/null) _ul_mbps=$(awk "BEGIN { printf \"%.1f\", ${_ul} / 125000 }" 2>/dev/null) + printf '%s %s\n' "${_dl_mbps}" "${_ul_mbps}" > "$_result_file" echo "${_dl_mbps} ${_ul_mbps}" return 0 fi fi fi - # Fallback: iperf3 single-stream test to public server - if which iperf3 >/dev/null 2>&1; then - _out=$(timeout 15 iperf3 -B "$_ip" -c iperf.he.net -p 5201 -J 2>/dev/null) - if [ -n "$_out" ]; then - _dl=$(_json_get_num "$_out" "bits_per_second") - if [ -n "$_dl" ]; then - _dl_mbps=$(awk "BEGIN { printf \"%.1f\", ${_dl} / 1000000 }" 2>/dev/null) - echo "${_dl_mbps} 0" - return 0 - fi - fi - fi # Fallback: curl a known file and measure throughput _tmp="/tmp/speedtest-$$" _start=$(date +%s) - if curl -s --interface "$_iface" --max-time 10 -o "$_tmp" http://speedtest.tele2.net/1MB.zip 2>/dev/null; then + if curl -s --interface "$_iface" --max-time 15 -o "$_tmp" http://speedtest.tele2.net/10MB.zip 2>/dev/null; then _end=$(date +%s) _bytes=$(wc -c < "$_tmp" 2>/dev/null) rm -f "$_tmp" @@ -405,6 +400,7 @@ run_speedtest() { _duration=$(( _end - _start )) [ "$_duration" -lt 1 ] && _duration=1 _dl_mbps=$(awk "BEGIN { printf \"%.1f\", ${_bytes} * 8 / ${_duration} / 1000000 }" 2>/dev/null) + printf '%s 0\n' "${_dl_mbps:-0}" > "$_result_file" echo "${_dl_mbps:-0} 0" return 0 fi @@ -413,6 +409,18 @@ run_speedtest() { echo "0 0" } +# Read cached speedtest result for an interface if < 35 minutes old. +_cached_speedtest() { + _iface="$1" + _f="${STATE_DIR}/speedtest-result-${_iface}" + [ -f "$_f" ] || return 1 + _mtime=$(stat -c %Y "$_f" 2>/dev/null || stat -f %m "$_f" 2>/dev/null) + [ -n "$_mtime" ] || return 1 + _age=$(( $(date +%s) - _mtime )) + [ "$_age" -lt 2100 ] || return 1 + cat "$_f" +} + # ── Telemetry Collection ─────────────────────────────────────────────────── telemetry_collect() { now=$(date +%s)