feat(scheduler): 1000+ node scaling with HA, sharding, and O(nodes) reconciliation

[miladhzzzz]

Scheduler Optimization for 1000+ Node Scaling with High Availability

This release is a comprehensive architectural overhaul enabling persys-scheduler to run at 1,000-5,000+ node fleet scales with high availability, sharding, and significantly reduced agent communication overhead. It implements leader election, active-active failover modes, a live node cache for placement, weighted placement scoring with in-flight reservations, etcd compare-and-swap on all concurrent writes, cluster-wide event streaming to Redis, connection pooling to agents, and bounded-concurrency reconciliation.

TL;DR:

• ✅ Multiple scheduler replicas with automatic failover (new HA modes: failover, active-active with sharding)
• ✅ Agent fan-out reduced from O(workloads) to O(nodes) per reconciliation cycle
• ✅ Live node watch cache eliminates O(nodes) etcd scan per placement decision
• ✅ In-flight reservation tracking prevents oversubscription during concurrent placement
• ✅ Weighted placement algorithm (CPU + memory + spread headroom)
• ✅ Connection pooling to agents (eliminates TLS handshake per RPC)
• ✅ Cluster-wide events in Redis, not etcd (includes new event API: ListEvents, WatchEvents)
• ✅ Standalone disk/volume management API (CreateDisk, ListDisks, etc.)
• ✅ Ceph RGW S3-compatible object storage proxying (CreateBucket, ListBuckets, etc.)
• ✅ No breaking changes; all new features default to prior behavior

Major Features

1. High Availability & Leader Election

Multiple scheduler replicas can run against the same etcd cluster with automatic failover:

• Failover mode (default, SCHEDULER_HA_MODE=failover): Exactly one active replica cluster-wide drives all singleton background loops (reconciliation, drift detection, monitoring, node watch). Others are hot standbys that take over automatically if the leader dies or its etcd lease expires (typically within 15s TTL). Preserves existing single-active-instance behavior exactly.

• Active-Active mode (SCHEDULER_HA_MODE=active-active): Nodes are partitioned across SCHEDULER_SHARD_COUNT shards by stable hash of node ID. Each replica is assigned a SCHEDULER_SHARD_INDEX and only drives reconciliation/monitoring/drift-detection for nodes in its shard. Multiple shards process workloads concurrently instead of one replica doing all the work. Run more than one replica per shard for HA within a shard.

In both modes, the gRPC API runs unconditionally on every replica (stateless, protected by etcd CAS); only the background singleton loops are gated by leadership. Safe to route agent traffic to any replica via load balancer.

Implementation:

• New files: leader.go (etcd-lease-based election), sharding.go (node partitioning)
• New config: SCHEDULER_HA_MODE, SCHEDULER_SHARD_COUNT, SCHEDULER_SHARD_INDEX
• New events: LeaderElected, LeaderLost
• Replaces StartReconciliation(ctx) with StartLeaderElectedBackgroundLoops(ctx) in main.go

2. Live Node Watch Cache for Placement

A leader-elected singleton background loop maintains a continuously updated in-memory node cache via etcd watch:

• Full resync on startup and after watch stream errors
• Live watch stream keeps cache current as nodes register/update/drain
• Placement algorithm reads from this cache (O(1) in-memory lookup) instead of scanning etcd for every decision
• Falls back to live etcd scan if cache isn't populated yet (startup, resync window)

Why: Without this optimization, every placement decision would require a full etcd scan. At 1000+ nodes, that's prohibitively expensive for a high-frequency placement path.

Implementation:

• New file: node_watch.go (watch loop + cache management)
• Updates to scheduler.go: StartNodeWatch(ctx), in-memory node cache, candidateNodeSnapshot() with fallback

3. Reconciliation Fan-Out: O(workloads) → O(nodes)

Reconciliation now batches workload queries per node instead of querying each workload individually:

• Prefetch: Call GetWorkloads once per node (batch RPC) at start of cycle = O(nodes) fan-out
• Cache: Store results for the cycle duration
• Query: getActualWorkloadState consults snapshot before falling back to live per-workload RPC
• Result: Dramatic reduction in agent load when workload_count >> node_count (typical case)

Bounded Concurrency:

• Processes at most SCHEDULER_RECONCILE_CONCURRENCY workloads concurrently (default 64)
• Prevents resource exhaustion; scales predictably with node/workload count
• Cycle-overlap guard: slow cycle won't stack with next tick

Applied same bounded-concurrency optimization to MonitorNodes, MonitorWorkloads, and detectDriftOnce.

Implementation:

• New type: nodeSnapshotEntry in reconciler.go
• New method: prefetchNodeSnapshots(ctx) batches GetWorkloads calls
• Updated: getActualWorkloadState() consults snapshot first
• New config: SCHEDULER_RECONCILE_CONCURRENCY (default 64)

4. etcd Compare-and-Swap on All Concurrent Write Paths

All etcd mutations now use atomic compare-and-swap instead of unconditional put, eliminating silent overwrites:

• New method: RetryableEtcdCASPut(key, value) (with conflict retry)
• Applied to all node mutations: heartbeat, drain/ready/taint/label/capability updates, NotReady transitions
• Applied to all workload-status mutations: UpdateWorkloadStatus, UpdateWorkloadLogs, UpdateWorkloadMetadata, UpdateWorkloadRuntimeDetails
• On conflict: reload fresh state and retry, rather than silently overwrite concurrent writer

Fixes:

• Bug: usage telemetry (Usage field) was silently dropped from status projection on every read → now preserved via CAS
• Concurrency: Multiple replicas or multiple reconciliation workers can't corrupt node/workload state

Implementation:

• New method: RetryableEtcdCASPut(key, value) in etcd.go
• Updated all node_control.go methods to use CAS
• Updated reconciler state update calls to use CAS

5. Weighted Placement Algorithm with In-Flight Reservations

Placement scoring replaced single-factor "lowest utilization" with multi-factor weighted score:

Weighted Score:

• CPU headroom factor (weight 0.4): (available_cpu - in_flight_cpu) / total_cpu
• Memory headroom factor (weight 0.4): (available_memory - in_flight_memory) / total_memory
• Spread factor (weight 0.2): workload_count / max_workload_count (avoid piling onto least-full node if tied)
• Deterministic tie-breaker: hash(workload_id + node_id) (stable ordering on ties)

In-Flight Reservations:
Node AvailableCPU and AvailableMemory only update via heartbeat, which lags placement decisions. When multiple placement decisions happen concurrently (reconciliation with bounded concurrency, monitoring, multiple replicas in active-active mode), they all read stale "available" numbers and pick the same node, causing oversubscription before heartbeats catch up.

In-flight reservations track resources committed to just-assigned workloads and subtract them from effective capacity during scoring. They self-expire after 90s (multiple heartbeat intervals), trading small scoring precision loss for not needing to hook every status-confirmation path.

Implementation:

• New file: placement.go (scoring functions, in-flight reservation tracking)
• New methods: reservePlacement(nodeID, workloadID, cpu, memory), pendingReservationFor(nodeID)
• New struct: pendingReservation with TTL-based expiry
• Updated: selectNodeForWorkload() to use weighted scoring

6. Agent Connection Pooling

gRPC connections to agents are now pooled and reused:

• Map: map[nodeID]*grpc.ClientConn (keyed by node ID)
• Lifecycle: Created on first RPC to a node, reused for all subsequent RPCs, closed on connection failure
• Keepalive: Connections are keepalive-checked and automatically recreated if stale
• TLS: Cert manager can force rotation + retry on handshake failures

Impact: Eliminates TLS handshake from every apply/delete/status/list call to every node every cycle. At 1000 nodes with 5s reconciliation interval, that's 200 TLS handshakes/second removed.

Implementation:

• New struct: agentConnEntry (wraps *grpc.ClientConn)
• New field: agentConns map[string]*agentConnEntry in Scheduler
• New methods: getAgentConn(nodeID), closeAgentConns()
• Updated: All agent RPC calls to reuse connections from pool
• Updated: main.go calls sched.SetCertManager(certMgr) for cert-aware retry on TLS errors

7. Cluster-Wide Event Streaming (Redis-Backed)

Events are now stored in a Redis Stream, not etcd, and streamed via gRPC:

Why Redis instead of etcd:

• Events are high-churn data (create-once, read-occasionally, discard after TTL)
• etcd optimized for mutable state; event-only workloads stress it unnecessarily
• Redis Streams purpose-built for event logging with atomic append + automatic TTL cleanup
• Graceful degradation: Redis down → events dropped (logged), scheduler continues; etcd down → scheduler degraded

Event Types:

• Topology: NodeJoined, NodeLost, NodeLeft
• Workload: WorkloadScheduled, WorkloadFailed, DriftDetected, RetryTriggered, Rescheduled, Relocated
• Operator: NodeDraining, NodeReady, NodeTainted, NodeUntainted, NodeLabelSet, NodeLabelDeleted
• Leadership: LeaderElected, LeaderLost
• Control-plane: SchedulerModeChanged

gRPC API:

• ListEvents(type, workload_id, node_id, limit) - Historical replay (oldest-first)
• WatchEvents(type, workload_id, node_id) - Server-streaming, replays recent history then follows live events

Implementation:

• New file: events.go (event emitting, watching, filtering)
• New file: redis_store.go extensions for event stream (XADD, XREAD, XLEN, XTRIM)
• New protobuf messages: SchedulerEventView, ListEventsRequest, ListEventsResponse, WatchEventsRequest
• New methods: emitEvent(), ListEvents(), WatchEvents(), readEventsFromStream(), watchEventStream()
• Updated: all state-changing methods to call emitEvent() after successful write
• Config: REDIS_EVENT_TTL (default 24h), REDIS_EVENT_MAX_ENTRIES (default 1000)

8. Standalone Disk/Volume Management

New gRPC API for managing block volumes and managed volumes:

• CreateDisk(name, driver, size_gb, fs_type, retain_policy) - Provision standalone volume
• ListDisks() - List all volumes
• GetDisk(disk_id) - Get volume details
• DeleteDisk(disk_id, force) - Delete volume (respects retain policy)

Drivers: local, nfs, ceph-rbd

State: Stored in etcd (/volumes/*, /attachments/*)

Implementation:

• New file: disk.go (disk CRUD operations)
• New file: persistence.go (workload storage classification helpers)
• New protobuf messages: DiskView, CreateDiskRequest, etc.
• New methods: CreateDisk(), ListDisks(), GetDisk(), DeleteDisk()

9. Ceph RGW S3-Compatible Object Storage Proxying

New gRPC API for managing Ceph RGW buckets (no etcd bucket inventory):

• CreateBucket(name, region, versioning) - Create S3 bucket on RGW
• ListBuckets() - List all buckets
• GetBucket(name) - Get bucket details (size, object count, versioning, owner)
• DeleteBucket(name) - Delete bucket on RGW
• GetBucketAccess(name) - Generate S3 access key/secret (stored in Vault)
• ListBucketObjects(bucket, prefix) - List objects in bucket

Architecture:

• Control-plane credentials (list/create/delete on RGW): env vars
• User access keys: generated on creation, stored in Vault KV only
• Vault auth: vault-manager AppRole (same pattern as certmanager)
• Optional RGW Admin Ops registration for S3 client auth

Implementation:

• New file: bucket.go (RGW API proxying, S3 signing, Vault integration)
• New protobuf messages: BucketView, CreateBucketRequest, GetBucketAccessRequest, etc.
• New methods: CreateBucket(), ListBuckets(), GetBucket(), DeleteBucket(), GetBucketAccess(), ListBucketObjects()
• Config: PERSYS_RGW_ENDPOINT, PERSYS_RGW_REGION, PERSYS_RGW_ACCESS_KEY, PERSYS_RGW_SECRET_KEY, PERSYS_RGW_ADMIN_PATH

10. CoreDNS Multi-Replica Fix

Scheduler self-registration for gateway → scheduler discovery now works with multiple replicas:

Before: Single fixed etcd key; every replica overwrote others on startup. Gateway could only resolve one replica.

After: Per-instance etcd key (mirroring agent node registration pattern). CoreDNS returns one A/SRV record per running replica. Deregistration on clean shutdown prevents lingering dead records.

Note: This DNS mechanism is for gateway → scheduler discovery only. Agents do not use CoreDNS to reach the scheduler (they use direct IPs from registration).

Implementation:

• Updated: RegisterSchedulerSelfInCoreDNS(), DeregisterSchedulerSelfFromCoreDNS() to use per-instance key
• Called: In main.go startup and defer Close()

Configuration Changes

New environment variables (all have safe defaults matching prior behavior):

# HA modes
SCHEDULER_HA_MODE=failover              # "failover" (default) or "active-active"
SCHEDULER_SHARD_COUNT=1                 # Number of shards (default 1 = no sharding)
SCHEDULER_SHARD_INDEX=0                 # This instance's shard (0..SHARD_COUNT-1)

# Reconciliation concurrency
SCHEDULER_RECONCILE_CONCURRENCY=64      # Workload processing parallelism (default 64)

# Event retention (Redis Stream)
REDIS_EVENT_TTL=24h                     # Event history TTL (default 24h)
REDIS_EVENT_MAX_ENTRIES=1000            # Event stream size limit (default 1000)

# Object storage (Ceph RGW / S3)
PERSYS_RGW_ENDPOINT=http://rgw:8080     # RGW endpoint
PERSYS_RGW_REGION=default               # RGW region (default "default")
PERSYS_RGW_ACCESS_KEY=...               # Control-plane admin key
PERSYS_RGW_SECRET_KEY=...               # Control-plane admin secret
PERSYS_RGW_ADMIN_PATH=admin             # Optional Admin Ops path (default "admin")
PERSYS_RGW_VAULT_PATH_PREFIX=...        # Vault KV path for bucket credentials

# Vault configuration (updated)
PERSYS_VAULT_MANAGER_ADDR=vault-manager:50069  # Explicit vault-manager address