From eb6c22af36aa199899f79f930221a6bc7eb9e3c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roberto=20Jim=C3=A9nez=20S=C3=A1nchez?= Date: Fri, 9 Jan 2026 09:08:49 +0100 Subject: [PATCH] Provisioning: Add connection operator with health check updates (#116028) * Add connection operator with health check updates - Add ConnectionController to watch and reconcile Connection resources - Add ConnectionStatusPatcher for updating connection status - Add connection_operator.go entry point for standalone operator - Register connection operator in pkg/operators/register.go - Add connection controller to in-process setup in register.go - Add unit tests for connection controller - Add integration tests for health check updates * Fix integration test: get latest version before update to avoid conflicts * refactor: move repoFactory to operator-specific configs - Remove repoFactory from shared provisioningControllerConfig - Add repoFactory to repoControllerConfig and jobsControllerConfig - This allows connection operator to run without repository setup * Remove unneccesary comments --- .../pkg/controller/connection_status.go | 40 +++ pkg/operators/provisioning/config.go | 12 - .../provisioning/connection_operator.go | 86 ++++++ pkg/operators/provisioning/repo_operator.go | 13 + pkg/operators/register.go | 6 + .../provisioning/controller/connection.go | 254 ++++++++++++++++ .../controller/connection_test.go | 287 ++++++++++++++++++ pkg/registry/apis/provisioning/register.go | 14 + .../apis/provisioning/connection_test.go | 148 +++++++++ 9 files changed, 848 insertions(+), 12 deletions(-) create mode 100644 apps/provisioning/pkg/controller/connection_status.go create mode 100644 pkg/operators/provisioning/connection_operator.go create mode 100644 pkg/registry/apis/provisioning/controller/connection.go create mode 100644 pkg/registry/apis/provisioning/controller/connection_test.go diff --git a/apps/provisioning/pkg/controller/connection_status.go b/apps/provisioning/pkg/controller/connection_status.go new file mode 100644 index 00000000000..0d8a0002f41 --- /dev/null +++ b/apps/provisioning/pkg/controller/connection_status.go @@ -0,0 +1,40 @@ +package controller + +import ( + "context" + "encoding/json" + "fmt" + + provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" + client "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" +) + +// ConnectionStatusPatcher provides methods to patch Connection status subresources. +type ConnectionStatusPatcher struct { + client client.ProvisioningV0alpha1Interface +} + +// NewConnectionStatusPatcher creates a new ConnectionStatusPatcher. +func NewConnectionStatusPatcher(client client.ProvisioningV0alpha1Interface) *ConnectionStatusPatcher { + return &ConnectionStatusPatcher{ + client: client, + } +} + +// Patch applies JSON patch operations to a Connection's status subresource. +func (p *ConnectionStatusPatcher) Patch(ctx context.Context, conn *provisioning.Connection, patchOperations ...map[string]interface{}) error { + patch, err := json.Marshal(patchOperations) + if err != nil { + return fmt.Errorf("unable to marshal patch data: %w", err) + } + + _, err = p.client.Connections(conn.Namespace). + Patch(ctx, conn.Name, types.JSONPatchType, patch, metav1.PatchOptions{}, "status") + if err != nil { + return fmt.Errorf("unable to update connection status: %w", err) + } + + return nil +} diff --git a/pkg/operators/provisioning/config.go b/pkg/operators/provisioning/config.go index 05552e56095..868d2a8d717 100644 --- a/pkg/operators/provisioning/config.go +++ b/pkg/operators/provisioning/config.go @@ -36,7 +36,6 @@ import ( type provisioningControllerConfig struct { provisioningClient *client.Clientset resyncInterval time.Duration - repoFactory repository.Factory unified resources.ResourceStore clients resources.ClientFactory tokenExchangeClient *authn.TokenExchangeClient @@ -129,16 +128,6 @@ func setupFromConfig(cfg *setting.Cfg, registry prometheus.Registerer) (controll return nil, fmt.Errorf("failed to create provisioning client: %w", err) } - decrypter, err := setupDecrypter(cfg, tracer, tokenExchangeClient) - if err != nil { - return nil, fmt.Errorf("failed to setup decrypter: %w", err) - } - - repoFactory, err := setupRepoFactory(cfg, decrypter, provisioningClient, registry) - if err != nil { - return nil, fmt.Errorf("failed to setup repository getter: %w", err) - } - // HACK: This logic directly connects to unified storage. We are doing this for now as there is no global // search endpoint. But controllers, in general, should not connect directly to unified storage and instead // go through the api server. Once there is a global search endpoint, we will switch to that here as well. @@ -195,7 +184,6 @@ func setupFromConfig(cfg *setting.Cfg, registry prometheus.Registerer) (controll return &provisioningControllerConfig{ provisioningClient: provisioningClient, - repoFactory: repoFactory, unified: unified, clients: clients, resyncInterval: operatorSec.Key("resync_interval").MustDuration(60 * time.Second), diff --git a/pkg/operators/provisioning/connection_operator.go b/pkg/operators/provisioning/connection_operator.go new file mode 100644 index 00000000000..34624f4fe47 --- /dev/null +++ b/pkg/operators/provisioning/connection_operator.go @@ -0,0 +1,86 @@ +package provisioning + +import ( + "context" + "fmt" + "log/slog" + "os" + "os/signal" + "syscall" + + "github.com/grafana/grafana-app-sdk/logging" + "github.com/prometheus/client_golang/prometheus" + "k8s.io/client-go/tools/cache" + + appcontroller "github.com/grafana/grafana/apps/provisioning/pkg/controller" + informer "github.com/grafana/grafana/apps/provisioning/pkg/generated/informers/externalversions" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/controller" + "github.com/grafana/grafana/pkg/server" + "github.com/grafana/grafana/pkg/setting" +) + +// RunConnectionController starts the connection controller operator. +func RunConnectionController(deps server.OperatorDependencies) error { + logger := logging.NewSLogLogger(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{ + Level: slog.LevelDebug, + })).With("logger", "provisioning-connection-controller") + logger.Info("Starting provisioning connection controller") + + controllerCfg, err := getConnectionControllerConfig(deps.Config, deps.Registerer) + if err != nil { + return fmt.Errorf("failed to setup operator: %w", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) + go func() { + <-sigChan + fmt.Println("Received shutdown signal, stopping controllers") + cancel() + }() + + informerFactory := informer.NewSharedInformerFactoryWithOptions( + controllerCfg.provisioningClient, + controllerCfg.resyncInterval, + ) + + statusPatcher := appcontroller.NewConnectionStatusPatcher(controllerCfg.provisioningClient.ProvisioningV0alpha1()) + connInformer := informerFactory.Provisioning().V0alpha1().Connections() + + connController, err := controller.NewConnectionController( + controllerCfg.provisioningClient.ProvisioningV0alpha1(), + connInformer, + statusPatcher, + ) + if err != nil { + return fmt.Errorf("failed to create connection controller: %w", err) + } + + informerFactory.Start(ctx.Done()) + if !cache.WaitForCacheSync(ctx.Done(), connInformer.Informer().HasSynced) { + return fmt.Errorf("failed to sync informer cache") + } + + connController.Run(ctx, controllerCfg.workerCount) + return nil +} + +type connectionControllerConfig struct { + provisioningControllerConfig + workerCount int +} + +func getConnectionControllerConfig(cfg *setting.Cfg, registry prometheus.Registerer) (*connectionControllerConfig, error) { + controllerCfg, err := setupFromConfig(cfg, registry) + if err != nil { + return nil, err + } + + return &connectionControllerConfig{ + provisioningControllerConfig: *controllerCfg, + workerCount: cfg.SectionWithEnvOverrides("operator").Key("worker_count").MustInt(1), + }, nil +} diff --git a/pkg/operators/provisioning/repo_operator.go b/pkg/operators/provisioning/repo_operator.go index 416eb5c0e3f..c3a038b9378 100644 --- a/pkg/operators/provisioning/repo_operator.go +++ b/pkg/operators/provisioning/repo_operator.go @@ -106,6 +106,7 @@ func RunRepoController(deps server.OperatorDependencies) error { type repoControllerConfig struct { provisioningControllerConfig + repoFactory repository.Factory workerCount int parallelOperations int allowedTargets []string @@ -119,6 +120,17 @@ func getRepoControllerConfig(cfg *setting.Cfg, registry prometheus.Registerer) ( return nil, err } + // Setup repository factory for repo controller + decrypter, err := setupDecrypter(cfg, tracing.NewNoopTracerService(), controllerCfg.tokenExchangeClient) + if err != nil { + return nil, fmt.Errorf("failed to setup decrypter: %w", err) + } + + repoFactory, err := setupRepoFactory(cfg, decrypter, controllerCfg.provisioningClient, registry) + if err != nil { + return nil, fmt.Errorf("failed to setup repository factory: %w", err) + } + allowedTargets := []string{} cfg.SectionWithEnvOverrides("provisioning").Key("allowed_targets").Strings("|") if len(allowedTargets) == 0 { @@ -127,6 +139,7 @@ func getRepoControllerConfig(cfg *setting.Cfg, registry prometheus.Registerer) ( return &repoControllerConfig{ provisioningControllerConfig: *controllerCfg, + repoFactory: repoFactory, allowedTargets: allowedTargets, workerCount: cfg.SectionWithEnvOverrides("operator").Key("worker_count").MustInt(1), parallelOperations: cfg.SectionWithEnvOverrides("operator").Key("parallel_operations").MustInt(10), diff --git a/pkg/operators/register.go b/pkg/operators/register.go index b31b9837fb0..4d42591ca7b 100644 --- a/pkg/operators/register.go +++ b/pkg/operators/register.go @@ -13,6 +13,12 @@ func init() { RunFunc: provisioning.RunRepoController, }) + server.RegisterOperator(server.Operator{ + Name: "provisioning-connection", + Description: "Watch provisioning connections", + RunFunc: provisioning.RunConnectionController, + }) + server.RegisterOperator(server.Operator{ Name: "iam-folder-reconciler", Description: "Reconcile folder resources into Zanzana", diff --git a/pkg/registry/apis/provisioning/controller/connection.go b/pkg/registry/apis/provisioning/controller/connection.go new file mode 100644 index 00000000000..be90908bd49 --- /dev/null +++ b/pkg/registry/apis/provisioning/controller/connection.go @@ -0,0 +1,254 @@ +package controller + +import ( + "context" + "errors" + "fmt" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" + + "github.com/grafana/grafana-app-sdk/logging" + provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" + client "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1" + informer "github.com/grafana/grafana/apps/provisioning/pkg/generated/informers/externalversions/provisioning/v0alpha1" + listers "github.com/grafana/grafana/apps/provisioning/pkg/generated/listers/provisioning/v0alpha1" +) + +const connectionLoggerName = "provisioning-connection-controller" + +const ( + connectionMaxAttempts = 3 + // connectionHealthyDuration defines how recent a health check must be to be considered "recent" when healthy + connectionHealthyDuration = 5 * time.Minute + // connectionUnhealthyDuration defines how recent a health check must be to be considered "recent" when unhealthy + connectionUnhealthyDuration = 1 * time.Minute +) + +type connectionQueueItem struct { + key string + attempts int +} + +// ConnectionStatusPatcher defines the interface for updating connection status. +// +//go:generate mockery --name=ConnectionStatusPatcher +type ConnectionStatusPatcher interface { + Patch(ctx context.Context, conn *provisioning.Connection, patchOperations ...map[string]interface{}) error +} + +// ConnectionController controls Connection resources. +type ConnectionController struct { + client client.ProvisioningV0alpha1Interface + connLister listers.ConnectionLister + connSynced cache.InformerSynced + logger logging.Logger + + statusPatcher ConnectionStatusPatcher + + queue workqueue.TypedRateLimitingInterface[*connectionQueueItem] +} + +// NewConnectionController creates a new ConnectionController. +func NewConnectionController( + provisioningClient client.ProvisioningV0alpha1Interface, + connInformer informer.ConnectionInformer, + statusPatcher ConnectionStatusPatcher, +) (*ConnectionController, error) { + cc := &ConnectionController{ + client: provisioningClient, + connLister: connInformer.Lister(), + connSynced: connInformer.Informer().HasSynced, + queue: workqueue.NewTypedRateLimitingQueueWithConfig( + workqueue.DefaultTypedControllerRateLimiter[*connectionQueueItem](), + workqueue.TypedRateLimitingQueueConfig[*connectionQueueItem]{ + Name: "provisioningConnectionController", + }, + ), + statusPatcher: statusPatcher, + logger: logging.DefaultLogger.With("logger", connectionLoggerName), + } + + _, err := connInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: cc.enqueue, + UpdateFunc: func(oldObj, newObj interface{}) { + cc.enqueue(newObj) + }, + }) + if err != nil { + return nil, err + } + + return cc, nil +} + +func (cc *ConnectionController) enqueue(obj interface{}) { + key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj) + if err != nil { + cc.logger.Error("failed to get key for object", "error", err) + return + } + cc.queue.Add(&connectionQueueItem{key: key}) +} + +// Run starts the ConnectionController. +func (cc *ConnectionController) Run(ctx context.Context, workerCount int) { + defer utilruntime.HandleCrash() + defer cc.queue.ShutDown() + + cc.logger.Info("starting connection controller", "workers", workerCount) + + for i := 0; i < workerCount; i++ { + go wait.UntilWithContext(ctx, cc.runWorker, time.Second) + } + + <-ctx.Done() + cc.logger.Info("shutting down connection controller") +} + +func (cc *ConnectionController) runWorker(ctx context.Context) { + for cc.processNextWorkItem(ctx) { + } +} + +func (cc *ConnectionController) processNextWorkItem(ctx context.Context) bool { + item, quit := cc.queue.Get() + if quit { + return false + } + defer cc.queue.Done(item) + + logger := logging.FromContext(ctx).With("work_key", item.key) + logger.Info("ConnectionController processing key") + + err := cc.process(ctx, item) + if err == nil { + cc.queue.Forget(item) + return true + } + + item.attempts++ + logger = logger.With("error", err, "attempts", item.attempts) + logger.Error("ConnectionController failed to process key") + + if item.attempts >= connectionMaxAttempts { + logger.Error("ConnectionController failed too many times") + cc.queue.Forget(item) + return true + } + + if !apierrors.IsServiceUnavailable(err) { + logger.Info("ConnectionController will not retry") + cc.queue.Forget(item) + return true + } + + logger.Info("ConnectionController will retry as service is unavailable") + utilruntime.HandleError(fmt.Errorf("%v failed with: %v", item, err)) + cc.queue.AddRateLimited(item) + + return true +} + +func (cc *ConnectionController) process(ctx context.Context, item *connectionQueueItem) error { + logger := cc.logger.With("key", item.key) + ctx = logging.Context(ctx, logger) + + namespace, name, err := cache.SplitMetaNamespaceKey(item.key) + if err != nil { + return err + } + + conn, err := cc.connLister.Connections(namespace).Get(name) + switch { + case apierrors.IsNotFound(err): + return errors.New("connection not found in cache") + case err != nil: + return err + } + + // Skip if being deleted + if conn.DeletionTimestamp != nil { + logger.Info("connection is being deleted, skipping") + return nil + } + + hasSpecChanged := conn.Generation != conn.Status.ObservedGeneration + shouldCheckHealth := cc.shouldCheckHealth(conn) + + // Determine the main triggering condition + switch { + case hasSpecChanged: + logger.Info("spec changed, reconciling", "generation", conn.Generation, "observedGeneration", conn.Status.ObservedGeneration) + case shouldCheckHealth: + logger.Info("health is stale, refreshing", "lastChecked", conn.Status.Health.Checked, "healthy", conn.Status.Health.Healthy) + default: + logger.Debug("skipping as conditions are not met", "generation", conn.Generation, "observedGeneration", conn.Status.ObservedGeneration) + return nil + } + + // For now, just update the state to connected, health to healthy, and observed generation + // Future: Add credential validation logic here + patchOperations := []map[string]interface{}{} + + // Only update observedGeneration when spec changes + if hasSpecChanged { + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "replace", + "path": "/status/observedGeneration", + "value": conn.Generation, + }) + } + + // Always update state and health + patchOperations = append(patchOperations, + map[string]interface{}{ + "op": "replace", + "path": "/status/state", + "value": provisioning.ConnectionStateConnected, + }, + map[string]interface{}{ + "op": "replace", + "path": "/status/health", + "value": provisioning.HealthStatus{ + Healthy: true, + Checked: time.Now().UnixMilli(), + }, + }, + ) + + if err := cc.statusPatcher.Patch(ctx, conn, patchOperations...); err != nil { + return fmt.Errorf("failed to update connection status: %w", err) + } + + logger.Info("connection reconciled successfully") + return nil +} + +// shouldCheckHealth determines if a connection health check should be performed. +func (cc *ConnectionController) shouldCheckHealth(conn *provisioning.Connection) bool { + // If the connection has been updated, always check health + if conn.Generation != conn.Status.ObservedGeneration { + return true + } + + // Check if health check is stale + return !cc.hasRecentHealthCheck(conn.Status.Health) +} + +// hasRecentHealthCheck checks if a health check was performed recently. +func (cc *ConnectionController) hasRecentHealthCheck(healthStatus provisioning.HealthStatus) bool { + if healthStatus.Checked == 0 { + return false // Never checked + } + + age := time.Since(time.UnixMilli(healthStatus.Checked)) + if healthStatus.Healthy { + return age <= connectionHealthyDuration + } + return age <= connectionUnhealthyDuration +} diff --git a/pkg/registry/apis/provisioning/controller/connection_test.go b/pkg/registry/apis/provisioning/controller/connection_test.go new file mode 100644 index 00000000000..b033ddb39a9 --- /dev/null +++ b/pkg/registry/apis/provisioning/controller/connection_test.go @@ -0,0 +1,287 @@ +package controller + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" +) + +func TestConnectionController_shouldCheckHealth(t *testing.T) { + testCases := []struct { + name string + conn *provisioning.Connection + expected bool + }{ + { + name: "should check health when generation differs from observed", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Generation: 2, + }, + Status: provisioning.ConnectionStatus{ + ObservedGeneration: 1, + }, + }, + expected: true, + }, + { + name: "should check health when never checked before", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Generation: 1, + }, + Status: provisioning.ConnectionStatus{ + ObservedGeneration: 1, + Health: provisioning.HealthStatus{ + Checked: 0, + }, + }, + }, + expected: true, + }, + { + name: "should check health when healthy check is stale (>5 min)", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Generation: 1, + }, + Status: provisioning.ConnectionStatus{ + ObservedGeneration: 1, + Health: provisioning.HealthStatus{ + Healthy: true, + Checked: time.Now().Add(-6 * time.Minute).UnixMilli(), + }, + }, + }, + expected: true, + }, + { + name: "should check health when unhealthy check is stale (>1 min)", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Generation: 1, + }, + Status: provisioning.ConnectionStatus{ + ObservedGeneration: 1, + Health: provisioning.HealthStatus{ + Healthy: false, + Checked: time.Now().Add(-2 * time.Minute).UnixMilli(), + }, + }, + }, + expected: true, + }, + { + name: "should not check health when healthy check is recent (<5 min)", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Generation: 1, + }, + Status: provisioning.ConnectionStatus{ + ObservedGeneration: 1, + Health: provisioning.HealthStatus{ + Healthy: true, + Checked: time.Now().Add(-2 * time.Minute).UnixMilli(), + }, + }, + }, + expected: false, + }, + { + name: "should not check health when unhealthy check is recent (<1 min)", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Generation: 1, + }, + Status: provisioning.ConnectionStatus{ + ObservedGeneration: 1, + Health: provisioning.HealthStatus{ + Healthy: false, + Checked: time.Now().Add(-30 * time.Second).UnixMilli(), + }, + }, + }, + expected: false, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + cc := &ConnectionController{} + result := cc.shouldCheckHealth(tc.conn) + assert.Equal(t, tc.expected, result) + }) + } +} + +func TestConnectionController_hasRecentHealthCheck(t *testing.T) { + testCases := []struct { + name string + healthStatus provisioning.HealthStatus + expected bool + }{ + { + name: "never checked", + healthStatus: provisioning.HealthStatus{ + Checked: 0, + }, + expected: false, + }, + { + name: "healthy and recent", + healthStatus: provisioning.HealthStatus{ + Healthy: true, + Checked: time.Now().Add(-2 * time.Minute).UnixMilli(), + }, + expected: true, + }, + { + name: "healthy and stale", + healthStatus: provisioning.HealthStatus{ + Healthy: true, + Checked: time.Now().Add(-10 * time.Minute).UnixMilli(), + }, + expected: false, + }, + { + name: "unhealthy and recent", + healthStatus: provisioning.HealthStatus{ + Healthy: false, + Checked: time.Now().Add(-30 * time.Second).UnixMilli(), + }, + expected: true, + }, + { + name: "unhealthy and stale", + healthStatus: provisioning.HealthStatus{ + Healthy: false, + Checked: time.Now().Add(-2 * time.Minute).UnixMilli(), + }, + expected: false, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + cc := &ConnectionController{} + result := cc.hasRecentHealthCheck(tc.healthStatus) + assert.Equal(t, tc.expected, result) + }) + } +} + +func TestConnectionController_reconcileConditions(t *testing.T) { + testCases := []struct { + name string + conn *provisioning.Connection + expectReconcile bool + expectSpecChanged bool + description string + }{ + { + name: "skip when being deleted", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-conn", + Namespace: "default", + DeletionTimestamp: &metav1.Time{Time: time.Now()}, + }, + }, + expectReconcile: false, + expectSpecChanged: false, + description: "deleted connections should be skipped", + }, + { + name: "skip when no changes needed", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-conn", + Namespace: "default", + Generation: 1, + }, + Status: provisioning.ConnectionStatus{ + ObservedGeneration: 1, + Health: provisioning.HealthStatus{ + Healthy: true, + Checked: time.Now().UnixMilli(), + }, + }, + }, + expectReconcile: false, + expectSpecChanged: false, + description: "no reconcile when generation matches and health is recent", + }, + { + name: "reconcile when spec changed", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-conn", + Namespace: "default", + Generation: 2, + }, + Status: provisioning.ConnectionStatus{ + ObservedGeneration: 1, + Health: provisioning.HealthStatus{ + Healthy: true, + Checked: time.Now().UnixMilli(), + }, + }, + }, + expectReconcile: true, + expectSpecChanged: true, + description: "reconcile when generation differs", + }, + { + name: "reconcile when health is stale", + conn: &provisioning.Connection{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-conn", + Namespace: "default", + Generation: 1, + }, + Status: provisioning.ConnectionStatus{ + ObservedGeneration: 1, + Health: provisioning.HealthStatus{ + Healthy: true, + Checked: time.Now().Add(-10 * time.Minute).UnixMilli(), + }, + }, + }, + expectReconcile: true, + expectSpecChanged: false, + description: "reconcile when health check is stale", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + cc := &ConnectionController{} + + // Test the core reconciliation conditions + if tc.conn.DeletionTimestamp != nil { + assert.False(t, tc.expectReconcile, tc.description) + return + } + + hasSpecChanged := tc.conn.Generation != tc.conn.Status.ObservedGeneration + shouldCheckHealth := cc.shouldCheckHealth(tc.conn) + + needsReconcile := hasSpecChanged || shouldCheckHealth + + assert.Equal(t, tc.expectReconcile, needsReconcile, tc.description) + assert.Equal(t, tc.expectSpecChanged, hasSpecChanged, "spec changed check") + }) + } +} + +func TestConnectionController_processNextWorkItem(t *testing.T) { + t.Run("returns false when queue is shut down", func(t *testing.T) { + cc := &ConnectionController{} + // This test verifies the structure is correct + assert.NotNil(t, cc) + }) +} diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 217a1933d15..026797eb474 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -817,8 +817,10 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH sharedInformerFactory := informers.NewSharedInformerFactory(c, 60*time.Second) repoInformer := sharedInformerFactory.Provisioning().V0alpha1().Repositories() jobInformer := sharedInformerFactory.Provisioning().V0alpha1().Jobs() + connInformer := sharedInformerFactory.Provisioning().V0alpha1().Connections() go repoInformer.Informer().Run(postStartHookCtx.Done()) go jobInformer.Informer().Run(postStartHookCtx.Done()) + go connInformer.Informer().Run(postStartHookCtx.Done()) // Create the repository resources factory repositoryListerWrapper := func(ctx context.Context) ([]provisioning.Repository, error) { @@ -939,6 +941,18 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH go repoController.Run(postStartHookCtx.Context, repoControllerWorkers) + // Create and run connection controller + connStatusPatcher := appcontroller.NewConnectionStatusPatcher(b.GetClient()) + connController, err := controller.NewConnectionController( + b.GetClient(), + connInformer, + connStatusPatcher, + ) + if err != nil { + return err + } + go connController.Run(postStartHookCtx.Context, repoControllerWorkers) + // If Loki not used, initialize the API client-based history writer and start the controller for history jobs if b.jobHistoryLoki == nil { // Create HistoryJobController for cleanup of old job history entries diff --git a/pkg/tests/apis/provisioning/connection_test.go b/pkg/tests/apis/provisioning/connection_test.go index ea28ac88359..02c5436badb 100644 --- a/pkg/tests/apis/provisioning/connection_test.go +++ b/pkg/tests/apis/provisioning/connection_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "testing" + "time" "github.com/grafana/grafana/pkg/util/testutil" "github.com/stretchr/testify/assert" @@ -11,6 +12,9 @@ import ( k8serrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + + provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" + clientset "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned" ) func TestIntegrationProvisioning_ConnectionCRUDL(t *testing.T) { @@ -411,3 +415,147 @@ func TestIntegrationProvisioning_ConnectionValidation(t *testing.T) { assert.Contains(t, err.Error(), "privateKey is forbidden in Gitlab connection") }) } + +func TestIntegrationConnectionController_HealthCheckUpdates(t *testing.T) { + testutil.SkipIntegrationTestInShortMode(t) + + helper := runGrafana(t) + ctx := context.Background() + namespace := "default" + + // Create typed client from REST config + restConfig := helper.Org1.Admin.NewRestConfig() + provisioningClient, err := clientset.NewForConfig(restConfig) + require.NoError(t, err) + connClient := provisioningClient.ProvisioningV0alpha1().Connections(namespace) + + t.Run("health check gets updated after initial creation", func(t *testing.T) { + // Create a connection using unstructured (like other connection tests) + connUnstructured := &unstructured.Unstructured{Object: map[string]any{ + "apiVersion": "provisioning.grafana.app/v0alpha1", + "kind": "Connection", + "metadata": map[string]any{ + "name": "test-connection-health", + "namespace": namespace, + }, + "spec": map[string]any{ + "type": "github", + "github": map[string]any{ + "appID": "12345", + "installationID": "67890", + }, + }, + "secure": map[string]any{ + "privateKey": map[string]any{ + "create": "test-private-key", + }, + }, + }} + + createdUnstructured, err := helper.Connections.Resource.Create(ctx, connUnstructured, metav1.CreateOptions{}) + require.NoError(t, err) + require.NotNil(t, createdUnstructured) + + connName := createdUnstructured.GetName() + + t.Cleanup(func() { + _ = helper.Connections.Resource.Delete(ctx, connName, metav1.DeleteOptions{}) + }) + + // Wait for initial reconciliation - controller should update status + require.Eventually(t, func() bool { + updated, err := connClient.Get(ctx, connName, metav1.GetOptions{}) + if err != nil { + return false + } + return updated.Status.ObservedGeneration == updated.Generation && + updated.Status.Health.Checked > 0 && + updated.Status.State == provisioning.ConnectionStateConnected && + updated.Status.Health.Healthy + }, 10*time.Second, 500*time.Millisecond, "connection should be initially reconciled with health status") + + // Verify initial health check was set + initial, err := connClient.Get(ctx, connName, metav1.GetOptions{}) + require.NoError(t, err) + assert.True(t, initial.Status.Health.Healthy, "connection should be healthy") + assert.Equal(t, provisioning.ConnectionStateConnected, initial.Status.State, "connection should be connected") + assert.Greater(t, initial.Status.Health.Checked, int64(0), "health check timestamp should be set") + assert.Equal(t, initial.Generation, initial.Status.ObservedGeneration, "observed generation should match") + }) + + t.Run("health check updates when spec changes", func(t *testing.T) { + // Create a connection using unstructured + connUnstructured := &unstructured.Unstructured{Object: map[string]any{ + "apiVersion": "provisioning.grafana.app/v0alpha1", + "kind": "Connection", + "metadata": map[string]any{ + "name": "test-connection-spec-change", + "namespace": namespace, + }, + "spec": map[string]any{ + "type": "github", + "github": map[string]any{ + "appID": "11111", + "installationID": "22222", + }, + }, + "secure": map[string]any{ + "privateKey": map[string]any{ + "create": "test-private-key-2", + }, + }, + }} + + createdUnstructured, err := helper.Connections.Resource.Create(ctx, connUnstructured, metav1.CreateOptions{}) + require.NoError(t, err) + require.NotNil(t, createdUnstructured) + + connName := createdUnstructured.GetName() + + t.Cleanup(func() { + _ = helper.Connections.Resource.Delete(ctx, connName, metav1.DeleteOptions{}) + }) + + // Wait for initial reconciliation + var initialHealthChecked int64 + require.Eventually(t, func() bool { + updated, err := connClient.Get(ctx, connName, metav1.GetOptions{}) + if err != nil { + return false + } + if updated.Status.ObservedGeneration == updated.Generation { + initialHealthChecked = updated.Status.Health.Checked + return true + } + return false + }, 10*time.Second, 500*time.Millisecond, "connection should be initially reconciled") + + // Get the latest version before updating to avoid conflicts with controller updates + latestUnstructured, err := helper.Connections.Resource.Get(ctx, connName, metav1.GetOptions{}) + require.NoError(t, err) + + // Update the connection spec using the latest version + updatedUnstructured := latestUnstructured.DeepCopy() + githubSpec := updatedUnstructured.Object["spec"].(map[string]any)["github"].(map[string]any) + githubSpec["appID"] = "99999" + _, err = helper.Connections.Resource.Update(ctx, updatedUnstructured, metav1.UpdateOptions{}) + require.NoError(t, err) + + // Wait for reconciliation after spec change + require.Eventually(t, func() bool { + reconciled, err := connClient.Get(ctx, connName, metav1.GetOptions{}) + if err != nil { + return false + } + return reconciled.Status.ObservedGeneration == reconciled.Generation && + reconciled.Status.Health.Checked > initialHealthChecked + }, 10*time.Second, 500*time.Millisecond, "connection should be reconciled after spec change") + + // Verify health check was updated + final, err := connClient.Get(ctx, connName, metav1.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, final.Generation, final.Status.ObservedGeneration, "observed generation should match generation") + assert.Greater(t, final.Status.Health.Checked, initialHealthChecked, "health check should be updated after spec change") + assert.True(t, final.Status.Health.Healthy, "connection should remain healthy") + }) +}