diff --git a/api/v1alpha1/workerdeployment_types.go b/api/v1alpha1/workerdeployment_types.go index 2f4930de..52542944 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/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. diff --git a/internal/controller/reconciler_events_test.go b/internal/controller/reconciler_events_test.go index dc7fd60c..a8ae079d 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,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) diff --git a/internal/controller/worker_controller.go b/internal/controller/worker_controller.go index a657fc36..74702749 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" @@ -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,7 +696,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 @@ -699,13 +707,57 @@ func (r *WorkerDeploymentReconciler) syncConditions(twd *temporaliov1alpha1.Work 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)) diff --git a/internal/controller/workers_healthy.go b/internal/controller/workers_healthy.go new file mode 100644 index 00000000..7c8d38cd --- /dev/null +++ b/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 +} diff --git a/internal/controller/workers_healthy_test.go b/internal/controller/workers_healthy_test.go new file mode 100644 index 00000000..3b7c8367 --- /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.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) + }) + } +} 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,