diff --git a/k8-operator/internal/api/api.go b/k8-operator/internal/api/api.go index 546b309a3..2e57d5930 100644 --- a/k8-operator/internal/api/api.go +++ b/k8-operator/internal/api/api.go @@ -1,7 +1,9 @@ package api import ( + "encoding/json" "fmt" + "net/http" "github.com/Infisical/infisical/k8-operator/internal/model" "github.com/go-resty/resty/v2" @@ -167,3 +169,65 @@ func CallGetProjectByIDv2(httpClient *resty.Client, request GetProjectByIDReques 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/controller/infisicalsecret_controller.go b/k8-operator/internal/controller/infisicalsecret_controller.go index 7a6b3497e..eb00cac7c 100644 --- a/k8-operator/internal/controller/infisicalsecret_controller.go +++ b/k8-operator/internal/controller/infisicalsecret_controller.go @@ -199,14 +199,16 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ } if infisicalSecretCRD.Spec.InstantUpdates { - logger.Info("Instant updates are enabled") - if err := handler.OpenInstantUpdatesStream(ctx, logger, &infisicalSecretCRD, infisicalSecretResourceVariablesMap, r.SourceCh); err != nil { - logger.Error(err, fmt.Sprintf("unable to ensure event stream. Will requeue after [requeueTime=%v]", requeueTime)) + requeueTime = time.Second * 10 + logger.Info(err.Error()) + logger.Info(fmt.Sprintf("unable to ensure event stream. Will requeue after [requeueTime=%v]", requeueTime)) return ctrl.Result{ RequeueAfter: requeueTime, }, nil } + + logger.Info("Instant updates are enabled") } else { handler.CloseInstantUpdatesStream(ctx, logger, &infisicalSecretCRD, infisicalSecretResourceVariablesMap) } @@ -223,7 +225,7 @@ func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). WatchesRawSource( - source.Channel[client.Object](r.SourceCh, &util.EnqueueDelayedEventHandler{Delay: time.Second * 3}), + 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 { diff --git a/k8-operator/internal/services/infisicalsecret/reconciler.go b/k8-operator/internal/services/infisicalsecret/reconciler.go index 6382f9dc7..3bd9de34a 100644 --- a/k8-operator/internal/services/infisicalsecret/reconciler.go +++ b/k8-operator/internal/services/infisicalsecret/reconciler.go @@ -3,7 +3,6 @@ package infisicalsecret import ( "bytes" "context" - "encoding/json" "errors" "fmt" "net/http" @@ -19,6 +18,7 @@ import ( "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" @@ -581,59 +581,24 @@ func (r *InfisicalSecretReconciler) OpenInstantUpdatesStream(ctx context.Context secretsPath = fmt.Sprint(secretsPath, "**") } - conditions := &api.SubscribeProjectEventsRequestCondition{ - SecretPath: secretsPath, - EnvironmentSlug: envSlug, - } - - body, err := json.Marshal(api.SubscribeProjectEventsRequest{ - ProjectID: project.ID, - Register: []api.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 fmt.Errorf("CallSubscribeProjectEvents: unable to marshal body [err=%s]", err) } - events, errors, err := sseRegistry.Subscribe(func() (*http.Request, error) { - headers := map[string]string{ - "User-Agent": api.USER_AGENT_NAME, - "Authorization": fmt.Sprint("Bearer ", token), - "Content-Type": "application/json", - } + events, errors, err := sseRegistry.Subscribe(func() (*http.Response, error) { + httpClient := resty.New() - req, err := http.NewRequest("POST", fmt.Sprintf("%s/v1/events/subscribe/project-events", api.API_HOST_URL), strings.NewReader(string(body))) + req, err := api.CallSubscribeProjectEvents(httpClient, project.ID, secretsPath, envSlug, token) if err != nil { return nil, err } - for k, v := range headers { - req.Header.Set(k, v) - } - return req, nil }) if err != nil { - return fmt.Errorf("unable to connect to SSE server [err=%s]", err) + return fmt.Errorf("unable to connect sse [err=%s]", err) } go func() { diff --git a/k8-operator/internal/util/sse/client.go b/k8-operator/internal/util/sse/client.go deleted file mode 100644 index c2bc6900f..000000000 --- a/k8-operator/internal/util/sse/client.go +++ /dev/null @@ -1,187 +0,0 @@ -package sse - -import ( - "bufio" - "context" - "fmt" - "io" - "net/http" - "strings" - "time" -) - -// Event represents a Server-Sent Event -type Event struct { - ID string - Event string - Data string - Retry int -} - -// Client handles SSE connections with high performance -type Client struct { - httpClient *http.Client - onPing func() // Callback for ping events -} - -// NewClient creates a new high-performance SSE client -func NewClient() Client { - return Client{ - httpClient: &http.Client{ - Timeout: 0, // No timeout for streaming - Transport: &http.Transport{ - MaxIdleConns: 100, - IdleConnTimeout: 90 * time.Second, - DisableCompression: true, // SSE typically doesn't benefit from compression - }, - }, - } -} - -// WithPingHandler sets a callback for ping events -func (c *Client) WithPingHandler(handler func()) *Client { - c.onPing = handler - return c -} - -// Connect establishes an SSE connection and returns event channels -func (c *Client) Connect(ctx context.Context, req *http.Request) (<-chan Event, <-chan error, error) { - // Configure SSE headers - req.Header.Set("Cache-Control", "no-cache") - req.Header.Set("Accept", "text/event-stream") - req.Header.Set("Connection", "keep-alive") - - // Add context to request - req = req.WithContext(ctx) - - resp, err := c.httpClient.Do(req) - if err != nil { - return nil, nil, fmt.Errorf("request failed: %w", err) - } - - if resp.StatusCode != http.StatusOK { - resp.Body.Close() - return nil, nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) - } - - eventChan := make(chan Event, 10) - errorChan := make(chan error, 1) - - go c.stream(ctx, resp.Body, eventChan, errorChan) - - return eventChan, errorChan, nil -} - -func (c *Client) stream(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 c.isPingEvent(currentEvent) { - if c.onPing != nil { - c.onPing() - } - } else { - // Send non-ping events - select { - case eventChan <- currentEvent: - case <-ctx.Done(): - return - } - } - - // Reset for next event - currentEvent = Event{} - } - continue - } - - // Parse line efficiently - c.parseLine(line, ¤tEvent, &dataBuilder) - } - - if err := scanner.Err(); err != nil { - select { - case errorChan <- err: - case <-ctx.Done(): - } - } -} - -// parseLine efficiently parses SSE protocol lines -func (c *Client) 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 (c *Client) 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 -} - -// WithHTTPClient sets a custom HTTP client -func (c *Client) WithHTTPClient(client *http.Client) *Client { - c.httpClient = client - return c -} diff --git a/k8-operator/internal/util/sse/sse.go b/k8-operator/internal/util/sse/sse.go index 90918e8ea..7bfbcef88 100644 --- a/k8-operator/internal/util/sse/sse.go +++ b/k8-operator/internal/util/sse/sse.go @@ -1,14 +1,24 @@ package sse import ( + "bufio" "context" - "fmt" + "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 @@ -39,13 +49,13 @@ func (c *ConnectionMeta) Cancel() { // ConnectionRegistry manages SSE connections with high performance type ConnectionRegistry struct { - client Client - 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 @@ -54,17 +64,16 @@ func NewConnectionRegistry(ctx context.Context) *ConnectionRegistry { monitorStop: make(chan struct{}), } - // Configure client with ping handler - r.client = NewClient() - r.client.WithPingHandler(func() { + // Configure ping handler + r.onPing = func() { r.UpdateLastPing() - }) + } return r } // Subscribe provides SSE events, creating a connection if needed -func (r *ConnectionRegistry) Subscribe(buildRequest func() (*http.Request, error)) (<-chan Event, <-chan error, error) { +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 @@ -79,12 +88,12 @@ func (r *ConnectionRegistry) Subscribe(buildRequest func() (*http.Request, error return r.conn.EventChan, r.conn.ErrorChan, nil } - req, err := buildRequest() + res, err := request() if err != nil { - return nil, nil, fmt.Errorf("failed to build request: %w", err) + return nil, nil, err } - conn, err := r.createConnection(req) + conn, err := r.createStream(res) if err != nil { return nil, nil, err } @@ -144,14 +153,13 @@ func (r *ConnectionRegistry) getConnection() *ConnectionMeta { return conn } -// createConnection creates a new SSE connection -func (r *ConnectionRegistry) createConnection(req *http.Request) (*ConnectionMeta, error) { +func (r *ConnectionRegistry) createStream(res *http.Response) (*ConnectionMeta, error) { ctx, cancel := context.WithCancel(context.Background()) - eventChan, errorChan, err := r.client.Connect(ctx, req) + eventChan, errorChan, err := r.stream(ctx, res) if err != nil { cancel() - return nil, fmt.Errorf("failed to connect: %w", err) + return nil, err } meta := &ConnectionMeta{ @@ -164,6 +172,125 @@ func (r *ConnectionRegistry) createConnection(req *http.Request) (*ConnectionMet 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 (