Skip to content
20 changes: 20 additions & 0 deletions api/v1alpha1/workerdeployment_types.go
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,26 @@ const (

// Deprecated: Use ReasonRolloutComplete on ConditionReady instead.
ReasonConnectionHealthy = "ConnectionHealthy"

// 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"
Comment on lines +147 to +151

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please change this to ReasonActivePollers. The actual health of the poller isn't known :) Just that it's actively polling the server.

Also, I have changed my mind about having this reason affect ConditionReady. Instead, let's make this reason only impact ConditionProgressing=False so that it matches the existing ReasonWaitingForPollers that is set on ConditionProgressing=True.


// 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"

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm wondering if we need ReasonNoActivePollers at all since we have the existing ReasonWaitingForPollers. thoughts?


// 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"
)

// VersionStatus indicates the status of a version.
Expand Down
43 changes: 39 additions & 4 deletions internal/controller/reconciler_events_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -348,7 +349,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)
Expand All @@ -357,10 +366,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)
Expand All @@ -371,7 +406,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)
Expand All @@ -382,7 +417,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)
Expand Down
75 changes: 65 additions & 10 deletions internal/controller/worker_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"context"
"errors"
"fmt"
"strings"
"time"

"github.com/go-logr/logr"
Expand Down Expand Up @@ -391,8 +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)
// 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).
Expand Down Expand Up @@ -685,14 +689,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,
Expand All @@ -704,21 +709,71 @@ 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:
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))
Expand Down
47 changes: 47 additions & 0 deletions internal/controller/workers_healthy.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
// 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 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.
// - 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
}
78 changes: 78 additions & 0 deletions internal/controller/workers_healthy_test.go
Original file line number Diff line number Diff line change
@@ -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)
})
}
}
Loading
Loading