From c1485ecf5fe7f53b503ddda780d58301cc3c503c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roberto=20Jim=C3=A9nez=20S=C3=A1nchez?= Date: Thu, 13 Nov 2025 17:58:33 +0100 Subject: [PATCH] Provisioning: detect stale sync status and trigger resync (#113826) * provisioning: detect stale sync status and trigger resync When sync jobs expire and are cleaned up by the expired job cleanup controller, the Repository sync status remains stuck in Pending or Working state. This prevents new sync jobs from being queued because shouldResync() blocks on these states. This change adds detection logic in shouldResync() to check if a sync job referenced in the sync status still exists. If the job doesn't exist (NotFound), we trigger a resync to reconcile the stale state. Fixes grafana/git-ui-sync-project#626 * test: remove unused mocks and fix test case - Remove unused mockRepositoryLister and mockRepositoryNamespaceLister types - Remove unused imports (labels, listers) - Remove test case for sync disabled scenario as we don't care about sync enabled state when detecting stale status --- .../provisioning/controller/repository.go | 38 +++- .../controller/repository_test.go | 212 +++++++++++++++++- 2 files changed, 243 insertions(+), 7 deletions(-) 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) + }) + } +}