diff --git a/cmd/gantry/agent_metrics.go b/cmd/gantry/agent_metrics.go index 91ea0425c..4aa36da96 100644 --- a/cmd/gantry/agent_metrics.go +++ b/cmd/gantry/agent_metrics.go @@ -389,7 +389,7 @@ func newPhase3Metrics(reg *metrics.Registry, infl *inflight.Map) *phase3Metrics }, []string{"digest_kind", "outcome"}), coldStartSeedContacted: reg.NewHistogramVec("coord", prometheus.HistogramOpts{ Name: "p2p_cold_start_seed_chairs_contacted", - Help: "Chairs contacted per Resolve before SeedCount accepted the digest. Equal to SeedCount when the top-8 accept; larger means declines pushed the requester deeper into the ranking, which is what makes more than SeedCount nodes fetch the same layer from origin.", + Help: "Chairs contacted per Resolve before the active cohort accepted the digest. Larger than the selectable cohort means declines pushed the requester deeper into the ranking and added origin fetchers.", Buckets: prometheus.LinearBuckets(4, 4, 17), }, []string{"digest_kind"}), coldStartSeedSelectable: reg.NewHistogramVec("coord", prometheus.HistogramOpts{ @@ -399,7 +399,7 @@ func newPhase3Metrics(reg *metrics.Registry, infl *inflight.Map) *phase3Metrics }, []string{"digest_kind"}), coldStartSeedAccepted: reg.NewHistogramVec("coord", prometheus.HistogramOpts{ Name: "p2p_cold_start_seed_chairs_accepted", - Help: "Chairs that accepted the digest per Resolve. Fewer than SeedCount is normal and harmless: the accepted chairs are already fetching, so the resolver no longer walks down the ranking to top the cohort back up.", + Help: "Chairs that accepted the digest per Resolve. Fewer than the active cohort is normal: accepted chairs are already fetching, so the resolver does not backfill silent chairs.", Buckets: prometheus.LinearBuckets(1, 1, 16), }, []string{"digest_kind"}), coldStartChairDispatch: reg.NewCounterVec("coord", prometheus.CounterOpts{ diff --git a/cmd/gantry/chair_capacity.go b/cmd/gantry/chair_capacity.go new file mode 100644 index 000000000..c674d5aa7 --- /dev/null +++ b/cmd/gantry/chair_capacity.go @@ -0,0 +1,50 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "fmt" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + appsv1 "k8s.io/client-go/kubernetes/typed/apps/v1" +) + +func proportionalChairSeedTarget(capacity, percentage, maximum int) int { + target := (capacity*percentage + 99) / 100 + if target < 1 { + target = 1 + } + + if target > maximum { + target = maximum + } + + return target +} + +type daemonSetChairCapacity struct { + daemonSets appsv1.DaemonSetInterface + name string + percentage int + maximum int +} + +func (c daemonSetChairCapacity) SeedTarget(ctx context.Context) (int, error) { + daemonSet, err := c.daemonSets.Get(ctx, c.name, metav1.GetOptions{}) + if err != nil { + return 0, fmt.Errorf("get Gantry DaemonSet capacity: %w", err) + } + + capacity := int(daemonSet.Status.DesiredNumberScheduled) + if capacity < 1 { + return 0, fmt.Errorf("gantry daemonset desired capacity is %d", capacity) + } + + return proportionalChairSeedTarget(capacity, c.percentage, c.maximum), nil +} + +func chairDHTReady(clusterSizeEstimate, routingTableSize int, holdingChair bool) bool { + return clusterSizeEstimate <= 1 || routingTableSize > 0 || holdingChair +} diff --git a/cmd/gantry/chair_capacity_test.go b/cmd/gantry/chair_capacity_test.go new file mode 100644 index 000000000..98bbe3249 --- /dev/null +++ b/cmd/gantry/chair_capacity_test.go @@ -0,0 +1,98 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "testing" + + appsv1 "k8s.io/api/apps/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +func TestProportionalChairSeedTarget(t *testing.T) { + tests := []struct { + name string + capacity int + percentage int + maximum int + want int + }{ + {name: "two nodes", capacity: 2, percentage: 10, maximum: 50, want: 1}, + {name: "three nodes", capacity: 3, percentage: 10, maximum: 50, want: 1}, + {name: "twenty nodes", capacity: 20, percentage: 10, maximum: 50, want: 2}, + {name: "round up", capacity: 21, percentage: 10, maximum: 50, want: 3}, + {name: "maximum", capacity: 1000, percentage: 10, maximum: 50, want: 50}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := proportionalChairSeedTarget(test.capacity, test.percentage, test.maximum); got != test.want { + t.Fatalf("target = %d; want %d", got, test.want) + } + }) + } +} + +func TestDaemonSetChairCapacitySeedTarget(t *testing.T) { + client := fake.NewSimpleClientset(&appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{Name: "gantry", Namespace: "gantry-system"}, + Status: appsv1.DaemonSetStatus{DesiredNumberScheduled: 21}, + }) + source := daemonSetChairCapacity{ + daemonSets: client.AppsV1().DaemonSets("gantry-system"), + name: "gantry", + percentage: 10, + maximum: 50, + } + + target, err := source.SeedTarget(context.Background()) + if err != nil { + t.Fatalf("SeedTarget: %v", err) + } + + if target != 3 { + t.Fatalf("target = %d; want 3", target) + } +} + +func TestDaemonSetChairCapacityRejectsZeroDesiredCapacity(t *testing.T) { + client := fake.NewSimpleClientset(&appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{Name: "gantry", Namespace: "gantry-system"}, + }) + source := daemonSetChairCapacity{ + daemonSets: client.AppsV1().DaemonSets("gantry-system"), + name: "gantry", + percentage: 10, + maximum: 50, + } + + if _, err := source.SeedTarget(context.Background()); err == nil { + t.Fatal("SeedTarget succeeded with zero desired capacity") + } +} + +func TestChairDHTReady(t *testing.T) { + tests := []struct { + name string + clusterEstimate int + routingTableSize int + holdingChair bool + want bool + }{ + {name: "connected non-chair", clusterEstimate: 100, routingTableSize: 1, want: true}, + {name: "isolated chair", clusterEstimate: 100, holdingChair: true, want: true}, + {name: "isolated non-chair", clusterEstimate: 100, want: false}, + {name: "single node", clusterEstimate: 1, want: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := chairDHTReady(test.clusterEstimate, test.routingTableSize, test.holdingChair); got != test.want { + t.Fatalf("ready = %t; want %t", got, test.want) + } + }) + } +} diff --git a/cmd/gantry/main.go b/cmd/gantry/main.go index d57f5ac79..6bc8198ad 100644 --- a/cmd/gantry/main.go +++ b/cmd/gantry/main.go @@ -249,6 +249,9 @@ func runAgent(args []string) error { slog.Bool("chairs_active", c.ChairNamespace != ""), slog.String("chair_namespace", c.ChairNamespace), slog.String("chair_listen", c.ChairListen), + slog.String("chair_capacity_daemonset", c.ChairCapacityDaemonSet), + slog.Int("chair_seed_percentage", c.ChairSeedPercentage), + slog.Int("chair_seed_maximum", c.ChairSeedCount), ) const kademliaMaxRoutingTable = 256 @@ -316,7 +319,14 @@ func runAgent(args []string) error { ) var chairManager *chairs.Manager + if chairStore != nil { + capacity := daemonSetChairCapacity{ + daemonSets: chairClient.AppsV1().DaemonSets(c.ChairNamespace), + name: c.ChairCapacityDaemonSet, + percentage: c.ChairSeedPercentage, + maximum: c.ChairSeedCount, + } chairManager = chairs.NewManager(chairs.ManagerOptions{ Store: chairStore, Cache: chairCache, @@ -327,6 +337,7 @@ func runAgent(args []string) error { return disco.ConnectPeers(connectCtx, addresses) }, BootstrapHealthy: func() bool { return disco.RoutingTableSize() > 0 }, + SeedTarget: capacity.SeedTarget, Logger: logger, LeaseDuration: c.ChairLeaseDuration, RenewPeriod: c.ChairRenewPeriod, @@ -467,8 +478,8 @@ func runAgent(args []string) error { coldStartResolver = coldStartAdapter{r: realResolver} layerPrefetcher = newLayerPrefetcher(realResolver, cstore, logger, layerProgress.observeManifest) logger.Info("Lease-chair cold-start orchestrator wired", - slog.Int("chairs", chairs.Count), - slog.Int("seeds", c.ChairSeedCount), + slog.Int("chair_slots", chairs.Count), + slog.Int("seed_maximum", c.ChairSeedCount), ) } else { logger.Info("Lease-chair cold-start orchestrator disabled (no Kubernetes namespace configured)") @@ -801,7 +812,7 @@ func runAgent(args []string) error { } if chairManager != nil && !chairManager.Ready() { - return fmt.Sprintf("no Lease chair held and fewer than %d are selectable", c.ChairSeedCount), false + return "no selectable Lease chair", false } if checkDialable && noDialableP2PAddrs { @@ -827,7 +838,12 @@ func runAgent(args []string) error { return "transfer listener family mismatches Pod IP; check transfer_listen vs Pod IP family", false } - if chairManager != nil && c.ChairClusterSizeEstimate > 1 && disco.RoutingTableSize() < 1 { + holdingChair := false + if chairManager != nil { + _, holdingChair = chairManager.Held() + } + + if chairManager != nil && !chairDHTReady(c.ChairClusterSizeEstimate, disco.RoutingTableSize(), holdingChair) { return "dht routing table empty", false } diff --git a/deploy/gantry/configmap.yaml.tmpl b/deploy/gantry/configmap.yaml.tmpl index 6bef5b12d..d9e98122e 100644 --- a/deploy/gantry/configmap.yaml.tmpl +++ b/deploy/gantry/configmap.yaml.tmpl @@ -80,6 +80,10 @@ data: chair_claim_jitter: "750ms" chair_claim_initial_divisor: 2048 chair_cluster_size_estimate: 100000 + # Keep 10% of the Gantry DaemonSet's desired capacity as chairs, capped at + # 50. Small clusters always retain at least one chair. + chair_capacity_daemonset: gantry + chair_seed_percentage: 10 chair_seed_count: 50 chair_api_timeout: "5s" diff --git a/deploy/gantry/render_test.go b/deploy/gantry/render_test.go index f5b539b4d..69ce2dff8 100644 --- a/deploy/gantry/render_test.go +++ b/deploy/gantry/render_test.go @@ -218,7 +218,8 @@ func TestChairRBACAllowsLeaseRecovery(t *testing.T) { } decoder := yaml.NewDecoder(bytes.NewReader(raw)) - found := false + foundLeaseRule := false + foundDaemonSetRule := false for { var object struct { @@ -249,14 +250,26 @@ func TestChairRBACAllowsLeaseRecovery(t *testing.T) { } } - found = true + foundLeaseRule = true + } + + if containsString(rule.APIGroups, "apps") && containsString(rule.Resources, "daemonsets") { + if !containsString(rule.Verbs, "get") { + t.Fatalf("DaemonSet RBAC verbs = %v, missing get", rule.Verbs) + } + + foundDaemonSetRule = true } } } - if !found { + if !foundLeaseRule { t.Fatal("no coordination Lease RBAC rule rendered") } + + if !foundDaemonSetRule { + t.Fatal("no apps DaemonSet RBAC rule rendered") + } } func containsString(values []string, target string) bool { diff --git a/deploy/gantry/serviceaccount.yaml.tmpl b/deploy/gantry/serviceaccount.yaml.tmpl index 8b58a40d9..736d7dbb5 100644 --- a/deploy/gantry/serviceaccount.yaml.tmpl +++ b/deploy/gantry/serviceaccount.yaml.tmpl @@ -1,7 +1,7 @@ # ServiceAccount + RBAC for the gantry agent. # -# The agent's only Kubernetes access is the 64 fixed Lease chairs. It opens no -# watches and reads no Pods or Nodes. +# The agent manages fixed Lease chairs and reads its DaemonSet's desired +# capacity. It opens no watches and reads no Pods or Nodes. --- apiVersion: v1 kind: Namespace @@ -29,6 +29,9 @@ rules: - apiGroups: ["coordination.k8s.io"] resources: ["leases"] verbs: ["get", "list", "create", "update"] + - apiGroups: ["apps"] + resources: ["daemonsets"] + verbs: ["get"] --- apiVersion: rbac.authorization.k8s.io/v1 kind: RoleBinding diff --git a/designs/gantry-cold-pull-explained.md b/designs/gantry-cold-pull-explained.md index 000aff0b6..bae1bf9d0 100644 --- a/designs/gantry-cold-pull-explained.md +++ b/designs/gantry-cold-pull-explained.md @@ -93,8 +93,8 @@ flowchart LR A chair is not a coordinator and does not decide anything for anyone else. It is simply a node that has agreed to be one of the designated registry fetchers. -Being a chair is a small extra duty: 64 of 1,000 nodes, and it never grows with -cluster size. +Being a chair is a small extra duty. Gantry selects 10% of its DaemonSet's +desired capacity, rounded up, with a minimum of one and a maximum of 50 chairs. ### Choosing which chairs seed which layer @@ -106,7 +106,7 @@ digest against the 64 chair names, producing a ranking. flowchart LR D["layer digest
sha256:abc..."] --> H["hash against
64 chair names"] H --> RK["ranked list
chair-41, chair-07, chair-29, ..."] - RK --> T["top 8 = seeds for this layer"] + RK --> T["active chairs = seeds for this layer"] style T fill:#cfe6ff,stroke:#3b7dd8 ``` @@ -115,8 +115,8 @@ Because the input is just the digest and the chair names, every node computes the **same** ranking independently. No election, no coordination, no chatter. A different layer hashes to a different ranking, so the 40 layers of an image -spread their seeding work across the chairs rather than piling onto the same -eight nodes. +spread their seeding work across the active chairs rather than piling onto the +same nodes. ### The cold pull, end to end @@ -125,7 +125,7 @@ sequenceDiagram participant CD as containerd participant G as Gantry (this node) participant IX as index - participant CH as top 8 chairs + participant CH as active chairs participant REG as registry CD->>G: get layer sha256:abc @@ -134,11 +134,11 @@ sequenceDiagram G->>G: rank 64 chairs for this digest G->>CH: please pull sha256:abc Note over CH: each chair checks:
am I already pulling this? - CH->>REG: fetch layer (8 copies total) + CH->>REG: fetch layer (one copy per active chair) REG-->>CH: layer bytes CH->>IX: publish "I have sha256:abc" G->>IX: who has sha256:abc? - IX-->>G: 8 chairs + IX-->>G: available providers G->>CH: fetch from a chair CH-->>G: layer bytes G-->>CD: layer bytes @@ -149,12 +149,20 @@ progress. It waits on the **index**: once any chair finishes and publishes, the normal peer path takes over and the layer spreads outward from the seeds. Meanwhile the other 999 nodes are doing the same thing. They all compute the -same top 8, so they all ask the same chairs, and each chair recognizes the +same active cohort, so they all ask the same chairs, and each chair recognizes the duplicate requests and pulls once. -### Why exactly 8 +### How many seeds -Eight is a deliberate trade between registry cost and resilience. +The target balances registry cost and resilience: + +$$ + ext{chairs}=\min\left(50,\max\left(1,\left\lceil0.10\times\text{desired Gantry pods}\right\rceil\right)\right) +$$ + +For example, 3 desired pods select 1 chair, 20 select 2, and 500 or more select +the maximum of 50. The table below records the earlier eight-seed benchmark, +not the current proportional default. | Seeds | Registry traffic | Risk | |---|---|---| @@ -162,9 +170,8 @@ Eight is a deliberate trade between registry cost and resilience. | **8** | **343.6 GB** | tolerates several slow or dead seeds | | 1,000 | 42 TB | no sharing at all | -Eight copies of a 40 GiB image is 343.6 GB, and that number does not change if -the cluster grows to 100,000 nodes. Only the peer-to-peer traffic grows, and -that is traffic the registry never sees. +Eight copies of a 40 GiB image is 343.6 GB. Under the proportional policy, +registry copies grow with Gantry capacity only until the 50-chair maximum. ## Failure handling @@ -183,22 +190,22 @@ flowchart TB direction LR B1["chair 5 slow to answer"] --> B2["recruit chair 9"] B2 --> B3["chair 5 IS pulling
chair 9 now pulling too"] - B3 --> B4["9 copies, not 8"] + B3 --> B4["one extra registry copy"] end subgraph good["keeping the cohort (correct)"] direction LR G1["chair 5 slow to answer"] --> G2["do nothing"] G2 --> G3["chair 5 finishes and publishes"] - G3 --> G4["8 copies"] + G3 --> G4["target unchanged"] end style bad fill:#ffe9e9,stroke:#d86b6b style good fill:#e9f7e9,stroke:#5c9c5c ``` -So Gantry contacts the top eight once and accepts partial answers. It only moves -further down the ranking when the entire cohort of eight answers nothing at all, +So Gantry contacts the active cohort once and accepts partial answers. It only +moves further down the ranking when the entire cohort answers nothing at all, which means they are genuinely unreachable rather than merely busy. Measured on 1,000 nodes, this single distinction is the difference between 13.4 @@ -326,7 +333,7 @@ also the cleanest way to confirm the watches are really gone. ## What this looks like in practice Measured across three runs of 1,000 nodes pulling a cold 40 GiB image -simultaneously: +simultaneously with the earlier eight-seed configuration: | | | |---|---| @@ -335,7 +342,7 @@ simultaneously: | Peer-to-peer traffic | 42.6 TB | | Bytes containerd received from the registry directly | **0** | -The design floor is 8 copies, or 343.6 GB. Two runs sat exactly on it and the +That configuration's floor is 8 copies, or 343.6 GB. Two runs sat exactly on it and the third was a few extra seed fetches above. In the most recent run, 99.3% of the bytes containerd received came from peers @@ -350,7 +357,7 @@ flowchart TB B -->|no| C{"does the index
know a peer?"} C -->|yes| D["fetch from peer"] C -->|no| E["hash digest to rank 64 chairs"] - E --> F["ask the top 8 to fetch from the registry"] + E --> F["ask the active chairs to fetch from the registry"] F --> G["chairs publish to the index"] G --> D D --> Z @@ -359,5 +366,6 @@ flowchart TB style D fill:#e9f7e9,stroke:#5c9c5c ``` -The registry path is the narrow red box, entered once per layer by eight nodes. -Everything else is the green box, and that is where the terabytes go. +The registry path is the narrow red box, entered once per layer by each active +chair, up to 50. Everything else is the green box, and that is where the +terabytes go. diff --git a/designs/gantry-cold-pull-rotating-chair.md b/designs/gantry-cold-pull-rotating-chair.md index 045a484b7..8d67019d0 100644 --- a/designs/gantry-cold-pull-rotating-chair.md +++ b/designs/gantry-cold-pull-rotating-chair.md @@ -4,7 +4,9 @@ New Lease based cold start design with deterministic puller selection and doesnt **Scope** - Target: `100,000` Gantry nodes. -- `64` fixed Kubernetes Lease names are stable chairs. +- `64` fixed Kubernetes Lease names are available chair slots. +- The active chair target is 10% of the Gantry DaemonSet's desired capacity, + rounded up, with a minimum of one and a maximum of 50. - Each chair has one current Gantry holder. - A holder is a final origin seed puller, not a coordinator that performs another DHT selection. - Each node may hold at most one chair. @@ -13,9 +15,9 @@ New Lease based cold start design with deterministic puller selection and doesnt **Per-Digest Selection** ```text -ranking = HRW(blob digest, 64 stable chair IDs) -top 8 = primary seed chairs -rest = ordered backup chairs +ranking = HRW(blob digest, active stable chair IDs) +active cohort = primary seed chairs, up to 50 +rest = ordered backup chairs ``` - Every requester using the same snapshot gets the same ordering. @@ -60,6 +62,9 @@ use cachedSnapshot - Concurrent pulls share one refresh. - Gantry agents do not watch Leases. The operator watches chair deletions only so the fixed objects are recreated; heartbeat updates do not reconcile. +- Eligible claimers read `DaemonSet.status.desiredNumberScheduled` to compute + the target. Holders refresh it during renewal; no Pod or Node reads or + watches are introduced. - No per-digest Lease reads. - Nodes without pulls perform no snapshot refresh. - A dial failure triggers an immediate targeted chair refresh, even if the global epoch is unchanged. @@ -75,16 +80,16 @@ When all 64 Leases are empty: 4. Each eligible node chooses one chair and may claim only that chair. 5. Claim uses a Lease `resourceVersion` update. 6. One node wins each chair. -7. Eligibility widens in later rounds until enough chairs are occupied. -8. Cold selection becomes available once at least eight chairs exist. +7. Eligibility widens in later rounds until the proportional target is occupied. +8. Cold selection becomes available once one selectable chair exists. **Cold Pull Path** 1. Check local content. 2. Query DHT for existing providers. 3. If a provider exists, peer-fetch from it. -4. If genuinely cold, rank the 64 chairs for that digest. -5. Contact the holders of the top eight chairs. +4. If genuinely cold, rank the active chairs for that digest. +5. Contact the ranked active cohort, up to the configured maximum. 6. Group multiple image children assigned to the same holder into one `please_pull` RPC. 7. Each holder validates the chair epoch and generation. 8. Each holder checks local content and its local in-flight map. @@ -93,23 +98,24 @@ When all 64 Leases are empty: 11. Completed content is committed and advertised through DHT. 12. Other nodes discover providers and peer-fetch normally. -Nominal cold-origin seeding is eight copies per digest. +Nominal cold-origin seeding is one copy per active chair, bounded by 50. The cohort is contacted in one pass and partial acceptance is sufficient. The -requester moves to the next eight chairs only when the entire cohort accepts -nothing; any acceptance means the pull is under way and the requester waits on -DHT for it. Recruiting until eight acceptances are collected would count a +requester moves to the next cohort only when the entire cohort accepts nothing; +any acceptance means the pull is under way and the requester waits on DHT for +it. Recruiting until a full set of acceptances is collected would count a silent chair as absent when it is in fact already fetching, so each timeout would add a seed instead of moving one. Measured on 1,000 nodes, that -distinction is the difference between 13.4 and 8.0 origin copies per layer. +distinction was the difference between 13.4 and 8.0 origin copies per layer in +the earlier eight-seed configuration. While the requester waits, it re-queries the accepting cohort. A chair that reports the pull is already under way is making progress, however long it takes: a large layer or a queued job routinely outlives several poll windows, -and escalating there would put eight more nodes on the origin for work that is +and escalating there would put another cohort on the origin for work that is already running. The requester therefore keeps waiting, bounded by the calling request, for as long as any chair reports work in flight. It escalates to the -next eight chairs only once no chair does, which covers a cohort that has gone +next cohort only once no chair does, which covers a cohort that has gone silent or whose accepted work ended without publishing. **Please-Pull Transport** @@ -184,20 +190,18 @@ no chair looks free and nothing is ever claimed. Startup therefore treats a chair unrenewed for five lease durations as abandoned and claimable, but only once no genuinely free chair remains, so a briefly slow holder keeps its seat. -Readiness reflects whether this agent can seed, not whether the cluster has a -full cohort. Requiring eight occupied chairs is unsatisfiable while fewer than -eight nodes run the new build, which would stall the first batch of a rolling -upgrade and any cluster smaller than the seed count. Holding a chair is -therefore sufficient on its own. +Readiness requires one selectable chair, not the complete proportional target. +The target continues converging after agents become ready, so small clusters +and rolling upgrades do not require every node to become a chair. **Accepted Limitations** -- Timeout ambiguity can temporarily activate more than eight seeds, though a +- Timeout ambiguity can temporarily activate extra seeds, though a silent chair no longer causes a substitute to be recruited. - Network partitions can temporarily produce old/new-holder overlap. - Direct duplicate `please_pull` traffic still requires 100,000-node measurement. - Snapshot refresh bursts require jitter and API-scale validation. -- Eight-seed dissemination performance at 100,000 nodes is unmeasured. +- Proportional-seed dissemination performance at 100,000 nodes is unmeasured. - Chair selection currently lacks zone-awareness. - Content integrity remains protected by digest verification. - The libp2p connection watermark bounds the DHT's connection appetite, which is @@ -216,9 +220,11 @@ therefore sufficient on its own. - Startup snapshot jitter: deterministic over `30s`. - Empty-chair claim rounds run every `1s` with up to `750ms` deterministic per-claim jitter. +- Chair target: 10% of `DaemonSet.status.desiredNumberScheduled`, rounded up, + with a minimum of one and a configurable maximum of 50. - The initial claim lottery admits `1/2048` of peers. The divisor halves each stage, using one stable peer/epoch ticket so eligibility only widens until - every node is eligible if fewer than eight chairs have been occupied. + every node is eligible if the proportional target has not been occupied. - Widening occurs in eight-round stages. Nonparticipants refresh their chair snapshot on deterministic slots in a cluster-sized observation window, targeting roughly 500 follow-up Lease lists per second at 100,000 nodes. diff --git a/e2e/gantry/auth_registry_test.go b/e2e/gantry/auth_registry_test.go index 2489b650a..7aecf7723 100644 --- a/e2e/gantry/auth_registry_test.go +++ b/e2e/gantry/auth_registry_test.go @@ -530,6 +530,7 @@ func (h *harness) applyConfigMapWithAuthRegistry(ctx context.Context) { " - \"/ip4/0.0.0.0/udp/4001/quic-v1\"", " libp2p_identity_path: \"/var/lib/gantry/libp2p/identity.key\"", " members_label_selector: \"app.kubernetes.io/name=gantry\"", + " chair_seed_percentage: 100", " chair_seed_count: 8", " storage_mode: \"containerd\"", " containerd_socket: \"/run/containerd/containerd.sock\"", diff --git a/e2e/gantry/harness_e2e.go b/e2e/gantry/harness_e2e.go index 8e86d60a9..e81afaddc 100644 --- a/e2e/gantry/harness_e2e.go +++ b/e2e/gantry/harness_e2e.go @@ -804,13 +804,12 @@ func patchConfigMapForE2E(raw string) (string, error) { return "", fmt.Errorf("patchConfigMapForE2E: upstream_registries anchor not found in deploy/configmap.yaml; update configMapUpstreamRegistriesAnchor in harness_e2e.go") } - // The shipped pacing staggers 100,000 nodes claiming 64 chairs. Readiness - // needs SeedCount chairs occupied, so on this 8-node cluster every pod must - // claim before the rollout completes, and the shipped values make that take - // ~118s per rollout: up to 30s of startup jitter plus 88s for the eligibility - // divisor to halve from 2048 down to 1. The suite rolls out a dozen times. + // The suite intentionally uses every kind node as a chair to exercise the + // eight-seed data path. Remove large-cluster claim pacing so repeated + // rollouts do not wait for the eligibility lottery to widen. for _, sub := range []struct{ from, to string }{ {" chair_cluster_size_estimate: 100000", " chair_cluster_size_estimate: 8"}, + {" chair_seed_percentage: 10", " chair_seed_percentage: 100"}, {" chair_seed_count: 50", " chair_seed_count: 8"}, {" chair_claim_initial_divisor: 2048", " chair_claim_initial_divisor: 1"}, {` chair_startup_jitter: "30s"`, ` chair_startup_jitter: "2s"`}, diff --git a/e2e/gantry/harness_patch_test.go b/e2e/gantry/harness_patch_test.go index 02ede5dc9..49dc5658f 100644 --- a/e2e/gantry/harness_patch_test.go +++ b/e2e/gantry/harness_patch_test.go @@ -215,9 +215,12 @@ func TestPatchConfigMapForE2E_RewritesUpstreamRegistries(t *testing.T) { t.Error("patched ConfigMap does not use an eight-chair seed cohort") } - // Readiness needs SeedCount chairs occupied, so on eight nodes every pod - // must claim before a rollout completes. The shipped divisor and jitter - // stagger 100,000 nodes and cost ~118s per rollout here. + if !strings.Contains(patched, "chair_seed_percentage: 100") { + t.Error("patched ConfigMap does not select every eight-node kind agent as a chair") + } + + // The E2E topology intentionally makes all eight nodes immediately eligible + // so its fixed eight-chair assertions do not depend on claim pacing. if !strings.Contains(patched, "chair_claim_initial_divisor: 1") { t.Error("patched ConfigMap does not make every kind node immediately claim-eligible") } diff --git a/e2e/gantry/kind-config.yaml b/e2e/gantry/kind-config.yaml index 0c34eecea..4720b2c46 100644 --- a/e2e/gantry/kind-config.yaml +++ b/e2e/gantry/kind-config.yaml @@ -1,7 +1,7 @@ # kind cluster topology for the gantry e2e suite. # -# One control-plane + seven workers provides eight independent Gantry agents, -# the minimum required to occupy the eight primary Lease chairs. +# One control-plane + seven workers provides eight independent Gantry agents. +# The E2E ConfigMap selects all eight as chairs to exercise seed fanout. # # No extraMounts are declared. Each kind node runs its own # containerd inside the node container, listening on diff --git a/internal/gantry/chairs/manager.go b/internal/gantry/chairs/manager.go index 50ca53e53..92a2b5df9 100644 --- a/internal/gantry/chairs/manager.go +++ b/internal/gantry/chairs/manager.go @@ -25,6 +25,7 @@ type ManagerOptions struct { Candidates func() []Holder Connect func(context.Context, []string) int BootstrapHealthy func() bool + SeedTarget func(context.Context) (int, error) Now func() time.Time Logger *slog.Logger LeaseDuration time.Duration @@ -58,6 +59,7 @@ type Manager struct { selectionReady bool electionEpoch int64 bootstrapReady bool + seedTarget int observationRounds uint64 duplicateChairs []ID } @@ -161,12 +163,8 @@ func (m *Manager) Held() (Chair, bool) { // Ready reports whether this agent can take part in cold start. // -// A full seed cohort is the healthy steady state, but requiring it outright -// deadlocks any cluster that cannot field SeedCount holders at once: a node -// holds one chair, so a cluster smaller than SeedCount, or the first batch of -// a rolling upgrade, would never report ready and the rollout would never -// proceed to create the holders it is waiting for. Holding a chair is -// therefore also sufficient - such a node is itself a usable seed. +// One selectable chair is sufficient for cold-start coordination. The manager +// continues filling the proportional target in the background. func (m *Manager) Ready() bool { snapshot := m.opts.Cache.Peek() epoch := m.CurrentEpoch() @@ -175,13 +173,7 @@ func (m *Manager) Ready() bool { return false } - if snapshot.SelectableCount() >= m.opts.SeedCount { - return true - } - - _, held := m.Held() - - return held + return snapshot.SelectableCount() > 0 } func (m *Manager) Run(ctx context.Context) { @@ -382,16 +374,25 @@ func (m *Manager) attemptClaim(ctx context.Context) { m.selectionReady = false } - // Whether this node should try to take a chair. A node that already holds - // one, or that has seen every chair taken, has nothing to claim. - skipClaim := m.held != nil || m.reserved != nil || m.knownFull || m.claiming || - (m.selectionReady && m.bootstrapReady && !m.participating) + held := m.held != nil + reserved := m.reserved != nil + selectionReady := m.selectionReady + bootstrapReady := m.bootstrapReady + legacySkipClaim := held || reserved || m.knownFull || m.claiming || + (selectionReady && bootstrapReady && !m.participating) // Bootstrap failure is a separate condition from a completed election. A // non-holder whose initial dials all failed still needs the snapshot below, // because observe is what retries Connect; returning here on knownFull // alone would leave it disconnected until the next epoch. - if skipClaim && m.bootstrapReady { + if m.opts.SeedTarget == nil && legacySkipClaim && bootstrapReady { + m.mu.Unlock() + return + } + + // Holders refresh capacity from maintain. Reserved successors cannot claim + // another chair while waiting for rotation. + if m.opts.SeedTarget != nil && (held || reserved) && bootstrapReady { m.mu.Unlock() return } @@ -421,7 +422,10 @@ func (m *Manager) attemptClaim(ctx context.Context) { // the shipped 100,000-node estimate this targets about 500 follow-up Lease // lists per second, while eligibility itself continues widening every stage. observerSlot := stableHash(string(m.opts.Self.PeerID), strconv.FormatInt(epoch, 10), "observe") % m.observationRounds - if !eligible && round%m.observationRounds != observerSlot { + observerTurn := round%m.observationRounds == observerSlot + + refreshingSatisfiedTarget := m.opts.SeedTarget != nil && selectionReady && bootstrapReady + if (refreshingSatisfiedTarget || !eligible) && !observerTurn { return } @@ -441,16 +445,24 @@ func (m *Manager) attemptClaim(ctx context.Context) { return } - // The snapshot above has now retried Connect; claiming stays suppressed. - if skipClaim { + refreshSeedTarget := refreshingSatisfiedTarget && snapshot.SelectableCount() < m.opts.SeedCount + + seedTarget, err := m.resolveSeedTarget(ctx, snapshot, refreshSeedTarget) + if err != nil { + m.opts.Logger.Warn("chair seed target unavailable", slog.Any("err", err)) + + return + } + + if snapshot.SelectableCount() >= seedTarget { return } - if !eligible { + if reserved || !eligible { return } - empty := make([]ID, 0, Count-snapshot.OccupiedCount()) + empty := make([]ID, 0, seedTarget) occupied := make(map[ID]struct{}, len(snapshot.Chairs)) @@ -460,7 +472,7 @@ func (m *Manager) attemptClaim(ctx context.Context) { } } - for index := range Count { + for index := range seedTarget { id := ID(index) if _, ok := occupied[id]; !ok { empty = append(empty, id) @@ -472,7 +484,7 @@ func (m *Manager) attemptClaim(ctx context.Context) { unresponsive := false if len(empty) == 0 { - if reclaimable := m.reclaimableChairs(snapshot); len(reclaimable) > 0 { + if reclaimable := m.reclaimableChairs(snapshot, seedTarget); len(reclaimable) > 0 { empty = reclaimable unresponsive = true } @@ -579,12 +591,78 @@ func (m *Manager) maintain(ctx context.Context) { if snapshotErr == nil { m.observe(ctx, snapshot) + cached = snapshot } } + seedTarget, targetErr := m.resolveSeedTarget(ctx, cached, true) + if targetErr != nil { + m.opts.Logger.Warn("chair seed target refresh failed", slog.Any("err", targetErr)) + + return + } + + if int(held.ID) >= seedTarget { + apiCtx, cancel := m.apiContext(ctx) + vacateErr := m.opts.Store.Vacate(apiCtx, held.ID, m.opts.Self.PeerID) + + cancel() + + if vacateErr == nil || errors.Is(vacateErr, ErrNotClaimable) { + m.mu.Lock() + m.held = nil + m.mu.Unlock() + m.opts.Cache.Invalidate() + } + + return + } + m.prepareRotation(ctx, updated) } +func (m *Manager) resolveSeedTarget(ctx context.Context, snapshot Snapshot, refresh bool) (int, error) { + if m.opts.SeedTarget == nil { + return m.opts.SeedCount, nil + } + + m.mu.Lock() + seedTarget := m.seedTarget + m.mu.Unlock() + + if !refresh && seedTarget > 0 { + return seedTarget, nil + } + + if !refresh && snapshot.SelectableCount() >= m.opts.SeedCount { + return m.opts.SeedCount, nil + } + + apiCtx, cancel := m.apiContext(ctx) + target, err := m.opts.SeedTarget(apiCtx) + + cancel() + + if err != nil { + return 0, err + } + + if target < 1 { + target = 1 + } + + if target > m.opts.SeedCount { + target = m.opts.SeedCount + } + + m.mu.Lock() + m.seedTarget = target + m.selectionReady = snapshot.SelectableCount() >= target + m.mu.Unlock() + + return target, nil +} + func (m *Manager) retryDuplicateVacates(ctx context.Context) { m.mu.Lock() pending := append([]ID(nil), m.duplicateChairs...) @@ -759,10 +837,15 @@ func (m *Manager) observe(ctx context.Context, snapshot Snapshot) { // claimable" rather than "every chair records a holder". Chairs abandoned by // departed nodes stay occupied forever and are exactly what a replacement // node needs to take over. - m.knownFull = snapshot.OccupiedCount() == Count && len(m.reclaimableChairs(snapshot)) == 0 + m.knownFull = snapshot.OccupiedCount() == Count && len(m.reclaimableChairs(snapshot, Count)) == 0 m.initialized = true - m.selectionReady = snapshot.SelectableCount() >= m.opts.SeedCount + seedTarget := m.seedTarget + if seedTarget == 0 { + seedTarget = m.opts.SeedCount + } + + m.selectionReady = snapshot.SelectableCount() >= seedTarget bootstrapHealthy := m.opts.BootstrapHealthy == nil || m.opts.BootstrapHealthy() if (m.opts.Connect == nil || connected > 0) && bootstrapHealthy { @@ -775,7 +858,7 @@ func (m *Manager) observe(ctx context.Context, snapshot Snapshot) { // ago to be treated as gone. Occupancy alone is not evidence of a live holder: // a node pool replaced wholesale leaves every Lease recording an absent one, so // without this no chair is ever free again and the deployment cannot recover. -func (m *Manager) reclaimableChairs(snapshot Snapshot) []ID { +func (m *Manager) reclaimableChairs(snapshot Snapshot, seedTarget int) []ID { if m.opts.LeaseDuration <= 0 { return nil } @@ -786,6 +869,10 @@ func (m *Manager) reclaimableChairs(snapshot Snapshot) []ID { out := make([]ID, 0, len(snapshot.Chairs)) for _, chair := range snapshot.Chairs { + if int(chair.ID) >= seedTarget { + continue + } + if !chair.Occupied() || chair.Holder.PeerID == m.opts.Self.PeerID { continue } diff --git a/internal/gantry/chairs/manager_internal_test.go b/internal/gantry/chairs/manager_internal_test.go index cf0c025dd..8d97f145f 100644 --- a/internal/gantry/chairs/manager_internal_test.go +++ b/internal/gantry/chairs/manager_internal_test.go @@ -70,6 +70,124 @@ func TestClaimEligibilityEventuallyIncludesEntireCluster(t *testing.T) { } } +func TestManagersClaimOnlyProportionalTargetSlots(t *testing.T) { + client := fake.NewClientset() + store := NewStore(client.CoordinationV1().Leases("gantry-system")) + + for index := range 10 { + manager := NewManager(ManagerOptions{ + Store: store, + Self: Holder{PeerID: ifaces.NodeID(fmt.Sprintf("peer-%d", index)), P2PAddrs: []string{fmt.Sprintf("/ip4/10.0.0.%d/tcp/4001", index+1)}, TransferAddr: fmt.Sprintf("10.0.0.%d:5001", index+1)}, + Now: func() time.Time { return time.Unix(0, 0) }, + ClaimJitter: time.Nanosecond, + ClaimInitialDivisor: 1, + RotationPeriod: time.Hour, + SeedCount: 50, + SeedTarget: func(context.Context) (int, error) { return 2, nil }, + }) + + if err := manager.Initialize(context.Background()); err != nil { + t.Fatalf("Initialize manager %d: %v", index, err) + } + + manager.attemptClaim(context.Background()) + } + + snapshot, err := store.Snapshot(context.Background(), 0) + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + + if snapshot.OccupiedCount() != 2 { + t.Fatalf("occupied chairs = %d; want 2", snapshot.OccupiedCount()) + } + + for _, chair := range snapshot.Chairs { + if chair.Occupied() && int(chair.ID) >= 2 { + t.Fatalf("chair %s occupied outside target slots", chair.ID.Name()) + } + } +} + +func TestManagerVacatesChairAboveReducedTarget(t *testing.T) { + client := fake.NewClientset() + store := NewStore(client.CoordinationV1().Leases("gantry-system")) + + self := Holder{PeerID: "self", P2PAddrs: []string{"/ip4/10.0.0.1/tcp/4001"}, TransferAddr: "10.0.0.1:5001"} + if _, err := store.Claim(context.Background(), 4, self, 0, time.Minute, false, time.Unix(0, 0)); err != nil { + t.Fatalf("Claim: %v", err) + } + + manager := NewManager(ManagerOptions{ + Store: store, + Self: self, + Now: func() time.Time { return time.Unix(1, 0) }, + RotationPeriod: time.Hour, + SeedCount: 50, + SeedTarget: func(context.Context) (int, error) { return 2, nil }, + }) + if err := manager.Initialize(context.Background()); err != nil { + t.Fatalf("Initialize: %v", err) + } + + manager.maintain(context.Background()) + + if _, held := manager.Held(); held { + t.Fatal("manager retained chair above reduced target") + } + + chair, err := store.Get(context.Background(), 4) + if err != nil { + t.Fatalf("Get: %v", err) + } + + if chair.Occupied() { + t.Fatal("chair above reduced target remains occupied") + } +} + +func TestManagerClaimsAfterProportionalTargetIncreases(t *testing.T) { + client := fake.NewClientset() + store := NewStore(client.CoordinationV1().Leases("gantry-system")) + + seed := Holder{PeerID: "seed", P2PAddrs: []string{"/ip4/10.0.0.1/tcp/4001"}, TransferAddr: "10.0.0.1:5001"} + if _, err := store.Claim(context.Background(), 0, seed, 0, time.Minute, false, time.Unix(0, 0)); err != nil { + t.Fatalf("Claim seed: %v", err) + } + + target := 1 + + manager := NewManager(ManagerOptions{ + Store: store, + Self: Holder{PeerID: "candidate", P2PAddrs: []string{"/ip4/10.0.0.2/tcp/4001"}, TransferAddr: "10.0.0.2:5001"}, + Now: func() time.Time { return time.Unix(0, 0) }, + ClaimJitter: time.Nanosecond, + ClaimInitialDivisor: 1, + RotationPeriod: time.Hour, + ClusterSizeEstimate: 1, + SeedCount: 50, + SeedTarget: func(context.Context) (int, error) { return target, nil }, + }) + if err := manager.Initialize(context.Background()); err != nil { + t.Fatalf("Initialize: %v", err) + } + + manager.attemptClaim(context.Background()) + + if _, held := manager.Held(); held { + t.Fatal("manager claimed while initial target was satisfied") + } + + target = 2 + + manager.attemptClaim(context.Background()) + + held, ok := manager.Held() + if !ok || held.ID != 1 { + t.Fatalf("held chair = %+v, %t; want chair 01", held, ok) + } +} + func TestManagerScalesObservationCadence(t *testing.T) { client := fake.NewClientset() manager := NewManager(ManagerOptions{ diff --git a/internal/gantry/chairs/ready_test.go b/internal/gantry/chairs/ready_test.go index 994249a69..784f6317c 100644 --- a/internal/gantry/chairs/ready_test.go +++ b/internal/gantry/chairs/ready_test.go @@ -13,15 +13,11 @@ import ( "github.com/Azure/unbounded/internal/gantry/chairs" ) -// TestManagerReadyWithFewerHoldersThanSeedCount covers the first batch of a -// rolling upgrade and any cluster smaller than SeedCount. +// TestManagerReadyWithFewerHoldersThanSeedCount covers the first chair during +// startup and the first batch of a rolling upgrade. // -// A node holds exactly one chair, so requiring SeedCount occupied chairs before -// reporting ready is unsatisfiable until SeedCount nodes are already running -// the new build. On an eight-node cluster at maxUnavailable 50% only four pods -// are replaced at a time: they would never become ready, the availability -// budget would stay spent, and the rollout could never create the holders it -// was waiting for. Holding a chair makes a node a usable seed on its own. +// Requiring the complete target before reporting ready would block the +// remaining agents that are still converging on that target. func TestManagerReadyWithFewerHoldersThanSeedCount(t *testing.T) { const ns = "gantry-system" @@ -72,9 +68,35 @@ func TestManagerReadyWithFewerHoldersThanSeedCount(t *testing.T) { cancel() <-done - // Far fewer than SeedCount chairs are occupied here: this node holds one - // and nothing else is running. + // Only this node's chair is occupied and selectable. if !ready { - t.Fatal("a chair holder reports not ready with fewer than SeedCount chairs occupied; a rolling upgrade of a small cluster would deadlock") + t.Fatal("a chair holder reports not ready with one selectable chair") + } +} + +func TestManagerReadyWithOneSelectableChairHeldByPeer(t *testing.T) { + const ns = "gantry-system" + + client := fake.NewClientset() + store := chairs.NewStore(client.CoordinationV1().Leases(ns)) + + peer := chairs.Holder{PeerID: "seed", P2PAddrs: []string{"/ip4/10.0.0.2/tcp/4001"}, TransferAddr: "10.0.0.2:5001"} + if _, err := store.Claim(context.Background(), 0, peer, 0, time.Minute, false, time.Unix(0, 0)); err != nil { + t.Fatalf("Claim: %v", err) + } + + manager := chairs.NewManager(chairs.ManagerOptions{ + Store: store, + Self: chairs.Holder{PeerID: "non-seed"}, + Now: func() time.Time { return time.Unix(0, 0) }, + RotationPeriod: time.Hour, + SeedCount: 50, + }) + if err := manager.Initialize(context.Background()); err != nil { + t.Fatalf("Initialize: %v", err) + } + + if !manager.Ready() { + t.Fatal("manager is not ready with one selectable peer chair") } } diff --git a/internal/gantry/coldstart/chair.go b/internal/gantry/coldstart/chair.go index 93a135223..a4577197e 100644 --- a/internal/gantry/coldstart/chair.go +++ b/internal/gantry/coldstart/chair.go @@ -141,7 +141,9 @@ func (r *ChairResolver) Resolve(ctx context.Context, d digest.Digest, kind iface } ranked := chairs.Rank(snapshot, d) - if len(ranked) < r.opts.SeedCount { + + seedCount := min(len(ranked), r.opts.SeedCount) + if seedCount == 0 { return nil, ErrExhausted } @@ -154,7 +156,7 @@ func (r *ChairResolver) Resolve(ctx context.Context, d digest.Digest, kind iface ) } - accepted := make([]chairs.Chair, 0, r.opts.SeedCount) + accepted := make([]chairs.Chair, 0, seedCount) sawTransientFailure := false // Recruit the seed cohort in one pass. A chair that does not answer inside @@ -165,7 +167,7 @@ func (r *ChairResolver) Resolve(ctx context.Context, d digest.Digest, kind iface // accepts nothing justifies moving down the ranking. next := 0 for len(accepted) == 0 && next < len(ranked) { - end := next + r.opts.SeedCount + end := next + seedCount if end > len(ranked) { end = len(ranked) } @@ -243,7 +245,7 @@ func (r *ChairResolver) Resolve(ctx context.Context, d digest.Digest, kind iface return nil, ErrExhausted } - end := next + r.opts.SeedCount + end := next + seedCount if end > len(ranked) { end = len(ranked) } @@ -259,7 +261,7 @@ func (r *ChairResolver) Resolve(ctx context.Context, d digest.Digest, kind iface next = end for len(accepted) == 0 && next < len(ranked) { - end = next + r.opts.SeedCount + end = next + seedCount if end > len(ranked) { end = len(ranked) } @@ -415,11 +417,13 @@ func (r *ChairResolver) PrefetchManifestChildren(ctx context.Context, _ digest.D seen[child.Digest] = struct{}{} ranked := chairs.Rank(snapshot, child.Digest) - if len(ranked) < r.opts.SeedCount { + + seedCount := min(len(ranked), r.opts.SeedCount) + if seedCount == 0 { continue } - for _, chair := range ranked[:r.opts.SeedCount] { + for _, chair := range ranked[:seedCount] { key := groupKey{ peer: chair.Holder.PeerID, chair: chair.ID, diff --git a/internal/gantry/coldstart/chair_test.go b/internal/gantry/coldstart/chair_test.go index 82581ded2..7272ebf37 100644 --- a/internal/gantry/coldstart/chair_test.go +++ b/internal/gantry/coldstart/chair_test.go @@ -145,6 +145,33 @@ func TestChairResolverDoesNotBackfillFailedSeeds(t *testing.T) { } } +func TestChairResolverUsesAvailableSmallCohort(t *testing.T) { + d := digest.MustParse("sha256:abababababababababababababababababababababababababababababababab") + snapshot := fullChairSnapshot(5) + snapshot.Chairs = snapshot.Chairs[:3] + coord := &chairCoordStub{} + resolver := coldstart.NewChairResolver(coldstart.ChairOptions{ + Chairs: &chairSnapshotStub{snapshot: snapshot}, + Discovery: &stubDisco{providers: [][]ifaces.Provider{{{NodeID: "seed", Addr: "seed:5001"}}}}, + Coord: coord, + Inflight: inflight.New(inflight.DefaultStalls(), nil), + SelfPeerID: "self", + CurrentEpoch: func() int64 { return 5 }, + SeedCount: 50, + }) + + if _, err := resolver.Resolve(context.Background(), d, ifaces.KindBlob, "registry.example.com", "repo/image", 0); err != nil { + t.Fatalf("Resolve: %v", err) + } + + coord.mu.Lock() + defer coord.mu.Unlock() + + if len(coord.calls) != 3 { + t.Fatalf("chair calls = %d; want 3 available chairs", len(coord.calls)) + } +} + func TestChairResolverRefreshesStaleChairBeforeUsingBackup(t *testing.T) { d := digest.MustParse("sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd") snapshot := fullChairSnapshot(8) diff --git a/internal/gantry/config/config.go b/internal/gantry/config/config.go index 83bcb18de..a306c7dc7 100644 --- a/internal/gantry/config/config.go +++ b/internal/gantry/config/config.go @@ -172,8 +172,13 @@ type Config struct { ChairClaimJitter time.Duration `yaml:"chair_claim_jitter"` ChairClaimInitialDivisor int `yaml:"chair_claim_initial_divisor"` ChairClusterSizeEstimate int `yaml:"chair_cluster_size_estimate"` - ChairSeedCount int `yaml:"chair_seed_count"` - ChairAPITimeout time.Duration `yaml:"chair_api_timeout"` + // ChairCapacityDaemonSet supplies actual Gantry scheduling capacity. + ChairCapacityDaemonSet string `yaml:"chair_capacity_daemonset"` + // ChairSeedPercentage selects a proportional chair cohort from capacity. + ChairSeedPercentage int `yaml:"chair_seed_percentage"` + // ChairSeedCount caps the proportional chair cohort. + ChairSeedCount int `yaml:"chair_seed_count"` + ChairAPITimeout time.Duration `yaml:"chair_api_timeout"` // ---------- Storage backend ---------- @@ -482,6 +487,8 @@ func NewDefault() *Config { ChairClaimJitter: 750 * time.Millisecond, ChairClaimInitialDivisor: 2048, ChairClusterSizeEstimate: 100_000, + ChairCapacityDaemonSet: "gantry", + ChairSeedPercentage: 10, ChairSeedCount: 50, ChairAPITimeout: 5 * time.Second, @@ -625,6 +632,8 @@ func (c *Config) LoadEnv(env func(string) string) error { setDur("CHAIR_CLAIM_JITTER", &c.ChairClaimJitter) setInt("CHAIR_CLAIM_INITIAL_DIVISOR", &c.ChairClaimInitialDivisor) setInt("CHAIR_CLUSTER_SIZE_ESTIMATE", &c.ChairClusterSizeEstimate) + setStr("CHAIR_CAPACITY_DAEMONSET", &c.ChairCapacityDaemonSet) + setInt("CHAIR_SEED_PERCENTAGE", &c.ChairSeedPercentage) setInt("CHAIR_SEED_COUNT", &c.ChairSeedCount) setDur("CHAIR_API_TIMEOUT", &c.ChairAPITimeout) @@ -705,7 +714,9 @@ func (c *Config) BindFlags(fs *flag.FlagSet) { fs.DurationVar(&c.ChairClaimJitter, "chair-claim-jitter", c.ChairClaimJitter, "maximum deterministic delay before a chair claim") fs.IntVar(&c.ChairClaimInitialDivisor, "chair-claim-initial-divisor", c.ChairClaimInitialDivisor, "initial hash-lottery divisor, halved each claim round") fs.IntVar(&c.ChairClusterSizeEstimate, "chair-cluster-size-estimate", c.ChairClusterSizeEstimate, "cluster size used to size direct-origin fallback jitter without pod watches") - fs.IntVar(&c.ChairSeedCount, "chair-seed-count", c.ChairSeedCount, "number of ranked chairs in each cold-start seed cohort") + fs.StringVar(&c.ChairCapacityDaemonSet, "chair-capacity-daemonset", c.ChairCapacityDaemonSet, "DaemonSet whose desired scheduled count sizes the chair cohort") + fs.IntVar(&c.ChairSeedPercentage, "chair-seed-percentage", c.ChairSeedPercentage, "percentage of Gantry DaemonSet capacity selected as chairs") + fs.IntVar(&c.ChairSeedCount, "chair-seed-count", c.ChairSeedCount, "maximum number of ranked chairs in each cold-start seed cohort") fs.DurationVar(&c.ChairAPITimeout, "chair-api-timeout", c.ChairAPITimeout, "timeout for one Kubernetes chair Lease API operation") // Deprecated cache flags (--cache-dir, --cache-budget-bytes, @@ -954,6 +965,14 @@ func (c *Config) Validate() error { errs = append(errs, fmt.Errorf("chair_cluster_size_estimate: must be >= 8, got %d", c.ChairClusterSizeEstimate)) } + if c.ChairCapacityDaemonSet == "" { + errs = append(errs, errors.New("chair_capacity_daemonset: must not be empty")) + } + + if c.ChairSeedPercentage < 1 || c.ChairSeedPercentage > 100 { + errs = append(errs, fmt.Errorf("chair_seed_percentage: must be between 1 and 100, got %d", c.ChairSeedPercentage)) + } + if c.ChairSeedCount < 1 || c.ChairSeedCount > 64 { errs = append(errs, fmt.Errorf("chair_seed_count: must be between 1 and 64, got %d", c.ChairSeedCount)) } diff --git a/internal/gantry/config/config_test.go b/internal/gantry/config/config_test.go index af189f308..2021afdca 100644 --- a/internal/gantry/config/config_test.go +++ b/internal/gantry/config/config_test.go @@ -54,8 +54,8 @@ func TestDefaultsValidateAfterMinimalUpstream(t *testing.T) { t.Fatalf("chair scale defaults = %d/%d, want 2048/100000", c.ChairClaimInitialDivisor, c.ChairClusterSizeEstimate) } - if c.ChairSeedCount != 50 { - t.Fatalf("ChairSeedCount = %d, want 50", c.ChairSeedCount) + if c.ChairCapacityDaemonSet != "gantry" || c.ChairSeedPercentage != 10 || c.ChairSeedCount != 50 { + t.Fatalf("chair capacity defaults = %q/%d/%d, want gantry/10/50", c.ChairCapacityDaemonSet, c.ChairSeedPercentage, c.ChairSeedCount) } if c.ChairAPITimeout != 5*time.Second { @@ -72,6 +72,80 @@ func TestDefaultsValidateAfterMinimalUpstream(t *testing.T) { } } +func TestChairCapacityConfig(t *testing.T) { + t.Run("environment", func(t *testing.T) { + c := NewDefault() + + err := c.LoadEnv(func(key string) string { + switch key { + case "GANTRY_CHAIR_CAPACITY_DAEMONSET": + return "edge-gantry" + case "GANTRY_CHAIR_SEED_PERCENTAGE": + return "25" + default: + return "" + } + }) + if err != nil { + t.Fatalf("LoadEnv: %v", err) + } + + if c.ChairCapacityDaemonSet != "edge-gantry" || c.ChairSeedPercentage != 25 { + t.Fatalf("chair capacity = %q/%d; want edge-gantry/25", c.ChairCapacityDaemonSet, c.ChairSeedPercentage) + } + }) + + t.Run("YAML", func(t *testing.T) { + c := NewDefault() + if err := c.LoadYAML(strings.NewReader("chair_capacity_daemonset: edge-gantry\nchair_seed_percentage: 25\n")); err != nil { + t.Fatalf("LoadYAML: %v", err) + } + + if c.ChairCapacityDaemonSet != "edge-gantry" || c.ChairSeedPercentage != 25 { + t.Fatalf("chair capacity = %q/%d; want edge-gantry/25", c.ChairCapacityDaemonSet, c.ChairSeedPercentage) + } + }) + + t.Run("flags", func(t *testing.T) { + c := NewDefault() + flags := flag.NewFlagSet("test", flag.ContinueOnError) + c.BindFlags(flags) + + if err := flags.Parse([]string{"--chair-capacity-daemonset=edge-gantry", "--chair-seed-percentage=25"}); err != nil { + t.Fatalf("Parse: %v", err) + } + + if c.ChairCapacityDaemonSet != "edge-gantry" || c.ChairSeedPercentage != 25 { + t.Fatalf("chair capacity = %q/%d; want edge-gantry/25", c.ChairCapacityDaemonSet, c.ChairSeedPercentage) + } + }) +} + +func TestValidateChairCapacity(t *testing.T) { + tests := []struct { + name string + mutate func(*Config) + field string + }{ + {name: "empty DaemonSet", mutate: func(c *Config) { c.ChairCapacityDaemonSet = "" }, field: "chair_capacity_daemonset"}, + {name: "zero percentage", mutate: func(c *Config) { c.ChairSeedPercentage = 0 }, field: "chair_seed_percentage"}, + {name: "percentage above 100", mutate: func(c *Config) { c.ChairSeedPercentage = 101 }, field: "chair_seed_percentage"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + c := NewDefault() + c.UpstreamRegistries = []UpstreamRegistry{{Name: "r", Endpoint: "https://r"}} + test.mutate(c) + + err := c.Validate() + if err == nil || !strings.Contains(err.Error(), test.field) { + t.Fatalf("Validate error = %v; want %s", err, test.field) + } + }) + } +} + func TestChairSeedCountConfig(t *testing.T) { t.Run("environment", func(t *testing.T) { c := NewDefault()