waitForWorkerTCPWithMetadata connects to a worker over TCP and verifies its health using the provided metadata payload. For adopted hot-idle workers, the payload must include the claimed epoch so the worker doesn't reject the health check with "stale owner epoch".
(addr, bearerToken string, serverCertPEM []byte, timeout time.Duration, hcPayload server.WorkerHealthCheckPayload)
| 498 | p.mu.Lock() |
| 499 | p.spawning-- |
| 500 | p.mu.Unlock() |
| 501 | if err != nil { |
| 502 | return nil, fmt.Errorf("spawn sized worker: %w", err) |
| 503 | } |
| 504 | |
| 505 | p.mu.Lock() |
| 506 | w, ok := p.workers[id] |
| 507 | if !ok { |
| 508 | p.mu.Unlock() |
| 509 | return nil, fmt.Errorf("sized worker %d missing after spawn", id) |
| 510 | } |
| 511 | // spawnWorker does not stamp the profile on the in-memory worker; do it here |
| 512 | // so reuse-matching and the reserved record carry the size + TTL. |
| 513 | w.profile = profile |
| 514 | nextState, err := w.SharedState().Transition(WorkerLifecycleReserved, assignment) |
| 515 | if err != nil { |
| 516 | p.mu.Unlock() |
| 517 | p.retireWorkerWithReason(id, RetireReasonCrash, LifecycleOriginReserveFailure) |
| 518 | return nil, err |
| 519 | } |
| 520 | if err := w.SetSharedState(nextState); err != nil { |
| 521 | p.mu.Unlock() |
| 522 | p.retireWorkerWithReason(id, RetireReasonCrash, LifecycleOriginReserveFailure) |
| 523 | return nil, err |
| 524 | } |
| 525 | w.SetOwnerCPInstanceID(p.cpInstanceID) |
| 526 | w.IncrementOwnerEpoch() |
| 527 | w.reservedAt = time.Now() |
| 528 | reservedRecord := p.workerRecordFor(id, w, w.OwnerEpoch(), configstore.WorkerStateReserved, "", nil) |
| 529 | p.mu.Unlock() |
| 530 | _ = p.persistWorkerRecord(reservedRecord) |
| 531 | return w, nil |
| 532 | } |
| 533 | |
| 534 | // workerResources returns resource requests and limits for a worker pod. |
| 535 | // Set via DUCKGRES_K8S_WORKER_CPU_REQUEST / DUCKGRES_K8S_WORKER_MEMORY_REQUEST. |
| 536 | // Returns empty (BestEffort) if neither is set. |
| 537 | // When set, limits are equal to requests (Guaranteed QoS). |
| 538 | func (p *K8sWorkerPool) workerResources() corev1.ResourceRequirements { |
| 539 | return p.workerResourcesForProfile(WorkerProfile{}) |
| 540 | } |
| 541 | |
| 542 | // workerPodEnv builds the explicit worker container env list. Worker pods do |
| 543 | // not inherit the CP process env — every startup flag must be mirrored here. |
| 544 | func (p *K8sWorkerPool) workerPodEnv(secretName string, workerResources corev1.ResourceRequirements) []corev1.EnvVar { |
| 545 | env := []corev1.EnvVar{ |
| 546 | { |
| 547 | Name: "DUCKGRES_DUCKDB_TOKEN", |
| 548 | ValueFrom: &corev1.EnvVarSource{ |
| 549 | SecretKeyRef: &corev1.SecretKeySelector{ |
no test coverage detected