Provisioning: use the sync job to finish the migrate job (#101107)

pkg/registry/apis/provisioning/jobs/migrate/import.go

@@ -3,53 +3,18 @@ package migrate
 import (
 	"context"
 	"fmt"
-	"path"
-	"strings"
 	"time"
 
 	"google.golang.org/grpc/metadata"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	"github.com/grafana/grafana-app-sdk/logging"
-	"github.com/grafana/grafana/pkg/apimachinery/utils"
 	dashboard "github.com/grafana/grafana/pkg/apis/dashboard"
 	folders "github.com/grafana/grafana/pkg/apis/folder/v0alpha1"
-	"github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs"
-	"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository"
-	"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
-	"github.com/grafana/grafana/pkg/registry/apis/provisioning/safepath"
 	"github.com/grafana/grafana/pkg/storage/legacysql/dualwrite"
 	"github.com/grafana/grafana/pkg/storage/unified/resource"
 )
 
-func (j *migrationJob) importFromRepo(ctx context.Context, dual dualwrite.Service) error {
-	err := j.wipeUnifiedAndSetMigratedFlag(ctx, dual)
-	if err != nil {
-		return fmt.Errorf("unable to reset unified storage %w", err)
-	}
-
-	tree, err := j.target.ReadTree(ctx, "")
-	if err != nil {
-		return err
-	}
-
-	folders, err := j.createFolders(ctx, tree)
-	if err != nil {
-		return err
-	}
-
-	for _, item := range tree {
-		if !item.Blob || resources.ShouldIgnorePath(item.Path) {
-			continue
-		}
-		_ = j.writeResourceFromFile(ctx, item.Path, folders)
-	}
-
-	return nil
-}
-
 func (j *migrationJob) wipeUnifiedAndSetMigratedFlag(ctx context.Context, dual dualwrite.Service) error {
 	kinds := []schema.GroupResource{{
 		Group:    folders.GROUP,
@@ -100,151 +65,3 @@ func (j *migrationJob) wipeUnifiedAndSetMigratedFlag(ctx context.Context, dual d
 
 	return nil
 }
-
-func (j *migrationJob) createFolders(ctx context.Context, tree []repository.FileTreeEntry) (*resources.FolderTree, error) {
-	folders := j.client.Resource(schema.GroupVersionResource{
-		Group:    folders.GROUP,
-		Version:  folders.VERSION,
-		Resource: folders.RESOURCE,
-	})
-
-	repoName := j.target.Config().Name
-	folderLookup := resources.NewEmptyFolderTree()
-	parent := resources.RootFolder(j.target.Config())
-	if parent != "" {
-		folderLookup.Add(resources.Folder{
-			ID: parent,
-		}, "")
-		obj := &unstructured.Unstructured{
-			Object: map[string]interface{}{
-				"spec": map[string]any{
-					"title": j.target.Config().Spec.Title,
-				},
-			},
-		}
-
-		meta, err := utils.MetaAccessor(obj)
-		if err != nil {
-			return nil, fmt.Errorf("create meta accessor for the object: %w", err)
-		}
-
-		obj.SetNamespace(j.namespace)
-		obj.SetName(parent)
-		meta.SetRepositoryInfo(&utils.ResourceRepositoryInfo{
-			Name: repoName,
-		})
-		if _, err := folders.Create(ctx, obj, metav1.CreateOptions{}); err != nil {
-			return nil, fmt.Errorf("failed to create root folder: %w", err)
-		}
-	}
-
-	var err error
-	for _, item := range tree {
-		if item.Blob {
-			continue
-		}
-
-		f := resources.ParseFolder(item.Path, repoName)
-		if folderLookup.In(f.ID) {
-			continue
-		}
-
-		parent := resources.RootFolder(j.target.Config())
-		var traverse string
-		for i, part := range strings.Split(f.Path, "/") {
-			if i == 0 {
-				traverse = part
-			} else {
-				traverse, err = safepath.Join(traverse, part)
-				if err != nil {
-					return nil, fmt.Errorf("unable to make path: %w", err)
-				}
-			}
-
-			f := resources.ParseFolder(traverse, repoName)
-			if folderLookup.In(f.ID) {
-				parent = f.ID
-				continue
-			}
-			folderLookup.Add(f, parent)
-
-			obj := &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"spec": map[string]any{
-						"title": part,
-					},
-				},
-			}
-
-			meta, err := utils.MetaAccessor(obj)
-			if err != nil {
-				return nil, fmt.Errorf("create meta accessor for the object: %w", err)
-			}
-
-			obj.SetNamespace(j.namespace)
-			obj.SetName(f.ID)
-			meta.SetFolder(parent)
-			meta.SetRepositoryInfo(&utils.ResourceRepositoryInfo{
-				Name:      repoName,
-				Path:      item.Path,
-				Hash:      "",  // FIXME: which hash?
-				Timestamp: nil, // ???&info.Modified.Time,
-			})
-
-			if _, err := folders.Create(ctx, obj, metav1.CreateOptions{}); err != nil {
-				return nil, fmt.Errorf("failed to create folder: %w", err)
-			}
-
-			// traversal
-			parent = f.ID
-		}
-	}
-	return folderLookup, err
-}
-
-// copied from sync
-func (j *migrationJob) writeResourceFromFile(ctx context.Context, filepath string, folders *resources.FolderTree) jobs.JobResourceResult {
-	result := jobs.JobResourceResult{
-		Path:   filepath,
-		Action: repository.FileActionCreated,
-	}
-
-	if resources.ShouldIgnorePath(filepath) {
-		result.Action = repository.FileActionIgnored
-		return result
-	}
-
-	// Read the referenced file
-	fileInfo, err := j.target.Read(ctx, filepath, "")
-	if err != nil {
-		result.Error = fmt.Errorf("failed to read file: %w", err)
-		return result
-	}
-
-	parsed, err := j.parser.Parse(ctx, fileInfo, false) // no validation (we are about to write anyway)
-	if err != nil {
-		result.Error = fmt.Errorf("failed to parse file: %w", err)
-		return result
-	}
-
-	dir := path.Dir(filepath)
-	if dir != "." {
-		f := resources.ParseFolder(dir, j.target.Config().Name)
-		if !folders.In(f.ID) {
-			result.Error = fmt.Errorf("parent folder does not exist: %s", filepath)
-			return result
-		}
-		parsed.Meta.SetFolder(f.ID)
-	}
-
-	parsed.Meta.SetUID("")             // clear identifiers
-	parsed.Meta.SetResourceVersion("") // clear identifiers
-
-	result.Name = parsed.Obj.GetName()
-	result.Resource = parsed.GVR.Resource
-	result.Group = parsed.GVK.Group
-
-	// migrate is always create
-	_, result.Error = parsed.Client.Create(ctx, parsed.Obj, metav1.CreateOptions{})
-	return result
-}

pkg/registry/apis/provisioning/jobs/migrate/resources.go

@@ -9,7 +9,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	"github.com/grafana/grafana/pkg/apimachinery/utils"
-	dashboards "github.com/grafana/grafana/pkg/apis/dashboard"
+	"github.com/grafana/grafana/pkg/apis/dashboard"
 	"github.com/grafana/grafana/pkg/infra/slugify"
 	"github.com/grafana/grafana/pkg/registry/apis/dashboard/legacy"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs"
@@ -38,14 +38,20 @@ func (r *resourceReader) CloseWithResults() (*resource.BatchResponse, error) {
 
 // Write implements resource.BatchResourceWriter.
 func (r *resourceReader) Write(ctx context.Context, key *resource.ResourceKey, value []byte) error {
-	item := &unstructured.Unstructured{}
-	err := item.UnmarshalJSON(value)
+	// Reuse the same parse+cleanup logic
+	parsed, err := r.job.parser.Parse(ctx, &repository.FileInfo{
+		Path: "", // empty path to ignore file system
+		Data: value,
+	}, false)
 	if err != nil {
 		// TODO: should we fail the entire execution?
 		return fmt.Errorf("failed to unmarshal unstructured: %w", err)
 	}
 
-	if result := r.job.write(ctx, item); result.Error != nil {
+	// clear anything so it will get written
+	parsed.Meta.SetRepositoryInfo(nil)
+
+	if result := r.job.write(ctx, parsed.Obj); result.Error != nil {
 		r.job.progress.Record(ctx, result)
 		if err := r.job.progress.TooManyErrors(); err != nil {
 			return err
@@ -57,8 +63,8 @@ func (r *resourceReader) Write(ctx context.Context, key *resource.ResourceKey, v
 
 func (j *migrationJob) loadResources(ctx context.Context) error {
 	kinds := []schema.GroupVersionResource{{
-		Group:    dashboards.GROUP,
-		Resource: dashboards.DASHBOARD_RESOURCE,
+		Group:    dashboard.GROUP,
+		Resource: dashboard.DASHBOARD_RESOURCE,
 		Version:  "v1alpha1",
 	}}
 

pkg/registry/apis/provisioning/jobs/migrate/worker.go

@@ -11,6 +11,7 @@ import (
 	provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
 	"github.com/grafana/grafana/pkg/registry/apis/dashboard/legacy"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs"
+	"github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/sync"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository"
 	gogit "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository/go-git"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
@@ -41,6 +42,9 @@ type MigrationWorker struct {
 
 	// Decrypt secret from config object
 	secrets secrets.Service
+
+	// Delegate the import to sync worker
+	syncWorker *sync.SyncWorker
 }
 
 func NewMigrationWorker(clients *resources.ClientFactory,
@@ -49,6 +53,7 @@ func NewMigrationWorker(clients *resources.ClientFactory,
 	storageStatus dualwrite.Service,
 	batch resource.BatchStoreClient,
 	secrets secrets.Service,
+	syncWorker *sync.SyncWorker,
 	clonedir string,
 ) *MigrationWorker {
 	return &MigrationWorker{
@@ -59,6 +64,7 @@ func NewMigrationWorker(clients *resources.ClientFactory,
 		legacyMigrator,
 		batch,
 		secrets,
+		syncWorker,
 	}
 }
 
@@ -153,10 +159,23 @@ func (w *MigrationWorker) Process(ctx context.Context, repo repository.Repositor
 		}
 	}
 
-	// Now import from out local checkout
-	// TODO: can we skip this and write while exporting?
-	// YES... but :( the export with *history* does not know the current value
-	return worker.importFromRepo(ctx, w.storageStatus)
+	// Clear unified and allow writing
+	err = worker.wipeUnifiedAndSetMigratedFlag(ctx, w.storageStatus)
+	if err != nil {
+		return fmt.Errorf("unable to reset unified storage %w", err)
+	}
+
+	// enable sync (won't be saved)
+	repo.Config().Spec.Sync.Enabled = true
+
+	// Delegate the import to a sync (from the already checked out go-git repository!)
+	return w.syncWorker.Process(ctx, repo, provisioning.Job{
+		Spec: provisioning.JobSpec{
+			Sync: &provisioning.SyncJobOptions{
+				Incremental: false,
+			},
+		},
+	}, progress)
 }
 
 // MigrationJob holds all context for a running job

pkg/registry/apis/provisioning/jobs/sync/worker.go

@@ -93,7 +93,7 @@ func (r *SyncWorker) Process(ctx context.Context, repo repository.Repository, jo
 	}
 
 	syncError := syncJob.run(ctx, *job.Spec.Sync)
-	jobStatus := progress.Complete(ctx, syncJob.run(ctx, *job.Spec.Sync))
+	jobStatus := progress.Complete(ctx, syncError)
 	syncStatus := jobStatus.ToSyncStatus(job.Name)
 
 	// Create sync status and set hash if successful
@@ -166,7 +166,7 @@ func (r *SyncWorker) createJob(ctx context.Context, repo repository.Repository,
 			Version:  "v1alpha1",
 			Resource: dashboard.DASHBOARD_RESOURCE,
 		}),
-		resourcesLookup: map[resourceID]bool{},
+		resourcesLookup: map[resourceID]string{},
 	}
 
 	return job, nil
@@ -202,7 +202,7 @@ type syncJob struct {
 	folders         dynamic.ResourceInterface
 	dashboards      dynamic.ResourceInterface
 	folderLookup    *resources.FolderTree
-	resourcesLookup map[resourceID]bool
+	resourcesLookup map[resourceID]string // the path with this k8s name
 }
 
 func (r *syncJob) run(ctx context.Context, options provisioning.SyncJobOptions) error {
@@ -266,6 +266,10 @@ func (r *syncJob) run(ctx context.Context, options provisioning.SyncJobOptions)
 }
 
 func (r *syncJob) applyChanges(ctx context.Context, changes []ResourceFileChange) error {
+	if len(r.resourcesLookup) > 0 {
+		return fmt.Errorf("this should be empty")
+	}
+
 	// Do the longest paths first (important for delete)
 	sort.Slice(changes, func(i, j int) bool {
 		return len(changes[i].Path) > len(changes[j].Path)
@@ -437,12 +441,12 @@ func (r *syncJob) writeResourceFromFile(ctx context.Context, path string, ref st
 		Resource: parsed.GVR.Resource,
 		Group:    parsed.GVK.Group,
 	}
-
-	if r.resourcesLookup[id] {
-		result.Error = fmt.Errorf("duplicate resource name: %s", parsed.Obj.GetName())
+	existing, found := r.resourcesLookup[id]
+	if found {
+		result.Error = fmt.Errorf("duplicate resource name: %s, %s and %s", parsed.Obj.GetName(), path, existing)
 		return result
 	}
-	r.resourcesLookup[id] = true
+	r.resourcesLookup[id] = path
 
 	// Make sure the parent folders exist
 	folder, err := r.ensureFolderPathExists(ctx, path)

pkg/registry/apis/provisioning/register.go

@@ -483,6 +483,19 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
 			}
 			b.repositoryLister = repoInformer.Lister()
 
+			b.jobs.Register(export.NewExportWorker(
+				b.client,
+				b.storageStatus,
+				b.secrets,
+				b.clonedir,
+			))
+			syncWorker := sync.NewSyncWorker(
+				c.ProvisioningV0alpha1(),
+				b.parsers,
+				b.resourceLister,
+				b.storageStatus,
+			)
+			b.jobs.Register(syncWorker)
 			b.jobs.Register(migrate.NewMigrationWorker(
 				b.client,
 				b.legacyMigrator,
@@ -490,20 +503,9 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
 				b.storageStatus,
 				b.unified,
 				b.secrets,
+				syncWorker,
 				b.clonedir,
 			))
-			b.jobs.Register(export.NewExportWorker(
-				b.client,
-				b.storageStatus,
-				b.secrets,
-				b.clonedir,
-			))
-			b.jobs.Register(sync.NewSyncWorker(
-				c.ProvisioningV0alpha1(),
-				b.parsers,
-				b.resourceLister,
-				b.storageStatus,
-			))
 
 			renderer := pullrequest.NewRenderer(b.render, b.blobstore)
 			pullRequestWorker, err := pullrequest.NewPullRequestWorker(b.parsers, renderer, b.urlProvider)

pkg/registry/apis/provisioning/repository/go-git/wrapper.go

@@ -208,7 +208,10 @@ func (g *GoGitRepo) Push(ctx context.Context, progress io.Writer) error {
 			All: true, // Add everything that changed
 		})
 		if err != nil {
-			return err
+			// empty commit is fine -- no change
+			if !errors.Is(err, git.ErrEmptyCommit) {
+				return err
+			}
 		}
 	}
 
@@ -321,6 +324,9 @@ func (g *GoGitRepo) Write(ctx context.Context, fpath string, ref string, data []
 		}
 	}
 	_, err = g.tree.Commit(message, opts)
+	if errors.Is(err, git.ErrEmptyCommit) {
+		return nil // empty commit is fine -- no change
+	}
 	return err
 }
 

pkg/registry/apis/provisioning/resources/parser.go

@@ -19,6 +19,7 @@ import (
 	"github.com/grafana/grafana-app-sdk/logging"
 	"github.com/grafana/grafana/pkg/apimachinery/apis/common/v0alpha1"
 	"github.com/grafana/grafana/pkg/apimachinery/utils"
+	"github.com/grafana/grafana/pkg/apis/dashboard"
 	provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/lint"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository"
@@ -126,7 +127,7 @@ func (r *Parser) Parse(ctx context.Context, info *repository.FileInfo, validate
 		Info: info,
 	}
 
-	if ShouldIgnorePath(info.Path) {
+	if ShouldIgnorePath(info.Path) && info.Path != "" {
 		return parsed, ErrUnableToReadResourceBytes
 	}
 
@@ -140,6 +141,13 @@ func (r *Parser) Parse(ctx context.Context, info *repository.FileInfo, validate
 		}
 	}
 
+	// Remove the internal UID,version and id if they exist
+	if parsed.GVK.Group == dashboard.GROUP && parsed.GVK.Kind == "Dashboard" {
+		unstructured.RemoveNestedField(parsed.Obj.Object, "spec", "uid")
+		unstructured.RemoveNestedField(parsed.Obj.Object, "spec", "version")
+		unstructured.RemoveNestedField(parsed.Obj.Object, "spec", "id") // now managed as a label
+	}
+
 	parsed.Meta, err = utils.MetaAccessor(parsed.Obj)
 	if err != nil {
 		return nil, err
@@ -160,13 +168,16 @@ func (r *Parser) Parse(ctx context.Context, info *repository.FileInfo, validate
 		Timestamp: nil, // ???&info.Modified.Time,
 	})
 
-	objName, folderName := NamesFromHashedRepoPath(cfg.Name, info.Path)
-	if obj.GetName() == "" {
-		obj.SetName(objName) // use the name saved in config
+	// Calculate name+folder from the file path
+	if info.Path != "" {
+		objName, folderName := NamesFromHashedRepoPath(cfg.Name, info.Path)
+		parsed.Meta.SetFolder(folderName)
+		if obj.GetName() == "" {
+			obj.SetName(objName) // use the name saved in config
+		}
 	}
 	obj.SetUID("")             // clear identifiers
 	obj.SetResourceVersion("") // clear identifiers
-	parsed.Meta.SetFolder(folderName)
 
 	// We can not do anything more if no kind is defined
 	if parsed.GVK == nil {