Provisioning: folder not found legacy migration (#104225)

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()

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)
 	}

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) {

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 {

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"))
 			},

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(),

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
 }
 

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{

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)
 	})

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

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
 		}
 	}

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)
+	})
 }