diff --git a/.github/workflows/docker-build.yml b/.github/workflows/docker-build.yml index e80fc17..4c324f6 100644 --- a/.github/workflows/docker-build.yml +++ b/.github/workflows/docker-build.yml @@ -3,6 +3,8 @@ name: Build and Push on: push: branches: [main] + pull_request: + branches: [main] env: REGISTRY: ghcr.io @@ -11,7 +13,22 @@ env: IMAGE_NAME: programmer-network/tapo-exporter jobs: + # The poller talks to real hardware, so the only thing worth testing is its + # error handling — and that is exactly where the ~50% poll-success bug lived. + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: pip install --no-cache-dir prometheus-client==0.21.1 + - run: python3 test_poller.py + build: + needs: test + # Pull requests only validate; publishing :latest happens on main. + if: github.event_name == 'push' runs-on: ubuntu-latest permissions: contents: read diff --git a/__pycache__/tapo-exporter.cpython-314.pyc b/__pycache__/tapo-exporter.cpython-314.pyc new file mode 100644 index 0000000..86db3e3 Binary files /dev/null and b/__pycache__/tapo-exporter.cpython-314.pyc differ diff --git a/tapo-exporter.py b/tapo-exporter.py index d8f05c5..795acda 100755 --- a/tapo-exporter.py +++ b/tapo-exporter.py @@ -13,9 +13,10 @@ import logging import os import sys +import time from tapo import ApiClient -from prometheus_client import CollectorRegistry, Gauge, start_http_server +from prometheus_client import CollectorRegistry, Counter, Gauge, start_http_server logging.basicConfig( level=logging.INFO, @@ -41,6 +42,12 @@ metric_on_since = Gauge("tapo_plug_on_since_seconds", "Seconds since powered on", ["plug"], registry=registry) metric_today = Gauge("tapo_plug_energy_today_kwh", "Energy today in kWh", ["plug"], registry=registry) metric_month = Gauge("tapo_plug_energy_month_kwh", "Energy this month in kWh", ["plug"], registry=registry) +# On failure the gauges above keep their last good values rather than going +# stale-but-plausible in silence. This is how a consumer tells the difference: +# time() - tapo_plug_last_success_seconds is the true age of every reading. +metric_last_ok = Gauge("tapo_plug_last_success_seconds", "Unix time of the last successful poll", ["plug"], registry=registry) +metric_reauths = Counter("tapo_plug_reauth_total", "Mid-poll re-handshakes after a dropped session", ["plug"], registry=registry) +metric_fails = Counter("tapo_plug_poll_failures_total", "Polls that failed even after a re-handshake", ["plug"], registry=registry) def load_plugs() -> dict: @@ -56,22 +63,45 @@ def load_plugs() -> dict: class PlugPoller: - """Holds a reused device session per plug; re-handshakes only on error.""" + """Holds a reused device session per plug, re-handshaking in place on error. - def __init__(self, client: ApiClient, name: str, host: str): - self.client = client + Each poller owns its own ApiClient. Sharing one across plugs makes their + handshakes race — the plugs answer 403 Forbidden for every request carrying + a session another plug's handshake has since superseded. + """ + + def __init__(self, user: str, password: str, name: str, host: str): + self.client = ApiClient(user, password) self.name = name self.host = host self.device = None + async def _read(self): + """One full read against the current session, handshaking if needed.""" + if self.device is None: + self.device = await asyncio.wait_for(self.client.p110(self.host), OP_TIMEOUT) + info = await asyncio.wait_for(self.device.get_device_info(), OP_TIMEOUT) + energy = await asyncio.wait_for(self.device.get_energy_usage(), OP_TIMEOUT) + power = await asyncio.wait_for(self.device.get_current_power(), OP_TIMEOUT) + return info, energy, power + async def poll(self): labels = {"plug": self.name} try: - if self.device is None: - self.device = await asyncio.wait_for(self.client.p110(self.host), OP_TIMEOUT) - info = await asyncio.wait_for(self.device.get_device_info(), OP_TIMEOUT) - energy = await asyncio.wait_for(self.device.get_energy_usage(), OP_TIMEOUT) - power = await asyncio.wait_for(self.device.get_current_power(), OP_TIMEOUT) + try: + info, energy, power = await self._read() + except Exception as first: + # A dropped session is the overwhelmingly common failure here and + # it is recoverable immediately — the plug is fine, our token is + # not. Retrying on the NEXT cycle instead of this one is what + # pinned poll success at ~50%: every session survived exactly one + # poll, so the loop alternated success, expiry, re-handshake, + # expiry, forever. Reconnect and retry once, in place. + log.info("session lost for plug %s (%s): %s — re-handshaking", + self.name, self.host, first) + metric_reauths.labels(**labels).inc() + self.device = None + info, energy, power = await self._read() metric_up.labels(**labels).set(1) metric_state.labels(**labels).set(1 if info.device_on else 0) @@ -80,10 +110,16 @@ async def poll(self): metric_on_since.labels(**labels).set(info.on_time) metric_today.labels(**labels).set(energy.today_energy / 1000.0) # Wh -> kWh metric_month.labels(**labels).set(energy.month_energy / 1000.0) # Wh -> kWh + metric_last_ok.labels(**labels).set(time.time()) except Exception as e: + # Both the original attempt and the retry failed, so this is not a + # stale session — the plug is genuinely unreachable, off the Wi-Fi, + # or the credentials are wrong. metric_up.labels(**labels).set(0) + metric_fails.labels(**labels).inc() self.device = None # force a fresh handshake next cycle - log.warning("poll failed for plug %s (%s): %s", self.name, self.host, e) + log.warning("poll failed for plug %s (%s) after re-handshake: %s", + self.name, self.host, e) async def poll_loop(pollers, interval: int): @@ -105,8 +141,7 @@ def main(): sys.exit(1) plugs = load_plugs() - client = ApiClient(user, password) - pollers = [PlugPoller(client, name, host) for name, host in plugs.items()] + pollers = [PlugPoller(user, password, name, host) for name, host in plugs.items()] # Start the HTTP server FIRST (daemon thread) so /metrics — and thus the k8s # readiness probe — is serving immediately, even if the plugs are unreachable. diff --git a/test_poller.py b/test_poller.py new file mode 100644 index 0000000..d958e8d --- /dev/null +++ b/test_poller.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python3 +"""Regression test for the ~50% poll-success bug. + +In the field, every Tapo session survived exactly one poll and then answered +403 Forbidden / Tapo(SessionTimeout). The old poll() reacted by clearing the +device and giving up until the NEXT cycle, so the loop alternated success, +expiry, re-handshake, expiry — pinning `avg_over_time(tapo_plug_up[3h])` at +0.503 on all three plugs while power gauges kept updating from the good half. + +This test models exactly that plug: a session that dies after one full read. +The poller must absorb it and report every poll as successful. + +Run: python3 test_poller.py (needs prometheus-client; no network, no plugs) +""" +import asyncio +import importlib.util +import os +import sys +import types + + +# ── a plug whose session dies after exactly one complete read ──────────────── +class FakeInfo: + device_on, rssi, on_time = True, -55, 1234 + + +class FakeEnergy: + today_energy, month_energy = 612.0, 18000.0 + + +class FakePower: + current_power = 50.0 + + +class FakeDevice: + def __init__(self): + self.spent = False + + async def _guard(self): + if self.spent: + raise RuntimeError("Tapo(SessionTimeout): 403 Forbidden") + + async def get_device_info(self): + await self._guard() + return FakeInfo() + + async def get_energy_usage(self): + await self._guard() + return FakeEnergy() + + async def get_current_power(self): + await self._guard() + self.spent = True # the read that consumes the session + return FakePower() + + +class FakeApiClient: + def __init__(self, user, password): + self.handshakes = 0 + + async def p110(self, host): + self.handshakes += 1 + return FakeDevice() + + +# The exporter is a single script, not a package — stub `tapo` before loading it. +fake_tapo = types.ModuleType("tapo") +fake_tapo.ApiClient = FakeApiClient +sys.modules["tapo"] = fake_tapo + +_here = os.path.dirname(os.path.abspath(__file__)) +_spec = importlib.util.spec_from_file_location( + "tapo_exporter", os.path.join(_here, "tapo-exporter.py")) +exporter = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(exporter) + +PLUG = "DGX Spark" + + +def gauge(metric): + return metric.labels(plug=PLUG)._value.get() + + +async def test_expiring_session_still_polls_cleanly(): + poller = exporter.PlugPoller("user", "password", PLUG, "192.168.10.217") + + ups = [] + for _ in range(10): + await poller.poll() + ups.append(gauge(exporter.metric_up)) + + rate = sum(ups) / len(ups) + print(f"up per poll : {[int(u) for u in ups]}") + print(f"success rate : {rate:.0%}") + print(f"reauths : {int(gauge(exporter.metric_reauths))}") + print(f"hard failures: {int(gauge(exporter.metric_fails))}") + + assert rate == 1.0, f"expected every poll to succeed, got {rate:.0%}" + assert gauge(exporter.metric_fails) == 0, "a recoverable expiry was reported as a failure" + assert gauge(exporter.metric_power) == 50.0, "power gauge not populated" + assert gauge(exporter.metric_last_ok) > 0, "last-success timestamp not set" + + +async def test_unreachable_plug_is_reported_down(): + """The retry must not paper over a plug that is genuinely gone.""" + class DeadClient(FakeApiClient): + async def p110(self, host): + raise RuntimeError("connection refused") + + poller = exporter.PlugPoller("user", "password", PLUG, "192.168.10.217") + poller.client = DeadClient("user", "password") + before = gauge(exporter.metric_fails) + await poller.poll() + + assert gauge(exporter.metric_up) == 0, "unreachable plug should report up=0" + assert gauge(exporter.metric_fails) == before + 1, "failure counter not incremented" + print("unreachable plug correctly reported down") + + +async def main(): + await test_expiring_session_still_polls_cleanly() + await test_unreachable_plug_is_reported_down() + print("\nOK") + + +if __name__ == "__main__": + asyncio.run(main())