diff --git a/pkg/registry/apis/provisioning/jobs/job_progress_recorder_mock.go b/pkg/registry/apis/provisioning/jobs/job_progress_recorder_mock.go index 48c03742d28..f7f4c53681f 100644 --- a/pkg/registry/apis/provisioning/jobs/job_progress_recorder_mock.go +++ b/pkg/registry/apis/provisioning/jobs/job_progress_recorder_mock.go @@ -237,6 +237,38 @@ func (_c *MockJobProgressRecorder_SetTotal_Call) RunAndReturn(run func(context.C return _c } +// Strict provides a mock function with no fields +func (_m *MockJobProgressRecorder) Strict() { + _m.Called() +} + +// MockJobProgressRecorder_Strict_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Strict' +type MockJobProgressRecorder_Strict_Call struct { + *mock.Call +} + +// Strict is a helper method to define mock.On call +func (_e *MockJobProgressRecorder_Expecter) Strict() *MockJobProgressRecorder_Strict_Call { + return &MockJobProgressRecorder_Strict_Call{Call: _e.mock.On("Strict")} +} + +func (_c *MockJobProgressRecorder_Strict_Call) Run(run func()) *MockJobProgressRecorder_Strict_Call { + _c.Call.Run(func(args mock.Arguments) { + run() + }) + return _c +} + +func (_c *MockJobProgressRecorder_Strict_Call) Return() *MockJobProgressRecorder_Strict_Call { + _c.Call.Return() + return _c +} + +func (_c *MockJobProgressRecorder_Strict_Call) RunAndReturn(run func()) *MockJobProgressRecorder_Strict_Call { + _c.Run(run) + return _c +} + // TooManyErrors provides a mock function with no fields func (_m *MockJobProgressRecorder) TooManyErrors() error { ret := _m.Called() diff --git a/pkg/registry/apis/provisioning/jobs/migrate/folders.go b/pkg/registry/apis/provisioning/jobs/migrate/folders.go index 68b89360483..6f58eff634b 100644 --- a/pkg/registry/apis/provisioning/jobs/migrate/folders.go +++ b/pkg/registry/apis/provisioning/jobs/migrate/folders.go @@ -88,9 +88,8 @@ func (f *legacyFoldersMigrator) Migrate(ctx context.Context, namespace string, r if !created { result.Action = repository.FileActionIgnored } - progress.Record(ctx, result) - return nil + return err }); err != nil { return fmt.Errorf("export folders from SQL: %w", err) } diff --git a/pkg/registry/apis/provisioning/jobs/migrate/folders_test.go b/pkg/registry/apis/provisioning/jobs/migrate/folders_test.go index 8bea8ad6c41..c5c11850638 100644 --- a/pkg/registry/apis/provisioning/jobs/migrate/folders_test.go +++ b/pkg/registry/apis/provisioning/jobs/migrate/folders_test.go @@ -6,6 +6,7 @@ import ( "fmt" "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" @@ -247,6 +248,66 @@ func TestLegacyFoldersMigrator_Migrate(t *testing.T) { require.NoError(t, err) progress.AssertExpectations(t) }) + t.Run("should fail when folder creation fails", func(t *testing.T) { + mockLegacyMigrator := legacy.NewMockLegacyMigrator(t) + mockLegacyMigrator.On("Migrate", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { + opts := args.Get(1).(legacy.MigrateOptions) + folder := &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": "folder.grafana.app/v1alpha1", + "kind": "Folder", + "metadata": map[string]interface{}{ + "name": "test-folder", + "annotations": map[string]interface{}{ + "folder.grafana.app/uid": "test-folder-uid", + }, + }, + }, + } + folder.SetKind("Folder") + folder.SetAPIVersion("folder.grafana.app/v1alpha1") + + data, err := folder.MarshalJSON() + require.NoError(t, err) + client, err := opts.Store.BulkProcess(context.Background()) + require.NoError(t, err) + require.NoError(t, client.Send(&resource.BulkRequest{ + Key: &resource.ResourceKey{Namespace: "test-namespace", Name: "test-folder"}, + Value: data, + })) + }).Return(&resource.BulkResponse{}, nil) + + mockRepositoryResources := resources.NewMockRepositoryResources(t) + expectedError := errors.New("folder creation failed") + mockRepositoryResources.On("EnsureFolderTreeExists", mock.Anything, "", "", mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + callback := args.Get(4).(func(folder resources.Folder, created bool, err error) error) + // Call the callback with an error and return its result + err := callback(resources.Folder{ + ID: "test-folder-uid", + Path: "/test-folder", + }, true, expectedError) + require.Equal(t, expectedError, err) + }).Return(expectedError) + + migrator := NewLegacyFoldersMigrator(mockLegacyMigrator) + progress := jobs.NewMockJobProgressRecorder(t) + progress.On("SetMessage", mock.Anything, "read folders from SQL").Return() + progress.On("SetMessage", mock.Anything, "export folders from SQL").Return() + progress.On("Record", mock.Anything, mock.MatchedBy(func(result jobs.JobResourceResult) bool { + return result.Action == repository.FileActionCreated && + result.Name == "test-folder-uid" && + result.Resource == resources.FolderResource.Resource && + result.Group == resources.FolderResource.Group && + result.Path == "/test-folder" && + assert.Equal(t, expectedError, result.Error) + })).Return() + + err := migrator.Migrate(context.Background(), "test-namespace", mockRepositoryResources, progress) + require.Error(t, err) + require.Contains(t, err.Error(), "export folders from SQL: folder creation failed") + progress.AssertExpectations(t) + }) } func TestLegacyFoldersMigrator_Close(t *testing.T) { diff --git a/pkg/registry/apis/provisioning/jobs/migrate/worker.go b/pkg/registry/apis/provisioning/jobs/migrate/worker.go index 925e34704bb..6b28955ad49 100644 --- a/pkg/registry/apis/provisioning/jobs/migrate/worker.go +++ b/pkg/registry/apis/provisioning/jobs/migrate/worker.go @@ -43,6 +43,7 @@ func (w *MigrationWorker) Process(ctx context.Context, repo repository.Repositor return errors.New("missing migrate settings") } + progress.Strict() progress.SetTotal(ctx, 10) // will show a progress bar rw, ok := repo.(repository.ReaderWriter) if !ok { diff --git a/pkg/registry/apis/provisioning/jobs/migrate/worker_test.go b/pkg/registry/apis/provisioning/jobs/migrate/worker_test.go index d0b0de64514..2b6ac3e08a2 100644 --- a/pkg/registry/apis/provisioning/jobs/migrate/worker_test.go +++ b/pkg/registry/apis/provisioning/jobs/migrate/worker_test.go @@ -61,6 +61,7 @@ func TestMigrationWorker_ProcessNotReaderWriter(t *testing.T) { } progressRecorder := jobs.NewMockJobProgressRecorder(t) progressRecorder.On("SetTotal", mock.Anything, 10).Return() + progressRecorder.On("Strict").Return() repo := repository.NewMockReader(t) err := worker.Process(context.Background(), repo, job, progressRecorder) @@ -98,6 +99,7 @@ func TestMigrationWorker_Process(t *testing.T) { isLegacyActive: true, setupMocks: func(lm *MockMigrator, um *MockMigrator, ds *dualwrite.MockService, pr *jobs.MockJobProgressRecorder) { pr.On("SetTotal", mock.Anything, 10).Return() + pr.On("Strict").Return() ds.On("ReadFromUnified", mock.Anything, mock.Anything).Return(false, nil) lm.On("Migrate", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) }, @@ -113,6 +115,7 @@ func TestMigrationWorker_Process(t *testing.T) { isLegacyActive: false, setupMocks: func(lm *MockMigrator, um *MockMigrator, ds *dualwrite.MockService, pr *jobs.MockJobProgressRecorder) { pr.On("SetTotal", mock.Anything, 10).Return() + pr.On("Strict").Return() ds.On("ReadFromUnified", mock.Anything, mock.Anything).Return(true, nil) um.On("Migrate", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) }, @@ -128,6 +131,7 @@ func TestMigrationWorker_Process(t *testing.T) { isLegacyActive: true, setupMocks: func(lm *MockMigrator, um *MockMigrator, ds *dualwrite.MockService, pr *jobs.MockJobProgressRecorder) { pr.On("SetTotal", mock.Anything, 10).Return() + pr.On("Strict").Return() ds.On("ReadFromUnified", mock.Anything, mock.Anything).Return(false, nil) lm.On("Migrate", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(errors.New("migration failed")) }, diff --git a/pkg/registry/apis/provisioning/jobs/progress.go b/pkg/registry/apis/provisioning/jobs/progress.go index 7d1e1e50c54..d23abcd9cd2 100644 --- a/pkg/registry/apis/provisioning/jobs/progress.go +++ b/pkg/registry/apis/provisioning/jobs/progress.go @@ -3,6 +3,7 @@ package jobs import ( "context" "fmt" + "sync" "time" "github.com/grafana/grafana-app-sdk/logging" @@ -14,10 +15,17 @@ import ( // or if the job completed func maybeNotifyProgress(threshold time.Duration, fn ProgressFn) ProgressFn { var last time.Time + var mu sync.Mutex return func(ctx context.Context, status provisioning.JobStatus) error { - if status.Finished != 0 || last.IsZero() || time.Since(last) > threshold { + mu.Lock() + shouldNotify := status.Finished != 0 || last.IsZero() || time.Since(last) > threshold + if shouldNotify { last = time.Now() + } + mu.Unlock() + + if shouldNotify { return fn(ctx, status) } @@ -36,8 +44,10 @@ type JobResourceResult struct { } type jobProgressRecorder struct { + mu sync.RWMutex started time.Time total int + maxErrors int message string finalMessage string resultCount int @@ -50,7 +60,8 @@ type jobProgressRecorder struct { func newJobProgressRecorder(ProgressFn ProgressFn) JobProgressRecorder { return &jobProgressRecorder{ - started: time.Now(), + maxErrors: 20, + started: time.Now(), // Have a faster notifier for messages and total notifyImmediatelyFn: maybeNotifyProgress(500*time.Millisecond, ProgressFn), maybeNotifyFn: maybeNotifyProgress(5*time.Second, ProgressFn), @@ -59,6 +70,7 @@ func newJobProgressRecorder(ProgressFn ProgressFn) JobProgressRecorder { } func (r *jobProgressRecorder) Record(ctx context.Context, result JobResourceResult) { + r.mu.Lock() r.resultCount++ logger := logging.FromContext(ctx).With("path", result.Path, "resource", result.Resource, "group", result.Group, "action", result.Action, "name", result.Name) @@ -73,11 +85,16 @@ func (r *jobProgressRecorder) Record(ctx context.Context, result JobResourceResu } r.updateSummary(result) + r.mu.Unlock() + r.maybeNotify(ctx) } // ResetResults will reset the results of the job func (r *jobProgressRecorder) ResetResults() { + r.mu.Lock() + defer r.mu.Unlock() + r.resultCount = 0 r.errorCount = 0 r.errors = nil @@ -85,24 +102,41 @@ func (r *jobProgressRecorder) ResetResults() { } func (r *jobProgressRecorder) SetMessage(ctx context.Context, msg string) { + r.mu.Lock() r.message = msg + r.mu.Unlock() + logging.FromContext(ctx).Info("job progress message", "message", msg) r.notifyImmediately(ctx) } func (r *jobProgressRecorder) SetFinalMessage(ctx context.Context, msg string) { + r.mu.Lock() r.finalMessage = msg + r.mu.Unlock() + logging.FromContext(ctx).Info("job final message", "message", msg) } func (r *jobProgressRecorder) SetTotal(ctx context.Context, total int) { + r.mu.Lock() r.total = total + r.mu.Unlock() r.notifyImmediately(ctx) } +func (r *jobProgressRecorder) Strict() { + r.mu.Lock() + r.maxErrors = 1 + r.mu.Unlock() +} + func (r *jobProgressRecorder) TooManyErrors() error { - if r.errorCount > 20 { + r.mu.RLock() + defer r.mu.RUnlock() + + if r.errorCount >= r.maxErrors { return fmt.Errorf("too many errors: %d", r.errorCount) } @@ -110,6 +144,9 @@ func (r *jobProgressRecorder) TooManyErrors() error { } func (r *jobProgressRecorder) summary() []*provisioning.JobResourceSummary { + r.mu.RLock() + defer r.mu.RUnlock() + if len(r.summaries) == 0 { return nil } @@ -123,6 +160,7 @@ func (r *jobProgressRecorder) summary() []*provisioning.JobResourceSummary { } func (r *jobProgressRecorder) updateSummary(result JobResourceResult) { + // Note: This method is called from Record() which already holds the lock key := result.Resource + ":" + result.Group summary, exists := r.summaries[key] if !exists { @@ -155,6 +193,7 @@ func (r *jobProgressRecorder) updateSummary(result JobResourceResult) { } func (r *jobProgressRecorder) progress() float64 { + // Note: This method is called from currentStatus() which already holds the lock if r.total == 0 { return 0 } @@ -162,15 +201,22 @@ func (r *jobProgressRecorder) progress() float64 { return float64(r.resultCount) / float64(r.total) * 100 } -func (r *jobProgressRecorder) notifyImmediately(ctx context.Context) { - jobStatus := provisioning.JobStatus{ +func (r *jobProgressRecorder) currentStatus() provisioning.JobStatus { + r.mu.RLock() + defer r.mu.RUnlock() + + return provisioning.JobStatus{ + Started: r.started.UnixMilli(), State: provisioning.JobStateWorking, Message: r.message, Errors: r.errors, Progress: r.progress(), Summary: r.summary(), } +} +func (r *jobProgressRecorder) notifyImmediately(ctx context.Context) { + jobStatus := r.currentStatus() logger := logging.FromContext(ctx) if err := r.notifyImmediatelyFn(ctx, jobStatus); err != nil { logger.Warn("error notifying immediate progress", "err", err) @@ -178,13 +224,7 @@ func (r *jobProgressRecorder) notifyImmediately(ctx context.Context) { } func (r *jobProgressRecorder) maybeNotify(ctx context.Context) { - jobStatus := provisioning.JobStatus{ - State: provisioning.JobStateWorking, - Message: r.message, - Errors: r.errors, - Progress: r.progress(), - Summary: r.summary(), - } + jobStatus := r.currentStatus() logger := logging.FromContext(ctx) if err := r.maybeNotifyFn(ctx, jobStatus); err != nil { @@ -193,6 +233,9 @@ func (r *jobProgressRecorder) maybeNotify(ctx context.Context) { } func (r *jobProgressRecorder) Complete(ctx context.Context, err error) provisioning.JobStatus { + r.mu.RLock() + defer r.mu.RUnlock() + // Initialize base job status jobStatus := provisioning.JobStatus{ Started: r.started.UnixMilli(), diff --git a/pkg/registry/apis/provisioning/jobs/queue.go b/pkg/registry/apis/provisioning/jobs/queue.go index 16c5173f3f4..0502fb7886a 100644 --- a/pkg/registry/apis/provisioning/jobs/queue.go +++ b/pkg/registry/apis/provisioning/jobs/queue.go @@ -24,6 +24,7 @@ type JobProgressRecorder interface { SetMessage(ctx context.Context, msg string) SetTotal(ctx context.Context, total int) TooManyErrors() error + Strict() Complete(ctx context.Context, err error) provisioning.JobStatus } diff --git a/pkg/registry/apis/provisioning/repository/go-git/wrapper.go b/pkg/registry/apis/provisioning/repository/go-git/wrapper.go index 0082ee60f58..d4d9af45c23 100644 --- a/pkg/registry/apis/provisioning/repository/go-git/wrapper.go +++ b/pkg/registry/apis/provisioning/repository/go-git/wrapper.go @@ -371,7 +371,9 @@ func (g *GoGitRepo) Delete(ctx context.Context, path string, ref string, message func (g *GoGitRepo) Read(ctx context.Context, path string, ref string) (*repository.FileInfo, error) { readPath := safepath.Join(g.config.Spec.GitHub.Path, path) stat, err := g.tree.Filesystem.Lstat(readPath) - if err != nil { + if errors.Is(err, fs.ErrNotExist) { + return nil, repository.ErrFileNotFound + } else if err != nil { return nil, fmt.Errorf("failed to stat path '%s': %w", readPath, err) } info := &repository.FileInfo{ diff --git a/pkg/registry/apis/provisioning/resources/folders.go b/pkg/registry/apis/provisioning/resources/folders.go index 6a065f2c5d4..598c73007ab 100644 --- a/pkg/registry/apis/provisioning/resources/folders.go +++ b/pkg/registry/apis/provisioning/resources/folders.go @@ -155,7 +155,7 @@ func (fm *FolderManager) GetFolder(ctx context.Context, name string) (*unstructu // If the folder already exists, the function is called with created set to false. // If the folder is created, the function is called with created set to true. func (fm *FolderManager) EnsureFolderTreeExists(ctx context.Context, ref, path string, tree FolderTree, fn func(folder Folder, created bool, err error) error) error { - return tree.Walk(ctx, func(ctx context.Context, folder Folder) error { + return tree.Walk(ctx, func(ctx context.Context, folder Folder, parent string) error { p := folder.Path if path != "" { p = safepath.Join(path, p) @@ -175,6 +175,8 @@ func (fm *FolderManager) EnsureFolderTreeExists(ctx context.Context, ref, path s if err := fm.repo.Create(ctx, p, ref, nil, msg); err != nil { return fn(folder, true, fmt.Errorf("write folder in repo: %w", err)) } + // Add it to the existing tree + fm.tree.Add(folder, parent) return fn(folder, true, nil) }) diff --git a/pkg/registry/apis/provisioning/resources/resources.go b/pkg/registry/apis/provisioning/resources/resources.go index 3ff01b0840e..7f819999b90 100644 --- a/pkg/registry/apis/provisioning/resources/resources.go +++ b/pkg/registry/apis/provisioning/resources/resources.go @@ -94,13 +94,10 @@ func (r *ResourcesManager) CreateResourceFileFromObject(ctx context.Context, obj folder := meta.GetFolder() // Get the absolute path of the folder - fid, ok := r.folders.Tree().DirPath(folder, "") + rootFolder := RootFolder(r.repo.Config()) + fid, ok := r.folders.Tree().DirPath(folder, rootFolder) if !ok { - // FIXME: Shouldn't this fail instead? - fid = Folder{ - Path: "__folder_not_found/" + slugify.Slugify(folder), - } - // r.logger.Error("folder of item was not in tree of repository") + return "", fmt.Errorf("folder not found in tree: %s", folder) } // Clear the metadata diff --git a/pkg/registry/apis/provisioning/resources/tree.go b/pkg/registry/apis/provisioning/resources/tree.go index ec7a5c22d88..e18ad25f873 100644 --- a/pkg/registry/apis/provisioning/resources/tree.go +++ b/pkg/registry/apis/provisioning/resources/tree.go @@ -86,7 +86,7 @@ func (t *folderTree) Count() int { return t.count } -type WalkFunc func(ctx context.Context, folder Folder) error +type WalkFunc func(ctx context.Context, folder Folder, parent string) error func (t *folderTree) Walk(ctx context.Context, fn WalkFunc) error { toWalk := make([]Folder, 0, len(t.folders)) @@ -101,7 +101,7 @@ func (t *folderTree) Walk(ctx context.Context, fn WalkFunc) error { }) for _, folder := range toWalk { - if err := fn(ctx, folder); err != nil { + if err := fn(ctx, folder, t.tree[folder.ID]); err != nil { return err } } diff --git a/pkg/registry/apis/provisioning/resources/tree_test.go b/pkg/registry/apis/provisioning/resources/tree_test.go index a98afbbbf3b..3407f644be0 100644 --- a/pkg/registry/apis/provisioning/resources/tree_test.go +++ b/pkg/registry/apis/provisioning/resources/tree_test.go @@ -1,6 +1,7 @@ package resources import ( + "context" "testing" "github.com/stretchr/testify/assert" @@ -89,4 +90,30 @@ func TestFolderTree(t *testing.T) { assert.Empty(t, id.Title) } }) + + t.Run("walk tree", func(t *testing.T) { + tree := &folderTree{ + tree: map[string]string{"a": "b", "b": "c", "c": "x", "x": ""}, + folders: map[string]Folder{ + "x": newFid("x", "X!"), + "c": newFid("c", "C :)"), + "b": newFid("b", "!!B#!"), + "a": newFid("a", "[€]@£a"), + }, + } + + visited := make(map[string]string) + err := tree.Walk(context.Background(), func(ctx context.Context, folder Folder, parent string) error { + visited[folder.ID] = parent + return nil + }) + + assert.NoError(t, err) + assert.Equal(t, map[string]string{ + "x": "", + "c": "x", + "b": "c", + "a": "b", + }, visited) + }) }