diff --git a/api/v1alpha1/conditions.go b/api/v1alpha1/conditions.go index 953ac72e..fe4f6ebc 100644 --- a/api/v1alpha1/conditions.go +++ b/api/v1alpha1/conditions.go @@ -11,6 +11,42 @@ const ( // ConditionProgressing is True while a rollout is actively in-flight — // i.e., the target version has not yet been promoted to current. ConditionProgressing = "Progressing" + + // ConditionStalled is True when reconciliation cannot progress and only a spec + // change can resolve it — an invalid spec, or a connection kind this controller + // cannot read. kstatus + // (https://github.com/kubernetes-sigs/cli-utils/tree/master/pkg/kstatus) reports + // Failed when it is True, so Argo Rollouts and Helm --wait abort instead of + // waiting out their timeout. + // + // kstatus's own convention is that such a condition is absent while things are + // normal, but it only ever tests for True, so this controller writes the + // condition on every path and sets it False when nothing is stalled. That reads + // identically to kstatus and keeps every condition the controller owns visible + // in kubectl describe, consistent with Ready and Progressing. + // + // It is set only for failures decidable from information already in hand. + // Failures that are waiting on another object to exist (a missing Connection or + // credential Secret) and transient infrastructure failures report Reconciling + // instead, because neither can be told apart from a normal few-second gap + // during a deploy. See stalledReasons in the controller package. + ConditionStalled = "Stalled" + + // ConditionReconciling is True while the controller is still working toward the + // spec, and False once it has caught up. kstatus reports InProgress when it is + // True, which is the path kstatus intends for custom resources. Without it, + // kstatus has to infer the same answer from Ready=False, a fallback its own + // documentation flags as unreliable. + // + // It is close to the inverse of Progressing but not identical: a transient + // blocking error sets Progressing=False (blocked) and Reconciling=True (still + // retrying), because those two vocabularies disagree about what a retry is. + // + // Reconciling and Stalled must never both be True on the same object. + // kstatus scans status.conditions in array order and returns on the first + // match, so the verdict would depend on insertion order. The controller + // writes both on every path and sets at most one of them to True. + ConditionReconciling = "Reconciling" ) // Deprecated condition type constants. Maintained for backward compatibility with diff --git a/api/v1alpha1/workerdeployment_types.go b/api/v1alpha1/workerdeployment_types.go index 2e48260c..82c83a1e 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/api/v1alpha1/workerdeployment_types.go @@ -121,6 +121,15 @@ const ( // when the target version has been successfully registered as the current version. ReasonRolloutComplete = "RolloutComplete" + // ReasonReconcileSucceeded is set on ConditionStalled=False after any reconcile + // that completed without a blocking error, whatever stage the rollout is at. + // It is what allows a WorkerDeployment that recovers from a blocking error to + // stop reporting Failed to kstatus consumers. + ReasonReconcileSucceeded = "ReconcileSucceeded" + + // ReasonWaitingForPollers is set on ConditionProgressing=True when the target + // version's Kubernetes Deployment has been created but the version is not yet + // registered with Temporal (workers have not started polling yet). // ReasonWaitingForPollers is set on ConditionProgressing=True when workers are // not yet (or are no longer) actively polling Temporal. This covers both: // (1) the target version's Kubernetes Deployment has been created but the diff --git a/api/v1alpha1/workerresourcetemplate_types.go b/api/v1alpha1/workerresourcetemplate_types.go index e41a6a75..f35f7e24 100644 --- a/api/v1alpha1/workerresourcetemplate_types.go +++ b/api/v1alpha1/workerresourcetemplate_types.go @@ -110,6 +110,16 @@ type WorkerResourceTemplateStatus struct { // +listType=map // +listMapKey=type Conditions []metav1.Condition `json:"conditions,omitempty"` + + // ObservedGeneration is the .metadata.generation the controller last + // reconciled. Compare against .metadata.generation to tell whether the + // controller has caught up with the latest spec change. + // + // This is the only generation field kstatus consults; the per-entry + // observedGeneration carried on each condition is not read by it. + // +optional + // +kubebuilder:validation:Minimum=0 + ObservedGeneration int64 `json:"observedGeneration,omitempty"` } //+kubebuilder:object:root=true diff --git a/docs/cd-rollouts.md b/docs/cd-rollouts.md index bbbf49ee..75ad7ae2 100644 --- a/docs/cd-rollouts.md +++ b/docs/cd-rollouts.md @@ -8,7 +8,11 @@ For migration help, see [migration-to-versioned.md](migration-to-versioned.md). ## Understanding the conditions -The `WorkerDeployment` resource exposes two standard conditions on `status.conditions` that CD tools and scripts can consume. +The `WorkerDeployment` resource exposes four standard conditions on `status.conditions` that CD tools and scripts can consume, in two pairs. + +`Ready` and `Progressing` describe the rollout in the controller's own terms. They are the ones to read in a script, a dashboard, or `kubectl describe`, and their `reason` fields carry the detail. + +`Stalled` and `Reconciling` say the same thing in the vocabulary [kstatus](https://github.com/kubernetes-sigs/cli-utils/tree/master/pkg/kstatus) understands, the library behind Helm 4 `--wait` and Flux health assessment. They follow kstatus's "abnormal-true" convention: each is present and `True` only while something unusual is happening, and absent otherwise. You rarely need to read them yourself; they exist so those tools reach the right verdict without a custom health check. ### `Ready` @@ -43,6 +47,34 @@ When `Progressing=False` due to an error, the `reason` field identifies what wen Once the underlying problem is fixed, the next successful reconcile will restore `Progressing` and `Ready` to the correct state. +### `Stalled` and `Reconciling` + +`Reconciling=True` means the controller is still working toward the spec. kstatus-based tools report the resource as **in progress** and keep waiting. `Stalled=True` means reconciliation cannot proceed and waiting will not help. The kstatus tools report **failed** and stop. Both are absent once a rollout is complete, and only one is ever set at a time. + +`Stalled` is set only for failures that are decidable from information already in hand, where nothing arriving later could change the answer: + +| Reason | Condition set | Why | +|---|---|---| +| `InvalidSpec` | `Stalled` | Settled by the spec you just applied | +| `ClusterConnectionUnsupported` | `Stalled` | Settled by the spec plus how the controller was deployed | +| `ConnectionNotFound` | `Reconciling` | Waiting on another object — the `Connection` may not exist *yet* | +| `AuthSecretInvalid` | `Reconciling` | Same, and this reason also covers a credential Secret that is simply absent | +| `TemporalClientCreationFailed` | `Reconciling` | Server unreachable; retried | +| `TemporalStateFetchFailed` | `Reconciling` | Includes rate limiting; retried | +| `PlanGenerationFailed`, `PlanExecutionFailed` | `Reconciling` | Retried with backoff | + +The reason a missing `Connection` is not treated as terminal is ordering. Applying a `WorkerDeployment` alongside its `Connection` and credentials in one release gives no guarantee about which lands first, so a missing reference is frequently a normal gap of a few seconds rather than a mistake. + +The trade-off is that an incorrect `connectionRef` or a `Connection` that was never created keeps reporting *in progress* until your tool's timeout expires rather than failing immediately. Set timeouts you are willing to wait out, and read the `reason` on `Ready`/`Progressing` (or the resource's Kubernetes Events) to see what is actually blocking. + +`WorkerResourceTemplate` follows the same pattern: a template that cannot render, or that the API server rejects outright, sets `Stalled`. A `WorkerResourceTemplate` waiting for its `WorkerDeployment` to appear, or retrying a transient apply failure, sets `Reconciling`. + +### `Connection` and `ClusterConnection` + +`Connection` and `ClusterConnection` are configuration-only resources. They have no controller of their own and expose no conditions, so tools that assess health from conditions like Helm `--wait`, Flux, and anything else built on [kstatus](https://github.com/kubernetes-sigs/cli-utils/tree/master/pkg/kstatus) treat them as healthy as soon as they exist. This is intentional as there is no reconcile loop behind them and therefore nothing to wait for. Kubernetes treats `ConfigMap` and `Secret` the same way. + +A broken connection is still reported, just on the `WorkerDeployment` that references it rather than on the connection itself (see the `ConnectionNotFound` and `AuthSecretInvalid` reasons above). Gate your rollouts on the `WorkerDeployment` as waiting on a `Connection` tells you only that the object was accepted by the API server, not that the credentials in it work. + ## Triggering a rollout A rollout starts when you change the pod template in your `WorkerDeployment` spec — a changed pod spec produces a new Build ID, which the controller treats as a new version to roll out. @@ -97,7 +129,7 @@ Set `--timeout` to exceed the longest expected rollout time — for progressive ### Helm 4 -Helm 4 uses [kstatus](https://github.com/kubernetes-sigs/cli-utils/tree/master/pkg/kstatus) for its `--wait` implementation ([HIP-0022](https://helm.sh/community/hips/hip-0022/)). kstatus understands the standard Kubernetes conditions contract and should block until `Ready=True` on your `WorkerDeployment`: +Helm 4 uses [kstatus](https://github.com/kubernetes-sigs/cli-utils/tree/master/pkg/kstatus) for its `--wait` implementation ([HIP-0022](https://helm.sh/community/hips/hip-0022/)). kstatus understands the standard Kubernetes conditions contract and should block until `Ready=True` on your `WorkerDeployment`. Because the controller also emits `Stalled` (see above), a rollout blocked by an invalid spec fails the release immediately instead of waiting out the timeout: ```bash helm upgrade my-worker ./chart --values values.yaml --wait --timeout 10m @@ -121,38 +153,42 @@ kubectl wait workerdeployment/my-worker \ ArgoCD does not have a generic fallback that automatically checks `status.conditions` on unknown CRD types. For any resource whose group (`temporal.io`) is not in ArgoCD's built-in health check registry, ArgoCD silently skips that resource when computing application health. A [custom Lua health check](https://argo-cd.readthedocs.io/en/stable/operator-manual/health/) is the standard mechanism for teaching ArgoCD how to assess a CRD's health. -The two standard conditions (`Ready`, `Progressing`) keep the Lua simple — it only needs to read the condition type and status, not any controller-specific status fields. The following script is a starting point; adapt it to your ArgoCD version and any site-specific requirements: +The standard conditions keep the Lua simple — it only needs to read condition types and statuses, not any controller-specific status fields. Reading `Stalled` and `Reconciling` rather than `Progressing` also makes ArgoCD agree with Helm and Flux about what counts as a failure, instead of showing **Degraded** for a `Connection` that is a second away from existing. The following script is a starting point; adapt it to your ArgoCD version and any site-specific requirements: ```yaml # In your argocd-cm ConfigMap data: resource.customizations.health.temporal.io_WorkerDeployment: | local ready = nil - local progressing = nil + local stalled = nil + local reconciling = nil if obj.status ~= nil and obj.status.conditions ~= nil then for _, c in ipairs(obj.status.conditions) do if c.type == "Ready" then ready = c end - if c.type == "Progressing" then progressing = c end + if c.type == "Stalled" then stalled = c end + if c.type == "Reconciling" then reconciling = c end end end + -- Check Stalled first: it is the only condition that means waiting will not help. + if stalled ~= nil and stalled.status == "True" then + return {status = "Degraded", message = stalled.message} + end if ready ~= nil and ready.status == "True" then return {status = "Healthy", message = ready.message} end - if progressing ~= nil then - if progressing.status == "True" then - return {status = "Progressing", message = progressing.message} - else - return {status = "Degraded", message = progressing.message} - end + if reconciling ~= nil and reconciling.status == "True" then + return {status = "Progressing", message = reconciling.message} end return {status = "Progressing", message = "Waiting for conditions"} ``` With a health check like this in place: +- ArgoCD shows **Degraded** when reconciliation is stalled (`Stalled=True`) — an invalid spec, or a connection kind this controller cannot read. - ArgoCD shows **Healthy** once `Ready=True`. -- ArgoCD shows **Progressing** while a rollout is in-flight (`Progressing=True`). -- ArgoCD shows **Degraded** when progress is blocked (`Progressing=False` with an error reason). +- ArgoCD shows **Progressing** while a rollout is in-flight, and also while the controller is retrying a recoverable problem such as a `Connection` that does not exist yet. Read the `reason` on `Ready` to tell those apart. + +The same script works for `WorkerResourceTemplate` — register it under `resource.customizations.health.temporal.io_WorkerResourceTemplate` as well, since it emits the same three conditions. If you use [sync waves](https://argo-cd.readthedocs.io/en/stable/user-guide/sync-waves/) and workers must be fully rolled out before a dependent service is updated, place the `WorkerDeployment` in an earlier wave. @@ -162,7 +198,7 @@ If you use [sync waves](https://argo-cd.readthedocs.io/en/stable/user-guide/sync ### Kustomization -Flux's `Kustomization` controller uses kstatus to assess resource health. Because `WorkerDeployment` emits a standard `Ready` condition, Flux should treat it as healthy when `Ready=True`. Adding an explicit `healthChecks` entry makes the dependency visible and ensures Flux waits on the `WorkerDeployment` before marking the Kustomization as ready: +Flux's `Kustomization` controller uses kstatus to assess resource health. Because `WorkerDeployment` emits the standard `Ready`, `Reconciling`, and `Stalled` conditions, Flux should treat it as healthy when `Ready=True`, keep waiting while `Reconciling=True`, and fail the health check rather than wait out the timeout when `Stalled=True`. Adding an explicit `healthChecks` entry makes the dependency visible and ensures Flux waits on the `WorkerDeployment` before marking the Kustomization as ready: ```yaml apiVersion: kustomize.toolkit.fluxcd.io/v1 diff --git a/docs/migration-crd-rename.md b/docs/migration-crd-rename.md index 44f4e92f..f2fd2f75 100644 --- a/docs/migration-crd-rename.md +++ b/docs/migration-crd-rename.md @@ -109,6 +109,8 @@ message: "Migration complete. Delete this TemporalWorkerDeployment." For `TemporalConnection`, the same `Deprecated` → `MigratedToConnection` pattern applies (there is no ownership transfer step, so there is no intermediate state). +> **CD health checks:** because these resources never report `Ready=True`, tools that assess health from conditions (Helm `--wait`, Flux, and anything else built on [kstatus](https://github.com/kubernetes-sigs/cli-utils/tree/master/pkg/kstatus)) treat an unmigrated `TemporalWorkerDeployment` or `TemporalConnection` as not-ready for as long as it exists — completing the migration is what resolves it. A resource already marked for deletion reports `Terminating` instead, so following the migration steps above does not leave a release waiting. + ## Deletion protection After upgrading to v1.7, the controller adds a `temporal.io/migration-guard` finalizer to every `TemporalWorkerDeployment` and `TemporalConnection`. This finalizer prevents the resource from being fully deleted until migration is confirmed: diff --git a/go.mod b/go.mod index c7652df1..28ebc354 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( k8s.io/apiextensions-apiserver v0.36.0 k8s.io/apimachinery v0.36.3 k8s.io/client-go v0.36.3 + sigs.k8s.io/cli-utils v0.37.2 k8s.io/utils v0.0.0-20260319190234-28399d86e0b5 sigs.k8s.io/controller-runtime v0.24.0 sigs.k8s.io/yaml v1.6.0 @@ -80,7 +81,6 @@ require ( github.com/emirpasic/gods v1.18.1 // indirect github.com/envoyproxy/go-control-plane/envoy v1.37.0 // indirect github.com/envoyproxy/protoc-gen-validate v1.3.3 // indirect - github.com/evanphx/json-patch v5.6.0+incompatible // indirect github.com/evanphx/json-patch/v5 v5.9.11 // indirect github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect github.com/felixge/httpsnoop v1.0.4 // indirect diff --git a/go.sum b/go.sum index 05d8153c..c1aca3de 100644 --- a/go.sum +++ b/go.sum @@ -702,6 +702,8 @@ modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= +sigs.k8s.io/cli-utils v0.37.2 h1:GOfKw5RV2HDQZDJlru5KkfLO1tbxqMoyn1IYUxqBpNg= +sigs.k8s.io/cli-utils v0.37.2/go.mod h1:V+IZZr4UoGj7gMJXklWBg6t5xbdThFBcpj4MrZuCYco= sigs.k8s.io/controller-runtime v0.24.0 h1:Ck6N2LdS8Lovy1o25BB4r1xjvLEKUl1s2o9kU+KWDE4= sigs.k8s.io/controller-runtime v0.24.0/go.mod h1:vFkfY5fGt5xAC/sKb8IBFKgWPNKG9OUG29dR8Y2wImw= sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg= diff --git a/helm/temporal-worker-controller-crds/templates/temporal.io_workerresourcetemplates.yaml b/helm/temporal-worker-controller-crds/templates/temporal.io_workerresourcetemplates.yaml index 7c606c88..07118280 100644 --- a/helm/temporal-worker-controller-crds/templates/temporal.io_workerresourcetemplates.yaml +++ b/helm/temporal-worker-controller-crds/templates/temporal.io_workerresourcetemplates.yaml @@ -111,6 +111,10 @@ spec: x-kubernetes-list-map-keys: - type x-kubernetes-list-type: map + observedGeneration: + format: int64 + minimum: 0 + type: integer versions: items: properties: diff --git a/internal/controller/execplan.go b/internal/controller/execplan.go index 0dac93d9..43b6a10d 100644 --- a/internal/controller/execplan.go +++ b/internal/controller/execplan.go @@ -27,6 +27,7 @@ import ( appsv1 "k8s.io/api/apps/v1" autoscalingv1 "k8s.io/api/autoscaling/v1" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" @@ -445,6 +446,10 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( hash string // rendered hash recorded on successful apply; "" on error err error skipped bool // true if the apply was skipped because the rendered hash is unchanged + // renderFailed distinguishes a spec.template render failure from an SSA apply + // failure. Render failures are always terminal as only a spec change can fix + // them while apply failures are classified by API error kind. + renderFailed bool } wrtResults := make(map[wrtKey][]applyResult) @@ -458,8 +463,9 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( "buildID", apply.BuildID, ) wrtResults[key] = append(wrtResults[key], applyResult{ - buildID: apply.BuildID, - err: apply.RenderError, + buildID: apply.BuildID, + err: apply.RenderError, + renderFailed: true, }) continue } @@ -564,6 +570,36 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( } } if allSkipped && len(deleted) == 0 { + // Every apply was a no-op and nothing was deleted, so the per-Build-ID + // status and conditions are already correct. The one thing that can still + // be stale is status.observedGeneration. + // + // metadata.generation tracks any semantic change to the spec, while the + // skip decision above is made on a hash of the rendered output. Those two + // can come apart. Switching spec.temporalWorkerDeploymentRef to + // spec.workerDeploymentRef with the same name is such a case: the webhook + // permits it (only the effective name is immutable) and it is step 4 of the + // CRD rename migration, but rendering does not depend on which ref field + // was used, so the hash is unchanged, every apply is skipped and no status + // write happens. A stale observedGeneration would make kstatus report + // InProgress forever. + // + // The Get is served from the informer cache, and the write still only + // happens when something has actually changed, so this branch's + // optimisation is preserved. + wrt := &temporaliov1alpha1.WorkerResourceTemplate{} + if err := r.Get(ctx, types.NamespacedName{Namespace: key.namespace, Name: key.name}, wrt); err != nil { + if !apierrors.IsNotFound(err) { + statusErrs = append(statusErrs, fmt.Errorf("get WRT %s/%s to refresh observedGeneration: %w", key.namespace, key.name, err)) + } + continue + } + if wrt.Status.ObservedGeneration != wrt.Generation { + wrt.Status.ObservedGeneration = wrt.Generation + if err := r.Status().Update(ctx, wrt); err != nil { + statusErrs = append(statusErrs, fmt.Errorf("refresh observedGeneration for WRT %s/%s: %w", key.namespace, key.name, err)) + } + } continue } @@ -585,6 +621,7 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( versions := make([]temporaliov1alpha1.WorkerResourceTemplateVersionStatus, 0, len(results)) coveredByApply := make(map[string]struct{}, len(results)) anyFailed := false + anyTerminal := false for _, result := range results { coveredByApply[result.buildID] = struct{}{} if result.skipped { @@ -599,6 +636,9 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( applyErrs = append(applyErrs, result.err) applyErr = result.err.Error() anyFailed = true + if isTerminalWorkerResourceError(result.err, result.renderFailed) { + anyTerminal = true + } // 0 means "unset" / "not yet successfully applied at current generation". // Failure Message and LastTransitionTime are still recorded below. appliedGeneration = 0 @@ -651,6 +691,43 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( ObservedGeneration: wrt.Generation, }) + // Translate the same outcome into the kstatus abnormal-true conditions, so + // Argo Rollouts and Helm --wait can tell a template that will never apply from + // one that is still being retried. Ready=False alone reads as InProgress to + // kstatus, which is why a bad template used to hang a deploy until timeout. + // + // Both are always written, and at most one of them is True. kstatus scans + // status.conditions in array order and returns on the first match, so an object + // carrying both as True would get a verdict decided by insertion order. + // Writing the inactive one as False rather than removing it keeps every + // condition this controller owns present on every object. + stalledStatus, reconcilingStatus := metav1.ConditionFalse, metav1.ConditionFalse + switch { + case anyTerminal: + stalledStatus = metav1.ConditionTrue + case anyFailed: + reconcilingStatus = metav1.ConditionTrue + } + apimeta.SetStatusCondition(&wrt.Status.Conditions, metav1.Condition{ + Type: temporaliov1alpha1.ConditionStalled, + Status: stalledStatus, + Reason: condReason, + Message: condMessage, + ObservedGeneration: wrt.Generation, + }) + apimeta.SetStatusCondition(&wrt.Status.Conditions, metav1.Condition{ + Type: temporaliov1alpha1.ConditionReconciling, + Status: reconcilingStatus, + Reason: condReason, + Message: condMessage, + ObservedGeneration: wrt.Generation, + }) + + // Record that this generation was processed, whatever the outcome. kstatus + // checks this before it looks at any condition, so leaving it behind would + // mask both of the conditions set above. + wrt.Status.ObservedGeneration = wrt.Generation + // Sort the versions by BuildID for deterministic status output. slices.SortFunc(versions, func(a, b temporaliov1alpha1.WorkerResourceTemplateVersionStatus) int { return strings.Compare(a.BuildID, b.BuildID) @@ -664,6 +741,35 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( return errors.Join(append(applyErrs, statusErrs...)...) } +// isTerminalWorkerResourceError reports whether a WorkerResourceTemplate failure can +// only be resolved by a human changing something, and so should be surfaced through the +// kstatus Stalled condition (making kstatus report Failed) rather than left looking +// like work still in progress. +// +// renderFailed covers a spec.template that could not be rendered at all. This is always +// terminal, since only a spec change can fix it. For SSA apply failures only the API +// server's own outright rejections count: Invalid (the rendered object does not satisfy +// the target schema), Forbidden and Unauthorized (the controller lacks RBAC for the +// templated kind), BadRequest, and the media-type/method rejections. +// +// Everything else (Conflict, timeouts, TooManyRequests, transport errors) is retried +// on the next reconcile and deliberately keeps reporting InProgress, so a blip cannot +// abort a deploy. This mirrors the stalledReasons split in worker_controller.go. +func isTerminalWorkerResourceError(err error, renderFailed bool) bool { + if err == nil { + return false + } + if renderFailed { + return true + } + return apierrors.IsInvalid(err) || + apierrors.IsForbidden(err) || + apierrors.IsUnauthorized(err) || + apierrors.IsBadRequest(err) || + apierrors.IsUnsupportedMediaType(err) || + apierrors.IsMethodNotSupported(err) +} + // deleteDrainedVersions prunes the Temporal server-side Worker Deployment Version // record for each k8s Deployment in DeleteDeployments, before executeK8sOperations // deletes them. It is mutated to remove the k8s Deployments that should not be diff --git a/internal/controller/kstatus_test.go b/internal/controller/kstatus_test.go new file mode 100644 index 00000000..c31a5564 --- /dev/null +++ b/internal/controller/kstatus_test.go @@ -0,0 +1,614 @@ +// Unless explicitly stated otherwise all files in this repository are licensed under the MIT License. +// +// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2024 Datadog, Inc. + +package controller + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + kstatus "sigs.k8s.io/cli-utils/pkg/kstatus/status" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// computeKstatus feeds obj through the real kstatus decision tree +// (sigs.k8s.io/cli-utils) and returns its verdict. This is the same code path +// Argo Rollouts and Helm --wait use to decide whether a custom resource is +// healthy, so whatever this returns is what those tools will conclude. +// +// The conversion to unstructured is not test scaffolding: kstatus is a generic +// library that has never heard of our types, so in production it reads our +// objects as untyped JSON off the wire. Converting here reproduces that. +func computeKstatus(t *testing.T, obj runtime.Object) *kstatus.Result { + t.Helper() + + content, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj) + require.NoError(t, err, "convert typed object to unstructured") + + res, err := kstatus.Compute(&unstructured.Unstructured{Object: content}) + require.NoError(t, err, "kstatus.Compute") + require.NotNil(t, res, "kstatus.Compute returned a nil result") + + return res +} + +// TestKstatusBaseline_WorkerDeployment records the verdict kstatus reaches for +// each rollout state today. It is a characterization test: `today` is what the +// current code produces, `desired` is what it should produce once issue #478 is +// resolved. Where they differ the test still passes and logs a KNOWN GAP, so +// this file is green on main and fails loudly when the behavior is fixed — +// at which point update `today` in the same commit as the fix. +func TestKstatusBaseline_WorkerDeployment(t *testing.T) { + ctx := context.Background() + + cases := []struct { + name string + build func() *temporaliov1alpha1.WorkerDeployment + today kstatus.Status + desired kstatus.Status + }{ + { + // Pods created, no worker has polled Temporal yet. + name: "TargetNotRegistered", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + wd.Status.ObservedGeneration = wd.Generation + wd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusNotRegistered + r.syncConditions(wd) + return wd + }, + today: kstatus.InProgressStatus, + desired: kstatus.InProgressStatus, + }, + { + // Registered with Temporal, receiving no traffic, awaiting promotion. + name: "TargetInactive", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + wd.Status.ObservedGeneration = wd.Generation + wd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusInactive + r.syncConditions(wd) + return wd + }, + today: kstatus.InProgressStatus, + desired: kstatus.InProgressStatus, + }, + { + // Mid-canary: receiving a percentage of new workflows. + name: "TargetRamping", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + wd.Status.ObservedGeneration = wd.Generation + wd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusRamping + r.syncConditions(wd) + return wd + }, + today: kstatus.InProgressStatus, + desired: kstatus.InProgressStatus, + }, + { + // The finish line. This is the case that makes `helm upgrade --wait` + // return successfully today; do not break it. + name: "TargetIsCurrent", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + wd.Status.ObservedGeneration = wd.Generation + wd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusCurrent + r.syncConditions(wd) + return wd + }, + today: kstatus.CurrentStatus, + desired: kstatus.CurrentStatus, + }, + { + // Waiting on another object to exist. Applying a WorkerDeployment + // alongside its Connection gives no ordering guarantee, so this can be a + // normal few-second gap rather than a mistake — reporting Failed would + // abort a deploy that was about to succeed. Stays InProgress, which is + // also what the controller did before Stalled existed. + name: "BlockedOnMissingConnection", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + r.recordWarningAndSetBlocked(ctx, wd, + temporaliov1alpha1.ReasonConnectionNotFound, + `Connection "conn" not found`, + `Connection "conn" not found`) + return wd + }, + today: kstatus.InProgressStatus, + desired: kstatus.InProgressStatus, + }, + { + // Same shape as the case above: ReasonAuthSecretInvalid also fires when + // the credential Secret is merely absent, which a deploy can resolve on + // its own moments later. + name: "BlockedOnMissingCredentials", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + r.recordWarningAndSetBlocked(ctx, wd, + temporaliov1alpha1.ReasonAuthSecretInvalid, + "Unable to resolve auth secret", + "Unable to resolve auth secret") + return wd + }, + today: kstatus.InProgressStatus, + desired: kstatus.InProgressStatus, + }, + { + // A previously-healthy WorkerDeployment that just got an invalid spec: + // generation advanced to 2 while observedGeneration was still 1. Uses a + // terminal reason deliberately — kstatus returns at its generation check + // before reading any condition, so only a case that should reach Failed + // can prove recordWarningAndSetBlocked advances observedGeneration. + name: "BlockedAfterBadSpecEdit", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + wd.Generation = 2 + wd.Status.ObservedGeneration = 1 + r.recordWarningAndSetBlocked(ctx, wd, + temporaliov1alpha1.ReasonInvalidSpec, + "Invalid WorkerDeployment spec: rampPercentage must increase between each step", + "rampPercentage must increase between each step") + return wd + }, + today: kstatus.FailedStatus, + desired: kstatus.FailedStatus, + }, + { + // Transient failures must NOT report Failed: the controller is still + // retrying (the ResourceExhausted paths requeue after 30s), and aborting + // a deploy because Temporal was briefly rate limited would be worse than + // waiting. ReasonTemporalStateFetchFailed is absent from stalledReasons, + // so no Stalled condition is set and kstatus stays on InProgress. + name: "BlockedOnTransientTemporalError", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + r.recordWarningAndSetBlocked(ctx, wd, + temporaliov1alpha1.ReasonTemporalStateFetchFailed, + "Got ResourceExhausted error fetching worker deployment state", + "Got ResourceExhausted error fetching worker deployment state") + return wd + }, + today: kstatus.InProgressStatus, + desired: kstatus.InProgressStatus, + }, + { + // Recovery: a WorkerDeployment that was blocked and then reconciled + // successfully. meta.SetStatusCondition only upserts, so unless a later + // successful reconcile writes Stalled=False, a Stalled=True would be + // permanent and kstatus would report Failed forever. Note this case uses a + // transient reason, which never sets Stalled at all — the terminal path is + // covered by AbnormalConditionsResolvedOnRecovery. + name: "RecoveredAfterBlocked", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + r.recordWarningAndSetBlocked(ctx, wd, + temporaliov1alpha1.ReasonConnectionNotFound, + `Connection "conn" not found`, + `Connection "conn" not found`) + // The user creates the Connection; the next reconcile succeeds. + wd.Status.ObservedGeneration = wd.Generation + wd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusCurrent + r.syncConditions(wd) + return wd + }, + today: kstatus.CurrentStatus, + desired: kstatus.CurrentStatus, + }, + { + // kstatus checks metadata.deletionTimestamp first and ignores + // everything else, so a Ready=True object still reports Terminating. + // This one is already correct with no work from us; pinned so nobody + // "fixes" it later with a condition that fights it. + name: "BeingDeleted", + build: func() *temporaliov1alpha1.WorkerDeployment { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + wd.DeletionTimestamp = &metav1.Time{Time: time.Now()} + wd.Finalizers = []string{finalizerName} + wd.Status.ObservedGeneration = wd.Generation + wd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusCurrent + r.syncConditions(wd) + return wd + }, + today: kstatus.TerminatingStatus, + desired: kstatus.TerminatingStatus, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + res := computeKstatus(t, tc.build()) + + assert.Equal(t, tc.today, res.Status, + "kstatus verdict for %s (message: %q)", tc.name, res.Message) + + if tc.today != tc.desired { + t.Logf("KNOWN GAP (issue #478): kstatus reports %s; should report %s. message=%q", + tc.today, tc.desired, res.Message) + } + }) + } +} + +// TestConditionsAreKstatusCompatible asserts the condition-level invariants the +// verdict table above cannot see. Chiefly: Reconciling and Stalled are never both +// True. kstatus scans status.conditions in array order and returns on the first +// match, so an object carrying both would get a verdict decided by insertion order. +// +// It also pins which route each state takes through kstatus. That cannot be checked +// from the verdict: reaching InProgress via Reconciling=True and via the Ready=False +// fallback produce a byte-identical kstatus.Result, so the assertion has to be made +// against the object's own conditions. +func TestConditionsAreKstatusCompatible(t *testing.T) { + ctx := context.Background() + + isTrue := func(wd *temporaliov1alpha1.WorkerDeployment, condType string) bool { + return apimeta.IsStatusConditionTrue(wd.Status.Conditions, condType) + } + + for _, st := range []temporaliov1alpha1.VersionStatus{ + temporaliov1alpha1.VersionStatusNotRegistered, + temporaliov1alpha1.VersionStatusInactive, + temporaliov1alpha1.VersionStatusRamping, + temporaliov1alpha1.VersionStatusCurrent, + } { + t.Run("Rollout"+string(st), func(t *testing.T) { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + wd.Status.ObservedGeneration = wd.Generation + wd.Status.TargetVersion.Status = st + r.syncConditions(wd) + + assert.False(t, isTrue(wd, temporaliov1alpha1.ConditionStalled), + "a successful reconcile must never leave Stalled=True") + + inFlight := st != temporaliov1alpha1.VersionStatusCurrent + assert.Equal(t, inFlight, isTrue(wd, temporaliov1alpha1.ConditionReconciling), + "Reconciling should be True exactly while the rollout is in flight") + assert.Equal(t, + isTrue(wd, temporaliov1alpha1.ConditionProgressing), + isTrue(wd, temporaliov1alpha1.ConditionReconciling), + "on the success path Reconciling must mirror Progressing") + }) + } + + // On the blocked path Progressing=False means "blocked" in the pre-kstatus + // vocabulary, so Reconciling deliberately does NOT mirror it: a transient + // failure is still being retried and must read as InProgress, not Failed. + for _, tc := range []struct { + name string + reason string + wantStalled bool + }{ + {"TerminalInvalidSpec", temporaliov1alpha1.ReasonInvalidSpec, true}, + {"TerminalClusterConnectionUnsupported", temporaliov1alpha1.ReasonClusterConnectionUnsupported, true}, + // Waiting on another object to exist: never terminal, because the deploy that + // creates it may simply not have got there yet. + {"WaitingOnConnection", temporaliov1alpha1.ReasonConnectionNotFound, false}, + {"WaitingOnCredentials", temporaliov1alpha1.ReasonAuthSecretInvalid, false}, + // Transient infrastructure failure: retried with backoff. + {"TransientTemporalError", temporaliov1alpha1.ReasonTemporalStateFetchFailed, false}, + } { + t.Run("Blocked"+tc.name, func(t *testing.T) { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + r.recordWarningAndSetBlocked(ctx, wd, tc.reason, "boom", "boom") + + assert.Equal(t, tc.wantStalled, isTrue(wd, temporaliov1alpha1.ConditionStalled), + "Stalled for reason %s", tc.reason) + assert.Equal(t, !tc.wantStalled, isTrue(wd, temporaliov1alpha1.ConditionReconciling), + "Reconciling for reason %s", tc.reason) + assert.False(t, + isTrue(wd, temporaliov1alpha1.ConditionStalled) && isTrue(wd, temporaliov1alpha1.ConditionReconciling), + "Stalled and Reconciling must never both be True") + assert.Equal(t, wd.Generation, wd.Status.ObservedGeneration, + "a blocked reconcile must still record the generation it observed") + }) + } + + // Recovery is the case that would break if nothing ever cleared Stalled: + // meta.SetStatusCondition only upserts, so a Stalled=True written during a + // blocking error would otherwise be permanent and kstatus would keep reporting + // Failed on a WorkerDeployment that is now healthy. + t.Run("AbnormalConditionsResolvedOnRecovery", func(t *testing.T) { + r, _ := newTestReconciler(nil) + wd := makeWD("wd", "default", "conn") + r.recordWarningAndSetBlocked(ctx, wd, temporaliov1alpha1.ReasonInvalidSpec, "boom", "boom") + require.True(t, isTrue(wd, temporaliov1alpha1.ConditionStalled), "precondition: Stalled was set") + + // The user fixes the spec; the next reconcile succeeds and completes. + wd.Status.ObservedGeneration = wd.Generation + wd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusCurrent + r.syncConditions(wd) + + // Both are set to False rather than removed, matching how every other + // condition syncConditions writes is handled. kstatus only ever tests for + // True, so present-and-False reads the same to it as absent. + stalled := apimeta.FindStatusCondition(wd.Status.Conditions, temporaliov1alpha1.ConditionStalled) + require.NotNil(t, stalled, "Stalled should still be reported, as False") + assert.Equal(t, metav1.ConditionFalse, stalled.Status) + assert.Equal(t, temporaliov1alpha1.ReasonReconcileSucceeded, stalled.Reason) + + reconciling := apimeta.FindStatusCondition(wd.Status.Conditions, temporaliov1alpha1.ConditionReconciling) + require.NotNil(t, reconciling, "Reconciling should still be reported, as False") + assert.Equal(t, metav1.ConditionFalse, reconciling.Status) + assert.Equal(t, temporaliov1alpha1.ReasonRolloutComplete, reconciling.Reason) + }) +} + +// TestKstatusBaseline_WorkerResourceTemplate covers the second CRD with +// conditions. WorkerResourceTemplateStatus has no top-level observedGeneration +// field (api/v1alpha1/workerresourcetemplate_types.go:100), so kstatus's +// generation check is always skipped for these objects. The per-condition +// ObservedGeneration set at execplan.go:649 does not help: kstatus reads only +// the top-level status.observedGeneration. +func TestKstatusBaseline_WorkerResourceTemplate(t *testing.T) { + newWRT := func() *temporaliov1alpha1.WorkerResourceTemplate { + return &temporaliov1alpha1.WorkerResourceTemplate{ + TypeMeta: metav1.TypeMeta{ + APIVersion: temporaliov1alpha1.GroupVersion.String(), + Kind: "WorkerResourceTemplate", + }, + ObjectMeta: metav1.ObjectMeta{Name: "wrt", Namespace: "default", Generation: 1}, + } + } + // setCond mirrors what the apply loop in execplan.go writes. + setCond := func(wrt *temporaliov1alpha1.WorkerResourceTemplate, condType string, st metav1.ConditionStatus, reason string) { + apimeta.SetStatusCondition(&wrt.Status.Conditions, metav1.Condition{ + Type: condType, + Status: st, + Reason: reason, + ObservedGeneration: wrt.Generation, + }) + } + + t.Run("AllVersionsApplied", func(t *testing.T) { + wrt := newWRT() + wrt.Status.ObservedGeneration = wrt.Generation + setCond(wrt, temporaliov1alpha1.ConditionReady, metav1.ConditionTrue, temporaliov1alpha1.ReasonWRTAllVersionsApplied) + + res := computeKstatus(t, wrt) + assert.Equal(t, kstatus.CurrentStatus, res.Status, "message: %q", res.Message) + }) + + t.Run("ApplyFailedTerminal", func(t *testing.T) { + // A render failure or an API rejection (Invalid, Forbidden): only a spec or + // RBAC change can fix it, so it must not read as work in progress. + wrt := newWRT() + wrt.Status.ObservedGeneration = wrt.Generation + setCond(wrt, temporaliov1alpha1.ConditionReady, metav1.ConditionFalse, temporaliov1alpha1.ReasonWRTApplyFailed) + setCond(wrt, temporaliov1alpha1.ConditionStalled, metav1.ConditionTrue, temporaliov1alpha1.ReasonWRTApplyFailed) + + res := computeKstatus(t, wrt) + assert.Equal(t, kstatus.FailedStatus, res.Status, "message: %q", res.Message) + }) + + t.Run("ApplyFailedTransient", func(t *testing.T) { + // Conflict, timeout, TooManyRequests: retried next reconcile, so InProgress. + wrt := newWRT() + wrt.Status.ObservedGeneration = wrt.Generation + setCond(wrt, temporaliov1alpha1.ConditionReady, metav1.ConditionFalse, temporaliov1alpha1.ReasonWRTApplyFailed) + setCond(wrt, temporaliov1alpha1.ConditionReconciling, metav1.ConditionTrue, temporaliov1alpha1.ReasonWRTApplyFailed) + + res := computeKstatus(t, wrt) + assert.Equal(t, kstatus.InProgressStatus, res.Status, "message: %q", res.Message) + }) + + t.Run("SpecEditNotYetObserved", func(t *testing.T) { + // Now reachable for WRTs at all, because the type finally has a top-level + // observedGeneration for kstatus to compare against. + wrt := newWRT() + wrt.Generation = 2 + wrt.Status.ObservedGeneration = 1 + setCond(wrt, temporaliov1alpha1.ConditionReady, metav1.ConditionTrue, temporaliov1alpha1.ReasonWRTAllVersionsApplied) + + res := computeKstatus(t, wrt) + assert.Equal(t, kstatus.InProgressStatus, res.Status) + assert.Contains(t, res.Message, "latest observed generation is 1") + }) + + // Unlike the cases above, this one drives the real writer. + t.Run("WorkerDeploymentNotFound", func(t *testing.T) { + wrt := newWRT() + wrt.Spec.WorkerDeploymentRef = &temporaliov1alpha1.WorkerDeploymentReference{Name: "missing-wd"} + r, _ := newTestReconciler([]client.Object{wrt}) + + require.NoError(t, r.markWRTsWDNotFound(context.Background(), + types.NamespacedName{Name: "missing-wd", Namespace: "default"})) + + var got temporaliov1alpha1.WorkerResourceTemplate + require.NoError(t, r.Get(context.Background(), + types.NamespacedName{Name: "wrt", Namespace: "default"}, &got)) + got.TypeMeta = newWRT().TypeMeta // the fake client strips TypeMeta on Get + + assert.True(t, apimeta.IsStatusConditionTrue(got.Status.Conditions, temporaliov1alpha1.ConditionReconciling), + "a WRT waiting for its WorkerDeployment is still reconciling") + assert.True(t, apimeta.IsStatusConditionFalse(got.Status.Conditions, temporaliov1alpha1.ConditionStalled), + "creation ordering is expected and self-resolving, so Stalled must be False") + assert.Equal(t, got.Generation, got.Status.ObservedGeneration) + + res := computeKstatus(t, &got) + assert.Equal(t, kstatus.InProgressStatus, res.Status, "message: %q", res.Message) + }) +} + +// TestKstatusBaseline_RemainingKinds covers the four CRD kinds that deliberately do +// not participate in the kstatus conditions contract, so that "every Temporal custom +// resource type" is accounted for rather than merely untested. +// +// Neither group is an oversight, but neither is self-evident from the code either, so +// each verdict is pinned here: if someone later adds a Ready condition to Connection, +// or makes a migration stub report ready, a test fails and points at this comment. +func TestKstatusBaseline_RemainingKinds(t *testing.T) { + ctx := context.Background() + req := func(name, namespace string) ctrl.Request { + return ctrl.Request{NamespacedName: types.NamespacedName{Name: name, Namespace: namespace}} + } + + // Connection and ClusterConnection are configuration only: no controller, no + // conditions, and no observedGeneration. kstatus finds nothing to assess and falls + // through to "current", which is how it treats ConfigMap and Secret. Connection + // problems surface on the referencing WorkerDeployment instead — see the + // ConnectionNotFound and AuthSecretInvalid cases above. + t.Run("Connection", func(t *testing.T) { + conn := &temporaliov1alpha1.Connection{ + TypeMeta: metav1.TypeMeta{ + APIVersion: temporaliov1alpha1.GroupVersion.String(), + Kind: "Connection", + }, + ObjectMeta: metav1.ObjectMeta{Name: "conn", Namespace: "default", Generation: 1}, + Spec: temporaliov1alpha1.ConnectionSpec{HostPort: "temporal.example.com:7233"}, + } + // ConnectionStatus has no Conditions field at all, so "carries no conditions" + // is enforced by the compiler rather than asserted here. + res := computeKstatus(t, conn) + assert.Equal(t, kstatus.CurrentStatus, res.Status, "message: %q", res.Message) + }) + + t.Run("ClusterConnection", func(t *testing.T) { + cc := &temporaliov1alpha1.ClusterConnection{ + TypeMeta: metav1.TypeMeta{ + APIVersion: temporaliov1alpha1.GroupVersion.String(), + Kind: "ClusterConnection", + }, + ObjectMeta: metav1.ObjectMeta{Name: "cluster-conn", Generation: 1}, + Spec: temporaliov1alpha1.ConnectionSpec{HostPort: "temporal.example.com:7233"}, + } + res := computeKstatus(t, cc) + assert.Equal(t, kstatus.CurrentStatus, res.Status, "message: %q", res.Message) + }) + + // The deprecated kinds are migration stubs: they are never reconciled against + // Temporal and never report Ready=True, which docs/migration-crd-rename.md states + // as intended behaviour. kstatus therefore reads them as InProgress for as long as + // they exist. That is acceptable because the documented migration replaces and + // deletes them in one step, and a resource with a deletionTimestamp reports + // Terminating instead (kstatus checks that first) — so a release following the + // guide is never left waiting. They are deliberately left alone rather than given + // Stalled/Reconciling: adding behaviour to a type scheduled for removal creates a + // contract someone can depend on. + t.Run("TemporalWorkerDeployment", func(t *testing.T) { + twd := makeTWDStub("my-worker", "default", nil) + r := newDeprecatedTWDReconciler(twd) + + // First reconcile adds the finalizer; the second writes the condition. + _, err := r.Reconcile(ctx, req("my-worker", "default")) + require.NoError(t, err) + _, err = r.Reconcile(ctx, req("my-worker", "default")) + require.NoError(t, err) + + var got temporaliov1alpha1.TemporalWorkerDeployment + require.NoError(t, r.Get(ctx, req("my-worker", "default").NamespacedName, &got)) + got.TypeMeta = metav1.TypeMeta{ + APIVersion: temporaliov1alpha1.GroupVersion.String(), + Kind: "TemporalWorkerDeployment", + } + + require.False(t, apimeta.IsStatusConditionTrue(got.Status.Conditions, temporaliov1alpha1.ConditionReady), + "a migration stub must never report Ready=True") + res := computeKstatus(t, &got) + assert.Equal(t, kstatus.InProgressStatus, res.Status, "message: %q", res.Message) + }) + + t.Run("TemporalConnection", func(t *testing.T) { + tc := makeTCStub("my-conn", "default") + r := newDeprecatedTCReconciler(tc) + + _, err := r.Reconcile(ctx, req("my-conn", "default")) + require.NoError(t, err) + _, err = r.Reconcile(ctx, req("my-conn", "default")) + require.NoError(t, err) + + var got temporaliov1alpha1.TemporalConnection + require.NoError(t, r.Get(ctx, req("my-conn", "default").NamespacedName, &got)) + got.TypeMeta = metav1.TypeMeta{ + APIVersion: temporaliov1alpha1.GroupVersion.String(), + Kind: "TemporalConnection", + } + + require.False(t, apimeta.IsStatusConditionTrue(got.Status.Conditions, temporaliov1alpha1.ConditionReady), + "a migration stub must never report Ready=True") + res := computeKstatus(t, &got) + assert.Equal(t, kstatus.InProgressStatus, res.Status, "message: %q", res.Message) + }) + + // A stub partway through the documented migration carries a deletionTimestamp, + // which kstatus checks before anything else. This is the case that keeps the + // InProgress verdicts above from stalling a real release. + t.Run("TemporalWorkerDeploymentBeingDeleted", func(t *testing.T) { + twd := makeTWDStub("my-worker", "default", nil) + twd.DeletionTimestamp = &metav1.Time{Time: time.Now()} + twd.Finalizers = []string{deprecatedMigrationFinalizer} + twd.TypeMeta = metav1.TypeMeta{ + APIVersion: temporaliov1alpha1.GroupVersion.String(), + Kind: "TemporalWorkerDeployment", + } + apimeta.SetStatusCondition(&twd.Status.Conditions, metav1.Condition{ + Type: temporaliov1alpha1.ConditionReady, + Status: metav1.ConditionFalse, + Reason: "DeletingPendingMigration", + ObservedGeneration: twd.Generation, + }) + + res := computeKstatus(t, twd) + assert.Equal(t, kstatus.TerminatingStatus, res.Status, "message: %q", res.Message) + }) +} + +// TestIsTerminalWorkerResourceError pins the split that decides whether a failed WRT +// apply reports Failed or InProgress to kstatus. Getting it wrong in the permissive +// direction hangs a deploy until timeout; in the strict direction it aborts a deploy +// over a retryable blip. +func TestIsTerminalWorkerResourceError(t *testing.T) { + gr := schema.GroupResource{Group: "apps", Resource: "deployments"} + + cases := []struct { + name string + err error + renderFailed bool + want bool + }{ + {"NoError", nil, false, false}, + {"RenderFailure", errors.New("template render failed: unknown field"), true, true}, + {"Invalid", apierrors.NewInvalid(schema.GroupKind{Group: "apps", Kind: "Deployment"}, "d", nil), false, true}, + {"Forbidden", apierrors.NewForbidden(gr, "d", errors.New("no RBAC")), false, true}, + {"Unauthorized", apierrors.NewUnauthorized("bad credentials"), false, true}, + {"BadRequest", apierrors.NewBadRequest("malformed patch"), false, true}, + {"MethodNotSupported", apierrors.NewMethodNotSupported(gr, "patch"), false, true}, + {"Conflict", apierrors.NewConflict(gr, "d", errors.New("modified")), false, false}, + {"TooManyRequests", apierrors.NewTooManyRequests("slow down", 1), false, false}, + {"ServerTimeout", apierrors.NewServerTimeout(gr, "patch", 1), false, false}, + {"InternalError", apierrors.NewInternalError(errors.New("boom")), false, false}, + {"PlainTransportError", errors.New("connection reset by peer"), false, false}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, isTerminalWorkerResourceError(tc.err, tc.renderFailed)) + }) + } +} diff --git a/internal/controller/worker_controller.go b/internal/controller/worker_controller.go index 1fe60d97..9352ecb4 100644 --- a/internal/controller/worker_controller.go +++ b/internal/controller/worker_controller.go @@ -266,12 +266,18 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req // finalizer from the previously-referenced connection if no other WD uses it. // The new connection is already protected by ensureConnectionFinalizer above, // so the WD is never left unprotected. - if workerDeploy.Generation != workerDeploy.Status.ObservedGeneration { - current := workerDeploy.Spec.WorkerOptions.ConnectionRef - if observed := workerDeploy.Status.ObservedConnectionRef; observed != nil && !sameConnectionRef(*observed, current) { - if err := r.releaseConnectionFinalizerIfUnused(ctx, l, *observed, workerDeploy.Namespace, workerDeploy.Name); err != nil { - return ctrl.Result{}, err - } + // + // This compares the observed ref directly rather than first gating on + // generation != observedGeneration. Blocked reconciles now advance + // observedGeneration (see recordWarningAndSetBlocked), so a connectionRef change + // whose first reconcile was blocked (by repointing at a Connection that does not + // exist yet for example) would otherwise never be noticed again and the old connection + // would keep the finalizer forever. ObservedConnectionRef is only written on a + // successful reconcile, so it remains the correct thing to compare against. + current := workerDeploy.Spec.WorkerOptions.ConnectionRef + if observed := workerDeploy.Status.ObservedConnectionRef; observed != nil && !sameConnectionRef(*observed, current) { + if err := r.releaseConnectionFinalizerIfUnused(ctx, l, *observed, workerDeploy.Namespace, workerDeploy.Name); err != nil { + return ctrl.Result{}, err } } @@ -576,6 +582,24 @@ func (r *WorkerDeploymentReconciler) markWRTsWDNotFound(ctx context.Context, wd Message: fmt.Sprintf("WorkerDeployment %q not found", wd.Name), ObservedGeneration: wrt.Generation, }) + // Reconciling, not Stalled: a WRT that references a WorkerDeployment which does + // not exist yet is an expected, self-resolving state during creation ordering. + // Reporting Failed here would abort any install that applies the WRT before the WD. + meta.SetStatusCondition(&wrt.Status.Conditions, metav1.Condition{ + Type: temporaliov1alpha1.ConditionStalled, + Status: metav1.ConditionFalse, + Reason: temporaliov1alpha1.ReasonWRTWDNotFound, + Message: fmt.Sprintf("WorkerDeployment %q not found", wd.Name), + ObservedGeneration: wrt.Generation, + }) + meta.SetStatusCondition(&wrt.Status.Conditions, metav1.Condition{ + Type: temporaliov1alpha1.ConditionReconciling, + Status: metav1.ConditionTrue, + Reason: temporaliov1alpha1.ReasonWRTWDNotFound, + Message: fmt.Sprintf("WorkerDeployment %q not found", wd.Name), + ObservedGeneration: wrt.Generation, + }) + wrt.Status.ObservedGeneration = wrt.Generation if err := r.Status().Update(ctx, wrt); err != nil { l.Error(err, "unable to update WorkerResourceTemplate status for missing WorkerDeployment", "WorkerResourceTemplate", wrt.Name, "WorkerDeployment", wd.Name) @@ -788,6 +812,17 @@ func (r *WorkerDeploymentReconciler) syncConditions( metav1.ConditionTrue, temporaliov1alpha1.ReasonConnectionHealthy, //nolint:staticcheck // backward compat "Connection is healthy and auth secret is resolved") + // Reaching this function means the reconcile completed without a blocking error, + // so nothing is stalled whatever stage the rollout is at. Set above the switch + // rather than in each arm for the same reason ConnectionHealthy is: the value does + // not vary by rollout state, and repeating it per arm invites one arm to drift. + // + // SetStatusCondition only upserts, so a Stalled=True from an earlier blocked reconcile + // stays until something sets it False. + r.setCondition(twd, temporaliov1alpha1.ConditionStalled, + metav1.ConditionFalse, temporaliov1alpha1.ReasonReconcileSucceeded, + "Reconcile succeeded") + switch twd.Status.TargetVersion.Status { case temporaliov1alpha1.VersionStatusCurrent: // Rollout itself has completed — Ready stays True regardless of poller @@ -797,6 +832,13 @@ func (r *WorkerDeploymentReconciler) syncConditions( r.setCondition(twd, temporaliov1alpha1.ConditionReady, metav1.ConditionTrue, temporaliov1alpha1.ReasonRolloutComplete, fmt.Sprintf("Rollout complete for buildID %s", twd.Status.TargetVersion.BuildID)) + r.setCondition(twd, temporaliov1alpha1.ConditionProgressing, + metav1.ConditionFalse, temporaliov1alpha1.ReasonRolloutComplete, + fmt.Sprintf("Target version %s is current", twd.Status.TargetVersion.BuildID)) + r.setCondition(twd, temporaliov1alpha1.ConditionReconciling, + metav1.ConditionFalse, temporaliov1alpha1.ReasonRolloutComplete, + fmt.Sprintf("Target version %s is current", twd.Status.TargetVersion.BuildID)) + // Deprecated: set RolloutComplete=True for v1.3.x compat. r.setConditionProgressingForCurrent(twd, temporalState) @@ -814,6 +856,9 @@ func (r *WorkerDeploymentReconciler) syncConditions( r.setCondition(twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionTrue, temporaliov1alpha1.ReasonRamping, fmt.Sprintf("Target version %s is receiving a percentage of new workflows", twd.Status.TargetVersion.BuildID)) + r.setCondition(twd, temporaliov1alpha1.ConditionReconciling, + metav1.ConditionTrue, temporaliov1alpha1.ReasonRamping, + fmt.Sprintf("Target version %s is receiving a percentage of new workflows", twd.Status.TargetVersion.BuildID)) case temporaliov1alpha1.VersionStatusInactive: r.setCondition(twd, temporaliov1alpha1.ConditionReady, metav1.ConditionFalse, temporaliov1alpha1.ReasonWaitingForPromotion, @@ -821,6 +866,9 @@ func (r *WorkerDeploymentReconciler) syncConditions( r.setCondition(twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionTrue, temporaliov1alpha1.ReasonWaitingForPromotion, fmt.Sprintf("Target version %s is waiting for promotion to current", twd.Status.TargetVersion.BuildID)) + r.setCondition(twd, temporaliov1alpha1.ConditionReconciling, + metav1.ConditionTrue, temporaliov1alpha1.ReasonWaitingForPromotion, + fmt.Sprintf("Target version %s is waiting for promotion to current", twd.Status.TargetVersion.BuildID)) default: // NotRegistered or unset: workers have not started polling yet r.setCondition(twd, temporaliov1alpha1.ConditionReady, metav1.ConditionFalse, temporaliov1alpha1.ReasonWaitingForPollers, @@ -828,12 +876,53 @@ func (r *WorkerDeploymentReconciler) syncConditions( r.setCondition(twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionTrue, temporaliov1alpha1.ReasonWaitingForPollers, fmt.Sprintf("Waiting for workers with buildID %s to start polling", twd.Status.TargetVersion.BuildID)) + r.setCondition(twd, temporaliov1alpha1.ConditionReconciling, + metav1.ConditionTrue, temporaliov1alpha1.ReasonWaitingForPollers, + fmt.Sprintf("Waiting for workers with buildID %s to start polling", twd.Status.TargetVersion.BuildID)) } } +// stalledReasons are the blocking reasons reported through the kstatus Stalled +// condition, which makes kstatus report Failed and lets Argo Rollouts and Helm --wait +// fail fast instead of waiting out their timeout. Reasons absent from this set are +// treated as transient: the controller keeps retrying and kstatus keeps reporting +// InProgress, exactly as it did before this condition existed. +// +// A reason belongs here only when it is decidable from information already in hand. +// Both entries below are settled by the spec the user just applied plus how the +// controller was deployed; nothing arriving later can change the answer, so calling +// them terminal can never be wrong. +// +// Everything else is deliberately excluded, in two groups: +// +// - Waiting on another object to exist — ReasonConnectionNotFound (a Connection) and +// ReasonAuthSecretInvalid (a Secret, which this reason also covers when simply +// absent). Applying a WorkerDeployment alongside its Connection and credentials in +// one release gives no ordering guarantee, so a missing reference can be a normal +// few-second gap rather than a mistake. Reporting Failed there would abort a deploy +// that was about to succeed, and a false failure costs more than a slow one. +// Distinguishing a real typo would need a grace period — report Stalled only once +// the reference has been missing for a while — which lastTransitionTime already +// makes measurable. Until then these keep pre-existing behaviour. +// +// - Transient infrastructure failures — ReasonTemporalClientCreationFailed and +// ReasonTemporalStateFetchFailed (server unreachable or rate limited; the +// ResourceExhausted paths requeue after 30s) and ReasonPlanGenerationFailed / +// ReasonPlanExecutionFailed (retried with backoff). Marking these Stalled would +// abort a deploy that is merely being throttled. +// +// WorkerResourceTemplate makes the same call for the same reason: a WRT whose +// WorkerDeployment does not exist yet reports Reconciling, not Stalled. See +// markWRTsWDNotFound. +var stalledReasons = map[string]bool{ + temporaliov1alpha1.ReasonInvalidSpec: true, + temporaliov1alpha1.ReasonClusterConnectionUnsupported: true, +} + // recordWarningAndSetBlocked emits a warning event, sets Progressing=False and Ready=False -// with the given reason, and persists the status immediately. Called on all error paths that -// block reconciliation progress. +// with the given reason, sets Stalled=True for reasons in stalledReasons, advances +// status.observedGeneration, and persists the status immediately. Called on all error paths +// that block reconciliation progress. func (r *WorkerDeploymentReconciler) recordWarningAndSetBlocked( ctx context.Context, workerDeploy *temporaliov1alpha1.WorkerDeployment, @@ -844,6 +933,35 @@ func (r *WorkerDeploymentReconciler) recordWarningAndSetBlocked( r.Recorder.Eventf(workerDeploy, corev1.EventTypeWarning, reason, "%s", eventMessage) r.setCondition(workerDeploy, temporaliov1alpha1.ConditionProgressing, metav1.ConditionFalse, reason, conditionMessage) r.setCondition(workerDeploy, temporaliov1alpha1.ConditionReady, metav1.ConditionFalse, reason, conditionMessage) + + // Report terminal failures through the kstatus Stalled condition so consumers + // (Argo Rollouts, Helm --wait) fail fast instead of waiting out their timeout. + // Progressing=False alone cannot express this: kstatus does not read it. + // + // Both are always written, and exactly one of them is True. kstatus scans + // status.conditions in array order and returns on the first match, so an object + // carrying both as True would get a verdict that depends on which was inserted + // first. Writing the other as False rather than removing it keeps every condition + // this controller owns present on every object, whatever path produced it. + if stalledReasons[reason] { + r.setCondition(workerDeploy, temporaliov1alpha1.ConditionStalled, metav1.ConditionTrue, reason, conditionMessage) + r.setCondition(workerDeploy, temporaliov1alpha1.ConditionReconciling, metav1.ConditionFalse, reason, conditionMessage) + } else { + // Transient failure: the controller is still retrying (with backoff, or an + // explicit RequeueAfter for rate limits), so say so directly instead of + // leaving kstatus to infer it from the Ready=False fallback. + r.setCondition(workerDeploy, temporaliov1alpha1.ConditionStalled, metav1.ConditionFalse, reason, conditionMessage) + r.setCondition(workerDeploy, temporaliov1alpha1.ConditionReconciling, metav1.ConditionTrue, reason, conditionMessage) + } + + // Record that this generation was observed even though it could not be + // reconciled: the controller has seen this spec and reached a verdict on it. + // Without this, status.observedGeneration lags metadata.generation on every + // error path that returns before generateStatus, and kstatus returns + // InProgress from its generation check before it ever reads the conditions + // set above. + workerDeploy.Status.ObservedGeneration = workerDeploy.Generation + // Deprecated: set ConnectionHealthy=False for v1.3.x compat, but only for // reasons that actually indicate connection/auth issues. Plan generation and execution // failures are unrelated to connection health and should not trigger this condition. diff --git a/internal/tests/go.mod b/internal/tests/go.mod index 4c7ebc0e..433454c1 100644 --- a/internal/tests/go.mod +++ b/internal/tests/go.mod @@ -12,6 +12,7 @@ require ( k8s.io/api v0.36.3 k8s.io/apimachinery v0.36.3 k8s.io/client-go v0.36.3 + sigs.k8s.io/cli-utils v0.37.2 sigs.k8s.io/controller-runtime v0.24.0 ) diff --git a/internal/tests/go.sum b/internal/tests/go.sum index b1e43a7d..14dfced1 100644 --- a/internal/tests/go.sum +++ b/internal/tests/go.sum @@ -682,6 +682,8 @@ modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= +sigs.k8s.io/cli-utils v0.37.2 h1:GOfKw5RV2HDQZDJlru5KkfLO1tbxqMoyn1IYUxqBpNg= +sigs.k8s.io/cli-utils v0.37.2/go.mod h1:V+IZZr4UoGj7gMJXklWBg6t5xbdThFBcpj4MrZuCYco= sigs.k8s.io/controller-runtime v0.24.0 h1:Ck6N2LdS8Lovy1o25BB4r1xjvLEKUl1s2o9kU+KWDE4= sigs.k8s.io/controller-runtime v0.24.0/go.mod h1:vFkfY5fGt5xAC/sKb8IBFKgWPNKG9OUG29dR8Y2wImw= sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg= diff --git a/internal/tests/internal/integration_test.go b/internal/tests/internal/integration_test.go index 7eee0923..3153e49e 100644 --- a/internal/tests/internal/integration_test.go +++ b/internal/tests/internal/integration_test.go @@ -991,6 +991,10 @@ func TestIntegration(t *testing.T) { // Conditions and events tests runConditionsAndEventsTests(t, k8sClient, mgr, ts, testNamespace.Name) + // kstatus verdict tests: assert what Helm --wait and Flux conclude from the + // conditions written above, read back from the real API server. + runKstatusTests(t, k8sClient, mgr, ts, testNamespace.Name) + // Version-summary divergence safety test runNotRegisteredVersionTests(t, k8sClient, clientPool, ts, testNamespace.Name) diff --git a/internal/tests/internal/kstatus_integration_test.go b/internal/tests/internal/kstatus_integration_test.go new file mode 100644 index 00000000..0aa34529 --- /dev/null +++ b/internal/tests/internal/kstatus_integration_test.go @@ -0,0 +1,299 @@ +package internal + +// This file verifies that the conditions the controller writes are read the way +// kstatus intends, against a real API server. The unit tests in +// internal/controller/kstatus_test.go cover the same states with a fake client; +// what only a real API server can prove is that the CRD schema does not prune +// status.conditions[type=Stalled|Reconciling] or status.observedGeneration on the +// way in, and that the values survive a status-subresource round trip. +// +// kstatus is the library Helm --wait and Flux use to decide whether a custom +// resource is healthy, so these verdicts are what those tools will conclude. +// +// Covered: +// - Current: target version promoted to current +// - InProgress: target registered but not yet promoted (Reconciling=True) +// - Failed: terminal blocking error (Stalled=True) +// - InProgress: transient blocking error (Reconciling=True, no Stalled) + +import ( + "context" + "fmt" + "testing" + "time" + + temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" + "github.com/temporalio/temporal-worker-controller/internal/testhelpers" + "go.temporal.io/server/temporaltest" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + kstatus "sigs.k8s.io/cli-utils/pkg/kstatus/status" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/manager" +) + +// waitForKstatus polls the named WorkerDeployment, converts what the API server +// actually returns into unstructured, and runs the real kstatus decision tree over +// it until the verdict matches want, or fatals on timeout. +func waitForKstatus( + t *testing.T, + ctx context.Context, + k8sClient client.Client, + name, namespace string, + want kstatus.Status, + timeout, interval time.Duration, +) { + t.Helper() + eventually(t, timeout, interval, func() error { + var wd temporaliov1alpha1.WorkerDeployment + if err := k8sClient.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, &wd); err != nil { + return fmt.Errorf("get WorkerDeployment: %w", err) + } + // Get strips TypeMeta; kstatus only uses it for message text, but set it so + // failure messages name the kind. + wd.TypeMeta = metav1.TypeMeta{ + APIVersion: temporaliov1alpha1.GroupVersion.String(), + Kind: "WorkerDeployment", + } + + content, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&wd) + if err != nil { + return fmt.Errorf("convert to unstructured: %w", err) + } + res, err := kstatus.Compute(&unstructured.Unstructured{Object: content}) + if err != nil { + return fmt.Errorf("kstatus.Compute: %w", err) + } + if res.Status != want { + return fmt.Errorf("kstatus verdict: want %s, got %s (message: %q, conditions: %v)", + want, res.Status, res.Message, conditionSummary(wd.Status.Conditions)) + } + return nil + }) +} + +// conditionSummary renders conditions compactly for failure messages. +func conditionSummary(conds []metav1.Condition) []string { + out := make([]string, 0, len(conds)) + for _, c := range conds { + out = append(out, fmt.Sprintf("%s=%s(%s)", c.Type, c.Status, c.Reason)) + } + return out +} + +// requireObservedGenerationCurrent fails if status.observedGeneration has not caught +// up with metadata.generation. kstatus checks this before it looks at any condition, +// so a lagging value masks every condition the controller wrote. +func requireObservedGenerationCurrent(t *testing.T, ctx context.Context, k8sClient client.Client, name, namespace string) { + t.Helper() + var wd temporaliov1alpha1.WorkerDeployment + if err := k8sClient.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, &wd); err != nil { + t.Fatalf("get WorkerDeployment: %v", err) + } + if wd.Status.ObservedGeneration != wd.Generation { + t.Fatalf("status.observedGeneration = %d, want %d (metadata.generation)", + wd.Status.ObservedGeneration, wd.Generation) + } +} + +func runKstatusTests( + t *testing.T, + k8sClient client.Client, + mgr manager.Manager, + ts *temporaltest.TestServer, + testNamespace string, +) { + cases := []testCase{ + { + // A completed rollout must read as Current, or Helm --wait never returns. + name: "kstatus-current-when-rollout-complete", + builder: testhelpers.NewTestCase(). + WithInput( + testhelpers.NewWorkerDeploymentBuilder(). + WithAllAtOnceStrategy(). + WithTargetTemplate("v1.0"), + ). + WithExpectedStatus( + testhelpers.NewStatusBuilder(). + WithTargetVersion("v1.0", temporaliov1alpha1.VersionStatusCurrent, -1, true, false). + WithCurrentVersion("v1.0", true, false), + ). + WithValidatorFunction(func(t *testing.T, ctx context.Context, tc testhelpers.TestCase, env testhelpers.TestEnv) { + twd := tc.GetTWD() + waitForKstatus(t, ctx, env.K8sClient, twd.Name, twd.Namespace, + kstatus.CurrentStatus, 30*time.Second, time.Second) + requireObservedGenerationCurrent(t, ctx, env.K8sClient, twd.Name, twd.Namespace) + }), + }, + { + // Registered but awaiting promotion: Reconciling=True should put kstatus on + // its intended path rather than the Ready=False fallback. + name: "kstatus-inprogress-while-awaiting-promotion", + builder: testhelpers.NewTestCase(). + WithInput( + testhelpers.NewWorkerDeploymentBuilder(). + WithManualStrategy(). + WithTargetTemplate("v1.0"), + ). + WithExpectedStatus( + testhelpers.NewStatusBuilder(). + WithTargetVersion("v1.0", temporaliov1alpha1.VersionStatusInactive, -1, true, false), + ). + WithValidatorFunction(func(t *testing.T, ctx context.Context, tc testhelpers.TestCase, env testhelpers.TestEnv) { + twd := tc.GetTWD() + waitForCondition(t, ctx, env.K8sClient, twd.Name, twd.Namespace, + temporaliov1alpha1.ConditionReconciling, + metav1.ConditionTrue, + temporaliov1alpha1.ReasonWaitingForPromotion, + 30*time.Second, time.Second) + waitForKstatus(t, ctx, env.K8sClient, twd.Name, twd.Namespace, + kstatus.InProgressStatus, 30*time.Second, time.Second) + }), + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + testWorkerDeploymentCreation(ctx, t, k8sClient, mgr, ts, tc.builder.BuildWithValues(tc.name, testNamespace, ts.GetDefaultNamespace())) + }) + } + + // The blocking-error cases run standalone for the same reason the existing + // condition tests do: the controller fails before creating any k8s Deployment, so + // testWorkerDeploymentCreation's status-validation machinery would time out. + + t.Run("kstatus-failed-on-invalid-spec", func(t *testing.T) { + ctx := context.Background() + name := "kstatus-failed-invalid-spec" + + conn := &temporaliov1alpha1.Connection{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testNamespace}, + Spec: temporaliov1alpha1.ConnectionSpec{HostPort: ts.GetFrontendHostPort()}, + } + if err := k8sClient.Create(ctx, conn); err != nil { + t.Fatalf("failed to create Connection: %v", err) + } + + // Progressive ramp steps must strictly increase, so 50 followed by 10 is + // invalid. The CRD schema cannot express that ordering rule (it does enforce + // the 30s minimum pause used here), and the validating webhook is not + // installed in this environment, so the API server accepts the object and the + // controller reports ReasonInvalidSpec. Nothing arriving later can make this + // spec valid, so it is terminal and must read as Failed rather than making a + // CD tool wait out its timeout. + twd := testhelpers.NewWorkerDeploymentBuilder(). + WithProgressiveStrategy( + temporaliov1alpha1.RolloutStep{RampPercentage: 50, PauseDuration: metav1.Duration{Duration: 30 * time.Second}}, + temporaliov1alpha1.RolloutStep{RampPercentage: 10, PauseDuration: metav1.Duration{Duration: 30 * time.Second}}, + ). + WithTargetTemplate("v1.0"). + WithName(name). + WithNamespace(testNamespace). + WithConnection(name). + WithTemporalNamespace(ts.GetDefaultNamespace()). + Build() + if err := k8sClient.Create(ctx, twd); err != nil { + t.Fatalf("failed to create WorkerDeployment: %v", err) + } + + waitForCondition(t, ctx, k8sClient, twd.Name, twd.Namespace, + temporaliov1alpha1.ConditionStalled, + metav1.ConditionTrue, + temporaliov1alpha1.ReasonInvalidSpec, + 30*time.Second, time.Second) + waitForKstatus(t, ctx, k8sClient, twd.Name, twd.Namespace, + kstatus.FailedStatus, 30*time.Second, time.Second) + // The blocked path must still advance observedGeneration, or kstatus returns + // InProgress from its generation check and never reads Stalled at all. + requireObservedGenerationCurrent(t, ctx, k8sClient, twd.Name, twd.Namespace) + }) + + t.Run("kstatus-inprogress-on-missing-connection", func(t *testing.T) { + ctx := context.Background() + name := "kstatus-inprogress-missing-conn" + + // No Connection is created, so the reference cannot resolve. This must stay + // InProgress: a WorkerDeployment applied alongside its Connection has no + // ordering guarantee, so the reference may simply not exist yet, and reporting + // Failed would abort a deploy that was about to succeed. + twd := testhelpers.NewWorkerDeploymentBuilder(). + WithManualStrategy(). + WithTargetTemplate("v1.0"). + WithName(name). + WithNamespace(testNamespace). + WithConnection(name). + WithTemporalNamespace(ts.GetDefaultNamespace()). + Build() + if err := k8sClient.Create(ctx, twd); err != nil { + t.Fatalf("failed to create WorkerDeployment: %v", err) + } + + waitForCondition(t, ctx, k8sClient, twd.Name, twd.Namespace, + temporaliov1alpha1.ConditionReconciling, + metav1.ConditionTrue, + temporaliov1alpha1.ReasonConnectionNotFound, + 30*time.Second, time.Second) + waitForKstatus(t, ctx, k8sClient, twd.Name, twd.Namespace, + kstatus.InProgressStatus, 30*time.Second, time.Second) + + var got temporaliov1alpha1.WorkerDeployment + if err := k8sClient.Get(ctx, types.NamespacedName{Name: twd.Name, Namespace: twd.Namespace}, &got); err != nil { + t.Fatalf("get WorkerDeployment: %v", err) + } + for _, c := range got.Status.Conditions { + if c.Type == temporaliov1alpha1.ConditionStalled && c.Status == metav1.ConditionTrue { + t.Fatalf("a missing Connection must not set Stalled=True (reason %q)", c.Reason) + } + } + }) + + t.Run("kstatus-inprogress-on-transient-error", func(t *testing.T) { + ctx := context.Background() + name := "kstatus-inprogress-transient" + + // The Connection resolves and the client dials, but the Temporal namespace does + // not exist, so the state fetch fails. ReasonTemporalStateFetchFailed is + // deliberately NOT in stalledReasons: the controller keeps retrying, so + // reporting Failed would abort a deploy over something self-resolving. + conn := &temporaliov1alpha1.Connection{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testNamespace}, + Spec: temporaliov1alpha1.ConnectionSpec{HostPort: ts.GetFrontendHostPort()}, + } + if err := k8sClient.Create(ctx, conn); err != nil { + t.Fatalf("failed to create Connection: %v", err) + } + + twd := testhelpers.NewWorkerDeploymentBuilder(). + WithManualStrategy(). + WithTargetTemplate("v1.0"). + WithName(name). + WithNamespace(testNamespace). + WithConnection(name). + WithTemporalNamespace("does-not-exist"). + Build() + if err := k8sClient.Create(ctx, twd); err != nil { + t.Fatalf("failed to create WorkerDeployment: %v", err) + } + + waitForCondition(t, ctx, k8sClient, twd.Name, twd.Namespace, + temporaliov1alpha1.ConditionReconciling, + metav1.ConditionTrue, + temporaliov1alpha1.ReasonTemporalStateFetchFailed, + 30*time.Second, time.Second) + waitForKstatus(t, ctx, k8sClient, twd.Name, twd.Namespace, + kstatus.InProgressStatus, 30*time.Second, time.Second) + + var got temporaliov1alpha1.WorkerDeployment + if err := k8sClient.Get(ctx, types.NamespacedName{Name: twd.Name, Namespace: twd.Namespace}, &got); err != nil { + t.Fatalf("get WorkerDeployment: %v", err) + } + for _, c := range got.Status.Conditions { + if c.Type == temporaliov1alpha1.ConditionStalled && c.Status == metav1.ConditionTrue { + t.Fatalf("a transient failure must not set Stalled=True (reason %q)", c.Reason) + } + } + }) +}