alexpires.me/a13labs.infra
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(devolo-exporter): adjust scrape interval and enhance health check functionality

Browse Source
This commit is contained in:
Alexandre Pires
2025-10-14 22:43:02 +02:00
parent d069fe3585
commit 161efa5230
4 changed files with 207 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
terraform/apps/dev-01/resources/prometheus.yml
+1 -1
View File
@@ -22,7 +22,7 @@ scrape_configs:
- labels:
cluster: "dev-01"
- job_name: devolo-exporter
scrape_interval: 10s
scrape_interval: 30s
static_configs:
- targets:
- "devolo-exporter-620.prometheus.svc.cluster.local:9100"
terraform/modules/utils/devolo-expporter/main.tf
+8 -5
View File
@@ -75,9 +75,9 @@ resource "kubernetes_deployment" "devolo_exporter" {
path = "/healthz"
port = var.exporter_port
}
initial_delay_seconds = 10
period_seconds = 10
timeout_seconds = 5
initial_delay_seconds = 15
period_seconds = 60
timeout_seconds = 10
failure_threshold = 3
}
@@ -86,7 +86,7 @@ resource "kubernetes_deployment" "devolo_exporter" {
path = "/healthz"
port = var.exporter_port
}
initial_delay_seconds = 5
initial_delay_seconds = 10
period_seconds = 10
timeout_seconds = 5
failure_threshold = 3
@@ -126,6 +126,9 @@ resource "kubernetes_deployment" "devolo_exporter" {
"--disable-dev-shm-usage",
"--disable-extensions",
"--disable-gpu",
"--enable-low-end-device-mode",
"--memory-pressure-thresholds=moderate:512,critical:256",
"--force-device-scale-factor=1",
"--disable-popup-blocking",
"--disable-prompt-on-repost",
"--disable-sync",
@@ -144,7 +147,7 @@ resource "kubernetes_deployment" "devolo_exporter" {
"--remote-allow-origins=http://localhost:9222",
"--safebrowsing-disable-auto-update",
"--user-data-dir=/home/chromium",
"http://${var.devolo_ip}/#/overview"
"about:blank"
]
# Optional: expose the DevTools port within the pod (not needed outside)
terraform/modules/utils/devolo-expporter/resources/exporter_326.py
+64 -26
View File
@@ -223,6 +223,70 @@ def _parse_uptime_seconds(s: str) -> float:
return float("nan")
def _health_check() -> tuple[bool, str]:
"""Quick health check for liveness/readiness.
Returns (ok, message). ok=False with a short message describing the first failure.
Criteria:
- Chromium DevTools endpoint reachable on localhost:RD_PORT
- Device HTTP reachable on http://DEVOLO_IP
"""
# 1) Check Chromium DevTools
try:
r = requests.get(f"http://localhost:{RD_PORT}/json", timeout=2)
if r.status_code >= 400:
return False, f"devtools http {r.status_code}"
targets = r.json()
if not isinstance(targets, list) or not targets:
return False, "devtools no targets"
ws_url = targets[0].get("webSocketDebuggerUrl")
if not ws_url:
return False, "devtools no ws url"
# Try executing a trivial command via WebSocket to ensure DevTools is functional
try:
ws = websocket.create_connection(ws_url, timeout=2)
ws.settimeout(2)
# Enable Runtime and evaluate a simple expression
ws.send(json.dumps({"id": 1, "method": "Runtime.enable"}))
# Drain until id 1 or timeout
while True:
msg = json.loads(ws.recv())
if msg.get("id") == 1:
break
ws.send(json.dumps({
"id": 2,
"method": "Runtime.evaluate",
"params": {"expression": "1+1", "returnByValue": True}
}))
ok_eval = False
while True:
msg = json.loads(ws.recv())
if msg.get("id") == 2:
val = (
msg.get("result", {})
.get("result", {})
.get("value")
)
ok_eval = (val == 2)
break
except Exception as we: # noqa: BLE001
try:
ws.close() # type: ignore[name-defined]
except Exception:
pass
return False, f"devtools ws error: {we}"
finally:
try:
ws.close()
except Exception:
pass
if not ok_eval:
return False, "devtools evaluate failed"
except Exception as e: # noqa: BLE001
return False, f"devtools error: {e}"
class MetricsHandler(BaseHTTPRequestHandler):
registry = CollectorRegistry()
dev_tx = Gauge(
@@ -343,31 +407,5 @@ def run_server():
pass
scraper.close()
def _health_check() -> tuple[bool, str]:
"""Quick health check for liveness/readiness.
Returns (ok, message). ok=False with a short message describing the first failure.
Criteria:
- Chromium DevTools endpoint reachable on localhost:RD_PORT
- Device HTTP reachable on http://DEVOLO_IP
"""
try:
r = requests.get(f"http://localhost:{RD_PORT}/json", timeout=2)
if r.status_code >= 400:
return False, f"devtools http {r.status_code}"
except Exception as e: # noqa: BLE001
return False, f"devtools error: {e}"
try:
r = requests.get(f"http://{DEVOLO_IP}", timeout=3)
if r.status_code >= 400:
return False, f"device http {r.status_code}"
except Exception as e: # noqa: BLE001
return False, f"device error: {e}"
return True, "ok"
if __name__ == "__main__":
run_server()
terraform/modules/utils/devolo-expporter/resources/exporter_620.py
+134 -102
View File
@@ -15,7 +15,7 @@ from prometheus_client import CollectorRegistry, Gauge, generate_latest, CONTENT
# Configuration via environment
DEVOLO_IP = os.getenv("DEVOLO_IP")
DEVOLO_IP = os.getenv("DEVOLO_IP","192.168.0.199")
RD_PORT = int(os.getenv("RD_PORT", "9222"))
EXPORTER_PORT = int(os.getenv("EXPORTER_PORT", "9100"))
@@ -28,57 +28,77 @@ class DevoloScraper:
def __init__(self, devolo_ip: str, rd_port: int) -> None:
self.devolo_ip = devolo_ip
self.rd_port = rd_port
self.ws = None
self.msg_id = 0
self.lock = threading.Lock()
self._id_lock = threading.Lock()
def _next_id(self) -> int:
with self.lock:
self.msg_id += 1
return self.msg_id
def _open_ephemeral(self, url: str):
"""Create a fresh temporary target for the given URL and return (ws, target_id).
def _connect(self) -> None:
# Find Chrome DevTools WebSocket endpoint
targets = requests.get(f"http://localhost:{self.rd_port}/json", timeout=5).json()
if not targets:
raise RuntimeError("No DevTools targets found")
ws_url = targets[0]["webSocketDebuggerUrl"]
self.ws = websocket.create_connection(ws_url, timeout=10)
The caller is responsible for closing the target and the WebSocket.
"""
info = requests.put(
f"http://localhost:{self.rd_port}/json/new?{url}", timeout=5
)
if info.status_code >= 400:
raise RuntimeError(f"Failed to create DevTools target: http {info.status_code}")
info = info.json()
ws_url = info.get("webSocketDebuggerUrl")
target_id = info.get("id")
if not ws_url or not target_id:
raise RuntimeError("Failed to create DevTools target")
ws = websocket.create_connection(ws_url, timeout=10)
# Enable Runtime domain
self._send({"method": "Runtime.enable"})
# Enable Page domain for navigation
self._send({"method": "Page.enable"})
# Simple send/recv helper with incremental ids per connection
msg_id = 0
def _send(self, payload: dict) -> dict:
if self.ws is None:
self._connect()
msg_id = self._next_id()
payload = {"id": msg_id, **payload}
self.ws.send(json.dumps(payload))
return self._recv_until_id(msg_id)
def send(method: str, params: dict | None = None, timeout: int = 10) -> dict:
nonlocal msg_id
with self._id_lock:
msg_id += 1
cur_id = msg_id
ws.settimeout(timeout)
ws.send(json.dumps({"id": cur_id, "method": method, **({"params": params} if params else {})}))
while True:
resp = json.loads(ws.recv())
if resp.get("id") == cur_id:
return resp
def _recv_until_id(self, expected_id: int, timeout: int = 10) -> dict:
self.ws.settimeout(timeout)
while True:
msg = json.loads(self.ws.recv())
if "id" not in msg:
# Log events but continue
continue
if msg["id"] == expected_id:
return msg
# Enable minimal domains and disable network cache to reduce memory/caching
send("Runtime.enable")
send("Page.enable")
try:
send("Network.enable")
send("Network.setCacheDisabled", {"cacheDisabled": True})
except Exception:
# Network domain might not be available in some builds; ignore
pass
return ws, target_id, send
def _close_target(self, target_id: str) -> None:
try:
requests.get(f"http://localhost:{self.rd_port}/json/close/{target_id}", timeout=5)
except Exception:
pass
def scrape(self) -> list[dict]:
try:
# Navigate to the Devolo page
self._send({
"method": "Page.navigate",
"params": {"url": f"http://{self.devolo_ip}/#/powerline"},
})
# Open a fresh ephemeral tab for each scrape to avoid memory growth
ws, target_id, send = self._open_ephemeral(f"http://{self.devolo_ip}/#/overview")
time.sleep(2)
# Evaluate JavaScript to extract device data and return by value
script = """
# Navigate to powerline page
send(
"Page.navigate",
{
"url": f"http://{self.devolo_ip}/#/powerline"
}
)
try:
time.sleep(2)
# Evaluate JavaScript to extract device data and return by value
script = """
Array.from(document.querySelectorAll('tr[id^="plc-"]')).map(tr => ({
id: tr.querySelector('[id^="plc-id-"]')?.innerText?.trim(),
mac: tr.querySelector('[id^="plc-mac-"]')?.innerText?.trim(),
@@ -86,54 +106,53 @@ Array.from(document.querySelectorAll('tr[id^="plc-"]')).map(tr => ({
rx: tr.querySelector('[id^="plc-rx-"]')?.innerText?.trim()
}))
"""
resp = self._send({
"method": "Runtime.evaluate",
"params": {
"expression": script,
"returnByValue": True,
"awaitPromise": True,
},
})
value = resp.get("result", {}).get("result", {}).get("value")
if isinstance(value, list):
# Normalize values
out = []
for it in value:
if not isinstance(it, dict):
continue
dev_id = (it.get("id") or "").strip()
if not dev_id or "(this device)" in dev_id.lower():
continue
out.append(
{
"id": (it.get("id") or "").strip(),
"mac": (it.get("mac") or "").strip(),
"tx": _to_float(it.get("tx")),
"rx": _to_float(it.get("rx")),
}
)
return out
return []
except Exception as e: # noqa: BLE001
# Reset connection next time
try:
if self.ws is not None:
self.ws.close()
resp = send(
"Runtime.evaluate",
{
"expression": script,
"returnByValue": True,
"awaitPromise": True,
},
)
print("Scrape response:", resp)
value = resp.get("result", {}).get("result", {}).get("value")
if isinstance(value, list):
# Normalize values
out = []
for it in value:
if not isinstance(it, dict):
continue
dev_id = (it.get("id") or "").strip()
if not dev_id or "(this device)" in dev_id.lower():
continue
out.append(
{
"id": (it.get("id") or "").strip(),
"mac": (it.get("mac") or "").strip(),
"tx": _to_float(it.get("tx")),
"rx": _to_float(it.get("rx")),
}
)
return out
return []
finally:
self.ws = None
try:
ws.close()
except Exception:
pass
self._close_target(target_id)
except Exception as e: # noqa: BLE001
print(f"Scrape error: {e}", file=sys.stderr)
return []
def scrape_overview(self) -> dict:
try:
# Open a fresh ephemeral tab for overview
ws, target_id, send = self._open_ephemeral(f"http://{self.devolo_ip}/#/overview")
try:
# Navigate to the Overview page
self._send({
"method": "Page.navigate",
"params": {"url": f"http://{self.devolo_ip}/#/overview"},
})
time.sleep(2)
time.sleep(2)
overview_js = """
overview_js = """
(() => {
const byId = (id) => {
const el = document.getElementById(id);
@@ -148,26 +167,29 @@ return {
};
})()
"""
resp = self._send({
"method": "Runtime.evaluate",
"params": {
"expression": overview_js,
"returnByValue": True,
"awaitPromise": True,
},
})
value = resp.get("result", {}).get("result", {}).get("value")
return value if isinstance(value, dict) else {}
except Exception as e: # noqa: BLE001
print(f"Overview scrape error: {e}", file=sys.stderr)
return {}
resp = send(
"Runtime.evaluate",
{
"expression": overview_js,
"returnByValue": True,
"awaitPromise": True,
},
)
value = resp.get("result", {}).get("result", {}).get("value")
return value if isinstance(value, dict) else {}
finally:
try:
ws.close()
except Exception:
pass
self._close_target(target_id)
except Exception as e: # noqa: BLE001
print(f"Overview scrape error: {e}", file=sys.stderr)
return {}
def close(self) -> None:
if self.ws is not None:
try:
self.ws.close()
finally:
self.ws = None
# No persistent connection held anymore
return None
def _to_float(v):
@@ -206,7 +228,6 @@ def _parse_uptime_seconds(s: str) -> float:
return float("nan")
def _health_check() -> tuple[bool, str]:
"""Quick health check for liveness/readiness.
@@ -220,6 +241,12 @@ def _health_check() -> tuple[bool, str]:
r = requests.get(f"http://localhost:{RD_PORT}/json", timeout=2)
if r.status_code >= 400:
return False, f"devtools http {r.status_code}"
try:
info = r.json()
if not isinstance(info, list) or len(info) == 0:
return False, "devtools invalid json"
except Exception: # noqa: BLE001
return False, "devtools invalid json"
except Exception as e: # noqa: BLE001
return False, f"devtools error: {e}"
@@ -264,6 +291,11 @@ class MetricsHandler(BaseHTTPRequestHandler):
ok, msg = _health_check()
status = 200 if ok else 503
body = ("ok" if ok else msg).encode()
if not ok:
try:
print(f"healthz failed: {msg}")
except Exception:
pass
self.send_response(status)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(body)))