createPodWithBackoff creates a pod, retrying transient K8s API errors with exponential backoff (500ms, 1s, 2s, 4s).
(ctx context.Context, pod *corev1.Pod)
| 420 | dialOpts = append(dialOpts, grpc.WithPerRPCCredentials(&workerTLSBearerCreds{token: bearerToken})) |
| 421 | } |
| 422 | |
| 423 | client, err := flightsql.NewClient(addr, nil, nil, dialOpts...) |
| 424 | if err == nil { |
| 425 | ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| 426 | _, err = doHealthCheckWithMetadata(ctx, client, hcPayload) |
| 427 | cancel() |
| 428 | if err == nil { |
| 429 | return client, nil |
| 430 | } |
| 431 | lastErr = err |
| 432 | _ = client.Close() |
| 433 | } else { |
| 434 | lastErr = fmt.Errorf("grpc dial: %w", err) |
| 435 | } |
| 436 | attempts++ |
| 437 | if attempts <= 3 || attempts%10 == 0 { |
| 438 | slog.Debug("waitForWorkerTCP health check attempt failed.", "addr", addr, "attempt", attempts, "error", lastErr) |
| 439 | } |
| 440 | time.Sleep(500 * time.Millisecond) |
| 441 | } |
| 442 | return nil, fmt.Errorf("timeout connecting to worker at %s (last error: %v, attempts: %d)", addr, lastErr, attempts) |
| 443 | } |
| 444 | |
| 445 | // spawnWorkerBackground spawns a worker pod without blocking AcquireWorker. |
| 446 | // The new worker becomes available for future sessions once ready. |
| 447 | func (p *K8sWorkerPool) spawnWorkerBackground(id int, image string) { |
| 448 | ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) |
| 449 | defer cancel() |
| 450 | |
| 451 | err := p.spawnWorker(ctx, id, image, WorkerProfile{}, true) |
no test coverage detected