From b91b7d91f6ddea0fd5a56d579dd1fb39c489566f Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Fri, 21 Feb 2025 11:54:24 +0300 Subject: [PATCH] Provisioning: use the sync job to finish the migrate job (#101107) --- .../apis/provisioning/jobs/migrate/import.go | 183 ------------------ .../provisioning/jobs/migrate/resources.go | 18 +- .../apis/provisioning/jobs/migrate/worker.go | 27 ++- .../apis/provisioning/jobs/sync/worker.go | 18 +- pkg/registry/apis/provisioning/register.go | 26 +-- .../provisioning/repository/go-git/wrapper.go | 8 +- .../apis/provisioning/resources/parser.go | 21 +- 7 files changed, 83 insertions(+), 218 deletions(-) diff --git a/pkg/registry/apis/provisioning/jobs/migrate/import.go b/pkg/registry/apis/provisioning/jobs/migrate/import.go index 31583a7c4a1..2ee02b697ff 100644 --- a/pkg/registry/apis/provisioning/jobs/migrate/import.go +++ b/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 -} diff --git a/pkg/registry/apis/provisioning/jobs/migrate/resources.go b/pkg/registry/apis/provisioning/jobs/migrate/resources.go index 78fed3e4d71..a14db4e750e 100644 --- a/pkg/registry/apis/provisioning/jobs/migrate/resources.go +++ b/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", }} diff --git a/pkg/registry/apis/provisioning/jobs/migrate/worker.go b/pkg/registry/apis/provisioning/jobs/migrate/worker.go index 7cec1f7931d..1377f210e55 100644 --- a/pkg/registry/apis/provisioning/jobs/migrate/worker.go +++ b/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 diff --git a/pkg/registry/apis/provisioning/jobs/sync/worker.go b/pkg/registry/apis/provisioning/jobs/sync/worker.go index 5bc78e1f8b9..d0d88ebfa98 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/worker.go +++ b/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) diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index fb03b780566..3fc34851366 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/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) diff --git a/pkg/registry/apis/provisioning/repository/go-git/wrapper.go b/pkg/registry/apis/provisioning/repository/go-git/wrapper.go index c66a88dc54b..de8825347b7 100644 --- a/pkg/registry/apis/provisioning/repository/go-git/wrapper.go +++ b/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 } diff --git a/pkg/registry/apis/provisioning/resources/parser.go b/pkg/registry/apis/provisioning/resources/parser.go index 201ea39f0d3..19991185c32 100644 --- a/pkg/registry/apis/provisioning/resources/parser.go +++ b/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 {