diff --git a/api/v1alpha1/workerdeployment_types.go b/api/v1alpha1/workerdeployment_types.go index 2f4930de..87187326 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/api/v1alpha1/workerdeployment_types.go @@ -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" + + // 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 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. diff --git a/internal/controller/reconciler_events_test.go b/internal/controller/reconciler_events_test.go index 92b374cd..d76d8d03 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" @@ -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) @@ -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) @@ -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) @@ -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) diff --git a/internal/controller/worker_controller.go b/internal/controller/worker_controller.go index c0acb5cc..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,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). @@ -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, @@ -704,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 @@ -712,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 new file mode 100644 index 00000000..6b06bbb2 --- /dev/null +++ b/internal/controller/workers_healthy.go @@ -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 +} 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) + }) + } +} diff --git a/internal/temporal/worker_deployment.go b/internal/temporal/worker_deployment.go index b05eaf40..6372bb0d 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,24 @@ 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 { + 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 + } + } + state.Versions[version.DeploymentVersion.BuildId] = versionInfo } @@ -228,6 +257,12 @@ 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). + 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 { // Skip non-workflow task queues @@ -333,6 +368,31 @@ func getPollers(ctx context.Context, return resp.GetPollers(), nil } +// computePollerHealth reports, per task queue, whether at least one poller was +// 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, err error) { + health = make(map[string]bool, len(tqs)) + for _, tqInfo := range tqs { + pollers, err := getPollers(ctx, client, tqInfo) + if err != nil { + return health, err + } + health[tqInfo.Name] = len(pollers) > 0 + } + return health, nil +} + func allTaskQueuesHaveUnversionedPoller( ctx context.Context, client temporalClient.Client,