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-<iface>); _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 <noreply@anthropic.com>
This commit is contained in:
+34
-72
@@ -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", "")
|
||||
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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user