From 4db1b53dc8c731ab0d747b9e1f5f6e376447773b Mon Sep 17 00:00:00 2001 From: Dongkeun Lee Date: Fri, 28 Aug 2026 06:49:28 -0400 Subject: [PATCH] perf(azure): eliminate fleet lists from slot reads --- bin/fm-azure-worker-provider.py | 194 ++++++++++++++++++++++++------ docs/azure-workers.md | 2 +- tests/fm-worker-lifecycle.test.sh | 65 +++++----- 3 files changed, 195 insertions(+), 66 deletions(-) diff --git a/bin/fm-azure-worker-provider.py b/bin/fm-azure-worker-provider.py index 64912628ad6..3ba8e750b9c 100755 --- a/bin/fm-azure-worker-provider.py +++ b/bin/fm-azure-worker-provider.py @@ -484,11 +484,52 @@ def resource_record(kind, value, power_state=None, tags_override=None): def azure_resource_not_found(stderr): return bool(re.search( - r"(?:\(ResourceNotFound\)|^Code:\s*ResourceNotFound\s*$)", + r"(?:\((?:ResourceNotFound|ParentResourceNotFound|ContainerNotFound)\)" + r"|^(?:Code|ErrorCode):\s*" + r"(?:ResourceNotFound|ParentResourceNotFound|ContainerNotFound)\s*$)", str(stderr or ""), re.MULTILINE, )) +def show_exact_optional(controller, args): + """Read one exact Azure object; only a recognized absence becomes None.""" + value, rc, stderr = az(controller, args, check=False) + if rc != 0: + if azure_resource_not_found(stderr): + return None + raise ProviderError("exact Azure inventory read failed or was malformed: {}".format(stderr)) + if not isinstance(value, dict): + raise ProviderError("exact Azure inventory read failed or was malformed: {}".format(stderr)) + return value + + +def show_exact_optional_many(controller, operations): + """Read independent exact objects concurrently, preserving error order.""" + if not operations: + return {} + keys = [key for key, _ in operations] + if len(keys) != len(set(keys)): + raise ProviderError("exact Azure inventory operation keys are not unique") + with concurrent.futures.ThreadPoolExecutor(max_workers=min(4, len(operations))) as executor: + futures = { + key: executor.submit(show_exact_optional, controller, args) + for key, args in operations + } + results = {} + failures = [] + for key in keys: + try: + results[key] = futures[key].result() + except Exception as exc: + failures.append((key, exc)) + if failures: + key, exc = failures[0] + if isinstance(exc, ProviderError): + raise ProviderError("{}: {}".format(key, exc)) + raise ProviderError("{}: exact Azure inventory read failed: {}".format(key, exc)) + return results + + def show_full(controller, resource_id, api_version=None, inventory_missing_ok=False): args = ["resource", "show", "--ids", resource_id] if api_version: @@ -1098,37 +1139,116 @@ def inventory(controller, include_metrics=True, target_slot=None): raise ProviderError("Azure subscription scope is not the exact enabled controller binding") prefix = re.escape(controller["prefix"]) - vms = list_json( - controller, - ["vm", "list", "--resource-group", controller["resource_group"], "--show-details"], - transient_not_found_attempts=4, - ) - nics = list_json(controller, ["network", "nic", "list", "--resource-group", controller["resource_group"]]) - disks = list_json(controller, ["disk", "list", "--resource-group", controller["resource_group"]]) - identities = list_json(controller, ["identity", "list", "--resource-group", controller["resource_group"]]) - extensions = list_json(controller, [ - "resource", "list", "--resource-group", controller["resource_group"], - "--resource-type", "Microsoft.Compute/virtualMachines/extensions", - ]) - run_commands = list_json(controller, [ - "resource", "list", "--resource-group", controller["resource_group"], - "--resource-type", "Microsoft.Compute/virtualMachines/runCommands", - ]) - schedules = list_json(controller, [ - "resource", "list", "--resource-group", controller["resource_group"], - "--resource-type", "Microsoft.DevTestLab/schedules", - ]) - # Scope casing varies across ARM responses, so filter client-side rather - # than with a case-sensitive JMESPath query. - scope_marker = "/resourcegroups/{}/".format(controller["resource_group"]).lower() - roles = [ - role for role in list_json(controller, ["role", "assignment", "list", "--all"]) - if scope_marker in str(role.get("scope") or "").lower() - ] - containers = list_json(controller, [ - "storage", "container", "list", "--auth-mode", "login", "--include-metadata", - "--account-name", os.environ.get("FM_AZURE_STORAGE_NAME", ""), - ]) + if target_slot is None: + vms = list_json( + controller, + ["vm", "list", "--resource-group", controller["resource_group"], "--show-details"], + transient_not_found_attempts=4, + ) + nics = list_json(controller, ["network", "nic", "list", "--resource-group", controller["resource_group"]]) + disks = list_json(controller, ["disk", "list", "--resource-group", controller["resource_group"]]) + identities = list_json(controller, ["identity", "list", "--resource-group", controller["resource_group"]]) + extensions = list_json(controller, [ + "resource", "list", "--resource-group", controller["resource_group"], + "--resource-type", "Microsoft.Compute/virtualMachines/extensions", + ]) + run_commands = list_json(controller, [ + "resource", "list", "--resource-group", controller["resource_group"], + "--resource-type", "Microsoft.Compute/virtualMachines/runCommands", + ]) + schedules = list_json(controller, [ + "resource", "list", "--resource-group", controller["resource_group"], + "--resource-type", "Microsoft.DevTestLab/schedules", + ]) + # Scope casing varies across ARM responses, so filter client-side rather + # than with a case-sensitive JMESPath query. + scope_marker = "/resourcegroups/{}/".format(controller["resource_group"]).lower() + roles = [ + role for role in list_json(controller, ["role", "assignment", "list", "--all"]) + if scope_marker in str(role.get("scope") or "").lower() + ] + containers = list_json(controller, [ + "storage", "container", "list", "--auth-mode", "login", "--include-metadata", + "--account-name", os.environ.get("FM_AZURE_STORAGE_NAME", ""), + ]) + else: + names = expected_names(controller, target_slot) + resource_group = controller["resource_group"] + storage = os.environ.get("FM_AZURE_STORAGE_NAME", "") + + vm_id = exact_id(controller, "Microsoft.Compute", "virtualMachines", names["vm"]) + exact = show_exact_optional_many(controller, [ + ("vm", [ + "vm", "show", "--resource-group", resource_group, + "--name", names["vm"], "--show-details", + ]), + ("nic", [ + "network", "nic", "show", "--resource-group", resource_group, + "--name", names["nic"], + ]), + ("os-disk", [ + "disk", "show", "--resource-group", resource_group, + "--name", names["os-disk"], + ]), + ("task-disk", [ + "disk", "show", "--resource-group", resource_group, + "--name", names["task-disk"], + ]), + ("account-disk", [ + "disk", "show", "--resource-group", resource_group, + "--name", names["account-disk"], + ]), + ("identity", [ + "identity", "show", "--resource-group", resource_group, + "--name", names["identity"], + ]), + ("monitor-extension", [ + "resource", "show", "--ids", + vm_id + "/extensions/" + names["monitor-extension"], + ]), + ("bootstrap-command", [ + "resource", "show", "--ids", + vm_id + "/runCommands/" + names["bootstrap-command"], + ]), + ("task-command", [ + "resource", "show", "--ids", + vm_id + "/runCommands/" + names["task-command"], + ]), + ("ttl-schedule", [ + "resource", "show", "--ids", exact_id( + controller, "Microsoft.DevTestLab", "schedules", names["ttl-schedule"], + ), "--api-version", "2018-09-15", + ]), + ("state-container", [ + "storage", "container", "show", "--auth-mode", "login", + "--account-name", storage, "--name", names["state-container"], + ]), + ]) + + def present(kind): + return [exact[kind]] if exact[kind] is not None else [] + + vms = present("vm") + nics = present("nic") + disks = [ + exact[kind] for kind in ("os-disk", "task-disk", "account-disk") + if exact[kind] is not None + ] + identities = present("identity") + extensions = present("monitor-extension") + run_commands = [ + exact[kind] for kind in ("bootstrap-command", "task-command") + if exact[kind] is not None + ] + schedules = present("ttl-schedule") + containers = present("state-container") + container_scope = ( + exact_id(controller, "Microsoft.Storage", "storageAccounts", storage) + + "/blobServices/default/containers/" + names["state-container"] + ) + roles = list_json(controller, [ + "role", "assignment", "list", "--scope", container_scope, + ]) if containers else [] workers = {} conflicts = [] @@ -1234,7 +1354,9 @@ def add(kind, value, slot, power=None, tags_override=None): ) # The generic resource listing omits properties and etag; only the full # object carries an immutable child identity. - value = show_full(controller, extension["id"], inventory_missing_ok=True) + value = dict(extension) if target_slot is not None else show_full( + controller, extension["id"], inventory_missing_ok=True, + ) if value is None: continue value["attached_to"] = vm_id @@ -1250,7 +1372,9 @@ def add(kind, value, slot, power=None, tags_override=None): if kind is None: conflicts.append({"kind": "run-command", "slot": slot, "reason": "undeclared worker Run Command child"}) continue - value = show_full(controller, command["id"], inventory_missing_ok=True) + value = dict(command) if target_slot is not None else show_full( + controller, command["id"], inventory_missing_ok=True, + ) if value is None: continue value["attached_to"] = exact_id( @@ -1263,7 +1387,7 @@ def add(kind, value, slot, power=None, tags_override=None): for schedule in schedules: slot = slot_from_name(schedule.get("name"), r"^shutdown-computevm-vm-{}-wkr-".format(prefix)) if slot is not None and (target_slot is None or slot == target_slot): - value = show_full( + value = dict(schedule) if target_slot is not None else show_full( controller, schedule["id"], api_version="2018-09-15", inventory_missing_ok=True, ) diff --git a/docs/azure-workers.md b/docs/azure-workers.md index 9e38c12a241..3aa21419d19 100644 --- a/docs/azure-workers.md +++ b/docs/azure-workers.md @@ -404,7 +404,7 @@ The guest verifies the runtime's exact file inventory, runs the role command wit The wrapper verifies the semantic bytes and head binding before writing the controller-facing result; a process exit, missing outcome, malformed outcome, or changed read-only head is a failed result, never `CLEAR` by inference. Repair results return one digest-bound single-ref bundle whose head must descend from the requested head, while review and test return no code bundle and must keep the exact requested head. The wrapper records a retryable local candidate before cleanup, releases through `service-complete` only after the lifecycle owns the exact execution result, and replays the candidate after a lost response instead of executing the step again. -Admission, execute recovery, and cleanup use `service-reconcile`, which advances only the caller's exact task generation or replays that task's own pending slot claim rather than converging unrelated fleet work. Once a service task owns a slot, its execution and cleanup inventory expands only that exact slot's Azure children; queued admission retains the whole-fleet quota, spend, and conflict census. +Admission, execute recovery, and cleanup use `service-reconcile`, which advances only the caller's exact task generation or replays that task's own pending slot claim rather than converging unrelated fleet work. Once a service task owns a slot, its execution and cleanup inventory reads that slot's deterministic Azure object paths in a bounded parallel batch; it neither lists the resource group nor expands peer children. Queued admission retains the whole-fleet quota, spend, and conflict census. The guest supervisor marks a no-mistakes Azure execution as the already-isolated test boundary, so the repository test command runs the focused service suite directly instead of recursively provisioning the general validation fleet or a Herdr lab. The root-owned supervisor stages the job, then runs the no-mistakes process as the dedicated non-root `fmworker` user with no supplementary groups; the sealed runtime remains root-owned and read-only while the exact repository and projected account are writable only by that service identity. `bin/fm-azure-service-test-scope.py` owns that focused inventory and the narrow source set eligible for focused pull-request CI; an empty, mixed, or unknown diff and every push to `main` retain the complete behavior suite. diff --git a/tests/fm-worker-lifecycle.test.sh b/tests/fm-worker-lifecycle.test.sh index 01247bb5423..9546f9119c8 100755 --- a/tests/fm-worker-lifecycle.test.sh +++ b/tests/fm-worker-lifecycle.test.sh @@ -9223,40 +9223,45 @@ controller = { tags = { "workload": "firstmate", "deployment-generation": "dep", "cleanup-owner": "owner", } -vms = [ - {"name": "vm-fixture-wkr-01", "id": "/vm/1", "vmId": "vm-id-1", - "powerState": "VM running", "tags": tags}, - {"name": "vm-fixture-wkr-02", "id": "/vm/2", "vmId": "vm-id-2", - "powerState": "VM running", "tags": tags}, -] -extensions = [ - {"name": "vm-fixture-wkr-01/AzureMonitorLinuxAgent", "id": "/vm/1/ext"}, - {"name": "vm-fixture-wkr-02/AzureMonitorLinuxAgent", "id": "/vm/2/ext"}, -] -expanded = [] - -provider.az = lambda _controller, args, check=False, timeout=provider.AZ_TIMEOUT_SECONDS: ( - ({"id": controller["subscription"], "state": "Enabled"}, 0, "") - if args[:2] == ["account", "show"] else (None, 1, "unexpected") +calls = [] +vm_id = provider.exact_id( + controller, "Microsoft.Compute", "virtualMachines", "vm-fixture-wkr-02", +) +extension_id = vm_id + "/extensions/AzureMonitorLinuxAgent" + +def exact_read(_controller, args, check=False, timeout=provider.AZ_TIMEOUT_SECONDS): + calls.append(tuple(args)) + if args[:2] == ["account", "show"]: + return {"id": controller["subscription"], "state": "Enabled"}, 0, "" + if args[:2] == ["vm", "show"]: + assert args[args.index("--name") + 1] == "vm-fixture-wkr-02", args + return { + "name": "vm-fixture-wkr-02", "id": vm_id, "vmId": "vm-id-2", + "powerState": "VM running", "tags": tags, "identity": {}, + }, 0, "" + if args[:3] == ["resource", "show", "--ids"] and args[3] == extension_id: + return { + "name": "vm-fixture-wkr-02/AzureMonitorLinuxAgent", + "id": extension_id, "tags": tags, + "properties": {"provisioningState": "Succeeded"}, + }, 0, "" + if args[:3] == ["storage", "container", "show"]: + return None, 1, "ErrorCode:ContainerNotFound" + return None, 1, "(ResourceNotFound) exact fixture absence" + +provider.az = exact_read +provider.list_json = lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("targeted inventory issued a list operation") ) -def listing(_controller, args, transient_not_found_attempts=1): - if args[:2] == ["vm", "list"]: - return vms - if args[:2] == ["resource", "list"] and args[-1] == "Microsoft.Compute/virtualMachines/extensions": - return extensions - return [] -provider.list_json = listing -def show(_controller, resource_id, api_version=None, inventory_missing_ok=False): - expanded.append(resource_id) - return { - "id": resource_id, "tags": tags, - "properties": {"provisioningState": "Succeeded"}, - } -provider.show_full = show snapshot = provider.inventory_slot(controller, 2) assert [worker["slot"] for worker in snapshot["workers"]] == [2], snapshot -assert expanded == ["/vm/2/ext"], expanded +assert snapshot["workers"][0]["resources"]["monitor-extension"]["id"] == extension_id +assert all("wkr-01" not in " ".join(call) for call in calls), calls +assert not any("list" in call[:3] for call in calls), calls +assert any(call[:2] == ("vm", "show") for call in calls), calls +assert any(call[:2] == ("network", "nic") for call in calls), calls +assert sum(call[:2] == ("disk", "show") for call in calls) == 3, calls assert snapshot["capacity_reservations"] == [] assert snapshot["metrics"]["actual_usd"] is None try: