Update To Global Eventing System

make generate manifests

revert test file since tests are being added in another PR

update controller

changes

update event path

code clean up

make genereate manifests

Update Secret Struct Name

update secret type to VaultStaticSecretConfig

changes for global event listener

updates

Remove unused stubVaultClient event methods

Add joinVaultPath helper to build Vault paths

Trim leading/trailing slashes and skip empty segments when joining

update tests

make fmt

Author: jaireddjawed

api/v1beta1/vaultdynamicsecret_types.go

@@ -62,7 +62,7 @@ type VaultDynamicSecretSpec struct {
 	// Destination provides configuration necessary for syncing the Vault secret to Kubernetes.
 	Destination Destination `json:"destination"`
 	// SyncConfig configures sync behavior from Vault to VSO
-	SyncConfig *DyanmicSecretSyncConfig `json:"syncConfig,omitempty"`
+	SyncConfig *VaultDynamicSecretSyncConfig `json:"syncConfig,omitempty"`
 	// RefreshAfter a period of time for VSO to sync the source secret data, in
 	// duration notation e.g. 30s, 1m, 24h. This value only needs to be set when
 	// syncing from a secret's engine that does not provide a lease TTL in its
@@ -131,10 +131,10 @@ type VaultStaticCredsMetaData struct {
 	TTL int64 `json:"ttl"`
 }
 
-// DyanmicSecretSyncConfig configures sync behavior from Vault to VSO for dynamic secrets
-type DyanmicSecretSyncConfig struct {
+// VaultDynamicSecretSyncConfig configures sync behavior from Vault to VSO for dynamic secrets
+type VaultDynamicSecretSyncConfig struct {
 	// InstantUpdates is a flag to indicate that event-driven updates are
-	// enabled for a VaultDynamicSecret
+	// enabled for this VaultDynamicSecret
 	InstantUpdates bool `json:"instantUpdates,omitempty"`
 }
 

api/v1beta1/vaultstaticsecret_types.go

@@ -41,7 +41,7 @@ type VaultStaticSecretSpec struct {
 	// Destination provides configuration necessary for syncing the Vault secret to Kubernetes.
 	Destination Destination `json:"destination"`
 	// SyncConfig configures sync behavior from Vault to VSO
-	SyncConfig *StaticSecretSyncConfig `json:"syncConfig,omitempty"`
+	SyncConfig *VaultStaticSecretSyncConfig `json:"syncConfig,omitempty"`
 
 	VaultStaticSecretCommon `json:",inline"`
 }
@@ -69,10 +69,10 @@ type VaultStaticSecretCollectable struct {
 	Transformation *Transformation `json:"transformation,omitempty"`
 }
 
-// StaticSecretSyncConfig configures sync behavior from Vault to VSO
-type StaticSecretSyncConfig struct {
+// VaultStaticSecretSyncConfig configures sync behavior from Vault to VSO
+type VaultStaticSecretSyncConfig struct {
 	// InstantUpdates is a flag to indicate that event-driven updates are
-	// enabled for a VaultStaticSecret
+	// enabled for this VaultStaticSecret
 	InstantUpdates bool `json:"instantUpdates,omitempty"`
 }
 

api/v1beta1/zz_generated.deepcopy.go

@@ -296,7 +296,7 @@ spec:
                   instantUpdates:
                     description: |-
                       InstantUpdates is a flag to indicate that event-driven updates are
-                      enabled for VaultStaticSecret/VaultDynamicSecret
+                      enabled for this VaultDynamicSecret
                     type: boolean
                 type: object
               vaultAuthRef:

chart/crds/secrets.hashicorp.com_vaultdynamicsecrets.yaml

@@ -257,7 +257,7 @@ spec:
                   instantUpdates:
                     description: |-
                       InstantUpdates is a flag to indicate that event-driven updates are
-                      enabled for VaultStaticSecret/VaultDynamicSecret
+                      enabled for this VaultStaticSecret
                     type: boolean
                 type: object
               type:

chart/crds/secrets.hashicorp.com_vaultstaticsecrets.yaml

@@ -296,7 +296,7 @@ spec:
                   instantUpdates:
                     description: |-
                       InstantUpdates is a flag to indicate that event-driven updates are
-                      enabled for VaultDynamicSecret
+                      enabled for this VaultDynamicSecret
                     type: boolean
                 type: object
               vaultAuthRef:

config/crd/bases/secrets.hashicorp.com_vaultdynamicsecrets.yaml

@@ -257,7 +257,7 @@ spec:
                   instantUpdates:
                     description: |-
                       InstantUpdates is a flag to indicate that event-driven updates are
-                      enabled for VaultStaticSecret
+                      enabled for this VaultStaticSecret
                     type: boolean
                 type: object
               type:

config/crd/bases/secrets.hashicorp.com_vaultstaticsecrets.yaml

@@ -5,7 +5,11 @@ package controllers
 
 import (
 	"context"
+	"fmt"
+	"time"
 
+	"github.com/cenkalti/backoff/v4"
+	"github.com/go-logr/logr"
 	gocache "github.com/patrickmn/go-cache"
 	"k8s.io/apimachinery/pkg/types"
 )
@@ -23,6 +27,19 @@ type eventWatcherMeta struct {
 	// LastClientID - vault client ID for the last successful connection, used
 	// to detect if the Vault client has changed since the event watcher started
 	LastClientID string
+	// ListenerID tracks the identifier for event listeners registered on a
+	// Vault client.
+	ListenerID string
+	// ErrorCount records the number of consecutive errors encountered while
+	// handling events for this watcher/listener.
+	ErrorCount int
+	// ErrorThreshold defines how many consecutive errors should trigger a
+	// requeue of the resource.
+	ErrorThreshold int
+	// Backoff defines the retry behavior when handling errors.
+	Backoff *backoff.ExponentialBackOff `json:"-"`
+	// RetryCancel cancels any pending requeue timers.
+	RetryCancel context.CancelFunc `json:"-"`
 }
 
 // eventWatcherRegistry - registry for keeping track of running event watcher
@@ -32,6 +49,8 @@ type eventWatcherRegistry struct {
 	registry *gocache.Cache
 }
 
+const defaultEventWatcherErrorThreshold = 5
+
 func newEventWatcherRegistry() *eventWatcherRegistry {
 	return &eventWatcherRegistry{
 		registry: gocache.New(gocache.NoExpiration, gocache.NoExpiration),
@@ -57,3 +76,46 @@ func (r *eventWatcherRegistry) Get(key types.NamespacedName) (*eventWatcherMeta,
 func (r *eventWatcherRegistry) Delete(key types.NamespacedName) {
 	r.registry.Delete(key.String())
 }
+
+func resetEventWatcherMeta(meta *eventWatcherMeta) {
+	if meta == nil {
+		return
+	}
+	meta.ErrorCount = 0
+	if meta.Backoff != nil {
+		meta.Backoff.Reset()
+	}
+	if meta.RetryCancel != nil {
+		meta.RetryCancel()
+		meta.RetryCancel = nil
+	}
+}
+
+// Stop cancels the watcher/listener referenced by meta, waits for it to finish,
+// and deletes the registry entry.
+func (r *eventWatcherRegistry) Stop(ctx context.Context, key types.NamespacedName, meta *eventWatcherMeta, logger logr.Logger) {
+	if meta == nil {
+		r.Delete(key)
+		return
+	}
+
+	if meta.Cancel != nil {
+		meta.Cancel()
+	} else if logger.GetSink() != nil {
+		logger.Error(fmt.Errorf("nil cancel function"), "event watcher has nil cancel function", "key", key)
+	}
+
+	if meta.StoppedCh != nil {
+		waitCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+		defer cancel()
+		if err := waitForStoppedCh(waitCtx, meta.StoppedCh); err != nil && logger.GetSink() != nil {
+			logger.Error(err, "Failed to stop event watcher", "key", key)
+		}
+	}
+	if meta.RetryCancel != nil {
+		meta.RetryCancel()
+		meta.RetryCancel = nil
+	}
+
+	r.Delete(key)
+}