feat(controller): surface poller and gate-workflow health as conditions and events

api/v1alpha1/workerdeployment_types.go

@@ -101,9 +101,11 @@ const (
 	// when the target version has been successfully registered as the current version.
 	ReasonRolloutComplete = "RolloutComplete"
 
-	// 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
+	// version is not yet registered with Temporal, and (2) the target version has
+	// become current but one or more of its task queues have no active poller.
 	ReasonWaitingForPollers = "WaitingForPollers"
 
 	// ReasonWaitingForPromotion is set on ConditionProgressing=True when the target
@@ -143,6 +145,22 @@ const (
 
 	// Deprecated: Use ReasonRolloutComplete on ConditionReady instead.
 	ReasonConnectionHealthy = "ConnectionHealthy"
+
+	// ReasonActivePollers is set on ConditionProgressing=False when the target
+	// version has become current and all known task queues have at least one
+	// active poller. This mirrors ReasonWaitingForPollers on
+	// ConditionProgressing=True: it reports only that workers are actively
+	// polling the server, not that the pollers themselves are otherwise healthy.
+	// Also used as the reason on the Normal Event emitted when a version
+	// transitions from having no active pollers (or unknown status) back to
+	// actively polling every known task queue.
+	ReasonActivePollers = "ActivePollers"
+
+	// ReasonPollerStatusUnknown is set on ConditionProgressing=False when poller
+	// status could not be determined for the current version (e.g. a transient
+	// DescribeTaskQueue error). This must NOT be treated as "no pollers" -- it
+	// means "don't know", not "broken".
+	ReasonPollerStatusUnknown = "PollerStatusUnknown"
 )
 
 // VersionStatus indicates the status of a version.

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"
@@ -348,19 +349,54 @@ 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 active -- 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)
+		assertCondition(t, twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionFalse, temporaliov1alpha1.ReasonActivePollers)
 		// Deprecated conditions
 		assertCondition(t, twd, temporaliov1alpha1.ConditionConnectionHealthy, metav1.ConditionTrue, temporaliov1alpha1.ReasonConnectionHealthy) //nolint:staticcheck // backward compat
 		assertCondition(t, twd, temporaliov1alpha1.ConditionRolloutComplete, metav1.ConditionTrue, temporaliov1alpha1.ReasonRolloutComplete)     //nolint:staticcheck // backward compat
 	})
 
+	t.Run("ProgressingWhenCurrentVersionHasNoActivePollers", 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)
+
+		// Ready stays about rollout completion; poller presence is surfaced on Progressing.
+		assertCondition(t, twd, temporaliov1alpha1.ConditionReady, metav1.ConditionTrue, temporaliov1alpha1.ReasonRolloutComplete)
+		assertCondition(t, twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionTrue, temporaliov1alpha1.ReasonWaitingForPollers)
+	})
+
+	t.Run("ProgressingUnknownWhenCurrentVersionPollerStatusUnknown", 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.ConditionTrue, temporaliov1alpha1.ReasonRolloutComplete)
+		assertCondition(t, twd, temporaliov1alpha1.ConditionProgressing, metav1.ConditionFalse, 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)
@@ -371,7 +407,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)
@@ -382,7 +418,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)

internal/controller/worker_controller.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -372,8 +373,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)
+	// Derive Ready/Progressing from rollout state before the final write. When the
+	// target version has become current, this also factors in whether workers are
+	// actively polling Temporal into ConditionProgressing (Ready itself remains
+	// about rollout completion only).
+	r.syncConditions(&workerDeploy, temporalState)
 
 	// Single status write per reconcile: persists the generated status and
 	// conditions set during this loop (Ready, Progressing).
@@ -672,14 +676,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,
@@ -691,21 +696,68 @@ 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
 		"Connection is healthy and auth secret is resolved")
 
 	switch twd.Status.TargetVersion.Status {
 	case temporaliov1alpha1.VersionStatusCurrent:
+		// Rollout itself has completed — Ready stays True regardless of poller
+		// presence. Poller presence is surfaced on ConditionProgressing instead,
+		// mirroring the existing ReasonWaitingForPollers (Progressing=True) /
+		// ReasonActivePollers (Progressing=False) pair.
 		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))
-		// Deprecated: set RolloutComplete=True for v1.3.x compat.
+
+		buildID := twd.Status.TargetVersion.BuildID
+		if twd.Status.CurrentVersion != nil {
+			buildID = twd.Status.CurrentVersion.BuildID
+		}
+		var pollerHealth map[string]bool
+		var pollerHealthUnknown bool
+		if temporalState != nil {
+			if versionInfo, exists := temporalState.Versions[buildID]; exists {
+				pollerHealth = versionInfo.PollerHealth
+				pollerHealthUnknown = versionInfo.PollerHealthUnknown
+			}
+		}
+		progressingStatus, progressingReason, affectedQueues := computePollerHealthCondition(pollerHealth, pollerHealthUnknown)
+
+		var progressingMessage string
+		switch progressingReason {
+		case temporaliov1alpha1.ReasonWaitingForPollers:
+			progressingMessage = fmt.Sprintf("Version %s has no active pollers on task queue(s): %s", buildID, strings.Join(affectedQueues, ", "))
+		case temporaliov1alpha1.ReasonActivePollers:
+			progressingMessage = fmt.Sprintf("Version %s has active pollers on all known task queues", buildID)
+		default:
+			progressingMessage = fmt.Sprintf("Poller status for version %s could not be determined", buildID)
+		}
+
+		progressingChanged := r.setCondition(twd, temporaliov1alpha1.ConditionProgressing, progressingStatus, progressingReason, progressingMessage)
+		if progressingChanged {
+			switch progressingReason {
+			case temporaliov1alpha1.ReasonWaitingForPollers:
+				r.Recorder.Eventf(twd, corev1.EventTypeWarning, temporaliov1alpha1.ReasonWaitingForPollers,
+					"Version %s has no active pollers on task queue(s): %s", buildID, strings.Join(affectedQueues, ", "))
+			case temporaliov1alpha1.ReasonActivePollers:
+				r.Recorder.Eventf(twd, corev1.EventTypeNormal, temporaliov1alpha1.ReasonActivePollers,
+					"Version %s has active pollers on all known task queues", buildID)
+			case temporaliov1alpha1.ReasonPollerStatusUnknown:
+				// Don't emit an event for Unknown -- it just means "don't know yet",
+				// not a state transition worth alerting on.
+			}
+		}
+
+		// Deprecated: set RolloutComplete=True for v1.3.x compat. This deliberately
+		// mirrors rollout completion only, not poller presence, 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))

internal/controller/workers_healthy.go

@@ -0,0 +1,50 @@
+// 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 (
+	"sort"
+
+	temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// computePollerHealthCondition derives a ConditionProgressing status/reason from a
+// version's poller presence data, for use as part of the ConditionProgressing
+// calculation when the target version is Current (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) -> Progressing=False, ReasonPollerStatusUnknown.
+//   - any task queue with a false value -> Progressing=True, ReasonWaitingForPollers,
+//     naming the affected queues. A known missing-poller problem takes precedence
+//     over an unrelated unknown elsewhere.
+//   - no false values, but unknown == true (some task queues errored) ->
+//     Progressing=False, ReasonPollerStatusUnknown.
+//   - all task queues true, unknown == false -> Progressing=False, ReasonActivePollers.
+func computePollerHealthCondition(
+	pollerHealth map[string]bool,
+	unknown bool,
+) (status metav1.ConditionStatus, reason string, affectedQueues []string) {
+	if pollerHealth == nil {
+		return metav1.ConditionFalse, temporaliov1alpha1.ReasonPollerStatusUnknown, nil
+	}
+
+	for tq, hasPoller := range pollerHealth {
+		if !hasPoller {
+			affectedQueues = append(affectedQueues, tq)
+		}
+	}
+
+	if len(affectedQueues) > 0 {
+		sort.Strings(affectedQueues)
+		return metav1.ConditionTrue, temporaliov1alpha1.ReasonWaitingForPollers, affectedQueues
+	}
+	if unknown {
+		return metav1.ConditionFalse, temporaliov1alpha1.ReasonPollerStatusUnknown, nil
+	}
+	return metav1.ConditionFalse, temporaliov1alpha1.ReasonActivePollers, nil
+}

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.ConditionFalse,
+			wantReason:   temporaliov1alpha1.ReasonPollerStatusUnknown,
+		},
+		{
+			name:         "all queues have active pollers",
+			pollerHealth: map[string]bool{"tq-1": true, "tq-2": true},
+			unknown:      false,
+			wantStatus:   metav1.ConditionFalse,
+			wantReason:   temporaliov1alpha1.ReasonActivePollers,
+		},
+		{
+			name:         "one queue has no pollers",
+			pollerHealth: map[string]bool{"tq-1": true, "tq-2": false},
+			unknown:      false,
+			wantStatus:   metav1.ConditionTrue,
+			wantReason:   temporaliov1alpha1.ReasonWaitingForPollers,
+			wantAffected: []string{"tq-2"},
+		},
+		{
+			name:         "empty map with no fetch errors is active (no task queues to check)",
+			pollerHealth: map[string]bool{},
+			unknown:      false,
+			wantStatus:   metav1.ConditionFalse,
+			wantReason:   temporaliov1alpha1.ReasonActivePollers,
+		},
+		{
+			name:         "fetch error with no confirmed-missing poller is unknown, not waiting",
+			pollerHealth: map[string]bool{"tq-1": true},
+			unknown:      true,
+			wantStatus:   metav1.ConditionFalse,
+			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.ConditionTrue,
+			wantReason:   temporaliov1alpha1.ReasonWaitingForPollers,
+			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)
+		})
+	}
+}