VSS/instant-updates: more stable event watcher

controllers/vaultstaticsecret_controller.go

@@ -292,7 +292,9 @@ func (r *VaultStaticSecretReconciler) ensureEventWatcher(ctx context.Context, o
 	// launch the goroutine to watch events
 	logger.V(consts.LogLevelDebug).Info("Starting event watcher", "meta", updatedMeta)
 	r.eventWatcherRegistry.Register(name, updatedMeta)
-	go r.getEvents(watchCtx, o, wsClient, stoppedCh)
+	// Pass a dereferenced VSS object here to avoid the effect of the VSS object
+	// being garbage collected before the goroutine is started/evaluated
+	go r.getEvents(watchCtx, *o, wsClient, stoppedCh)
 
 	return nil
 }
@@ -312,9 +314,9 @@ func (r *VaultStaticSecretReconciler) unWatchEvents(o *secretsv1beta1.VaultStati
 
 // getEvents calls streamStaticSecretEvents in a loop, collecting and responding
 // to any errors returned.
-func (r *VaultStaticSecretReconciler) getEvents(ctx context.Context, o *secretsv1beta1.VaultStaticSecret, wsClient *vault.WebsocketClient, stoppedCh chan struct{}) {
+func (r *VaultStaticSecretReconciler) getEvents(ctx context.Context, o secretsv1beta1.VaultStaticSecret, wsClient *vault.WebsocketClient, stoppedCh chan struct{}) {
 	logger := log.FromContext(ctx).WithName("getEvents")
-	name := client.ObjectKeyFromObject(o)
+	name := client.ObjectKeyFromObject(&o)
 	defer func() {
 		r.eventWatcherRegistry.Delete(name)
 		close(stoppedCh)
@@ -343,7 +345,7 @@ eventLoop:
 				}
 				time.Sleep(retryBackoff.NextBackOff())
 			}
-			err := r.streamStaticSecretEvents(ctx, o, wsClient)
+			err := r.streamStaticSecretEvents(ctx, &o, wsClient)
 			if err != nil {
 				if strings.Contains(err.Error(), "use of closed network connection") ||
 					strings.Contains(err.Error(), "context canceled") {
@@ -362,15 +364,15 @@ eventLoop:
 				// For any other errors, we emit the error as an event on the
 				// VaultStaticSecret, reload the client and try connecting
 				// again.
-				r.Recorder.Eventf(o, corev1.EventTypeWarning, consts.ReasonEventWatcherError,
+				r.Recorder.Eventf(&o, corev1.EventTypeWarning, consts.ReasonEventWatcherError,
 					"Error while watching events: %s", err)
 
 				if errorCount >= errorThreshold {
 					logger.Error(err, "Too many errors while watching events, requeuing")
 					break eventLoop
 				}
 
-				newVaultClient, err := r.ClientFactory.Get(ctx, r.Client, o)
+				newVaultClient, err := r.ClientFactory.Get(ctx, r.Client, &o)
 				if err != nil {
 					logger.Error(err, "Failed to retrieve Vault client")
 					break eventLoop
@@ -383,7 +385,7 @@ eventLoop:
 				}
 
 				// Update the LastClientID in the event registry
-				key := client.ObjectKeyFromObject(o)
+				key := client.ObjectKeyFromObject(&o)
 				meta, ok := r.eventWatcherRegistry.Get(key)
 				if !ok {
 					logger.Error(

test/integration/vaultstaticsecret_integration_test.go

@@ -434,6 +434,28 @@ func TestVaultStaticSecret(t *testing.T) {
 				awaitRolloutRestarts(t, ctx, crdClient, obj, obj.Spec.RolloutRestartTargets)
 			}
 		}
+
+		if expectInitial && obj.Spec.SyncConfig != nil && obj.Spec.SyncConfig.InstantUpdates {
+			// Ensure the event watcher has started so that subsequent updates
+			// are detected and synced.
+			assert.NoError(t, backoff.Retry(func() error {
+				objEvents := corev1.EventList{}
+				err := crdClient.List(ctx, &objEvents,
+					ctrlclient.InNamespace(obj.Namespace),
+					ctrlclient.MatchingFields{
+						"involvedObject.name": obj.Name,
+						"reason":              consts.ReasonEventWatcherStarted,
+					},
+				)
+				if err != nil {
+					return err
+				}
+				if len(objEvents.Items) == 0 {
+					return fmt.Errorf("no EventWatcherStarted event for %s", obj.Name)
+				}
+				return nil
+			}, backoff.WithMaxRetries(backoff.NewConstantBackOff(time.Millisecond*500), 200)))
+		}
 	}
 
 	for _, tt := range tests {