diff --git a/api/v1beta1/nodeinventory_types.go b/api/v1beta1/nodeinventory_types.go index d192c6d..33f4f1d 100644 --- a/api/v1beta1/nodeinventory_types.go +++ b/api/v1beta1/nodeinventory_types.go @@ -87,6 +87,9 @@ type NodeInventory struct { // already holds a warm microVM for a requested (template, net, size). // +optional Pools []PoolCapacity `json:"pools,omitempty"` + // publishedAt is the publisher's last apply; the aggregated apiserver skips an inventory older than its staleness window. + // +optional + PublishedAt metav1.Time `json:"publishedAt,omitempty"` } // +kubebuilder:object:root=true diff --git a/api/v1beta1/zz_generated.deepcopy.go b/api/v1beta1/zz_generated.deepcopy.go index e5a6822..5858c20 100644 --- a/api/v1beta1/zz_generated.deepcopy.go +++ b/api/v1beta1/zz_generated.deepcopy.go @@ -63,6 +63,7 @@ func (in *NodeInventory) DeepCopyInto(out *NodeInventory) { *out = make([]PoolCapacity, len(*in)) copy(*out, *in) } + in.PublishedAt.DeepCopyInto(&out.PublishedAt) } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new NodeInventory. diff --git a/cmd/sandbox-apiserver/main.go b/cmd/sandbox-apiserver/main.go index fd9fcf8..ee19d5c 100644 --- a/cmd/sandbox-apiserver/main.go +++ b/cmd/sandbox-apiserver/main.go @@ -53,6 +53,8 @@ type options struct { E2BAPI bool E2B *e2bcompat.Flags + + Inventory kubeinventory.Options } func newOptions() *options { @@ -88,6 +90,7 @@ func (o *options) addFlags(fs *pflag.FlagSet) { fs.BoolVar(&o.E2BAPI, "enable-e2b-api", o.E2BAPI, "Serve the e2b-compatible REST surface, so an unmodified e2b SDK can claim from the same warm pools (point E2B_API_URL at it).") o.E2B.AddFlags(fs) + o.Inventory.AddFlags(fs) } func (o *options) serverConfig() (*genericapiserver.Config, error) { @@ -139,7 +142,7 @@ func run() error { if err != nil { return err } - invSource := kubeinventory.New(reader) + invSource := kubeinventory.New(reader, o.Inventory) store := scale.NewScatterGatherStore( invSource, scale.WithClaimRouting(token, scale.NewSandboxdClientFactory()), diff --git a/cmd/sandbox-envd-proxy/main.go b/cmd/sandbox-envd-proxy/main.go index c6a1149..d7d418b 100644 --- a/cmd/sandbox-envd-proxy/main.go +++ b/cmd/sandbox-envd-proxy/main.go @@ -32,10 +32,12 @@ type options struct { Domain string Namespace string Proxy envdproxy.Flags + Inventory kubeinventory.Options } func (o *options) addFlags(fs *pflag.FlagSet) { o.Proxy.AddFlags(fs, "") + o.Inventory.AddFlags(fs) fs.StringVar(&o.Domain, "domain", o.Domain, "Base domain sandbox hosts are derived from, as {port}-{sandboxID}.{domain}. Must match the apiserver's --e2b-domain.") fs.StringVar(&o.Namespace, "namespace", o.Namespace, @@ -70,7 +72,7 @@ func run(ctx context.Context, o *options) error { if err != nil { return err } - inv := kubeinventory.New(reader) + inv := kubeinventory.New(reader, o.Inventory) resolver, err := envdproxy.NewResolver(scale.NewScatterGatherStore(inv), inv, o.Namespace) if err != nil { return err diff --git a/docs/api.md b/docs/api.md index 58493aa..3977722 100644 --- a/docs/api.md +++ b/docs/api.md @@ -67,6 +67,7 @@ _Appears in:_ | `entries` _[InventoryEntry](#inventoryentry) array_ | entries summarizes the node's live sandboxes. | | Optional: \{\}
| | `address` _string_ | address is the node's sandboxd advertise address ("host:port"); the
aggregated apiserver routes a claim to this node's sandboxd through it. | | Optional: \{\}
| | `pools` _[PoolCapacity](#poolcapacity) array_ | pools is the node's per-pool warm capacity, used to pick a node that
already holds a warm microVM for a requested (template, net, size). | | Optional: \{\}
| +| `publishedAt` _[Time](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.37/#time-v1-meta)_ | publishedAt is the publisher's last apply; the aggregated apiserver skips an inventory older than its staleness window. | | Optional: \{\}
| #### NodeInventoryList diff --git a/docs/configuration.md b/docs/configuration.md index 6a1f595..b3a5888 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -134,6 +134,31 @@ The driver watches `SandboxWarmPool` and `NodeInventory`, resolves each pool's under leader election (`cocoon-warmpool-driver`), so one replica drives the pools. An empty sandboxd token leaves it fail-closed: it logs and sets no pools. +### Node inventory freshness + +| Flag | Default | Help | +|---|---|---| +| `--inventory-stale-after` | `90s` | Drop a node from this process's inventory reads once its NodeInventory `publishedAt` trails the newest publish in the fleet by more than this; set the same value on sandbox-apiserver and sandbox-envd-proxy. An inventory without `publishedAt`, from a vk-sandbox that predates the field, always stays. | + +A node that stops publishing, because it died or its vk-sandbox stopped, +leaves every read and claim path once its `publishedAt` trails the newest +publish by more than this. At vk-sandbox's default 30 s publish cadence that is +three missed publishes, and since the newest publish only advances when a live +node publishes, a dead node leaves between this age and one publish interval +later. Keep it above vk-sandbox's `--publish-interval`. + +The age is measured against the newest publish the process has seen, capped by +its own clock. A control-plane write outage freezes every stamp together, so it +never empties the fleet, and a node's clock matters only relative to the other +nodes: one running ahead cannot evict the rest. A fleet whose nodes all stop +publishing at once, or a one-node fleet, keeps its last inventory until the +nodes publish again. + +The window works on a node once the `nodeinventories` CRD from `helm/crds` is +applied and that node runs a vk-sandbox that stamps `publishedAt`. An older CRD +prunes the field and an older vk-sandbox never sets it, and either way that +node never goes stale, so an upgrade never empties the fleet. + ### e2b-compatible surface | Flag | Default | Help | @@ -176,6 +201,7 @@ informer and relays the request into the owning node's guest-port endpoint. | `--tls-cert-file` | — | Wildcard certificate for `*.{domain}`. Omit to serve cleartext h2c behind an edge that terminates TLS. | | `--tls-private-key-file` | — | Private key for `--tls-cert-file`. | | `--guest-http2` | `false` | Forward to the guest over cleartext HTTP/2. Off by default: envd 0.8.0 installs no h2c handler and refuses it. Clients still reach this proxy over HTTP/2. | +| `--inventory-stale-after` | `90s` | Drop a node from this process's inventory reads once its NodeInventory `publishedAt` trails the newest publish in the fleet by more than this; set the same value on sandbox-apiserver and sandbox-envd-proxy. An inventory without `publishedAt`, from a vk-sandbox that predates the field, always stays. | The two TLS flags must be set together. Routing, authorization and failure mapping are in [envd-proxy](envd-proxy.md). diff --git a/docs/e2b-compat.md b/docs/e2b-compat.md index 297ead5..b0aeb89 100644 --- a/docs/e2b-compat.md +++ b/docs/e2b-compat.md @@ -143,6 +143,13 @@ const sandbox = await Sandbox.create('registry.example.com/rt:24.04') once by `files`/`commands` works. The proxy's node lookups share one budget per replica (200 new sandboxes/s, see [envd-proxy](envd-proxy.md)); past it a sandbox its node has not published answers `502` until the node publishes. +- **A node that stops publishing leaves after `--inventory-stale-after`** + (90 s by default). Until then a dead node's sandboxes stay listed and a + claim can still sample it; past it they leave `GET /sandboxes`, and + `GET`, `DELETE` and the verbs on them answer `404`. The window is measured + against the fleet's newest publish, so a control-plane outage never drops + a node, and a vk-sandbox restart inside the window is invisible. A + vk-sandbox that predates `publishedAt` never goes stale. - **`envdVersion`** is reported as `0.4.0` unless `--e2b-envd-version` says otherwise. The SDK version-compares it and *kills the sandbox* if it cannot parse it, so it is always sent. Set it to the version actually installed in diff --git a/docs/scaling-design.md b/docs/scaling-design.md index 566c55e..9a86345 100644 --- a/docs/scaling-design.md +++ b/docs/scaling-design.md @@ -161,6 +161,8 @@ v1beta1 the `APIService` hands to the aggregated server, which serves only | Node partitioned from aggregated server | Its sandboxes briefly absent from `List` (eventual consistency, same as an informer lag) | No | | A node `inventory` object lost | Rebuilt from the node's own live state on next publish | No | | A node's `Node` object deleted | Its `NodeInventory` is garbage-collected with it: the node leaves the claim path and its sandboxes leave the read view, though sandboxd keeps serving them and their leases still expire there; restarting vk-sandbox registers the node again | No | +| A node dies, its `Node` object kept | Its `NodeInventory` is no longer republished. Once its `publishedAt` trails the newest publish in the fleet by more than `--inventory-stale-after` (90 s), the node leaves the claim path and its sandboxes leave the read view, the watch emits Deleted for each, Get and the verbs answer 404, and the warm-pool driver and envd-proxy stop calling it. Its next publish brings it back. A vk-sandbox restart shorter than the window changes nothing | No | +| The control plane stops accepting writes | Every publish fails and every stamp freezes together, so no node goes stale. envd-proxy and every apiserver replica but one keep claims, lists and lookups running from the informer cache, which shows the inventory as of the outage. The replica holding the warm-pool driver's lease exits once it cannot renew it (about 10 s) and starts again when the control plane returns | No | | Aggregated server restart | Stateless; rebuilds from node fan-out | No | | Client reads before the owning node republishes inventory | Lookups by name and by claim id (e2b, envd proxy), and a list or watch pinned to one name in a namespace, ask the nodes; other lists and watches show it at the next publish | No — fleet `list` and `watch`, and a deleted sandbox until its node publishes, stay eventually consistent | diff --git a/helm/README.md b/helm/README.md index 59525ac..150972a 100644 --- a/helm/README.md +++ b/helm/README.md @@ -51,9 +51,10 @@ kubectl apply -f helm/crds/ helm upgrade sandbox-operator ./helm --namespace sandbox-system --reuse-values ``` -An older `nodeinventories` CRD silently prunes entry fields it does not know -(`claimedAt` is one), so until the CRD is applied every node publishes like an -older node and the read view falls back accordingly. +An older `nodeinventories` CRD silently prunes fields it does not know +(`claimedAt` and `publishedAt` are two), so until the CRD is applied every node +publishes like an older node: the read view falls back accordingly, and no node +ever goes stale under `--inventory-stale-after`. `NodeInventory` moved from `extensions.agents.x-k8s.io` to `sandbox.cocoonstack.io`. A fleet coming from the old group runs a vk-sandbox diff --git a/helm/crds/sandbox.cocoonstack.io_nodeinventories.yaml b/helm/crds/sandbox.cocoonstack.io_nodeinventories.yaml index 0449971..f851098 100644 --- a/helm/crds/sandbox.cocoonstack.io_nodeinventories.yaml +++ b/helm/crds/sandbox.cocoonstack.io_nodeinventories.yaml @@ -80,6 +80,9 @@ spec: - warm type: object type: array + publishedAt: + format: date-time + type: string required: - node type: object diff --git a/pkg/scale/inventorycache_bench_test.go b/pkg/scale/inventorycache_bench_test.go index c6a2fac..61b8834 100644 --- a/pkg/scale/inventorycache_bench_test.go +++ b/pkg/scale/inventorycache_bench_test.go @@ -96,6 +96,7 @@ func benchCachedStore(b *testing.B, nodes, perNode int, noCopy bool) (*scale.Sca invs, pool := scale.BenchInventories(nodes, perNode) list := &unstructured.UnstructuredList{} for _, inv := range invs { + inv.PublishedAt = metav1.Now() raw, err := runtime.DefaultUnstructuredConverter.ToUnstructured(inv) if err != nil { b.Fatalf("encode node inventory: %v", err) @@ -137,5 +138,5 @@ func benchCachedStore(b *testing.B, nodes, perNode int, noCopy bool) (*scale.Sca if !invCache.WaitForCacheSync(syncCtx) { b.Fatal("node inventory cache did not sync") } - return scale.NewScatterGatherStore(kubeinventory.New(invCache)).(*scale.ScatterGatherStore), pool + return scale.NewScatterGatherStore(kubeinventory.New(invCache, kubeinventory.Options{})).(*scale.ScatterGatherStore), pool } diff --git a/pkg/scale/kubeinventory/source.go b/pkg/scale/kubeinventory/source.go index 15ecddd..b46beca 100644 --- a/pkg/scale/kubeinventory/source.go +++ b/pkg/scale/kubeinventory/source.go @@ -2,29 +2,53 @@ package kubeinventory import ( + "cmp" "context" "fmt" "slices" + "sync/atomic" + "time" + "github.com/spf13/pflag" + k8serrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" "github.com/cocoonstack/sandbox-operator/pkg/scale" ) +const defaultStaleAfter = 90 * time.Second + +var nodeInventories = schema.GroupResource{Group: scale.NodeInventoryGVK.Group, Resource: "nodeinventories"} + +// Options tunes a Source. +type Options struct { + // StaleAfter is how old publishedAt may get before its node leaves the fleet; zero means 90 s. + StaleAfter time.Duration +} + +// AddFlags registers the source flags on fs. +func (o *Options) AddFlags(fs *pflag.FlagSet) { + fs.DurationVar(&o.StaleAfter, "inventory-stale-after", cmp.Or(o.StaleAfter, defaultStaleAfter), + "Drop a node from this process's inventory reads once its NodeInventory publishedAt trails the newest publish in the fleet by more than this; set the same value on sandbox-apiserver and sandbox-envd-proxy. An inventory without publishedAt, from a vk-sandbox that predates the field, always stays.") +} + var _ scale.InventorySource = (*Source)(nil) // Source is the production InventorySource over a cache-fed NodeInventory reader. // It never mutates what it reads, since NewCache hands out its cached objects themselves. type Source struct { - reader client.Reader + reader client.Reader + staleAfter time.Duration + reference atomic.Int64 } // New builds a Source over reader. -func New(reader client.Reader) *Source { - return &Source{reader: reader} +func New(reader client.Reader, opts Options) *Source { + return &Source{reader: reader, staleAfter: cmp.Or(opts.StaleAfter, defaultStaleAfter)} } func (s *Source) ListNodes(ctx context.Context) ([]string, error) { @@ -33,19 +57,29 @@ func (s *Source) ListNodes(ctx context.Context) ([]string, error) { if err := s.reader.List(ctx, ul); err != nil { return nil, fmt.Errorf("kubeinventory: list node inventories: %w", err) } - nodes := make([]string, 0, len(ul.Items)) + nodes := make([]string, len(ul.Items)) + stamps := make([]int64, len(ul.Items)) + var newest int64 for i := range ul.Items { - nodes = append(nodes, ul.Items[i].GetName()) + nodes[i], stamps[i] = ul.Items[i].GetName(), publishedAt(ul.Items[i].Object) + newest = max(newest, stamps[i]) + } + ref := min(newest, time.Now().UnixNano()) + s.reference.Store(ref) + kept := nodes[:0] + for i, node := range nodes { + if !s.stale(stamps[i], ref) { + kept = append(kept, node) + } } - slices.Sort(nodes) - return nodes, nil + slices.Sort(kept) + return kept, nil } func (s *Source) NodeInventory(ctx context.Context, node string) (*scale.NodeInventory, error) { - u := &unstructured.Unstructured{} - u.SetGroupVersionKind(scale.NodeInventoryGVK) - if err := s.reader.Get(ctx, types.NamespacedName{Name: node}, u); err != nil { - return nil, fmt.Errorf("kubeinventory: get node %q inventory: %w", node, err) + u, err := s.get(ctx, node) + if err != nil { + return nil, err } inv := &scale.NodeInventory{} if err := runtime.DefaultUnstructuredConverter.FromUnstructured(u.Object, inv); err != nil { @@ -55,10 +89,9 @@ func (s *Source) NodeInventory(ctx context.Context, node string) (*scale.NodeInv } func (s *Source) NodeCapacity(ctx context.Context, node string) (string, []scale.PoolCapacity, error) { - u := &unstructured.Unstructured{} - u.SetGroupVersionKind(scale.NodeInventoryGVK) - if err := s.reader.Get(ctx, types.NamespacedName{Name: node}, u); err != nil { - return "", nil, fmt.Errorf("kubeinventory: get node %q inventory: %w", node, err) + u, err := s.get(ctx, node) + if err != nil { + return "", nil, err } addr, _, err := unstructured.NestedString(u.Object, "address") if err != nil { @@ -82,3 +115,31 @@ func (s *Source) NodeCapacity(ctx context.Context, node string) (string, []scale } return addr, pools, nil } + +func (s *Source) get(ctx context.Context, node string) (*unstructured.Unstructured, error) { + u := &unstructured.Unstructured{} + u.SetGroupVersionKind(scale.NodeInventoryGVK) + if err := s.reader.Get(ctx, types.NamespacedName{Name: node}, u); err != nil { + return nil, fmt.Errorf("kubeinventory: get node %q inventory: %w", node, err) + } + if s.stale(publishedAt(u.Object), s.reference.Load()) { + return nil, fmt.Errorf("kubeinventory: node %q inventory is stale: %w", node, k8serrors.NewNotFound(nodeInventories, node)) + } + return u, nil +} + +func (s *Source) stale(stamp, ref int64) bool { + return stamp != 0 && ref-stamp > int64(s.staleAfter) +} + +func publishedAt(obj map[string]any) int64 { + raw, _ := obj["publishedAt"].(string) + if raw == "" { + return 0 + } + at, err := time.Parse(time.RFC3339, raw) + if err != nil { + return 0 + } + return at.UnixNano() +} diff --git a/pkg/scale/kubeinventory/source_test.go b/pkg/scale/kubeinventory/source_test.go new file mode 100644 index 0000000..380b602 --- /dev/null +++ b/pkg/scale/kubeinventory/source_test.go @@ -0,0 +1,122 @@ +package kubeinventory + +import ( + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + k8serrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + cocoonv1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1" + "github.com/cocoonstack/sandbox-operator/pkg/scale" +) + +func TestSourceDropsADeadNodeAgainstTheNewestPublish(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx := t.Context() + c := fakeClient(t, inventory("live", metav1.Now()), inventory("dead", metav1.Now()), inventory("unstamped", metav1.Time{})) + src := New(c, Options{}) + + time.Sleep(90 * time.Second) + republish(t, c, "live") + nodes, err := src.ListNodes(ctx) + require.NoError(t, err) + assert.Equal(t, []string{"dead", "live", "unstamped"}, nodes) + + time.Sleep(time.Second) + republish(t, c, "live") + nodes, err = src.ListNodes(ctx) + require.NoError(t, err) + assert.Equal(t, []string{"live", "unstamped"}, nodes) + _, err = src.NodeInventory(ctx, "dead") + assert.True(t, k8serrors.IsNotFound(err), "NodeInventory on a stale node: %v", err) + _, _, err = src.NodeCapacity(ctx, "dead") + assert.True(t, k8serrors.IsNotFound(err), "NodeCapacity on a stale node: %v", err) + addr, pools, err := src.NodeCapacity(ctx, "live") + require.NoError(t, err) + assert.Equal(t, "live:7777", addr) + assert.Len(t, pools, 1) + }) +} + +func TestSourceKeepsEveryNodeThroughAPublishOutage(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx := t.Context() + src := New(fakeClient(t, inventory("a", metav1.Now()), inventory("b", metav1.Now())), Options{}) + time.Sleep(10 * time.Minute) + nodes, err := src.ListNodes(ctx) + require.NoError(t, err) + assert.Equal(t, []string{"a", "b"}, nodes) + _, err = src.NodeInventory(ctx, "a") + require.NoError(t, err, "a frozen fleet still answers lookups") + }) +} + +func TestSourceCapsTheReferenceAtTheClock(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ahead := metav1.NewTime(time.Now().Add(time.Hour)) + src := New(fakeClient(t, inventory("rogue", ahead), inventory("a", metav1.Now()), inventory("b", metav1.Now())), Options{}) + nodes, err := src.ListNodes(t.Context()) + require.NoError(t, err) + assert.Equal(t, []string{"a", "b", "rogue"}, nodes, "a stamp ahead of the clock must not evict the others") + }) +} + +func TestSourceLookupBeforeAnyListCountsFresh(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + src := New(fakeClient(t, inventory("a", metav1.Now())), Options{}) + time.Sleep(time.Hour) + _, _, err := src.NodeCapacity(t.Context(), "a") + require.NoError(t, err) + }) +} + +func TestSourceHonorsStaleAfter(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := fakeClient(t, inventory("live", metav1.Now()), inventory("dead", metav1.Now())) + src := New(c, Options{StaleAfter: 10 * time.Second}) + time.Sleep(11 * time.Second) + republish(t, c, "live") + nodes, err := src.ListNodes(t.Context()) + require.NoError(t, err) + assert.Equal(t, []string{"live"}, nodes) + }) +} + +func TestPublishedAtTreatsAMissingOrUnparsableStampAsUnset(t *testing.T) { + for _, obj := range []map[string]any{{}, {"publishedAt": nil}, {"publishedAt": "not a time"}} { + assert.Zero(t, publishedAt(obj), "%v", obj) + } + assert.Equal(t, time.Date(2026, 9, 27, 9, 0, 0, 0, time.UTC).UnixNano(), publishedAt(map[string]any{"publishedAt": "2026-09-27T09:00:00Z"})) +} + +func fakeClient(t *testing.T, objs ...client.Object) client.Client { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, cocoonv1beta1.AddToScheme(scheme)) + return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build() +} + +func republish(t *testing.T, c client.Client, node string) { + t.Helper() + inv := &cocoonv1beta1.NodeInventory{} + require.NoError(t, c.Get(t.Context(), client.ObjectKey{Name: node}, inv)) + inv.PublishedAt = metav1.Now() + require.NoError(t, c.Update(t.Context(), inv)) +} + +func inventory(node string, publishedAt metav1.Time) *cocoonv1beta1.NodeInventory { + return &cocoonv1beta1.NodeInventory{ + Name: node, + Node: node, + Address: node + ":7777", + Pools: []scale.PoolCapacity{{Template: "rt", Warm: 1, Target: 1}}, + PublishedAt: publishedAt, + } +}