diff --git a/backend/src/ee/services/event/event-sse-stream.ts b/backend/src/ee/services/event/event-sse-stream.ts index dc7d15c79..13e18374f 100644 --- a/backend/src/ee/services/event/event-sse-stream.ts +++ b/backend/src/ee/services/event/event-sse-stream.ts @@ -123,7 +123,7 @@ export function createEventStreamClient(redis: Redis, options: IEventStreamClien await redis.set(key, "1", "EX", 60); - stream.push("1"); + send({ type: "ping" }); }; const close = () => { diff --git a/backend/src/ee/services/permission/default-roles.ts b/backend/src/ee/services/permission/default-roles.ts index cca4efaf2..349130d8e 100644 --- a/backend/src/ee/services/permission/default-roles.ts +++ b/backend/src/ee/services/permission/default-roles.ts @@ -13,6 +13,7 @@ import { ProjectPermissionPkiSubscriberActions, ProjectPermissionPkiTemplateActions, ProjectPermissionSecretActions, + ProjectPermissionSecretEventActions, ProjectPermissionSecretRotationActions, ProjectPermissionSecretScanningConfigActions, ProjectPermissionSecretScanningDataSourceActions, @@ -252,6 +253,16 @@ const buildAdminPermissionRules = () => { ProjectPermissionSub.SecretScanningConfigs ); + can( + [ + ProjectPermissionSecretEventActions.SubscribeCreated, + ProjectPermissionSecretEventActions.SubscribeDeleted, + ProjectPermissionSecretEventActions.SubscribeUpdated, + ProjectPermissionSecretEventActions.SubscribeImportMutations + ], + ProjectPermissionSub.SecretEvents + ); + return rules; }; @@ -455,6 +466,16 @@ const buildMemberPermissionRules = () => { can([ProjectPermissionSecretScanningConfigActions.Read], ProjectPermissionSub.SecretScanningConfigs); + can( + [ + ProjectPermissionSecretEventActions.SubscribeCreated, + ProjectPermissionSecretEventActions.SubscribeDeleted, + ProjectPermissionSecretEventActions.SubscribeUpdated, + ProjectPermissionSecretEventActions.SubscribeImportMutations + ], + ProjectPermissionSub.SecretEvents + ); + return rules; }; @@ -505,6 +526,16 @@ const buildViewerPermissionRules = () => { can([ProjectPermissionSecretScanningConfigActions.Read], ProjectPermissionSub.SecretScanningConfigs); + can( + [ + ProjectPermissionSecretEventActions.SubscribeCreated, + ProjectPermissionSecretEventActions.SubscribeDeleted, + ProjectPermissionSecretEventActions.SubscribeUpdated, + ProjectPermissionSecretEventActions.SubscribeImportMutations + ], + ProjectPermissionSub.SecretEvents + ); + return rules; }; diff --git a/k8-operator/api/v1alpha1/infisicalsecret_types.go b/k8-operator/api/v1alpha1/infisicalsecret_types.go index e4da2911b..e2e0c7b66 100644 --- a/k8-operator/api/v1alpha1/infisicalsecret_types.go +++ b/k8-operator/api/v1alpha1/infisicalsecret_types.go @@ -160,6 +160,9 @@ type InfisicalSecretSpec struct { // +kubebuilder:validation:Optional TLS TLSConfig `json:"tls"` + + // +kubebuilder:default:=false + InstantUpdates bool `json:"instantUpdates"` } // InfisicalSecretStatus defines the observed state of InfisicalSecret diff --git a/k8-operator/config/crd/bases/secrets.infisical.com_infisicalsecrets.yaml b/k8-operator/config/crd/bases/secrets.infisical.com_infisicalsecrets.yaml index 3cf97dbaf..0dcc39f17 100644 --- a/k8-operator/config/crd/bases/secrets.infisical.com_infisicalsecrets.yaml +++ b/k8-operator/config/crd/bases/secrets.infisical.com_infisicalsecrets.yaml @@ -314,6 +314,9 @@ spec: hostAPI: description: Infisical host to pull secrets from type: string + instantUpdates: + default: false + type: boolean managedKubeConfigMapReferences: items: properties: @@ -469,6 +472,7 @@ spec: - secretNamespace type: object required: + - instantUpdates - resyncInterval type: object status: diff --git a/k8-operator/config/samples/crd/infisicalsecret/infisicalSecretCrd.yaml b/k8-operator/config/samples/crd/infisicalsecret/infisicalSecretCrd.yaml index b18f5df82..6ea6dc561 100644 --- a/k8-operator/config/samples/crd/infisicalsecret/infisicalSecretCrd.yaml +++ b/k8-operator/config/samples/crd/infisicalsecret/infisicalSecretCrd.yaml @@ -9,6 +9,7 @@ metadata: spec: hostAPI: http://localhost:8080/api resyncInterval: 10 + instantUpdates: false # tls: # caRef: # secretName: custom-ca-certificate diff --git a/k8-operator/config/samples/k8s-auth/sample.yaml b/k8-operator/config/samples/k8s-auth/sample.yaml index 6dcbae13a..91f910aff 100644 --- a/k8-operator/config/samples/k8s-auth/sample.yaml +++ b/k8-operator/config/samples/k8s-auth/sample.yaml @@ -29,4 +29,4 @@ spec: secretName: managed-secret-k8s secretNamespace: default creationPolicy: "Orphan" ## Owner | Orphan - # secretType: kubernetes.io/dockerconfigjson + # secretType: kubernetes.io/dockerconfigjson \ No newline at end of file diff --git a/k8-operator/config/samples/serviceTokenSecret.yaml b/k8-operator/config/samples/serviceTokenSecret.yaml index 3fd2e1b8a..dc03abcc2 100644 --- a/k8-operator/config/samples/serviceTokenSecret.yaml +++ b/k8-operator/config/samples/serviceTokenSecret.yaml @@ -1,7 +1,7 @@ -apiVersion: v1 -kind: Secret -metadata: - name: service-token -type: Opaque -data: - infisicalToken: \ No newline at end of file +# apiVersion: v1 +# kind: Secret +# metadata: +# name: service-token +# type: Opaque +# data: +# infisicalToken: \ No newline at end of file diff --git a/k8-operator/config/samples/universalAuthIdentitySecret.yaml b/k8-operator/config/samples/universalAuthIdentitySecret.yaml index 88d60e6ab..741de34ee 100644 --- a/k8-operator/config/samples/universalAuthIdentitySecret.yaml +++ b/k8-operator/config/samples/universalAuthIdentitySecret.yaml @@ -4,5 +4,5 @@ metadata: name: universal-auth-credentials type: Opaque stringData: - clientId: da81e27e-1885-47d9-9ea3-ec7d4d807bb6 - clientSecret: 2772414d440fe04d8b975f5fe25acd0fbfe71b2a4a420409eb9ac6f5ae6c1e98 + clientId: your-client-id-here + clientSecret: your-client-secret-here \ No newline at end of file diff --git a/k8-operator/internal/api/api.go b/k8-operator/internal/api/api.go index 36edfa5c1..2e57d5930 100644 --- a/k8-operator/internal/api/api.go +++ b/k8-operator/internal/api/api.go @@ -1,8 +1,11 @@ package api import ( + "encoding/json" "fmt" + "net/http" + "github.com/Infisical/infisical/k8-operator/internal/model" "github.com/go-resty/resty/v2" ) @@ -146,3 +149,85 @@ func CallGetProjectByID(httpClient *resty.Client, request GetProjectByIDRequest) return projectResponse, nil } + +func CallGetProjectByIDv2(httpClient *resty.Client, request GetProjectByIDRequest) (model.Project, error) { + var projectResponse model.Project + + response, err := httpClient. + R().SetResult(&projectResponse). + SetHeader("User-Agent", USER_AGENT_NAME). + Get(fmt.Sprintf("%s/v2/workspace/%s", API_HOST_URL, request.ProjectID)) + + if err != nil { + return model.Project{}, fmt.Errorf("CallGetProject: Unable to complete api request [err=%s]", err) + } + + if response.IsError() { + return model.Project{}, fmt.Errorf("CallGetProject: Unsuccessful response: [response=%s]", response) + } + + return projectResponse, nil + +} + +func CallSubscribeProjectEvents(httpClient *resty.Client, projectId, secretsPath, envSlug, token string) (*http.Response, error) { + conditions := &SubscribeProjectEventsRequestCondition{ + SecretPath: secretsPath, + EnvironmentSlug: envSlug, + } + + body, err := json.Marshal(&SubscribeProjectEventsRequest{ + ProjectID: projectId, + Register: []SubscribeProjectEventsRequestRegister{ + { + Event: "secret:create", + Conditions: conditions, + }, + { + Event: "secret:update", + Conditions: conditions, + }, + { + Event: "secret:delete", + Conditions: conditions, + }, + { + Event: "secret:import-mutation", + Conditions: conditions, + }, + }, + }) + + if err != nil { + return nil, fmt.Errorf("CallSubscribeProjectEvents: Unable to marshal body [err=%s]", err) + } + + response, err := httpClient. + R(). + SetDoNotParseResponse(true). + SetHeader("User-Agent", USER_AGENT_NAME). + SetHeader("Content-Type", "application/json"). + SetHeader("Accept", "text/event-stream"). + SetHeader("Connection", "keep-alive"). + SetHeader("Authorization", fmt.Sprint("Bearer ", token)). + SetBody(body). + Post(fmt.Sprintf("%s/v1/events/subscribe/project-events", API_HOST_URL)) + + if err != nil { + return nil, fmt.Errorf("CallSubscribeProjectEvents: Unable to complete api request [err=%s]", err) + } + + if response.IsError() { + data := struct { + Message string `json:"message"` + }{} + + if err := json.NewDecoder(response.RawBody()).Decode(&data); err != nil { + return nil, err + } + + return nil, fmt.Errorf("CallSubscribeProjectEvents: Unsuccessful response: [message=%s]", data.Message) + } + + return response.RawResponse, nil +} diff --git a/k8-operator/internal/api/models.go b/k8-operator/internal/api/models.go index 2128aac2a..49e1a8c8c 100644 --- a/k8-operator/internal/api/models.go +++ b/k8-operator/internal/api/models.go @@ -206,3 +206,20 @@ type GetProjectByIDRequest struct { type GetProjectByIDResponse struct { Project model.Project `json:"workspace"` } + +type SubscribeProjectEventsRequestRegister struct { + Event string `json:"event"` + Conditions *SubscribeProjectEventsRequestCondition `json:"conditions"` +} + +type SubscribeProjectEventsRequestCondition struct { + EnvironmentSlug string `json:"environmentSlug"` + SecretPath string `json:"secretPath"` +} + +type SubscribeProjectEventsRequest struct { + ProjectID string `json:"projectId"` + Register []SubscribeProjectEventsRequestRegister `json:"register"` +} + +type SubscribeProjectEventsResponse struct{} diff --git a/k8-operator/internal/controller/infisicalpushsecret_controller.go b/k8-operator/internal/controller/infisicalpushsecret_controller.go index e8665b5d9..a8854d880 100644 --- a/k8-operator/internal/controller/infisicalpushsecret_controller.go +++ b/k8-operator/internal/controller/infisicalpushsecret_controller.go @@ -231,7 +231,6 @@ func (r *InfisicalPushSecretReconciler) Reconcile(ctx context.Context, req ctrl. } func (r *InfisicalPushSecretReconciler) SetupWithManager(mgr ctrl.Manager) error { - // Custom predicate that allows both spec changes and deletions specChangeOrDelete := predicate.Funcs{ UpdateFunc: func(e event.UpdateEvent) bool { diff --git a/k8-operator/internal/controller/infisicalsecret_controller.go b/k8-operator/internal/controller/infisicalsecret_controller.go index d5d9c5d39..07552f9cf 100644 --- a/k8-operator/internal/controller/infisicalsecret_controller.go +++ b/k8-operator/internal/controller/infisicalsecret_controller.go @@ -31,6 +31,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/source" secretsv1alpha1 "github.com/Infisical/infisical/k8-operator/api/v1alpha1" "github.com/Infisical/infisical/k8-operator/internal/controllerhelpers" @@ -41,8 +42,10 @@ import ( // InfisicalSecretReconciler reconciles a InfisicalSecret object type InfisicalSecretReconciler struct { client.Client - BaseLogger logr.Logger - Scheme *runtime.Scheme + BaseLogger logr.Logger + Scheme *runtime.Scheme + + SourceCh chan event.TypedGenericEvent[client.Object] Namespace string IsNamespaceScoped bool } @@ -74,7 +77,6 @@ func (r *InfisicalSecretReconciler) GetLogger(req ctrl.Request) logr.Logger { // For more details, check Reconcile and its Result here: // - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.21.0/pkg/reconcile func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { - logger := r.GetLogger(req) var infisicalSecretCRD secretsv1alpha1.InfisicalSecret @@ -196,6 +198,20 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ }, nil } + if infisicalSecretCRD.Spec.InstantUpdates { + if err := handler.OpenInstantUpdatesStream(ctx, logger, &infisicalSecretCRD, infisicalSecretResourceVariablesMap, r.SourceCh); err != nil { + requeueTime = time.Second * 10 + logger.Info(fmt.Sprintf("event stream failed. Will requeue after [requeueTime=%v] [error=%s]", requeueTime, err.Error())) + return ctrl.Result{ + RequeueAfter: requeueTime, + }, nil + } + + logger.Info("Instant updates are enabled") + } else { + handler.CloseInstantUpdatesStream(ctx, logger, &infisicalSecretCRD, infisicalSecretResourceVariablesMap) + } + // Sync again after the specified time logger.Info(fmt.Sprintf("Successfully synced %d secrets. Operator will requeue after [%v]", secretsCount, requeueTime)) return ctrl.Result{ @@ -204,7 +220,12 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ } func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error { + r.SourceCh = make(chan event.TypedGenericEvent[client.Object]) + return ctrl.NewControllerManagedBy(mgr). + WatchesRawSource( + source.Channel[client.Object](r.SourceCh, &util.EnqueueDelayedEventHandler{Delay: time.Second * 10}), + ). For(&secretsv1alpha1.InfisicalSecret{}, builder.WithPredicates(predicate.Funcs{ UpdateFunc: func(e event.UpdateEvent) bool { if e.ObjectOld.GetGeneration() == e.ObjectNew.GetGeneration() { @@ -230,4 +251,5 @@ func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error { }, })). Complete(r) + } diff --git a/k8-operator/internal/services/infisicalsecret/handler.go b/k8-operator/internal/services/infisicalsecret/handler.go index 5657d701d..7fa75581a 100644 --- a/k8-operator/internal/services/infisicalsecret/handler.go +++ b/k8-operator/internal/services/infisicalsecret/handler.go @@ -7,6 +7,7 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" "github.com/Infisical/infisical/k8-operator/api/v1alpha1" "github.com/Infisical/infisical/k8-operator/internal/api" @@ -100,3 +101,22 @@ func (h *InfisicalSecretHandler) SetInfisicalAutoRedeploymentReady(ctx context.C } reconciler.SetInfisicalAutoRedeploymentReady(ctx, logger, infisicalSecret, numDeployments, errorToConditionOn) } + +func (h *InfisicalSecretHandler) CloseInstantUpdatesStream(ctx context.Context, logger logr.Logger, infisicalSecret *v1alpha1.InfisicalSecret, resourceVariablesMap map[string]util.ResourceVariables) error { + reconciler := &InfisicalSecretReconciler{ + Client: h.Client, + Scheme: h.Scheme, + IsNamespaceScoped: h.IsNamespaceScoped, + } + return reconciler.CloseInstantUpdatesStream(ctx, logger, infisicalSecret, resourceVariablesMap) +} + +// Ensures that SSE stream is open, incase if the stream is already opened - this is a noop +func (h *InfisicalSecretHandler) OpenInstantUpdatesStream(ctx context.Context, logger logr.Logger, infisicalSecret *v1alpha1.InfisicalSecret, resourceVariablesMap map[string]util.ResourceVariables, eventCh chan<- event.TypedGenericEvent[client.Object]) error { + reconciler := &InfisicalSecretReconciler{ + Client: h.Client, + Scheme: h.Scheme, + IsNamespaceScoped: h.IsNamespaceScoped, + } + return reconciler.OpenInstantUpdatesStream(ctx, logger, infisicalSecret, resourceVariablesMap, eventCh) +} diff --git a/k8-operator/internal/services/infisicalsecret/reconciler.go b/k8-operator/internal/services/infisicalsecret/reconciler.go index 94596fb8c..3bd9de34a 100644 --- a/k8-operator/internal/services/infisicalsecret/reconciler.go +++ b/k8-operator/internal/services/infisicalsecret/reconciler.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "net/http" "strings" tpl "text/template" @@ -15,11 +16,14 @@ import ( "github.com/Infisical/infisical/k8-operator/internal/model" "github.com/Infisical/infisical/k8-operator/internal/template" "github.com/Infisical/infisical/k8-operator/internal/util" + "github.com/Infisical/infisical/k8-operator/internal/util/sse" "github.com/go-logr/logr" + "github.com/go-resty/resty/v2" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" infisicalSdk "github.com/infisical/go-sdk" corev1 "k8s.io/api/core/v1" @@ -409,9 +413,10 @@ func (r *InfisicalSecretReconciler) getResourceVariables(infisicalSecret v1alpha }) resourceVariablesMap[string(infisicalSecret.UID)] = util.ResourceVariables{ - InfisicalClient: client, - CancelCtx: cancel, - AuthDetails: util.AuthenticationDetails{}, + InfisicalClient: client, + CancelCtx: cancel, + AuthDetails: util.AuthenticationDetails{}, + ServerSentEvents: sse.NewConnectionRegistry(ctx), } resourceVariables = resourceVariablesMap[string(infisicalSecret.UID)] @@ -421,7 +426,6 @@ func (r *InfisicalSecretReconciler) getResourceVariables(infisicalSecret v1alpha } return resourceVariables - } func (r *InfisicalSecretReconciler) updateResourceVariables(infisicalSecret v1alpha1.InfisicalSecret, resourceVariables util.ResourceVariables, resourceVariablesMap map[string]util.ResourceVariables) { @@ -454,9 +458,10 @@ func (r *InfisicalSecretReconciler) ReconcileInfisicalSecret(ctx context.Context } r.updateResourceVariables(*infisicalSecret, util.ResourceVariables{ - InfisicalClient: infisicalClient, - CancelCtx: cancelCtx, - AuthDetails: authDetails, + InfisicalClient: infisicalClient, + CancelCtx: cancelCtx, + AuthDetails: authDetails, + ServerSentEvents: sse.NewConnectionRegistry(ctx), }, resourceVariablesMap) } @@ -525,3 +530,94 @@ func (r *InfisicalSecretReconciler) ReconcileInfisicalSecret(ctx context.Context return secretsCount, nil } + +func (r *InfisicalSecretReconciler) CloseInstantUpdatesStream(ctx context.Context, logger logr.Logger, infisicalSecret *v1alpha1.InfisicalSecret, resourceVariablesMap map[string]util.ResourceVariables) error { + if infisicalSecret == nil { + return fmt.Errorf("infisicalSecret is nil") + } + + variables := r.getResourceVariables(*infisicalSecret, resourceVariablesMap) + + if !variables.AuthDetails.IsMachineIdentityAuth { + return fmt.Errorf("only machine identity is supported for subscriptions") + } + + conn := variables.ServerSentEvents + + if _, ok := conn.Get(); ok { + conn.Close() + } + + return nil +} + +func (r *InfisicalSecretReconciler) OpenInstantUpdatesStream(ctx context.Context, logger logr.Logger, infisicalSecret *v1alpha1.InfisicalSecret, resourceVariablesMap map[string]util.ResourceVariables, eventCh chan<- event.TypedGenericEvent[client.Object]) error { + if infisicalSecret == nil { + return fmt.Errorf("infisicalSecret is nil") + } + + variables := r.getResourceVariables(*infisicalSecret, resourceVariablesMap) + + if !variables.AuthDetails.IsMachineIdentityAuth { + return fmt.Errorf("only machine identity is supported for subscriptions") + } + + projectSlug := variables.AuthDetails.MachineIdentityScope.ProjectSlug + secretsPath := variables.AuthDetails.MachineIdentityScope.SecretsPath + envSlug := variables.AuthDetails.MachineIdentityScope.EnvSlug + + infiscalClient := variables.InfisicalClient + sseRegistry := variables.ServerSentEvents + + token := infiscalClient.Auth().GetAccessToken() + + project, err := util.GetProjectBySlug(token, projectSlug) + + if err != nil { + return fmt.Errorf("failed to get project [err=%s]", err) + } + + if variables.AuthDetails.MachineIdentityScope.Recursive { + secretsPath = fmt.Sprint(secretsPath, "**") + } + + if err != nil { + return fmt.Errorf("CallSubscribeProjectEvents: unable to marshal body [err=%s]", err) + } + + events, errors, err := sseRegistry.Subscribe(func() (*http.Response, error) { + httpClient := resty.New() + + req, err := api.CallSubscribeProjectEvents(httpClient, project.ID, secretsPath, envSlug, token) + + if err != nil { + return nil, err + } + + return req, nil + }) + + if err != nil { + return fmt.Errorf("unable to connect sse [err=%s]", err) + } + + go func() { + outer: + for { + select { + case ev := <-events: + logger.Info("Received SSE Event", "event", ev) + eventCh <- event.TypedGenericEvent[client.Object]{ + Object: infisicalSecret, + } + case err := <-errors: + logger.Error(err, "Error occurred") + break outer + case <-ctx.Done(): + break outer + } + } + }() + + return nil +} diff --git a/k8-operator/internal/util/handler.go b/k8-operator/internal/util/handler.go new file mode 100644 index 000000000..d1cc86562 --- /dev/null +++ b/k8-operator/internal/util/handler.go @@ -0,0 +1,59 @@ +package util + +import ( + "context" + "math/rand" + "time" + + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/workqueue" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +// computeMaxJitterDuration returns a random duration between 0 and max. +// This is useful for introducing jitter to event processing. +func computeMaxJitterDuration(max time.Duration) (time.Duration, time.Duration) { + if max <= 0 { + return 0, 0 + } + jitter := time.Duration(rand.Int63n(int64(max))) + return max, jitter +} + +// EnqueueDelayedEventHandler enqueues reconcile requests with a random delay (jitter) +// to spread the load and avoid thundering herd issues. +type EnqueueDelayedEventHandler struct { + Delay time.Duration +} + +func (e *EnqueueDelayedEventHandler) Create(_ context.Context, _ event.TypedCreateEvent[client.Object], _ workqueue.TypedRateLimitingInterface[reconcile.Request]) { +} + +func (e *EnqueueDelayedEventHandler) Update(_ context.Context, _ event.TypedUpdateEvent[client.Object], _ workqueue.TypedRateLimitingInterface[reconcile.Request]) { +} + +func (e *EnqueueDelayedEventHandler) Delete(_ context.Context, _ event.TypedDeleteEvent[client.Object], _ workqueue.TypedRateLimitingInterface[reconcile.Request]) { +} + +func (e *EnqueueDelayedEventHandler) Generic(_ context.Context, evt event.TypedGenericEvent[client.Object], q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + if evt.Object == nil { + return + } + + req := reconcile.Request{ + NamespacedName: types.NamespacedName{ + Namespace: evt.Object.GetNamespace(), + Name: evt.Object.GetName(), + }, + } + + _, delay := computeMaxJitterDuration(e.Delay) + + if delay > 0 { + q.AddAfter(req, delay) + } else { + q.Add(req) + } +} diff --git a/k8-operator/internal/util/models.go b/k8-operator/internal/util/models.go index 8030731c2..e583858fa 100644 --- a/k8-operator/internal/util/models.go +++ b/k8-operator/internal/util/models.go @@ -3,11 +3,13 @@ package util import ( "context" + "github.com/Infisical/infisical/k8-operator/internal/util/sse" infisicalSdk "github.com/infisical/go-sdk" ) type ResourceVariables struct { - InfisicalClient infisicalSdk.InfisicalClientInterface - CancelCtx context.CancelFunc - AuthDetails AuthenticationDetails + InfisicalClient infisicalSdk.InfisicalClientInterface + CancelCtx context.CancelFunc + AuthDetails AuthenticationDetails + ServerSentEvents *sse.ConnectionRegistry } diff --git a/k8-operator/internal/util/sse/sse.go b/k8-operator/internal/util/sse/sse.go new file mode 100644 index 000000000..7bfbcef88 --- /dev/null +++ b/k8-operator/internal/util/sse/sse.go @@ -0,0 +1,331 @@ +package sse + +import ( + "bufio" + "context" + "io" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" +) + +// Event represents a Server-Sent Event +type Event struct { + ID string + Event string + Data string + Retry int +} + +// ConnectionMeta holds metadata about an SSE connection +type ConnectionMeta struct { + EventChan <-chan Event + ErrorChan <-chan error + lastPingAt atomic.Value // stores time.Time + cancel context.CancelFunc +} + +// LastPing returns the last ping time +func (c *ConnectionMeta) LastPing() time.Time { + if t, ok := c.lastPingAt.Load().(time.Time); ok { + return t + } + return time.Time{} +} + +// UpdateLastPing atomically updates the last ping time +func (c *ConnectionMeta) UpdateLastPing() { + c.lastPingAt.Store(time.Now()) +} + +// Cancel terminates the connection +func (c *ConnectionMeta) Cancel() { + if c.cancel != nil { + c.cancel() + } +} + +// ConnectionRegistry manages SSE connections with high performance +type ConnectionRegistry struct { + mu sync.RWMutex + conn *ConnectionMeta + + monitorOnce sync.Once + monitorStop chan struct{} + + onPing func() // Callback for ping events +} + +// NewConnectionRegistry creates a new high-performance connection registry +func NewConnectionRegistry(ctx context.Context) *ConnectionRegistry { + r := &ConnectionRegistry{ + monitorStop: make(chan struct{}), + } + + // Configure ping handler + r.onPing = func() { + r.UpdateLastPing() + } + + return r +} + +// Subscribe provides SSE events, creating a connection if needed +func (r *ConnectionRegistry) Subscribe(request func() (*http.Response, error)) (<-chan Event, <-chan error, error) { + // Fast path: check if connection exists + if conn := r.getConnection(); conn != nil { + return conn.EventChan, conn.ErrorChan, nil + } + + // Slow path: create new connection under lock + r.mu.Lock() + defer r.mu.Unlock() + + // Double-check after acquiring lock + if r.conn != nil { + return r.conn.EventChan, r.conn.ErrorChan, nil + } + + res, err := request() + if err != nil { + return nil, nil, err + } + + conn, err := r.createStream(res) + if err != nil { + return nil, nil, err + } + + r.conn = conn + + // Start monitor once + r.monitorOnce.Do(func() { + go r.monitorConnections() + }) + + return conn.EventChan, conn.ErrorChan, nil +} + +// Get retrieves the current connection +func (r *ConnectionRegistry) Get() (*ConnectionMeta, bool) { + conn := r.getConnection() + return conn, conn != nil +} + +// IsConnected checks if there's an active connection +func (r *ConnectionRegistry) IsConnected() bool { + return r.getConnection() != nil +} + +// UpdateLastPing updates the last ping time for the current connection +func (r *ConnectionRegistry) UpdateLastPing() { + if conn := r.getConnection(); conn != nil { + conn.UpdateLastPing() + } +} + +// Close gracefully shuts down the registry +func (r *ConnectionRegistry) Close() { + // Stop monitor first + select { + case <-r.monitorStop: + // Already closed + default: + close(r.monitorStop) + } + + // Close connection + r.mu.Lock() + if r.conn != nil { + r.conn.Cancel() + r.conn = nil + } + r.mu.Unlock() +} + +// getConnection returns the current connection without locking +func (r *ConnectionRegistry) getConnection() *ConnectionMeta { + r.mu.RLock() + conn := r.conn + r.mu.RUnlock() + return conn +} + +func (r *ConnectionRegistry) createStream(res *http.Response) (*ConnectionMeta, error) { + ctx, cancel := context.WithCancel(context.Background()) + + eventChan, errorChan, err := r.stream(ctx, res) + if err != nil { + cancel() + return nil, err + } + + meta := &ConnectionMeta{ + EventChan: eventChan, + ErrorChan: errorChan, + cancel: cancel, + } + meta.UpdateLastPing() + + return meta, nil +} + +// stream processes SSE data from an HTTP response +func (r *ConnectionRegistry) stream(ctx context.Context, res *http.Response) (<-chan Event, <-chan error, error) { + eventChan := make(chan Event, 10) + errorChan := make(chan error, 1) + + go r.processStream(ctx, res.Body, eventChan, errorChan) + + return eventChan, errorChan, nil +} + +// processStream reads and parses SSE events from the response body +func (r *ConnectionRegistry) processStream(ctx context.Context, body io.ReadCloser, eventChan chan<- Event, errorChan chan<- error) { + defer body.Close() + defer close(eventChan) + defer close(errorChan) + + scanner := bufio.NewScanner(body) + + var currentEvent Event + var dataBuilder strings.Builder + + for scanner.Scan() { + select { + case <-ctx.Done(): + return + default: + } + + line := scanner.Text() + + // Empty line indicates end of event + if len(line) == 0 { + if currentEvent.Data != "" || currentEvent.Event != "" { + // Finalize data + if dataBuilder.Len() > 0 { + currentEvent.Data = dataBuilder.String() + dataBuilder.Reset() + } + + // Handle ping events + if r.isPingEvent(currentEvent) { + if r.onPing != nil { + r.onPing() + } + } else { + // Send non-ping events + select { + case eventChan <- currentEvent: + case <-ctx.Done(): + return + } + } + + // Reset for next event + currentEvent = Event{} + } + continue + } + + // Parse line efficiently + r.parseLine(line, ¤tEvent, &dataBuilder) + } + + if err := scanner.Err(); err != nil { + select { + case errorChan <- err: + case <-ctx.Done(): + } + } +} + +// parseLine efficiently parses SSE protocol lines +func (r *ConnectionRegistry) parseLine(line string, event *Event, dataBuilder *strings.Builder) { + colonIndex := strings.IndexByte(line, ':') + if colonIndex == -1 { + return // Invalid line format + } + + field := line[:colonIndex] + value := line[colonIndex+1:] + + // Trim leading space from value (SSE spec) + if len(value) > 0 && value[0] == ' ' { + value = value[1:] + } + + switch field { + case "data": + if dataBuilder.Len() > 0 { + dataBuilder.WriteByte('\n') + } + dataBuilder.WriteString(value) + case "event": + event.Event = value + case "id": + event.ID = value + case "retry": + // Parse retry value if needed + // This could be used to configure reconnection delay + case "": + // Comment line, ignore + } +} + +// isPingEvent checks if an event is a ping/keepalive +func (r *ConnectionRegistry) isPingEvent(event Event) bool { + // Check for common ping patterns + if event.Event == "ping" { + return true + } + + // Check for heartbeat data (common pattern is "1" or similar) + if event.Event == "" && strings.TrimSpace(event.Data) == "1" { + return true + } + + return false +} + +// monitorConnections checks connection health periodically +func (r *ConnectionRegistry) monitorConnections() { + const ( + checkInterval = 30 * time.Second + pingTimeout = 2 * time.Minute + ) + + ticker := time.NewTicker(checkInterval) + defer ticker.Stop() + + for { + select { + case <-r.monitorStop: + return + case <-ticker.C: + r.checkConnectionHealth(pingTimeout) + } + } +} + +// checkConnectionHealth verifies connection is still alive +func (r *ConnectionRegistry) checkConnectionHealth(timeout time.Duration) { + conn := r.getConnection() + if conn == nil { + return + } + + if time.Since(conn.LastPing()) > timeout { + // Connection is stale, close it + r.mu.Lock() + if r.conn == conn { // Verify it's still the same connection + r.conn.Cancel() + r.monitorStop <- struct{}{} + r.conn = nil + } + r.mu.Unlock() + } +} diff --git a/k8-operator/internal/util/workspace.go b/k8-operator/internal/util/workspace.go index d62c0288a..d014a4c4e 100644 --- a/k8-operator/internal/util/workspace.go +++ b/k8-operator/internal/util/workspace.go @@ -9,7 +9,6 @@ import ( ) func GetProjectByID(accessToken string, projectId string) (model.Project, error) { - httpClient := resty.New() httpClient. SetAuthScheme("Bearer"). @@ -25,3 +24,21 @@ func GetProjectByID(accessToken string, projectId string) (model.Project, error) return projectDetails.Project, nil } + +func GetProjectBySlug(accessToken string, projectSlug string) (model.Project, error) { + httpClient := resty.New() + httpClient. + SetAuthScheme("Bearer"). + SetAuthToken(accessToken). + SetHeader("Accept", "application/json") + + project, err := api.CallGetProjectByIDv2(httpClient, api.GetProjectByIDRequest{ + ProjectID: projectSlug, + }) + + if err != nil { + return model.Project{}, fmt.Errorf("unable to get project by slug. [err=%v]", err) + } + + return project, nil +}