CAEVES Copilot Connector Fleet Scaling

How It Works

 The CAEVES Copilot Connector Fleet Scaling script automates scaling Azure Container Instances (ACI) up or down based on the required fleet size.  

• Fleet Discovery: The script identifies connector fleets across the Azure subscription by filtering container groups based on the connector image (caeves-connector). Each fleet is identified by a common name prefix and numeric suffix (e.g., prd-cvs-we-connector-1, prd-cvs-we-connector-02, and prd-cvs-we-connector-03). No resource group is hardcoded.
• Instance Creation: New instances are cloned from the lowest-numbered healthy instance in the Running state. The script copies its network settings, managed identity, registry access, Log Analytics configuration, and environment variables.
• Microsoft Graph Rate-Limit Management: The script distributes the approximate Graph request budget of 25 requests per second across the fleet. Each instance receives a rate limit calculated as 25 / N, rounded down, with a minimum of one request per second.
• Automatic Rebuilding: When an instance's rate limit needs to change, the script deletes and recreates it because ACI environment variables are immutable.
• Controlled Scaling Sequence: The script deletes surplus instances first, rebuilds existing instances when their rate limits need to change, and then creates new instances. This helps prevent the fleet from exceeding the Graph request budget and triggering 429s.
• Reliable Queue Processing: Work items are stored in a durable queue. If an instance is removed while processing an item, the item is redelivered after the queue visibility timeout, helping prevent work from being lost.

Prerequisites 

 Before starting, ensure the following requirements are met:  

 Step-by-Step Procedure 

Step 1: Get the Script

Copy the complete scale_fleet.py script from the Script: scale_fleet.py section at the end of this KB article.

Save the file as: scripts/scale_fleet.py

Step 2: Set the Subscription and Default Fleet

The published script contains a placeholder subscription ID and will not run until a valid subscription ID is configured. 

Configure the following values: 

 The default fleet prefix is prd-cvs-we-connector. 

 PowerShell 

$env:CAEVES_SUBSCRIPTION = "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX"
$env:CAEVES_FLEET        = "prd-cvs-we-connector"

 Bash 

export CAEVES_SUBSCRIPTION="XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX"
export CAEVES_FLEET="prd-cvs-we-connector"

Replace <subscription-id> with the actual Azure subscription ID.

Alternatively, update the SUBSCRIPTION and DEFAULT_FLEET values near the top of the script.

Note:  The script passes --subscription to every Azure CLI command, so it does not change the active Azure CLI subscription.

Step 3: Check the Current State

 Run the following command to view all discovered connector fleets and their instance status. 

python scripts/scale_fleet.py status

Verify the following:

• At least one instance is in the Running state before scaling up.
• Restart counts are stable.
• The QUEUE column displays yes. Without CAEVES_WORKER_QUEUE_URL the fleet does table-scan ingestion and extra replicas will not share the work. 

 Important: Use the script's status output to assess runtime health.  az container list reports provisioning state and may show Succeeded even when an instance is crash-looping. 

Step 4: Preview the Plan (Dry Run)

 Before applying changes, preview the scaling plan. 

python scripts/scale_fleet.py scale 8 --dry-run

The plan displays:

• The reference instance used for cloning.
• The connector image.
• The new per-instance Graph rate limit.
• Instances scheduled for deletion.
• Existing instances that need rebuilding.
• New instances to be created.

Step 5: Scale the Fleet

Scale Up

 To scale the fleet to eight instances: 

python scripts/scale_fleet.py scale 8     # scale to 8 instances

 To add two instances to the current fleet: 

python scripts/scale_fleet.py scale +2    # add 2 instances

 New instances are named with the next free suffix (-02, -03 ...). Survivors whose rate changes are rebuilt first, then new instances are created. 

Scale Down

 To scale the fleet to four instances: 
python scripts/scale_fleet.py scale 4     # scale to 4 instances

 To remove one instance from the current fleet: 
python scripts/scale_fleet.py scale -1    # remove 1 instance

 The highest-numbered instances are removed first. The remaining instances are rebuilt if their share of the Graph budget goes up. 

 Interactive 

python scripts/scale_fleet.py

Lists every fleet, lets you pick one, several (1,2) or all, and asks for a target for each.

Every run prints its plan and asks Apply? [y/N] before changing anything. After applying, the script waits (polling every 45 seconds, up to 25 minutes) until every desired instance is Running, then prints the fleet status again.

Step 6: Verify the Fleet

 After scaling completes, run:  python scripts/scale_fleet.py status  

Verify that:

• The desired number of instances is present.
• All expected instances are in the Running state.
• Restart counts are stable.
• The QUEUE column displays yes.
• No rate limit is marked with !.
• The correct connector image is running.
•  If any instance is not running, check its logs: 
  az container logs --subscription "<subscription-id>" -g "<resource-group>" -n "<instance-name>"
   Replace the placeholders with the actual subscription ID, resource group, and instance name. 
  
  

 Command-Line Options 

Environment Variables

 Rate limit per fleet size

Instances (N)	Rate per instance	Fleet total
1	25/s	25/s
2	12/s	24/s
3	8/s	24/s
4	6/s	24/s
5	5/s	25/s
6	4/s	24/s
7 to 8	3/s	21 to 24/s
9 to 12	2/s	18 to 24/s
13 to 25	1/s	13 to 25/s
26 to 60	1/s (floor)	26 to 60/s, over budget

 Important considerations: 

• The per-instance rate is calculated as 25 // N, with a minimum of one request per second.
• ACI environment variables are immutable. Instances must be recreated when their configured rate limits change.
• The script automatically rebuilds affected instances unless --no-rebalance is used.
• Scaling between 13 and 25 instances does not change the per-instance rate limit, so a rate-only rebuild is not required.
• Above 25 instances, the minimum rate of  1/s per instance exceeds the approximate Graph budget and may result in 429s.

 Timing and downtime

• Each new or recreated instance pulls a ~1.8 GB image. Expect roughly 3 to 10 minutes per batch, with the image pull dominating.
• Instances are created and rebuilt in batches of 8.
• A rebuild deletes instances before recreating them. At a fleet size of 1 that means a few minutes of zero capacity; at larger sizes, reduced capacity. Resize at quiet times.

Limits

• Minimum 1, maximum 60 instances. The script cannot scale to zero.
• A Running instance is needed to grow or rebuild. If nothing is Running, the fleet can only be scaled down until one instance is fixed.
• Secure environment variables cannot be cloned (Azure never returns their values); the script stops with an error if the reference group has any.
• Only the image and rate change on clones. To change any other setting, fix it on the reference instance first.

 Troubleshooting 

Symptom	Cause and Recommended Action
Error: no subscription set	Step 2 was skipped. Set CAEVES_SUBSCRIPTION.
No fleet named ...	The --fleet or CAEVES_FLEET prefix doesn't match. The error lists the fleets that were found.
no Running instance to clone from	Every instance is down. Check logs, fix or redeploy one instance, then scale.
... pulls with a registry username/password	Set CAEVES_REGISTRY_PASSWORD and rerun.
... is on ACR but ... has no credential	Pass --registry-identity with a managed identity that has AcrPull.
Log Analytics workspace ... is not visible	The workspace is in another subscription or you lack access to its keys.
Instance quiet for 10+ minutes	Usually a long OCR job rather than a hang. Check its logs before acting.
Rate shows ! in status	A previous run used --no-rebalance or failed midway. Rescale to the current count to rebuild them.

 Script: scale_fleet.py 

 Subscription id masked. Set CAEVES_SUBSCRIPTION and CAEVES_FLEET (Step 2) before use. Save as scripts/scale_fleet.py. 

#!/usr/bin/env python3
"""
Scale CAEVES connector fleets on Azure Container Instances up or down.

    python scripts/scale_fleet.py                     # interactive: pick fleet(s), set targets
    python scripts/scale_fleet.py status              # every fleet and instance, with real state
    python scripts/scale_fleet.py scale 8             # plan, confirm, apply (default fleet)
    python scripts/scale_fleet.py scale +2 --fleet stg-cvs-we-connector-acr
    python scripts/scale_fleet.py scale 8 --dry-run   # print the plan and stop
    python scripts/scale_fleet.py scale 1 --image caevesprd.azurecr.io/caeves/caeves-connector:latest
                                                      # move to ACR, pulling by managed identity

A fleet is a set of container groups that share a name prefix and differ only
by a numeric suffix -- prd-cvs-we-connector-1, -02, -03 ... -- all running the
connector image and draining the same storage queue. A lone group without a
suffix is a fleet of one; replicas added to it are named <name>-02, -03 ....
Fleets are discovered across the whole subscription by image, so nothing about
resource groups is hardcoded.

New groups are cloned from a group in the fleet that is already healthy, read
live with `az container show`. Nothing about the shape is hardcoded here:
subnet, managed identity, registry credentials, Log Analytics, image and every
environment variable come from whatever the reference is running. That keeps
this script correct as the connector's configuration changes, and means it
cannot invent a half-configured replica -- the failure mode that once left three
of four groups crash-looping while `az container list` reported them all as
"Succeeded".

Two things make scaling more than a create/delete loop:

MS_GRAPH_UPLOAD_RATE_LIMIT is a *per-process* rate in requests/sec (main.py,
GraphRateLimiter) drawing on a fleet-wide Graph budget of ~25/s. The correct
per-replica value is therefore GRAPH_BUDGET // N, and it changes every time N
changes. ACI environment variables are immutable, so a survivor whose rate is
now wrong has to be deleted and recreated -- that is what "rebuild" means below.

Order is deletes, then rebuilds, then creates. Rebuilding before creating means
the fleet never briefly runs with both the old (higher) rate and the new replicas
on top of it, which would push the whole app registration past the Graph budget
and earn 429s. The cost is a short dip in capacity while survivors recycle;
rebuilds are batched so they do not all go down at once. Work is not lost: the
queue is durable and an in-flight item whose claim dies is redelivered after the
visibility timeout.

Instances are matched by their numeric suffix, not their exact name, so a
fleet whose first group is "-1" keeps it when scaling rather than replacing it
with "-01".
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Optional

# REQUIRED: set both for your environment, either here or with the
# CAEVES_SUBSCRIPTION and CAEVES_FLEET environment variables.
#   SUBSCRIPTION   the Azure subscription id that holds the connector fleet
#   DEFAULT_FLEET  the fleet name prefix used when --fleet is not given
SUBSCRIPTION = os.getenv("CAEVES_SUBSCRIPTION", "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX")
DEFAULT_FLEET = os.getenv("CAEVES_FLEET", "prd-cvs-we-connector")

# Container groups whose image path contains this are connector replicas. The
# admin UI (Caddy) and anything else in the subscription are ignored.
CONNECTOR_IMAGE = "caeves-connector"
CONNECTOR_IMAGE_EXCLUDE = "caeves-connector-ui"

# Fleet-wide Microsoft Graph write budget, requests/sec. Mirrors the
# MS_GRAPH_UPLOAD_RATE_LIMIT default in main.py.
GRAPH_BUDGET = 25

# ACI pulls a ~1.8 GB image per group; ARM deploys of more than this many at once
# take long enough that a single failure costs a lot of waiting.
BATCH_SIZE = 8
MAX_REPLICAS = 60
PARALLEL_AZ = 8

_SUFFIX = re.compile(r"^(?P<prefix>.+)-(?P<index>\d+)$")

# On Windows the CLI is az.cmd, which CreateProcess won't find from a bare "az".
AZ = shutil.which("az") or "az"


def _az(*args: str, check: bool = True) -> Optional[str]:
    """
    Run an az command pinned to the connector subscription.

    Always passes --subscription rather than relying on the CLI's active
    context, which belongs to the user and must not be switched underneath them.
    """
    proc = subprocess.run(
        [AZ, *args, "--subscription", SUBSCRIPTION], capture_output=True, text=True
    )
    if proc.returncode != 0:
        if check:
            raise RuntimeError(
                f"az {' '.join(args)} failed ({proc.returncode}): "
                f"{(proc.stderr or '').strip()[:800]}"
            )
        return None
    return (proc.stdout or "").strip()


def per_replica_rate(count: int) -> int:
    """Each replica's share of the Graph budget. Never below 1: 0 disables the limiter."""
    return max(1, GRAPH_BUDGET // max(1, count))


def _split_name(name: str) -> tuple[str, int]:
    """'prd-cvs-we-connector-07' -> ('prd-cvs-we-connector', 7); no suffix -> (name, 1)."""
    m = _SUFFIX.match(name)
    return (m["prefix"], int(m["index"])) if m else (name, 1)


class Fleet:
    def __init__(self, prefix: str, resource_group: str):
        self.prefix = prefix
        self.resource_group = resource_group
        self.groups: list[dict] = []

    @property
    def by_index(self) -> dict[int, dict]:
        return {g["index"]: g for g in self.groups}

    @property
    def running(self) -> int:
        return sum(1 for g in self.groups if g["state"] == "Running")

    def new_name(self, index: int) -> str:
        return f"{self.prefix}-{index:02d}"


def _show(resource_group: str, name: str) -> Optional[dict]:
    raw = _az("container", "show", "--resource-group", resource_group, "--name", name,
              "-o", "json", check=False)
    if not raw:
        return None
    d = json.loads(raw)
    c = d["containers"][0]
    iv = c.get("instanceView") or {}
    env = {e["name"]: e.get("value") for e in (c.get("environmentVariables") or [])}
    _, index = _split_name(name)
    return {
        "name": name,
        "index": index,
        "provisioning": d.get("provisioningState"),
        "state": (iv.get("currentState") or {}).get("state"),
        "restarts": iv.get("restartCount") or 0,
        "image": c["image"],
        "rate": env.get("MS_GRAPH_UPLOAD_RATE_LIMIT"),
        "queue": bool(env.get("CAEVES_WORKER_QUEUE_URL")),
        "nvars": len(env),
        "raw": d,
    }


def discover() -> list[Fleet]:
    """
    Every connector fleet in the subscription, with the container state that
    `az container list` does not return (its "Succeeded" is provisioning state only).
    """
    listed = json.loads(_az(
        "container", "list", "--query",
        "[].{name:name, rg:resourceGroup, image:containers[0].image}", "-o", "json") or "[]")
    candidates = [
        g for g in listed
        if CONNECTOR_IMAGE in (g["image"] or "") and CONNECTOR_IMAGE_EXCLUDE not in g["image"]
    ]
    with ThreadPoolExecutor(PARALLEL_AZ) as pool:
        shown = list(pool.map(lambda g: (g, _show(g["rg"], g["name"])), candidates))

    fleets: dict[tuple[str, str], Fleet] = {}
    for g, detail in shown:
        if detail is None:
            continue
        prefix, _ = _split_name(g["name"])
        key = (prefix.lower(), g["rg"].lower())
        fleets.setdefault(key, Fleet(prefix, g["rg"])).groups.append(detail)
    for f in fleets.values():
        f.groups.sort(key=lambda g: g["index"])
    return sorted(fleets.values(), key=lambda f: f.prefix)


def pick_reference(fleet: Fleet) -> Optional[dict]:
    """
    The group new replicas are cloned from: the lowest-numbered one that is
    Running. A crash-looping group is a bad template -- it may be crash-looping
    precisely because its configuration is incomplete. None if nothing is
    Running, in which case the fleet can only shrink.
    """
    healthy = [g for g in fleet.groups if g["state"] == "Running"]
    return healthy[0] if healthy else None


def registry_server(image: str) -> Optional[str]:
    """'caevesprd.azurecr.io/caeves/x:tag' -> 'caevesprd.azurecr.io'; Docker Hub shorthand -> None."""
    first = image.split("/", 1)[0]
    return first.lower() if "/" in image and ("." in first or ":" in first) else None


def build_template(reference: dict, registry: Optional[tuple[str, str]] = None) -> tuple[dict, bool]:
    """
    An ARM template that clones the reference group across a list of names.
    `registry` is (server, identity resource id) to pull a new image with,
    replacing any credential the reference has for that server.
    Returns (template, needs_registry_password).
    """
    src = reference["raw"]
    c = src["containers"][0]
    env = {e["name"]: e.get("value") for e in c["environmentVariables"]}
    secure = [e["name"] for e in c["environmentVariables"]
              if e.get("value") is None and e.get("secureValue") is None]
    if secure:
        # `show` never returns secure values, so a clone would silently drop them.
        raise RuntimeError(
            f"{reference['name']} has secure environment variables ({', '.join(secure)}) whose\n"
            "values Azure will not return; this script cannot clone them."
        )
    # Both are per-replica or per-scale, and are supplied as parameters below.
    env.pop("CONNECTOR_ID", None)
    env.pop("MS_GRAPH_UPLOAD_RATE_LIMIT", None)

    # Registry credentials. Managed-identity pulls clone verbatim; a password is
    # write-only and must be supplied again.
    registries, needs_password = [], False
    for cred in src.get("imageRegistryCredentials") or []:
        if registry and cred["server"].lower() == registry[0]:
            continue
        if cred.get("identity"):
            registries.append({"server": cred["server"], "identity": cred["identity"]})
        else:
            registries.append({"server": cred["server"], "username": cred["username"],
                               "password": "[parameters('registryPassword')]"})
            needs_password = True
    if registry:
        registries.append({"server": registry[0], "identity": registry[1]})

    # ACI rejects a new group whose IP address exposes no ports, even though an
    # existing group can be in that state (prd-cvs-we-connector-1 is). Nothing
    # connects in over the private IP, so fall back to the port 80 the fleet has
    # always been created with.
    ports = c.get("ports") or [{"port": 80, "protocol": "TCP"}]

    properties = {
        "osType": src["osType"],
        "sku": src["sku"],
        "restartPolicy": src["restartPolicy"],
        "containers": [{
            "name": "[parameters('names')[copyIndex()]]",
            "properties": {
                "image": "[parameters('image')]",
                "ports": ports,
                "resources": {"requests": {
                    "cpu": c["resources"]["requests"]["cpu"],
                    "memoryInGB": c["resources"]["requests"]["memoryInGb"],
                }},
                "environmentVariables":
                    [{"name": k, "value": v} for k, v in env.items()] + [
                        {"name": "CONNECTOR_ID", "value": "[parameters('names')[copyIndex()]]"},
                        {"name": "MS_GRAPH_UPLOAD_RATE_LIMIT", "value": "[parameters('rateLimit')]"},
                    ],
            },
        }],
    }
    if registries:
        properties["imageRegistryCredentials"] = registries
    if src.get("subnetIds"):
        properties["subnetIds"] = [{"id": s["id"]} for s in src["subnetIds"]]
    if src.get("ipAddress"):
        ip = src["ipAddress"]
        properties["ipAddress"] = {"type": ip["type"], "ports": ip.get("ports") or ports}
    la = (src.get("diagnostics") or {}).get("logAnalytics")
    if la:
        properties["diagnostics"] = {"logAnalytics": {
            "workspaceId": la["workspaceId"],
            "workspaceKey": "[parameters('workspaceKey')]",
        }}

    resource = {
        "type": "Microsoft.ContainerInstance/containerGroups",
        "apiVersion": "2023-05-01",
        "name": "[parameters('names')[copyIndex()]]",
        "location": src["location"],
        "copy": {"name": "fleet", "count": "[length(parameters('names'))]", "mode": "Parallel"},
        "properties": properties,
    }
    ident = src.get("identity") or {}
    assigned = list(ident.get("userAssignedIdentities") or {})
    if registry and registry[1].lower() not in (k.lower() for k in assigned):
        # The pull identity must be assigned to the group itself.
        assigned.append(registry[1])
    if ident or assigned:
        kinds = set((ident.get("type") or "").replace(" ", "").split(",")) - {"", "None"}
        if assigned:
            kinds.add("UserAssigned")
        resource["identity"] = {"type": ", ".join(sorted(kinds))}
        if assigned:
            resource["identity"]["userAssignedIdentities"] = {k: {} for k in assigned}
    if src.get("tags"):
        resource["tags"] = src["tags"]

    template = {
        "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
        "contentVersion": "1.0.0.0",
        "parameters": {
            "names": {"type": "array"},
            "image": {"type": "string"},
            "rateLimit": {"type": "string"},
            "workspaceKey": {"type": "securestring", "defaultValue": ""},
            "registryPassword": {"type": "securestring", "defaultValue": ""},
        },
        "resources": [resource],
    }
    return template, needs_password


def workspace_key(reference: dict) -> str:
    """
    The Log Analytics key for the reference's workspace, refetched every run:
    `az container show` returns it as null, it is write-only.
    """
    la = (reference["raw"].get("diagnostics") or {}).get("logAnalytics")
    if not la:
        return ""
    found = json.loads(_az(
        "monitor", "log-analytics", "workspace", "list",
        "--query", f"[?customerId=='{la['workspaceId']}'].",
        "-o", "json") or "[]")
    if not found:
        raise RuntimeError(
            f"Log Analytics workspace {la['workspaceId']} used by {reference['name']} is not "
            "visible in this subscription; cannot fetch its key.")
    return _az("monitor", "log-analytics", "workspace", "get-shared-keys",
               "--resource-group", found[0]["rg"], "--workspace-name", found[0]["name"],
               "--query", "primarySharedKey", "-o", "tsv")


def create(fleet: Fleet, names: list[str], template: dict, image: str, rate: int,
           key: str, registry_password: str) -> None:
    # Template and parameters go via 0600 temp files so the workspace key and any
    # registry password never appear on a command line (visible in `ps`).
    with tempfile.TemporaryDirectory() as tmp:
        tpath = os.path.join(tmp, "template.json")
        with open(tpath, "w") as fh:
            json.dump(template, fh)
        for i in range(0, len(names), BATCH_SIZE):
            batch = names[i:i + BATCH_SIZE]
            ppath = os.path.join(tmp, f"params-{i}.json")
            with open(ppath, "w") as fh:
                json.dump({"$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentParameters.json#",
                           "contentVersion": "1.0.0.0",
                           "parameters": {
                               "names": {"value": batch}, "image": {"value": image},
                               "rateLimit": {"value": str(rate)}, "workspaceKey": {"value": key},
                               "registryPassword": {"value": registry_password}}}, fh)
            print(f"  creating {', '.join(batch)} (~3-10 min, image pull dominates)", flush=True)
            _az("deployment", "group", "create",
                "--resource-group", fleet.resource_group,
                "--name", f"scale-{int(time.time())}-{i // BATCH_SIZE}",
                "--template-file", tpath, "--parameters", f"@{ppath}", "-o", "none")


def delete(fleet: Fleet, names: list[str]) -> None:
    procs = []
    for name in names:
        procs.append(subprocess.Popen(
            [AZ, "container", "delete", "--subscription", SUBSCRIPTION,
             "--resource-group", fleet.resource_group, "--name", name, "--yes", "-o", "none"],
            stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True))
    for name, proc in zip(names, procs):
        _, err = proc.communicate()
        print(f"  deleted {name}" if proc.returncode == 0 else
              f"  FAILED to delete {name}: {(err or '').strip()[:200]}", flush=True)


def wait_running(fleet: Fleet, names: list[str], timeout: int = 1500) -> dict[str, str]:
    """Poll until every named group reports Running, or time out."""
    def state(name: str) -> str:
        return _az("container", "show", "--resource-group", fleet.resource_group, "--name", name,
                   "--query", "containers[0].instanceView.currentState.state",
                   "-o", "tsv", check=False) or "?"

    deadline = time.time() + timeout
    states: dict[str, str] = {}
    while time.time() < deadline:
        with ThreadPoolExecutor(PARALLEL_AZ) as pool:
            states = dict(zip(names, pool.map(state, names)))
        up = sum(1 for s in states.values() if s == "Running")
        print(f"  [{time.strftime('%H:%M:%S')}] {fleet.prefix}: running {up}/{len(names)}", flush=True)
        if up == len(names):
            return states
        time.sleep(45)
    return states


# --- planning ----------------------------------------------------------------

class Plan:
    def __init__(self, fleet: Fleet, target: int, image_override: Optional[str], rebalance: bool,
                 registry_identity: Optional[str] = None):
        self.fleet = fleet
        self.target = target
        self.reference = pick_reference(fleet)
        self.image = image_override or (self.reference or fleet.groups[0])["image"]
        self.rate = per_replica_rate(target)
        self.registry = self._resolve_registry(registry_identity)

        existing = fleet.by_index
        desired = set(range(1, target + 1))
        survivors = sorted(set(existing) & desired)
        self.to_delete = [existing[i]["name"] for i in sorted(set(existing) - desired)]
        self.to_create = [fleet.new_name(i) for i in sorted(desired - set(existing))]
        self.to_rebuild = [] if not rebalance else [
            existing[i]["name"] for i in survivors
            if existing[i]["rate"] != str(self.rate)
            or (image_override and existing[i]["image"] != self.image)
        ]
        if self.reference is None:
            if self.to_create:
                raise RuntimeError(
                    f"{fleet.prefix}: no Running instance to clone from, so it can only be scaled\n"
                    "down. Fix or redeploy one instance first (see its logs).")
            # Rebuilding also needs a template; shrink only and say so.
            self.skipped_rebuild, self.to_rebuild = self.to_rebuild, []
        else:
            self.skipped_rebuild = []
        self.desired_names = [existing[i]["name"] if i in existing else fleet.new_name(i)
                              for i in sorted(desired)]

    def _resolve_registry(self, identity: Optional[str]) -> Optional[tuple[str, str]]:
        """
        (server, identity) to pull self.image with, when the reference has no
        credential for its registry -- i.e. when moving to a private registry.
        For ACR the group's own managed identity is used if it has exactly one.
        """
        server = registry_server(self.image)
        if self.reference is None or server is None:
            return None
        src = self.reference["raw"]
        has_cred = any(c["server"].lower() == server for c in src.get("imageRegistryCredentials") or [])
        if identity:
            return server, identity
        if has_cred or not server.endswith(".azurecr.io"):
            return None  # already configured, or a public registry such as ghcr.io
        assigned = list(((src.get("identity") or {}).get("userAssignedIdentities")) or {})
        if len(assigned) != 1:
            raise RuntimeError(
                f"{self.image} is on ACR but {self.reference['name']} has no credential for {server}.\n"
                "Pass --registry-identity <managed identity resource id> (it needs AcrPull).")
        return server, assigned[0]

    @property
    def empty(self) -> bool:
        return not (self.to_delete or self.to_create or self.to_rebuild)

    def print(self) -> None:
        f, ref = self.fleet, self.reference
        print(f"\nPlan for {f.prefix} ({f.resource_group}): {len(f.groups)} -> {self.target} instance(s)")
        if ref:
            print(f"  reference group   {ref['name']} (cloning its config, {ref['nvars']} env vars)")
        else:
            print("  reference group   none Running -- delete only")
        print(f"  image             {self.image}")
        if self.registry:
            print(f"  pull identity     {self.registry[1].rsplit('/', 1)[-1]} -> {self.registry[0]} (new)")
        print(f"  rate limit        {self.rate}/s per replica  "
              f"({self.rate} x {self.target} vs {GRAPH_BUDGET}/s Graph budget)")
        print(f"  delete            {len(self.to_delete):>2}  {', '.join(self.to_delete) or '-'}")
        print(f"  rebuild           {len(self.to_rebuild):>2}  {', '.join(self.to_rebuild) or '-'}")
        print(f"  create            {len(self.to_create):>2}  {', '.join(self.to_create) or '-'}")
        if self.to_rebuild:
            print("  Rebuild deletes and recreates a running replica so its immutable rate limit or\n"
                  "  image can change. In-flight items are redelivered after the queue visibility timeout.")
        if self.skipped_rebuild:
            print(f"  NOTE: {len(self.skipped_rebuild)} survivor(s) keep a stale rate limit; with no Running\n"
                  "  instance there is nothing to rebuild them from.")
        if ref and self.target > 1 and not ref["queue"]:
            print("  WARNING: the reference has no CAEVES_WORKER_QUEUE_URL, so this fleet runs\n"
                  "  table-scan ingestion and extra replicas will not fan the work out.")
        if self.empty:
            print("  Nothing to do.")

    def apply(self, registry_password: str) -> bool:
        f = self.fleet
        if self.reference is None:
            print(f"\n{f.prefix}: deleting surplus groups...")
            delete(f, self.to_delete)
            print(f"\n{f.prefix}: now {self.target} instance(s); none were Running, so not waiting.")
            return True
        template, needs_password = build_template(self.reference, self.registry)
        if needs_password and not registry_password:
            raise RuntimeError(
                f"{self.reference['name']} pulls with a registry username/password, which Azure\n"
                "will not return. Set CAEVES_REGISTRY_PASSWORD and rerun.")
        key = workspace_key(self.reference)

        if self.to_delete:
            print(f"\n{f.prefix}: deleting surplus groups...")
            delete(f, self.to_delete)
        if self.to_rebuild:
            print(f"\n{f.prefix}: rebuilding groups whose rate limit or image is changing...")
            for i in range(0, len(self.to_rebuild), BATCH_SIZE):
                batch = self.to_rebuild[i:i + BATCH_SIZE]
                delete(f, batch)
                create(f, batch, template, self.image, self.rate, key, registry_password)
        if self.to_create:
            print(f"\n{f.prefix}: creating new groups...")
            create(f, self.to_create, template, self.image, self.rate, key, registry_password)

        print(f"\n{f.prefix}: waiting for the fleet to come up...")
        states = wait_running(f, self.desired_names)
        bad = {n: s for n, s in states.items() if s != "Running"}
        if bad:
            print(f"\n{f.prefix}: not Running: {bad}")
            print("Check with: az container logs --subscription "
                  f"{SUBSCRIPTION} -g {f.resource_group} -n <name>")
            return False
        print(f"\n{f.prefix}: at {self.target} instance(s), all running.")
        return True


def parse_target(text: str, current: int) -> int:
    """'8' -> 8, '+2' -> current + 2, '-1' -> current - 1; blank keeps current."""
    text = text.strip()
    if not text:
        return current
    if text[0] in "+-":
        return current + int(text)
    return int(text)


# --- output -------------------------------------------------------------------

def print_fleets(fleets: list[Fleet], numbered: bool = False) -> None:
    if not fleets:
        print("No connector container groups found in this subscription.")
        return
    for n, f in enumerate(fleets, 1):
        want = per_replica_rate(len(f.groups))
        head = f"[{n}] " if numbered else ""
        print(f"\n{head}{f.prefix}  ({f.resource_group})  "
              f"{len(f.groups)} instance(s), {f.running} running, "
              f"rate should be {want}/s")
        if not f.running:
            print("    ! nothing Running: can be scaled down, but not up, until one instance is fixed")
        print(f"    {'NAME':<34} {'STATE':<11} {'RESTARTS':>8}  {'RATE':>6}  {'QUEUE':<5}  IMAGE")
        for g in f.groups:
            rate = str(g["rate"]) + ("" if g["rate"] == str(want) else " !")
            print(f"    {g['name']:<34} {str(g['state']):<11} {g['restarts']:>8}  "
                  f"{rate:>6}  {'yes' if g['queue'] else 'no':<5}  {g['image']}")
        stale = [g for g in f.groups if g["rate"] != str(want)]
        if stale:
            print(f"    ! rate mismatch on {len(stale)} group(s); rescaling to "
                  f"{len(f.groups)} will rebuild them.")


def find_fleet(fleets: list[Fleet], prefix: str) -> Fleet:
    for f in fleets:
        if f.prefix.lower() == prefix.lower():
            return f
    names = ", ".join(f.prefix for f in fleets) or "none"
    raise RuntimeError(f"No fleet named {prefix!r}. Fleets found: {names}")


def confirm(prompt: str) -> bool:
    try:
        return input(prompt).strip().lower() in ("y", "yes")
    except (EOFError, KeyboardInterrupt):
        print()
        return False


def run_plans(plans: list[Plan], args) -> int:
    for p in plans:
        p.print()
    plans = [p for p in plans if not p.empty]
    if not plans:
        return 0
    if args.dry_run:
        print("\n--dry-run: stopping here.")
        return 0
    if not args.yes and not confirm("\nApply? [y/N] "):
        print("Aborted.")
        return 1
    password = os.getenv("CAEVES_REGISTRY_PASSWORD", "")
    ok = all([p.apply(password) for p in plans])
    print_fleets(discover())
    return 0 if ok else 1


# --- commands -----------------------------------------------------------------

def cmd_status(_args) -> int:
    print_fleets(discover())
    return 0


def cmd_scale(args) -> int:
    fleet = find_fleet(discover(), args.fleet)
    print(f"Current fleet: {fleet.prefix}, {len(fleet.groups)} group(s), {fleet.running} running.")
    target_text = args.count
    if target_text is None:
        try:
            target_text = input(f"Scale to how many instances? (N, +N or -N) [{len(fleet.groups)}] ")
        except (EOFError, KeyboardInterrupt):
            print("\nAborted.")
            return 1
    target = _checked_target(target_text, len(fleet.groups))
    if target is None:
        return 1
    return run_plans([Plan(fleet, target, args.image, not args.no_rebalance, args.registry_identity)], args)


def _checked_target(text: str, current: int) -> Optional[int]:
    try:
        target = parse_target(text, current)
    except ValueError:
        print(f"Not a number: {text!r}")
        return None
    if not 1 <= target <= MAX_REPLICAS:
        print(f"Target must be between 1 and {MAX_REPLICAS} (got {target}).")
        return None
    return target


def cmd_interactive(args) -> int:
    print(f"Discovering connector fleets in subscription {SUBSCRIPTION}...")
    fleets = discover()
    print_fleets(fleets, numbered=True)
    if not fleets:
        return 1
    try:
        raw = input("\nSelect fleet(s) to scale (e.g. 1 or 1,2 or all; blank to quit): ").strip()
        if not raw:
            return 0
        if raw.lower() == "all":
            chosen = list(fleets)
        else:
            picks = [int(x) for x in re.split(r"[,\s]+", raw) if x]
            if any(not 1 <= p <= len(fleets) for p in picks):
                print("Selection out of range.")
                return 1
            chosen = [fleets[p - 1] for p in dict.fromkeys(picks)]

        plans = []
        for f in chosen:
            text = input(f"Target for {f.prefix} [{len(f.groups)}] (N, +N or -N): ")
            target = _checked_target(text, len(f.groups))
            if target is None:
                return 1
            plans.append(Plan(f, target, args.image, not args.no_rebalance, args.registry_identity))
    except ValueError:
        print("Selection must be numbers separated by commas.")
        return 1
    except (EOFError, KeyboardInterrupt):
        print("\nAborted.")
        return 1
    return run_plans(plans, args)


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Scale CAEVES connector fleets on ACI. Run with no command for the interactive selector.")
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--image", help="image override (default: whatever the reference group runs)")
    common.add_argument("--yes", action="store_true", help="skip the confirmation prompt")
    common.add_argument("--dry-run", action="store_true", help="print the plan and stop")
    common.add_argument("--registry-identity",
                        help="managed identity resource id to pull --image with (needs AcrPull); "
                             "defaults to the group's own identity for an ACR image")
    common.add_argument("--no-rebalance", action="store_true",
                        help="leave survivors alone even if their rate limit is now wrong")

    sub = parser.add_subparsers(dest="cmd")
    p_status = sub.add_parser("status", help="show every fleet and instance")
    p_status.set_defaults(func=cmd_status)

    p_scale = sub.add_parser("scale", parents=[common], help="scale one fleet to a target count")
    p_scale.add_argument("count", nargs="?",
                         help="target instances: N, +N or -N (prompts if omitted)")
    p_scale.add_argument("--fleet", default=DEFAULT_FLEET,
                         help=f"fleet name prefix (default: {DEFAULT_FLEET})")
    p_scale.set_defaults(func=cmd_scale)

    p_int = sub.add_parser("interactive", parents=[common], help="pick fleets and targets interactively")
    p_int.set_defaults(func=cmd_interactive)

    argv = sys.argv[1:]
    # No command (or only flags such as --dry-run) means the interactive selector.
    if not argv or (argv[0].startswith("-") and argv[0] not in ("-h", "--help")):
        argv = ["interactive", *argv]
    # `scale -1` would otherwise be read as an unknown option: move it behind `--`.
    if argv[0] == "scale":
        relative = [a for a in argv[1:] if re.fullmatch(r"-\d+", a)]
        if relative:
            argv = ["scale", *[a for a in argv[1:] if a not in relative], "--", *relative]
    args = parser.parse_args(argv)
    if "XXXX" in SUBSCRIPTION.upper():
        print("Error: no subscription set. Set CAEVES_SUBSCRIPTION=<subscription id>, or edit\n"
              "SUBSCRIPTION at the top of this script.", file=sys.stderr)
        return 1
    try:
        return args.func(args)
    except RuntimeError as exc:
        print(f"\nError: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())

Completion 

After completing the scaling operation, verify that all expected instances are in the Running state, the rate limits are correct, and the connector queue is configured properly.

The CAEVES Copilot Connector fleet is ready for operation once all verification checks pass and no errors remain.

Note: Always review the scaling plan using --dry-run before applying changes, and perform scaling operations during quieter periods to minimize service disruption.