From 243cbafb7129e8d18304793119e809613e11dcc7 Mon Sep 17 00:00:00 2001 From: frossbeamish <6519237+frossbeamish@users.noreply.github.com> Date: Fri, 17 Jul 2026 16:35:03 -0700 Subject: [PATCH] fix(controller): re-apply worker resources when tracked object is missing The WRT apply path skips the SSA apply whenever the rendered hash matches the LastAppliedHash recorded in the WRT status. The hash alone is not a safe signal: when a version is sunset its rendered resources are deleted, but the status entry for that build ID can survive (the status update that would drop it can hit a conflict, or the build can be re-registered before the next successful status write). If the same build ID returns, the stale hash matches the unchanged render and the resource is never re-created. For a WRT rendering a ScaledObject this leaves the returning version's Deployment with no autoscaler: the controller starts it at 1 replica expecting the scaler to own replicas, and nothing ever scales it. Guard the skip with an existence check on the rendered object (a cache read), falling through to the SSA apply on NotFound or on an indeterminate error. Extract the WRT apply block into applyWorkerResourceTemplates so it is unit-testable in isolation. --- internal/controller/execplan.go | 59 +++++++-- internal/controller/execplan_wrt_test.go | 121 ++++++++++++++++++ internal/controller/reconciler_events_test.go | 2 +- 3 files changed, 171 insertions(+), 11 deletions(-) create mode 100644 internal/controller/execplan_wrt_test.go diff --git a/internal/controller/execplan.go b/internal/controller/execplan.go index 466ad814..357f11e1 100644 --- a/internal/controller/execplan.go +++ b/internal/controller/execplan.go @@ -23,6 +23,7 @@ import ( appsv1 "k8s.io/api/apps/v1" autoscalingv1 "k8s.io/api/autoscaling/v1" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" @@ -337,9 +338,14 @@ func (r *WorkerDeploymentReconciler) executePlan(ctx context.Context, l logr.Log } } - // Apply worker resource templates via Server-Side Apply. - // Partial failure isolation: all resources are attempted even if some fail; - // errors are collected and returned together. + return r.applyWorkerResourceTemplates(ctx, l, p) +} + +// applyWorkerResourceTemplates applies rendered worker resource templates via +// Server-Side Apply and records per-Build-ID results in each WRT's status. +// Partial failure isolation: all resources are attempted even if some fail; +// errors are collected and returned together. +func (r *WorkerDeploymentReconciler) applyWorkerResourceTemplates(ctx context.Context, l logr.Logger, p *plan) error { type wrtKey struct{ namespace, name string } type applyResult struct { buildID string @@ -370,14 +376,47 @@ func (r *WorkerDeploymentReconciler) executePlan(ctx context.Context, l logr.Log // successfully applied. This avoids unnecessary API server load at scale // (hundreds of TWDs × hundreds of versions × multiple WRTs). // An empty RenderedHash means hashing failed; always apply in that case. + // + // The hash match alone is not sufficient: the tracked resource may have been + // deleted since the hash was recorded. In particular, when a version is + // sunset its rendered resources are deleted (DeleteWorkerResources), but the + // WRT status entry for that build can survive — e.g. the status update that + // would drop it hits a conflict, or the version is re-registered before the + // next successful status write. If the same build ID then comes back, the + // stale LastAppliedHash matches the unchanged render and the resource is + // never re-created, leaving the returning version without its scaler + // (observed as a current version pinned at 1 replica with no ScaledObject). + // Guard the skip with an existence check on the rendered object. if apply.RenderedHash != "" && apply.RenderedHash == apply.LastAppliedHash { - wrtResults[key] = append(wrtResults[key], applyResult{ - buildID: apply.BuildID, - resourceName: apply.Resource.GetName(), - hash: apply.RenderedHash, - skipped: true, - }) - continue + existing := &unstructured.Unstructured{} + existing.SetGroupVersionKind(apply.Resource.GroupVersionKind()) + getErr := r.Get(ctx, types.NamespacedName{ + Namespace: apply.Resource.GetNamespace(), + Name: apply.Resource.GetName(), + }, existing) + if getErr == nil { + wrtResults[key] = append(wrtResults[key], applyResult{ + buildID: apply.BuildID, + resourceName: apply.Resource.GetName(), + hash: apply.RenderedHash, + skipped: true, + }) + continue + } + if !apierrors.IsNotFound(getErr) { + l.Error(getErr, "unable to confirm worker resource exists; re-applying", + "name", apply.Resource.GetName(), + "kind", apply.Resource.GetKind(), + ) + } else { + l.Info("worker resource missing despite unchanged hash; re-applying", + "name", apply.Resource.GetName(), + "kind", apply.Resource.GetKind(), + "buildID", apply.BuildID, + ) + } + // Fall through to the SSA apply: it is create-or-update, so re-applying + // is safe in both the NotFound and the indeterminate-error case. } l.Info("applying rendered worker resource template", diff --git a/internal/controller/execplan_wrt_test.go b/internal/controller/execplan_wrt_test.go new file mode 100644 index 00000000..0d684e63 --- /dev/null +++ b/internal/controller/execplan_wrt_test.go @@ -0,0 +1,121 @@ +// 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 ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" + "github.com/temporalio/temporal-worker-controller/internal/planner" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" +) + +// newRenderedConfigMap returns an unstructured ConfigMap standing in for a +// rendered WRT resource (the controller treats rendered objects generically). +func newRenderedConfigMap(namespace, name string) *unstructured.Unstructured { + u := &unstructured.Unstructured{} + u.SetAPIVersion("v1") + u.SetKind("ConfigMap") + u.SetNamespace(namespace) + u.SetName(name) + return u +} + +func newTestWRT(namespace, name string) *temporaliov1alpha1.WorkerResourceTemplate { + return &temporaliov1alpha1.WorkerResourceTemplate{ + ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, + } +} + +func newTestTWD(namespace, name string) *temporaliov1alpha1.WorkerDeployment { + return &temporaliov1alpha1.WorkerDeployment{ + ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, + } +} + +// A version whose rendered hash matches the recorded LastAppliedHash but whose +// rendered resource is absent from the cluster (version sunset deletes rendered +// resources, and the build ID can be re-registered while its stale status entry +// survives) must be re-applied, not skipped — otherwise the returning build ID +// runs without its rendered resources. +func TestExecutePlan_WRTApply_ReappliesWhenResourceMissing(t *testing.T) { + const ns = "default" + wrt := newTestWRT(ns, "test-wrt") + twd := newTestTWD(ns, "test-twd") + + r, _ := newTestReconciler([]client.Object{wrt, twd}) + + rendered := newRenderedConfigMap(ns, "test-wrt-rendered") + p := &plan{ + ApplyWorkerResources: []planner.WorkerResourceApply{{ + Resource: rendered, + WRTName: wrt.Name, + WRTNamespace: ns, + BuildID: "build-1", + RenderedHash: "hash-1", + LastAppliedHash: "hash-1", // matches, but the resource is gone + }}, + } + + err := r.applyWorkerResourceTemplates(context.Background(), ctrl.Log, p) + require.NoError(t, err) + + // The rendered resource must have been re-created despite the hash match. + got := &corev1.ConfigMap{} + require.NoError(t, r.Get(context.Background(), + types.NamespacedName{Namespace: ns, Name: "test-wrt-rendered"}, got)) + + // The WRT status must record the build as applied (not skipped-with-stale-state). + gotWRT := &temporaliov1alpha1.WorkerResourceTemplate{} + require.NoError(t, r.Get(context.Background(), + types.NamespacedName{Namespace: ns, Name: wrt.Name}, gotWRT)) + require.Len(t, gotWRT.Status.Versions, 1) + assert.Equal(t, "build-1", gotWRT.Status.Versions[0].BuildID) + assert.Equal(t, "hash-1", gotWRT.Status.Versions[0].LastAppliedHash) +} + +// The skip fast-path must still hold when the resource exists and the hash is +// unchanged: no SSA apply call is made. +func TestExecutePlan_WRTApply_SkipsWhenResourceExistsAndHashUnchanged(t *testing.T) { + const ns = "default" + wrt := newTestWRT(ns, "test-wrt") + twd := newTestTWD(ns, "test-twd") + existing := newRenderedConfigMap(ns, "test-wrt-rendered") + + patchCalls := 0 + r, _ := newTestReconcilerWithInterceptors( + []client.Object{wrt, twd, existing}, + interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + patchCalls++ + return c.Patch(ctx, obj, patch, opts...) + }, + }, + ) + + p := &plan{ + ApplyWorkerResources: []planner.WorkerResourceApply{{ + Resource: newRenderedConfigMap(ns, "test-wrt-rendered"), + WRTName: wrt.Name, + WRTNamespace: ns, + BuildID: "build-1", + RenderedHash: "hash-1", + LastAppliedHash: "hash-1", + }}, + } + + err := r.applyWorkerResourceTemplates(context.Background(), ctrl.Log, p) + require.NoError(t, err) + assert.Zero(t, patchCalls, "hash-unchanged apply with existing resource must be skipped") +} diff --git a/internal/controller/reconciler_events_test.go b/internal/controller/reconciler_events_test.go index e9056697..94134c46 100644 --- a/internal/controller/reconciler_events_test.go +++ b/internal/controller/reconciler_events_test.go @@ -59,7 +59,7 @@ func newTestReconcilerWithInterceptors(objs []client.Object, funcs interceptor.F fakeClient := fake.NewClientBuilder(). WithScheme(scheme). WithObjects(objs...). - WithStatusSubresource(&temporaliov1alpha1.WorkerDeployment{}). + WithStatusSubresource(&temporaliov1alpha1.WorkerDeployment{}, &temporaliov1alpha1.WorkerResourceTemplate{}). WithIndex(&appsv1.Deployment{}, deployOwnerKey, func(rawObj client.Object) []string { deploy := rawObj.(*appsv1.Deployment) owner := metav1.GetControllerOf(deploy)