diff --git a/packages/core/canyonos_core/controller/global_controller.py b/packages/core/canyonos_core/controller/global_controller.py index 7e9807be..1d4e92af 100644 --- a/packages/core/canyonos_core/controller/global_controller.py +++ b/packages/core/canyonos_core/controller/global_controller.py @@ -1092,7 +1092,10 @@ def _trigger_cleanup(self): payload = json.dumps({"request_ids": list(all_completed)}) def _send(instance): - endpoint = instance["endpoint"] + # Container-reachable address (runtime_id:CONTAINER_PORT), not + # instance["endpoint"] -- that's the host-published port + # (e.g. localhost:8001), which the GC container can't reach. + endpoint = self.instance_manager._routing_endpoint_for(instance) try: stub = self._get_lc_stub(endpoint) stub.Cleanup(local_controler_pb2.JsonResponse(resonse=payload)) @@ -1101,20 +1104,33 @@ def _send(instance): len(all_completed), endpoint, ) + return True except Exception as e: logger.warning("Failed to trigger cleanup on %s: %s", endpoint, e) + return False instances = self.instance_manager.list_instances() - if instances: - with ThreadPoolExecutor(max_workers=len(instances)) as executor: - list(executor.map(_send, instances)) + if not instances: + logger.warning( + "No instances to broadcast cleanup to; leaving %d request(s) queued.", + len(all_completed), + ) + return + + with ThreadPoolExecutor(max_workers=len(instances)) as executor: + if not all(executor.map(_send, instances)): + logger.warning( + "Cleanup broadcast failed for at least one instance; leaving %d " + "request(s) queued for retry on the next cycle.", + len(all_completed), + ) + return logger.info( "Triggered cleanup for %d completed request(s) across %d node(s)", len(all_completed), len(completed_by_client), ) - # Drain each node's own set from the same client it was read from. for client, completed in completed_by_client.items(): client.srem("request:completed", *completed) diff --git a/packages/core/canyonos_core/controller/local_controller_frontend.py b/packages/core/canyonos_core/controller/local_controller_frontend.py index 5334f1a8..3d829265 100644 --- a/packages/core/canyonos_core/controller/local_controller_frontend.py +++ b/packages/core/canyonos_core/controller/local_controller_frontend.py @@ -22,6 +22,15 @@ logger = logging.getLogger(__name__) +POLL_INTERVAL_SECONDS = float(os.environ.get("CANYONOS_POLL_INTERVAL", 5)) +FUTURE_CLEANUP_GRACE_MULTIPLIER = 3 +FUTURE_CLEANUP_GRACE_MIN_SECONDS = 30 +FUTURE_CLEANUP_GRACE_SECONDS = max( + FUTURE_CLEANUP_GRACE_MIN_SECONDS, + POLL_INTERVAL_SECONDS * FUTURE_CLEANUP_GRACE_MULTIPLIER, +) + + class LocalControllerServicer(local_controler_pb2_grpc.LocalControllerServicer): """gRPC servicer that accepts requests and pushes them into a queue.""" @@ -132,35 +141,22 @@ def _cleanup_request(self, request_id): logger.info("No futures found for request %s on this node.", request_id) return - keys_to_delete = [futures_key] + keys_to_expire = [futures_key] for fid in future_ids: - future_key = f"future:{fid}" - # Delete sibling collection keys (e.g. future:{fid}:consumers) - # but handle the main hash separately to preserve logs. - keys_to_delete.extend(self.redis.scan_keys(f"{future_key}:*")) - - logs = self.redis.hget(future_key, "logs") - if logs: - # Replace the hash with a minimal snapshot so the global - # controller's next poll can persist logs to SQLite before - # they vanish. The TTL guarantees cleanup even if the poll - # never reads it (e.g. controller restarts). - self.redis.delete(future_key) - self.redis.hset_multiple( - future_key, - { - "id": fid, - "request_id": request_id, - "logs": logs, - }, - ) - self.redis.expire(future_key, 30) - else: - keys_to_delete.append(future_key) - - self.redis.delete(*keys_to_delete) + keys_to_expire.extend( + [ + f"future:{fid}", + f"future:{fid}:children", + f"future:{fid}:consumers", + ] + ) + for key in keys_to_expire: + self.redis.expire(key, FUTURE_CLEANUP_GRACE_SECONDS, nx=True) logger.info( - "Cleaned up %d future(s) for request %s", len(future_ids), request_id + "Scheduled %d future(s) for request %s to expire in %ds", + len(future_ids), + request_id, + FUTURE_CLEANUP_GRACE_SECONDS, ) # Clean up affinity bindings for this request diff --git a/packages/core/canyonos_core/controller/utils/redis_client.py b/packages/core/canyonos_core/controller/utils/redis_client.py index 9b52d8ee..3ba079e8 100644 --- a/packages/core/canyonos_core/controller/utils/redis_client.py +++ b/packages/core/canyonos_core/controller/utils/redis_client.py @@ -28,9 +28,9 @@ def setnx(self, key, value): """Set key to value only if it does not already exist. Returns True if set, False otherwise.""" return self.client.setnx(key, value) - def expire(self, key, seconds): + def expire(self, key, seconds, nx=False): """Set a TTL (in seconds) on a key. No-op if the key does not exist.""" - return self.client.expire(key, seconds) + return self.client.expire(key, seconds, nx=nx) # --- Hash operations --- diff --git a/packages/core/tests/test_global_controller_cleanup.py b/packages/core/tests/test_global_controller_cleanup.py index cdb02c8e..3a65da3c 100644 --- a/packages/core/tests/test_global_controller_cleanup.py +++ b/packages/core/tests/test_global_controller_cleanup.py @@ -70,6 +70,15 @@ def __init__(self, instances): def list_instances(self): return self._instances + def _routing_endpoint_for(self, instance): + # Real InstanceManager resolves the container-reachable address + # (runtime_id:CONTAINER_PORT); these fixtures use "endpoint" as + # that already-resolved stand-in, since these tests are about + # batching/draining semantics, not address resolution itself. + # A fixture that needs "endpoint" and the routing address to differ + # (see RoutingEndpointTests) sets "routing_endpoint" explicitly. + return instance.get("routing_endpoint", instance["endpoint"]) + def _bare_controller(redis, instances, node_redis=None): """Build a GlobalController without running its heavy __init__. @@ -115,7 +124,12 @@ def test_sends_one_batched_call_per_instance_not_per_request(self): # Drained after broadcasting, same as before. self.assertEqual(redis.smembers("request:completed"), set()) - def test_one_instance_failing_does_not_block_others_or_stop_draining(self): + def test_one_instance_failing_leaves_the_batch_queued_for_retry(self): + # CAN-391: a batch is only ever removed from "request:completed" once + # every instance has confirmed receipt. If even one Cleanup RPC fails + # (e.g. an unreachable endpoint), the whole batch must stay queued -- + # dropping it here is how cleanup entries went missing and Redis grew + # unbounded. completed = {"reqA", "reqB"} expected = set(completed) # snapshot -- see note in the test above redis = _FakeRedis({"request:completed": completed}) @@ -128,9 +142,11 @@ def test_one_instance_failing_does_not_block_others_or_stop_draining(self): controller._trigger_cleanup() # must not raise + # The reachable instance still gets the batch... self.assertEqual(len(good_stub.calls), 1) self.assertEqual(set(good_stub.calls[0]["request_ids"]), expected) - self.assertEqual(redis.smembers("request:completed"), set()) + # ...but nothing is drained until every instance has confirmed. + self.assertEqual(redis.smembers("request:completed"), expected) def test_noop_when_nothing_completed(self): redis = _FakeRedis() @@ -144,12 +160,13 @@ def test_noop_when_nothing_completed(self): self.assertEqual(stub.calls, []) def test_noop_when_no_instances_registered(self): + # CAN-391: with nothing to broadcast to, nothing has confirmed the + # batch -- it must stay queued rather than being silently dropped. redis = _FakeRedis({"request:completed": {"req1"}}) controller = _bare_controller(redis, []) - # Should still drain the completed set even with nothing to broadcast to. controller._trigger_cleanup() - self.assertEqual(redis.smembers("request:completed"), set()) + self.assertEqual(redis.smembers("request:completed"), {"req1"}) class MultiNodeTriggerCleanupTests(unittest.TestCase): @@ -252,6 +269,70 @@ def test_falls_back_to_self_redis_when_node_redis_is_empty_dict(self): self.assertEqual(redis.smembers("request:completed"), set()) +class RoutingEndpointTests(unittest.TestCase): + """CAN-391: the GC container must send Cleanup to each instance's + container-reachable routing endpoint, never its host-published endpoint + (e.g. localhost:8001) -- the GC can't reach that from inside Docker. + Every other test in this file gives an instance the same value for both, + so a regression back to instance["endpoint"] would pass them unnoticed.""" + + def test_cleanup_uses_routing_endpoint_not_published_endpoint(self): + completed = {"req1"} + redis = _FakeRedis({"request:completed": set(completed)}) + instances = [ + {"endpoint": "localhost:8001", "routing_endpoint": "runtime-abc:50051"} + ] + controller = _bare_controller(redis, instances) + + routing_stub = _FakeStub() + + def _get_lc_stub(endpoint): + if endpoint == "localhost:8001": + raise AssertionError( + "Cleanup must not be sent to the host-published endpoint; " + "the GC container cannot reach it." + ) + return routing_stub + + controller._get_lc_stub = _get_lc_stub + + controller._trigger_cleanup() + + self.assertEqual(len(routing_stub.calls), 1) + self.assertEqual(set(routing_stub.calls[0]["request_ids"]), completed) + self.assertEqual(redis.smembers("request:completed"), set()) + + def test_distinct_endpoints_across_multiple_instances(self): + completed = {"req1", "req2"} + redis = _FakeRedis({"request:completed": set(completed)}) + instances = [ + {"endpoint": f"localhost:800{i}", "routing_endpoint": f"runtime-{i}:50051"} + for i in range(3) + ] + controller = _bare_controller(redis, instances) + + stubs = {inst["routing_endpoint"]: _FakeStub() for inst in instances} + published = {inst["endpoint"] for inst in instances} + + def _get_lc_stub(endpoint): + self.assertNotIn( + endpoint, published, "must route by routing_endpoint, not endpoint" + ) + return stubs[endpoint] + + controller._get_lc_stub = _get_lc_stub + + controller._trigger_cleanup() + + for routing_endpoint, stub in stubs.items(): + self.assertEqual( + len(stub.calls), + 1, + f"expected exactly one Cleanup call to {routing_endpoint}", + ) + self.assertEqual(set(stub.calls[0]["request_ids"]), completed) + + class StaleContainerNameTests(unittest.TestCase): """Bug 14: the cleanup used to build "canyonos--", which never matched the name the Local runtime actually creates, so no stale agent diff --git a/packages/core/tests/test_local_controller_cleanup.py b/packages/core/tests/test_local_controller_cleanup.py index 6be88d6c..dddc20a8 100644 --- a/packages/core/tests/test_local_controller_cleanup.py +++ b/packages/core/tests/test_local_controller_cleanup.py @@ -11,7 +11,10 @@ 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "grpc_stubs")) ) -from canyonos_core.controller.local_controller_frontend import LocalControllerServicer +from canyonos_core.controller.local_controller_frontend import ( + FUTURE_CLEANUP_GRACE_SECONDS, + LocalControllerServicer, +) import local_controler_pb2 @@ -102,6 +105,7 @@ class _FakeRedisStore: def __init__(self, strings=None, sets=None): self.strings = strings or {} self.sets = sets or {} + self.expirations = {} # key -> seconds, as recorded by expire() def setnx(self, key, value): if key in self.strings: @@ -123,6 +127,12 @@ def delete(self, *keys): self.strings.pop(key, None) self.sets.pop(key, None) + def expire(self, key, seconds, nx=False): + # Real Redis: schedules removal after `seconds`, doesn't touch the + # value now. This fake just records the call so tests can assert on + # it without needing to fake time passing. + self.expirations[key] = seconds + def _bare_servicer(redis): servicer = LocalControllerServicer.__new__(LocalControllerServicer) @@ -132,7 +142,12 @@ def _bare_servicer(redis): class CleanupRequestTests(unittest.TestCase): - def test_cleanup_deletes_consolidated_future_hashes_and_bookkeeping(self): + def test_cleanup_expires_consolidated_future_hashes_and_bookkeeping(self): + # CAN-391 follow-up: GlobalController's poll loop reads future:{id} to + # build an OTel span (see telemetry_logging.pull_runtime_information). + # Deleting it immediately here races that read and silently drops the + # span. Expiring with a grace period keeps memory bounded without + # deleting out from under the poll loop. redis = _FakeRedisStore( sets={"request:req1:futures": {"fut1", "fut2"}}, strings={ @@ -146,14 +161,21 @@ def test_cleanup_deletes_consolidated_future_hashes_and_bookkeeping(self): servicer._cleanup_request("req1") - # Future-resolution bookkeeping: gone. - self.assertNotIn("request:req1:futures", redis.sets) - self.assertNotIn("future:fut1", redis.strings) - self.assertNotIn("future:fut2", redis.strings) - self.assertNotIn("future:fut1:children", redis.strings) - self.assertNotIn("future:fut1:consumers", redis.strings) + # Not deleted outright -- still readable until the TTL elapses. + for key in ( + "request:req1:futures", + "future:fut1", + "future:fut2", + "future:fut1:children", + "future:fut1:consumers", + ): + self.assertEqual(redis.expirations.get(key), FUTURE_CLEANUP_GRACE_SECONDS) + self.assertIn("request:req1:futures", redis.sets) + self.assertIn("future:fut1", redis.strings) - def test_cleanup_still_deletes_affinity_bindings(self): + def test_cleanup_still_deletes_affinity_bindings_outright(self): + # Affinity bindings aren't read by the telemetry poll loop, so there's + # no race to protect against here -- immediate deletion is still fine. redis = _FakeRedisStore( sets={"request:req1:futures": {"fut1"}}, strings={"future:fut1": "x", "affinity:req1": "some-host"}, @@ -163,7 +185,11 @@ def test_cleanup_still_deletes_affinity_bindings(self): servicer._cleanup_request("req1") self.assertNotIn("affinity:req1", redis.strings) - self.assertNotIn("future:fut1", redis.strings) + # The future itself is expired (grace period), not deleted outright. + self.assertIn("future:fut1", redis.strings) + self.assertEqual( + redis.expirations.get("future:fut1"), FUTURE_CLEANUP_GRACE_SECONDS + ) def test_cleanup_releases_its_lock_even_with_no_futures(self): redis = _FakeRedisStore(sets={"request:req1:futures": set()})