Provisioning: Fix race condition causing unhealthy repository message to be lost (#115150)

* Fix race condition causing unhealthy repository message to be lost This commit fixes a race condition in the provisioning repository controller where the "Repository is unhealthy" message in the sync status could be lost due to status updates being based on stale repository objects. ## Problem The issue occurred in the `process` function when: 1. Repository object was fetched from cache with old status 2. `RefreshHealth` immediately patched the health status to "unhealthy" 3. `determineSyncStatusOps` used the stale object to check if unhealthy message was already set 4. A second patch operation based on stale data would overwrite the health status update ## Solution Introduced `RefreshHealthWithPatchOps` method that returns patch operations instead of immediately applying them. This allows batching all status updates (health + sync) into a single atomic patch operation, eliminating the race condition. ## Changes - Added `HealthCheckerInterface` for better testability - Added `RefreshHealthWithPatchOps` method to return patch ops without applying - Updated `process` function to batch health and sync status updates - Added comprehensive unit tests for the fix Fixes the issue where unhealthy repositories don't show the "Repository is unhealthy" message in their sync status. * Fix staticcheck lint error: remove unnecessary nil check for slice
2025-12-12 13:24:58 +02:00
parent c7c052480d
commit b863acab05
6 changed files with 518 additions and 2 deletions
@@ -26,6 +26,18 @@ type StatusPatcher interface {
 	Patch(ctx context.Context, repo *provisioning.Repository, patchOperations ...map[string]interface{}) error
 }

+// HealthCheckerInterface defines the interface for health checking operations
+//
+//go:generate mockery --name=HealthCheckerInterface --structname=MockHealthChecker
+type HealthCheckerInterface interface {
+	ShouldCheckHealth(repo *provisioning.Repository) bool
+	RefreshHealth(ctx context.Context, repo repository.Repository) (*provisioning.TestResults, provisioning.HealthStatus, error)
+	RefreshHealthWithPatchOps(ctx context.Context, repo repository.Repository) (*provisioning.TestResults, provisioning.HealthStatus, []map[string]interface{}, error)
+	RefreshTimestamp(ctx context.Context, repo *provisioning.Repository) error
+	RecordFailure(ctx context.Context, failureType provisioning.HealthFailureType, err error, repo *provisioning.Repository) error
+	HasRecentFailure(healthStatus provisioning.HealthStatus, failureType provisioning.HealthFailureType) bool
+}
+
 // HealthChecker provides unified health checking for repositories
 type HealthChecker struct {
 	statusPatcher StatusPatcher
@@ -162,6 +174,33 @@ func (hc *HealthChecker) RefreshHealth(ctx context.Context, repo repository.Repo
 	return testResults, newHealthStatus, nil
 }

+// RefreshHealthWithPatchOps performs a health check on an existing repository
+// and returns the test results, health status, and patch operations to apply.
+// This method does NOT apply the patch itself, allowing the caller to batch
+// multiple status updates together to avoid race conditions.
+func (hc *HealthChecker) RefreshHealthWithPatchOps(ctx context.Context, repo repository.Repository) (*provisioning.TestResults, provisioning.HealthStatus, []map[string]interface{}, error) {
+	cfg := repo.Config()
+
+	// Use health checker to perform comprehensive health check with existing status
+	testResults, newHealthStatus, err := hc.refreshHealth(ctx, repo, cfg.Status.Health)
+	if err != nil {
+		return nil, provisioning.HealthStatus{}, nil, fmt.Errorf("health check failed: %w", err)
+	}
+
+	var patchOps []map[string]interface{}
+
+	// Only return patch operation if health status actually changed
+	if hc.hasHealthStatusChanged(cfg.Status.Health, newHealthStatus) {
+		patchOps = append(patchOps, map[string]interface{}{
+			"op":    "replace",
+			"path":  "/status/health",
+			"value": newHealthStatus,
+		})
+	}
+
+	return testResults, newHealthStatus, patchOps, nil
+}
+
 // RefreshTimestamp updates the health status timestamp without changing other fields
 func (hc *HealthChecker) RefreshTimestamp(ctx context.Context, repo *provisioning.Repository) error {
 	// Update the timestamp on the existing health status
@@ -532,6 +532,136 @@ func TestRefreshHealth(t *testing.T) {
 	}
 }

+func TestRefreshHealthWithPatchOps(t *testing.T) {
+	tests := []struct {
+		name              string
+		testResult        *provisioning.TestResults
+		testError         error
+		existingStatus    provisioning.HealthStatus
+		expectError       bool
+		expectedHealth    bool
+		expectPatchOps    bool
+		expectedPatchPath string
+	}{
+		{
+			name: "successful health check with status change",
+			testResult: &provisioning.TestResults{
+				Success: true,
+				Code:    200,
+			},
+			testError: nil,
+			existingStatus: provisioning.HealthStatus{
+				Healthy: false,
+				Error:   provisioning.HealthFailureHealth,
+				Checked: time.Now().Add(-time.Hour).UnixMilli(),
+			},
+			expectError:       false,
+			expectedHealth:    true,
+			expectPatchOps:    true,
+			expectedPatchPath: "/status/health",
+		},
+		{
+			name: "failed health check with status change",
+			testResult: &provisioning.TestResults{
+				Success: false,
+				Code:    500,
+				Errors: []provisioning.ErrorDetails{
+					{Detail: "connection failed"},
+				},
+			},
+			testError: nil,
+			existingStatus: provisioning.HealthStatus{
+				Healthy: true,
+				Checked: time.Now().Add(-time.Hour).UnixMilli(),
+			},
+			expectError:       false,
+			expectedHealth:    false,
+			expectPatchOps:    true,
+			expectedPatchPath: "/status/health",
+		},
+		{
+			name: "no status change - no patch ops returned",
+			testResult: &provisioning.TestResults{
+				Success: true,
+				Code:    200,
+			},
+			testError: nil,
+			existingStatus: provisioning.HealthStatus{
+				Healthy: true,
+				Checked: time.Now().Add(-15 * time.Second).UnixMilli(),
+			},
+			expectError:    false,
+			expectedHealth: true,
+			expectPatchOps: false,
+		},
+		{
+			name:       "test repository error",
+			testResult: nil,
+			testError:  errors.New("repository test failed"),
+			existingStatus: provisioning.HealthStatus{
+				Healthy: true,
+				Checked: time.Now().Add(-time.Hour).UnixMilli(),
+			},
+			expectError:    true,
+			expectedHealth: false,
+			expectPatchOps: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create mock repository
+			mockRepo := &mockRepository{
+				config: &provisioning.Repository{
+					Spec: provisioning.RepositorySpec{
+						Title: "Test Repository",
+						Type:  provisioning.LocalRepositoryType,
+					},
+					Status: provisioning.RepositoryStatus{
+						Health: tt.existingStatus,
+					},
+				},
+				testResult: tt.testResult,
+				testError:  tt.testError,
+			}
+
+			// Create health checker with validator and tester
+			validator := repository.NewValidator(30*time.Second, []provisioning.SyncTargetType{provisioning.SyncTargetTypeFolder, provisioning.SyncTargetTypeInstance}, true)
+			hc := NewHealthChecker(nil, prometheus.NewPedanticRegistry(), repository.NewSimpleRepositoryTester(validator))
+
+			// Call RefreshHealthWithPatchOps
+			testResults, healthStatus, patchOps, err := hc.RefreshHealthWithPatchOps(context.Background(), mockRepo)
+
+			// Verify error
+			if tt.expectError {
+				assert.Error(t, err)
+				assert.Nil(t, testResults)
+				return
+			}
+			assert.NoError(t, err)
+
+			// Verify health status
+			assert.Equal(t, tt.expectedHealth, healthStatus.Healthy)
+
+			// Verify patch operations
+			if tt.expectPatchOps {
+				assert.NotEmpty(t, patchOps, "expected patch operations to be returned")
+				assert.Len(t, patchOps, 1)
+				assert.Equal(t, "replace", patchOps[0]["op"])
+				assert.Equal(t, tt.expectedPatchPath, patchOps[0]["path"])
+				assert.Equal(t, healthStatus, patchOps[0]["value"])
+			} else {
+				assert.Empty(t, patchOps, "expected no patch operations to be returned")
+			}
+
+			// Verify test results
+			if tt.testResult != nil {
+				assert.Equal(t, tt.testResult, testResults)
+			}
+		})
+	}
+}
+
 func TestHasHealthStatusChanged(t *testing.T) {
 	tests := []struct {
 		name     string
@@ -0,0 +1,187 @@
+// Code generated by mockery v2.53.4. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+
+	repository "github.com/grafana/grafana/apps/provisioning/pkg/repository"
+
+	v0alpha1 "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+)
+
+// MockHealthChecker is an autogenerated mock type for the HealthCheckerInterface type
+type MockHealthChecker struct {
+	mock.Mock
+}
+
+// HasRecentFailure provides a mock function with given fields: healthStatus, failureType
+func (_m *MockHealthChecker) HasRecentFailure(healthStatus v0alpha1.HealthStatus, failureType v0alpha1.HealthFailureType) bool {
+	ret := _m.Called(healthStatus, failureType)
+
+	if len(ret) == 0 {
+		panic("no return value specified for HasRecentFailure")
+	}
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(v0alpha1.HealthStatus, v0alpha1.HealthFailureType) bool); ok {
+		r0 = rf(healthStatus, failureType)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	return r0
+}
+
+// RecordFailure provides a mock function with given fields: ctx, failureType, err, repo
+func (_m *MockHealthChecker) RecordFailure(ctx context.Context, failureType v0alpha1.HealthFailureType, err error, repo *v0alpha1.Repository) error {
+	ret := _m.Called(ctx, failureType, err, repo)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RecordFailure")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, v0alpha1.HealthFailureType, error, *v0alpha1.Repository) error); ok {
+		r0 = rf(ctx, failureType, err, repo)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// RefreshHealth provides a mock function with given fields: ctx, repo
+func (_m *MockHealthChecker) RefreshHealth(ctx context.Context, repo repository.Repository) (*v0alpha1.TestResults, v0alpha1.HealthStatus, error) {
+	ret := _m.Called(ctx, repo)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RefreshHealth")
+	}
+
+	var r0 *v0alpha1.TestResults
+	var r1 v0alpha1.HealthStatus
+	var r2 error
+	if rf, ok := ret.Get(0).(func(context.Context, repository.Repository) (*v0alpha1.TestResults, v0alpha1.HealthStatus, error)); ok {
+		return rf(ctx, repo)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, repository.Repository) *v0alpha1.TestResults); ok {
+		r0 = rf(ctx, repo)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*v0alpha1.TestResults)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, repository.Repository) v0alpha1.HealthStatus); ok {
+		r1 = rf(ctx, repo)
+	} else {
+		r1 = ret.Get(1).(v0alpha1.HealthStatus)
+	}
+
+	if rf, ok := ret.Get(2).(func(context.Context, repository.Repository) error); ok {
+		r2 = rf(ctx, repo)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
+}
+
+// RefreshHealthWithPatchOps provides a mock function with given fields: ctx, repo
+func (_m *MockHealthChecker) RefreshHealthWithPatchOps(ctx context.Context, repo repository.Repository) (*v0alpha1.TestResults, v0alpha1.HealthStatus, []map[string]interface{}, error) {
+	ret := _m.Called(ctx, repo)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RefreshHealthWithPatchOps")
+	}
+
+	var r0 *v0alpha1.TestResults
+	var r1 v0alpha1.HealthStatus
+	var r2 []map[string]interface{}
+	var r3 error
+	if rf, ok := ret.Get(0).(func(context.Context, repository.Repository) (*v0alpha1.TestResults, v0alpha1.HealthStatus, []map[string]interface{}, error)); ok {
+		return rf(ctx, repo)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, repository.Repository) *v0alpha1.TestResults); ok {
+		r0 = rf(ctx, repo)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*v0alpha1.TestResults)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, repository.Repository) v0alpha1.HealthStatus); ok {
+		r1 = rf(ctx, repo)
+	} else {
+		r1 = ret.Get(1).(v0alpha1.HealthStatus)
+	}
+
+	if rf, ok := ret.Get(2).(func(context.Context, repository.Repository) []map[string]interface{}); ok {
+		r2 = rf(ctx, repo)
+	} else {
+		if ret.Get(2) != nil {
+			r2 = ret.Get(2).([]map[string]interface{})
+		}
+	}
+
+	if rf, ok := ret.Get(3).(func(context.Context, repository.Repository) error); ok {
+		r3 = rf(ctx, repo)
+	} else {
+		r3 = ret.Error(3)
+	}
+
+	return r0, r1, r2, r3
+}
+
+// RefreshTimestamp provides a mock function with given fields: ctx, repo
+func (_m *MockHealthChecker) RefreshTimestamp(ctx context.Context, repo *v0alpha1.Repository) error {
+	ret := _m.Called(ctx, repo)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RefreshTimestamp")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, *v0alpha1.Repository) error); ok {
+		r0 = rf(ctx, repo)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// ShouldCheckHealth provides a mock function with given fields: repo
+func (_m *MockHealthChecker) ShouldCheckHealth(repo *v0alpha1.Repository) bool {
+	ret := _m.Called(repo)
+
+	if len(ret) == 0 {
+		panic("no return value specified for ShouldCheckHealth")
+	}
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(*v0alpha1.Repository) bool); ok {
+		r0 = rf(repo)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	return r0
+}
+
+// NewMockHealthChecker creates a new instance of MockHealthChecker. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewMockHealthChecker(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *MockHealthChecker {
+	mock := &MockHealthChecker{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
@@ -1,4 +1,4 @@
-// Code generated by mockery v2.52.4. DO NOT EDIT.
+// Code generated by mockery v2.53.4. DO NOT EDIT.

 package mocks

@@ -561,11 +561,16 @@ func (rc *RepositoryController) process(item *queueItem) error {
 	}

 	// Handle health checks using the health checker
-	_, healthStatus, err := rc.healthChecker.RefreshHealth(ctx, repo)
+	_, healthStatus, healthPatchOps, err := rc.healthChecker.RefreshHealthWithPatchOps(ctx, repo)
 	if err != nil {
 		return fmt.Errorf("update health status: %w", err)
 	}

+	// Add health patch operations first
+	if len(healthPatchOps) > 0 {
+		patchOperations = append(patchOperations, healthPatchOps...)
+	}
+
 	// determine the sync strategy and sync status to apply
 	syncOptions := rc.determineSyncStrategy(ctx, obj, repo, shouldResync, healthStatus)
 	patchOperations = append(patchOperations, rc.determineSyncStatusOps(obj, syncOptions, healthStatus)...)
@@ -350,6 +350,161 @@ type mockJobsQueueStore struct {
 	*jobs.MockStore
 }

+func TestRepositoryController_process_UnhealthyRepositoryStatusUpdate(t *testing.T) {
+	testCases := []struct {
+		name                     string
+		repo                     *provisioning.Repository
+		healthStatus             provisioning.HealthStatus
+		hasHealthStatusChanged   bool
+		expectedUnhealthyMessage bool
+		description              string
+	}{
+		{
+			name: "unhealthy repository should set unhealthy message in sync status",
+			repo: &provisioning.Repository{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-repo",
+					Namespace:  "default",
+					Generation: 1,
+				},
+				Spec: provisioning.RepositorySpec{
+					Sync: provisioning.SyncOptions{
+						Enabled:         true,
+						IntervalSeconds: 300,
+					},
+				},
+				Status: provisioning.RepositoryStatus{
+					ObservedGeneration: 1,
+					Health: provisioning.HealthStatus{
+						Healthy: true,
+						Checked: time.Now().Add(-10 * time.Minute).UnixMilli(),
+					},
+					Sync: provisioning.SyncStatus{
+						State:    provisioning.JobStateSuccess,
+						Finished: time.Now().Add(-1 * time.Minute).UnixMilli(),
+						Message:  []string{},
+					},
+				},
+			},
+			healthStatus: provisioning.HealthStatus{
+				Healthy: false,
+				Error:   provisioning.HealthFailureHealth,
+				Checked: time.Now().UnixMilli(),
+				Message: []string{"connection failed"},
+			},
+			hasHealthStatusChanged:   true,
+			expectedUnhealthyMessage: true,
+			description:              "should set unhealthy message when repository becomes unhealthy",
+		},
+		{
+			name: "unhealthy repository should not duplicate unhealthy message",
+			repo: &provisioning.Repository{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-repo",
+					Namespace:  "default",
+					Generation: 1,
+				},
+				Spec: provisioning.RepositorySpec{
+					Sync: provisioning.SyncOptions{
+						Enabled:         true,
+						IntervalSeconds: 300,
+					},
+				},
+				Status: provisioning.RepositoryStatus{
+					ObservedGeneration: 1,
+					Health: provisioning.HealthStatus{
+						Healthy: false,
+						Checked: time.Now().Add(-2 * time.Minute).UnixMilli(),
+					},
+					Sync: provisioning.SyncStatus{
+						State:    provisioning.JobStateError,
+						Finished: time.Now().Add(-1 * time.Minute).UnixMilli(),
+						Message:  []string{"Repository is unhealthy"},
+					},
+				},
+			},
+			healthStatus: provisioning.HealthStatus{
+				Healthy: false,
+				Error:   provisioning.HealthFailureHealth,
+				Checked: time.Now().UnixMilli(),
+				Message: []string{"connection failed"},
+			},
+			hasHealthStatusChanged:   false,
+			expectedUnhealthyMessage: false,
+			description:              "should not set unhealthy message when it already exists",
+		},
+		{
+			name: "healthy repository should clear unhealthy message",
+			repo: &provisioning.Repository{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-repo",
+					Namespace:  "default",
+					Generation: 1,
+				},
+				Spec: provisioning.RepositorySpec{
+					Sync: provisioning.SyncOptions{
+						Enabled:         true,
+						IntervalSeconds: 300,
+					},
+				},
+				Status: provisioning.RepositoryStatus{
+					ObservedGeneration: 1,
+					Health: provisioning.HealthStatus{
+						Healthy: false,
+						Checked: time.Now().Add(-2 * time.Minute).UnixMilli(),
+					},
+					Sync: provisioning.SyncStatus{
+						State:    provisioning.JobStateError,
+						Finished: time.Now().Add(-1 * time.Minute).UnixMilli(),
+						Message:  []string{"Repository is unhealthy"},
+					},
+				},
+			},
+			healthStatus: provisioning.HealthStatus{
+				Healthy: true,
+				Checked: time.Now().UnixMilli(),
+				Message: []string{},
+			},
+			hasHealthStatusChanged:   true,
+			expectedUnhealthyMessage: false,
+			description:              "should clear unhealthy message when repository becomes healthy",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Create controller
+			rc := &RepositoryController{}
+
+			// Determine sync status ops (this is a pure function, no mocks needed)
+			syncOps := rc.determineSyncStatusOps(tc.repo, nil, tc.healthStatus)
+
+			// Verify expectations
+			hasUnhealthyOp := false
+			hasClearUnhealthyOp := false
+			for _, op := range syncOps {
+				if path, ok := op["path"].(string); ok {
+					if path == "/status/sync/message" {
+						if messages, ok := op["value"].([]string); ok {
+							if len(messages) > 0 && messages[0] == "Repository is unhealthy" {
+								hasUnhealthyOp = true
+							} else if len(messages) == 0 {
+								hasClearUnhealthyOp = true
+							}
+						}
+					}
+				}
+			}
+
+			if tc.expectedUnhealthyMessage {
+				assert.True(t, hasUnhealthyOp, tc.description+": expected unhealthy message operation")
+			} else if len(tc.repo.Status.Sync.Message) > 0 && tc.healthStatus.Healthy {
+				assert.True(t, hasClearUnhealthyOp, tc.description+": expected clear unhealthy message operation")
+			}
+		})
+	}
+}
+
 func TestRepositoryController_shouldResync_StaleSyncStatus(t *testing.T) {
 	testCases := []struct {
 		name           string