diff --git a/README.MD b/README.MD index 8dacd1f..0fc7d03 100644 --- a/README.MD +++ b/README.MD @@ -8,7 +8,7 @@ Cattery follows a **server/agent** architecture: ``` GitHub Actions Cattery Server Infrastructure - (Scale Set API) | (Docker / GCE) + (Scale Set API) | (Docker / GCE / Nomad) | | | |-- job demand ----------->| | | |-- provision tray ----------->| @@ -28,23 +28,24 @@ GitHub Actions Cattery Server Infrastructure A tray progresses through: **Creating** -> **Registering** -> **Registered** -> **Running** -> **Deleting** -Stale trays (no heartbeat for 2+ minutes) are automatically cleaned up. +A tray that sits in any non-running status for longer than a configurable threshold is treated as stale and deleted (defaults: 5 minutes for Creating and Registering, 15 minutes for Registered and Deleting; see the `stale` section in [Configuration](docs/configuration.md)). An idle Registered tray is also told to shut itself down after 15 minutes. ## Features -- **Multi-provider** — Ships with Docker and Google Compute Engine providers, new providers can be added +- **Multi-provider** — Ships with Docker, Google Compute Engine and Nomad providers, new providers can be added - **Scale-to-zero** — Runners are only provisioned when jobs are queued - **Ephemeral runners** — Each job gets a fresh, isolated environment via JIT configuration - **Multi-org support** — Manage runners across multiple GitHub organizations from a single server -- **Automatic failed job restart** — Agents can request reruns of failed workflow jobs +- **Preemption recovery** — Jobs lost to a preempted or terminated VM are re-run once their workflow run completes (skipped for merge queue runs, closed pull requests, and runs cancelled by a person or a concurrency group) +- **High availability** — Multiple server replicas with leader election backed by MongoDB or Kubernetes Leases - **Monitoring** — Built-in status page (`/status`) and Prometheus metrics (`/metrics`) ## Prerequisites -- Go 1.25+ +- Go 1.26+ - MongoDB -- A [GitHub App](https://docs.github.com/en/apps/creating-github-apps) with Actions read/write, Checks read and Pull requests read permissions, installed on your organization -- Docker and/or GCP credentials depending on your provider choice +- A [GitHub App](https://docs.github.com/en/apps/creating-github-apps) installed on your organization with these organization permissions: Self-hosted runners (read and write, for the runner scale set and JIT runner configuration), Actions (read and write, to re-run failed jobs), Checks (read) and Pull requests (read) for the preemption recovery checks +- Docker, GCP credentials, or a Nomad cluster depending on your provider choice ## Quick start @@ -95,6 +96,8 @@ trayTypes: image: my-runner-image:latest ``` +`maxTrays` must be set: with the default of 0 no trays are ever created. A complete walkthrough with a local docker runner is in [examples/example.md](examples/example.md). + ### Run **Server:** @@ -130,7 +133,7 @@ See [charts/cattery/values.yaml](charts/cattery/values.yaml) for the available v ## Configuration -See [docs/configuration.md](docs/configuration.md) for the full configuration reference covering server, database, GitHub App, providers, and tray types. +See [docs/configuration.md](docs/configuration.md) for the full configuration reference covering server, database, stale tray cleanup, leader election, GitHub App, providers, and tray types. ## License diff --git a/charts/cattery/README.md b/charts/cattery/README.md index dfbee12..83dc55f 100644 --- a/charts/cattery/README.md +++ b/charts/cattery/README.md @@ -180,8 +180,9 @@ CRDs installed in the cluster). The Deployment defaults to `strategy: Recreate` with a single replica: each tray type's poller holds a long-running GitHub session, so two overlapping pods would double-poll. To run multiple replicas, set `config.coordination.backend` to -`mongo` or `k8s` — which leases each tray type's session to one replica at a -time — and switch `updateStrategy` to RollingUpdate. The tray API is served by +`mongo` or `k8s` — which leases each tray type's session, the workflow +restarter, and the stale tray cleanup to one replica at a time — and switch +`updateStrategy` to RollingUpdate. The tray API is served by every replica regardless, so trays stay served throughout rollouts and failovers. The `k8s` backend additionally needs `rbac.create` (default `true`) for Lease access and mounts the service account token automatically. diff --git a/docs/configuration.md b/docs/configuration.md index 53ed6f5..8aacf4d 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1,7 +1,6 @@ # Configuration -Cattery uses YAML configuration files and searches for the file named config.yaml first in the working directory and then in the /etc/cattery directory. -Cattery will use the first config file found. +Cattery reads a YAML configuration file. Pass its path with `cattery server -c /path/to/config.yaml`; without `-c`, cattery looks for `config.yaml` first in `/etc/cattery/` and then in the working directory, and uses the first one found. ### Example: @@ -10,12 +9,22 @@ server: listenAddress: "0.0.0.0:5137" statusListenAddress: "0.0.0.0:5138" advertiseUrl: https://example.org - agentSecret: my-secret-token database: uri: mongodb://localhost:27017/ database: cattery +stale: + pollInterval: 1m + thresholds: + creating: 5m + registering: 5m + registered: 15m + deleting: 15m + +coordination: + backend: memory + github: - name: my-org appId: 123456 @@ -48,7 +57,6 @@ trayTypes: maxParallelCreation: 5 config: image: cattery-runner-tiny:latest - namePrefix: cattery - name: cattery-gce-prod provider: gce-prod @@ -59,7 +67,6 @@ trayTypes: extraMetadata: cattery-agent-version: 0.0.4 config: - project: my-gcp-project zones: - europe-west1-c - europe-west1-d @@ -88,7 +95,7 @@ trayTypes: | listenAddress | string | yes | Host:port for the HTTP server to bind (e.g., 0.0.0.0:5137). | | statusListenAddress | string | no | Separate host:port for the /status and /metrics endpoints. If empty or equal to listenAddress, served on the agent port. | | advertiseUrl | string | yes | Public base URL where the server is reachable. Passed to agents. | -| agentSecret | string | no | Bearer token that agents must present to register/unregister. If empty, agent auth is disabled. | +| agentSecret | string | no | Bearer token the server requires on every `/agent/*` request. **Leave empty:** the agent does not send this token yet, so setting it rejects every agent registration. | #### database @@ -97,6 +104,30 @@ trayTypes: | uri | string | yes | MongoDB connection string (e.g., mongodb://localhost:27017/). | | database | string | yes | Database name (e.g., cattery). | +#### stale +Optional. Configures the cleanup loop that deletes trays stuck in a non-running status. A tray whose status has not changed for longer than its threshold is deleted (the provider resource is cleaned up and the record removed). `running` trays are never stale. + +| Key | Type | Required | Description | +|--------------|-------------------------|----------|-------------------------------------------------------------------------------------------------------------------------------------------------| +| pollInterval | duration | no | How often stale trays are looked for. Default `1m`. | +| thresholds | map[status]duration | no | Per-status age after which a tray is stale. Keys: `creating`, `registering`, `registered`, `deleting`. Defaults: `5m`, `5m`, `15m`, `15m`. Statuses missing from the map are not checked. | + +Durations use Go syntax (`30s`, `5m`, `1h30m`). Independently of this loop, an agent whose tray has sat in `registered` (idle, no job) for 15 minutes is told to shut down on its next ping. + +#### coordination +Optional. Selects the leader-election backend for running more than one server replica. Each tray type's scale set poller (the GitHub session), the workflow restarter and the stale tray cleanup each run on exactly one replica at a time; every replica serves the agent HTTP API regardless. + +| Key | Type | Required | Description | +|-----------------------------|----------|----------|----------------------------------------------------------------------------------------------------------------------------------| +| backend | enum | no | `memory` (default), `mongo` or `k8s`. `memory` always leads and is correct **only for a single replica**. `mongo` stores leases in the configured database. `k8s` uses `coordination.k8s.io` Leases and must run in-cluster with RBAC to get/create/update leases. | +| lease.ttl | duration | no | Lease validity; bounds worst-case failover after a leader dies. Default `30s`. Ignored by `memory`. | +| lease.renewInterval | duration | no | How often a leader renews. Default `ttl/3`. Must stay well below `ttl`. | +| lease.retryInterval | duration | no | How often a non-leader retries acquisition. Default `5s`. | +| kubernetes.namespace | string | no | `k8s` only. Namespace for Lease objects. Defaults to the pod's namespace. | +| kubernetes.leaseNamePrefix | string | no | `k8s` only. Prefix prepended to the sanitized lease key to form the Lease name. | + +Lease keys are the tray type names plus `cattery-restarter` and `cattery-stale-trays`; do not name a tray type after those two. + #### github A list of GitHub organizations/accounts the server manages via a GitHub App. @@ -142,7 +173,7 @@ Provider-specific fields: | namespace | string | no | Nomad namespace to dispatch into. Defaults to `default`. | | region | string | no | Nomad region. Defaults to the agent's region. | | tlsCaFile | string | no | Path to a PEM CA bundle for verifying the Nomad agent's TLS certificate. | - | insecure | bool | no | Skip TLS verification. Dev-only. | + | insecure | bool | no | Skip TLS verification. Dev-only. Accepts `true`/`false`, quoted or not. | #### trayTypes Defines one or more tray "profiles" that the Tray Manager can maintain. @@ -155,29 +186,30 @@ Defines one or more tray "profiles" that the Tray Manager can maintain. | runnerGroupId | int | yes | GitHub Runner Group ID to register runners into. | | githubOrg | string | yes | The GitHub org key, matching one of the entries under `github`. | | shutdown | bool | no | Whether instances should self-terminate when the job completes. | -| maxTrays | int | no | Maximum number of concurrent trays of this type. | +| maxTrays | int | yes | Maximum number of concurrent trays of this type. Also the scale set's runner capacity reported to GitHub. Must be greater than 0: with the default of 0 no trays are ever created. | | maxParallelCreation | int | no | Maximum number of trays to create in parallel. Defaults to 10. | -| extraMetadata | map[string]string | no | Extra key-value metadata passed to the provider (e.g., GCE instance metadata). | +| extraMetadata | map[string]string | no | Extra key-value metadata passed to the provider (GCE instance metadata, Nomad dispatch meta; ignored by docker). Keys are lowercased when the file is read. | | config | provider-dependent | yes | Provider-specific configuration for how to create a tray (see below). | Provider-specific config under trayType.config: - docker config - + + Containers are named after the tray id and started as ` /action-runner/cattery/cattery agent -i -s --runner-folder /action-runner` with `host.docker.internal` mapped to the host, so the image must contain the cattery binary and the Actions runner at those paths (see [examples/Dockerfile-cattery-tiny](../examples/Dockerfile-cattery-tiny)). + | Key | Type | Required | Description | |------------|--------|----------|-----------------------------------------------------------------------------| | image | string | yes | Docker image to run for the agent/runner (e.g., cattery-runner-tiny:latest) | - | namePrefix | string | no | Prefix for container names | - google (GCE) config - + + Instances are named after the tray id and receive `cattery-url` (the `advertiseUrl`) and `cattery-agent-id` (the tray id) as instance metadata, plus any `extraMetadata`. The GCP project comes from the provider. + | Key | Type | Required | Description | |------------------|----------|----------|---------------------------------------------------------------------------------| - | project | string | no | GCP project ID (can also be set at provider level) | - | zones | []string | yes | List of zones to create instances in (e.g. `europe-west1-c`) | + | zones | []string | yes | List of zones to create instances in (e.g. `europe-west1-c`); one is picked at random per tray | | machineType | string | yes | Instance machine type (e.g. `e2-standard-4`) | | instanceTemplate | string | yes | Template to base instances on (e.g. `global/instanceTemplates/cattery-default`) | - | namePrefix | string | no | Prefix for VM names | - nomad config @@ -189,7 +221,7 @@ Provider-specific config under trayType.config: **Bootstrap composition.** The provider builds the dispatched payload from three pieces: - 1. A fixed prelude that downloads the cattery agent binary from `$CATTERY_URL/agent/binary` to `/usr/local/bin/cattery`. + 1. A fixed prelude that downloads the cattery agent binary from `$CATTERY_URL/agent/download` (the server serves its own executable) to `/usr/local/bin/cattery`. 2. The optional `script` field, executed as a pre-agent hook. 3. An `exec /usr/local/bin/cattery agent -i "$TRAY_NAME" -s "$CATTERY_URL" --runner-folder `, where `` defaults to `/cattery`. diff --git a/examples/Dockerfile-cattery-tiny b/examples/Dockerfile-cattery-tiny index ab62069..4957f5a 100644 --- a/examples/Dockerfile-cattery-tiny +++ b/examples/Dockerfile-cattery-tiny @@ -1,24 +1,30 @@ -FROM ubuntu:24.04 - -RUN apt-get update && apt-get install -y \ - git dotnet-runtime-8.0 golang-go tar curl - -WORKDIR /action-runner +# Minimal runner image for the docker provider. Build from the repository root: +# docker build -t cattery-runner-tiny:latest -f examples/Dockerfile-cattery-tiny . +# +# The docker provider starts the container with +# /action-runner/cattery/cattery agent -i -s --runner-folder /action-runner +# so the agent binary and the GitHub Actions runner must live at those paths. -ARG RUNNER_VERISON='2.333' -ENV RUNNER_VERISON=$RUNNER_VERISON +ARG GO_VERSION=1.26 -RUN curl -sL -o actions-runner-linux-x64-${RUNNER_VERISON}.0.tar.gz https://github.com/actions/runner/releases/download/v${RUNNER_VERISON}.0/actions-runner-linux-x64-${RUNNER_VERISON}.0.tar.gz -RUN ls -al -RUN tar xzf ./actions-runner-linux-x64-${RUNNER_VERISON}.0.tar.gz +FROM golang:${GO_VERSION} AS build +WORKDIR /cattery +COPY src/ . +RUN CGO_ENABLED=0 go build -o /out/cattery . -WORKDIR /cattery-agent +FROM ubuntu:24.04 -COPY . . +RUN apt-get update && apt-get install -y --no-install-recommends \ + git curl ca-certificates tar \ + && rm -rf /var/lib/apt/lists/* -RUN ls -al /action-runner -RUN ls -al /cattery-agent -RUN ls -al . +WORKDIR /action-runner +ARG RUNNER_VERSION='2.333.0' +RUN curl -sL -o actions-runner.tar.gz \ + https://github.com/actions/runner/releases/download/v${RUNNER_VERSION}/actions-runner-linux-x64-${RUNNER_VERSION}.tar.gz \ + && tar xzf actions-runner.tar.gz \ + && rm actions-runner.tar.gz \ + && ./bin/installdependencies.sh -RUN cd src && go build -o /action-runner/cattery/cattery \ No newline at end of file +COPY --from=build /out/cattery /action-runner/cattery/cattery diff --git a/examples/example-config.yaml b/examples/example-config.yaml index 60af93f..c6ecca7 100644 --- a/examples/example-config.yaml +++ b/examples/example-config.yaml @@ -1,9 +1,13 @@ server: listenAddress: "0.0.0.0:5137" + # Optional. Serve /status and /metrics on a separate port. If omitted (or + # equal to listenAddress) they are served on the agent port. + # statusListenAddress: "0.0.0.0:5138" + # Public URL the agents use to reach this server. advertiseUrl: https://cattery.my-org.com database: - uri: mongodb://localhost:27017/cattery + uri: mongodb://localhost:27017/ database: cattery # Optional. Stale-tray cleanup loop. Defaults shown below. @@ -18,9 +22,10 @@ stale: deleting: 15m # Optional. Leader election for running multiple cattery replicas. Each tray -# type's scale set poller (the GitHub session) runs on exactly one replica — the -# leader for that tray type — while every replica keeps serving the tray HTTP -# API, so trays stay served across rollouts and failovers. +# type's scale set poller (the GitHub session), the workflow restarter and the +# stale tray cleanup run on exactly one replica at a time, while every replica +# keeps serving the agent HTTP API, so trays stay served across rollouts and +# failovers. # # backend "memory" (default) is a no-op that always leads — correct ONLY for a # single replica. Running multiple replicas on "memory" makes every replica @@ -41,32 +46,32 @@ coordination: # k8s backend only; both optional: # kubernetes: # namespace: cattery # defaults to the pod's namespace - # leaseNamePrefix: "cattery-" # prepended to the sanitized tray type name + # leaseNamePrefix: "cattery-" # prepended to the sanitized lease key +# GitHub organizations, each managed through a GitHub App installation. +# The App needs: Self-hosted runners (read/write), Actions (read/write), +# Checks (read), Pull requests (read). github: - - name: paritytech-stg + - name: my-org appId: 123456 installationId: 654321 appClientId: Iv123abC privateKeyPath: path/to/github-private-key.pem - webhookSecret: my-webhook-secret - - name: paritytech + - name: my-org-staging appId: 123456 - installationId: 654321 + installationId: 654322 appClientId: Iv123abC - privateKeyPath: path/to/github-private-key.pem - webhookSecret: my-webhook-secret + privateKeyPath: path/to/github-private-key-staging.pem providers: - name: docker-local type: docker - catteryUrl: http://host.containers.internal:5137 # example for podman - name: gce-stg type: google project: my-gcp-project - credentialsFile: path/to/credentials.json + credentialsFile: path/to/credentials.json # optional; Application Default Credentials otherwise - name: nomad-scw type: nomad @@ -82,37 +87,42 @@ providers: insecure: false # optional, skip TLS verification (dev only) trayTypes: + # The tray type name is the runner scale set name: target it with + # `runs-on: ` in workflows that can use the runner group. - name: cattery-tiny # free-text, shown on the status page (Tray Types tab) description: Small local Docker runner for lint and quick checks provider: docker-local shutdown: false - runnerGroupId: 3 - githubOrg: My-Github-Org + runnerGroupId: 3 # org Settings -> Actions -> Runner groups, id is in the URL + githubOrg: my-org + maxTrays: 2 # required: 0 (the default) never creates trays config: image: cattery-runner-tiny:latest - name: cattery-gce + provider: gce-stg + githubOrg: my-org-staging + runnerGroupId: 3 + maxTrays: 3 # max number of VMs of this type + maxParallelCreation: 5 # optional, defaults to 10 + shutdown: true + # Forwarded as GCE instance metadata alongside cattery-url and + # cattery-agent-id; consumed by the image's startup script. + extraMetadata: + # can be: version (0.0.2), server (to download binary from server) or a commit hash + cattery-agent-version: 0.0.4 + # cattery-agent-version: 13d197aa1e73db09514772e55794b3a0f9b7952b config: instanceTemplate: global/instanceTemplates/ machineType: e2-standard-2 - project: my-gcp-project - zones: + zones: # one is picked at random per tray - us-west1-a - us-west1-b - extraMetadata: - # can be: version (0.0.2), server (to download binary from server) or a commit hash - cattery-agent-version: 0.0.4 - # cattery-agent-version: 13d197aa1e73db09514772e55794b3a0f9b7952b - githubOrg: My-Github-Org - maxTrays: 3 # max number of VMs of this type - provider: gce - runnerGroupId: 3 # check in github org settings -> Runner groups - shutdown: true - name: cattery-nomad provider: nomad-scw - githubOrg: My-Github-Org + githubOrg: my-org runnerGroupId: 3 maxTrays: 5 shutdown: true diff --git a/examples/example.md b/examples/example.md index e89f79f..486b23c 100644 --- a/examples/example.md +++ b/examples/example.md @@ -1,42 +1,63 @@ ### Local docker -create config file -```yaml -#config.yaml -listenAddress: http://0.0.0.0:5137 -appID: 123456 -appSecret: 123456 -privateKeyPath: /path/to/private.key - -providers: - docker-local: - type: docker - -trayTypes: - cattery-tiny: - provider: docker-local - config: - image: cattery-runner-tiny:latest - namePrefix: cattery -``` - - -build docker image -```bash -docker build -t cattery-runner-tiny:latest -f .\Dockerfile-cattery-tiny ..\src\ -``` - -run cattery -```bash -cattery server -``` - -push GitHub webhook payload. Example payload can be found at `https://github.com/organizations//settings/hooks/` - -example output: -```bash -time="2025-03-26T22:10:02+04:00" level=info msg="Starting webhook server on 0.0.0.0:5137" -time="2025-03-26T22:10:04+04:00" level=debug msg="Create tray 14081360499" name=trayProviderFactory -time="2025-03-26T22:10:04+04:00" level=debug msg="Agent registration request, cattery-14081360499" agentId=cattery-14081360499 hostname=cattery-14081360499 -time="2025-03-26T22:10:06+04:00" level=info msg="Agent cattery-14081360499, cattery-14081360499 registered with runner ID 17251" agentId=cattery-14081360499 hostname=cattery-14081360499 -``` +A minimal end-to-end setup: cattery server on your machine, MongoDB in docker +compose, and runners as local docker containers. + +1. Start MongoDB (single-node replica set): + + ```bash + docker compose -f examples/docker-compose.yaml up -d + ``` + +2. Build the runner image. The build context is the repository root, because + the image compiles the cattery agent from `src/`: + + ```bash + docker build -t cattery-runner-tiny:latest -f examples/Dockerfile-cattery-tiny . + ``` + +3. Create `config.yaml`. The docker provider starts each container with + `--add-host=host.docker.internal:host-gateway`, so `advertiseUrl` must point + at that name for the agent inside the container to reach the server: + + ```yaml + server: + listenAddress: "0.0.0.0:5137" + advertiseUrl: http://host.docker.internal:5137 + + database: + uri: mongodb://localhost:27017/ + database: cattery + + github: + - name: my-org + appId: 123456 + appClientId: Iv123abC + installationId: 987654321 + privateKeyPath: /path/to/private-key.pem + + providers: + - name: docker-local + type: docker + + trayTypes: + - name: cattery-tiny + provider: docker-local + githubOrg: my-org + runnerGroupId: 1 + maxTrays: 2 + config: + image: cattery-runner-tiny:latest + ``` + +4. Run the server: + + ```bash + cd src && go build -o cattery && ./cattery server -c ../config.yaml + ``` + +5. Queue a job in a workflow that targets the runner group with + `runs-on: cattery-tiny` (the tray type name is the scale set name). The + server picks the job up through the scale set API, starts a container, the + agent registers and runs the job, and the container is removed when the + job completes. Progress is visible on `http://localhost:5137/status`. diff --git a/src/lib/election/leaseElector.go b/src/lib/election/leaseElector.go index 2dab284..39345f4 100644 --- a/src/lib/election/leaseElector.go +++ b/src/lib/election/leaseElector.go @@ -80,6 +80,10 @@ func NewLeaseElector(store LeaseStore, holder string, cfg LeaseConfig) Elector { func (e *LeaseElector) Run(ctx context.Context, key string, onElected OnElected) error { logger := e.logger.WithField("key", key) for ctx.Err() == nil { + // The lease is valid for TTL from the store's clock at the time of the + // call. Measuring from before the call (local clock) gives a + // conservative local expiry: the store's lease can only outlive it. + attemptedAt := time.Now() acquired, err := e.store.Acquire(ctx, key, e.holder, e.cfg.TTL) if err != nil { logger.Warnf("Lease acquire failed: %v", err) @@ -92,17 +96,26 @@ func (e *LeaseElector) Run(ctx context.Context, key string, onElected OnElected) } logger.Info("Acquired leadership") - e.lead(ctx, key, logger, onElected) + e.lead(ctx, key, logger, onElected, attemptedAt.Add(e.cfg.TTL)) logger.Info("Lost leadership") } return ctx.Err() } -// lead runs onElected for one leadership term: it renews on a ticker, cancels -// leaderCtx the instant a renew fails (or ctx ends), waits for onElected to -// return, and best-effort releases the lease. It returns when the term ends; -// Run then loops to attempt reacquisition. -func (e *LeaseElector) lead(ctx context.Context, key string, logger *log.Entry, onElected OnElected) { +// lead runs onElected for one leadership term. It renews on a ticker and +// cancels leaderCtx when the lease is lost or ctx ends, waits for onElected +// to return, and best-effort releases the lease. It returns when the term +// ends; Run then loops to attempt reacquisition. +// +// expiresAt is the local, conservative estimate of when the lease lapses +// without a renew. Renew errors (a slow or unreachable store) do not end the +// term by themselves: the lease is still ours until expiresAt, and dropping +// leadership on a single transient error would needlessly flap the poller. +// The term ends the instant expiresAt passes without a successful renew, and +// each renew call is bounded by expiresAt, so neither a stalled store nor a +// fast-failing one can keep leaderCtx alive past the point where another +// replica may take the key. +func (e *LeaseElector) lead(ctx context.Context, key string, logger *log.Entry, onElected OnElected, expiresAt time.Time) { leaderCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -112,9 +125,20 @@ func (e *LeaseElector) lead(ctx context.Context, key string, logger *log.Entry, onElected(leaderCtx) }() + // stepDown stops the work without releasing: the lease is either expired + // or owned by someone else, and releasing could clobber a new owner. + stepDown := func(why string) { + logger.Warnf("Stepping down: %s", why) + cancel() + <-done + } + ticker := time.NewTicker(e.cfg.RenewInterval) defer ticker.Stop() + expiry := time.NewTimer(time.Until(expiresAt)) + defer expiry.Stop() + for { select { case <-ctx.Done(): @@ -129,18 +153,28 @@ func (e *LeaseElector) lead(ctx context.Context, key string, logger *log.Entry, // lease so another replica can pick the key up immediately. e.release(key, logger) return + case <-expiry.C: + stepDown("lease expired without a successful renew") + return case <-ticker.C: - acquired, err := e.store.Acquire(ctx, key, e.holder, e.cfg.TTL) + attemptedAt := time.Now() + renewCtx, renewCancel := context.WithDeadline(ctx, expiresAt) + acquired, err := e.store.Acquire(renewCtx, key, e.holder, e.cfg.TTL) + renewCancel() + if err != nil { - logger.Warnf("Lease renew failed: %v", err) + // Still ours until expiresAt; the expiry timer ends the term + // if no later renew succeeds. + logger.Warnf("Lease renew failed, lease still valid for %s: %v", + time.Until(expiresAt).Round(time.Millisecond), err) + continue } if !acquired { - // Lost the lease (expired before renew, or stolen). Stop the - // work immediately; do NOT release — someone else may own it now. - cancel() - <-done + stepDown("lease no longer held (expired before renew, or taken over)") return } + expiresAt = attemptedAt.Add(e.cfg.TTL) + expiry.Reset(time.Until(expiresAt)) } } } diff --git a/src/lib/election/leaseElector_test.go b/src/lib/election/leaseElector_test.go index 7e090bd..4bdeeae 100644 --- a/src/lib/election/leaseElector_test.go +++ b/src/lib/election/leaseElector_test.go @@ -2,6 +2,7 @@ package election import ( "context" + "errors" "sync" "sync/atomic" "testing" @@ -19,14 +20,21 @@ type fakeStore struct { lastRelKey string lastRelHold string acquireFn func(call int) (bool, error) + // acquireCtxFn, when set, takes precedence over acquireFn and receives the + // per-call ctx so a test can simulate a store that stalls. + acquireCtxFn func(ctx context.Context, call int) (bool, error) } -func (f *fakeStore) Acquire(_ context.Context, _, _ string, _ time.Duration) (bool, error) { +func (f *fakeStore) Acquire(ctx context.Context, _, _ string, _ time.Duration) (bool, error) { f.mu.Lock() f.acquireCalls++ n := f.acquireCalls fn := f.acquireFn + ctxFn := f.acquireCtxFn f.mu.Unlock() + if ctxFn != nil { + return ctxFn(ctx, n) + } if fn != nil { return fn(n) } @@ -141,6 +149,88 @@ func TestLeaseElector_LostLeadershipCancelsButDoesNotRelease(t *testing.T) { assert.Equal(t, 0, releases, "losing the lease must not release it") } +// A renew that errors while the lease is still within its TTL must not end the +// term: the lease is still ours, and flapping on one transient store error +// would tear down the poller for nothing. +func TestLeaseElector_TransientRenewErrorKeepsLeadership(t *testing.T) { + store := &fakeStore{ + acquireFn: func(call int) (bool, error) { + if call == 1 { + return true, nil + } + // Every renew fails "transiently"; the TTL (100ms) is what bounds + // the term, not the error. + return false, errors.New("store unavailable") + }, + } + elector := newTestElector(store) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + started := make(chan struct{}) + stopped := make(chan struct{}) + var termStart, termEnd time.Time + + go elector.Run(ctx, "type-a", func(lctx context.Context) { + termStart = time.Now() + close(started) + <-lctx.Done() + termEnd = time.Now() + close(stopped) + }) + + waitClosed(t, started, "leadership start") + // The term must end once the TTL lapses without a successful renew... + waitClosed(t, stopped, "step-down at lease expiry") + + // ...but not before: with a 5ms renew interval, the old behaviour ended the + // term on the first failed renew (~5ms). Allow slack below the 100ms TTL + // for timer granularity; anything under half the TTL means a transient + // error ended the term. + term := termEnd.Sub(termStart) + assert.GreaterOrEqual(t, term, 50*time.Millisecond, "transient renew errors ended the term early (term lasted %s)", term) + acquires, releases := store.counts() + assert.Greater(t, acquires, 3, "renews kept being attempted during the term") + assert.Equal(t, 0, releases, "an expired lease must not be released") +} + +// A store that stalls on renew must not keep leaderCtx alive past the lease's +// expiry: another replica may take the key the moment the TTL lapses, and two +// live leaders would double-poll the same GitHub session. +func TestLeaseElector_StalledRenewStepsDownAtExpiry(t *testing.T) { + store := &fakeStore{ + acquireCtxFn: func(ctx context.Context, call int) (bool, error) { + if call == 1 { + return true, nil + } + <-ctx.Done() // hang until the elector gives up on this call + return false, ctx.Err() + }, + } + elector := newTestElector(store) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + started := make(chan struct{}) + stopped := make(chan struct{}) + + go elector.Run(ctx, "type-a", func(lctx context.Context) { + close(started) + <-lctx.Done() + close(stopped) + }) + + waitClosed(t, started, "leadership start") + termStart := time.Now() + waitClosed(t, stopped, "step-down after stalled renew") + + // TTL is 100ms; allow generous slack for scheduling but reject anything + // that looks like the stall being waited out indefinitely. + assert.Less(t, time.Since(termStart), 500*time.Millisecond, "leaderCtx outlived the lease TTL") + _, releases := store.counts() + assert.Equal(t, 0, releases, "an expired lease must not be released") +} + // If onElected returns on its own (the poller exited), the lease is released so // another replica can take the key immediately. func TestLeaseElector_OnElectedReturnReleasesLease(t *testing.T) { diff --git a/src/lib/metrics/metrics.go b/src/lib/metrics/metrics.go index 5247a8e..ba71f87 100644 --- a/src/lib/metrics/metrics.go +++ b/src/lib/metrics/metrics.go @@ -35,7 +35,7 @@ var ( scaleSetPollErrors = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "cattery_scaleset_poll_errors", - Help: "Number of scale set polling errors", + Help: "Number of errors while handling scale set messages (job started/completed, scaling)", }, []string{"org", "tray_type"}) // Gauges diff --git a/src/lib/restarter/workflowRestarter.go b/src/lib/restarter/workflowRestarter.go index 15aae74..04e68b5 100644 --- a/src/lib/restarter/workflowRestarter.go +++ b/src/lib/restarter/workflowRestarter.go @@ -28,30 +28,33 @@ func (wr *WorkflowRestarter) RequestRestart(ctx context.Context, workflowRunId i return wr.repository.SaveRestartRequest(ctx, workflowRunId, orgName, repoName) } -// StartPoller starts a background goroutine that periodically checks pending restart -// requests and triggers restarts when workflows have completed with failure. -func (wr *WorkflowRestarter) StartPoller(ctx context.Context) { +// RunPoller periodically checks pending restart requests and triggers restarts +// when workflows have completed with failure. It blocks until ctx is done. +// +// It must run on a single replica at a time (the caller gates it behind +// leader election): two replicas polling the same requests would both call +// GitHub's re-run endpoint for the same run. +func (wr *WorkflowRestarter) RunPoller(ctx context.Context) { const pollInterval = 30 * time.Second // Must exceed the longest expected workflow run: a job preempted early in // a run can only be re-run after the whole run completes. const requestTTL = 6 * time.Hour logger := log.WithField("component", "restarterPoller") + logger.Info("Restart poller started") - go func() { - for { - select { - case <-ctx.Done(): - logger.Info("Restart poller shutting down") - return - default: - time.Sleep(pollInterval) - wr.pollPendingRestarts(ctx, logger, requestTTL) - } - } - }() + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() - logger.Info("Restart poller started") + for { + select { + case <-ctx.Done(): + logger.Info("Restart poller shutting down") + return + case <-ticker.C: + wr.pollPendingRestarts(ctx, logger, requestTTL) + } + } } func (wr *WorkflowRestarter) pollPendingRestarts(ctx context.Context, logger *log.Entry, ttl time.Duration) { @@ -62,6 +65,13 @@ func (wr *WorkflowRestarter) pollPendingRestarts(ctx context.Context, logger *lo } for _, req := range requests { + // Shutdown (or loss of leadership) waits for this loop to return, and + // each request costs several GitHub calls; stop between requests + // rather than finishing the whole batch. + if ctx.Err() != nil { + return + } + if time.Since(req.CreatedAt) > ttl { logger.Warnf("Restart request for workflow %d expired (age: %v), deleting", req.WorkflowRunId, time.Since(req.CreatedAt)) if err := wr.repository.DeleteRestartRequest(ctx, req.WorkflowRunId); err != nil { diff --git a/src/lib/scaleSetPoller/manager.go b/src/lib/scaleSetPoller/manager.go index e0ecf3c..4601aea 100644 --- a/src/lib/scaleSetPoller/manager.go +++ b/src/lib/scaleSetPoller/manager.go @@ -8,7 +8,6 @@ import ( type Manager struct { mu sync.RWMutex pollers map[string]*Poller - wg sync.WaitGroup } func NewManager() *Manager { @@ -47,15 +46,3 @@ func (m *Manager) MessageHistory() []*Message { }) return all } - -func (m *Manager) Add(delta int) { - m.wg.Add(delta) -} - -func (m *Manager) Done() { - m.wg.Done() -} - -func (m *Manager) Wait() { - m.wg.Wait() -} diff --git a/src/lib/scaleSetPoller/poller.go b/src/lib/scaleSetPoller/poller.go index 914932e..01639a5 100644 --- a/src/lib/scaleSetPoller/poller.go +++ b/src/lib/scaleSetPoller/poller.go @@ -134,13 +134,30 @@ func (cs *catteryScaler) RecordDesiredRunners(count int) {} func (cs *catteryScaler) HandleDesiredRunnerCount(ctx context.Context, count int) (int, error) { cs.recordScaleMessage(count) - err := cs.poller.trayManager.ScaleForDemand(ctx, cs.poller.trayType, count) + // Scaling failures (provider or DB) are transient from the session's point + // of view: the next statistics message triggers another attempt. Failing + // the session would only add a 30s+ gap during which job events are lost. + if err := cs.poller.trayManager.ScaleForDemand(ctx, cs.poller.trayType, count); err != nil { + cs.recordHandlerError(fmt.Sprintf("scale for demand (%d)", count), err) + } + + active, err := cs.poller.trayManager.CountTrays(ctx, cs.poller.trayType.Name) if err != nil { - cs.poller.logger.Errorf("Failed to scale for demand (%d): %v", count, err) - return 0, err + cs.recordHandlerError("count trays", err) + return 0, nil } + return active, nil +} - return cs.poller.trayManager.CountTrays(ctx, cs.poller.trayType.Name) +// recordHandlerError logs a failure inside a listener callback and counts it +// in cattery_scaleset_poll_errors. The scaleset listener acks a message before +// invoking the callbacks and aborts the whole session on any callback error, +// so per-tray failures must never propagate back to it: the acked batch would +// be lost, and every JobStarted arriving before the session is recreated would +// leave its tray without a workflow run id (and so unrestartable on preemption). +func (cs *catteryScaler) recordHandlerError(what string, err error) { + cs.poller.logger.Errorf("Failed to %s: %v", what, err) + metrics.ScaleSetPollErrorsInc(cs.poller.trayType.GitHubOrg, cs.poller.trayType.Name) } func (cs *catteryScaler) recordScaleMessage(count int) { @@ -173,13 +190,17 @@ func (cs *catteryScaler) HandleJobStarted(ctx context.Context, jobInfo *scaleset // The tray stores the bare repository name: the restarter passes it to // GitHub API calls that take the owner separately. Storing the full name // here broke the restarter (duplicated owner in the API URL). + // + // A failure here is per-tray, not session-level, so it is logged and + // counted but not returned: the listener has already acked the message, + // and a returned error would make it drop the session and every job + // event that arrives before the session is recreated (see recordHandlerError). tray, err := cs.poller.trayManager.SetJob(ctx, jobInfo.RunnerName, jobID, jobInfo.WorkflowRunID, jobInfo.RepositoryName, jobInfo.JobDisplayName, workflowName) if err != nil { - cs.poller.logger.Errorf("Failed to set job on tray %s: %v", jobInfo.RunnerName, err) - return err + cs.recordHandlerError("set job on tray "+jobInfo.RunnerName, err) } - if tray == nil { + if err == nil && tray == nil { cs.poller.logger.Warnf("Tray %s not found for job %s (workflow run %d) — tray already removed", jobInfo.RunnerName, jobInfo.JobDisplayName, jobInfo.WorkflowRunID) } @@ -208,10 +229,10 @@ func (cs *catteryScaler) HandleJobCompleted(ctx context.Context, jobInfo *scales return nil } - _, err := cs.poller.trayManager.DeleteTray(ctx, jobInfo.RunnerName) - if err != nil { - cs.poller.logger.Errorf("Failed to delete tray %s: %v", jobInfo.RunnerName, err) - return err + // Per-tray failure: log and count, do not fail the session. The stale + // handler and the agent's own unregister will still remove the tray. + if _, err := cs.poller.trayManager.DeleteTray(ctx, jobInfo.RunnerName); err != nil { + cs.recordHandlerError("delete tray "+jobInfo.RunnerName, err) } jobID, _ := strconv.ParseInt(jobInfo.JobID, 10, 64) diff --git a/src/lib/scaleSetPoller/poller_test.go b/src/lib/scaleSetPoller/poller_test.go index 29ba69d..e442300 100644 --- a/src/lib/scaleSetPoller/poller_test.go +++ b/src/lib/scaleSetPoller/poller_test.go @@ -7,11 +7,13 @@ import ( "cattery/lib/trays" "cattery/lib/trays/providers" "context" + "errors" "sync" "testing" "time" "github.com/actions/scaleset" + log "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -116,6 +118,50 @@ func TestHandleDesiredRunnerCount_RecordsScaleBeforeProviderWaitCompletes(t *tes } } +// The scaleset listener acks a message before invoking the scaler callbacks +// and aborts the whole session on any callback error. Per-tray failures (DB or +// provider) must therefore be swallowed: returning them would drop every job +// event until the session is recreated. +func TestScalerCallbacks_PerTrayErrorsDoNotFailSession(t *testing.T) { + repo := testutil.NewMockTrayRepository() + repo.UpdateErr = errors.New("mongo unavailable") + repo.CountErr = errors.New("mongo unavailable") + + trayType := &config.TrayType{Name: "test-type", Provider: "blocking", MaxTrays: 1} + tm := trayManager.NewTrayManager(repo, &pollerTestProviderFactory{provider: &blockingProvider{}}) + poller := &Poller{ + trayType: trayType, + trayManager: tm, + history: &History{}, + logger: log.WithField("test", t.Name()), + } + scaler := &catteryScaler{poller: poller} + ctx := context.Background() + + err := scaler.HandleJobStarted(ctx, &scaleset.JobStarted{ + RunnerName: "test-type-abc", + JobMessageBase: scaleset.JobMessageBase{WorkflowRunID: 42, JobID: "7"}, + }) + assert.NoError(t, err, "SetJob failure must not propagate to the listener") + + err = scaler.HandleJobCompleted(ctx, &scaleset.JobCompleted{ + RunnerName: "test-type-abc", + JobMessageBase: scaleset.JobMessageBase{WorkflowRunID: 42, JobID: "7"}, + }) + assert.NoError(t, err, "DeleteTray failure must not propagate to the listener") + + count, err := scaler.HandleDesiredRunnerCount(ctx, 1) + assert.NoError(t, err, "scaling failure must not propagate to the listener") + assert.Equal(t, 0, count) + + // The events are still recorded so the status page reflects what GitHub sent. + kinds := []MessageKind{} + for _, m := range poller.History().Recent() { + kinds = append(kinds, m.Kind) + } + assert.ElementsMatch(t, []MessageKind{MessageKindJobStarted, MessageKindJobCompleted, MessageKindScale}, kinds) +} + func TestNewPollerInitializesHistory(t *testing.T) { poller := NewPoller(nil, &config.TrayType{Name: "test-type"}, nil) require.NotNil(t, poller.History()) diff --git a/src/lib/trayManager/trayManager.go b/src/lib/trayManager/trayManager.go index 7ba2095..56e762d 100644 --- a/src/lib/trayManager/trayManager.go +++ b/src/lib/trayManager/trayManager.go @@ -224,41 +224,46 @@ func (tm *TrayManager) DeleteTray(ctx context.Context, trayId string) (*trays.Tr return tray, nil } +// HandleStale periodically deletes trays stuck in a non-running status for +// longer than the configured threshold. It blocks until ctx is done. +// +// It must run on a single replica at a time (the caller gates it behind +// leader election): two replicas would each pick up the same stale trays and +// call the provider's cleanup twice. func (tm *TrayManager) HandleStale(ctx context.Context) { cfg := config.Get().Stale.WithDefaults() thresholds := resolveStaleThresholds(cfg.Thresholds) log.Infof("Stale handler starting: pollInterval=%s thresholds=%v", cfg.PollInterval, formatThresholds(thresholds)) - go func() { - ticker := time.NewTicker(cfg.PollInterval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - stale, err := tm.trayRepository.GetStale(ctx, thresholds) - if err != nil { - log.Errorf("Failed to get stale trays: %v", err) - continue - } + ticker := time.NewTicker(cfg.PollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + log.Info("Stale handler shutting down") + return + case <-ticker.C: + stale, err := tm.trayRepository.GetStale(ctx, thresholds) + if err != nil { + log.Errorf("Failed to get stale trays: %v", err) + continue + } - if len(stale) > 0 { - log.Infof("Found %d stale trays: %v", len(stale), stale) - } + if len(stale) > 0 { + log.Infof("Found %d stale trays: %v", len(stale), stale) + } - for _, tray := range stale { - log.Debugf("Deleting stale tray: %s (status=%s)", tray.Id, tray.Status) - if _, err := tm.DeleteTray(ctx, tray.Id); err != nil { - log.Errorf("Failed to delete tray %s: %v", tray.Id, err) - } - metrics.StaleTraysInc(tray.GitHubOrgName, tray.TrayTypeName) + for _, tray := range stale { + log.Debugf("Deleting stale tray: %s (status=%s)", tray.Id, tray.Status) + if _, err := tm.DeleteTray(ctx, tray.Id); err != nil { + log.Errorf("Failed to delete tray %s: %v", tray.Id, err) } + metrics.StaleTraysInc(tray.GitHubOrgName, tray.TrayTypeName) } } - }() + } } // resolveStaleThresholds converts the string-keyed config map into a diff --git a/src/lib/trayManager/trayManager_test.go b/src/lib/trayManager/trayManager_test.go index 7b4595b..8c3bf8c 100644 --- a/src/lib/trayManager/trayManager_test.go +++ b/src/lib/trayManager/trayManager_test.go @@ -668,7 +668,7 @@ func TestHandleStale_DeletesStaleTrays(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - tm.HandleStale(ctx) + go tm.HandleStale(ctx) // Expect both stale trays to be cleaned within a few poll cycles. assert.Eventually(t, func() bool { @@ -699,7 +699,7 @@ func TestHandleStale_GetStaleErrorDoesNotCrashLoop(t *testing.T) { tm := newTestManager(repo, &mockProviderFactory{provider: prov}) ctx, cancel := context.WithCancel(context.Background()) - tm.HandleStale(ctx) + go tm.HandleStale(ctx) // Let it tick a few times under error condition. time.Sleep(50 * time.Millisecond) @@ -737,7 +737,7 @@ func TestHandleStale_ContextCancellationStopsLoop(t *testing.T) { tm := newTestManager(repo, &mockProviderFactory{provider: prov}) ctx, cancel := context.WithCancel(context.Background()) - tm.HandleStale(ctx) + go tm.HandleStale(ctx) time.Sleep(30 * time.Millisecond) cancel() diff --git a/src/lib/trays/providers/nomadProvider.go b/src/lib/trays/providers/nomadProvider.go index 975379b..2ef3f88 100644 --- a/src/lib/trays/providers/nomadProvider.go +++ b/src/lib/trays/providers/nomadProvider.go @@ -8,6 +8,7 @@ import ( "encoding/hex" "errors" "fmt" + "strconv" "strings" "github.com/hashicorp/nomad/api" @@ -88,11 +89,21 @@ func NewNomadProvider(name string, providerConfig config.ProviderConfig) *NomadP } cfg.TLSConfig.CACert = caFile } - if strings.EqualFold(providerConfig.Get("insecure"), "true") { - if cfg.TLSConfig == nil { - cfg.TLSConfig = &api.TLSConfig{} + // Provider config is a string map; viper renders an unquoted YAML bool + // into it as "1"/"0", a quoted one as "true"/"false". ParseBool accepts + // both spellings. + if raw := providerConfig.Get("insecure"); raw != "" { + insecure, err := strconv.ParseBool(raw) + if err != nil { + logger.Errorf("nomad provider has invalid 'insecure' value %q: %v", raw, err) + return nil + } + if insecure { + if cfg.TLSConfig == nil { + cfg.TLSConfig = &api.TLSConfig{} + } + cfg.TLSConfig.Insecure = true } - cfg.TLSConfig.Insecure = true } client, err := api.NewClient(cfg) diff --git a/src/lib/trays/providers/nomadProvider_test.go b/src/lib/trays/providers/nomadProvider_test.go index 89754c5..7e53d25 100644 --- a/src/lib/trays/providers/nomadProvider_test.go +++ b/src/lib/trays/providers/nomadProvider_test.go @@ -205,7 +205,8 @@ func TestNewNomadProvider(t *testing.T) { // The TLS config lives inside an unexported nomad client field, so // we don't introspect it — we'd need a real handshake, which an // httptest server below covers. - for _, v := range []string{"true", "TRUE", "True"} { + // "1"/"0" is how viper hands an unquoted YAML bool to the string map. + for _, v := range []string{"true", "TRUE", "True", "1", "false", "0"} { p := NewNomadProvider("toaster", config.ProviderConfig{ "name": "toaster", "type": "nomad", @@ -215,6 +216,16 @@ func TestNewNomadProvider(t *testing.T) { assert.NotNil(t, p, "insecure=%q should be accepted", v) } }) + + t.Run("insecure with a non-bool value is rejected", func(t *testing.T) { + p := NewNomadProvider("toaster", config.ProviderConfig{ + "name": "toaster", + "type": "nomad", + "address": "https://example.invalid:4646", + "insecure": "yes", + }) + assert.Nil(t, p, "a value ParseBool cannot read must not be silently ignored") + }) } func TestGetProviderName(t *testing.T) { diff --git a/src/server/server.go b/src/server/server.go index a38e464..a1ac708 100644 --- a/src/server/server.go +++ b/src/server/server.go @@ -17,6 +17,7 @@ import ( "net/http" "os" "os/signal" + "sync" "syscall" "time" @@ -87,6 +88,10 @@ func Start() { logger.Fatalf("Failed to initialize leader election: %v", err) } + // Every leader-only background loop is tracked here so shutdown waits for + // all of them to step down and release their leases. + var leaderTasks sync.WaitGroup + for _, trayType := range config.Get().TrayTypes { org := config.Get().GetGitHubOrg(trayType.GitHubOrg) if org == nil { @@ -102,25 +107,20 @@ func Start() { poller := scaleSetPoller.NewPoller(ssClient, trayType, tm) ssm.Register(trayType.Name, poller) - ssm.Add(1) - go func(p *scaleSetPoller.Poller, name string) { - defer ssm.Done() - // Run the poller only while this replica holds the lease for this - // tray type; leaderCtx is cancelled the moment leadership is lost. - err := elector.Run(ctx, name, func(leaderCtx context.Context) { - runPoller(leaderCtx, p, name, logger) - }) - if err != nil && ctx.Err() == nil { - logger.Errorf("Leader election for '%s' exited: %v", name, err) - } - }(poller, trayType.Name) + // Run the poller only while this replica holds the lease for this + // tray type; leaderCtx is cancelled the moment leadership is lost. + name := trayType.Name + runLeaderTask(ctx, &leaderTasks, elector, name, logger, func(leaderCtx context.Context) { + runPoller(leaderCtx, poller, name, logger) + }) } - // Start restart poller (replaces workflow_run webhook) - rm.StartPoller(ctx) - - // Start stale tray cleanup - tm.HandleStale(ctx) + // The restart poller and the stale tray cleanup are cluster-wide singletons: + // two replicas running them would re-run the same workflows and clean the + // same trays twice. Each is leased under its own key, independent of the + // per-tray-type poller leases, so they may land on any replica. + runLeaderTask(ctx, &leaderTasks, elector, leaseKeyRestarter, logger, rm.RunPoller) + runLeaderTask(ctx, &leaderTasks, elector, leaseKeyStaleTrays, logger, tm.HandleStale) h := &handlers.Handlers{ TrayManager: tm, @@ -147,9 +147,9 @@ func Start() { } } - logger.Info("Waiting for pollers to shut down...") - ssm.Wait() - logger.Info("All pollers stopped") + logger.Info("Waiting for leader tasks to shut down...") + leaderTasks.Wait() + logger.Info("All leader tasks stopped") disconnectCtx, disconnectCancel := context.WithTimeout(context.Background(), 5*time.Second) defer disconnectCancel() @@ -159,6 +159,27 @@ func Start() { logger.Info("MongoDB connection closed") } +// Lease keys for the cluster-wide singleton loops. They share the key space +// with tray type names, so tray types must not use these names. +const ( + leaseKeyRestarter = "cattery-restarter" + leaseKeyStaleTrays = "cattery-stale-trays" +) + +// runLeaderTask runs task in the background for as long as this replica holds +// the lease for key: task receives a context that is cancelled the moment +// leadership is lost and is re-invoked whenever leadership is regained. The +// goroutine is tracked in wg so shutdown can wait for the task to step down. +func runLeaderTask(ctx context.Context, wg *sync.WaitGroup, elector election.Elector, key string, logger *log.Logger, task election.OnElected) { + wg.Add(1) + go func() { + defer wg.Done() + if err := elector.Run(ctx, key, task); err != nil && ctx.Err() == nil { + logger.Errorf("Leader election for '%s' exited: %v", key, err) + } + }() +} + // runPoller runs a tray type's scale set listener until ctx is cancelled — // either leadership was lost (leaderCtx) or the process is shutting down — // restarting the listener after transient errors.