diff --git a/internal/controller/execplan.go b/internal/controller/execplan.go index 6e9d8607..a60a42ff 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" @@ -351,9 +352,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 @@ -384,14 +390,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") +}