gantry: size chair seeds by cluster capacity

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{

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
+}

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)
+			}
+		})
+	}
+}

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
 		}
 

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"
 

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 {

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