wip: k8s operator changes

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

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)
 }

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,
+			InfisicalClient:   client,
-			CancelCtx:       cancel,
+			CancelCtx:         cancel,
-			AuthDetails:     util.AuthenticationDetails{},
+			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
+}

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
+}

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{}

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
+	InfisicalClient   infisicalSdk.InfisicalClientInterface
-	CancelCtx       context.CancelFunc
+	CancelCtx         context.CancelFunc
-	AuthDetails     AuthenticationDetails
+	AuthDetails       AuthenticationDetails
+	EventStreamClient *sse.SSEClient
 }

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
+	}
+}