Report stalled agent lanes and bound health lock waits

agent/ec_agent.py

@@ -33,7 +33,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.5.2"
+VERSION = "1.5.3"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -41,6 +41,7 @@ MANIFEST_MAX_BYTES = 16 * 1024 * 1024
 MANIFEST_MAX_SECONDS = 15
 MANIFEST_MAX_DEPTH = 64
 LIBRARY_DOWNLOAD_WORKERS = 4
+HEALTH_LOCK_TIMEOUT_SECONDS = 5
 ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
 AGENT_CONFIG = Path("/etc/electric-cherry/agent.json")
 AGENT_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-agent.py")
@@ -923,6 +924,28 @@ class Agent:
         self.inventory_manifest_lock = threading.Lock()
         self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
         self.executing_id = None
+        self.lane_lock = threading.Lock()
+        self.lanes = {name: {"thread": None, "lastPollAt": None, "polled": time.monotonic(), "busy": False}
+                      for name in ("health-1", "health-2", "read", "write")}
+
+    def lane_started(self, name):
+        with self.lane_lock:
+            self.lanes[name]["thread"] = threading.current_thread()
+
+    def lane_polled(self, name):
+        with self.lane_lock:
+            self.lanes[name].update(lastPollAt=datetime.now(timezone.utc).isoformat(timespec="milliseconds"), polled=time.monotonic())
+
+    def lane_busy(self, name, busy):
+        with self.lane_lock:
+            self.lanes[name]["busy"] = busy
+
+    def lane_inventory(self):
+        with self.lane_lock:
+            return [{"name": name, "alive": lane["thread"] is None or lane["thread"].is_alive(),
+                     "busy": lane["busy"], "lastPollAt": lane["lastPollAt"],
+                     "pollAgeSeconds": max(0, time.monotonic() - lane["polled"])}
+                    for name, lane in self.lanes.items()]
 
     def can_uninstall(self):
         try:
@@ -1334,7 +1357,7 @@ class Agent:
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
         capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
-        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
+        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
 
     def safe_to_disrupt(self, instance_id, generation):
         health = self.inspect(instance_id, generation)
@@ -2366,13 +2389,29 @@ class Agent:
         key = operation.get("instanceId")
         return self.instance_locks.setdefault(key, threading.RLock()) if key else contextlib.nullcontext()
 
+    @contextlib.contextmanager
+    def observation_lock(self, operation, health=False):
+        lock = self.instance_lock(operation)
+        if not health or not operation.get("instanceId"):
+            with lock:
+                yield
+            return
+        if not lock.acquire(timeout=HEALTH_LOCK_TIMEOUT_SECONDS):
+            raise AgentError("Health check timed out waiting for the instance operation")
+        try:
+            yield
+        finally:
+            lock.release()
+
     @staticmethod
     def health_operation(operation):
         return operation.get("kind") in ("instance.inspect", "rcon.status-batch") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
 
-    def read_loop(self, lane="read"):
+    def read_loop(self, lane="read", lane_name=None):
         # Each lane owns its SQLite connection and never writes executing_id,
         # checkpoints, or current_operation used by the mutation state machine.
+        lane_name = lane_name or ("health-1" if lane == "health" else "read")
+        self.lane_started(lane_name)
         journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False)
         next_prune = time.monotonic()
         try:
@@ -2388,7 +2427,9 @@ class Agent:
                                 if journal.prune_observations(self.observation_keep_days) == 500:
                                     next_prune = time.monotonic() + 15
                     response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20")
+                    self.lane_polled(lane_name)
                     handled = False
+                    self.lane_busy(lane_name, bool(response.get("operations")))
                     for operation in response.get("operations", []):
                         if self.stopping.is_set() or not self.read_operation(operation):
                             continue
@@ -2405,12 +2446,12 @@ class Agent:
                                 row, _ = journal.accept(operation)
                                 if row["state"] in ("accepted", "unknown", "running"):
                                     self.report(journal.update(operation["id"], "running"), journal=journal)
-                                    with self.read_gate.read(), self.instance_lock(operation):
-                                        try:
+                                    try:
+                                        with self.read_gate.read(), self.observation_lock(operation, health=lane == "health"):
                                             result = self.dispatch(operation)
                                             row = journal.update(operation["id"], "completed", result)
-                                        except Exception as err:
-                                            row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable")
+                                    except Exception as err:
+                                        row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable")
                                 # Receipt delivery can take seconds; it must not
                                 # hold the instance lock needed by health checks.
                                 self.report(row, journal=journal)
@@ -2427,6 +2468,8 @@ class Agent:
                 except Exception as err:
                     print(f"{lane} lane recovered from {type(err).__name__}", file=sys.stderr)
                     self.stopping.wait(3)
+                finally:
+                    self.lane_busy(lane_name, False)
         finally:
             journal.close()
 
@@ -2440,6 +2483,8 @@ class Agent:
                 return
         # Never hold the updater lock while waiting for delivery.
         response = self.request("GET", "/api/agent/operations?lane=write&wait=20")
+        self.lane_polled("write")
+        self.lane_busy("write", bool(response.get("operations")))
         with self.activity() as acquired:
             if not acquired or self.stopping.is_set() or self.update_pending():
                 return
@@ -2467,6 +2512,7 @@ class Agent:
             return completed
 
     def run(self):
+        self.lane_started("write")
         # Obtain the configured host port range before accepting any deployment.
         # Enrollment's default values must never race the first heartbeat.
         while not self.stopping.is_set():
@@ -2487,7 +2533,8 @@ class Agent:
         heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True)
         heartbeats.start()
         # Preserve two concurrent health checks even while files are being read.
-        readers = [threading.Thread(target=self.read_loop, args=(lane,), daemon=True) for lane in ("health", "health", "read")]
+        readers = [threading.Thread(target=self.read_loop, args=(lane, name), daemon=True)
+                   for name, lane in (("health-1", "health"), ("health-2", "health"), ("read", "read"))]
         for reader in readers:
             reader.start()
         while not self.stopping.is_set():
@@ -2499,6 +2546,8 @@ class Agent:
                 print("Host controller contact unavailable; existing containers continue running", file=sys.stderr)
             except Exception as err:
                 print(f"Write lane recovered from {type(err).__name__}", file=sys.stderr)
+            finally:
+                self.lane_busy("write", False)
             self.stopping.wait(0.25 if completed else 3)
         for reader in readers:
             reader.join(timeout=2)

agent/tests/test_agent.py

@@ -446,6 +446,67 @@ class AgentTests(unittest.TestCase):
         self.assertEqual(logs.getvalue().count("RuntimeError"), 2)
         self.assertNotIn("secret", logs.getvalue())
 
+    def test_lane_inventory_tracks_each_health_worker_and_detects_dead_threads(self):
+        def poll():
+            self.agent.lane_started("health-1")
+            self.agent.lane_polled("health-1")
+
+        with patch.object(module.time, "monotonic", return_value=100):
+            thread = threading.Thread(target=poll)
+            thread.start()
+            thread.join(timeout=2)
+        self.assertFalse(thread.is_alive())
+        with patch.object(module.time, "monotonic", return_value=150):
+            self.agent.lane_started("health-2")
+            self.agent.lane_polled("health-2")
+            self.agent.lane_busy("health-2", True)
+        with patch.object(module.time, "monotonic", return_value=200):
+            lanes = {lane["name"]: lane for lane in self.agent.lane_inventory()}
+        self.assertEqual(set(lanes), {"health-1", "health-2", "read", "write"})
+        self.assertFalse(lanes["health-1"]["alive"])
+        self.assertEqual(lanes["health-1"]["pollAgeSeconds"], 100)
+        self.assertTrue(lanes["health-2"]["alive"])
+        self.assertTrue(lanes["health-2"]["busy"])
+        self.assertEqual(lanes["health-2"]["pollAgeSeconds"], 50)
+        self.assertIsNotNone(lanes["health-2"]["lastPollAt"])
+        self.assertEqual(len(self.agent.inventory()["agentLanes"]), 4)
+
+    def test_health_lane_times_out_on_instance_lock_and_processes_next_check(self):
+        blocked = self.operation("instance.inspect", op_id="blocked-health")
+        fresh = {**self.operation("instance.inspect", op_id="fresh-health"), "instanceId": "other-instance"}
+        pending = iter([blocked, fresh])
+        failures = []
+
+        def request(method, path, payload=None):
+            if method == "POST":
+                return {"operation": {"id": path.rsplit("/", 1)[1], "state": payload["state"]}}
+            operation = next(pending, None)
+            if operation is None:
+                self.agent.stopping.set()
+            return {"operations": [operation] if operation else []}
+
+        def read():
+            try:
+                self.agent.read_loop("health")
+            except BaseException as error:
+                failures.append(error)
+
+        with patch.object(module, "HEALTH_LOCK_TIMEOUT_SECONDS", 0.01), patch.object(self.agent, "request", side_effect=request), patch.object(self.agent, "dispatch", return_value={"healthy": True}) as dispatch:
+            with self.agent.instance_lock(blocked):
+                thread = threading.Thread(target=read)
+                thread.start()
+                thread.join(timeout=2)
+                finished_while_locked = not thread.is_alive()
+            self.agent.stopping.set()
+            thread.join(timeout=2)
+        self.assertTrue(finished_while_locked)
+        self.assertEqual(failures, [])
+        self.assertEqual(self.agent.journal.get(blocked["id"])["state"], "failed")
+        self.assertIn("waiting for the instance", self.agent.journal.get(blocked["id"])["error"])
+        self.assertEqual(self.agent.journal.get(fresh["id"])["state"], "completed")
+        dispatch.assert_called_once_with(fresh)
+        self.assertEqual(self.agent.read_claims, set())
+
     def test_inventory_reuses_manifests_until_mtime_changes(self):
         marker = self.library / "manifest.json"
         saved = json.loads(marker.read_text())