diff --git a/deploy/operator/internal/controller/builders.go b/deploy/operator/internal/controller/builders.go index c28c3376..01c8b0f8 100644 --- a/deploy/operator/internal/controller/builders.go +++ b/deploy/operator/internal/controller/builders.go @@ -317,11 +317,16 @@ func envFor(idep *inferav1alpha1.InferaDeployment, svc inferav1alpha1.ServiceSpe } if useK8sDiscovery(idep) { // Pod identity for self-registration (worker) + selector for the server. + // POD_IP is what a worker advertises: it binds 0.0.0.0, and the address + // it registers is the one the router dials, so without this it + // registers its bind host and every request to it is unreachable. env = append(env, corev1.EnvVar{Name: "POD_NAME", ValueFrom: &corev1.EnvVarSource{ FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.name"}}}, corev1.EnvVar{Name: "POD_NAMESPACE", ValueFrom: &corev1.EnvVarSource{ FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.namespace"}}}, + corev1.EnvVar{Name: "POD_IP", ValueFrom: &corev1.EnvVarSource{ + FieldRef: &corev1.ObjectFieldSelector{FieldPath: "status.podIP"}}}, ) if svc.ComponentType == inferav1alpha1.ComponentTypeServer { env = append(env, corev1.EnvVar{ @@ -424,6 +429,24 @@ func injectWorkerRolloutDefaults( // exposes the service port (so buildServerService has a target). Used when // ServiceSpec.ExtraPodSpec is set (an external orchestrator renders the full // pod template). +// appendEnvIfAbsent adds each variable the container does not already declare. +// +// A template supplied by an external orchestrator may well set these itself, +// and a duplicate name in a container's env is not an error -- the last one +// wins, silently overriding what the author wrote. +func appendEnvIfAbsent(env []corev1.EnvVar, add ...corev1.EnvVar) []corev1.EnvVar { + present := make(map[string]bool, len(env)) + for _, e := range env { + present[e.Name] = true + } + for _, e := range add { + if !present[e.Name] { + env = append(env, e) + } + } + return env +} + func podTemplateFromExtra(idep *inferav1alpha1.InferaDeployment, svcName string, svc inferav1alpha1.ServiceSpec) corev1.PodTemplateSpec { spec := *svc.ExtraPodSpec.DeepCopy() port := servicePort(svc) @@ -448,11 +471,29 @@ func podTemplateFromExtra(idep *inferav1alpha1.InferaDeployment, svcName string, spec.Containers[idx].Ports = append(spec.Containers[idx].Ports, corev1.ContainerPort{ContainerPort: port}) } - // k8s discovery: the server reads its watch scope from an env var so we - // don't have to rewrite the externally-supplied entrypoint command. - if useK8sDiscovery(idep) && svc.ComponentType == inferav1alpha1.ComponentTypeServer { - spec.Containers[idx].Env = append(spec.Containers[idx].Env, corev1.EnvVar{ - Name: "INFERA_K8S_LABEL_SELECTOR", Value: discoveryLabelSelector(idep.Name)}) + if useK8sDiscovery(idep) { + // Pod identity, for both component types: a worker registers by + // patching its own Pod annotation, and the server finds the + // deployment it belongs to from its own Pod labels. Rendering the + // template elsewhere does not change that either needs to know + // which Pod it is. + spec.Containers[idx].Env = appendEnvIfAbsent(spec.Containers[idx].Env, + corev1.EnvVar{Name: "POD_NAME", ValueFrom: &corev1.EnvVarSource{ + FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.name"}}}, + corev1.EnvVar{Name: "POD_NAMESPACE", ValueFrom: &corev1.EnvVarSource{ + FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.namespace"}}}, + corev1.EnvVar{Name: "POD_IP", ValueFrom: &corev1.EnvVarSource{ + FieldRef: &corev1.ObjectFieldSelector{FieldPath: "status.podIP"}}}, + ) + // The server reads its watch scope from an env var so we don't have + // to rewrite the externally-supplied entrypoint command. + if svc.ComponentType == inferav1alpha1.ComponentTypeServer { + spec.Containers[idx].Env = appendEnvIfAbsent(spec.Containers[idx].Env, + corev1.EnvVar{ + Name: "INFERA_K8S_LABEL_SELECTOR", + Value: discoveryLabelSelector(idep.Name), + }) + } } } // Bind the discovery ServiceAccount (workers patch their own Pod; the @@ -615,11 +656,30 @@ func buildDiscoveryRole(idep *inferav1alpha1.InferaDeployment) *rbacv1.Role { Namespace: idep.Namespace, Labels: labelsFor(idep.Name, "disc"), }, - Rules: []rbacv1.PolicyRule{{ - APIGroups: []string{""}, - Resources: []string{"pods"}, - Verbs: []string{"get", "list", "watch", "patch"}, - }}, + Rules: []rbacv1.PolicyRule{ + { + APIGroups: []string{""}, + Resources: []string{"pods"}, + Verbs: []string{"get", "list", "watch", "patch"}, + }, + { + // The server's scaling API writes replica counts back to the CR + // it belongs to. Named via ResourceNames so the grant reaches + // exactly this deployment: every Pod here shares one identity, + // so an unrestricted grant would also let any worker resize the + // fleet, and a worker has no business doing that. + // + // Present whether or not --enable-scaling-api is set. The flag + // lives on the server's command line and the operator does not + // parse it; a permission nothing exercises costs nothing, while + // discovering it is absent only after enabling the feature + // costs a redeploy. + APIGroups: []string{inferav1alpha1.GroupVersion.Group}, + Resources: []string{"inferadeployments"}, + ResourceNames: []string{idep.Name}, + Verbs: []string{"get", "patch"}, + }, + }, } } diff --git a/deploy/operator/internal/controller/builders_test.go b/deploy/operator/internal/controller/builders_test.go index daf0d36d..7cf184f5 100644 --- a/deploy/operator/internal/controller/builders_test.go +++ b/deploy/operator/internal/controller/builders_test.go @@ -10,6 +10,8 @@ import ( "testing" corev1 "k8s.io/api/core/v1" + + inferav1alpha1 "github.com/amd/infera/deploy/operator/api/v1alpha1" ) // The grace period is the only thing standing between a graceful drain and a @@ -248,3 +250,106 @@ func TestDrainTimeoutStillAcceptsOrdinaryValues(t *testing.T) { } } } + +// Pod identity is what k8s discovery is built on: a worker patches its own Pod +// annotation to register, and the server reads its own labels to find the +// deployment it belongs to. Both need POD_NAME, which the operator injects -- +// on the path that renders the pod itself. A template supplied through +// extraPodSpec took a different path and got the watch selector but not the +// identity, so registration and the scaling API both failed on exactly the +// deployments the PD example tells people to write. +func TestExtraPodSpecStillGetsPodIdentity(t *testing.T) { + for _, ct := range []inferav1alpha1.ComponentType{ + inferav1alpha1.ComponentTypeServer, + inferav1alpha1.ComponentTypeWorker, + } { + idep := idepWith(1) + idep.Spec.DiscoveryBackend = "kubernetes" + svc := inferav1alpha1.ServiceSpec{ + ComponentType: ct, + ExtraPodSpec: &corev1.PodSpec{ + Containers: []corev1.Container{{Name: "main", Image: "x"}}, + }, + } + tmpl := podTemplateFromExtra(idep, "svc", svc) + got := map[string]bool{} + for _, e := range tmpl.Spec.Containers[0].Env { + got[e.Name] = true + } + for _, want := range []string{"POD_NAME", "POD_NAMESPACE"} { + if !got[want] { + t.Errorf("%s: extraPodSpec container has no %s; "+ + "self-registration and the scaling API both need it", ct, want) + } + } + } +} + +// A template that sets these itself keeps its own values: a duplicate env name +// is not an error, the last one wins, and appending ours would silently +// override whatever the author had in mind. +func TestExtraPodSpecKeepsItsOwnPodIdentity(t *testing.T) { + idep := idepWith(1) + idep.Spec.DiscoveryBackend = "kubernetes" + svc := inferav1alpha1.ServiceSpec{ + ComponentType: inferav1alpha1.ComponentTypeWorker, + ExtraPodSpec: &corev1.PodSpec{Containers: []corev1.Container{{ + Name: "main", + Image: "x", + Env: []corev1.EnvVar{{Name: "POD_NAME", Value: "chosen-by-the-author"}}, + }}}, + } + tmpl := podTemplateFromExtra(idep, "svc", svc) + + seen := 0 + for _, e := range tmpl.Spec.Containers[0].Env { + if e.Name != "POD_NAME" { + continue + } + seen++ + if e.Value != "chosen-by-the-author" { + t.Errorf("POD_NAME = %q, want the template's own value", e.Value) + } + } + if seen != 1 { + t.Errorf("POD_NAME appears %d times, want 1", seen) + } +} + +// A worker binds 0.0.0.0 and advertises something else, because the address it +// registers is the one the router dials. Under k8s discovery it resolves that +// from POD_IP -- the logic is already there and reads the downward API -- so +// leaving the variable out makes the worker register 0.0.0.0 and every request +// to it fail with "worker unreachable". +// +// Measured before this was injected: the worker came up healthy, registered, +// and the router returned {"error":"worker 0.0.0.0:8080 unreachable"} for the +// first inference request. +func TestWorkersLearnTheirOwnAddress(t *testing.T) { + idep := idepWith(1) + idep.Spec.DiscoveryBackend = "kubernetes" + + check := func(t *testing.T, env []corev1.EnvVar, where string) { + t.Helper() + for _, e := range env { + if e.Name != "POD_IP" { + continue + } + if e.ValueFrom == nil || e.ValueFrom.FieldRef == nil || + e.ValueFrom.FieldRef.FieldPath != "status.podIP" { + t.Errorf("%s: POD_IP is not read from the downward API", where) + } + return + } + t.Errorf("%s: no POD_IP; the worker would advertise its bind address", where) + } + + svc := inferav1alpha1.ServiceSpec{ComponentType: inferav1alpha1.ComponentTypeWorker} + check(t, envFor(idep, svc), "rendered pod") + + svc.ExtraPodSpec = &corev1.PodSpec{ + Containers: []corev1.Container{{Name: "main", Image: "x"}}, + } + tmpl := podTemplateFromExtra(idep, "worker", svc) + check(t, tmpl.Spec.Containers[0].Env, "extraPodSpec") +} diff --git a/deploy/operator/internal/controller/scale_paths_test.go b/deploy/operator/internal/controller/scale_paths_test.go index 1d8a48d7..e9f7dc30 100644 --- a/deploy/operator/internal/controller/scale_paths_test.go +++ b/deploy/operator/internal/controller/scale_paths_test.go @@ -314,3 +314,53 @@ func TestEditingTheChildLWSIsAlsoReverted(t *testing.T) { t.Fatalf("LWS scale survived reconciliation: %d groups, want it reverted to 2", got) } } + +// The server's scaling API writes replica counts back to the CR, which needs a +// grant the discovery identity did not previously carry. Every Pod in a +// deployment shares that identity, so the grant has to name the one CR it may +// touch: without ResourceNames a worker could resize the fleet it belongs to, +// or any other deployment in the namespace. +func TestTheDiscoveryRoleCanWriteOnlyItsOwnDeployment(t *testing.T) { + idep := idepWith(2) + role := buildDiscoveryRole(idep) + + var found *rbacv1.PolicyRule + for i := range role.Rules { + for _, res := range role.Rules[i].Resources { + if res == "inferadeployments" { + found = &role.Rules[i] + } + } + } + if found == nil { + t.Fatal("no grant for inferadeployments: the scaling API would 403") + } + if len(found.ResourceNames) != 1 || found.ResourceNames[0] != idep.Name { + t.Fatalf("ResourceNames = %v, want exactly [%s]: an unscoped grant lets "+ + "any Pod here resize any deployment in the namespace", + found.ResourceNames, idep.Name) + } + for _, verb := range found.Verbs { + switch verb { + case "get", "patch": + default: + t.Errorf("verb %q is more than the scaling API needs", verb) + } + } +} + +// Pods are a separate rule and must stay unscoped: the server lists and watches +// every worker Pod, which ResourceNames cannot express. +func TestThePodGrantIsUnchanged(t *testing.T) { + role := buildDiscoveryRole(idepWith(1)) + for _, rule := range role.Rules { + for _, res := range rule.Resources { + if res != "pods" { + continue + } + if len(rule.ResourceNames) != 0 { + t.Fatal("the Pod grant must not be scoped by name; discovery lists all of them") + } + } + } +} diff --git a/infera/server/__main__.py b/infera/server/__main__.py index 499d346f..75057355 100644 --- a/infera/server/__main__.py +++ b/infera/server/__main__.py @@ -26,6 +26,7 @@ from infera.router.policy.factory import build_policy from infera.server.app import init_app from infera.server.args import parse_server_args +from infera.server.scaling import DeploymentScaler logging.basicConfig(level=logging.INFO) # httpx logs every outbound request (etcd keepalives every ~10s, and one @@ -260,12 +261,24 @@ def on_worker_removed(worker_id: str) -> None: request_max_retries=args.request_max_retries, breaker=breaker, ) + scaler = None + if args.enable_scaling_api: + # Resolved lazily on first call: the deployment is read from this Pod's + # labels, and failing at startup over a permission the operator grants + # would take down a server whose main job does not need it. + scaler = DeploymentScaler(namespace=args.k8s_namespace or None) + logger.info( + "scaling API enabled: GET/POST /v1/admin/scale resizes this " + "deployment's pools (needs RBAC to patch inferadeployments)" + ) + app = init_app( registry, router, kv=policy.kv_client, kvd_socket_path=args.kvd_socket_path, enable_profiling=args.enable_profiling, + scaler=scaler, ) app.include_router( make_stats_router( diff --git a/infera/server/app.py b/infera/server/app.py index 38cfffcc..13961b88 100644 --- a/infera/server/app.py +++ b/infera/server/app.py @@ -22,6 +22,7 @@ from infera.router.policy.target import expand_targets from infera.server import metrics from infera.server.profiling import fan_out_profile, select_targets +from infera.server.scaling import DeploymentScaler, ScalingError logger = logging.getLogger(__name__) @@ -66,6 +67,10 @@ def _stash_direct_worker(body: dict, request: Request) -> None: _enable_profiling: bool = False _profile_client: httpx.AsyncClient | None = None +# Pool resizing. Off by default: it writes to the cluster, and /v1/admin carries +# no authentication of its own, so it is opt-in the way profiling is. +_scaler: DeploymentScaler | None = None + def init_app( reg: Registry, @@ -73,14 +78,16 @@ def init_app( kv: KvEventClient | None = None, kvd_socket_path: str | None = None, enable_profiling: bool = False, + scaler: DeploymentScaler | None = None, ) -> FastAPI: global registry, router, kv_client, _kvd_socket_path - global _enable_profiling + global _enable_profiling, _scaler registry = reg router = rtr kv_client = kv _kvd_socket_path = kvd_socket_path _enable_profiling = enable_profiling + _scaler = scaler return app @@ -93,9 +100,10 @@ def _get_profile_client() -> httpx.AsyncClient: # ------------------------------------------------------------------ -# Read-only worker inspection -# (Workers register themselves directly with etcd; the server is purely -# a reader of that state, so there are no write endpoints here.) +# Worker inspection +# (Workers register themselves, so the registry is read-only here. The one +# write in this file is /v1/admin/scale, which edits the InferaDeployment +# rather than any registry state -- see infera.server.scaling.) # ------------------------------------------------------------------ @@ -173,6 +181,64 @@ async def profile_stop(request: Request) -> dict: return await _profile_action("stop", request) +# ------------------------------------------------------------------ +# Pool sizes +# (Off unless --enable-scaling-api. Edits the InferaDeployment the operator +# reconciles from; see infera.server.scaling for why not the workloads.) +# ------------------------------------------------------------------ + + +def _require_scaler() -> DeploymentScaler: + if _scaler is None: + raise HTTPException( + status_code=403, + detail="scaling API disabled; start the server with --enable-scaling-api", + ) + return _scaler + + +@app.get("/v1/admin/scale") +async def scale_get() -> dict: + """Each pool's requested and observed size.""" + scaler = _require_scaler() + try: + return await scaler.snapshot() + except ScalingError as exc: + raise HTTPException(status_code=exc.status, detail=str(exc)) from exc + + +@app.post("/v1/admin/scale") +async def scale_set(request: Request) -> dict: + """Resize pools, all of the named ones or none. + + Body: ``{"services": {"prefill": {"replicas": 4}, "decode": {"replicas": 8}}}`` + Returns the snapshot after the write, so a caller sees both what it asked + for and how far the cluster has got. + """ + scaler = _require_scaler() + try: + body = await request.json() + except (json.JSONDecodeError, ValueError) as exc: + raise HTTPException(status_code=400, detail="body must be JSON") from exc + + services = (body or {}).get("services") + if not isinstance(services, dict): + raise HTTPException( + status_code=400, + detail='body must be {"services": {"": {"replicas": N}}}', + ) + requested: dict[str, object] = {} + for name, cfg in services.items(): + # Accept the nested form the CR uses, so a caller can paste from one to + # the other, and the bare number for the common single-value case. + requested[name] = cfg.get("replicas") if isinstance(cfg, dict) else cfg + + try: + return await scaler.scale(requested) + except ScalingError as exc: + raise HTTPException(status_code=exc.status, detail=str(exc)) from exc + + @app.get("/v1/workers") async def list_workers() -> dict: workers = [ diff --git a/infera/server/args.py b/infera/server/args.py index bddfa807..7a059f74 100644 --- a/infera/server/args.py +++ b/infera/server/args.py @@ -196,6 +196,17 @@ def parse_server_args(argv: list[str] | None = None) -> argparse.Namespace: "unless DYN_SYSTEM_PORT is set. Also enabled via " "$INFERA_ENABLE_PROFILING=1.", ) + parser.add_argument( + "--enable-scaling-api", + action="store_true", + default=os.environ.get("INFERA_ENABLE_SCALING_API", "").lower() in ("1", "true", "yes"), + help="Enable pool resizing over the admin API: GET/POST /v1/admin/scale " + "reads and writes the replica counts on the InferaDeployment this " + "server belongs to. Needs the operator (the deployment is found from " + "this Pod's labels) and RBAC to patch inferadeployments. Default OFF " + "(returns 403), since /v1/admin carries no authentication of its own. " + "Also enabled via $INFERA_ENABLE_SCALING_API=1.", + ) parser.add_argument( "--request-max-retries", type=int, diff --git a/infera/server/scaling.py b/infera/server/scaling.py new file mode 100644 index 00000000..47a18cd4 --- /dev/null +++ b/infera/server/scaling.py @@ -0,0 +1,285 @@ +############################################################################### +# Copyright (c) 2026, Advanced Micro Devices, Inc. All rights reserved. +# +# SPDX-License-Identifier: MIT +############################################################################### +"""Changing pool sizes through the router. + +The supported way to resize a pool is to edit `spec.services..replicas` on +the InferaDeployment, which the operator then reconciles. That is a +``kubectl patch`` -- fine for a human, awkward for the orchestration layer that +usually decides these things, which then needs cluster credentials and has to +know the CR's shape. + +This puts that same edit behind the router's admin API. It writes the CR and +nothing else: the generated Deployment and LeaderWorkerSet are derived state, +rewritten on every reconcile pass, so scaling those directly succeeds, reports +no error, and is silently undone a few seconds later. + +**Not an autoscaler, and not a step toward one.** It is a way to issue a +decision that has already been made. Two things make the automated version a +different problem: a worker takes around two minutes to become useful, which is +longer than most bursts last, and Kubernetes offers no way to choose *which* +replica to remove, so a scale-down is as likely to take the worker holding the +warmest cache as any other. +""" + +from __future__ import annotations + +import json +import logging +import os + +import httpx + +from infera.common.k8s_client import in_cluster_namespace, make_client + +logger = logging.getLogger(__name__) + +# Written by the operator onto every workload it generates; the router reads its +# own to find the deployment it belongs to, rather than being told separately. +LABEL_DEPLOYMENT = "infera.amd.com/deployment" +LABEL_SERVICE = "infera.amd.com/service" + +_CR_GROUP = "infera.amd.com" +_CR_VERSION = "v1alpha1" +_CR_PLURAL = "inferadeployments" + +# Container states a Pod does not recover from on its own. Distinguished from +# the ordinary ones (ContainerCreating, PodInitializing) because those resolve +# in seconds and reporting them would make every scale-up look stuck. +_TERMINAL_WAITS = frozenset( + {"ImagePullBackOff", "ErrImagePull", "CrashLoopBackOff", "CreateContainerConfigError"} +) + + +def _one_line(reason: str | None, message: str | None) -> str: + """A reason and its detail on one line, short enough to log. + + Scheduler messages enumerate every node and run to several hundred + characters; the head of one carries the verdict. + """ + text = " ".join((message or "").split()) + if len(text) > 300: + text = text[:297] + "..." + if reason and text: + return f"{reason}: {text}" + return reason or text or "unknown" + + +class ScalingError(Exception): + """A scale request that cannot be honoured, with a reason to return.""" + + def __init__(self, message: str, *, status: int = 400) -> None: + super().__init__(message) + self.status = status + + +class DeploymentScaler: + """Reads and resizes the pools of one InferaDeployment.""" + + def __init__(self, *, namespace: str | None = None, pod_name: str | None = None) -> None: + self._namespace = namespace or in_cluster_namespace() + self._pod_name = pod_name or os.environ.get("POD_NAME", "") + self._deployment: str | None = None + + async def _resolve_deployment(self, client: httpx.AsyncClient) -> str: + """Find the InferaDeployment this router was created by. + + Read from the router's own Pod labels rather than configured, so the + two cannot disagree: a name passed by flag would keep pointing at the + old deployment after a rename, and resize a fleet nobody asked about. + """ + if self._deployment: + return self._deployment + if not self._pod_name: + raise ScalingError( + "scaling needs POD_NAME (downward API) to find its own deployment", + status=503, + ) + resp = await client.get(f"/api/v1/namespaces/{self._namespace}/pods/{self._pod_name}") + if resp.status_code == 403: + raise ScalingError( + "the router's ServiceAccount cannot read its own Pod; re-apply the operator RBAC", + status=403, + ) + if resp.status_code == 404: + raise ScalingError( + f"router Pod {self._namespace}/{self._pod_name} not found; cannot resolve deployment to scale", + status=503, + ) + resp.raise_for_status() + labels = (resp.json().get("metadata") or {}).get("labels") or {} + name = labels.get(LABEL_DEPLOYMENT) + if not name: + raise ScalingError( + "this router was not created by the operator " + f"(no {LABEL_DEPLOYMENT} label), so there is no deployment to scale", + status=409, + ) + self._deployment = name + return name + + def _cr_path(self, name: str) -> str: + return f"/apis/{_CR_GROUP}/{_CR_VERSION}/namespaces/{self._namespace}/{_CR_PLURAL}/{name}" + + async def _fetch(self, client: httpx.AsyncClient, name: str) -> dict: + resp = await client.get(self._cr_path(name)) + if resp.status_code == 403: + raise ScalingError( + "the router's ServiceAccount cannot read InferaDeployments; " + "re-apply the operator RBAC", + status=403, + ) + if resp.status_code == 404: + raise ScalingError(f"InferaDeployment {name!r} not found", status=404) + resp.raise_for_status() + return resp.json() + + async def snapshot(self) -> dict: + """Every pool's requested and observed size. + + Both, because they answer different questions: the spec is what was + asked for, the status is what the cluster has managed so far, and a + caller deciding whether to scale again needs to know a previous request + is still landing. + """ + async with make_client() as client: + name = await self._resolve_deployment(client) + cr = await self._fetch(client, name) + spec = (cr.get("spec") or {}).get("services") or {} + status = (cr.get("status") or {}).get("services") or {} + pools = {} + for svc, cfg in sorted(spec.items()): + observed = status.get(svc) or {} + want = int(cfg.get("replicas", 1)) + pools[svc] = pool = { + "role": cfg.get("role") or "mixed", + "replicas": want, + "current_replicas": int(observed.get("replicas", 0)), + "ready_replicas": int(observed.get("readyReplicas", 0)), + "nodes_per_replica": int(cfg.get("numberOfNodes", 1)), + } + if pool["ready_replicas"] < want: + # Short of what was asked for. Whether that is a scale-up + # still landing or one that never will is not visible in the + # counts -- a replica the scheduler could not place is + # missing from both -- so the reason comes from the Pods. + blocked = await self._blocked_reason(client, name, svc) + if blocked: + pool["blocked"] = blocked + + return { + "deployment": name, + "namespace": self._namespace, + # The operator's own verdict, which compares ready replicas against + # the spec rather than against the ones that exist. That is the + # comparison a caller wants and the easy one to get wrong. + "state": (cr.get("status") or {}).get("state") or "unknown", + "services": pools, + } + + async def _blocked_reason(self, client: httpx.AsyncClient, name: str, svc: str) -> str | None: + """Why this pool's Pods are not running, if the cluster says so. + + Reported rather than diagnosed: the message is the scheduler's or the + kubelet's, so a caller polling this endpoint can tell "still starting" + from "will never start" without reaching for kubectl. Best-effort -- + a pool short of replicas is worth reporting even when the reason is not + available. + """ + selector = f"{LABEL_DEPLOYMENT}={name},{LABEL_SERVICE}={svc}" + try: + resp = await client.get( + f"/api/v1/namespaces/{self._namespace}/pods", + params={"labelSelector": selector}, + ) + resp.raise_for_status() + pods = resp.json().get("items") or [] + except Exception as exc: # noqa: BLE001 - a missing reason is not an error + logger.debug("could not read Pods for %s/%s: %s", name, svc, exc) + return None + + for pod in pods: + status = pod.get("status") or {} + if status.get("phase") != "Pending": + continue + for cond in status.get("conditions") or []: + if cond.get("type") == "PodScheduled" and cond.get("status") == "False": + return _one_line(cond.get("reason"), cond.get("message")) + for cs in status.get("containerStatuses") or []: + waiting = (cs.get("state") or {}).get("waiting") or {} + # ImagePullBackOff and friends: scheduled, but never going to + # run, which reads the same as a slow start from the counts. + if waiting.get("reason") in _TERMINAL_WAITS: + return _one_line(waiting.get("reason"), waiting.get("message")) + return None + + async def scale(self, requested: dict[str, int]) -> dict: + """Resize the named pools, all of them or none. + + One patch carries the whole request, so a caller rebalancing prefill + against decode cannot end up with one applied and the other rejected -- + a state neither the caller nor the cluster asked for, and one that could + leave a PD deployment lopsided until someone noticed. + """ + if not requested: + raise ScalingError("no services named") + + async with make_client() as client: + name = await self._resolve_deployment(client) + cr = await self._fetch(client, name) + known = (cr.get("spec") or {}).get("services") or {} + _validate(requested, known) + + patch = {"spec": {"services": {s: {"replicas": n} for s, n in requested.items()}}} + resp = await client.patch( + self._cr_path(name), + content=json.dumps(patch), + headers={"Content-Type": "application/merge-patch+json"}, + ) + if resp.status_code == 403: + raise ScalingError( + "the router's ServiceAccount cannot patch InferaDeployments; " + "re-apply the operator RBAC", + status=403, + ) + resp.raise_for_status() + + logger.info( + "scaled %s: %s", + name, + ", ".join(f"{s}={n}" for s, n in sorted(requested.items())), + ) + return await self.snapshot() + + +def _validate(requested: dict[str, int], known: dict) -> None: + """Reject what the cluster would accept but nobody wants. + + The API server would take any of these -- the CR's own validation only + bounds `replicas` at zero -- and the damage would show up later as workers + that never arrive or a pool that stops serving. + """ + unknown = sorted(set(requested) - set(known)) + if unknown: + raise ScalingError( + f"no such service(s): {', '.join(unknown)}. " + f"This deployment has: {', '.join(sorted(known))}" + ) + + for svc, count in sorted(requested.items()): + if not isinstance(count, int) or isinstance(count, bool): + raise ScalingError(f"{svc}: replicas must be an integer, got {count!r}") + if count < 1: + # Zero is a valid CR value and a valid thing to want -- for a pool + # being retired. It is not something to reach through this API by + # accident, and for a PD role it takes the whole deployment down: + # dispatch fails closed when either side is empty, so emptying one + # returns 503 for every request, not just the ones that would have + # landed there. + raise ScalingError( + f"{svc}: refusing to scale to {count}. A pool at zero stops serving, " + "and in a PD deployment an empty prefill or decode pool fails every " + "request. Edit the InferaDeployment directly if that is the intent." + ) diff --git a/manual/features/scaling.md b/manual/features/scaling.md index 4c4653cf..19cbc5a9 100644 --- a/manual/features/scaling.md +++ b/manual/features/scaling.md @@ -416,23 +416,148 @@ continuous traffic: **200 requests, 0 failures**, both pools scaling independently and the drained workers finishing their in-flight work. Taking the last prefill away then returns 503 naming the empty pool. +**Scaling over the API.** A PD deployment at 1 prefill / 2 decode resized to +3 / 4 by one `POST /v1/admin/scale`: the CR carried both new counts immediately +and the Pods reached them **within 35 s**. During that window the read reported +`replicas` 3 and 4 against `current_replicas` 1 and 2. + +**Scaling over the API, real engine.** One vLLM worker (Qwen3-8B, one MI300X) +scaled to two and back. The added worker loaded weights, registered, and +**served 4 of the 8 requests** issued after it joined; the remaining worker +served normally after the scale-down. + +A pool scaled to zero, an unknown service name and a negative count are each +refused with the CR unchanged, and a server without `--enable-scaling-api` +answers 403. + ```{warning} **Not measured:** multi-node workers, TP > 1, PD scaling with a *real* engine -(the run above used GPU-free stand-ins, so no KV moved), and scale-down during an -active KV transfer. The PD handoff queues are counted in the drain, but that +(the runs above used GPU-free stand-ins, so no KV moved), and scale-down during +an active KV transfer. The PD handoff queues are counted in the drain, but that path has not been exercised on hardware. + +The PD API run used stand-ins; the real-engine API run was a single mixed pool. +Neither covers a scale-down draining real work — that is the drain measurement +above. ``` ## Scaling a deployment -Edit the service's `replicas` in the `InferaDeployment`. That is the only -supported way in, and it is the only write that survives: +This section is about deployments the operator manages. Elsewhere — external +etcd, no CR — scaling is starting and stopping worker processes, as above. + +Edit the service's `replicas` in the `InferaDeployment`. That is the write that +counts — every supported path ends there: ```bash kubectl patch inferadeployment qwen --type=merge \ -p '{"spec":{"services":{"decode":{"replicas":5}}}}' ``` +Or over the server's admin API, for an orchestration layer that would otherwise +need cluster credentials and knowledge of the CR's shape: + +```bash +infera-server --enable-scaling-api # off by default + +curl -X POST http://router:8000/v1/admin/scale \ + -H 'Content-Type: application/json' \ + -d '{"services": {"prefill": 4, "decode": 8}}' +``` + +Counts are absolute rather than deltas, and every pool named in one request moves +in a single write, so a rebalance cannot half-apply. Pools not named are left +alone. + +`GET` on the same path returns the same shape, as does the `POST` once it has +written: + +```json +{ + "deployment": "qwen", + "namespace": "infera", + "state": "pending", + "services": { + "decode": { + "role": "decode", + "replicas": 8, + "current_replicas": 3, + "ready_replicas": 3, + "nodes_per_replica": 1, + "blocked": "Unschedulable: 0/13 nodes are available: 8 Insufficient amd.com/gpu." + } + } +} +``` + +The three replica counts are separate: + +- **`replicas`** — what was asked for, the value on the CR. +- **`current_replicas`** — Pods that exist. +- **`ready_replicas`** — workers registered and serving. + +A scale-up has landed when **`ready_replicas` reaches `replicas`**, and `state` +says the same thing for the deployment as a whole. Compare against `replicas` +rather than against `current_replicas`: a replica the scheduler could not place +is missing from *both* observed counts, so they agree with each other while the +pool is half its requested size — which is exactly the response above. + +`blocked` appears on a pool short of what it asked for, when the cluster says +why: the scheduler's message for a Pod it could not place, or the kubelet's for +one that will not start. It is absent while Pods are merely starting, since a +model takes minutes to load and that is not a fault. + +`nodes_per_replica` above 1 makes `replicas` a count of **groups** rather than +Pods — see below. + +### Waiting for a scale-up + +The write returns as soon as the CR is updated; the operator does the rest, and +reconciles every 15 seconds. Poll until `ready_replicas` reaches `replicas`. + +Nothing times out. A request for more replicas than the cluster can place is not +rejected and does not expire — the Pods stay `Pending` and the operator keeps +trying, indefinitely, until capacity appears or the count is lowered. That is +the usual behaviour of a Kubernetes controller, and it means **deciding when to +give up is the caller's**. + +`blocked` is what makes that decision early rather than on a timeout: + +- **absent, counts rising** — the scale-up is landing. A Pod exists within + seconds and serves once its model is loaded, which is minutes. +- **present** — the cluster has said this will not start on its own. Waiting + longer will not help. + +To give up, write the previous count back. The operator removes the surplus +Pods, `Pending` ones included. Existing workers are untouched throughout: Pods +that cannot be placed never displace ones that are already serving, so a +scale-up that fails leaves the pool exactly as it was. + +The API refuses to take a pool to zero: an empty pool stops serving, and in a PD +deployment an empty prefill or decode pool fails *every* request rather than +only the ones that would have landed there. Retiring a pool is a `kubectl` edit, +where the intent is unambiguous. A negative count and a service name that is not +in the CR are refused the same way, the latter listing the names that are: + +```json +{"detail": "no such service(s): prefil. This deployment has: decode, prefill, server"} +``` + +A refused request writes nothing, so nothing is left half-applied. + +```{note} +Both paths write the `InferaDeployment`, so both need one. Deployments outside +Kubernetes — the external-etcd shape described in [Across +machines](#across-machines) — have no CR and no operator to reconcile it, and +scale there by starting and stopping worker processes, as +[Scaling up](#scaling-up) and [Scaling down](#scaling-down) describe. The API +answers 409 on a server that was not created by the operator. + +Where there is an operator, the API also needs the RBAC that ships with it. A +cluster running an operator from before this feature has a server that cannot +write; re-apply the operator manifests to get the grant. +``` + For a multi-node service the count is **groups**, not pods: `replicas: 5` with `numberOfNodes: 3` is fifteen pods and five servable instances, since only node-rank 0 of each group registers. @@ -452,15 +577,18 @@ snapping back. ## Autoscaling -Infera ships no autoscaler, and there is currently no `/scale` surface for an -external one to drive. - -An `InferaDeployment` cannot carry `/scale` itself, and that is a property of -its shape rather than an omission: `spec.services` is a map with user-chosen -keys, while the scale subresource requires `specReplicasPath` to be a *static* -dot-notation JSONPath, and a CRD may declare only one. A single path could name -one service — hardcoding `decode`, say — which leaves every other pool, and in -a PD deployment specifically the prefill pool, with no handle at all. +Infera ships no autoscaler. `/v1/admin/scale` gives an external one somewhere to +push a decision to, but it is an entry point rather than a control loop: nothing +in Infera watches load and decides. + +An `InferaDeployment` cannot carry a Kubernetes `/scale` subresource, and that +is a property of its shape rather than an omission: `spec.services` is a map +with user-chosen keys, while the scale subresource requires `specReplicasPath` +to be a *static* dot-notation JSONPath, and a CRD may declare only one. A single +path could name one service — hardcoding `decode`, say — which leaves every +other pool, and in a PD deployment specifically the prefill pool, with no handle +at all. The admin API sidesteps that by naming the service in the request +instead of the path. Pointing an autoscaler at the generated workload does not work either, for the reason in the warning above: those objects are derived state and are rewritten diff --git a/manual/getting_started/quickstart.md b/manual/getting_started/quickstart.md index 7f5c8f73..c3338132 100644 --- a/manual/getting_started/quickstart.md +++ b/manual/getting_started/quickstart.md @@ -195,6 +195,18 @@ The router picked one of the two workers for you — that's a working Infera sta (`--discovery-backend etcd --request-transport http --kv-event-transport zmq`). ``` +## 6. Scale it + +Repeat step 3 on another port to add a worker, `SIGTERM` one to remove it. +Nothing needs to be told: workers register and deregister themselves, and the +router routes to whatever is registered at that instant. + +On Kubernetes the server can do it for you, over +[an admin API](../features/scaling.md#scaling-a-deployment) that resizes the +pools of the `InferaDeployment` it belongs to — including prefill and decode +independently. That path needs a deployment the operator created, so it is not +available on this page's etcd setup. + ## Where to go next - **Route by cache locality** instead of round-robin → @@ -203,5 +215,7 @@ The router picked one of the two workers for you — that's a working Infera sta [PD disaggregation](../features/pd_disaggregation.md) - **Keep KV warm** across restarts on RAM/NVMe → [KV-Cache Offload](../features/kv_cache_offload.md) +- **Add and remove workers** while traffic flows → + [Scaling a fleet](../features/scaling.md) - **Deploy for real** with Kubernetes → [Deployment](../serving/deployment.md) diff --git a/tests/unit/server/test_scaling.py b/tests/unit/server/test_scaling.py new file mode 100644 index 00000000..06d38221 --- /dev/null +++ b/tests/unit/server/test_scaling.py @@ -0,0 +1,421 @@ +############################################################################### +# Copyright (c) 2026, Advanced Micro Devices, Inc. All rights reserved. +# +# SPDX-License-Identifier: MIT +############################################################################### +"""Resizing pools through the router. + +The write itself is one PATCH, so most of what matters here is what never +reaches the cluster: a request that would empty a pool, name one that does not +exist, or apply to some services and not others. +""" + +from __future__ import annotations + +import json + +import pytest + +from infera.server.scaling import DeploymentScaler, ScalingError + +POD = {"metadata": {"labels": {"infera.amd.com/deployment": "qwen"}}} + +CR = { + "spec": { + "services": { + "server": {"componentType": "server", "replicas": 1}, + "prefill": {"componentType": "worker", "role": "prefill", "replicas": 2}, + "decode": {"componentType": "worker", "role": "decode", "replicas": 4}, + } + }, + "status": { + "state": "ready", + "services": { + "server": {"replicas": 1, "readyReplicas": 1}, + "prefill": {"replicas": 2, "readyReplicas": 2}, + "decode": {"replicas": 4, "readyReplicas": 3}, + }, + }, +} + + +class _Resp: + def __init__(self, status=200, payload=None): + self.status_code = status + self._payload = payload if payload is not None else {} + + def json(self): + return self._payload + + def raise_for_status(self): + if self.status_code >= 400: + raise AssertionError(f"unexpected {self.status_code}") + + +class _FakeApi: + """Stands in for the API server, recording what was written to it.""" + + def __init__(self, *, pod=POD, cr=None, get_status=200, patch_status=200, pods=None): + self.pod = pod + self.pods = pods or [] + self.cr = json.loads(json.dumps(cr if cr is not None else CR)) + self.get_status = get_status + self.patch_status = patch_status + self.patches: list[dict] = [] + + async def get(self, path, params=None): + if path.endswith("/pods"): + return _Resp(200, {"items": self.pods}) + if "/pods/" in path: + return _Resp(200 if self.pod else 404, self.pod) + if self.get_status != 200: + return _Resp(self.get_status) + return _Resp(200, self.cr) + + async def patch(self, path, content=None, headers=None): + if self.patch_status != 200: + return _Resp(self.patch_status) + body = json.loads(content) + self.patches.append(body) + # Reflect the write, so the snapshot returned afterwards is the new state. + for svc, cfg in body["spec"]["services"].items(): + self.cr["spec"]["services"][svc]["replicas"] = cfg["replicas"] + return _Resp(200, self.cr) + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +@pytest.fixture +def api(monkeypatch): + fake = _FakeApi() + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + monkeypatch.setattr("infera.server.scaling.in_cluster_namespace", lambda *a, **k: "infera") + return fake + + +def _scaler(): + return DeploymentScaler(namespace="infera", pod_name="router-abc") + + +@pytest.mark.asyncio +async def test_a_snapshot_reports_both_asked_for_and_achieved(api): + """A caller deciding whether to scale again needs to know whether the last + request has landed.""" + snap = await _scaler().snapshot() + + assert snap["deployment"] == "qwen" + assert snap["services"]["decode"] == { + "role": "decode", + "replicas": 4, + "current_replicas": 4, + "ready_replicas": 3, + "nodes_per_replica": 1, + } + + +@pytest.mark.asyncio +async def test_prefill_and_decode_move_in_one_write(api): + """Rebalancing one against the other must not be able to half-apply: a PD + deployment left lopsided serves badly until someone notices.""" + await _scaler().scale({"prefill": 4, "decode": 8}) + + assert len(api.patches) == 1, "a partial application must not be possible" + assert api.patches[0] == { + "spec": {"services": {"prefill": {"replicas": 4}, "decode": {"replicas": 8}}} + } + + +@pytest.mark.asyncio +async def test_only_the_named_services_are_touched(api): + await _scaler().scale({"decode": 6}) + + written = api.patches[0]["spec"]["services"] + assert set(written) == {"decode"} + assert api.cr["spec"]["services"]["prefill"]["replicas"] == 2 + + +@pytest.mark.asyncio +async def test_the_deployment_is_found_from_the_routers_own_pod(api): + """Configuring the name separately would let the two disagree -- and a stale + one would resize a fleet nobody asked about.""" + await _scaler().scale({"decode": 5}) + assert api.cr["spec"]["services"]["decode"]["replicas"] == 5 + + +@pytest.mark.asyncio +async def test_an_unknown_service_is_refused_before_anything_is_written(api): + with pytest.raises(ScalingError) as err: + await _scaler().scale({"decode": 5, "prefil": 2}) # typo + + assert "prefil" in str(err.value) + assert "prefill" in str(err.value), "the message should say what does exist" + assert api.patches == [], "a rejected request must not partially apply" + + +@pytest.mark.asyncio +async def test_scaling_to_zero_is_refused(api): + """An empty pool stops serving, and in PD an empty side fails every request + rather than only the ones that would have landed there.""" + with pytest.raises(ScalingError) as err: + await _scaler().scale({"decode": 0}) + + assert "stops serving" in str(err.value) + assert api.patches == [] + + +@pytest.mark.asyncio +async def test_a_negative_count_is_refused(api): + with pytest.raises(ScalingError): + await _scaler().scale({"decode": -1}) + assert api.patches == [] + + +@pytest.mark.asyncio +async def test_a_non_integer_count_is_refused(api): + for bad in ("4", 4.5, None, True): + with pytest.raises(ScalingError): + await _scaler().scale({"decode": bad}) + assert api.patches == [] + + +@pytest.mark.asyncio +async def test_an_empty_request_is_refused(api): + with pytest.raises(ScalingError): + await _scaler().scale({}) + + +@pytest.mark.asyncio +async def test_a_router_the_operator_did_not_create_says_so(monkeypatch): + """Deployed by hand, there is no deployment to resize -- and no CR that + would be the right one to guess at.""" + fake = _FakeApi(pod={"metadata": {"labels": {}}}) + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + + with pytest.raises(ScalingError) as err: + await _scaler().snapshot() + + assert err.value.status == 409 + assert "not created by the operator" in str(err.value) + + +@pytest.mark.asyncio +async def test_a_missing_permission_names_the_fix(monkeypatch): + """The RBAC for this ships with the operator, so an older deployment has a + router that cannot write. That is a setup step, not a bug to debug.""" + fake = _FakeApi(get_status=403) + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + + with pytest.raises(ScalingError) as err: + await _scaler().snapshot() + + assert err.value.status == 403 + assert "re-apply the operator RBAC" in str(err.value) + + +@pytest.mark.asyncio +async def test_a_missing_deployment_is_reported_as_such(monkeypatch): + fake = _FakeApi(get_status=404) + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + + with pytest.raises(ScalingError) as err: + await _scaler().snapshot() + assert err.value.status == 404 + + +# ------------------------------------------------------------------ +# Telling "still starting" from "never will" +# ------------------------------------------------------------------ + + +def pending_pod(reason, message, *, kind="scheduling"): + if kind == "scheduling": + return { + "status": { + "phase": "Pending", + "conditions": [ + { + "type": "PodScheduled", + "status": "False", + "reason": reason, + "message": message, + } + ], + } + } + return { + "status": { + "phase": "Pending", + "conditions": [{"type": "PodScheduled", "status": "True"}], + "containerStatuses": [{"state": {"waiting": {"reason": reason, "message": message}}}], + } + } + + +def short_cr(want=8, ready=3): + """A pool the cluster could not fill: the replicas that were never created + are missing from the status, so the two counts agree with each other and + disagree with the spec.""" + return { + "spec": {"services": {"decode": {"role": "decode", "replicas": want}}}, + "status": { + "state": "pending", + "services": {"decode": {"replicas": ready, "readyReplicas": ready}}, + }, + } + + +@pytest.mark.asyncio +async def test_the_operators_own_verdict_is_reported(api): + """It compares ready replicas against the spec, which is the comparison a + caller wants and the easy one to get wrong.""" + assert (await _scaler().snapshot())["state"] == "ready" + + +@pytest.mark.asyncio +async def test_a_pool_the_scheduler_could_not_place_says_so(monkeypatch): + """The counts alone cannot distinguish this from a slow start: a replica + that was never created is absent from both of them.""" + fake = _FakeApi( + cr=short_cr(), + pods=[ + pending_pod("Unschedulable", "0/13 nodes are available: 8 Insufficient amd.com/gpu.") + ], + ) + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + + pool = (await _scaler().snapshot())["services"]["decode"] + + assert pool["replicas"] == 8 + assert pool["ready_replicas"] == 3 + assert "Insufficient amd.com/gpu" in pool["blocked"] + assert pool["blocked"].startswith("Unschedulable:") + + +@pytest.mark.asyncio +async def test_a_pod_that_will_never_start_is_reported_too(monkeypatch): + """Scheduled but stuck reads the same as starting, from the counts.""" + fake = _FakeApi( + cr=short_cr(), + pods=[pending_pod("ImagePullBackOff", "Back-off pulling image", kind="waiting")], + ) + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + + blocked = (await _scaler().snapshot())["services"]["decode"]["blocked"] + assert blocked.startswith("ImagePullBackOff:") + + +@pytest.mark.asyncio +async def test_a_pod_merely_starting_is_not_reported_as_blocked(monkeypatch): + """ContainerCreating resolves in seconds. Reporting it would make every + scale-up look stuck for as long as a model takes to load.""" + fake = _FakeApi( + cr=short_cr(), + pods=[pending_pod("ContainerCreating", "", kind="waiting")], + ) + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + + assert "blocked" not in (await _scaler().snapshot())["services"]["decode"] + + +@pytest.mark.asyncio +async def test_a_pool_that_is_up_is_not_investigated(monkeypatch): + """No gap, no Pod query: the common path should not pay for the rare one.""" + fake = _FakeApi( + cr=short_cr(want=3, ready=3), + pods=[pending_pod("Unschedulable", "should not be read")], + ) + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + + assert "blocked" not in (await _scaler().snapshot())["services"]["decode"] + + +@pytest.mark.asyncio +async def test_an_unreadable_pod_list_does_not_hide_the_counts(monkeypatch): + """The reason is a convenience; the numbers are the answer.""" + + class _NoPods(_FakeApi): + async def get(self, path, params=None): + if path.endswith("/pods"): + raise RuntimeError("forbidden") + return await super().get(path, params) + + fake = _NoPods(cr=short_cr()) + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + + pool = (await _scaler().snapshot())["services"]["decode"] + assert pool["ready_replicas"] == 3 + assert "blocked" not in pool + + +@pytest.mark.asyncio +async def test_a_long_scheduler_message_is_cut_to_one_line(monkeypatch): + """Scheduler messages enumerate every node; the head carries the verdict.""" + fake = _FakeApi( + cr=short_cr(), + pods=[pending_pod("Unschedulable", "node-a: no gpu.\n" + "x" * 900)], + ) + monkeypatch.setattr("infera.server.scaling.make_client", lambda **kw: fake) + + blocked = (await _scaler().snapshot())["services"]["decode"]["blocked"] + assert len(blocked) < 350 + assert "\n" not in blocked + + +# ------------------------------------------------------------------ +# The HTTP surface +# ------------------------------------------------------------------ + + +def _client(scaler): + from fastapi.testclient import TestClient + + from infera.server import app as app_module + + app_module._scaler = scaler + return TestClient(app_module.app) + + +def test_the_endpoints_are_off_unless_enabled(): + """It writes to the cluster and /v1/admin has no authentication of its own, + so an operator has to ask for it.""" + c = _client(None) + assert c.get("/v1/admin/scale").status_code == 403 + assert c.post("/v1/admin/scale", json={"services": {"decode": 2}}).status_code == 403 + assert "--enable-scaling-api" in c.get("/v1/admin/scale").json()["detail"] + + +def test_a_scale_request_reports_the_state_it_produced(api): + c = _client(_scaler()) + resp = c.post("/v1/admin/scale", json={"services": {"decode": {"replicas": 6}}}) + + assert resp.status_code == 200 + assert resp.json()["services"]["decode"]["replicas"] == 6 + assert api.patches[0]["spec"]["services"] == {"decode": {"replicas": 6}} + + +def test_a_bare_number_is_accepted_like_the_nested_form(api): + """The nested form matches the CR, so a caller can paste between the two; + the bare number is what anyone writes by hand.""" + c = _client(_scaler()) + assert c.post("/v1/admin/scale", json={"services": {"decode": 3}}).status_code == 200 + assert api.patches[0]["spec"]["services"] == {"decode": {"replicas": 3}} + + +def test_a_refused_request_says_why(api): + c = _client(_scaler()) + resp = c.post("/v1/admin/scale", json={"services": {"decode": 0}}) + + assert resp.status_code == 400 + assert "stops serving" in resp.json()["detail"] + assert api.patches == [] + + +def test_a_malformed_body_is_rejected(api): + c = _client(_scaler()) + assert c.post("/v1/admin/scale", json={"decode": 3}).status_code == 400 + assert c.post("/v1/admin/scale", json={"services": 3}).status_code == 400 + assert api.patches == []