Add rollout restart time annotation to Brokercell and update Brokercell deployments

pkg/reconciler/brokercell/brokercell.go

@@ -240,6 +240,7 @@ func (r *Reconciler) makeIngressArgs(bc *intv1alpha1.BrokerCell) resources.Ingre
 			CPULimit:           bc.Spec.Components.Ingress.CPULimit,
 			MemoryRequest:      bc.Spec.Components.Ingress.MemoryRequest,
 			MemoryLimit:        bc.Spec.Components.Ingress.MemoryLimit,
+			RolloutRestartTime: bc.GetAnnotations()[resources.IngressRestartTimeAnnotationKey],
 		},
 		Port: r.env.IngressPort,
 	}
@@ -269,6 +270,7 @@ func (r *Reconciler) makeFanoutArgs(bc *intv1alpha1.BrokerCell) resources.Fanout
 			CPULimit:           bc.Spec.Components.Fanout.CPULimit,
 			MemoryRequest:      bc.Spec.Components.Fanout.MemoryRequest,
 			MemoryLimit:        bc.Spec.Components.Fanout.MemoryLimit,
+			RolloutRestartTime: bc.GetAnnotations()[resources.FanoutRestartTimeAnnotationKey],
 		},
 	}
 }
@@ -297,6 +299,7 @@ func (r *Reconciler) makeRetryArgs(bc *intv1alpha1.BrokerCell) resources.RetryAr
 			CPULimit:           bc.Spec.Components.Retry.CPULimit,
 			MemoryRequest:      bc.Spec.Components.Retry.MemoryRequest,
 			MemoryLimit:        bc.Spec.Components.Retry.MemoryLimit,
+			RolloutRestartTime: bc.GetAnnotations()[resources.RetryRestartTimeAnnotationKey],
 		},
 	}
 }

pkg/reconciler/brokercell/brokercell_test.go

@@ -56,7 +56,12 @@ const (
 var (
 	testKey = fmt.Sprintf("%s/%s", testNS, brokerCellName)
 
-	creatorAnnotation = map[string]string{"internal.events.cloud.google.com/creator": "googlecloud"}
+	creatorAnnotation       = map[string]string{"internal.events.cloud.google.com/creator": "googlecloud"}
+	restartedTimeAnnotation = map[string]string{
+		"events.cloud.google.com/ingressRestartRequestedAt": "2020-09-25T16:28:36-04:00",
+		"events.cloud.google.com/fanoutRestartRequestedAt":  "2020-09-25T16:28:36-04:00",
+		"events.cloud.google.com/retryRestartRequestedAt":   "2020-09-25T16:28:36-04:00",
+	}
 
 	brokerCellReconciledEvent     = Eventf(corev1.EventTypeNormal, "BrokerCellReconciled", `BrokerCell reconciled: "testnamespace/test-brokercell"`)
 	brokerCellGCEvent             = Eventf(corev1.EventTypeNormal, "BrokerCellGarbageCollected", `BrokerCell garbage collected: "testnamespace/test-brokercell"`)
@@ -878,6 +883,42 @@ func TestAllCases(t *testing.T) {
 				},
 			},
 			WantEvents: []string{brokerCellGCEvent},
+		}, {
+			Name: "Brokercell has restart time annotation, deployments are updated with restart time annotation successfully",
+			Key:  testKey,
+			Objects: []runtime.Object{
+				NewBrokerCell(brokerCellName, testNS, WithBrokerCellSetDefaults,
+					WithBrokerCellAnnotations(restartedTimeAnnotation)),
+				testingdata.EmptyConfig(t, NewBrokerCell(brokerCellName, testNS, WithBrokerCellSetDefaults)),
+				NewEndpoints(brokerCellName+"-brokercell-ingress", testNS,
+					WithEndpointsAddresses(corev1.EndpointAddress{IP: "127.0.0.1"})),
+				testingdata.IngressDeploymentWithStatus(t),
+				testingdata.IngressServiceWithStatus(t),
+				testingdata.FanoutDeploymentWithStatus(t),
+				testingdata.RetryDeploymentWithStatus(t),
+				testingdata.IngressHPA(t),
+				testingdata.FanoutHPA(t),
+				testingdata.RetryHPA(t),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: NewBrokerCell(brokerCellName, testNS,
+					WithBrokerCellReady,
+					WithBrokerCellAnnotations(restartedTimeAnnotation),
+					WithIngressTemplate("http://test-brokercell-brokercell-ingress.testnamespace.svc.cluster.local/{namespace}/{name}"),
+					WithBrokerCellSetDefaults,
+				),
+			}},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{Object: testingdata.IngressDeploymentWithRestartAnnotation(t)},
+				{Object: testingdata.FanoutDeploymentWithRestartAnnotation(t)},
+				{Object: testingdata.RetryDeploymentWithRestartAnnotation(t)},
+			},
+			WantEvents: []string{
+				ingressDeploymentUpdatedEvent,
+				fanoutDeploymentUpdatedEvent,
+				retryDeploymentUpdatedEvent,
+				brokerCellReconciledEvent,
+			},
 		},
 	}
 

pkg/reconciler/brokercell/resources/args.go

@@ -32,6 +32,12 @@ const (
 	// RetryName is the name used for the retry container.
 	RetryName          = "retry"
 	BrokerCellLabelKey = "brokerCell"
+
+	// the annotation key of the time to trigger rollout restart of the deployments
+	IngressRestartTimeAnnotationKey = "events.cloud.google.com/ingressRestartRequestedAt"
+	FanoutRestartTimeAnnotationKey  = "events.cloud.google.com/fanoutRestartRequestedAt"
+	RetryRestartTimeAnnotationKey   = "events.cloud.google.com/retryRestartRequestedAt"
+	RolloutRestartTimeAnnotationKey = "events.cloud.google.com/RestartRequestedAt"
 )
 
 var (
@@ -50,6 +56,7 @@ type Args struct {
 	CPULimit           string
 	MemoryRequest      string
 	MemoryLimit        string
+	RolloutRestartTime string
 }
 
 // IngressArgs are the arguments to create a Broker's ingress Deployment.

pkg/reconciler/brokercell/resources/deployments.go

@@ -127,6 +127,12 @@ func MakeRetryDeployment(args RetryArgs) *appsv1.Deployment {
 
 // deploymentTemplate creates a template for data plane deployments.
 func deploymentTemplate(args Args, containers []corev1.Container) *appsv1.Deployment {
+	annotation := map[string]string{
+		"sidecar.istio.io/inject": strconv.FormatBool(args.AllowIstioSidecar),
+	}
+	if args.RolloutRestartTime != "" {
+		annotation[RolloutRestartTimeAnnotationKey] = args.RolloutRestartTime
+	}
 	return &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:       args.BrokerCell.Namespace,
@@ -145,10 +151,8 @@ func deploymentTemplate(args Args, containers []corev1.Container) *appsv1.Deploy
 			MinReadySeconds: 60,
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
-					Labels: Labels(args.BrokerCell.Name, args.ComponentName),
-					Annotations: map[string]string{
-						"sidecar.istio.io/inject": strconv.FormatBool(args.AllowIstioSidecar),
-					},
+					Labels:      Labels(args.BrokerCell.Name, args.ComponentName),
+					Annotations: annotation,
 				},
 				Spec: corev1.PodSpec{
 					ServiceAccountName: args.ServiceAccountName,

pkg/reconciler/brokercell/testingdata/fanout_deployment_with_restart_annotation.yaml

@@ -0,0 +1,107 @@
+# Copyright 2020 Google LLC
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This yaml matches the fanout deployment objected created by the reconciler with
+# additional status so that reconciler will mark readiness based on the status.
+metadata:
+  name: test-brokercell-brokercell-fanout
+  namespace: testnamespace
+  labels:
+    app: cloud-run-events
+    brokerCell: test-brokercell
+    role: fanout
+  ownerReferences:
+    - apiVersion: internal.events.cloud.google.com/v1alpha1
+      kind: BrokerCell
+      name: test-brokercell
+      controller: true
+      blockOwnerDeletion: true
+spec:
+  selector:
+    matchLabels: &labels
+      app: cloud-run-events
+      brokerCell: test-brokercell
+      role: fanout
+  minReadySeconds: 60
+  strategy:
+    rollingUpdate:
+      maxUnavailable: 0
+      maxSurge: 1
+  template:
+    metadata:
+      labels: *labels
+      annotations:
+        sidecar.istio.io/inject: "true"
+        events.cloud.google.com/RestartRequestedAt: "2020-09-25T16:28:36-04:00"
+    spec:
+      serviceAccountName: broker
+      terminationGracePeriodSeconds: 60
+      containers:
+        - name: fanout
+          image: fanout
+          livenessProbe:
+            failureThreshold: 3
+            httpGet:
+              path: /healthz
+              port: 8080
+              scheme: HTTP
+            initialDelaySeconds: 15
+            periodSeconds: 15
+            successThreshold: 1
+            timeoutSeconds: 5
+          env:
+            - name: GOOGLE_APPLICATION_CREDENTIALS
+              value: /var/secrets/google/key.json
+            - name: SYSTEM_NAMESPACE
+              value: knative-testing
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: CONFIG_LOGGING_NAME
+              value: config-logging
+            - name: CONFIG_OBSERVABILITY_NAME
+              value: config-observability
+            - name: METRICS_DOMAIN
+              value: knative.dev/internal/eventing
+            - name: MAX_CONCURRENCY_PER_EVENT
+              value: "100"
+          volumeMounts:
+            - name: broker-config
+              mountPath: /var/run/cloud-run-events/broker
+            - name: google-broker-key
+              mountPath: /var/secrets/google
+          resources:
+            limits:
+              memory: 3000Mi
+            requests:
+              cpu: 1500m
+              memory: 3000Mi
+          ports:
+            - name: metrics
+              containerPort: 9090
+            - name: http-health
+              containerPort: 8080
+      volumes:
+        - name: broker-config
+          configMap:
+            name: test-brokercell-brokercell-broker-targets
+        - name: google-broker-key
+          secret:
+            secretName: google-broker-key
+            optional: true
+status:
+  conditions:
+    - status: "True"
+      type: Available

pkg/reconciler/brokercell/testingdata/ingress_deployment_with_restart_annotation.yaml

@@ -0,0 +1,116 @@
+# Copyright 2020 Google LLC
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This yaml matches the ingress deployment objected created by the reconciler with
+# additional status so that reconciler will mark readiness based on the status.
+metadata:
+  name: test-brokercell-brokercell-ingress
+  namespace: testnamespace
+  labels:
+    app: cloud-run-events
+    brokerCell: test-brokercell
+    role: ingress
+  ownerReferences:
+    - apiVersion: internal.events.cloud.google.com/v1alpha1
+      kind: BrokerCell
+      name: test-brokercell
+      controller: true
+      blockOwnerDeletion: true
+spec:
+  selector:
+    matchLabels: &labels
+      app: cloud-run-events
+      brokerCell: test-brokercell
+      role: ingress
+  minReadySeconds: 60
+  strategy:
+    rollingUpdate:
+      maxUnavailable: 0
+      maxSurge: 1
+  template:
+    metadata:
+      labels: *labels
+      annotations:
+        sidecar.istio.io/inject: "true"
+        events.cloud.google.com/RestartRequestedAt: "2020-09-25T16:28:36-04:00"
+    spec:
+      serviceAccountName: broker
+      terminationGracePeriodSeconds: 60
+      containers:
+        - name: ingress
+          image: ingress
+          livenessProbe:
+            failureThreshold: 5
+            httpGet:
+              path: /healthz
+              port: 8080
+              scheme: HTTP
+            initialDelaySeconds: 15
+            periodSeconds: 15
+            successThreshold: 1
+            timeoutSeconds: 5
+          readinessProbe:
+            failureThreshold: 5
+            httpGet:
+              path: /healthz
+              port: 8080
+              scheme: HTTP
+            periodSeconds: 15
+            successThreshold: 1
+            timeoutSeconds: 5
+          env:
+            - name: GOOGLE_APPLICATION_CREDENTIALS
+              value: /var/secrets/google/key.json
+            - name: SYSTEM_NAMESPACE
+              value: knative-testing
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: CONFIG_LOGGING_NAME
+              value: config-logging
+            - name: CONFIG_OBSERVABILITY_NAME
+              value: config-observability
+            - name: METRICS_DOMAIN
+              value: knative.dev/internal/eventing
+            - name: PORT
+              value: "8080"
+          volumeMounts:
+            - name: broker-config
+              mountPath: /var/run/cloud-run-events/broker
+            - name: google-broker-key
+              mountPath: /var/secrets/google
+          resources:
+            limits:
+              memory: 2000Mi
+            requests:
+              cpu: 2000m
+              memory: 2000Mi
+          ports:
+            - name: metrics
+              containerPort: 9090
+            - name: http
+              containerPort: 8080
+      volumes:
+        - name: broker-config
+          configMap:
+            name: test-brokercell-brokercell-broker-targets
+        - name: google-broker-key
+          secret:
+            secretName: google-broker-key
+            optional: true
+status:
+  conditions:
+    - status: "True"
+      type: Available

pkg/reconciler/brokercell/testingdata/read_yaml.go

@@ -57,6 +57,18 @@ func RetryDeploymentWithStatus(t *testing.T) *appsv1.Deployment {
 	return getDeployment(t, "testingdata/retry_deployment_with_status.yaml")
 }
 
+func IngressDeploymentWithRestartAnnotation(t *testing.T) *appsv1.Deployment {
+	return getDeployment(t, "testingdata/ingress_deployment_with_restart_annotation.yaml")
+}
+
+func FanoutDeploymentWithRestartAnnotation(t *testing.T) *appsv1.Deployment {
+	return getDeployment(t, "testingdata/fanout_deployment_with_restart_annotation.yaml")
+}
+
+func RetryDeploymentWithRestartAnnotation(t *testing.T) *appsv1.Deployment {
+	return getDeployment(t, "testingdata/retry_deployment_with_restart_annotation.yaml")
+}
+
 func IngressServiceWithStatus(t *testing.T) *corev1.Service {
 	return getService(t, "testingdata/ingress_service_with_status.yaml")
 }