From 94d63d4bb5780b1f42a743f5c0bc166fc4c5e139 Mon Sep 17 00:00:00 2001 From: wankhede04 Date: Tue, 21 Jul 2026 10:49:06 +0530 Subject: [PATCH 1/6] feat(temporal): capture poller health per task queue from existing DescribeTaskQueue calls Extend VersionInfo with PollerHealth/PollerHealthUnknown, populated for the current version (via DescribeVersion) and the target version (reusing task queues already fetched in GetTestWorkflowStatus), so poller presence data is no longer discarded after being reduced to AllTaskQueuesHaveUnversionedPoller. --- internal/temporal/worker_deployment.go | 53 ++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/internal/temporal/worker_deployment.go b/internal/temporal/worker_deployment.go index b05eaf40..20e27576 100644 --- a/internal/temporal/worker_deployment.go +++ b/internal/temporal/worker_deployment.go @@ -38,6 +38,17 @@ type VersionInfo struct { // - Strategy is Progressive, and // - Presence of unversioned pollers in all task queues of target version cannot be confirmed. AllTaskQueuesHaveUnversionedPoller bool + + // PollerHealth summarizes, per task queue, whether at least one poller was observed. + // Keyed by task queue name; true = has at least one poller. A task queue is absent + // from the map if its poller status could not be determined (e.g. a transient + // DescribeTaskQueue error) -- see PollerHealthUnknown. + PollerHealth map[string]bool + + // PollerHealthUnknown is true if poller status could not be determined for one or + // more of this version's task queues. Callers must not interpret this as unhealthy; + // it means "don't know", not "broken". + PollerHealthUnknown bool } // TemporalWorkerState represents the state of a worker deployment in Temporal @@ -177,6 +188,22 @@ func GetWorkerDeploymentState( } + // Poller health for the current version reflects whether workers are actively + // receiving tasks for the version serving production traffic, as opposed to + // merely being Ready at the Kubernetes level. Only checked for the current + // version to avoid an extra DescribeVersion/DescribeTaskQueue round trip per + // version on every reconcile. + if versionInfo.Status == temporaliov1alpha1.VersionStatusCurrent { + currentDesc, descErr := deploymentHandler.DescribeVersion(ctx, temporalClient.WorkerDeploymentDescribeVersionOptions{ + BuildID: version.DeploymentVersion.BuildId, + }) + if descErr == nil { + versionInfo.PollerHealth, versionInfo.PollerHealthUnknown = computePollerHealth(ctx, client, currentDesc.Info.TaskQueuesInfos) + } else { + versionInfo.PollerHealthUnknown = true + } + } + state.Versions[version.DeploymentVersion.BuildId] = versionInfo } @@ -228,6 +255,11 @@ func GetTestWorkflowStatus( return nil, fmt.Errorf("unable to describe worker deployment version for buildID %q: %w", buildID, err) } + // Poller health for the target version, computed from the task queues already + // fetched above via DescribeVersion (no additional per-version round trip). + temporalState.Versions[buildID].PollerHealth, temporalState.Versions[buildID].PollerHealthUnknown = + computePollerHealth(ctx, client, versionResp.Info.TaskQueuesInfos) + // Check test workflows for each task queue for _, tq := range versionResp.Info.TaskQueuesInfos { // Skip non-workflow task queues @@ -333,6 +365,27 @@ func getPollers(ctx context.Context, return resp.GetPollers(), nil } +// computePollerHealth reports, per task queue, whether at least one poller was +// observed. A task queue is omitted from the returned map (and unknown is set +// to true) if its poller status could not be determined, e.g. a transient +// DescribeTaskQueue error -- this must not be interpreted as unhealthy. +func computePollerHealth( + ctx context.Context, + client temporalClient.Client, + tqs []temporalClient.WorkerDeploymentTaskQueueInfo, +) (health map[string]bool, unknown bool) { + health = make(map[string]bool, len(tqs)) + for _, tqInfo := range tqs { + pollers, err := getPollers(ctx, client, tqInfo) + if err != nil { //nolint:revive // TODO(carlydf): consider logging this error + unknown = true + continue + } + health[tqInfo.Name] = len(pollers) > 0 + } + return health, unknown +} + func allTaskQueuesHaveUnversionedPoller( ctx context.Context, client temporalClient.Client, From c7ff05c518ea488cb2f7c5e2972f4daad894229d Mon Sep 17 00:00:00 2001 From: wankhede04 Date: Tue, 21 Jul 2026 10:49:34 +0530 Subject: [PATCH 2/6] feat(api): add WorkersHealthy condition type and reasons Add ConditionWorkersHealthy plus ReasonPollersHealthy/ReasonNoActivePollers/ ReasonPollerStatusUnknown, so the WorkerDeployment CRD can report whether workers are actually polling Temporal rather than merely Ready at the Kubernetes level. --- api/v1alpha1/conditions.go | 6 ++++++ api/v1alpha1/workerdeployment_types.go | 14 ++++++++++++++ 2 files changed, 20 insertions(+) diff --git a/api/v1alpha1/conditions.go b/api/v1alpha1/conditions.go index 953ac72e..8572b55e 100644 --- a/api/v1alpha1/conditions.go +++ b/api/v1alpha1/conditions.go @@ -11,6 +11,12 @@ 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" + + // ConditionWorkersHealthy indicates whether workers for the version currently + // serving production traffic (or, before the first rollout completes, the target + // version) are actively polling Temporal -- as opposed to merely being Ready at + // the Kubernetes level. + ConditionWorkersHealthy = "WorkersHealthy" ) // Deprecated condition type constants. Maintained for backward compatibility with diff --git a/api/v1alpha1/workerdeployment_types.go b/api/v1alpha1/workerdeployment_types.go index 2f4930de..f002ffc1 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/api/v1alpha1/workerdeployment_types.go @@ -143,6 +143,20 @@ const ( // Deprecated: Use ReasonRolloutComplete on ConditionReady instead. ReasonConnectionHealthy = "ConnectionHealthy" + + // ReasonPollersHealthy is set on ConditionWorkersHealthy=True when all known + // task queues for the version have at least one active poller. + ReasonPollersHealthy = "PollersHealthy" + + // ReasonNoActivePollers is set on ConditionWorkersHealthy=False when one or more + // of a version's task queues have no active poller. + ReasonNoActivePollers = "NoActivePollers" + + // ReasonPollerStatusUnknown is set on ConditionWorkersHealthy=Unknown when poller + // status could not be determined (e.g. a transient DescribeTaskQueue error, or the + // version is not yet registered with Temporal). This must NOT be treated as + // unhealthy -- it means "don't know", not "broken". + ReasonPollerStatusUnknown = "PollerStatusUnknown" ) // VersionStatus indicates the status of a version. From 713bc6dac089d801a0d5692c3e4d1db323c7441c Mon Sep 17 00:00:00 2001 From: wankhede04 Date: Tue, 21 Jul 2026 10:49:48 +0530 Subject: [PATCH 3/6] feat(controller): surface poller and gate-workflow health as conditions and events Set ConditionWorkersHealthy from poller health of the version serving production traffic (current, falling back to target pre-rollout), emitting a Normal/Warning event only on transition. Also emit a Warning event the first time a gate/test workflow ends in Failed/Canceled/Terminated/TimedOut. This is reporting-only: no rollout/reconciliation logic changes. Closes #447 --- internal/controller/genstatus.go | 34 +++++++ internal/controller/util.go | 1 + internal/controller/worker_controller.go | 10 +- internal/controller/workers_healthy.go | 101 ++++++++++++++++++++ internal/controller/workers_healthy_test.go | 78 +++++++++++++++ 5 files changed, 221 insertions(+), 3 deletions(-) create mode 100644 internal/controller/workers_healthy.go create mode 100644 internal/controller/workers_healthy_test.go diff --git a/internal/controller/genstatus.go b/internal/controller/genstatus.go index c6caa95b..d1da7a61 100644 --- a/internal/controller/genstatus.go +++ b/internal/controller/genstatus.go @@ -12,9 +12,24 @@ import ( "github.com/temporalio/temporal-worker-controller/internal/k8s" "github.com/temporalio/temporal-worker-controller/internal/temporal" temporalclient "go.temporal.io/sdk/client" + corev1 "k8s.io/api/core/v1" ctrl "sigs.k8s.io/controller-runtime" ) +// isGateWorkflowTerminalFailure reports whether a test/gate workflow status +// represents an ended-but-not-successful terminal state worth alerting on. +func isGateWorkflowTerminalFailure(status temporaliov1alpha1.WorkflowExecutionStatus) bool { + switch status { + case temporaliov1alpha1.WorkflowExecutionStatusFailed, + temporaliov1alpha1.WorkflowExecutionStatusCanceled, + temporaliov1alpha1.WorkflowExecutionStatusTerminated, + temporaliov1alpha1.WorkflowExecutionStatusTimedOut: + return true + default: + return false + } +} + func (r *WorkerDeploymentReconciler) generateStatus( ctx context.Context, l logr.Logger, @@ -42,6 +57,25 @@ func (r *WorkerDeploymentReconciler) generateStatus( // Continue without test workflow status } + // Emit a Warning event the first time a gate/test workflow is observed to have + // ended in a non-successful terminal state. Compare against the previous + // reconcile's recorded status (still on workerDeploy.Status at this point, since + // it hasn't been overwritten yet) so this doesn't re-fire on every loop. + prevStatusByWorkflowID := make(map[string]temporaliov1alpha1.WorkflowExecutionStatus, len(workerDeploy.Status.TargetVersion.TestWorkflows)) + for _, wf := range workerDeploy.Status.TargetVersion.TestWorkflows { + prevStatusByWorkflowID[wf.WorkflowID] = wf.Status + } + for _, wf := range testWorkflows { + if !isGateWorkflowTerminalFailure(wf.Status) { + continue + } + if prevStatusByWorkflowID[wf.WorkflowID] == wf.Status { + continue + } + r.Recorder.Eventf(workerDeploy, corev1.EventTypeWarning, ReasonGateWorkflowFailed, + "Gate/test workflow %s for version %s ended with status %s", wf.WorkflowID, targetBuildID, wf.Status) + } + // Add test workflow status to version info if it doesn't exist if versionInfo, exists := temporalState.Versions[targetBuildID]; exists { versionInfo.TestWorkflows = append(versionInfo.TestWorkflows, testWorkflows...) diff --git a/internal/controller/util.go b/internal/controller/util.go index e91fcbb5..0f6525ce 100644 --- a/internal/controller/util.go +++ b/internal/controller/util.go @@ -28,6 +28,7 @@ const ( ReasonVersionPromotionFailed = "VersionPromotionFailed" ReasonMetadataUpdateFailed = "MetadataUpdateFailed" ReasonManagerIdentityClaimFailed = "ManagerIdentityClaimFailed" + ReasonGateWorkflowFailed = "GateWorkflowFailed" ) const ( diff --git a/internal/controller/worker_controller.go b/internal/controller/worker_controller.go index c0acb5cc..cbf641a5 100644 --- a/internal/controller/worker_controller.go +++ b/internal/controller/worker_controller.go @@ -393,6 +393,9 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req // Derive Ready/Progressing from rollout state before the final write. r.syncConditions(&workerDeploy) + // Surface poller health (are workers actually polling Temporal, not just Ready + // at the Kubernetes level) as a condition, separate from rollout progress. + r.syncWorkersHealthyCondition(&workerDeploy, temporalState) // Single status write per reconcile: persists the generated status and // conditions set during this loop (Ready, Progressing). @@ -685,14 +688,15 @@ func (r *WorkerDeploymentReconciler) handleDeletion( return nil } -// setCondition sets a condition on the WorkerDeployment status. +// setCondition sets the given condition and reports whether it actually changed +// (differed in Status/Reason/Message/ObservedGeneration from what was already set). func (r *WorkerDeploymentReconciler) setCondition( workerDeploy *temporaliov1alpha1.WorkerDeployment, conditionType string, status metav1.ConditionStatus, reason, message string, -) { - meta.SetStatusCondition(&workerDeploy.Status.Conditions, metav1.Condition{ +) bool { + return meta.SetStatusCondition(&workerDeploy.Status.Conditions, metav1.Condition{ Type: conditionType, Status: status, ObservedGeneration: workerDeploy.Generation, diff --git a/internal/controller/workers_healthy.go b/internal/controller/workers_healthy.go new file mode 100644 index 00000000..6767666b --- /dev/null +++ b/internal/controller/workers_healthy.go @@ -0,0 +1,101 @@ +// 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 ( + "fmt" + "sort" + "strings" + + temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" + "github.com/temporalio/temporal-worker-controller/internal/temporal" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// computePollerHealthCondition derives the ConditionWorkersHealthy status/reason from +// a version's poller health data. It is a pure function so the decision logic can be +// unit tested without an envtest environment. +// +// - pollerHealth == nil: poller status was never checked for this version (e.g. not +// yet registered with Temporal) -> Unknown. +// - any task queue with a false value -> False, naming the affected queues. A known +// problem takes precedence over an unrelated unknown elsewhere. +// - no false values, but unknown == true (some task queues errored) -> Unknown. +// - all task queues true, unknown == false -> True. +func computePollerHealthCondition( + pollerHealth map[string]bool, + unknown bool, +) (status metav1.ConditionStatus, reason string, affectedQueues []string) { + if pollerHealth == nil { + return metav1.ConditionUnknown, temporaliov1alpha1.ReasonPollerStatusUnknown, nil + } + + for tq, healthy := range pollerHealth { + if !healthy { + affectedQueues = append(affectedQueues, tq) + } + } + + if len(affectedQueues) > 0 { + sort.Strings(affectedQueues) + return metav1.ConditionFalse, temporaliov1alpha1.ReasonNoActivePollers, affectedQueues + } + if unknown { + return metav1.ConditionUnknown, temporaliov1alpha1.ReasonPollerStatusUnknown, nil + } + return metav1.ConditionTrue, temporaliov1alpha1.ReasonPollersHealthy, nil +} + +// syncWorkersHealthyCondition sets ConditionWorkersHealthy from poller health observed +// for the version currently serving production traffic (CurrentVersion), falling back +// to TargetVersion before the first rollout has ever completed (CurrentVersion nil). +// It emits a Normal/Warning event only when the condition actually transitions, to +// avoid spamming an Event on every reconcile loop. +func (r *WorkerDeploymentReconciler) syncWorkersHealthyCondition( + workerDeploy *temporaliov1alpha1.WorkerDeployment, + temporalState *temporal.TemporalWorkerState, +) { + buildID := workerDeploy.Status.TargetVersion.BuildID + if workerDeploy.Status.CurrentVersion != nil { + buildID = workerDeploy.Status.CurrentVersion.BuildID + } + if buildID == "" { + return + } + + var pollerHealth map[string]bool + var unknown bool + if versionInfo, exists := temporalState.Versions[buildID]; exists { + pollerHealth = versionInfo.PollerHealth + unknown = versionInfo.PollerHealthUnknown + } + + status, reason, affectedQueues := computePollerHealthCondition(pollerHealth, unknown) + + var message string + switch status { + case metav1.ConditionFalse: + message = fmt.Sprintf("Version %s has no active pollers on task queue(s): %s", buildID, strings.Join(affectedQueues, ", ")) + case metav1.ConditionTrue: + message = fmt.Sprintf("Version %s pollers are healthy", buildID) + default: + message = fmt.Sprintf("Poller status for version %s could not be determined", buildID) + } + + changed := r.setCondition(workerDeploy, temporaliov1alpha1.ConditionWorkersHealthy, status, reason, message) + if !changed { + return + } + + switch status { + case metav1.ConditionFalse: + r.Recorder.Eventf(workerDeploy, corev1.EventTypeWarning, temporaliov1alpha1.ReasonNoActivePollers, + "Version %s has no active pollers on task queue(s): %s", buildID, strings.Join(affectedQueues, ", ")) + case metav1.ConditionTrue: + r.Recorder.Eventf(workerDeploy, corev1.EventTypeNormal, temporaliov1alpha1.ReasonPollersHealthy, + "Version %s pollers are healthy", buildID) + } +} diff --git a/internal/controller/workers_healthy_test.go b/internal/controller/workers_healthy_test.go new file mode 100644 index 00000000..d1f690c7 --- /dev/null +++ b/internal/controller/workers_healthy_test.go @@ -0,0 +1,78 @@ +// 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 ( + "testing" + + "github.com/stretchr/testify/assert" + temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestComputePollerHealthCondition(t *testing.T) { + tests := []struct { + name string + pollerHealth map[string]bool + unknown bool + wantStatus metav1.ConditionStatus + wantReason string + wantAffected []string + }{ + { + name: "nil map means never checked", + pollerHealth: nil, + unknown: false, + wantStatus: metav1.ConditionUnknown, + wantReason: temporaliov1alpha1.ReasonPollerStatusUnknown, + }, + { + name: "all queues healthy", + pollerHealth: map[string]bool{"tq-1": true, "tq-2": true}, + unknown: false, + wantStatus: metav1.ConditionTrue, + wantReason: temporaliov1alpha1.ReasonPollersHealthy, + }, + { + name: "one queue has no pollers", + pollerHealth: map[string]bool{"tq-1": true, "tq-2": false}, + unknown: false, + wantStatus: metav1.ConditionFalse, + wantReason: temporaliov1alpha1.ReasonNoActivePollers, + wantAffected: []string{"tq-2"}, + }, + { + name: "empty map with no fetch errors is healthy (no task queues to check)", + pollerHealth: map[string]bool{}, + unknown: false, + wantStatus: metav1.ConditionTrue, + wantReason: temporaliov1alpha1.ReasonPollersHealthy, + }, + { + name: "fetch error with no confirmed-unhealthy queue is unknown, not unhealthy", + pollerHealth: map[string]bool{"tq-1": true}, + unknown: true, + wantStatus: metav1.ConditionUnknown, + wantReason: temporaliov1alpha1.ReasonPollerStatusUnknown, + }, + { + name: "a confirmed no-poller queue wins over an unrelated fetch error", + pollerHealth: map[string]bool{"tq-1": false}, + unknown: true, + wantStatus: metav1.ConditionFalse, + wantReason: temporaliov1alpha1.ReasonNoActivePollers, + wantAffected: []string{"tq-1"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + status, reason, affected := computePollerHealthCondition(tt.pollerHealth, tt.unknown) + assert.Equal(t, tt.wantStatus, status) + assert.Equal(t, tt.wantReason, reason) + assert.Equal(t, tt.wantAffected, affected) + }) + } +} From dc7a237b62473a3ef5cafa847468477dee7cc61d Mon Sep 17 00:00:00 2001 From: wankhede04 Date: Fri, 24 Jul 2026 12:37:32 +0530 Subject: [PATCH 4/6] refactor(controller): split gate-workflow-failure events out of this PR Per review feedback from @jaypipes on PR #448: this PR should stay scoped to the poller-health-checking work. The gate/test workflow terminal-failure Warning event emission (isGateWorkflowTerminalFailure and the associated event loop in generateStatus, plus ReasonGateWorkflowFailed) is orthogonal to poller health and will be proposed in a separate PR instead. --- internal/controller/genstatus.go | 34 -------------------------------- internal/controller/util.go | 1 - 2 files changed, 35 deletions(-) diff --git a/internal/controller/genstatus.go b/internal/controller/genstatus.go index d1da7a61..c6caa95b 100644 --- a/internal/controller/genstatus.go +++ b/internal/controller/genstatus.go @@ -12,24 +12,9 @@ import ( "github.com/temporalio/temporal-worker-controller/internal/k8s" "github.com/temporalio/temporal-worker-controller/internal/temporal" temporalclient "go.temporal.io/sdk/client" - corev1 "k8s.io/api/core/v1" ctrl "sigs.k8s.io/controller-runtime" ) -// isGateWorkflowTerminalFailure reports whether a test/gate workflow status -// represents an ended-but-not-successful terminal state worth alerting on. -func isGateWorkflowTerminalFailure(status temporaliov1alpha1.WorkflowExecutionStatus) bool { - switch status { - case temporaliov1alpha1.WorkflowExecutionStatusFailed, - temporaliov1alpha1.WorkflowExecutionStatusCanceled, - temporaliov1alpha1.WorkflowExecutionStatusTerminated, - temporaliov1alpha1.WorkflowExecutionStatusTimedOut: - return true - default: - return false - } -} - func (r *WorkerDeploymentReconciler) generateStatus( ctx context.Context, l logr.Logger, @@ -57,25 +42,6 @@ func (r *WorkerDeploymentReconciler) generateStatus( // Continue without test workflow status } - // Emit a Warning event the first time a gate/test workflow is observed to have - // ended in a non-successful terminal state. Compare against the previous - // reconcile's recorded status (still on workerDeploy.Status at this point, since - // it hasn't been overwritten yet) so this doesn't re-fire on every loop. - prevStatusByWorkflowID := make(map[string]temporaliov1alpha1.WorkflowExecutionStatus, len(workerDeploy.Status.TargetVersion.TestWorkflows)) - for _, wf := range workerDeploy.Status.TargetVersion.TestWorkflows { - prevStatusByWorkflowID[wf.WorkflowID] = wf.Status - } - for _, wf := range testWorkflows { - if !isGateWorkflowTerminalFailure(wf.Status) { - continue - } - if prevStatusByWorkflowID[wf.WorkflowID] == wf.Status { - continue - } - r.Recorder.Eventf(workerDeploy, corev1.EventTypeWarning, ReasonGateWorkflowFailed, - "Gate/test workflow %s for version %s ended with status %s", wf.WorkflowID, targetBuildID, wf.Status) - } - // Add test workflow status to version info if it doesn't exist if versionInfo, exists := temporalState.Versions[targetBuildID]; exists { versionInfo.TestWorkflows = append(versionInfo.TestWorkflows, testWorkflows...) diff --git a/internal/controller/util.go b/internal/controller/util.go index 0f6525ce..e91fcbb5 100644 --- a/internal/controller/util.go +++ b/internal/controller/util.go @@ -28,7 +28,6 @@ const ( ReasonVersionPromotionFailed = "VersionPromotionFailed" ReasonMetadataUpdateFailed = "MetadataUpdateFailed" ReasonManagerIdentityClaimFailed = "ManagerIdentityClaimFailed" - ReasonGateWorkflowFailed = "GateWorkflowFailed" ) const ( From 03d55bd7c70808c599942db2a8461f050892b12a Mon Sep 17 00:00:00 2001 From: wankhede04 Date: Fri, 24 Jul 2026 12:37:56 +0530 Subject: [PATCH 5/6] refactor(api): fold poller health into ConditionReady instead of a new condition Per review feedback from @jaypipes: rather than introducing a new ConditionWorkersHealthy condition type, reuse the existing ConditionReady and factor poller health into its calculation. - Drop ConditionWorkersHealthy from api/v1alpha1/conditions.go. - ReasonNoActivePollers and ReasonPollerStatusUnknown are now set on ConditionReady=False/Unknown (instead of a separate condition) when the target version has otherwise completed rollout (Status=Current) but its workers are not confirmed to be actively polling Temporal. ReasonPollersHealthy remains for the Normal event emitted on recovery. - syncConditions now takes the TemporalWorkerState so it can look up PollerHealth/PollerHealthUnknown for the relevant build ID and reflect it in Ready before declaring rollout success; Progressing and the deprecated RolloutComplete condition are unaffected, since those track rollout state only. - workers_healthy.go now holds only the pure, unit-tested computePollerHealthCondition helper; the condition-setting/event logic that used to live in syncWorkersHealthyCondition moved into syncConditions. - Updated TestSyncConditions plus added NotReadyWhenCurrentVersionHasNoActivePollers and ReadyUnknownWhenCurrentVersionPollerStatusUnknown cases. --- api/v1alpha1/conditions.go | 6 -- api/v1alpha1/workerdeployment_types.go | 22 +++--- internal/controller/reconciler_events_test.go | 43 +++++++++-- internal/controller/worker_controller.go | 71 ++++++++++++++++--- internal/controller/workers_healthy.go | 62 ++-------------- 5 files changed, 118 insertions(+), 86 deletions(-) diff --git a/api/v1alpha1/conditions.go b/api/v1alpha1/conditions.go index 8572b55e..953ac72e 100644 --- a/api/v1alpha1/conditions.go +++ b/api/v1alpha1/conditions.go @@ -11,12 +11,6 @@ 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" - - // ConditionWorkersHealthy indicates whether workers for the version currently - // serving production traffic (or, before the first rollout completes, the target - // version) are actively polling Temporal -- as opposed to merely being Ready at - // the Kubernetes level. - ConditionWorkersHealthy = "WorkersHealthy" ) // Deprecated condition type constants. Maintained for backward compatibility with diff --git a/api/v1alpha1/workerdeployment_types.go b/api/v1alpha1/workerdeployment_types.go index f002ffc1..87187326 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/api/v1alpha1/workerdeployment_types.go @@ -144,18 +144,24 @@ const ( // Deprecated: Use ReasonRolloutComplete on ConditionReady instead. ReasonConnectionHealthy = "ConnectionHealthy" - // ReasonPollersHealthy is set on ConditionWorkersHealthy=True when all known - // task queues for the version have at least one active poller. + // ReasonPollersHealthy is used on ConditionReady=True (in place of + // ReasonRolloutComplete) and on the Normal Event emitted when a version's workers + // transition from having no active pollers (or unknown status) back to actively + // polling every known task queue. ReasonPollersHealthy = "PollersHealthy" - // ReasonNoActivePollers is set on ConditionWorkersHealthy=False when one or more - // of a version's task queues have no active poller. + // ReasonNoActivePollers is set on ConditionReady=False -- even though the target + // version has otherwise completed rollout and become current -- when one or more + // of the version's task queues have no active poller. A Deployment can be fully + // Ready at the Kubernetes level while its workers are misconfigured, stuck, or + // unable to reach Temporal; this reason distinguishes that case from a healthy + // rollout. ReasonNoActivePollers = "NoActivePollers" - // ReasonPollerStatusUnknown is set on ConditionWorkersHealthy=Unknown when poller - // status could not be determined (e.g. a transient DescribeTaskQueue error, or the - // version is not yet registered with Temporal). This must NOT be treated as - // unhealthy -- it means "don't know", not "broken". + // ReasonPollerStatusUnknown is set on ConditionReady=Unknown when poller status + // could not be determined for the current version (e.g. a transient + // DescribeTaskQueue error). This must NOT be treated as unhealthy -- it means + // "don't know", not "broken". ReasonPollerStatusUnknown = "PollerStatusUnknown" ) diff --git a/internal/controller/reconciler_events_test.go b/internal/controller/reconciler_events_test.go index e9056697..07c00694 100644 --- a/internal/controller/reconciler_events_test.go +++ b/internal/controller/reconciler_events_test.go @@ -18,6 +18,7 @@ import ( temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" "github.com/temporalio/temporal-worker-controller/internal/controller/clientpool" "github.com/temporalio/temporal-worker-controller/internal/planner" + "github.com/temporalio/temporal-worker-controller/internal/temporal" deploymentpb "go.temporal.io/api/deployment/v1" "go.temporal.io/api/serviceerror" "go.temporal.io/api/workflowservice/v1" @@ -344,7 +345,15 @@ func TestSyncConditions(t *testing.T) { t.Run("ReadyWhenVersionIsCurrent", func(t *testing.T) { twd := makeWD("test-worker", "default", "my-connection") twd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusCurrent - r.syncConditions(twd) + // An empty (but non-nil) PollerHealth map represents a version with no task + // queues seen yet to check -- vacuously healthy -- as opposed to nil, which + // means poller status was never checked at all (Unknown). + temporalState := &temporal.TemporalWorkerState{ + Versions: map[string]*temporal.VersionInfo{ + twd.Status.TargetVersion.BuildID: {PollerHealth: map[string]bool{}}, + }, + } + r.syncConditions(twd, temporalState) assertCondition(t, twd, temporaliov1alpha1.ConditionReady, metav1.ConditionTrue, temporaliov1alpha1.ReasonRolloutComplete) assertCondition(t, twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionFalse, temporaliov1alpha1.ReasonRolloutComplete) @@ -353,10 +362,36 @@ func TestSyncConditions(t *testing.T) { assertCondition(t, twd, temporaliov1alpha1.ConditionRolloutComplete, metav1.ConditionTrue, temporaliov1alpha1.ReasonRolloutComplete) //nolint:staticcheck // backward compat }) + t.Run("NotReadyWhenCurrentVersionHasNoActivePollers", func(t *testing.T) { + twd := makeWD("test-worker", "default", "my-connection") + twd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusCurrent + temporalState := &temporal.TemporalWorkerState{ + Versions: map[string]*temporal.VersionInfo{ + twd.Status.TargetVersion.BuildID: {PollerHealth: map[string]bool{"tq-1": false}}, + }, + } + r.syncConditions(twd, temporalState) + + assertCondition(t, twd, temporaliov1alpha1.ConditionReady, metav1.ConditionFalse, temporaliov1alpha1.ReasonNoActivePollers) + // Progressing is about rollout state, not poller health -- rollout has still completed. + assertCondition(t, twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionFalse, temporaliov1alpha1.ReasonRolloutComplete) + }) + + t.Run("ReadyUnknownWhenCurrentVersionPollerStatusUnknown", func(t *testing.T) { + twd := makeWD("test-worker", "default", "my-connection") + twd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusCurrent + temporalState := &temporal.TemporalWorkerState{ + Versions: map[string]*temporal.VersionInfo{}, + } + r.syncConditions(twd, temporalState) + + assertCondition(t, twd, temporaliov1alpha1.ConditionReady, metav1.ConditionUnknown, temporaliov1alpha1.ReasonPollerStatusUnknown) + }) + t.Run("ProgressingWhenVersionIsRamping", func(t *testing.T) { twd := makeWD("test-worker", "default", "my-connection") twd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusRamping - r.syncConditions(twd) + r.syncConditions(twd, nil) assertCondition(t, twd, temporaliov1alpha1.ConditionReady, metav1.ConditionFalse, temporaliov1alpha1.ReasonRamping) assertCondition(t, twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionTrue, temporaliov1alpha1.ReasonRamping) @@ -367,7 +402,7 @@ func TestSyncConditions(t *testing.T) { t.Run("ProgressingWhenVersionIsInactive", func(t *testing.T) { twd := makeWD("test-worker", "default", "my-connection") twd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusInactive - r.syncConditions(twd) + r.syncConditions(twd, nil) assertCondition(t, twd, temporaliov1alpha1.ConditionReady, metav1.ConditionFalse, temporaliov1alpha1.ReasonWaitingForPromotion) assertCondition(t, twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionTrue, temporaliov1alpha1.ReasonWaitingForPromotion) @@ -378,7 +413,7 @@ func TestSyncConditions(t *testing.T) { t.Run("ProgressingWhenVersionIsNotRegistered", func(t *testing.T) { twd := makeWD("test-worker", "default", "my-connection") twd.Status.TargetVersion.Status = temporaliov1alpha1.VersionStatusNotRegistered - r.syncConditions(twd) + r.syncConditions(twd, nil) assertCondition(t, twd, temporaliov1alpha1.ConditionReady, metav1.ConditionFalse, temporaliov1alpha1.ReasonWaitingForPollers) assertCondition(t, twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionTrue, temporaliov1alpha1.ReasonWaitingForPollers) diff --git a/internal/controller/worker_controller.go b/internal/controller/worker_controller.go index cbf641a5..824442b7 100644 --- a/internal/controller/worker_controller.go +++ b/internal/controller/worker_controller.go @@ -8,6 +8,7 @@ import ( "context" "errors" "fmt" + "strings" "time" "github.com/go-logr/logr" @@ -391,11 +392,11 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrl.Result{}, err } - // Derive Ready/Progressing from rollout state before the final write. - r.syncConditions(&workerDeploy) - // Surface poller health (are workers actually polling Temporal, not just Ready - // at the Kubernetes level) as a condition, separate from rollout progress. - r.syncWorkersHealthyCondition(&workerDeploy, temporalState) + // Derive Ready/Progressing from rollout state before the final write. When the + // target version has become current, this also factors in poller health (are + // workers actually polling Temporal, not just Ready at the Kubernetes level) + // before declaring Ready=True. + r.syncConditions(&workerDeploy, temporalState) // Single status write per reconcile: persists the generated status and // conditions set during this loop (Ready, Progressing). @@ -708,7 +709,10 @@ func (r *WorkerDeploymentReconciler) setCondition( // syncConditions sets Ready and Progressing based on the current rollout state. // It must be called at the end of a successful reconcile (no errors) so that // Progressing/Ready reflect the latest Temporal version status. -func (r *WorkerDeploymentReconciler) syncConditions(twd *temporaliov1alpha1.WorkerDeployment) { +func (r *WorkerDeploymentReconciler) syncConditions( + twd *temporaliov1alpha1.WorkerDeployment, + temporalState *temporal.TemporalWorkerState, +) { // Deprecated: set ConnectionHealthy=True on all successful reconciles for v1.3.x compat. r.setCondition(twd, temporaliov1alpha1.ConditionConnectionHealthy, //nolint:staticcheck // backward compat metav1.ConditionTrue, temporaliov1alpha1.ReasonConnectionHealthy, //nolint:staticcheck // backward compat @@ -716,13 +720,60 @@ func (r *WorkerDeploymentReconciler) syncConditions(twd *temporaliov1alpha1.Work switch twd.Status.TargetVersion.Status { case temporaliov1alpha1.VersionStatusCurrent: - r.setCondition(twd, temporaliov1alpha1.ConditionReady, - metav1.ConditionTrue, temporaliov1alpha1.ReasonRolloutComplete, - fmt.Sprintf("Rollout complete for buildID %s", twd.Status.TargetVersion.BuildID)) + // The rollout itself has completed, but before declaring Ready=True, confirm + // that workers for this version are actually polling Temporal -- as opposed to + // merely being Ready at the Kubernetes level. A Deployment can be fully Ready + // while its workers are misconfigured, stuck, or unable to reach Temporal. + buildID := twd.Status.TargetVersion.BuildID + if twd.Status.CurrentVersion != nil { + buildID = twd.Status.CurrentVersion.BuildID + } + var pollerHealth map[string]bool + var pollerHealthUnknown bool + if versionInfo, exists := temporalState.Versions[buildID]; exists { + pollerHealth = versionInfo.PollerHealth + pollerHealthUnknown = versionInfo.PollerHealthUnknown + } + pollerStatus, pollerReason, affectedQueues := computePollerHealthCondition(pollerHealth, pollerHealthUnknown) + + var readyStatus metav1.ConditionStatus + var readyReason, readyMessage string + switch pollerStatus { + case metav1.ConditionFalse: + readyStatus = metav1.ConditionFalse + readyReason = pollerReason + readyMessage = fmt.Sprintf("Version %s has no active pollers on task queue(s): %s", buildID, strings.Join(affectedQueues, ", ")) + case metav1.ConditionUnknown: + readyStatus = metav1.ConditionUnknown + readyReason = pollerReason + readyMessage = fmt.Sprintf("Poller status for version %s could not be determined", buildID) + default: + readyStatus = metav1.ConditionTrue + readyReason = temporaliov1alpha1.ReasonRolloutComplete + readyMessage = fmt.Sprintf("Rollout complete for buildID %s", buildID) + } + + readyChanged := r.setCondition(twd, temporaliov1alpha1.ConditionReady, readyStatus, readyReason, readyMessage) + if readyChanged { + switch readyStatus { + case metav1.ConditionFalse: + r.Recorder.Eventf(twd, corev1.EventTypeWarning, temporaliov1alpha1.ReasonNoActivePollers, + "Version %s has no active pollers on task queue(s): %s", buildID, strings.Join(affectedQueues, ", ")) + case metav1.ConditionTrue: + r.Recorder.Eventf(twd, corev1.EventTypeNormal, temporaliov1alpha1.ReasonPollersHealthy, + "Version %s pollers are healthy", buildID) + case metav1.ConditionUnknown: + // Don't emit an event for Unknown -- it just means "don't know yet", + // not a state transition worth alerting on. + } + } + r.setCondition(twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionFalse, temporaliov1alpha1.ReasonRolloutComplete, fmt.Sprintf("Target version %s is current", twd.Status.TargetVersion.BuildID)) - // Deprecated: set RolloutComplete=True for v1.3.x compat. + // Deprecated: set RolloutComplete=True for v1.3.x compat. This deliberately + // mirrors rollout completion only, not poller health, matching its pre-existing + // (Kubernetes-readiness-only) semantics for v1.3.x compat consumers. r.setCondition(twd, temporaliov1alpha1.ConditionRolloutComplete, //nolint:staticcheck // backward compat metav1.ConditionTrue, temporaliov1alpha1.ReasonRolloutComplete, fmt.Sprintf("Rollout complete for buildID %s", twd.Status.TargetVersion.BuildID)) diff --git a/internal/controller/workers_healthy.go b/internal/controller/workers_healthy.go index 6767666b..6b06bbb2 100644 --- a/internal/controller/workers_healthy.go +++ b/internal/controller/workers_healthy.go @@ -5,19 +5,16 @@ package controller import ( - "fmt" "sort" - "strings" temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" - "github.com/temporalio/temporal-worker-controller/internal/temporal" - corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -// computePollerHealthCondition derives the ConditionWorkersHealthy status/reason from -// a version's poller health data. It is a pure function so the decision logic can be -// unit tested without an envtest environment. +// computePollerHealthCondition derives a poller-health status/reason from a version's +// poller health data, for use as part of the ConditionReady calculation (see +// syncConditions). It is a pure function so the decision logic can be unit tested +// without an envtest environment. // // - pollerHealth == nil: poller status was never checked for this version (e.g. not // yet registered with Temporal) -> Unknown. @@ -48,54 +45,3 @@ func computePollerHealthCondition( } return metav1.ConditionTrue, temporaliov1alpha1.ReasonPollersHealthy, nil } - -// syncWorkersHealthyCondition sets ConditionWorkersHealthy from poller health observed -// for the version currently serving production traffic (CurrentVersion), falling back -// to TargetVersion before the first rollout has ever completed (CurrentVersion nil). -// It emits a Normal/Warning event only when the condition actually transitions, to -// avoid spamming an Event on every reconcile loop. -func (r *WorkerDeploymentReconciler) syncWorkersHealthyCondition( - workerDeploy *temporaliov1alpha1.WorkerDeployment, - temporalState *temporal.TemporalWorkerState, -) { - buildID := workerDeploy.Status.TargetVersion.BuildID - if workerDeploy.Status.CurrentVersion != nil { - buildID = workerDeploy.Status.CurrentVersion.BuildID - } - if buildID == "" { - return - } - - var pollerHealth map[string]bool - var unknown bool - if versionInfo, exists := temporalState.Versions[buildID]; exists { - pollerHealth = versionInfo.PollerHealth - unknown = versionInfo.PollerHealthUnknown - } - - status, reason, affectedQueues := computePollerHealthCondition(pollerHealth, unknown) - - var message string - switch status { - case metav1.ConditionFalse: - message = fmt.Sprintf("Version %s has no active pollers on task queue(s): %s", buildID, strings.Join(affectedQueues, ", ")) - case metav1.ConditionTrue: - message = fmt.Sprintf("Version %s pollers are healthy", buildID) - default: - message = fmt.Sprintf("Poller status for version %s could not be determined", buildID) - } - - changed := r.setCondition(workerDeploy, temporaliov1alpha1.ConditionWorkersHealthy, status, reason, message) - if !changed { - return - } - - switch status { - case metav1.ConditionFalse: - r.Recorder.Eventf(workerDeploy, corev1.EventTypeWarning, temporaliov1alpha1.ReasonNoActivePollers, - "Version %s has no active pollers on task queue(s): %s", buildID, strings.Join(affectedQueues, ", ")) - case metav1.ConditionTrue: - r.Recorder.Eventf(workerDeploy, corev1.EventTypeNormal, temporaliov1alpha1.ReasonPollersHealthy, - "Version %s pollers are healthy", buildID) - } -} From 844a2be9e364b000a423d67802f6f4d115c9c6b5 Mon Sep 17 00:00:00 2001 From: wankhede04 Date: Fri, 24 Jul 2026 12:38:08 +0530 Subject: [PATCH 6/6] fix(temporal): stop checking further task queues after a DescribeTaskQueue error Per review feedback from @jaypipes: a DescribeTaskQueue failure is most likely a rate limit or a network partition/connectivity issue, so continuing on to call DescribeTaskQueue for the remaining task queues right after one of those errors would only make things worse. computePollerHealth now returns an error and stops on the first getPollers failure instead of setting an 'unknown' flag and continuing the loop. Task queues already checked before the error are still returned, and callers set PollerHealthUnknown=true on a non-nil error, preserving the existing 'unknown, not unhealthy' semantics. --- internal/temporal/worker_deployment.go | 29 ++++++++++++++++---------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/internal/temporal/worker_deployment.go b/internal/temporal/worker_deployment.go index 20e27576..6372bb0d 100644 --- a/internal/temporal/worker_deployment.go +++ b/internal/temporal/worker_deployment.go @@ -198,7 +198,9 @@ func GetWorkerDeploymentState( BuildID: version.DeploymentVersion.BuildId, }) if descErr == nil { - versionInfo.PollerHealth, versionInfo.PollerHealthUnknown = computePollerHealth(ctx, client, currentDesc.Info.TaskQueuesInfos) + var pollerErr error + versionInfo.PollerHealth, pollerErr = computePollerHealth(ctx, client, currentDesc.Info.TaskQueuesInfos) //nolint:revive // TODO(carlydf): consider logging this error + versionInfo.PollerHealthUnknown = pollerErr != nil } else { versionInfo.PollerHealthUnknown = true } @@ -257,8 +259,9 @@ func GetTestWorkflowStatus( // Poller health for the target version, computed from the task queues already // fetched above via DescribeVersion (no additional per-version round trip). - temporalState.Versions[buildID].PollerHealth, temporalState.Versions[buildID].PollerHealthUnknown = - computePollerHealth(ctx, client, versionResp.Info.TaskQueuesInfos) + var pollerErr error + temporalState.Versions[buildID].PollerHealth, pollerErr = computePollerHealth(ctx, client, versionResp.Info.TaskQueuesInfos) //nolint:revive // TODO(carlydf): consider logging this error + temporalState.Versions[buildID].PollerHealthUnknown = pollerErr != nil // Check test workflows for each task queue for _, tq := range versionResp.Info.TaskQueuesInfos { @@ -366,24 +369,28 @@ func getPollers(ctx context.Context, } // computePollerHealth reports, per task queue, whether at least one poller was -// observed. A task queue is omitted from the returned map (and unknown is set -// to true) if its poller status could not be determined, e.g. a transient -// DescribeTaskQueue error -- this must not be interpreted as unhealthy. +// observed. It stops and returns an error on the first DescribeTaskQueue failure +// rather than continuing on to the remaining task queues: such an error is most +// likely a rate limit or a network partition/connectivity failure, and continuing +// to hammer Temporal with more DescribeTaskQueue calls right after one of those +// errors would only make things worse. Task queues successfully checked before the +// error are still returned in health. Callers must not interpret a non-nil error +// (or a task queue missing from health) as unhealthy -- it means "don't know", not +// "broken". func computePollerHealth( ctx context.Context, client temporalClient.Client, tqs []temporalClient.WorkerDeploymentTaskQueueInfo, -) (health map[string]bool, unknown bool) { +) (health map[string]bool, err error) { health = make(map[string]bool, len(tqs)) for _, tqInfo := range tqs { pollers, err := getPollers(ctx, client, tqInfo) - if err != nil { //nolint:revive // TODO(carlydf): consider logging this error - unknown = true - continue + if err != nil { + return health, err } health[tqInfo.Name] = len(pollers) > 0 } - return health, unknown + return health, nil } func allTaskQueuesHaveUnversionedPoller(