Repository: GoogleCloudPlatform/cloud-sql-proxy-operator
Pull request #197 (2023-02-07 15:37:24)

feat: Automatically trigger pod rollout for appsv1 resources when AuthProxyWorkload changes.
Given an AuthProxyWorkload resource that has been applied to StatefulSet, DaemonSet, 
ReplicaSet, and/or Deployment workloads, when that AuthProxyWorkload is updated, the operator 
will add or modify an annotation on the PodTemplateSpec of that workload, causing k8s to
replace the workload's pods following the workload's rollout strategy, thus applying the
AuthProxyWorkload's configuration change.  

Related to #187. 

The code for this feature will be split across a few PRs for clarity. These are the first 4:

1. (This PR) Automatically rollout pod changes on AuthProxyWorkload update.
2. Automatically rollout pod changes on AuthProxyWorkload delete.
3. Add `AutomaticRolloutEnabled` field to the AuthProxyWorkload resource.
4. Update AuthProxyWorkload status conditions to show if a workload has out-of-date pods

hessjcg: /gcbrun
hessjcg: /gcbrun


Patch:
diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml
index 7f68c2a..bc0f82c 100644
--- a/config/rbac/role.yaml
+++ b/config/rbac/role.yaml
@@ -45,6 +45,16 @@ rules:
       - get
       - list
       - watch
+  - apiGroups:
+      - apps
+    resources:
+      - daemonsets
+      - deployments
+      - replicasets
+      - statefulsets
+    verbs:
+      - patch
+      - update
   - apiGroups:
       - batch
     resources:
diff --git a/installer/cloud-sql-proxy-operator.yaml b/installer/cloud-sql-proxy-operator.yaml
index 811f436..c58e32e 100644
--- a/installer/cloud-sql-proxy-operator.yaml
+++ b/installer/cloud-sql-proxy-operator.yaml
@@ -1161,6 +1161,16 @@ rules:
       - get
       - list
       - watch
+  - apiGroups:
+      - apps
+    resources:
+      - daemonsets
+      - deployments
+      - replicasets
+      - statefulsets
+    verbs:
+      - patch
+      - update
   - apiGroups:
       - batch
     resources:
diff --git a/internal/api/v1alpha1/authproxyworkload_types.go b/internal/api/v1alpha1/authproxyworkload_types.go
index 86ddfee..04c7348 100644
--- a/internal/api/v1alpha1/authproxyworkload_types.go
+++ b/internal/api/v1alpha1/authproxyworkload_types.go
@@ -45,6 +45,12 @@ const (
 	// when the resource reconcile has finished running.
 	ReasonFinishedReconcile = "FinishedReconcile"
 
+	// ReasonWorkloadNeedsUpdate relates to condition UpToDate, this reason is set
+	// when the resource reconcile found existing workloads related to this
+	// AuthProxyWorkload resource that are not yet configured with an up-to-date
+	// proxy configuration.
+	ReasonWorkloadNeedsUpdate = "WorkloadNeedsUpdate"
+
 	// ReasonNoWorkloadsFound relates to condition UpToDate, this reason is set
 	// when there are no workloads related to this AuthProxyWorkload resource.
 	ReasonNoWorkloadsFound = "NoWorkloadsFound"
diff --git a/internal/controller/authproxyworkload_controller.go b/internal/controller/authproxyworkload_controller.go
index d5168d4..e3a9ad5 100644
--- a/internal/controller/authproxyworkload_controller.go
+++ b/internal/controller/authproxyworkload_controller.go
@@ -94,6 +94,7 @@ func (r *AuthProxyWorkloadReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Complete(r)
 }
 
+//+kubebuilder:rbac:groups=apps,resources=deployments;statefulsets;daemonsets;replicasets,verbs=update;patch
 //+kubebuilder:rbac:groups=apps,resources=*,verbs=get;list;watch
 //+kubebuilder:rbac:groups=batch,resources=*,verbs=get;list;watch
 //+kubebuilder:rbac:groups="",resources=*,verbs=get;list;watch
@@ -198,19 +199,24 @@ func (r *AuthProxyWorkloadReconciler) doDelete(ctx context.Context, resource *cl
 // - the condition `UpToDate` status and reason
 //
 // States:
-// |  state  | finalizer| fetch err | len(wl)      | Name                      |
-// |---------|----------|-----------|--------------|---------------------      |
-// | 0       | *        | *         | *            | start                     |
-// | 1.1     | absent   | *         | *            | needs finalizer           |
-// | 1.2     | present  | error     | *            | can't list workloads      |
-// | 2.1     | present  | nil       | == 0         | no workloads to reconcile |
-// | 3.1     | present  | nil       | > 0          | workloads reconciled      |
+// |  state  | finalizer| fetch err | len(wl) | outOfDateCount | Name                                  |
+// |---------|----------|-----------|---------|----------------|---------------------------------------|
+// | 0       | *        | *         | *       |                | start                                 |
+// | 1.1     | absent   | *         | *       |                | needs finalizer                       |
+// | 1.2     | present  | error     | *       |                | can't list workloads                  |
+// | 2.1     | present  | nil       | == 0    |                | no workloads to reconcile             |
+// | 3.1     | present  | nil       | > 0     | > 0 , err      | workload update needed, and failed    |
+// | 3.2     | present  | nil       | > 0     | > 0            | workload update needed, and succeeded |
+// | 3.3     | present  | nil       | > 0     | == 0           | workloads reconciled                  |
 //
-//	start ---x
-//	          \---> 1.1 --> (requeue, goto start)
-//	           \---> 1.2 --> (requeue, goto start)
-//	            \---> 2.1 --> (end)
-//	             \---> 3.1 --> (end)
+//		start ----x
+//		          |---> 1.1 --> (requeue, goto start)
+//		          |---> 1.2 --> (requeue, goto start)
+//		          |---> 2.1 --> (end)
+//		          |
+//	            |---> 3.1 ---> (requeue, goto start)
+//	            |---> 3.2 ---> (requeue, goto start)
+//	            |---> 3.3 ---> (end)
 func (r *AuthProxyWorkloadReconciler) doCreateUpdate(ctx context.Context, l logr.Logger, resource *cloudsqlapi.AuthProxyWorkload) (ctrl.Result, error) {
 	orig := resource.DeepCopy()
 	var err error
@@ -236,36 +242,111 @@ func (r *AuthProxyWorkloadReconciler) doCreateUpdate(ctx context.Context, l logr
 	// State 2.1: When there are no workloads, then mark this as "UpToDate" true,
 	// do not requeue.
 	if len(allWorkloads) == 0 {
-		return r.reconcileResult(ctx, l, resource, orig, cloudsqlapi.ReasonNoWorkloadsFound, "No workload updates needed")
+		return r.reconcileResult(ctx, l, resource, orig, cloudsqlapi.ReasonNoWorkloadsFound, "No workload updates needed", true)
 	}
 
-	// State 3.1: Workload updates are in progress. Check if the workload updates
-	// are complete.
-	//
+	// State 3.*: Workloads already exist. Some may need to be updated to roll out
+	// changes.
+	var outOfDateCount int
+	for _, wl := range allWorkloads {
+		wlChanged := r.needsAnnotationUpdate(wl, resource)
+		if !wlChanged {
+			continue
+		}
+
+		outOfDateCount++
+		_, err = controllerutil.CreateOrPatch(ctx, r.Client, wl.Object(), func() error {
+			r.updateAnnotation(wl, resource)
+			return nil
+		})
+
+		// State 3.1 Failed to update one of the workloads PodTemplateSpec annotations, requeue.
+		if err != nil {
+			message := fmt.Sprintf("Reconciled %d matching workloads. Error updating workload %v: %v", len(allWorkloads), wl.Object().GetName(), err)
+			return r.reconcileResult(ctx, l, resource, orig, cloudsqlapi.ReasonWorkloadNeedsUpdate, message, false)
+		}
+
+	}
+
+	// State 3.2 Successfully updated all workload PodTemplateSpec annotations, requeue
+	if outOfDateCount > 0 {
+		message := fmt.Sprintf("Reconciled %d matching workloads. %d workloads need updates", len(allWorkloads), outOfDateCount)
+		return r.reconcileResult(ctx, l, resource, orig, cloudsqlapi.ReasonWorkloadNeedsUpdate, message, false)
+	}
+
+	// State 3.3 Workload PodTemplateSpec annotations are all up to date
 	message := fmt.Sprintf("Reconciled %d matching workloads complete", len(allWorkloads))
+	return r.reconcileResult(ctx, l, resource, orig, cloudsqlapi.ReasonFinishedReconcile, message, true)
+}
+
+// needsAnnotationUpdate returns true when the workload was annotated with
+// a different generation of the resource.
+func (r *AuthProxyWorkloadReconciler) needsAnnotationUpdate(wl workload.Workload, resource *cloudsqlapi.AuthProxyWorkload) bool {
+
+	// This workload is not mutable. Ignore it.
+	if _, ok := wl.(workload.WithMutablePodTemplate); !ok {
+		return false
+	}
 
-	return r.reconcileResult(ctx, l, resource, orig, cloudsqlapi.ReasonFinishedReconcile, message)
+	k, v := workload.PodAnnotation(resource)
+
+	// Check if the correct annotation exists
+	an := wl.PodTemplateAnnotations()
+	if an != nil && an[k] == v {
+		return false
+	}
+
+	return true
+}
+
+// updateAnnotation applies an annotation to the workload for the resource.
+func (r *AuthProxyWorkloadReconciler) updateAnnotation(wl workload.Workload, resource *cloudsqlapi.AuthProxyWorkload) {
+	mpt, ok := wl.(workload.WithMutablePodTemplate)
+
+	// This workload is not mutable. Ignore it.
+	if !ok {
+		return
+	}
+
+	k, v := workload.PodAnnotation(resource)
+
+	// add the annotation if needed...
+	an := wl.PodTemplateAnnotations()
+	if an == nil {
+		an = make(map[string]string)
+	}
+
+	an[k] = v
+	mpt.SetPodTemplateAnnotations(an)
 }
 
 // workloadsReconciled  State 3.1: If workloads are all up to date, mark the condition
 // "UpToDate" true and do not requeue.
-func (r *AuthProxyWorkloadReconciler) reconcileResult(ctx context.Context, l logr.Logger, resource *cloudsqlapi.AuthProxyWorkload, orig *cloudsqlapi.AuthProxyWorkload, reason, message string) (ctrl.Result, error) {
+func (r *AuthProxyWorkloadReconciler) reconcileResult(ctx context.Context, l logr.Logger, resource, orig *cloudsqlapi.AuthProxyWorkload, reason, message string, upToDate bool) (ctrl.Result, error) {
+
+	status := metav1.ConditionFalse
+	result := requeueNow
+	if upToDate {
+		status = metav1.ConditionTrue
+		result = ctrl.Result{}
+	}
 
 	// Workload updates are complete, update the status
 	resource.Status.Conditions = replaceCondition(resource.Status.Conditions, &metav1.Condition{
 		Type:               cloudsqlapi.ConditionUpToDate,
-		Status:             metav1.ConditionTrue,
+		Status:             status,
 		ObservedGeneration: resource.GetGeneration(),
 		Reason:             reason,
 		Message:            message,
 	})
+
 	err := r.patchAuthProxyWorkloadStatus(ctx, resource, orig)
 	if err != nil {
 		l.Error(err, "Unable to patch status before beginning workloads", "AuthProxyWorkload", resource.GetNamespace()+"/"+resource.GetName())
-		return ctrl.Result{}, err
+		return result, err
 	}
 
-	return ctrl.Result{}, nil
+	return result, nil
 }
 
 // applyFinalizer adds the finalizer so that the operator is notified when
diff --git a/internal/workload/podspec_updates.go b/internal/workload/podspec_updates.go
index 9637dd7..2a175c6 100644
--- a/internal/workload/podspec_updates.go
+++ b/internal/workload/podspec_updates.go
@@ -47,6 +47,14 @@ const (
 
 var l = logf.Log.WithName("internal.workload")
 
+// PodAnnotation returns the key and value for an annotation on a pod
+// to indicate the specific AuthProxyWorkload resource that was configured
+// on the pod.
+func PodAnnotation(r *cloudsqlapi.AuthProxyWorkload) (string, string) {
+	return fmt.Sprintf("%s/%s", cloudsqlapi.AnnotationPrefix, r.Name),
+		fmt.Sprintf("%d", r.Generation)
+}
+
 // Updater holds global state used while reconciling workloads.
 type Updater struct {
 	// userAgent is the userAgent of the operator
@@ -388,6 +396,12 @@ func (s *updateState) update(wl *PodWorkload, matches []*cloudsqlapi.AuthProxyWo
 		}
 	}
 
+	// Copy the existing pod annotation map
+	ann := map[string]string{}
+	for k, v := range wl.PodTemplateAnnotations() {
+		ann[k] = v
+	}
+
 	// add all new containers and update existing containers
 	for i := range matches {
 		inst := matches[i]
@@ -395,10 +409,18 @@ func (s *updateState) update(wl *PodWorkload, matches []*cloudsqlapi.AuthProxyWo
 		newContainer := corev1.Container{}
 		s.updateContainer(inst, wl, &newContainer)
 		containers = append(containers, newContainer)
+
+		// Add pod annotation for each instance
+		k, v := PodAnnotation(inst)
+		ann[k] = v
 	}
 
 	podSpec.Containers = containers
 
+	if len(ann) != 0 {
+		wl.SetPodTemplateAnnotations(ann)
+	}
+
 	for i := range podSpec.Containers {
 		c := &podSpec.Containers[i]
 		s.updateContainerEnv(c)


Test patch:
diff --git a/internal/controller/authproxyworkload_controller_test.go b/internal/controller/authproxyworkload_controller_test.go
index 0f5098a..2e013fc 100644
--- a/internal/controller/authproxyworkload_controller_test.go
+++ b/internal/controller/authproxyworkload_controller_test.go
@@ -146,8 +146,51 @@ func TestReconcileState21BySelector(t *testing.T) {
 
 }
 
-func TestReconcileState31(t *testing.T) {
-	var wantRequeue bool
+func TestReconcileState32(t *testing.T) {
+	wantRequeue := true
+	wantStatus := metav1.ConditionFalse
+	wantReason := v1alpha1.ReasonWorkloadNeedsUpdate
+
+	p := testhelpers.BuildAuthProxyWorkload(types.NamespacedName{
+		Namespace: "default",
+		Name:      "test",
+	}, "project:region:db")
+	p.Generation = 2
+	p.Finalizers = []string{finalizerName}
+	p.Spec.Workload = v1alpha1.WorkloadSelectorSpec{
+		Kind: "Deployment",
+		Selector: &metav1.LabelSelector{
+			MatchLabels: map[string]string{"app": "things"},
+		},
+	}
+	p.Status.Conditions = []*metav1.Condition{{
+		Type:   v1alpha1.ConditionUpToDate,
+		Reason: v1alpha1.ReasonStartedReconcile,
+		Status: metav1.ConditionFalse,
+	}}
+
+	// mimic a pod that was updated by the webhook
+	reqName := v1alpha1.AnnotationPrefix + "/" + p.Name
+	pod := &appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "thing",
+			Namespace: "default",
+			Labels:    map[string]string{"app": "things"},
+		},
+		Spec: appsv1.DeploymentSpec{Template: corev1.PodTemplateSpec{
+			ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{reqName: "1"}},
+		}},
+	}
+
+	err := runReconcileTestcase(p, []client.Object{p, pod}, wantRequeue, wantStatus, wantReason)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+}
+
+func TestReconcileState33(t *testing.T) {
+	wantRequeue := false
 	wantStatus := metav1.ConditionTrue
 	wantReason := v1alpha1.ReasonFinishedReconcile
 
@@ -170,8 +213,7 @@ func TestReconcileState31(t *testing.T) {
 	}}
 
 	// mimic a pod that was updated by the webhook
-	reqName := v1alpha1.AnnotationPrefix + "/" +
-		workload.SafePrefixedName("req-", p.Namespace+"-"+p.Name)
+	reqName := v1alpha1.AnnotationPrefix + "/" + p.Name
 	pod := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "thing",
diff --git a/internal/workload/podspec_updates_test.go b/internal/workload/podspec_updates_test.go
index 813fc06..763fb81 100644
--- a/internal/workload/podspec_updates_test.go
+++ b/internal/workload/podspec_updates_test.go
@@ -775,3 +775,38 @@ func assertContainerArgsContains(t *testing.T, gotArgs, wantArgs []string) {
 		}
 	}
 }
+
+func TestPodTemplateAnnotations(t *testing.T) {
+	var (
+		wantAnnotations = map[string]string{
+			"cloudsql.cloud.google.com/instance1": "1",
+			"cloudsql.cloud.google.com/instance2": "2",
+		}
+
+		u = workload.NewUpdater("cloud-sql-proxy-operator/dev")
+	)
+
+	// Create a pod
+	wl := podWorkload()
+	wl.Pod.Spec.Containers[0].Ports =
+		[]corev1.ContainerPort{{Name: "http", ContainerPort: 8080}}
+
+	// Create a AuthProxyWorkload that matches the deployment
+	csqls := []*v1alpha1.AuthProxyWorkload{
+		simpleAuthProxy("instance1", "project:server:db"),
+		simpleAuthProxy("instance2", "project:server2:db2")}
+	csqls[0].ObjectMeta.Generation = 1
+	csqls[1].ObjectMeta.Generation = 2
+
+	// update the containers
+	err := configureProxies(u, wl, csqls)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// test that annotation was set properly
+	if !reflect.DeepEqual(wl.PodTemplateAnnotations(), wantAnnotations) {
+		t.Errorf("got %v, want %v for proxy container command", wl.PodTemplateAnnotations(), wantAnnotations)
+	}
+
+}
diff --git a/tests/e2e_test.go b/tests/e2e_test.go
index 45b8793..aa7eacd 100644
--- a/tests/e2e_test.go
+++ b/tests/e2e_test.go
@@ -26,7 +26,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
 func TestMain(m *testing.M) {
@@ -255,21 +254,6 @@ func TestProxyAppliedOnExistingWorkload(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			// if this is an apps/v1 resource with a mutable pod template,
-			// force a rolling update.
-			if wl, ok := test.o.(workload.WithMutablePodTemplate); ok {
-				// patch the workload, add an annotation to the podspec
-				t.Log("Customer updates the workload triggering a rollout")
-				controllerutil.CreateOrPatch(ctx, tp.Client, test.o.Object(), func() error {
-					wl.SetPodTemplateAnnotations(map[string]string{"customer": "updated"})
-					return nil
-				})
-
-				if err != nil {
-					t.Fatal(err)
-				}
-			}
-
 			t.Logf("Wait for %v pods to have 2 containers", test.allOrAny)
 			err = tp.ExpectPodContainerCount(ctx, selector, 2, test.allOrAny)
 			if err != nil {