feat: circuit breaker, unary failover, and graceful scale-down

.pre-commit-config.yaml

@@ -39,6 +39,14 @@ repos:
         files: ^rust/.*\.rs$
         pass_filenames: false
 
+      # Go formatting gate (needs a local toolchain). Only the operator is Go.
+      # `go vet` and the tests are heavier, so they run in CI.
+      - id: gofmt
+        name: gofmt (go)
+        entry: scripts/check-gofmt.sh
+        language: script
+        files: ^deploy/operator/.*\.go$
+
       # Refuse a commit whose git email is a machine-generated local hostname
       # (git's fallback when user.email is unset), so an internal build-host
       # name can't leak into permanent public history. Skipped in CI.

README.md

@@ -49,7 +49,8 @@ Around them:
 
 - **Multi-engine** — run vLLM, SGLang, or ATOM behind one common serving interface.
 - **OpenAI- and Anthropic-compatible API** — `/v1/chat/completions`, `/v1/completions`, and `/v1/messages` (Anthropic Messages, translated in-process).
-- **Self-registering fleet** — workers register into etcd and heartbeat, so the router works from a live view and never routes to a worker that is gone; run any number of stateless server replicas.
+- **Self-registering fleet** — workers register into etcd (or their own Pod annotation on Kubernetes) and heartbeat, so the router works from a live view and never routes to a worker that is gone; run any number of stateless server replicas.
+- **Scale without dropping requests** — a worker joins when it is ready and leaves by draining: it announces `DRAINING`, finishes the generations it already accepted, and only then deregisters. Measured on MI355X: a worker stops receiving new work **under a second** after `SIGTERM` while its in-flight 4000-token generations all complete, and adding or removing instances under continuous traffic costs **zero failed requests**. See [Scaling a fleet](https://rocm.docs.amd.com/projects/infera/en/latest/features/scaling.html).
 - **Kubernetes-native** — an operator reconciles an `InferaDeployment` CRD (aggregated / PD / multi-node), with an optional Gateway API (GAIE) endpoint picker.
 
 ## Architecture
@@ -146,6 +147,24 @@ kubectl apply -f examples/k8s-deployments/single-node-aggregated.yaml
 Ready-to-fill deployment templates (single-node, prefill/decode, multi-node TP, GAIE) and their
 placeholders are in [`examples/k8s-deployments/`](examples/k8s-deployments/README.md).
 
+**Scaling.** There is no scaling controller — workers self-register when ready and deregister when
+they drain, and the router routes to whatever is registered at that instant. Scaling is therefore
+`kubectl scale` on the CR, or starting and stopping workers; nothing has to be told about it.
+
+The two directions cost very different things, and it shapes everything built on top:
+
+| | measured |
+|---|---|
+| scale up: `docker run` → serving | **140 s**, almost all of it weight loading |
+| scale down: `SIGTERM` → stops receiving | **< 1 s** |
+| scale down: in-flight generations | run to completion, bounded by `--drain-timeout` |
+| adding + removing under traffic | **260 requests, 0 failures** |
+
+Because a cold start is minutes and the control loop is seconds, **a burst shorter than a cold
+start cannot be answered by adding workers** — keep headroom, or steer traffic to instances that
+are already running. Infera does not ship an autoscaler; [Scaling a
+fleet](https://rocm.docs.amd.com/projects/infera/en/latest/features/scaling.html) documents what is in place for one, and what is not.
+
 ## Engine images
 
 Prebuilt images are published to the `rocm/infera` repository on

deploy/operator/cmd/main.go

@@ -9,9 +9,9 @@ import (
 	"flag"
 	"os"
 
-	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"

deploy/operator/internal/controller/apply_idempotence_test.go

@@ -0,0 +1,243 @@
+/*
+Copyright (c) 2026, Advanced Micro Devices, Inc. All rights reserved.
+
+SPDX-License-Identifier: MIT
+*/
+
+package controller
+
+import (
+	"context"
+	"testing"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	inferav1alpha1 "github.com/amd/infera/deploy/operator/api/v1alpha1"
+)
+
+// The operator writes three fields of a LeaderWorkerSet; the API server fills
+// in the rest from the CRD's defaults -- startupPolicy, rolloutStrategy,
+// leaderWorkerTemplate.restartPolicy and more, nine of them on LWS v1.
+//
+// Replacing the whole .spec strips every one of those on each pass, the API
+// server puts them back, and the next pass strips them again. That was a
+// wasted write every resync; it becomes a hot loop now that the reconciler
+// watches LeaderWorkerSet, because the write it just made enqueues the
+// request that makes the next one.
+//
+// So: reconciling an object that is already in the desired state must not
+// write to it.
+func TestApplyingTheSameLwsTwiceDoesNotWriteAgain(t *testing.T) {
+	s := testScheme(t)
+	if err := inferav1alpha1.AddToScheme(s); err != nil {
+		t.Fatalf("scheme: %v", err)
+	}
+	c := fake.NewClientBuilder().WithScheme(s).Build()
+	r := &InferaDeploymentReconciler{Client: c, Scheme: s}
+	ctx := context.Background()
+
+	idep := &inferav1alpha1.InferaDeployment{}
+	idep.Name = "qwen"
+	idep.Namespace = "default"
+	idep.UID = "uid-1"
+
+	desired := func() *unstructured.Unstructured {
+		u := &unstructured.Unstructured{}
+		u.SetGroupVersionKind(lwsGVK())
+		u.SetName("qwen-worker")
+		u.SetNamespace("default")
+		_ = unstructured.SetNestedField(u.Object, int64(2), "spec", "replicas")
+		_ = unstructured.SetNestedField(u.Object, int64(2),
+			"spec", "leaderWorkerTemplate", "size")
+		return u
+	}
+
+	if err := r.applyUnstructured(ctx, idep, desired()); err != nil {
+		t.Fatalf("first apply: %v", err)
+	}
+
+	// Stand in for the API server defaulting the fields the operator omits.
+	live := &unstructured.Unstructured{}
+	live.SetGroupVersionKind(lwsGVK())
+	if err := c.Get(ctx, types.NamespacedName{Name: "qwen-worker", Namespace: "default"}, live); err != nil {
+		t.Fatalf("get after create: %v", err)
+	}
+	_ = unstructured.SetNestedField(live.Object, "LeaderCreated", "spec", "startupPolicy")
+	_ = unstructured.SetNestedField(live.Object, "RollingUpdate", "spec", "rolloutStrategy", "type")
+	_ = unstructured.SetNestedField(live.Object, "RecreateGroupOnPodRestart",
+		"spec", "leaderWorkerTemplate", "restartPolicy")
+	if err := c.Update(ctx, live); err != nil {
+		t.Fatalf("apply defaults: %v", err)
+	}
+	before := live.GetResourceVersion()
+
+	if err := r.applyUnstructured(ctx, idep, desired()); err != nil {
+		t.Fatalf("second apply: %v", err)
+	}
+
+	after := &unstructured.Unstructured{}
+	after.SetGroupVersionKind(lwsGVK())
+	if err := c.Get(ctx, types.NamespacedName{Name: "qwen-worker", Namespace: "default"}, after); err != nil {
+		t.Fatalf("get after reconcile: %v", err)
+	}
+
+	if got := after.GetResourceVersion(); got != before {
+		t.Errorf("reconcile rewrote an unchanged object (resourceVersion %s -> %s); "+
+			"with the LWS watch registered this is a write loop", before, got)
+	}
+	for _, f := range [][]string{
+		{"spec", "startupPolicy"},
+		{"spec", "rolloutStrategy", "type"},
+		{"spec", "leaderWorkerTemplate", "restartPolicy"},
+	} {
+		if v, ok, _ := unstructured.NestedString(after.Object, f...); !ok || v == "" {
+			t.Errorf("%v was stripped; the API server will re-default it and the "+
+				"next pass strips it again", f)
+		}
+	}
+}
+
+// Merging must not turn into "never update": a genuine spec change still has
+// to reach the child, or scaling through the CR would silently do nothing.
+func TestApplyStillPushesAChangedField(t *testing.T) {
+	s := testScheme(t)
+	c := fake.NewClientBuilder().WithScheme(s).Build()
+	r := &InferaDeploymentReconciler{Client: c, Scheme: s}
+	ctx := context.Background()
+
+	idep := &inferav1alpha1.InferaDeployment{}
+	idep.Name = "qwen"
+	idep.Namespace = "default"
+	idep.UID = "uid-1"
+
+	build := func(replicas int64) *unstructured.Unstructured {
+		u := &unstructured.Unstructured{}
+		u.SetGroupVersionKind(lwsGVK())
+		u.SetName("qwen-worker")
+		u.SetNamespace("default")
+		_ = unstructured.SetNestedField(u.Object, replicas, "spec", "replicas")
+		return u
+	}
+
+	if err := r.applyUnstructured(ctx, idep, build(2)); err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	if err := r.applyUnstructured(ctx, idep, build(5)); err != nil {
+		t.Fatalf("scale: %v", err)
+	}
+
+	got := &unstructured.Unstructured{}
+	got.SetGroupVersionKind(lwsGVK())
+	if err := c.Get(ctx, types.NamespacedName{Name: "qwen-worker", Namespace: "default"}, got); err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if v, _, _ := unstructured.NestedInt64(got.Object, "spec", "replicas"); v != 5 {
+		t.Fatalf("replicas = %d, want 5 -- scaling through the CR did not land", v)
+	}
+}
+
+// A field the builder emits only when the CR asks for it -- HTTPRoute's
+// hostnames -- has to disappear from the child when it disappears from the CR.
+// Merging alone cannot do that: an absent field looks the same as one the
+// operator does not manage, so the old value would survive and the route would
+// keep matching a host the user deleted.
+func TestRemovingAConditionalFieldClearsItOnTheChild(t *testing.T) {
+	s := testScheme(t)
+	c := fake.NewClientBuilder().WithScheme(s).Build()
+	r := &InferaDeploymentReconciler{Client: c, Scheme: s}
+	ctx := context.Background()
+
+	idep := &inferav1alpha1.InferaDeployment{}
+	idep.Name = "qwen"
+	idep.Namespace = "default"
+	idep.UID = "uid-1"
+
+	route := func(hostnames []any) *unstructured.Unstructured {
+		u := &unstructured.Unstructured{}
+		u.SetAPIVersion(httpRouteAPIVersion)
+		u.SetKind(httpRouteKind)
+		u.SetName("qwen-route")
+		u.SetNamespace("default")
+		spec := map[string]any{"rules": []any{}}
+		if len(hostnames) > 0 {
+			spec["hostnames"] = hostnames
+		}
+		_ = unstructured.SetNestedMap(u.Object, spec, "spec")
+		return u
+	}
+
+	if err := r.applyUnstructured(ctx, idep, route([]any{"a.example.com"})); err != nil {
+		t.Fatalf("create with hostnames: %v", err)
+	}
+	if err := r.applyUnstructured(ctx, idep, route(nil)); err != nil {
+		t.Fatalf("reapply without hostnames: %v", err)
+	}
+
+	got := &unstructured.Unstructured{}
+	got.SetAPIVersion(httpRouteAPIVersion)
+	got.SetKind(httpRouteKind)
+	if err := c.Get(ctx, types.NamespacedName{Name: "qwen-route", Namespace: "default"}, got); err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if v, ok, _ := unstructured.NestedSlice(got.Object, "spec", "hostnames"); ok {
+		t.Fatalf("hostnames still %v after removal from the CR; the route keeps "+
+			"matching a host the user deleted", v)
+	}
+}
+
+// Pruning owned fields must not start pruning the server's defaults again --
+// that is the write loop this merge exists to stop.
+func TestPruningLeavesServerDefaultsAlone(t *testing.T) {
+	s := testScheme(t)
+	c := fake.NewClientBuilder().WithScheme(s).Build()
+	r := &InferaDeploymentReconciler{Client: c, Scheme: s}
+	ctx := context.Background()
+
+	idep := &inferav1alpha1.InferaDeployment{}
+	idep.Name = "qwen"
+	idep.Namespace = "default"
+	idep.UID = "uid-1"
+
+	desired := func() *unstructured.Unstructured {
+		u := &unstructured.Unstructured{}
+		u.SetGroupVersionKind(lwsGVK())
+		u.SetName("qwen-worker")
+		u.SetNamespace("default")
+		_ = unstructured.SetNestedField(u.Object, int64(2), "spec", "replicas")
+		_ = unstructured.SetNestedField(u.Object, int64(2), "spec", "leaderWorkerTemplate", "size")
+		return u
+	}
+
+	if err := r.applyUnstructured(ctx, idep, desired()); err != nil {
+		t.Fatalf("create: %v", err)
+	}
+
+	live := &unstructured.Unstructured{}
+	live.SetGroupVersionKind(lwsGVK())
+	if err := c.Get(ctx, types.NamespacedName{Name: "qwen-worker", Namespace: "default"}, live); err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	_ = unstructured.SetNestedField(live.Object, "LeaderCreated", "spec", "startupPolicy")
+	if err := c.Update(ctx, live); err != nil {
+		t.Fatalf("apply defaults: %v", err)
+	}
+	before := live.GetResourceVersion()
+
+	if err := r.applyUnstructured(ctx, idep, desired()); err != nil {
+		t.Fatalf("second apply: %v", err)
+	}
+
+	after := &unstructured.Unstructured{}
+	after.SetGroupVersionKind(lwsGVK())
+	if err := c.Get(ctx, types.NamespacedName{Name: "qwen-worker", Namespace: "default"}, after); err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if v, ok, _ := unstructured.NestedString(after.Object, "spec", "startupPolicy"); !ok || v == "" {
+		t.Error("a server default the operator does not set was pruned")
+	}
+	if got := after.GetResourceVersion(); got != before {
+		t.Errorf("reconcile rewrote an unchanged object (%s -> %s)", before, got)
+	}
+}