package move import ( "context" "errors" "fmt" "path/filepath" "time" "k8s.io/apimachinery/pkg/runtime/schema" "github.com/grafana/grafana-app-sdk/logging" provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" "github.com/grafana/grafana/apps/provisioning/pkg/repository" "github.com/grafana/grafana/apps/provisioning/pkg/safepath" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/grafana/grafana/pkg/registry/apis/provisioning/utils" ) type Worker struct { syncWorker jobs.Worker wrapFn repository.WrapWithStageFn resourcesFactory resources.RepositoryResourcesFactory metrics jobs.JobMetrics } func NewWorker(syncWorker jobs.Worker, wrapFn repository.WrapWithStageFn, resourcesFactory resources.RepositoryResourcesFactory, metrics jobs.JobMetrics) *Worker { return &Worker{ syncWorker: syncWorker, wrapFn: wrapFn, resourcesFactory: resourcesFactory, metrics: metrics, } } func (w *Worker) IsSupported(ctx context.Context, job provisioning.Job) bool { return job.Spec.Action == provisioning.JobActionMove } func (w *Worker) Process(ctx context.Context, repo repository.Repository, job provisioning.Job, progress jobs.JobProgressRecorder) error { if job.Spec.Move == nil { return errors.New("missing move settings") } opts := *job.Spec.Move logger := logging.FromContext(ctx).With("job", job.GetName(), "namespace", job.GetNamespace()) outcome := utils.ErrorOutcome start := time.Now() resourcesMoved := 0 defer func() { w.metrics.RecordJob(string(provisioning.JobActionMove), outcome, resourcesMoved, time.Since(start).Seconds()) }() if opts.TargetPath == "" { return errors.New("target path is required for move operation") } // Validate that target path is a directory (ends with slash) if !safepath.IsDir(opts.TargetPath) { return errors.New("target path must be a directory (should end with '/')") } paths := opts.Paths progress.SetTotal(ctx, len(paths)+len(opts.Resources)) progress.StrictMaxErrors(1) // Fail fast on any error during move fn := func(repo repository.Repository, _ bool) error { rw, ok := repo.(repository.ReaderWriter) if !ok { logger.Error("move job submitted targeting repository that is not a ReaderWriter") return errors.New("move job submitted targeting repository that is not a ReaderWriter") } // Resolve ResourceRef entries to file paths using RepositoryResources if len(opts.Resources) > 0 { resolvedPaths, err := w.resolveResourcesToPaths(ctx, rw, progress, opts.Resources) if err != nil { return err } paths = append(paths, resolvedPaths...) } // Deduplicate paths to avoid attempting to move the same file multiple times paths = deduplicatePaths(paths) return w.moveFiles(ctx, rw, progress, opts, paths...) } msg := fmt.Sprintf("Move files from Grafana %s", job.Name) stageOptions := repository.StageOptions{ Mode: repository.StageModeCommitOnlyOnce, CommitOnlyOnceMessage: msg, PushOnWrites: false, Timeout: 10 * time.Minute, Ref: opts.Ref, } err := w.wrapFn(ctx, repo, stageOptions, fn) if err != nil { logger.Error("failed to move files in repository", "error", err) return fmt.Errorf("move files in repository: %w", err) } // Set RefURLs if the repository supports it and we have a target ref if opts.Ref != "" { if repoWithURLs, ok := repo.(repository.RepositoryWithURLs); ok { if refURLs, urlErr := repoWithURLs.RefURLs(ctx, opts.Ref); urlErr == nil && refURLs != nil { progress.SetRefURLs(ctx, refURLs) } } } if opts.Ref == "" { progress.ResetResults() progress.SetMessage(ctx, "pull resources") syncJob := provisioning.Job{ Spec: provisioning.JobSpec{ Pull: &provisioning.SyncJobOptions{ // Full sync because it's the only one that supports empty folder deletion Incremental: false, }, }, } if err := w.syncWorker.Process(ctx, repo, syncJob, progress); err != nil { logger.Error("failed to pull resources", "error", err) return fmt.Errorf("pull resources: %w", err) } } outcome = utils.SuccessOutcome jobStatus := progress.Complete(ctx, nil) for _, summary := range jobStatus.Summary { // FileActionRenamed increments both delete & create, use create here resourcesMoved += int(summary.Create) } return nil } func (w *Worker) moveFiles(ctx context.Context, rw repository.ReaderWriter, progress jobs.JobProgressRecorder, opts provisioning.MoveJobOptions, paths ...string) error { for _, path := range paths { result := jobs.JobResourceResult{ Path: path, Action: repository.FileActionRenamed, } // Construct the target path by combining the job's target path with the file/folder name targetPath := w.constructTargetPath(opts.TargetPath, path) progress.SetMessage(ctx, "Moving "+path+" to "+targetPath) if err := rw.Move(ctx, path, targetPath, opts.Ref, "Move "+path+" to "+targetPath); err != nil { result.Error = fmt.Errorf("moving file %s to %s: %w", path, targetPath, err) } progress.Record(ctx, result) if err := progress.TooManyErrors(); err != nil { return err } } return nil } // constructTargetPath combines the job's target path with the file/folder name from the source path func (w *Worker) constructTargetPath(jobTargetPath, sourcePath string) string { // Extract the file/folder name from the source path fileName := filepath.Base(sourcePath) // If the source path is a directory (ends with slash), preserve the trailing slash in target if safepath.IsDir(sourcePath) { return jobTargetPath + fileName + "/" } // For files, just append the filename return jobTargetPath + fileName } // resolveResourcesToPaths converts ResourceRef entries to file paths, recording errors for individual resources func (w *Worker) resolveResourcesToPaths(ctx context.Context, rw repository.ReaderWriter, progress jobs.JobProgressRecorder, resources []provisioning.ResourceRef) ([]string, error) { if len(resources) == 0 { return nil, nil } progress.SetMessage(ctx, "Resolving resource paths") repositoryResources, err := w.resourcesFactory.Client(ctx, rw) if err != nil { return nil, fmt.Errorf("create repository resources client: %w", err) } resolvedPaths := make([]string, 0, len(resources)) for _, resource := range resources { result := jobs.JobResourceResult{ Name: resource.Name, Group: resource.Group, Action: repository.FileActionRenamed, // Will be used for move later } gvk := schema.GroupVersionKind{ Group: resource.Group, Kind: resource.Kind, // Version is left empty so ForKind will use the preferred version } progress.SetMessage(ctx, fmt.Sprintf("Finding path for resource %s/%s/%s", resource.Group, resource.Kind, resource.Name)) resourcePath, err := repositoryResources.FindResourcePath(ctx, resource.Name, gvk) if err != nil { result.Error = fmt.Errorf("find path for resource %s/%s/%s: %w", resource.Group, resource.Kind, resource.Name, err) progress.Record(ctx, result) // Continue with next resource instead of failing fast if err := progress.TooManyErrors(); err != nil { return resolvedPaths, err } continue } result.Path = resourcePath resolvedPaths = append(resolvedPaths, resourcePath) } return resolvedPaths, nil } // deduplicatePaths removes duplicate file paths from the slice while preserving order func deduplicatePaths(paths []string) []string { if len(paths) <= 1 { return paths } seen := make(map[string]struct{}, len(paths)) result := make([]string, 0, len(paths)) for _, path := range paths { if _, exists := seen[path]; !exists { seen[path] = struct{}{} result = append(result, path) } } return result }