diff --git a/pkg/registry/apis/provisioning/controller/repository.go b/pkg/registry/apis/provisioning/controller/repository.go index ffeb3f1fb9f..d624879b984 100644 --- a/pkg/registry/apis/provisioning/controller/repository.go +++ b/pkg/registry/apis/provisioning/controller/repository.go @@ -55,7 +55,10 @@ type RepositoryController struct { logger logging.Logger dualwrite dualwrite.Service - jobs jobs.Queue + jobs interface { + jobs.Queue + jobs.Store + } finalizer finalizerProcessor statusPatcher StatusPatcher @@ -79,7 +82,10 @@ func NewRepositoryController( repoFactory repository.Factory, resourceLister resources.ResourceLister, clients resources.ClientFactory, - jobs jobs.Queue, + jobs interface { + jobs.Queue + jobs.Store + }, dualwrite dualwrite.Service, healthChecker *HealthChecker, statusPatcher StatusPatcher, @@ -273,7 +279,7 @@ func (rc *RepositoryController) updateDeleteStatus(ctx context.Context, obj *pro }) } -func (rc *RepositoryController) shouldResync(obj *provisioning.Repository) bool { +func (rc *RepositoryController) shouldResync(ctx context.Context, obj *provisioning.Repository) bool { // don't trigger resync if a sync was never started if obj.Status.Sync.Finished == 0 && obj.Status.Sync.State == "" { return false @@ -283,6 +289,30 @@ func (rc *RepositoryController) shouldResync(obj *provisioning.Repository) bool syncInterval := time.Duration(obj.Spec.Sync.IntervalSeconds) * time.Second tolerance := time.Second + // Check for stale sync status - if sync status indicates a job is running but the job no longer exists + // Only check if Finished is set (meaning a sync has completed before) to avoid interfering with initial syncs + // Only trigger resync if sync is enabled and sync interval has elapsed (to avoid unnecessary operations) + if obj.Status.Sync.Finished > 0 && + obj.Spec.Sync.Enabled && + (obj.Status.Sync.State == provisioning.JobStatePending || obj.Status.Sync.State == provisioning.JobStateWorking) && + obj.Status.Sync.JobID != "" { + _, err := rc.jobs.Get(ctx, obj.Namespace, obj.Status.Sync.JobID) + if apierrors.IsNotFound(err) { + // Job was cleaned up but sync status wasn't updated - trigger resync to reconcile + // Only trigger if sync interval has elapsed to avoid unnecessary operations + if syncAge >= (syncInterval - tolerance) { + logger := logging.FromContext(ctx) + logger.Info("detected stale sync status", "job_id", obj.Status.Sync.JobID) + return true + } + } + // For other errors, log but continue with normal logic + if err != nil { + logger := logging.FromContext(ctx) + logger.Warn("failed to check job existence for stale sync status", "error", err, "job_id", obj.Status.Sync.JobID) + } + } + // HACK: how would this work in a multi-tenant world or under heavy load? // It will start queueing up jobs and we will have to deal with that pendingForTooLong := syncAge >= syncInterval/2 && obj.Status.Sync.State == provisioning.JobStatePending @@ -490,7 +520,7 @@ func (rc *RepositoryController) process(item *queueItem) error { return rc.handleDelete(ctx, obj) } - shouldResync := rc.shouldResync(obj) + shouldResync := rc.shouldResync(ctx, obj) shouldCheckHealth := rc.healthChecker.ShouldCheckHealth(obj) hasSpecChanged := obj.Generation != obj.Status.ObservedGeneration patchOperations := []map[string]interface{}{} diff --git a/pkg/registry/apis/provisioning/controller/repository_test.go b/pkg/registry/apis/provisioning/controller/repository_test.go index e5789672e9d..8eab43d847f 100644 --- a/pkg/registry/apis/provisioning/controller/repository_test.go +++ b/pkg/registry/apis/provisioning/controller/repository_test.go @@ -3,15 +3,20 @@ package controller import ( "context" "testing" + "time" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/controller/mocks" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/mock" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/rest" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/controller/mocks" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" provisioningv0alpha1 "github.com/grafana/grafana/apps/provisioning/pkg/generated/applyconfiguration/provisioning/v0alpha1" client "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1" @@ -338,3 +343,204 @@ func TestShouldUseIncrementalSync(t *testing.T) { assert.False(t, got) }) } + +// mockJobsQueueStore implements both jobs.Queue and jobs.Store for testing +type mockJobsQueueStore struct { + *jobs.MockQueue + *jobs.MockStore +} + +func TestRepositoryController_shouldResync_StaleSyncStatus(t *testing.T) { + testCases := []struct { + name string + repo *provisioning.Repository + jobGetError error + expectedResync bool + description string + }{ + { + name: "stale sync status with Pending state - job not found", + repo: &provisioning.Repository{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-repo", + Namespace: "default", + }, + Spec: provisioning.RepositorySpec{ + Sync: provisioning.SyncOptions{ + Enabled: true, + IntervalSeconds: 300, + }, + }, + Status: provisioning.RepositoryStatus{ + Sync: provisioning.SyncStatus{ + State: provisioning.JobStatePending, + JobID: "test-job-123", + Started: time.Now().Add(-10 * time.Minute).UnixMilli(), + Finished: time.Now().Add(-10 * time.Minute).UnixMilli(), + }, + }, + }, + jobGetError: apierrors.NewNotFound(schema.GroupResource{Resource: "jobs"}, "test-job-123"), + expectedResync: true, + description: "should return true to trigger resync when job is not found", + }, + { + name: "stale sync status with Working state - job not found", + repo: &provisioning.Repository{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-repo", + Namespace: "default", + }, + Spec: provisioning.RepositorySpec{ + Sync: provisioning.SyncOptions{ + Enabled: true, + IntervalSeconds: 300, + }, + }, + Status: provisioning.RepositoryStatus{ + Sync: provisioning.SyncStatus{ + State: provisioning.JobStateWorking, + JobID: "test-job-456", + Started: time.Now().Add(-5 * time.Minute).UnixMilli(), + Finished: time.Now().Add(-5 * time.Minute).UnixMilli(), + }, + }, + }, + jobGetError: apierrors.NewNotFound(schema.GroupResource{Resource: "jobs"}, "test-job-456"), + expectedResync: true, + description: "should return true to trigger resync when working job is not found", + }, + { + name: "non-stale sync status - job exists", + repo: &provisioning.Repository{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-repo", + Namespace: "default", + }, + Spec: provisioning.RepositorySpec{ + Sync: provisioning.SyncOptions{ + Enabled: true, + IntervalSeconds: 300, + }, + }, + Status: provisioning.RepositoryStatus{ + Sync: provisioning.SyncStatus{ + State: provisioning.JobStatePending, + JobID: "test-job-789", + Started: time.Now().Add(-2 * time.Minute).UnixMilli(), + Finished: time.Now().Add(-2 * time.Minute).UnixMilli(), + }, + }, + }, + jobGetError: nil, // Job exists + expectedResync: false, // Should continue with normal logic + description: "should continue with normal logic when job exists", + }, + { + name: "non-stale sync status - no JobID", + repo: &provisioning.Repository{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-repo", + Namespace: "default", + }, + Spec: provisioning.RepositorySpec{ + Sync: provisioning.SyncOptions{ + Enabled: true, + IntervalSeconds: 300, + }, + }, + Status: provisioning.RepositoryStatus{ + Sync: provisioning.SyncStatus{ + State: provisioning.JobStatePending, + JobID: "", + Started: time.Now().Add(-2 * time.Minute).UnixMilli(), + Finished: time.Now().Add(-2 * time.Minute).UnixMilli(), + }, + }, + }, + jobGetError: nil, + expectedResync: false, + description: "should not check when JobID is empty", + }, + { + name: "non-stale sync status - already finished", + repo: &provisioning.Repository{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-repo", + Namespace: "default", + }, + Spec: provisioning.RepositorySpec{ + Sync: provisioning.SyncOptions{ + Enabled: true, + IntervalSeconds: 300, + }, + }, + Status: provisioning.RepositoryStatus{ + Sync: provisioning.SyncStatus{ + State: provisioning.JobStateSuccess, + JobID: "test-job-999", + Finished: time.Now().Add(-1 * time.Minute).UnixMilli(), + }, + }, + }, + jobGetError: nil, + expectedResync: false, + description: "should not check when sync status is already finished", + }, + { + name: "stale sync status - job lookup error (non-NotFound)", + repo: &provisioning.Repository{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-repo", + Namespace: "default", + }, + Spec: provisioning.RepositorySpec{ + Sync: provisioning.SyncOptions{ + Enabled: true, + IntervalSeconds: 300, + }, + }, + Status: provisioning.RepositoryStatus{ + Sync: provisioning.SyncStatus{ + State: provisioning.JobStatePending, + JobID: "test-job-error", + Started: time.Now().Add(-2 * time.Minute).UnixMilli(), + Finished: time.Now().Add(-2 * time.Minute).UnixMilli(), + }, + }, + }, + jobGetError: assert.AnError, // Non-NotFound error + expectedResync: false, // Should continue with normal logic + description: "should handle non-NotFound errors gracefully and continue with normal logic", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + // Create mocks + mockQueue := jobs.NewMockQueue(t) + mockStore := jobs.NewMockStore(t) + mockJobs := &mockJobsQueueStore{ + MockQueue: mockQueue, + MockStore: mockStore, + } + + // Set up job Get mock + if tc.repo.Status.Sync.JobID != "" && (tc.repo.Status.Sync.State == provisioning.JobStatePending || tc.repo.Status.Sync.State == provisioning.JobStateWorking) { + mockStore.On("Get", mock.Anything, tc.repo.Namespace, tc.repo.Status.Sync.JobID).Return(nil, tc.jobGetError).Once() + } + + // Create controller + rc := &RepositoryController{ + jobs: mockJobs, + } + + // Test shouldResync + ctx := context.Background() + result := rc.shouldResync(ctx, tc.repo) + + // Verify + assert.Equal(t, tc.expectedResync, result, tc.description) + }) + } +}