diff --git a/api/v1alpha1/workerdeployment_cel_validation_test.go b/api/v1alpha1/workerdeployment_cel_validation_test.go index 0723cd3d..1ba5dca6 100644 --- a/api/v1alpha1/workerdeployment_cel_validation_test.go +++ b/api/v1alpha1/workerdeployment_cel_validation_test.go @@ -64,6 +64,22 @@ var _ = Describe("WorkerDeployment CRD CEL validation", func() { Expect(err.Error()).To(ContainSubstring("spec.drainingDeployment.replicas")) }) + It("accepts custom deploymentLabels", func() { + twd := baseTWD("custom-deployment-labels") + twd.Spec.DeploymentLabels = map[string]string{"team": "payments"} + Expect(k8sClient.Create(ctx, twd)).To(Succeed()) + }) + + It("rejects deploymentLabels that set controller-managed keys", func() { + for i, key := range []string{"temporal.io/build-id", "temporal.io/deployment-name"} { + twd := baseTWD(fmt.Sprintf("reserved-deployment-label-%d", i)) + twd.Spec.DeploymentLabels = map[string]string{key: "user-supplied"} + err := k8sClient.Create(ctx, twd) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("are managed by the controller")) + } + }) + It("rejects name longer than 63 characters", func() { twd := baseTWD(strings.Repeat("a", 64)) err := k8sClient.Create(ctx, twd) diff --git a/api/v1alpha1/workerdeployment_types.go b/api/v1alpha1/workerdeployment_types.go index 8e8b6fdd..fd4a9ab3 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/api/v1alpha1/workerdeployment_types.go @@ -103,6 +103,16 @@ type WorkerDeploymentSpec struct { // +optional DrainingDeployment *DrainingDeploymentSpec `json:"drainingDeployment,omitempty"` + // DeploymentLabels are added to the metadata of each versioned Kubernetes + // Deployment that the controller creates. Pod labels belong in the pod + // template instead. Use this for tools that act on the Deployment object + // itself, such as admission webhooks that opt workloads in or out by label. + // Changes are applied in place to the Deployments of existing versions, + // without restarting their pods. + // +optional + // +kubebuilder:validation:XValidation:rule="!('temporal.io/build-id' in self) && !('temporal.io/deployment-name' in self)",message="temporal.io/build-id and temporal.io/deployment-name are managed by the controller" + DeploymentLabels map[string]string `json:"deploymentLabels,omitempty"` + // How to rollout new workflow executions to the target version. RolloutStrategy RolloutStrategy `json:"rollout"` diff --git a/api/v1alpha1/workerdeployment_webhook.go b/api/v1alpha1/workerdeployment_webhook.go index d9b40bae..cfd79553 100644 --- a/api/v1alpha1/workerdeployment_webhook.go +++ b/api/v1alpha1/workerdeployment_webhook.go @@ -11,6 +11,7 @@ import ( "github.com/temporalio/temporal-worker-controller/internal/defaults" apierrors "k8s.io/apimachinery/pkg/api/errors" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + metav1validation "k8s.io/apimachinery/pkg/apis/meta/v1/validation" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/validation/field" ctrl "sigs.k8s.io/controller-runtime" @@ -82,6 +83,7 @@ func (r *WorkerDeployment) validateForUpdateOrCreate(ctx context.Context, obj ru func validateForUpdateOrCreate(dep *WorkerDeployment) (admission.Warnings, error) { allErrs := validateRolloutStrategy(dep.Spec.RolloutStrategy) + allErrs = append(allErrs, metav1validation.ValidateLabels(dep.Spec.DeploymentLabels, field.NewPath("spec.deploymentLabels"))...) if len(allErrs) > 0 { return nil, newInvalidErr(dep, allErrs) } diff --git a/api/v1alpha1/workerdeployment_webhook_test.go b/api/v1alpha1/workerdeployment_webhook_test.go index 536c402b..d0d048e9 100644 --- a/api/v1alpha1/workerdeployment_webhook_test.go +++ b/api/v1alpha1/workerdeployment_webhook_test.go @@ -128,6 +128,28 @@ func TestWorkerDeployment_ValidateCreate(t *testing.T) { "valid temporal worker deployment": { obj: testhelpers.MakeWDWithName("valid-worker", ""), }, + "deploymentLabels with valid keys and values": { + obj: testhelpers.ModifyObj(testhelpers.MakeWDWithName("valid-deployment-labels", ""), func(obj *temporaliov1alpha1.WorkerDeployment) *temporaliov1alpha1.WorkerDeployment { + obj.Spec.DeploymentLabels = map[string]string{"team": "payments", "example.com/instrumentation": "false", "empty": ""} + return obj + }), + }, + // Without this check the CRD accepts the label and the controller then fails + // to create the versioned Deployment, which stalls the rollout. + "deploymentLabels with an invalid value": { + obj: testhelpers.ModifyObj(testhelpers.MakeWDWithName("invalid-deployment-label-value", ""), func(obj *temporaliov1alpha1.WorkerDeployment) *temporaliov1alpha1.WorkerDeployment { + obj.Spec.DeploymentLabels = map[string]string{"team": "foo bar"} + return obj + }), + errorMsg: `spec.deploymentLabels: Invalid value: "foo bar"`, + }, + "deploymentLabels with an invalid key": { + obj: testhelpers.ModifyObj(testhelpers.MakeWDWithName("invalid-deployment-label-key", ""), func(obj *temporaliov1alpha1.WorkerDeployment) *temporaliov1alpha1.WorkerDeployment { + obj.Spec.DeploymentLabels = map[string]string{"-team": "payments"} + return obj + }), + errorMsg: `spec.deploymentLabels: Invalid value: "-team"`, + }, "invalid object type": { obj: &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 9085986c..14b1a7d0 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -881,6 +881,13 @@ func (in *WorkerDeploymentSpec) DeepCopyInto(out *WorkerDeploymentSpec) { *out = new(DrainingDeploymentSpec) (*in).DeepCopyInto(*out) } + if in.DeploymentLabels != nil { + in, out := &in.DeploymentLabels, &out.DeploymentLabels + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } in.RolloutStrategy.DeepCopyInto(&out.RolloutStrategy) in.SunsetStrategy.DeepCopyInto(&out.SunsetStrategy) in.WorkerOptions.DeepCopyInto(&out.WorkerOptions) diff --git a/docs/configuration.md b/docs/configuration.md index 09c48b52..96dd7a9e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -600,6 +600,31 @@ example after a rollback, scales back to `deployment.replicas`. > `terminationGracePeriodSeconds`. Long-running activities should heartbeat so > they resume on another pod. +### Labels on versioned Deployments + +Labels in `spec.deployment.template.metadata.labels` go on the pods. To label the +versioned Kubernetes Deployment objects themselves, set `spec.deploymentLabels`. +Use this for tools that act on the Deployment rather than its pods, such as +admission webhooks that opt workloads in or out by label, or cost allocation. + +```yaml +spec: + deploymentLabels: + team: payments + example.com/instrumentation: "false" +``` + +The controller keeps these labels in sync on the Deployments of every version it +manages, including Draining ones. A change is applied in place: it doesn't create +a new Build ID or restart pods. Removing a key from `spec.deploymentLabels` removes +it from the Deployments. Labels that other tools add to the Deployments are left +alone; the controller tracks the keys it applied in the +`temporal.io/managed-deployment-labels` annotation. + +`temporal.io/build-id` and `temporal.io/deployment-name` are managed by the +controller and cannot be set. Keys and values must be valid Kubernetes labels; +the validating webhook rejects invalid ones. + ### Environment-Specific Configurations **Production Configuration:** diff --git a/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml b/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml index c0f509cc..167602f9 100644 --- a/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml +++ b/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml @@ -3927,6 +3927,15 @@ spec: required: - template type: object + deploymentLabels: + additionalProperties: + type: string + type: object + x-kubernetes-validations: + - message: temporal.io/build-id and temporal.io/deployment-name are + managed by the controller + rule: '!(''temporal.io/build-id'' in self) && !(''temporal.io/deployment-name'' + in self)' drainingDeployment: properties: replicas: diff --git a/internal/k8s/deployments.go b/internal/k8s/deployments.go index 622c9738..37472e74 100644 --- a/internal/k8s/deployments.go +++ b/internal/k8s/deployments.go @@ -10,6 +10,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "maps" "regexp" "slices" "sort" @@ -44,6 +45,10 @@ const ( MaxDeploymentNameLen = 47 ConnectionSpecHashAnnotation = "temporal.io/connection-spec-hash" PodTemplateSpecHashAnnotation = "temporal.io/pod-template-spec-hash" + // ManagedDeploymentLabelsAnnotation lists the spec.deploymentLabels keys the controller + // applied, so a key removed from the spec can be removed without touching labels that + // other tools put on the Deployment. + ManagedDeploymentLabelsAnnotation = "temporal.io/managed-deployment-labels" // Environment variables read by the Temporal Go SDK's envconfig package // (go.temporal.io/sdk/contrib/envconfig) to configure the worker's connection. @@ -317,13 +322,13 @@ func NewDeploymentWithOwnerRef( ) depSpec.Template.Spec = *podSpec - return &appsv1.Deployment{ + deployment := &appsv1.Deployment{ ObjectMeta: metav1.ObjectMeta{ Name: ComputeVersionedDeploymentName(objectMeta.Name, buildID), Namespace: objectMeta.Namespace, DeletionGracePeriodSeconds: nil, - Labels: selectorLabels, - Annotations: depSpec.Template.Annotations, + Labels: maps.Clone(selectorLabels), + Annotations: maps.Clone(depSpec.Template.Annotations), OwnerReferences: []metav1.OwnerReference{{ APIVersion: typeMeta.APIVersion, Kind: typeMeta.Kind, @@ -337,6 +342,63 @@ func NewDeploymentWithOwnerRef( }, Spec: depSpec, } + ApplyDeploymentLabels(deployment, spec.DeploymentLabels) + return deployment +} + +// ApplyDeploymentLabels brings the Deployment's labels in line with spec.deploymentLabels +// and reports whether anything changed. It removes only keys it applied earlier, so labels +// added by other tools survive, and it never changes the selector labels. +func ApplyDeploymentLabels(deployment *appsv1.Deployment, desired map[string]string) bool { + isSelectorLabel := func(key string) bool { + if deployment.Spec.Selector == nil { + return false + } + _, ok := deployment.Spec.Selector.MatchLabels[key] + return ok + } + + changed := false + for _, key := range strings.Split(deployment.Annotations[ManagedDeploymentLabelsAnnotation], ",") { + if _, keep := desired[key]; keep || key == "" || isSelectorLabel(key) { + continue + } + if _, ok := deployment.Labels[key]; ok { + delete(deployment.Labels, key) + changed = true + } + } + + managed := make([]string, 0, len(desired)) + for key, value := range desired { + if isSelectorLabel(key) { + continue + } + managed = append(managed, key) + if current, ok := deployment.Labels[key]; ok && current == value { + continue + } + if deployment.Labels == nil { + deployment.Labels = map[string]string{} + } + deployment.Labels[key] = value + changed = true + } + + sort.Strings(managed) + managedValue := strings.Join(managed, ",") + if deployment.Annotations[ManagedDeploymentLabelsAnnotation] != managedValue { + if managedValue == "" { + delete(deployment.Annotations, ManagedDeploymentLabelsAnnotation) + } else { + if deployment.Annotations == nil { + deployment.Annotations = map[string]string{} + } + deployment.Annotations[ManagedDeploymentLabelsAnnotation] = managedValue + } + changed = true + } + return changed } // TODO (Shivam): Change hash when secret name is updated as well. diff --git a/internal/k8s/deployments_test.go b/internal/k8s/deployments_test.go index 85236276..5401d59a 100644 --- a/internal/k8s/deployments_test.go +++ b/internal/k8s/deployments_test.go @@ -13,6 +13,7 @@ import ( "crypto/x509/pkix" "encoding/pem" "fmt" + "maps" "math/big" "os" "path/filepath" @@ -586,7 +587,9 @@ func TestNewDeploymentWithOwnerRef_Labels(t *testing.T) { deployment := k8s.NewDeploymentWithOwnerRef( &metav1.TypeMeta{}, &metav1.ObjectMeta{Name: "test-worker", Namespace: "default"}, - &temporaliov1alpha1.WorkerDeploymentSpec{}, + &temporaliov1alpha1.WorkerDeploymentSpec{ + DeploymentLabels: map[string]string{"team": "payments"}, + }, "test-deployment", "build123", temporaliov1alpha1.ConnectionSpec{}, @@ -594,6 +597,95 @@ func TestNewDeploymentWithOwnerRef_Labels(t *testing.T) { assert.Equal(t, "test-worker", deployment.Labels[k8s.WorkerDeploymentNameLabel]) assert.Equal(t, "build123", deployment.Labels[k8s.BuildIDLabel]) + assert.Equal(t, "payments", deployment.Labels["team"]) + assert.NotContains(t, deployment.Spec.Template.Labels, "team", "deployment labels must not leak into pod labels") + assert.Equal(t, "team", deployment.Annotations[k8s.ManagedDeploymentLabelsAnnotation]) + assert.NotContains(t, deployment.Spec.Template.Annotations, k8s.ManagedDeploymentLabelsAnnotation, + "a pod template change would roll the pods") + assert.Equal(t, map[string]string{ + k8s.WorkerDeploymentNameLabel: "test-worker", + k8s.BuildIDLabel: "build123", + }, deployment.Spec.Selector.MatchLabels) +} + +func TestApplyDeploymentLabels(t *testing.T) { + selector := map[string]string{ + k8s.WorkerDeploymentNameLabel: "test-worker", + k8s.BuildIDLabel: "build123", + } + tests := map[string]struct { + labels map[string]string + managed string + desired map[string]string + wantLabels map[string]string + wantManaged string + wantChanged bool + }{ + "adds labels to a Deployment created before the field existed": { + labels: map[string]string{"argocd": "x"}, + desired: map[string]string{"team": "payments"}, + wantLabels: map[string]string{"argocd": "x", "team": "payments"}, + wantManaged: "team", + wantChanged: true, + }, + "updates a changed value": { + labels: map[string]string{"team": "payments"}, + managed: "team", + desired: map[string]string{"team": "billing"}, + wantLabels: map[string]string{"team": "billing"}, + wantManaged: "team", + wantChanged: true, + }, + "removes a key dropped from the spec and keeps labels other tools added": { + labels: map[string]string{"team": "payments", "opt-out": "true", "argocd": "x"}, + managed: "opt-out,team", + desired: map[string]string{"team": "payments"}, + wantLabels: map[string]string{"team": "payments", "argocd": "x"}, + wantManaged: "team", + wantChanged: true, + }, + "removes the annotation when the spec is emptied": { + labels: map[string]string{"team": "payments"}, + managed: "team", + wantLabels: map[string]string{}, + wantChanged: true, + }, + "never overrides or removes selector labels": { + managed: k8s.BuildIDLabel, + desired: map[string]string{k8s.BuildIDLabel: "user-supplied"}, + wantLabels: map[string]string{}, + wantChanged: true, + }, + "reports no change when in sync": { + labels: map[string]string{"b": "2", "a": "1"}, + managed: "a,b", + desired: map[string]string{"a": "1", "b": "2"}, + wantLabels: map[string]string{"a": "1", "b": "2"}, + wantManaged: "a,b", + }, + } + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + labels := maps.Clone(selector) + maps.Copy(labels, tc.labels) + deployment := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Labels: labels, Annotations: map[string]string{"other": "kept"}}, + Spec: appsv1.DeploymentSpec{Selector: &metav1.LabelSelector{MatchLabels: maps.Clone(selector)}}, + } + if tc.managed != "" { + deployment.Annotations[k8s.ManagedDeploymentLabelsAnnotation] = tc.managed + } + + changed := k8s.ApplyDeploymentLabels(deployment, tc.desired) + + wantLabels := maps.Clone(selector) + maps.Copy(wantLabels, tc.wantLabels) + assert.Equal(t, tc.wantChanged, changed) + assert.Equal(t, wantLabels, deployment.Labels) + assert.Equal(t, tc.wantManaged, deployment.Annotations[k8s.ManagedDeploymentLabelsAnnotation]) + assert.Equal(t, "kept", deployment.Annotations["other"]) + }) + } } func TestNewDeploymentWithOwnerRef_Strategy(t *testing.T) { diff --git a/internal/planner/planner.go b/internal/planner/planner.go index ef1528aa..030b4851 100644 --- a/internal/planner/planner.go +++ b/internal/planner/planner.go @@ -708,6 +708,23 @@ func checkAndUpdateDeploymentStrategy( return existingDeployment } +// checkAndUpdateDeploymentLabels updates an owned Deployment when its labels +// differ from spec.deploymentLabels. +func checkAndUpdateDeploymentLabels( + buildID string, + k8sState *k8s.DeploymentState, + spec *temporaliov1alpha1.WorkerDeploymentSpec, +) *appsv1.Deployment { + existingDeployment, exists := k8sState.Deployments[buildID] + if !exists { + return nil + } + if !k8s.ApplyDeploymentLabels(existingDeployment, spec.DeploymentLabels) { + return nil + } + return existingDeployment +} + func getUpdateDeployments( k8sState *k8s.DeploymentState, status *temporaliov1alpha1.WorkerDeploymentStatus, @@ -763,6 +780,22 @@ func getUpdateDeployments( } } + // Sync deployment labels the same way, after the strategy sync, so a + // Kubernetes Deployment already queued for update picks up label changes in + // the same write. + for _, buildID := range ownedBuildIDs(status) { + if updatedBuildIDs[buildID] { + if deployment, exists := k8sState.Deployments[buildID]; exists { + k8s.ApplyDeploymentLabels(deployment, spec.DeploymentLabels) + } + continue + } + if deployment := checkAndUpdateDeploymentLabels(buildID, k8sState, spec); deployment != nil { + updateDeployments = append(updateDeployments, deployment) + updatedBuildIDs[buildID] = true + } + } + return updateDeployments } diff --git a/internal/planner/planner_test.go b/internal/planner/planner_test.go index 5ec90a3b..9cc22f88 100644 --- a/internal/planner/planner_test.go +++ b/internal/planner/planner_test.go @@ -5187,3 +5187,58 @@ func TestGetDeleteDeployments_Inactive(t *testing.T) { }) } } + +func TestGetUpdateDeployments_DeploymentLabelsReconcile(t *testing.T) { + spec := &temporaliov1alpha1.WorkerDeploymentSpec{ + DeploymentLabels: map[string]string{"team": "payments"}, + } + newDeployment := func(name string) *appsv1.Deployment { + return &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: appsv1.DeploymentSpec{ + Strategy: spec.DeploymentSpec().Strategy, + Template: corev1.PodTemplateSpec{ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{}}}, + }, + } + } + target := newDeployment("worker-v2") + deprecated := newDeployment("worker-v1") + + status := &temporaliov1alpha1.WorkerDeploymentStatus{ + TargetVersion: temporaliov1alpha1.TargetWorkerDeploymentVersion{ + BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{BuildID: "v2"}, + }, + DeprecatedVersions: []*temporaliov1alpha1.DeprecatedWorkerDeploymentVersion{ + {BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{BuildID: "v1"}}, + }, + } + k8sState := &k8s.DeploymentState{ + Deployments: map[string]*appsv1.Deployment{"v1": deprecated, "v2": target}, + } + + // A connection change queues the target Deployment first; its labels must land in the same write. + connection := temporaliov1alpha1.ConnectionSpec{HostPort: "temporal:7233"} + updates := getUpdateDeployments(k8sState, status, spec, connection) + require.Len(t, updates, 2) + assert.Equal(t, map[string]string{"team": "payments"}, target.Labels) + assert.Equal(t, map[string]string{"team": "payments"}, deprecated.Labels) + + assert.Empty(t, getUpdateDeployments(k8sState, status, spec, connection), "no update once labels are in sync") + + // A strategy change and a label change on the same Deployment go out in one update. + maxSurge := intstr.FromInt32(0) + changed := &temporaliov1alpha1.WorkerDeploymentSpec{ + DeploymentLabels: map[string]string{"team": "billing"}, + Deployment: &appsv1.DeploymentSpec{Strategy: appsv1.DeploymentStrategy{ + Type: appsv1.RollingUpdateDeploymentStrategyType, + RollingUpdate: &appsv1.RollingUpdateDeployment{MaxSurge: &maxSurge}, + }}, + } + updates = getUpdateDeployments(k8sState, status, changed, connection) + require.Len(t, updates, 2) + for _, d := range updates { + assert.Equal(t, "billing", d.Labels["team"], d.Name) + require.NotNil(t, d.Spec.Strategy.RollingUpdate, d.Name) + assert.Equal(t, maxSurge, *d.Spec.Strategy.RollingUpdate.MaxSurge, d.Name) + } +}