diff --git a/k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go b/k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go index 343f91b38..075b25d37 100644 --- a/k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go +++ b/k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go @@ -7,14 +7,11 @@ import ( "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/event" - "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/predicate" - "sigs.k8s.io/controller-runtime/pkg/reconcile" "sigs.k8s.io/controller-runtime/pkg/source" defaultErrors "errors" @@ -62,6 +59,8 @@ func (r *InfisicalSecretReconciler) GetLogger(req ctrl.Request) logr.Logger { func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { logger := r.GetLogger(req) + logger.Info("Reconcile called") + var infisicalSecretCRD secretsv1alpha1.InfisicalSecret requeueTime := time.Minute // seconds @@ -181,8 +180,8 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ } if infisicalSecretCRD.Spec.InstantUpdates { - logger.Info("Event watcher enabled") - // ensure event watcher is open + logger.Info("Instant updates are enabled") + if err := r.EnsureEventStream(ctx, logger, &infisicalSecretCRD); err != nil { logger.Error(err, fmt.Sprintf("unable to ensure event stream. Will requeue after [requeueTime=%v]", requeueTime)) return ctrl.Result{ @@ -190,7 +189,6 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ }, nil } } else { - // ensure event stream is closed r.CloseEventStream(ctx, logger, &infisicalSecretCRD) } @@ -207,15 +205,10 @@ func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). Watches( &source.Channel{Source: r.SourceCh}, - handler.EnqueueRequestsFromMapFunc(r.findSecretsForCluster), + &util.EnqueueDelayedEventHandler{Delay: time.Second * 3}, ). For(&secretsv1alpha1.InfisicalSecret{}, builder.WithPredicates(predicate.Funcs{ - GenericFunc: func(ge event.GenericEvent) bool { - fmt.Println("Generic event recieved") - return true - }, UpdateFunc: func(e event.UpdateEvent) bool { - println("UpdateFunc event recieved") if e.ObjectOld.GetGeneration() == e.ObjectNew.GetGeneration() { return false // Skip reconciliation for status-only changes } @@ -241,31 +234,3 @@ func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error { Complete(r) } - -func (r *InfisicalSecretReconciler) findSecretsForCluster(o client.Object) []reconcile.Request { - ctx := context.Background() - secrets := &secretsv1alpha1.InfisicalSecretList{} - - requests := []reconcile.Request{} - - if err := r.List(ctx, secrets); err != nil { - fmt.Println(err) - return requests - } - - for _, sec := range secrets.Items { - if sec.GetName() == o.GetName() && sec.GetNamespace() == o.GetNamespace() { - requests = append(requests, reconcile.Request{ - NamespacedName: types.NamespacedName{ - Namespace: o.GetNamespace(), - Name: o.GetName(), - }, - }) - break - } - } - - fmt.Println(requests) - - return requests -} diff --git a/k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go b/k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go index 377a5c9fa..5c2f75e5d 100644 --- a/k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go +++ b/k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go @@ -469,7 +469,7 @@ func (r *InfisicalSecretReconciler) getResourceVariables(infisicalSecret v1alpha InfisicalClient: client, CancelCtx: cancel, AuthDetails: util.AuthenticationDetails{}, - ServerSentEvents: sse.NewConnectionRegistry(), + ServerSentEvents: sse.NewConnectionRegistry(ctx), } resourceVariables = infisicalSecretResourceVariablesMap[string(infisicalSecret.UID)] @@ -509,7 +509,7 @@ func (r *InfisicalSecretReconciler) ReconcileInfisicalSecret(ctx context.Context InfisicalClient: infisicalClient, CancelCtx: cancelCtx, AuthDetails: authDetails, - ServerSentEvents: sse.NewConnectionRegistry(), + ServerSentEvents: sse.NewConnectionRegistry(ctx), }) } @@ -638,41 +638,43 @@ func (r *InfisicalSecretReconciler) EnsureEventStream(ctx context.Context, logge req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("%s/v1/events/subscribe/project-events", api.API_HOST_URL), strings.NewReader(string(body))) + if err != nil { + return nil, err + } + for k, v := range headers { req.Header.Set(k, v) } - return req, err + return req, nil }) if err != nil { return fmt.Errorf("unable to connect to SSE server [err=%s]", err) } -outer: - for { - select { - case ev := <-events: - logger.Info("Received event", "secret", secret, "event", ev) - r.SourceCh <- event.GenericEvent{ - Object: secret, + go func() { + outer: + for { + select { + case ev := <-events: + logger.Info("Received SSE Event", "event", ev) + r.SourceCh <- event.GenericEvent{ + Object: secret, + } + case err := <-errors: + logger.Error(err, "Error occurred") + break outer + case <-ctx.Done(): + break outer } - logger.Info("Send to channel") - case err := <-errors: - logger.Error(err, "Error occurred") - break outer - case <-ctx.Done(): - logger.Info("Context done") - break outer } - } + }() return nil } func (r *InfisicalSecretReconciler) CloseEventStream(ctx context.Context, logger logr.Logger, secret *secretsv1alpha1.InfisicalSecret) error { - logger.Info("Event watcher disabled") - if secret == nil { return fmt.Errorf("infisicalSecret is nil") } diff --git a/k8-operator/packages/util/handler.go b/k8-operator/packages/util/handler.go index 4de763d04..ca5e82dcc 100644 --- a/k8-operator/packages/util/handler.go +++ b/k8-operator/packages/util/handler.go @@ -1,7 +1,6 @@ package util import ( - "fmt" "math/rand" "time" @@ -37,7 +36,6 @@ func (e *EnqueueDelayedEventHandler) Delete(_ event.DeleteEvent, _ workqueue.Rat } func (e *EnqueueDelayedEventHandler) Generic(evt event.GenericEvent, q workqueue.RateLimitingInterface) { - fmt.Println(evt) if evt.Object == nil { return } diff --git a/k8-operator/packages/util/models.go b/k8-operator/packages/util/models.go index a104fb957..16de4608d 100644 --- a/k8-operator/packages/util/models.go +++ b/k8-operator/packages/util/models.go @@ -11,5 +11,5 @@ type ResourceVariables struct { InfisicalClient infisicalSdk.InfisicalClientInterface CancelCtx context.CancelFunc AuthDetails AuthenticationDetails - ServerSentEvents sse.ConnectionRegistry + ServerSentEvents *sse.ConnectionRegistry } diff --git a/k8-operator/packages/util/sse/sse.go b/k8-operator/packages/util/sse/sse.go index 471509f7d..ec4a5dbce 100644 --- a/k8-operator/packages/util/sse/sse.go +++ b/k8-operator/packages/util/sse/sse.go @@ -4,134 +4,162 @@ import ( "context" "fmt" "net/http" + "sync" "time" ) -// ConnectionMeta holds metadata about an active SSE connection type ConnectionMeta struct { EventChan <-chan SSEEvent ErrorChan <-chan error - Cancel context.CancelFunc LastPingAt time.Time + Cancel context.CancelFunc } -// ConnectionRegistry manages a single SSE connection with a shared client type ConnectionRegistry struct { - Ticker *time.Ticker + Ctx context.Context meta *ConnectionMeta client SSEClient + mu sync.RWMutex + + monitorCancel context.CancelFunc + monitorCtx context.Context } -// NewConnectionRegistry creates a new registry -func NewConnectionRegistry() ConnectionRegistry { - return ConnectionRegistry{ - Ticker: time.NewTicker(time.Second * 30), - client: NewClient(), +func NewConnectionRegistry(ctx context.Context) *ConnectionRegistry { + monitorCtx, monitorCancel := context.WithCancel(ctx) + return &ConnectionRegistry{ + Ctx: ctx, + client: NewClient(), + monitorCtx: monitorCtx, + monitorCancel: monitorCancel, } } -// GetOrCreate returns existing connection or creates a new one -func (r *ConnectionRegistry) GetOrCreate( - onBuild func() (*http.Request, error), -) (*ConnectionMeta, error) { - // First try to get existing connection - if r.meta != nil { - return r.meta, nil - } - - // Create new connection - req, err := onBuild() - if err != nil { - return nil, fmt.Errorf("failed to build request: %w", err) - } - - // Add cancellation context - ctx, cancel := context.WithCancel(context.Background()) - +// create creates a new connection +func (r *ConnectionRegistry) create(req *http.Request) (*ConnectionMeta, error) { + // Create new connection using provided request eventChan, errorChan, err := r.client.Connect(req) if err != nil { - cancel() return nil, fmt.Errorf("failed to connect: %w", err) } meta := &ConnectionMeta{ EventChan: eventChan, ErrorChan: errorChan, - Cancel: cancel, LastPingAt: time.Now(), } r.meta = meta - // Start cleanup monitor for this connection - go r.monitor(ctx, meta) + // Start cleanup monitor for this connection (NON-BLOCKING) + go r.monitor(meta) + println("Creating new connection\n") return meta, nil } // Get retrieves the existing connection func (r *ConnectionRegistry) Get() (*ConnectionMeta, bool) { + r.mu.RLock() + defer r.mu.RUnlock() return r.meta, r.meta != nil } // Close closes the connection func (r *ConnectionRegistry) Close() { + r.mu.Lock() + defer r.mu.Unlock() + if r.meta != nil { - r.meta.Cancel() + if r.meta.Cancel != nil { + r.meta.Cancel() + } r.meta = nil } + + // Cancel the monitor + if r.monitorCancel != nil { + r.monitorCancel() + } } // IsConnected returns whether there's an active connection func (r *ConnectionRegistry) IsConnected() bool { + r.mu.RLock() + defer r.mu.RUnlock() return r.meta != nil } -// monitorConnection watches for connection closure and cleans up -func (r *ConnectionRegistry) monitor(ctx context.Context, meta *ConnectionMeta) { -outer: - for range r.Ticker.C { +// UpdateLastPing updates the last ping time +func (r *ConnectionRegistry) UpdateLastPing() { + r.mu.Lock() + defer r.mu.Unlock() + + if r.meta != nil { + r.meta.LastPingAt = time.Now() + } +} + +// monitor watches for connection closure and cleans up +func (r *ConnectionRegistry) monitor(meta *ConnectionMeta) { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + for { select { - case <-ctx.Done(): - break outer - default: - if r.IsConnected() && time.Since(r.meta.LastPingAt) > 2*time.Minute { - fmt.Println("Last ping was more than 2 minutes ago") - r.Close() - break outer - } else { - fmt.Println("Last ping was within the last 2 minutes") + case <-r.monitorCtx.Done(): + // Context cancelled, exit monitor + return + + case <-ticker.C: + r.mu.RLock() + currentMeta := r.meta + r.mu.RUnlock() + + // Check if this monitor is still relevant + if currentMeta != meta { + // This connection has been replaced, exit monitor + return + } + + if currentMeta != nil && time.Since(currentMeta.LastPingAt) > 2*time.Minute { + fmt.Println("Last ping was more than 2 minutes ago, closing connection") + r.mu.Lock() + if r.meta == meta { // Double-check under lock + if r.meta.Cancel != nil { + r.meta.Cancel() + } + r.meta = nil + } + r.mu.Unlock() + return // Exit monitor after cleanup } } } - - // Clean up from registry - if r.meta == meta { - r.meta = nil - } -} - -// ConnectionInfo provides read-only info about a connection -type ConnectionInfo struct { - LastPingAt time.Time } // Subscribe provides a convenient way to get events from the connection -func (r *ConnectionRegistry) Subscribe( - onBuild func() (*http.Request, error), -) (<-chan SSEEvent, <-chan error, error) { - meta, err := r.GetOrCreate(onBuild) +func (r *ConnectionRegistry) Subscribe(build func() (*http.Request, error)) (<-chan SSEEvent, <-chan error, error) { + r.mu.Lock() + defer r.mu.Unlock() + + // Get existing connection if available + if r.meta != nil { + return r.meta.EventChan, r.meta.ErrorChan, nil + } + + req, err := build() + + if err != nil { + return nil, nil, err + } + + // Create new connection if none exists + meta, err := r.create(req) + if err != nil { return nil, nil, err } return meta.EventChan, meta.ErrorChan, nil } - -// Reconnect closes existing connection and creates a new one -func (r *ConnectionRegistry) Reconnect( - onBuild func() (*http.Request, error), -) (*ConnectionMeta, error) { - r.Close() - return r.GetOrCreate(onBuild) -}