public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Provisioning: Add connection operator with health check updates (#116028)

Browse Source 
* 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
This commit is contained in:
Roberto Jiménez Sánchez
2026-01-09 09:08:49 +01:00
committed by GitHub
parent 125cc5fddd
commit eb6c22af36
9 changed files with 848 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apis/provisioning/controller/connection.go
+254
View File
@@ -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
}
pkg/registry/apis/provisioning/controller/connection_test.go
+287
View File
@@ -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)
})
}