From 67abcbfe7afcc2c061a2cac8b781c833667f647e Mon Sep 17 00:00:00 2001 From: sidwebworks <58144379+sidwebworks@users.noreply.github.com> Date: Tue, 5 Aug 2025 00:16:47 +0530 Subject: [PATCH] wip: k8s operator changes --- .../api/v1alpha1/infisicalsecret_types.go | 3 + .../infisicalsecret_controller.go | 22 ++++ .../infisicalsecret/infisicalsecret_helper.go | 41 ++++++- k8-operator/packages/api/api.go | 22 ++++ k8-operator/packages/api/models.go | 15 +++ k8-operator/packages/util/models.go | 8 +- k8-operator/packages/util/sse/sse.go | 110 ++++++++++++++++++ 7 files changed, 215 insertions(+), 6 deletions(-) create mode 100644 k8-operator/packages/util/sse/sse.go diff --git a/k8-operator/api/v1alpha1/infisicalsecret_types.go b/k8-operator/api/v1alpha1/infisicalsecret_types.go index ff26a878c..22255e410 100644 --- a/k8-operator/api/v1alpha1/infisicalsecret_types.go +++ b/k8-operator/api/v1alpha1/infisicalsecret_types.go @@ -149,6 +149,9 @@ type InfisicalSecretSpec struct { // +kubebuilder:validation:Optional TLS TLSConfig `json:"tls"` + + // +kubebuilder:default:=true + InstantUpdates bool `json:"instantUpdates"` } // InfisicalSecretStatus defines the observed state of InfisicalSecret diff --git a/k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go b/k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go index bf7d75830..c8a39f5fc 100644 --- a/k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go +++ b/k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go @@ -11,7 +11,9 @@ import ( "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/source" defaultErrors "errors" @@ -27,6 +29,8 @@ type InfisicalSecretReconciler struct { client.Client BaseLogger logr.Logger Scheme *runtime.Scheme + + SourceCh chan event.GenericEvent } const FINALIZER_NAME = "secrets.finalizers.infisical.com" @@ -175,6 +179,20 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ }, nil } + if infisicalSecretCRD.Spec.InstantUpdates { + logger.Info("Event watcher enabled") + // ensure event watcher is open + 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{ + RequeueAfter: requeueTime, + }, nil + } + } else { + // ensure event stream is closed + r.CloseEventStream(ctx, logger, &infisicalSecretCRD) + } + // Sync again after the specified time logger.Info(fmt.Sprintf("Successfully synced %d secrets. Operator will requeue after [%v]", secretsCount, requeueTime)) return ctrl.Result{ @@ -208,5 +226,9 @@ func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error { return true }, })). + Watches( + &source.Channel{Source: r.SourceCh}, + &handler.EnqueueRequestForObject{}, + ). Complete(r) } diff --git a/k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go b/k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go index f2197b0ed..2b01e5f72 100644 --- a/k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go +++ b/k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go @@ -9,12 +9,14 @@ import ( tpl "text/template" "github.com/Infisical/infisical/k8-operator/api/v1alpha1" + secretsv1alpha1 "github.com/Infisical/infisical/k8-operator/api/v1alpha1" "github.com/Infisical/infisical/k8-operator/packages/api" "github.com/Infisical/infisical/k8-operator/packages/constants" "github.com/Infisical/infisical/k8-operator/packages/crypto" "github.com/Infisical/infisical/k8-operator/packages/model" "github.com/Infisical/infisical/k8-operator/packages/template" "github.com/Infisical/infisical/k8-operator/packages/util" + "github.com/Infisical/infisical/k8-operator/packages/util/sse" "github.com/go-logr/logr" "k8s.io/apimachinery/pkg/types" @@ -461,9 +463,10 @@ func (r *InfisicalSecretReconciler) getResourceVariables(infisicalSecret v1alpha }) infisicalSecretResourceVariablesMap[string(infisicalSecret.UID)] = util.ResourceVariables{ - InfisicalClient: client, - CancelCtx: cancel, - AuthDetails: util.AuthenticationDetails{}, + InfisicalClient: client, + CancelCtx: cancel, + AuthDetails: util.AuthenticationDetails{}, + EventStreamClient: sse.NewClient(api.API_HOST_URL, api.USER_AGENT_NAME), } resourceVariables = infisicalSecretResourceVariablesMap[string(infisicalSecret.UID)] @@ -567,3 +570,35 @@ func (r *InfisicalSecretReconciler) ReconcileInfisicalSecret(ctx context.Context return secretsCount, nil } + +func (r *InfisicalSecretReconciler) EnsureEventStream(ctx context.Context, logger logr.Logger, secret *v1alpha1.InfisicalSecret) error { + if secret == nil { + return fmt.Errorf("infisicalSecret is nil") + } + + resourceVariables := r.getResourceVariables(*secret) + infiscalClient := resourceVariables.InfisicalClient + // sseClient := resourceVariables.EventStreamClient + + projectSlug := resourceVariables.AuthDetails.MachineIdentityScope.ProjectSlug + + proj, err := util.GetProjectByID(infiscalClient.Auth().GetAccessToken(), projectSlug) + + logger.Info("Project", proj) + if err != nil { + return fmt.Errorf("failed to get project [err=%s]", err) + } + + api.CallSubscribeProjectEvents(secret.Spec.Authentication.ServiceAccount.ProjectId, api.SubProjectEventsRequest{ + ProjectID: proj.ID, + }) + + return nil +} + +func (r *InfisicalSecretReconciler) CloseEventStream(ctx context.Context, logger logr.Logger, infisicalSecretCRD *secretsv1alpha1.InfisicalSecret) error { + logger.Info("Event watcher enabled") + // ensure event watcher is running + + return nil +} diff --git a/k8-operator/packages/api/api.go b/k8-operator/packages/api/api.go index 36edfa5c1..ddefeeb51 100644 --- a/k8-operator/packages/api/api.go +++ b/k8-operator/packages/api/api.go @@ -1,8 +1,11 @@ package api import ( + "encoding/json" "fmt" + "strings" + "github.com/Infisical/infisical/k8-operator/packages/util/sse" "github.com/go-resty/resty/v2" ) @@ -146,3 +149,22 @@ func CallGetProjectByID(httpClient *resty.Client, request GetProjectByIDRequest) return projectResponse, nil } + +func CallSubscribeProjectEvents(projectID string, body SubProjectEventsRequest) (<-chan sse.SSEEvent, <-chan error, error) { + client := sse.NewClient(fmt.Sprintf("%s/api/v1/events/subscribe/project-events", API_HOST_URL), USER_AGENT_NAME) + + b, err := json.Marshal(body) + if err != nil { + return nil, nil, fmt.Errorf("CallSubscribeProjectEvents: Unable to marshal body [err=%s]", err) + } + + events, errors, err := client.Connect("POST", map[string]string{ + "User-Agent": USER_AGENT_NAME, + }, strings.NewReader(string(b))) + + if err != nil { + return nil, nil, fmt.Errorf("CallSubscribeProjectEvents: Unable to connect to SSE server [err=%s]", err) + } + + return events, errors, err +} diff --git a/k8-operator/packages/api/models.go b/k8-operator/packages/api/models.go index 01f835397..971561683 100644 --- a/k8-operator/packages/api/models.go +++ b/k8-operator/packages/api/models.go @@ -206,3 +206,18 @@ type GetProjectByIDRequest struct { type GetProjectByIDResponse struct { Project model.Project `json:"workspace"` } + +type SubProjectEventsRequestRegister struct { + EventType string `json:"eventType"` + Conditions *struct { + EnvironmentSlug string `json:"environmentSlug"` + SecretPath string `json:"secretPath"` + } `json:"conditions"` +} + +type SubProjectEventsRequest struct { + ProjectID string `json:"projectId"` + Register []SubProjectEventsRequestRegister `json:"register"` +} + +type SubProjectEventsResponse struct{} diff --git a/k8-operator/packages/util/models.go b/k8-operator/packages/util/models.go index 8030731c2..758f6e262 100644 --- a/k8-operator/packages/util/models.go +++ b/k8-operator/packages/util/models.go @@ -3,11 +3,13 @@ package util import ( "context" + "github.com/Infisical/infisical/k8-operator/packages/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 + EventStreamClient *sse.SSEClient } diff --git a/k8-operator/packages/util/sse/sse.go b/k8-operator/packages/util/sse/sse.go new file mode 100644 index 000000000..a0ac2127f --- /dev/null +++ b/k8-operator/packages/util/sse/sse.go @@ -0,0 +1,110 @@ +package sse + +import ( + "bufio" + "fmt" + "io" + "net/http" + "strings" +) + +type SSEEvent struct { + ID string + Event string + Data string +} + +// SSEClient handles SSE connections +type SSEClient struct { + URL string + UserAgent string + Client *http.Client +} + +// NewSSEClient creates a new SSE client +func NewClient(url, ua string) *SSEClient { + return &SSEClient{ + URL: url, + UserAgent: ua, + Client: &http.Client{ + Timeout: 0, // No timeout for streaming + }, + } +} + +// Connect establishes SSE connection and returns a channel of events +func (c *SSEClient) Connect(method string, headers map[string]string, body io.Reader) (<-chan SSEEvent, <-chan error, error) { + req, err := http.NewRequest(method, c.URL, body) + if err != nil { + return nil, nil, err + } + + // Set required headers for SSE + req.Header.Set("Cache-Control", "no-cache") + req.Header.Set("Connection", "keep-alive") + + for key, value := range headers { + req.Header.Set(key, value) + } + + resp, err := c.Client.Do(req) + if err != nil { + return nil, nil, err + } + + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + return nil, nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + eventChan := make(chan SSEEvent) + errorChan := make(chan error) + + go c.readEvents(resp.Body, eventChan, errorChan) + + return eventChan, errorChan, nil +} + +// readEvents reads and parses SSE events from the response body +func (c *SSEClient) readEvents(body io.ReadCloser, eventChan chan<- SSEEvent, errorChan chan<- error) { + defer body.Close() + defer close(eventChan) + defer close(errorChan) + + scanner := bufio.NewScanner(body) + var event SSEEvent + + for scanner.Scan() { + line := scanner.Text() + + // Empty line indicates end of event + if line == "" { + if event.Data != "" || event.Event != "" { + eventChan <- event + event = SSEEvent{} // Reset for next event + } + continue + } + + // Parse event fields + if strings.HasPrefix(line, "data: ") { + if event.Data != "" { + event.Data += "\n" + } + event.Data += strings.TrimPrefix(line, "data: ") + } else if strings.HasPrefix(line, "event: ") { + event.Event = strings.TrimPrefix(line, "event: ") + } else if strings.HasPrefix(line, "id: ") { + event.ID = strings.TrimPrefix(line, "id: ") + } else if strings.HasPrefix(line, "retry: ") { + // Parse retry value (implementation omitted for brevity) + } else if strings.HasPrefix(line, ": ") { + // Comment line, ignore + continue + } + } + + if err := scanner.Err(); err != nil { + errorChan <- err + } +}