From 4d948246750d9e5dc652e70f27837d5dafa30ebc Mon Sep 17 00:00:00 2001 From: Matthew Zhao Date: Thu, 10 Sep 2026 08:03:05 +0000 Subject: [PATCH] fix: harden reconciliation and document production risks --- README.md | 2 + docs/production-audit.md | 234 ++++++++++++++++++ go.mod | 30 +-- go.sum | 64 ++--- hack/cas-hammer/main.go | 36 ++- hack/cas-hammer/main_test.go | 80 ++++++ internal/controller/deploytracker.go | 4 +- internal/controller/fleetstate.go | 8 +- internal/controller/reconciliation_test.go | 123 +++++++++ internal/controller/rollout.go | 8 +- internal/controller/workerapp_controller.go | 71 +++++- .../controller/workerapp_controller_test.go | 59 ++++- 12 files changed, 645 insertions(+), 74 deletions(-) create mode 100644 docs/production-audit.md create mode 100644 hack/cas-hammer/main_test.go create mode 100644 internal/controller/reconciliation_test.go diff --git a/README.md b/README.md index a3605b1..7d4b915 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,8 @@ the WorkerApp reference, networking, autoscaling, and operations guides. > in [docs/celld-behaviors.md](docs/celld-behaviors.md); this README is the > operational guide. +The [production audit](docs/production-audit.md) records remaining operational risks and recommended feature changes. + ## How it works celld nodes coordinate through a bucket — deployments, cell state (one SQLite diff --git a/docs/production-audit.md b/docs/production-audit.md new file mode 100644 index 0000000..e917629 --- /dev/null +++ b/docs/production-audit.md @@ -0,0 +1,234 @@ +# Production audit — 2026-09-10 + +Scope: operator source at `1905ee7`, CRD/API, Helm and Kustomize packaging, +CI, tests, and the storage qualification helper. The celld integration was +cross-checked against the adjacent celld checkout at tag `v0.4.0`, notably +`actor.rs`, `generation.rs`, `main.rs`, and `docs/guarantees.md`. No production +cluster or application bucket was accessed. This is a source review and test +report, not a certification of storage, CNI, mesh, or runtime behavior. + +## Cleanup included + +- Upgrade `golang.org/x/text` to v0.39.0 and gRPC to v1.82.1 (with their + required transitive updates). The initial govulncheck v1.8.0 scan reported + reachable affected symbols for [GO-2026-5970](https://pkg.go.dev/vuln/GO-2026-5970) + and [GO-2026-6061](https://pkg.go.dev/vuln/GO-2026-6061). These are static + call-graph findings; exploitability was not reproduced. + +- Reject updates to typed resources, optional integration resources, and + StatefulSets unless their controller owner UID is the current WorkerApp. + ScaledObject cleanup also checks ownership and uses UID/resource-version + delete preconditions. Previously, name collisions could overwrite or delete + unrelated objects, including resources left by a previous WorkerApp UID. + Existing installations with unowned children now get an error: inspect and + resolve the collision rather than having the operator silently adopt it. +- Skip reconciliation of deleting WorkerApps and skip unchanged status writes. + The latter avoids status events repeatedly waking the primary resource watch. +- Preserve allocated Service node ports and foreign ServiceAccount annotations. + In `iamRole: auto` mode, preserve the externally provisioned identity promised + by the existing condition message; explicit identity configuration still wins. +- Watch owned networking.k8s.io Ingress objects, retry failed ScaledObject + cleanup, distinguish a missing map key from an empty value, handle IPv6 + addresses in the state client, and tolerate nonpositive poll intervals. + IPv6 URL handling alone does not provide IPv6-only fleet support: listener + addresses still bind IPv4 and cluster DNS configuration is hardcoded. +- Correct misleading internal comments about deployment polling and settle time. +- Fix `cas-hammer` to read the seed ETag once before releasing writers. Reading + an ETag independently in each writer permitted valid sequential writes to + appear as multiple winners. Reject zero rounds or fewer than two writers. + +## Feature changes recommended before production + +Priorities below describe impact when the stated trigger applies. They are +not claims that every installation will experience the failure. + +### P1: Protect acknowledged writes from correlated node loss + +[`buildPodTemplate`](../internal/controller/fleet_resources.go) uses `emptyDir` +for local state and `ScheduleAnyway` hostname spread. With fleet durability, +acknowledged writes may still reside only on the owner/follower disks before +bucket upload. Soft placement permits those copies on one host, and there is +no zone constraint. A correlated loss can remove every unuploaded copy. + +Add an explicit production storage/scheduling profile: persistent local state, +required host/zone placement with a documented minimum topology, configurable +resources and ephemeral-storage limits, and a tested recovery procedure. For +workloads that need bucket-backed acknowledgement regardless of placement, +use the existing `spec.durability: bucket` and measure the latency tradeoff. +Test host loss and fleet restart while writes are being acknowledged. Upstream's +[durability contract](https://github.com/denoland/celld/blob/v0.4.0/docs/guarantees.md) +should define the acceptance criteria. + +### P1: Make KEDA pause an acknowledged step before changing the fleet + +[`Reconcile`](../internal/controller/workerapp_controller.go) changes the +StatefulSet before calling `ensureScaledObject`. A pause error therefore occurs +after a template or replica change. Even a successful annotation write does not +prove KEDA/HPA has stopped changing replicas. `recreateStep` also relies on +StatefulSet status replica count to decide that old pods have exited; it does +not list old pods or check status freshness at that transition. + +Persist a rollout phase that requests pause, observes KEDA's paused state, +then changes the StatefulSet. Confirm old pods (including terminating pods) +are absent before starting an incompatible runtime. Resume scaling only after +convergence. Cover delayed KEDA reconciliation, API write failures, concurrent +scale-up/down, controller restart, and rapid spec changes in integration tests. +Treat lease/durability shutdown verification for lossy downgrades separately: +`Recreate` by itself does not prove every old node successfully sealed its log. + +### P1: Scale only from complete, fresh fleet measurements + +[`StatePoller.sweep`](../internal/controller/fleetstate.go) resets gauges before +sequentially polling every pod. Scrapes can observe an empty or partial fleet; +unreachable pods disappear from utilization averages. Poll time can grow by +three seconds per unreachable pod. [`promTrigger`](../internal/controller/fleet_resources.go) +does not set `ignoreNullValues`, so KEDA's default tolerates empty results. +The latency query also filters service name without destination namespace, +mixing identically named apps in different namespaces. + +Publish complete snapshots atomically, bound concurrent polling, expose sample +age/completeness, and prohibit scale-down on missing or stale data. Set +`ignoreNullValues: "false"`, namespace-scope latency queries, and test Prometheus +outage, partial pod failure, and leader transitions. See the +[KEDA Prometheus scaler contract](https://keda.sh/docs/2.18/scalers/prometheus/). + +### P1: Fail closed on invalid state responses at rollout gates + +[`StateClient.Fetch`](../internal/controller/fleetstate.go) accepts any JSON +object that decodes into `PodState`. Missing counters become zero: `{}` or an +error object is a successful sample with `restoring=0`. celld's `actor.rs` +can return an `actor_stopped` error JSON. The rollout sweep also skips pods +that lack an IP or are not Running, and does not require a complete census +of expected old ordinals before releasing a replacement. + +Validate the required safety fields and explicit runtime errors, bound response +size, and represent unavailable state distinctly from zero. Define which +unhealthy pods may safely be skipped; gate on expected membership, readiness, +and observed revision. Test malformed responses, actor failure, missing pods, +and overlapping pod termination. Introduce a persisted settle deadline if a +minimum dwell time is required: `RequeueAfter` cannot enforce one because +watch events may arrive earlier. + +### P1: Define and enforce the tenant/security boundary + +The fleet container has no security context and the celld v0.4 image has no +non-root USER. Pods do not meet Restricted Pod Security requirements and mount +a service-account token by default. WorkerApp writers can select arbitrary +images, reference namespace Secrets, choose IAM role annotations, and supply an +S3 endpoint. Deploy tracking can use the operator's ambient AWS credentials +against that endpoint. NetworkPolicy permits the entire operator namespace to +access the unauthenticated internal API; same-namespace fleet labels are not +an identity boundary for users allowed to create arbitrary pods. + +Treat WorkerApp creation as privileged workload deployment until admission and +RBAC boundaries are implemented. Add non-root writable-volume support, seccomp, +capability drops, token opt-out with workload-identity compatibility, restricted +image/role/endpoint policies, namespace-scoped operator deployments where +needed, and narrowly selected operator ingress. Verify CNI enforcement and +encrypted peer transport in the target cluster. Do not grant broad ambient +bucket authority to an operator serving untrusted WorkerApp authors. + +### P2: Separate expected deployment from observed serving version + +[`updateStatus`](../internal/controller/workerapp_controller.go) sets +`rolledOutAppVersion` from the desired annotation once StatefulSet replica +counts converge. It never reads `/state.deployment.version`. celld v0.4 polls +`deploy/current.json` and adopts in place, so a pinned CR value does not pin +the running application. A stalled adoption or a second publication can make +status disagree with reality. Some API comments still describe startup-only +loading or assert that all pods serve the reported version. + +Expose expected and per-pod observed deployment versions/generations, including +draining/swapping state; report convergence only after verifying live nodes. +Offer a documented in-place adoption policy versus a conservative restart +policy. Align API descriptions and generated chart schemas with that contract. +See [celld deployment behavior](https://github.com/denoland/celld/blob/v0.4.0/docs/README.md). + +### P2: Make deploy tracking bounded and credential-correct + +[`DeployTracker`](../internal/controller/deploytracker.go) caches by namespace +and name, without UID or bucket configuration. Deletion/recreation or a bucket +change can reuse an unrelated cached version; entries are never evicted. +Failures after TTL retry on every reconcile without a failure backoff. S3 reads +have no operator-level deadline or JSON size bound. With no static Secret the +tracker uses the operator's ambient identity, not the fleet ServiceAccount's +IRSA/AKS identity. GCS and Azure tracking are unsupported. + +Key cache entries by resource identity and relevant configuration, evict deleted +apps, bound requests and response size, back off failures, and choose an explicit +read-only tracking identity per fleet. Implement store adapters or make unsupported +auto tracking an admission error. Test credential rotation and bucket migration; +consider making bucket identity immutable unless a migration workflow is selected. + +### P2: Reconcile removal and actual readiness of integrations + +`ensureIngress` returns early when hostnames are cleared or mode is `none`, +leaving old routes live. Changing ingress mode leaves the previous resource kind. +Merged annotations retain removed WebSocket/certificate/LB settings. Conditions +not emitted on a later pass remain in status (for example a resolved pinned +version mismatch or disabled autoscaling). Successful object creation is labeled +IngressReady without checking Gateway acceptance, references, or programming. + +Track owned integration fields/resources, remove obsolete routes with ownership +checks, retire no-longer-applicable conditions, and derive readiness from actual +integration status. Avoid repeatedly updating HTTPRoute retry fields that the +standard-channel CRD prunes. Add removal/mode-switch and rejected-route tests. + +### P2: Validate API invariants and runtime configuration + +The API allows names too long for derived Services/labels, empty image/Secret +references, contradictory credential families, and inconsistent autoscaling +bounds. Very large `memoryGi` can overflow the int32 RSS calculation. Unknown +image tags bypass the compatibility table. CLI ingress typos silently select +HTTPRoute, and DNS assumes `cluster.local`. + +Add CEL/schema checks and a documented supported image/version policy; expose +cluster domain and scheduling/startup-probe configuration. Handle existing invalid +objects through actionable conditions before tightening admission on upgrades. +Test startup under slow storage; the current TCP liveness probe has no startup +probe to protect lengthy initialization. + +### P2: Make packaging and CI exercise the configuration users install + +The Helm ServiceMonitor always uses HTTPS even when `metrics.insecure` selects +an HTTP Service port. Enabling cert-manager mounts certificates but does not +pass `--metrics-cert-path` to the manager, so it can continue serving a self-signed +certificate while the monitor verifies the configured CA. The chart CI builds +and loads `celld-operator:v0.1.0`, then installs the chart without overriding its +image and without waiting for readiness. It can pass without running the build. + +Unify metrics TLS settings and flags; test HTTP, self-signed development, and +CA-verified configurations. Install the locally built image with `--wait` and +assert the running image. Add a real WorkerApp e2e suite covering admission, +rollout, storage recovery, NetworkPolicy, and autoscaling: the existing e2e suite +checks manager startup and metrics only. Add dependency/container vulnerability +scanning and pin build/CI tools by immutable versions. The e2e command logger also prints a temporary bearer token inside the curl +command; use mounted credentials and redact command output before reusing that +pattern outside disposable clusters. Manager memory limits +(128 MiB) and serial polling need load qualification for the intended fleet count. + +## Validation + +Regression tests cover API-backed no-op status/Service updates, preserved IAM +annotations, resource ownership collisions (including prior WorkerApp UIDs), +IPv4/IPv6 state URLs, and concurrent CAS writers sharing one seed ETag. +`make test` passed (Kubernetes 1.36.2 envtest), `make lint-fix` reported +zero issues, and `helm lint dist/chart` passed. The existing e2e suite passed +both tests on the dedicated `celld-operator-audit-20260910` Kind cluster with +an isolated kubeconfig; cert-manager installation was skipped. The cluster was +removed afterward. Helm certificate rendering was also inspected. +No real application durability, external identity provider, mesh, KEDA, or +production-scale load test was performed in this audit. + +The final govulncheck v1.8.0 scan (Go 1.26.8) reports zero reachable affected +symbols. It still lists these advisories without detected affected call paths; +track compatible upgrades, since this result is not a proof of non-exploitability: + +| Dependency | Remaining advisory | First fixed version | +| --- | --- | --- | +| cel-go v0.26.0 | [GO-2026-6094](https://pkg.go.dev/vuln/GO-2026-6094) | v0.30.0 | +| OpenTelemetry v1.43.0 | [GO-2026-5158](https://pkg.go.dev/vuln/GO-2026-5158) | v1.44.0 | +| x/mod v0.37.0 | [GO-2026-6180](https://pkg.go.dev/vuln/GO-2026-6180), [GO-2026-6179](https://pkg.go.dev/vuln/GO-2026-6179) | v0.40.0 | + +`go test -race ./hack/cas-hammer` also passed. Container/base-image and Rust +runtime dependency vulnerability scans were not part of this audit. diff --git a/go.mod b/go.mod index d4af06c..1984562 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.26.0 require ( github.com/aws/aws-sdk-go-v2 v1.43.5 github.com/aws/aws-sdk-go-v2/config v1.32.36 + github.com/aws/aws-sdk-go-v2/credentials v1.19.35 github.com/aws/aws-sdk-go-v2/service/s3 v1.107.1 github.com/aws/smithy-go v1.27.7 github.com/onsi/ginkgo/v2 v2.27.4 @@ -23,7 +24,6 @@ require ( github.com/Masterminds/semver/v3 v3.4.0 // indirect github.com/antlr4-go/antlr/v4 v4.13.0 // indirect github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.17 // indirect - github.com/aws/aws-sdk-go-v2/credentials v1.19.35 // indirect github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.36 // indirect github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.36 // indirect github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.36 // indirect @@ -85,31 +85,31 @@ require ( github.com/x448/float16 v0.8.4 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.65.0 // indirect - go.opentelemetry.io/otel v1.41.0 // indirect + go.opentelemetry.io/otel v1.43.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.40.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.40.0 // indirect - go.opentelemetry.io/otel/metric v1.41.0 // indirect - go.opentelemetry.io/otel/sdk v1.40.0 // indirect - go.opentelemetry.io/otel/trace v1.41.0 // indirect + go.opentelemetry.io/otel/metric v1.43.0 // indirect + go.opentelemetry.io/otel/sdk v1.43.0 // indirect + go.opentelemetry.io/otel/trace v1.43.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.27.1 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/exp v0.0.0-20251219203646-944ab1f22d93 // indirect - golang.org/x/mod v0.35.0 // indirect - golang.org/x/net v0.55.0 // indirect + golang.org/x/mod v0.37.0 // indirect + golang.org/x/net v0.56.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sync v0.20.0 // indirect - golang.org/x/sys v0.45.0 // indirect - golang.org/x/term v0.43.0 // indirect - golang.org/x/text v0.37.0 // indirect + golang.org/x/sync v0.21.0 // indirect + golang.org/x/sys v0.46.0 // indirect + golang.org/x/term v0.44.0 // indirect + golang.org/x/text v0.39.0 // indirect golang.org/x/time v0.15.0 // indirect - golang.org/x/tools v0.44.0 // indirect + golang.org/x/tools v0.47.0 // indirect gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260128011058-8636f8732409 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409 // indirect - google.golang.org/grpc v1.79.3 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect + google.golang.org/grpc v1.82.1 // indirect google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect diff --git a/go.sum b/go.sum index 67d8451..9c86796 100644 --- a/go.sum +++ b/go.sum @@ -211,20 +211,20 @@ go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.65.0 h1:7iP2uCb7sGddAr30RRS6xjKy7AZ2JtTOPA3oolgVSw8= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.65.0/go.mod h1:c7hN3ddxs/z6q9xwvfLPk+UHlWRQyaeR1LdgfL/66l0= -go.opentelemetry.io/otel v1.41.0 h1:YlEwVsGAlCvczDILpUXpIpPSL/VPugt7zHThEMLce1c= -go.opentelemetry.io/otel v1.41.0/go.mod h1:Yt4UwgEKeT05QbLwbyHXEwhnjxNO6D8L5PQP51/46dE= +go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.40.0 h1:QKdN8ly8zEMrByybbQgv8cWBcdAarwmIPZ6FThrWXJs= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.40.0/go.mod h1:bTdK1nhqF76qiPoCCdyFIV+N/sRHYXYCTQc+3VCi3MI= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.40.0 h1:DvJDOPmSWQHWywQS6lKL+pb8s3gBLOZUtw4N+mavW1I= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.40.0/go.mod h1:EtekO9DEJb4/jRyN4v4Qjc2yA7AtfCBuz2FynRUWTXs= -go.opentelemetry.io/otel/metric v1.41.0 h1:rFnDcs4gRzBcsO9tS8LCpgR0dxg4aaxWlJxCno7JlTQ= -go.opentelemetry.io/otel/metric v1.41.0/go.mod h1:xPvCwd9pU0VN8tPZYzDZV/BMj9CM9vs00GuBjeKhJps= -go.opentelemetry.io/otel/sdk v1.40.0 h1:KHW/jUzgo6wsPh9At46+h4upjtccTmuZCFAc9OJ71f8= -go.opentelemetry.io/otel/sdk v1.40.0/go.mod h1:Ph7EFdYvxq72Y8Li9q8KebuYUr2KoeyHx0DRMKrYBUE= -go.opentelemetry.io/otel/sdk/metric v1.40.0 h1:mtmdVqgQkeRxHgRv4qhyJduP3fYJRMX4AtAlbuWdCYw= -go.opentelemetry.io/otel/sdk/metric v1.40.0/go.mod h1:4Z2bGMf0KSK3uRjlczMOeMhKU2rhUqdWNoKcYrtcBPg= -go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa9TIN0= -go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= @@ -239,34 +239,34 @@ go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/exp v0.0.0-20251219203646-944ab1f22d93 h1:fQsdNF2N+/YewlRZiricy4P1iimyPKZ/xwniHj8Q2a0= golang.org/x/exp v0.0.0-20251219203646-944ab1f22d93/go.mod h1:EPRbTFwzwjXj9NpYyyrvenVh9Y+GFeEvMNh7Xuz7xgU= -golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM= -golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU= -golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= -golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= +golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= +golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= +golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= +golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= -golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= -golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= -golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= -golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= +golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= +golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= +golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.44.0 h1:0rLvDRCtNj0gZkyIXhCyOb2OAzEhLVqc4B+hrsBhrmc= +golang.org/x/term v0.44.0/go.mod h1:7ze4MdzUzLXpSAoFP1H0bOI9aXDqveSvatT5vKcFh2Y= +golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus= +golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= -golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c= -golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= +golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= +golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= gomodules.xyz/jsonpatch/v2 v2.4.0 h1:Ci3iUJyx9UeRx7CeFN8ARgGbkESwJK+KB9lLcWxY/Zw= gomodules.xyz/jsonpatch/v2 v2.4.0/go.mod h1:AH3dM2RI6uoBZxn3LVrfvJ3E0/9dG4cSrbuBJT4moAY= -gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= -gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= -google.golang.org/genproto/googleapis/api v0.0.0-20260128011058-8636f8732409 h1:merA0rdPeUV3YIIfHHcH4qBkiQAc1nfCKSI7lB4cV2M= -google.golang.org/genproto/googleapis/api v0.0.0-20260128011058-8636f8732409/go.mod h1:fl8J1IvUjCilwZzQowmw2b7HQB2eAuYBabMXzWurF+I= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409 h1:H86B94AW+VfJWDqFeEbBPhEtHzJwJfTbgE2lZa54ZAQ= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ= -google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE= -google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478 h1:yQugLulqltosq0B/f8l4w9VryjV+N/5gcW0jQ3N8Qec= +google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478/go.mod h1:C6ADNqOxbgdUUeRTU+LCHDPB9ttAMCTff6auwCVa4uc= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE= +google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af h1:+5/Sw3GsDNlEmu7TfklWKPdQ0Ykja5VEmq2i817+jbI= google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/hack/cas-hammer/main.go b/hack/cas-hammer/main.go index dcd5c29..c8a1857 100644 --- a/hack/cas-hammer/main.go +++ b/hack/cas-hammer/main.go @@ -76,6 +76,11 @@ func main() { os.Exit(2) } + if f.writers < 2 || f.rounds < 1 { + fmt.Fprintln(os.Stderr, "--writers must be at least 2 and --rounds must be positive") + os.Exit(2) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) defer cancel() @@ -97,13 +102,15 @@ func main() { violations += runPhase(ctx, cli, f, "update", raceUpdate) if violations > 0 { - fmt.Printf("\nFAIL: %d rounds admitted more than one winner — this store cannot fence celld cells\n", violations) + fmt.Printf("\nFAIL: %d rounds did not have exactly one winner — this store cannot fence celld cells\n", violations) os.Exit(1) } fmt.Println("\nOK: every round had exactly one winner (atomicity holds for this run; rerun per store release)") } -type racer func(ctx context.Context, cli *s3.Client, f flags, key string, writer int) (won bool, err error) +type racer func( + ctx context.Context, cli *s3.Client, f flags, key string, writer int, etag *string, +) (won bool, err error) func runPhase(ctx context.Context, cli *s3.Client, f flags, name string, race racer) int { violations := 0 @@ -112,6 +119,7 @@ func runPhase(ctx context.Context, cli *s3.Client, f flags, name string, race ra // The update phase races If-Match against a seeded object; the // create phase races If-None-Match:* against an absent key. + var etag *string if name == "update" { if _, err := cli.PutObject(ctx, &s3.PutObjectInput{ Bucket: &f.bucket, Key: &key, Body: bytes.NewReader([]byte("seed")), @@ -119,6 +127,14 @@ func runPhase(ctx context.Context, cli *s3.Client, f flags, name string, race ra fmt.Fprintf(os.Stderr, "round %d: seeding: %v\n", round, err) os.Exit(2) } + // Read once before releasing any writer. Per-writer HEADs can + // observe earlier winners and turn the race into valid serial CASes. + head, err := cli.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &f.bucket, Key: &key}) + if err != nil || head == nil || head.ETag == nil { + fmt.Fprintf(os.Stderr, "round %d: reading seed ETag: %v\n", round, err) + os.Exit(2) + } + etag = head.ETag } var mu sync.Mutex @@ -131,7 +147,7 @@ func runPhase(ctx context.Context, cli *s3.Client, f flags, name string, race ra go func(w int) { defer wg.Done() <-start - won, err := race(ctx, cli, f, key, w) + won, err := race(ctx, cli, f, key, w, etag) mu.Lock() defer mu.Unlock() if err != nil { @@ -164,7 +180,7 @@ func runPhase(ctx context.Context, cli *s3.Client, f flags, name string, race ra // raceCreate: N writers race a conditional create (If-None-Match: *) on an // absent key — celld's ownership-record acquisition and epoch seal. -func raceCreate(ctx context.Context, cli *s3.Client, f flags, key string, writer int) (bool, error) { +func raceCreate(ctx context.Context, cli *s3.Client, f flags, key string, writer int, etag *string) (bool, error) { body := fmt.Sprintf("writer-%d", writer) _, err := cli.PutObject(ctx, &s3.PutObjectInput{ Bucket: &f.bucket, @@ -175,20 +191,16 @@ func raceCreate(ctx context.Context, cli *s3.Client, f flags, key string, writer return classify(err) } -// raceUpdate: N writers read the seeded object's ETag, then race a +// raceUpdate: N writers share the seeded object's ETag, then race a // conditional overwrite (If-Match) — celld's ownership compare-and-swap. // Every writer holds the SAME valid ETag, so atomicity alone decides. -func raceUpdate(ctx context.Context, cli *s3.Client, f flags, key string, writer int) (bool, error) { - head, err := cli.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &f.bucket, Key: &key}) - if err != nil { - return false, fmt.Errorf("head before CAS: %w", err) - } +func raceUpdate(ctx context.Context, cli *s3.Client, f flags, key string, writer int, etag *string) (bool, error) { body := fmt.Sprintf("writer-%d", writer) - _, err = cli.PutObject(ctx, &s3.PutObjectInput{ + _, err := cli.PutObject(ctx, &s3.PutObjectInput{ Bucket: &f.bucket, Key: &key, Body: bytes.NewReader([]byte(body)), - IfMatch: head.ETag, + IfMatch: etag, }) return classify(err) } diff --git a/hack/cas-hammer/main_test.go b/hack/cas-hammer/main_test.go new file mode 100644 index 0000000..2ee5de3 --- /dev/null +++ b/hack/cas-hammer/main_test.go @@ -0,0 +1,80 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package main + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" +) + +func TestUpdateRaceSharesSeedETag(t *testing.T) { + var mu sync.Mutex + heads, attempts := 0, 0 + etag := `"seed"` + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + switch r.Method { + case http.MethodHead: + heads++ + w.Header().Set("ETag", etag) + case http.MethodPut: + match := r.Header.Get("If-Match") + if match == "" { + return + } // Seed. + attempts++ + if match != `"seed"` { + t.Errorf("writer used a later ETag: %s", match) + } + if match != etag { + w.Header().Set("Content-Type", "application/xml") + w.WriteHeader(http.StatusPreconditionFailed) + _, _ = w.Write([]byte("PreconditionFailed")) + return + } + etag = `"updated"` + w.Header().Set("ETag", etag) + case http.MethodDelete: + w.WriteHeader(http.StatusNoContent) + default: + t.Errorf("unexpected method: %s", r.Method) + } + })) + defer server.Close() + cli := s3.NewFromConfig(aws.Config{ + Region: "us-east-1", + Credentials: credentials.NewStaticCredentialsProvider("test", "test", ""), + }, func(o *s3.Options) { + o.BaseEndpoint = aws.String(server.URL) + o.UsePathStyle = true + }) + f := flags{bucket: "test-bucket", writers: 8, rounds: 1, prefix: "test"} + if violations := runPhase(context.Background(), cli, f, "update", raceUpdate); violations != 0 { + t.Fatalf("valid store reported %d violations", violations) + } + if heads != 1 || attempts != f.writers { + t.Fatalf("got %d HEADs and %d attempts", heads, attempts) + } +} diff --git a/internal/controller/deploytracker.go b/internal/controller/deploytracker.go index 35b58a0..969a45e 100644 --- a/internal/controller/deploytracker.go +++ b/internal/controller/deploytracker.go @@ -38,10 +38,10 @@ import ( // Deploy tracking closes the loop between `celld deploy` and the rollout: // with spec.appVersion "auto", the operator follows the fleet bucket's -// deploy/current.json — the same pointer celld nodes read at startup — and +// deploy/current.json — the same pointer celld v0.4 nodes poll in place — and // a new publication becomes an ordinary gated rollout with no CR edit. // With a pinned appVersion, the same read powers a mismatch warning, since -// nodes always load what current.json names, not what the CR says. +// nodes adopt what current.json names, not what the CR says. // AppVersionAuto is the spec.appVersion sentinel that enables tracking. const AppVersionAuto = "auto" diff --git a/internal/controller/fleetstate.go b/internal/controller/fleetstate.go index 93f3fae..a026d03 100644 --- a/internal/controller/fleetstate.go +++ b/internal/controller/fleetstate.go @@ -20,7 +20,9 @@ import ( "context" "encoding/json" "fmt" + "net" "net/http" + "strconv" "time" "github.com/prometheus/client_golang/prometheus" @@ -157,7 +159,7 @@ func NewStateClient() *StateClient { } func (c *StateClient) Fetch(ctx context.Context, podIP string) (*PodState, error) { - url := fmt.Sprintf("http://%s:%d/state", podIP, internalPort) + url := "http://" + net.JoinHostPort(podIP, strconv.Itoa(internalPort)) + "/state" req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { return nil, err @@ -212,7 +214,7 @@ func (p *StatePoller) NeedLeaderElection() bool { return true } func (p *StatePoller) Start(ctx context.Context) error { log := logf.FromContext(ctx).WithName("state-poller") interval := p.Interval - if interval == 0 { + if interval <= 0 { interval = 15 * time.Second } ticker := time.NewTicker(interval) @@ -223,7 +225,7 @@ func (p *StatePoller) Start(ctx context.Context) error { return nil case <-ticker.C: if err := p.sweep(ctx); err != nil { - log.Error(err, "fleet state sweep failed") + log.Error(err, "Failed to sweep fleet state") } } } diff --git a/internal/controller/reconciliation_test.go b/internal/controller/reconciliation_test.go new file mode 100644 index 0000000..1b6172d --- /dev/null +++ b/internal/controller/reconciliation_test.go @@ -0,0 +1,123 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + "context" + "io" + "net/http" + "strconv" + "strings" + "testing" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + platformv1alpha1 "github.com/ezgamehost/celld-operator/api/v1alpha1" +) + +const testStatefulSetKind = "StatefulSet" + +func TestOwnershipCollisions(t *testing.T) { + for _, ownedByOther := range []bool{false, true} { + for _, kind := range []string{testStatefulSetKind, "ScaledObject"} { + t.Run(kind+"/foreign="+strconv.FormatBool(ownedByOther), func(t *testing.T) { + scheme := runtime.NewScheme() + for _, add := range []func(*runtime.Scheme) error{corev1.AddToScheme, appsv1.AddToScheme, platformv1alpha1.AddToScheme} { + if err := add(scheme); err != nil { + t.Fatal(err) + } + } + app := &platformv1alpha1.WorkerApp{ObjectMeta: metav1.ObjectMeta{Name: "app", Namespace: "default", UID: "app-uid"}} + app.Spec.Celld.Image = testCelldImage + app.Spec.Bucket.Name = "s3://bucket/app" + var obj client.Object + if kind == testStatefulSetKind { + obj = buildStatefulSet(app, "", "old") + } else { + obj = newUnstructuredObject("keda.sh/v1alpha1", kind, fleetName(app), app.Namespace, nil, nil, map[string]any{"sentinel": "preserve"}) + } + if ownedByOther { + obj.SetOwnerReferences([]metav1.OwnerReference{{APIVersion: platformv1alpha1.SchemeGroupVersion.String(), Kind: "WorkerApp", Name: app.Name, UID: "previous-app-uid", Controller: ptr.To(true)}}) + } + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(obj).Build() + r := &WorkerAppReconciler{Client: c, Reader: c, Scheme: scheme} + var err error + if kind == testStatefulSetKind { + _, err = r.reconcileFleet(context.Background(), app, "new") + } else { + desired := obj.DeepCopyObject().(*unstructured.Unstructured) + desired.Object["spec"] = map[string]any{"changed": true} + err = r.ensureUnstructured(context.Background(), app, desired) + var conditions []metav1.Condition + if !r.ensureScaledObject(context.Background(), app, fleetOutcome{}, &conditions) || len(conditions) != 1 { + t.Fatal("cleanup must report an ownership conflict") + } + } + if err == nil || !strings.Contains(err.Error(), "not controlled by WorkerApp") { + t.Fatalf("expected ownership error, got %v", err) + } + live := obj.DeepCopyObject().(client.Object) + if err := c.Get(context.Background(), client.ObjectKeyFromObject(obj), live); err != nil { + t.Fatal(err) + } + if kind == testStatefulSetKind && live.(*appsv1.StatefulSet).Spec.Template.Annotations[appVersionAnnotation] != "old" { + t.Fatal("foreign StatefulSet was changed") + } + if kind == "ScaledObject" && live.(*unstructured.Unstructured).Object["spec"].(map[string]any)["sentinel"] != "preserve" { + t.Fatal("foreign ScaledObject was changed") + } + }) + } + } +} + +type stateRoundTripper func(*http.Request) (*http.Response, error) + +func (f stateRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +func TestStateClientPodAddresses(t *testing.T) { + for _, address := range []struct{ ip, host string }{{"10.0.0.1", "10.0.0.1:8081"}, {"fd00::1", "[fd00::1]:8081"}} { + t.Run(address.ip, func(t *testing.T) { + c := &StateClient{HTTP: &http.Client{Transport: stateRoundTripper(func(req *http.Request) (*http.Response, error) { + if req.URL.Host != address.host || req.URL.Path != "/state" { + t.Fatalf("unexpected URL %s", req.URL) + } + return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(`{"occupied":2,"restoring":1}`)), Header: make(http.Header)}, nil + })}} + state, err := c.Fetch(context.Background(), address.ip) + if err != nil { + t.Fatal(err) + } + if state.Occupied != 2 || state.Restoring != 1 { + t.Fatalf("unexpected state: %+v", state) + } + }) + } +} + +func TestStringMapSubsetRequiresKey(t *testing.T) { + if stringMapSubset(map[string]string{"key": ""}, nil) { + t.Fatal("missing key is not an empty-valued key") + } +} diff --git a/internal/controller/rollout.go b/internal/controller/rollout.go index 9af5503..9b42e96 100644 --- a/internal/controller/rollout.go +++ b/internal/controller/rollout.go @@ -285,6 +285,10 @@ func (r *WorkerAppReconciler) reconcileFleet(ctx context.Context, app *platformv return fleetOutcome{}, err } + if err := requireOwnership(app, &existing); err != nil { + return fleetOutcome{}, err + } + desiredHash := desired.Annotations[templateHashAnnotation] templateChanged := existing.Annotations[templateHashAnnotation] != desiredHash oldImage := stsContainerImage(&existing) @@ -360,8 +364,8 @@ func (r *WorkerAppReconciler) reconcileFleet(ctx context.Context, app *platformv } // rolloutStep advances a gated rolling update by at most one ordinal. -// Reconciles arrive at least Requeue apart, so steps are naturally paced — -// the damper docs/celld-behaviors.md calls settle time. +// Watch events can reconcile sooner than Requeue; the live gates below, +// rather than the requeue interval, decide whether another step is safe. func (r *WorkerAppReconciler) rolloutStep(ctx context.Context, app *platformv1alpha1.WorkerApp, sts *appsv1.StatefulSet) (fleetOutcome, error) { partition := stsPartition(sts) liveReplicas := ptr.Deref(sts.Spec.Replicas, 0) diff --git a/internal/controller/workerapp_controller.go b/internal/controller/workerapp_controller.go index cd3bc26..bae5959 100644 --- a/internal/controller/workerapp_controller.go +++ b/internal/controller/workerapp_controller.go @@ -135,6 +135,10 @@ func (r *WorkerAppReconciler) Reconcile(ctx context.Context, req ctrl.Request) ( return ctrl.Result{}, client.IgnoreNotFound(err) } + if !app.DeletionTimestamp.IsZero() { + return ctrl.Result{}, nil + } + var conditions []metav1.Condition // Foundation objects first: the StatefulSet references the service @@ -178,7 +182,7 @@ func (r *WorkerAppReconciler) Reconcile(ctx context.Context, req ctrl.Request) ( } if outcome.WaitingOn != "" { - log.Info("fleet reconciled", "phase", outcome.Phase, "waitingOn", outcome.WaitingOn) + log.Info("Reconciled fleet", "phase", outcome.Phase, "waitingOn", outcome.WaitingOn) } return ctrl.Result{RequeueAfter: outcome.Requeue}, nil } @@ -186,7 +190,17 @@ func (r *WorkerAppReconciler) Reconcile(ctx context.Context, req ctrl.Request) ( func (r *WorkerAppReconciler) ensureFoundation(ctx context.Context, app *platformv1alpha1.WorkerApp, conditions *[]metav1.Condition) error { sa := buildServiceAccount(app) if err := r.ensureObject(ctx, app, sa, func(live, desired client.Object) { - live.SetAnnotations(desired.GetAnnotations()) + annotations := mergeStringMaps(live.GetAnnotations(), desired.GetAnnotations()) + // Preserve externally provisioned identity in auto mode and foreign metadata. + for _, key := range []string{"eks.amazonaws.com/role-arn", azureClientIDAnnotation} { + if _, wanted := desired.GetAnnotations()[key]; !wanted && app.Spec.Bucket.CredentialsFrom.IAMRole != iamRoleAuto { + delete(annotations, key) + } + } + if len(annotations) == 0 { + annotations = nil + } + live.SetAnnotations(annotations) }); err != nil { return fmt.Errorf("service account: %w", err) } @@ -219,6 +233,16 @@ func (r *WorkerAppReconciler) ensureFoundation(ctx context.Context, app *platfor if err := r.ensureObject(ctx, app, public, func(live, desired client.Object) { l, d := live.(*corev1.Service), desired.(*corev1.Service) l.Spec.Selector = d.Spec.Selector + // Preserve API-allocated node ports while the Service still uses them. + if d.Spec.Type == corev1.ServiceTypeNodePort || d.Spec.Type == corev1.ServiceTypeLoadBalancer { + for i := range d.Spec.Ports { + for _, port := range l.Spec.Ports { + if port.Name == d.Spec.Ports[i].Name { + d.Spec.Ports[i].NodePort = port.NodePort + } + } + } + } l.Spec.Ports = d.Spec.Ports l.Spec.Type = d.Spec.Type // Cloud LB configuration rides on annotations; merge so the cloud @@ -265,6 +289,9 @@ func (r *WorkerAppReconciler) ensureObject(ctx context.Context, app *platformv1a if err != nil { return err } + if err := requireOwnership(app, live); err != nil { + return err + } before := live.DeepCopyObject().(client.Object) mutate(live, desired) live.SetLabels(mergeStringMaps(live.GetLabels(), desired.GetLabels())) @@ -274,6 +301,14 @@ func (r *WorkerAppReconciler) ensureObject(ctx context.Context, app *platformv1a return r.Update(ctx, live) } +// requireOwnership prevents name collisions from modifying another workload. +func requireOwnership(app *platformv1alpha1.WorkerApp, obj client.Object) error { + if !metav1.IsControlledBy(obj, app) { + return fmt.Errorf("refusing to modify %T %s: not controlled by WorkerApp %s", obj, client.ObjectKeyFromObject(obj), client.ObjectKeyFromObject(app)) + } + return nil +} + // The ensure helpers return true when they hit a transient error worth a // fast requeue (anything but a missing CRD, which only changes when a // human installs something). @@ -422,8 +457,23 @@ func (r *WorkerAppReconciler) ensureScaledObject(ctx context.Context, app *platf obj.SetKind("ScaledObject") obj.SetName(fleetName(app)) obj.SetNamespace(app.Namespace) - if err := r.Delete(ctx, obj); err != nil && !apierrors.IsNotFound(err) && !meta.IsNoMatchError(err) { - logf.FromContext(ctx).Error(err, "deleting stale ScaledObject") + err := r.Get(ctx, client.ObjectKeyFromObject(obj), obj) + if apierrors.IsNotFound(err) || meta.IsNoMatchError(err) { + return false + } + if err == nil { + err = requireOwnership(app, obj) + } + if err == nil { + uid, rv := obj.GetUID(), obj.GetResourceVersion() + err = r.Delete(ctx, obj, client.Preconditions{UID: &uid, ResourceVersion: &rv}) + } + if err != nil && !apierrors.IsNotFound(err) { + *conditions = append(*conditions, metav1.Condition{ + Type: condAutoscalingReady, Status: metav1.ConditionFalse, + Reason: "ScaledObjectError", Message: err.Error(), + }) + return true } return false } @@ -466,6 +516,9 @@ func (r *WorkerAppReconciler) ensureUnstructured(ctx context.Context, app *platf if err != nil { return err } + if err := requireOwnership(app, live); err != nil { + return err + } // Merge, never replace, metadata — other controllers annotate and // label these objects (KEDA stamps ScaledObjects with a name label), // and wiping their keys puts both controllers in a conflict loop. @@ -485,6 +538,9 @@ func (r *WorkerAppReconciler) ensureUnstructured(ctx context.Context, app *platf // mergeStringMaps overlays desired onto live: our keys win, foreign keys // survive. func mergeStringMaps(live, desired map[string]string) map[string]string { + if len(live) == 0 && len(desired) == 0 { + return nil + } out := make(map[string]string, len(live)+len(desired)) maps.Copy(out, live) maps.Copy(out, desired) @@ -493,7 +549,7 @@ func mergeStringMaps(live, desired map[string]string) map[string]string { func stringMapSubset(sub, of map[string]string) bool { for k, v := range sub { - if of[k] != v { + if actual, ok := of[k]; !ok || actual != v { return false } } @@ -501,6 +557,7 @@ func stringMapSubset(sub, of map[string]string) bool { } func (r *WorkerAppReconciler) updateStatus(ctx context.Context, app *platformv1alpha1.WorkerApp, outcome fleetOutcome, appVersion string, conditions []metav1.Condition) error { + before := app.DeepCopy().Status app.Status.Phase = outcome.Phase app.Status.Rollout = platformv1alpha1.RolloutStatus{ Partition: outcome.Partition, @@ -550,6 +607,9 @@ func (r *WorkerAppReconciler) updateStatus(ctx context.Context, app *platformv1a cond.ObservedGeneration = app.Generation meta.SetStatusCondition(&app.Status.Conditions, cond) } + if apiequality.Semantic.DeepEqual(before, app.Status) { + return nil + } return r.Status().Update(ctx, app) } @@ -580,6 +640,7 @@ func (r *WorkerAppReconciler) SetupWithManager(mgr ctrl.Manager) error { Owns(&corev1.ServiceAccount{}). Owns(&networkingv1.NetworkPolicy{}). Owns(&policyv1.PodDisruptionBudget{}). + Owns(&networkingv1.Ingress{}). Named("workerapp"). Complete(r) } diff --git a/internal/controller/workerapp_controller_test.go b/internal/controller/workerapp_controller_test.go index 25cfa89..f8f9bcc 100644 --- a/internal/controller/workerapp_controller_test.go +++ b/internal/controller/workerapp_controller_test.go @@ -29,6 +29,7 @@ import ( "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/reconcile" platformv1alpha1 "github.com/ezgamehost/celld-operator/api/v1alpha1" @@ -109,10 +110,62 @@ var _ = Describe("WorkerApp Controller", func() { Expect(k8sClient.Delete(ctx, resource)).To(Succeed()) // envtest has no garbage collector; remove the owned fleet so // the next spec starts clean. - sts := &appsv1.StatefulSet{} - if err := k8sClient.Get(ctx, fleetKey, sts); err == nil { - Expect(k8sClient.Delete(ctx, sts)).To(Succeed()) + for _, obj := range []client.Object{ + &appsv1.StatefulSet{}, &corev1.Service{}, &corev1.ServiceAccount{}, + &networkingv1.NetworkPolicy{}, &policyv1.PodDisruptionBudget{}, + } { + if err := k8sClient.Get(ctx, fleetKey, obj); err == nil { + Expect(k8sClient.Delete(ctx, obj)).To(Succeed()) + } } + internal := &corev1.Service{} + if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: resourceNamespace, Name: resourceName + "-celld-internal"}, internal); err == nil { + Expect(k8sClient.Delete(ctx, internal)).To(Succeed()) + } + }) + + It("should leave a steady reconcile unchanged and preserve Service metadata", func() { + r := reconciler() + resource := &platformv1alpha1.WorkerApp{} + Expect(k8sClient.Get(ctx, typeNamespacedName, resource)).To(Succeed()) + resource.Spec.Service.Type = corev1.ServiceTypeNodePort + resource.Spec.Bucket.CredentialsFrom.IAMRole = iamRoleAuto + Expect(k8sClient.Update(ctx, resource)).To(Succeed()) + _, err := r.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) + Expect(err).NotTo(HaveOccurred()) + sa := &corev1.ServiceAccount{} + Expect(k8sClient.Get(ctx, fleetKey, sa)).To(Succeed()) + sa.Annotations = map[string]string{"eks.amazonaws.com/role-arn": "external-role", "example.com/owner": "platform"} + Expect(k8sClient.Update(ctx, sa)).To(Succeed()) + // The second pass observes the newly created StatefulSet. + _, err = r.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) + Expect(err).NotTo(HaveOccurred()) + Expect(k8sClient.Get(ctx, typeNamespacedName, resource)).To(Succeed()) + version := resource.ResourceVersion + svc := &corev1.Service{} + Expect(k8sClient.Get(ctx, fleetKey, svc)).To(Succeed()) + svcVersion, nodePort := svc.ResourceVersion, svc.Spec.Ports[0].NodePort + Expect(nodePort).NotTo(BeZero()) + _, err = r.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) + Expect(err).NotTo(HaveOccurred()) + Expect(k8sClient.Get(ctx, typeNamespacedName, resource)).To(Succeed()) + Expect(resource.ResourceVersion).To(Equal(version)) + Expect(k8sClient.Get(ctx, fleetKey, svc)).To(Succeed()) + Expect(svc.ResourceVersion).To(Equal(svcVersion)) + Expect(svc.Spec.Ports[0].NodePort).To(Equal(nodePort)) + Expect(k8sClient.Get(ctx, fleetKey, sa)).To(Succeed()) + Expect(sa.Annotations).To(HaveKeyWithValue("eks.amazonaws.com/role-arn", "external-role")) + Expect(sa.Annotations).To(HaveKeyWithValue("example.com/owner", "platform")) + }) + + It("should refuse to modify an unowned ServiceAccount", func() { + sa := &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: fleetKey.Name, Namespace: fleetKey.Namespace}} + Expect(k8sClient.Create(ctx, sa)).To(Succeed()) + version := sa.ResourceVersion + _, err := reconciler().Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) + Expect(err).To(MatchError(ContainSubstring("not controlled by WorkerApp"))) + Expect(k8sClient.Get(ctx, fleetKey, sa)).To(Succeed()) + Expect(sa.ResourceVersion).To(Equal(version)) }) It("should reconcile the full fleet", func() {