diff --git a/cmd/sandbox-operator/main.go b/cmd/sandbox-operator/main.go index 000a9d1..7aae742 100644 --- a/cmd/sandbox-operator/main.go +++ b/cmd/sandbox-operator/main.go @@ -57,6 +57,22 @@ import ( var setupLog = ctrl.Log.WithName("setup") +func main() { + var o options + o.bindFlags() + flag.Parse() + + if o.printVersion { + fmt.Println(version.Print("sandbox-operator")) + return + } + ctrl.SetLogger(zap.New(zap.UseFlagOptions(&o.zap))) + if err := o.run(); err != nil { + setupLog.Error(err, "sandbox-operator exited") + os.Exit(1) + } +} + // options holds every operator flag. It is one struct so main stays a // sequence of named phases rather than a 400-line body. type options struct { @@ -155,22 +171,6 @@ func (o *options) totalWorkers() int { return o.sandboxWorkers + o.claimWorkers + o.warmPoolWorkers + o.templateWorkers } -func main() { - var o options - o.bindFlags() - flag.Parse() - - if o.printVersion { - fmt.Println(version.Print("sandbox-operator")) - return - } - ctrl.SetLogger(zap.New(zap.UseFlagOptions(&o.zap))) - if err := o.run(); err != nil { - setupLog.Error(err, "sandbox-operator exited") - os.Exit(1) - } -} - func (o *options) run() error { if err := o.validate(); err != nil { return err @@ -269,15 +269,6 @@ func (o *options) logSettings() { } } -func setupTracing(ctx context.Context, enabled bool) (asmetrics.Instrumenter, func(), error) { - if !enabled { - return asmetrics.NewNoOp(), func() {}, nil - } - initCtx, cancel := context.WithTimeout(ctx, 10*time.Second) - defer cancel() - return asmetrics.SetupOTel(initCtx, "sandbox-operator") -} - func (o *options) pprofHandlers() map[string]http.Handler { if !o.enablePprof && !o.enablePprofDebug { return nil @@ -407,6 +398,15 @@ func (o *options) setupExtensionControllers(mgr ctrl.Manager, instrumenter asmet return nil } +func setupTracing(ctx context.Context, enabled bool) (asmetrics.Instrumenter, func(), error) { + if !enabled { + return asmetrics.NewNoOp(), func() {}, nil + } + initCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + return asmetrics.SetupOTel(initCtx, "sandbox-operator") +} + // readAllowedLabelDomains reads the optional label-domain allowlist mounted by // the deployment; a missing file leaves the allowlist empty. func readAllowedLabelDomains() ([]string, error) { diff --git a/controllers/sandbox_controller.go b/controllers/sandbox_controller.go index 8564568..1c85a7a 100644 --- a/controllers/sandbox_controller.go +++ b/controllers/sandbox_controller.go @@ -77,15 +77,6 @@ var Scheme = func() *runtime.Scheme { // resourceOwnership represents the ownership state of a Kubernetes resource relative to a Sandbox. type resourceOwnership int -// SandboxReconciler reconciles a Sandbox object. -type SandboxReconciler struct { - client.Client - Scheme *runtime.Scheme - Tracer asmetrics.Instrumenter - ClusterDomain string - PodMutator PodMutator -} - // PodMutator applies runtime-specific defaults to a newly-created Sandbox Pod. // Implementations must preserve every field explicitly supplied through the // Sandbox API and return an error instead of silently overriding conflicts. @@ -103,6 +94,15 @@ type PodMutator interface { //+kubebuilder:rbac:groups=coordination.k8s.io,resources=leases,verbs=get;list;watch;create;update;patch //+kubebuilder:rbac:groups=apiextensions.k8s.io,resources=customresourcedefinitions,verbs=get;update;patch,resourceNames=sandboxes.agents.x-k8s.io;sandboxclaims.extensions.agents.x-k8s.io;sandboxtemplates.extensions.agents.x-k8s.io;sandboxwarmpools.extensions.agents.x-k8s.io +// SandboxReconciler reconciles a Sandbox object. +type SandboxReconciler struct { + client.Client + Scheme *runtime.Scheme + Tracer asmetrics.Instrumenter + ClusterDomain string + PodMutator PodMutator +} + // Reconcile drives a Sandbox toward its declared spec: child Service/Pod/PVCs, // status conditions, and expiry. func (r *SandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { @@ -1022,6 +1022,54 @@ func (r *SandboxReconciler) deleteExpiredChild(ctx context.Context, sandbox *san return nil } +// checkOwnership determines whether a Kubernetes resource is owned by the given Sandbox, +// has no controller, or is owned by a different controller. +// It returns both the ownership classification and the controller reference (if any), +// so callers can log owner details without redundant GetControllerOf calls. +func checkOwnership(obj client.Object, sandbox *sandboxv1beta1.Sandbox) (resourceOwnership, *metav1.OwnerReference) { + controllerRef := metav1.GetControllerOf(obj) + if controllerRef == nil { + return resourceUnowned, nil + } + if controllerRef.UID == sandbox.UID { + return resourceOwnedBySandbox, controllerRef + } + return resourceOwnedByOther, controllerRef +} + +// MergeVolumeClaimVolumes merges PVC-backed volumes into an existing volume +// list, replacing any volumes with matching names. This follows StatefulSet +// semantics where volumeClaimTemplate volumes take priority. +func MergeVolumeClaimVolumes(existing []corev1.Volume, pvcVolumes []corev1.Volume) []corev1.Volume { + if len(pvcVolumes) == 0 { + return existing + } + vctNames := make(map[string]struct{}, len(pvcVolumes)) + for _, v := range pvcVolumes { + vctNames[v.Name] = struct{}{} + } + filtered := make([]corev1.Volume, 0, len(existing)) + for _, v := range existing { + if _, ok := vctNames[v.Name]; !ok { + filtered = append(filtered, v) + } + } + return append(filtered, pvcVolumes...) +} + +// GetNumericHash generates a raw FNV-1a hash value. +func GetNumericHash(input string) uint32 { + h := fnv.New32a() + _, _ = h.Write([]byte(input)) + return h.Sum32() +} + +// NameHash generates an FNV-1a hash from a string and returns +// it as a fixed-length hexadecimal string. +func NameHash(objectName string) string { + return fmt.Sprintf("%08x", GetNumericHash(objectName)) +} + // checks if the sandbox has expired // returns true if expired, false otherwise // if not expired, also returns the duration to requeue after. @@ -1090,21 +1138,6 @@ func podReadiness(pod *corev1.Pod) (message string, ready bool) { return "Pod is Ready", true } -// checkOwnership determines whether a Kubernetes resource is owned by the given Sandbox, -// has no controller, or is owned by a different controller. -// It returns both the ownership classification and the controller reference (if any), -// so callers can log owner details without redundant GetControllerOf calls. -func checkOwnership(obj client.Object, sandbox *sandboxv1beta1.Sandbox) (resourceOwnership, *metav1.OwnerReference) { - controllerRef := metav1.GetControllerOf(obj) - if controllerRef == nil { - return resourceUnowned, nil - } - if controllerRef.UID == sandbox.UID { - return resourceOwnedBySandbox, controllerRef - } - return resourceOwnedByOther, controllerRef -} - // isAdoptable reports whether obj carries the warm-pool adoptable label. func isAdoptable(obj client.Object) bool { return obj.GetLabels()[sandboxv1beta1.SandboxAdoptableLabel] == "true" @@ -1120,26 +1153,6 @@ func resolvePodName(sandbox *sandboxv1beta1.Sandbox) string { return sandbox.Name } -// MergeVolumeClaimVolumes merges PVC-backed volumes into an existing volume -// list, replacing any volumes with matching names. This follows StatefulSet -// semantics where volumeClaimTemplate volumes take priority. -func MergeVolumeClaimVolumes(existing []corev1.Volume, pvcVolumes []corev1.Volume) []corev1.Volume { - if len(pvcVolumes) == 0 { - return existing - } - vctNames := make(map[string]struct{}, len(pvcVolumes)) - for _, v := range pvcVolumes { - vctNames[v.Name] = struct{}{} - } - filtered := make([]corev1.Volume, 0, len(existing)) - for _, v := range existing { - if _, ok := vctNames[v.Name]; !ok { - filtered = append(filtered, v) - } - } - return append(filtered, pvcVolumes...) -} - // podIPsFromStatus converts the K8s PodIP slice to a plain string slice. func podIPsFromStatus(podIPs []corev1.PodIP) []string { if len(podIPs) == 0 { @@ -1152,19 +1165,6 @@ func podIPsFromStatus(podIPs []corev1.PodIP) []string { return ips } -// GetNumericHash generates a raw FNV-1a hash value. -func GetNumericHash(input string) uint32 { - h := fnv.New32a() - _, _ = h.Write([]byte(input)) - return h.Sum32() -} - -// NameHash generates an FNV-1a hash from a string and returns -// it as a fixed-length hexadecimal string. -func NameHash(objectName string) string { - return fmt.Sprintf("%08x", GetNumericHash(objectName)) -} - // hasSystemReservedPrefix reports whether a key uses a label/annotation prefix // reserved for the sandbox system or its extensions. func hasSystemReservedPrefix(key string) bool { diff --git a/extensions/controllers/sandboxclaim_controller.go b/extensions/controllers/sandboxclaim_controller.go index 0b0aa3c..27fb9a7 100644 --- a/extensions/controllers/sandboxclaim_controller.go +++ b/extensions/controllers/sandboxclaim_controller.go @@ -112,6 +112,12 @@ type triggeredAdoptionEntry struct { sandbox string } +// failure is the reason/message pair behind a not-Ready claim. +type failure struct { + reason string + message string +} + // SandboxClaimReconciler reconciles a SandboxClaim object. type SandboxClaimReconciler struct { client.Client @@ -560,12 +566,6 @@ func (r *SandboxClaimReconciler) updateStatus(ctx context.Context, oldStatus *ex return nil } -// failure is the reason/message pair behind a not-Ready claim. -type failure struct { - reason string - message string -} - func (r *SandboxClaimReconciler) computeReadyCondition(claim *extensionsv1beta1.SandboxClaim, sandbox *v1beta1.Sandbox, err error, isClaimExpired bool) metav1.Condition { if err != nil { return notReady(claim, readyFailure(claim, err)) @@ -1694,6 +1694,29 @@ func (h *sandboxEventHandler) Generic(_ context.Context, _ event.GenericEvent, _ // Generic events are not typically used for pod lifecycle changes we care about. } +func (h *sandboxEventHandler) Delete(ctx context.Context, e event.DeleteEvent, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) { + sandbox, ok := e.Object.(*v1beta1.Sandbox) + if !ok { + return + } + + warmPoolName := getWarmPoolName(sandbox) + + if warmPoolName != "" { + key := queue.SandboxKey{ + Namespace: sandbox.Namespace, + Name: sandbox.Name, + } + + namespacedWarmPoolName := queue.GetNamespacedWarmPoolName(sandbox.Namespace, warmPoolName) + + // Actively delete the Ghost Pod from the memory queue + logger := log.FromContext(ctx) + logger.V(1).Info("Removing deleted sandbox from warm pool queue", "namespace", sandbox.Namespace, "sandbox", key) + h.sandboxQueue.RemoveItem(namespacedWarmPoolName, key) + } +} + func verifySandboxCandidate(candidate *v1beta1.Sandbox, claim *extensionsv1beta1.SandboxClaim) error { if candidate.Namespace != claim.Namespace { return fmt.Errorf("%w: sandbox is in %q, claim is in %q", ErrCrossNamespaceAdoption, candidate.Namespace, claim.Namespace) @@ -1731,29 +1754,6 @@ func isAdoptable(candidate *v1beta1.Sandbox) error { return nil } -func (h *sandboxEventHandler) Delete(ctx context.Context, e event.DeleteEvent, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) { - sandbox, ok := e.Object.(*v1beta1.Sandbox) - if !ok { - return - } - - warmPoolName := getWarmPoolName(sandbox) - - if warmPoolName != "" { - key := queue.SandboxKey{ - Namespace: sandbox.Namespace, - Name: sandbox.Name, - } - - namespacedWarmPoolName := queue.GetNamespacedWarmPoolName(sandbox.Namespace, warmPoolName) - - // Actively delete the Ghost Pod from the memory queue - logger := log.FromContext(ctx) - logger.V(1).Info("Removing deleted sandbox from warm pool queue", "namespace", sandbox.Namespace, "sandbox", key) - h.sandboxQueue.RemoveItem(namespacedWarmPoolName, key) - } -} - type warmPoolEventHandler struct { sandboxQueue *queue.SimpleSandboxQueue } diff --git a/pkg/e2bcompat/lifecycle.go b/pkg/e2bcompat/lifecycle.go index 3e297c3..7c25a4a 100644 --- a/pkg/e2bcompat/lifecycle.go +++ b/pkg/e2bcompat/lifecycle.go @@ -306,17 +306,6 @@ func (s *Server) nodesWithSandboxes(r *http.Request) ([]string, error) { return s.opts.Inventory.ListNodes(r.Context()) } -// decodeOptional decodes a JSON body that the schema allows to be absent. An -// empty body leaves the target at its zero value rather than failing, which is -// what pause and fork require. -func decodeOptional(w http.ResponseWriter, r *http.Request, out any) error { - err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxBodyBytes)).Decode(out) - if errors.Is(err, io.EOF) { - return nil - } - return err -} - // isPaused reports whether the sandbox is hibernated, asking its owning node // rather than trusting the synthesized read view. // @@ -339,6 +328,17 @@ func (s *Server) isPaused(ctx context.Context, sb *sandboxv1beta1.Sandbox) bool return sb.Labels[scale.PhaseLabel] == phaseHibernated } +// decodeOptional decodes a JSON body that the schema allows to be absent. An +// empty body leaves the target at its zero value rather than failing, which is +// what pause and fork require. +func decodeOptional(w http.ResponseWriter, r *http.Request, out any) error { + err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxBodyBytes)).Decode(out) + if errors.Is(err, io.EOF) { + return nil + } + return err +} + // claimIDOf reports the node-local claim id the store's verbs address. func claimIDOf(sb *sandboxv1beta1.Sandbox) string { return sb.Annotations[scale.ClaimIDAnnotation] diff --git a/pkg/scale/sandboxd/client.go b/pkg/scale/sandboxd/client.go index 399ba36..182d76c 100644 --- a/pkg/scale/sandboxd/client.go +++ b/pkg/scale/sandboxd/client.go @@ -39,15 +39,6 @@ func (e *HTTPError) Error() string { return fmt.Sprintf("sandboxd: http %d", e.StatusCode) } -// Client talks to a single sandboxd instance. It is safe for concurrent use. -type Client struct { - baseURL string - // token is the node api_token (root or tenant) presented on the claim verb. - // Release authenticates with the sandbox's own token, passed per call. - token string - hc *http.Client -} - // Option configures a Client. type Option func(*Client) @@ -59,20 +50,6 @@ func WithHTTPClient(hc *http.Client) Option { return func(c *Client) { c.hc = hc } } -// New returns a Client for the sandboxd at baseURL, authenticating resource verbs -// with the node api_token (may be empty when sandboxd runs without auth). -func New(baseURL, token string, opts ...Option) *Client { - c := &Client{ - baseURL: strings.TrimRight(baseURL, "/"), - token: token, - hc: &http.Client{}, - } - for _, o := range opts { - o(c) - } - return c -} - // ClaimSpec is the POST /v1/claim body. Net defaults to "none" and Size to // "small" server-side; TTLSeconds 0 means the server default. type ClaimSpec struct { @@ -103,6 +80,58 @@ type ClaimResult struct { Redirect []string `json:"redirect,omitempty"` } +// PoolSpec is one entry of the PUT /v1/pools body: the desired warm watermark +// for a single (template, net, size) pool on this node. It mirrors the claim +// key so a SandboxWarmPool's target lands on the exact pool a Create claims from. +type PoolSpec struct { + Template string `json:"template"` + Net string `json:"net,omitempty"` + Size string `json:"size,omitempty"` + Warm int `json:"warm"` +} + +// NodePool is one pool's live state in a NodeInfo. Only the fields the +// warm-pool driver reports on are decoded. +type NodePool struct { + Key PoolKey `json:"key"` + Warm int `json:"warm"` + Refilling int `json:"refilling"` + Target int `json:"target"` + Golden bool `json:"golden"` +} + +// NodeInfo is the PUT /v1/pools (and GET /v1/info) response: the node's live +// per-pool warm state plus its lifecycle counters. +type NodeInfo struct { + Pools []NodePool `json:"pools"` + Claimed int `json:"claimed"` + Hibernated int `json:"hibernated"` + Archived int `json:"archived"` +} + +// Client talks to a single sandboxd instance. It is safe for concurrent use. +type Client struct { + baseURL string + // token is the node api_token (root or tenant) presented on the claim verb. + // Release authenticates with the sandbox's own token, passed per call. + token string + hc *http.Client +} + +// New returns a Client for the sandboxd at baseURL, authenticating resource verbs +// with the node api_token (may be empty when sandboxd runs without auth). +func New(baseURL, token string, opts ...Option) *Client { + c := &Client{ + baseURL: strings.TrimRight(baseURL, "/"), + token: token, + hc: &http.Client{}, + } + for _, o := range opts { + o(c) + } + return c +} + // Claim performs POST /v1/claim, returning the delivered sandbox on success. // A 429, or a 200 that carries only a peer redirect, yields ErrNodeAtCapacity. func (c *Client) Claim(ctx context.Context, spec ClaimSpec) (ClaimResult, error) { @@ -142,35 +171,6 @@ func (c *Client) Claim(ctx context.Context, spec ClaimSpec) (ClaimResult, error) } } -// PoolSpec is one entry of the PUT /v1/pools body: the desired warm watermark -// for a single (template, net, size) pool on this node. It mirrors the claim -// key so a SandboxWarmPool's target lands on the exact pool a Create claims from. -type PoolSpec struct { - Template string `json:"template"` - Net string `json:"net,omitempty"` - Size string `json:"size,omitempty"` - Warm int `json:"warm"` -} - -// NodePool is one pool's live state in a NodeInfo. Only the fields the -// warm-pool driver reports on are decoded. -type NodePool struct { - Key PoolKey `json:"key"` - Warm int `json:"warm"` - Refilling int `json:"refilling"` - Target int `json:"target"` - Golden bool `json:"golden"` -} - -// NodeInfo is the PUT /v1/pools (and GET /v1/info) response: the node's live -// per-pool warm state plus its lifecycle counters. -type NodeInfo struct { - Pools []NodePool `json:"pools"` - Claimed int `json:"claimed"` - Hibernated int `json:"hibernated"` - Archived int `json:"archived"` -} - // SetPools performs PUT /v1/pools, replacing this node's desired warm targets // with the supplied set (an omitted pool is drained). It authenticates with the // node api_token. The whole set is sent in one request because sandboxd replaces diff --git a/pkg/scale/sandboxstore_impl.go b/pkg/scale/sandboxstore_impl.go index bba0a95..99cd61f 100644 --- a/pkg/scale/sandboxstore_impl.go +++ b/pkg/scale/sandboxstore_impl.go @@ -228,29 +228,6 @@ func (s *scatterGatherStore) List(ctx context.Context, opts ListOptions) (*sandb return list, nil } -// fanOutNodes enumerates the node inventories and runs work per node with -// List's bounded concurrency, concatenating per-node results in node order. -// A node the work skips contributes nil. -func fanOutNodes[T any](ctx context.Context, s *scatterGatherStore, work func(ctx context.Context, node string) []T) ([]T, error) { - nodes, err := s.src.ListNodes(ctx) - if err != nil { - return nil, fmt.Errorf("scale: enumerate node inventories: %w", err) - } - perNode := make([][]T, len(nodes)) - g, gctx := errgroup.WithContext(ctx) - if s.concurrency > 0 { - g.SetLimit(s.concurrency) - } - for i, node := range nodes { - g.Go(func() error { - perNode[i] = work(gctx, node) - return nil - }) - } - _ = g.Wait() - return slices.Concat(perNode...), nil -} - // Get routes to the owning node's authoritative inventory. It resolves which // node holds namespace/name and returns that entry synthesized as a Sandbox. // The per-node sweep fans out like List and cancels on the first hit — a @@ -557,121 +534,6 @@ func pickPowerOfTwo(candidates []warmCandidate) (warmCandidate, int) { return candidates[i], i } -// poolCapacityMatches reports whether a node's advertised pool capacity serves the -// requested pool key, normalizing the net/size defaults on both sides so an unset -// axis matches its default-named pool. -func poolCapacityMatches(pc PoolCapacity, key PoolKey) bool { - return pc.Template == key.Template && - cmp.Or(pc.Net, NetDefault) == cmp.Or(key.Net, NetDefault) && - cmp.Or(pc.Size, SizeClassSmall) == cmp.Or(key.Size, SizeClassSmall) -} - -// parseSelectors turns the string selectors on ListOptions into matchers. Empty -// strings become everything-matchers. -func parseSelectors(opts ListOptions) (labels.Selector, fields.Selector, error) { - labelSel, err := labels.Parse(opts.LabelSelector) - if err != nil { - return nil, nil, fmt.Errorf("scale: parse label selector %q: %w", opts.LabelSelector, err) - } - fieldSel, err := fields.ParseSelector(opts.FieldSelector) - if err != nil { - return nil, nil, fmt.Errorf("scale: parse field selector %q: %w", opts.FieldSelector, err) - } - return labelSel, fieldSel, nil -} - -// entryToSandbox synthesizes the Sandbox object served for one inventory entry. -// The entry name is the sandbox's "/"; an unqualified name -// lands in the default namespace. -func entryToSandbox(node string, e InventoryEntry) *sandboxv1beta1.Sandbox { - ns, name := splitNamespacedName(e.Name) - sb := &sandboxv1beta1.Sandbox{ - ObjectMeta: metav1.ObjectMeta{ - Namespace: ns, - Name: name, - Labels: synthLabels(node, e), - Annotations: synthAnnotations(e), - ResourceVersion: resourceVersionFor(ns, name, e), - }, - Status: sandboxv1beta1.SandboxStatus{ - NodeName: node, - PodIPs: AddressIPs(e.Address), - Conditions: []metav1.Condition{{ - Type: string(sandboxv1beta1.SandboxConditionReady), - Status: readyStatus(e.Phase), - Reason: readyReason(e.Phase), - Message: fmt.Sprintf("phase %q reported by node %q inventory", e.Phase, node), - }}, - }, - } - return sb -} - -func synthLabels(node string, e InventoryEntry) map[string]string { - l := map[string]string{NodeLabel: node} - if e.Phase != "" { - l[PhaseLabel] = e.Phase - } - if e.ClaimRef != "" { - _, claim := splitNamespacedName(e.ClaimRef) - if claim != "" { - l[ClaimLabel] = claim - } - } - return l -} - -// synthAnnotations carries the opaque node-local handles a synthesized Sandbox -// needs but that are not selector axes — the sandboxd claim id, which Delete -// uses to release the right microVM. Nil when the node has not published an id -// yet, so Delete refuses to release rather than guessing by name. -func synthAnnotations(e InventoryEntry) map[string]string { - if e.ID == "" { - return nil - } - return map[string]string{ClaimIDAnnotation: e.ID} -} - -func sandboxFields(sb *sandboxv1beta1.Sandbox) fields.Set { - return fields.Set{ - "metadata.name": sb.Name, - "metadata.namespace": sb.Namespace, - "status.nodeName": sb.Status.NodeName, - } -} - -func readyStatus(phase string) metav1.ConditionStatus { - if strings.EqualFold(phase, "Running") || strings.EqualFold(phase, "Ready") { - return metav1.ConditionTrue - } - return metav1.ConditionFalse -} - -func readyReason(phase string) string { - if phase == "" { - return "Unknown" - } - return phase -} - -// resourceVersionFor derives a deterministic, content-sensitive ResourceVersion -// so watch can detect a Modified entry and clients see a stable version for an -// unchanged one. It is opaque, as the API contract requires. -func resourceVersionFor(ns, name string, e InventoryEntry) string { - h := fnv.New64a() - _, _ = h.Write([]byte(ns + "/" + name + "|" + e.ID + "|" + e.Phase + "|" + e.ClaimRef + "|" + e.Address)) - return strconv.FormatUint(h.Sum64(), 10) -} - -func splitNamespacedName(s string) (namespace, name string) { - if before, after, ok := strings.Cut(s, "/"); ok { - return before, after - } - return metav1.NamespaceDefault, s -} - -func objKey(sb *sandboxv1beta1.Sandbox) string { return sb.Namespace + "/" + sb.Name } - // inventoryWatcher is a watch.Interface fed by the store's poll-diff goroutine. type inventoryWatcher struct { result chan watch.Event @@ -900,3 +762,141 @@ func (s *ClientInventorySource) NodeInventory(ctx context.Context, node string) } return inv, nil } + +// fanOutNodes enumerates the node inventories and runs work per node with +// List's bounded concurrency, concatenating per-node results in node order. +// A node the work skips contributes nil. +func fanOutNodes[T any](ctx context.Context, s *scatterGatherStore, work func(ctx context.Context, node string) []T) ([]T, error) { + nodes, err := s.src.ListNodes(ctx) + if err != nil { + return nil, fmt.Errorf("scale: enumerate node inventories: %w", err) + } + perNode := make([][]T, len(nodes)) + g, gctx := errgroup.WithContext(ctx) + if s.concurrency > 0 { + g.SetLimit(s.concurrency) + } + for i, node := range nodes { + g.Go(func() error { + perNode[i] = work(gctx, node) + return nil + }) + } + _ = g.Wait() + return slices.Concat(perNode...), nil +} + +// poolCapacityMatches reports whether a node's advertised pool capacity serves the +// requested pool key, normalizing the net/size defaults on both sides so an unset +// axis matches its default-named pool. +func poolCapacityMatches(pc PoolCapacity, key PoolKey) bool { + return pc.Template == key.Template && + cmp.Or(pc.Net, NetDefault) == cmp.Or(key.Net, NetDefault) && + cmp.Or(pc.Size, SizeClassSmall) == cmp.Or(key.Size, SizeClassSmall) +} + +// parseSelectors turns the string selectors on ListOptions into matchers. Empty +// strings become everything-matchers. +func parseSelectors(opts ListOptions) (labels.Selector, fields.Selector, error) { + labelSel, err := labels.Parse(opts.LabelSelector) + if err != nil { + return nil, nil, fmt.Errorf("scale: parse label selector %q: %w", opts.LabelSelector, err) + } + fieldSel, err := fields.ParseSelector(opts.FieldSelector) + if err != nil { + return nil, nil, fmt.Errorf("scale: parse field selector %q: %w", opts.FieldSelector, err) + } + return labelSel, fieldSel, nil +} + +// entryToSandbox synthesizes the Sandbox object served for one inventory entry. +// The entry name is the sandbox's "/"; an unqualified name +// lands in the default namespace. +func entryToSandbox(node string, e InventoryEntry) *sandboxv1beta1.Sandbox { + ns, name := splitNamespacedName(e.Name) + sb := &sandboxv1beta1.Sandbox{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: ns, + Name: name, + Labels: synthLabels(node, e), + Annotations: synthAnnotations(e), + ResourceVersion: resourceVersionFor(ns, name, e), + }, + Status: sandboxv1beta1.SandboxStatus{ + NodeName: node, + PodIPs: AddressIPs(e.Address), + Conditions: []metav1.Condition{{ + Type: string(sandboxv1beta1.SandboxConditionReady), + Status: readyStatus(e.Phase), + Reason: readyReason(e.Phase), + Message: fmt.Sprintf("phase %q reported by node %q inventory", e.Phase, node), + }}, + }, + } + return sb +} + +func synthLabels(node string, e InventoryEntry) map[string]string { + l := map[string]string{NodeLabel: node} + if e.Phase != "" { + l[PhaseLabel] = e.Phase + } + if e.ClaimRef != "" { + _, claim := splitNamespacedName(e.ClaimRef) + if claim != "" { + l[ClaimLabel] = claim + } + } + return l +} + +// synthAnnotations carries the opaque node-local handles a synthesized Sandbox +// needs but that are not selector axes — the sandboxd claim id, which Delete +// uses to release the right microVM. Nil when the node has not published an id +// yet, so Delete refuses to release rather than guessing by name. +func synthAnnotations(e InventoryEntry) map[string]string { + if e.ID == "" { + return nil + } + return map[string]string{ClaimIDAnnotation: e.ID} +} + +func sandboxFields(sb *sandboxv1beta1.Sandbox) fields.Set { + return fields.Set{ + "metadata.name": sb.Name, + "metadata.namespace": sb.Namespace, + "status.nodeName": sb.Status.NodeName, + } +} + +func readyStatus(phase string) metav1.ConditionStatus { + if strings.EqualFold(phase, "Running") || strings.EqualFold(phase, "Ready") { + return metav1.ConditionTrue + } + return metav1.ConditionFalse +} + +func readyReason(phase string) string { + if phase == "" { + return "Unknown" + } + return phase +} + +// resourceVersionFor derives a deterministic, content-sensitive ResourceVersion +// so watch can detect a Modified entry and clients see a stable version for an +// unchanged one. It is opaque, as the API contract requires. +func resourceVersionFor(ns, name string, e InventoryEntry) string { + h := fnv.New64a() + _, _ = h.Write([]byte(ns + "/" + name + "|" + e.ID + "|" + e.Phase + "|" + e.ClaimRef + "|" + e.Address)) + return strconv.FormatUint(h.Sum64(), 10) +} + +func splitNamespacedName(s string) (namespace, name string) { + if before, after, ok := strings.Cut(s, "/"); ok { + return before, after + } + return metav1.NamespaceDefault, s +} + +func objKey(sb *sandboxv1beta1.Sandbox) string { return sb.Namespace + "/" + sb.Name }