[Provisioning] Use ProgressRecorder in Export Job (#100703)

* Use progress recorder instead
* Use recorder for folders
* More refactoring
* Move things to resources
* More situations
* Fix more TODOs
* Refactor progress to precalculate summary
* Do not store results
* Fix bug with dashboards error

pkg/registry/apis/provisioning/jobs/export/folders.go

@@ -11,22 +11,19 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	folders "github.com/grafana/grafana/pkg/apis/folder/v0alpha1"
-	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/repository"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
 	"github.com/grafana/grafana/pkg/storage/unified/parquet"
 	"github.com/grafana/grafana/pkg/storage/unified/resource"
 )
 
-var (
-	_ resource.BatchResourceWriter = (*folderReader)(nil)
-)
+var _ resource.BatchResourceWriter = (*folderReader)(nil)
 
 type folderReader struct {
 	tree           *resources.FolderTree
 	targetRepoName string
-	summary        *provisioning.JobResourceSummary
 }
 
 // Close implements resource.BatchResourceWriter.
@@ -44,33 +41,25 @@ func (f *folderReader) Write(ctx context.Context, key *resource.ResourceKey, val
 	item := &unstructured.Unstructured{}
 	err := item.UnmarshalJSON(value)
 	if err != nil {
-		return err
+		return fmt.Errorf("unmarshal unstructured to JSON: %w", err)
 	}
-	err = f.tree.AddUnstructured(item, f.targetRepoName)
-	if err != nil {
-		f.summary.Errors = append(f.summary.Errors, err.Error())
-	}
-	return nil
+
+	return f.tree.AddUnstructured(item, f.targetRepoName)
 }
 
+// FIXME: revise logging in this method
 func (r *exportJob) loadFolders(ctx context.Context) error {
 	logger := r.logger
-	status := r.jobStatus
-	status.Message = "reading folder tree"
-	r.maybeNotify(ctx)
+	r.progress.SetMessage("reading folder tree")
 
-	summary := r.getSummary(schema.GroupResource{
-		Group:    folders.GROUP,
-		Resource: folders.RESOURCE,
-	})
-
-	reader := &folderReader{
-		tree:           resources.NewEmptyFolderTree(),
-		targetRepoName: r.target.Config().Name,
-		summary:        summary,
-	}
+	repoName := r.target.Config().Name
 
 	if r.legacy != nil {
+		r.progress.SetMessage("migrate folder tree from legacy")
+		reader := &folderReader{
+			tree:           r.folderTree,
+			targetRepoName: repoName,
+		}
 		_, err := r.legacy.Migrate(ctx, legacy.MigrateOptions{
 			Namespace: r.namespace,
 			Resources: []schema.GroupResource{{
@@ -83,6 +72,8 @@ func (r *exportJob) loadFolders(ctx context.Context) error {
 			return fmt.Errorf("unable to read folders from legacy storage %w", err)
 		}
 	} else {
+		// TODO: should this be logging or message or both?
+		r.progress.SetMessage("read folder tree from unified storage")
 		client := r.client.Resource(schema.GroupVersionResource{
 			Group:    folders.GROUP,
 			Version:  folders.VERSION,
@@ -96,46 +87,63 @@ func (r *exportJob) loadFolders(ctx context.Context) error {
 		if rawList.GetContinue() != "" {
 			return fmt.Errorf("unable to list all folders in one request: %s", rawList.GetContinue())
 		}
+
 		for _, item := range rawList.Items {
-			err = reader.tree.AddUnstructured(&item, reader.targetRepoName)
+			err = r.folderTree.AddUnstructured(&item, repoName)
 			if err != nil {
-				summary.Errors = append(summary.Errors, err.Error())
+				r.progress.Record(ctx, jobs.JobResourceResult{
+					Name:     item.GetName(),
+					Resource: folders.RESOURCE,
+					Group:    folders.GROUP,
+					Error:    err,
+				})
 			}
 		}
 	}
 
-	// first create folders
-	// NOTE: this is required so that empty folders exist when finished
-	status.Message = "writing folders"
-	err := reader.tree.Walk(ctx, func(ctx context.Context, folder resources.Folder) error {
+	// create folders first is required so that empty folders exist when finished
+	r.progress.SetMessage("write folders")
+
+	err := r.folderTree.Walk(ctx, func(ctx context.Context, folder resources.Folder) error {
 		p := folder.Path + "/"
 		if r.prefix != "" {
 			p = r.prefix + "/" + p
 		}
 		logger := logger.With("path", p)
 
+		result := jobs.JobResourceResult{
+			Name:     folder.ID,
+			Resource: folders.RESOURCE,
+			Group:    folders.GROUP,
+			Path:     p,
+		}
+
 		_, err := r.target.Read(ctx, p, r.ref)
 		if err != nil && !(errors.Is(err, repository.ErrFileNotFound) || apierrors.IsNotFound(err)) {
-			logger.Error("failed to check if folder exists before writing", "error", err)
-			return fmt.Errorf("failed to check if folder exists before writing: %w", err)
+			result.Error = fmt.Errorf("failed to check if folder exists before writing: %w", err)
+			return result.Error
 		} else if err == nil {
 			logger.Info("folder already exists")
-			summary.Noop++
+			result.Action = repository.FileActionIgnored
+			r.progress.Record(ctx, result)
 			return nil
 		}
 
+		result.Action = repository.FileActionCreated
+		msg := fmt.Sprintf("export folder %s", p)
 		// Create with an empty body will make a folder (or .keep file if unsupported)
-		if err := r.target.Create(ctx, p, r.ref, nil, "export folder `"+p+"`"); err != nil {
-			logger.Error("failed to write a folder in repository", "error", err)
-			return fmt.Errorf("failed to write folder in repo: %w", err)
+		if err := r.target.Create(ctx, p, r.ref, nil, msg); err != nil {
+			result.Error = fmt.Errorf("failed to write folder in repo: %w", err)
+			r.progress.Record(ctx, result)
+			return result.Error
 		}
-		summary.Create++
-		logger.Debug("successfully exported folder")
+
+		r.progress.Record(ctx, result)
 		return nil
 	})
 	if err != nil {
 		return fmt.Errorf("failed to write folders: %w", err)
 	}
-	r.foldersTree = reader.tree
+
 	return nil
 }

pkg/registry/apis/provisioning/jobs/export/job.go

@@ -2,17 +2,10 @@ package export
 
 import (
 	"context"
-	"encoding/json"
-	"fmt"
-	"time"
-
-	"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"
 	provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
-	"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"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository"
@@ -28,165 +21,41 @@ type exportJob struct {
 	legacy    legacy.LegacyMigrator
 	namespace string
 
-	progress         jobs.ProgressFn
-	progressInterval time.Duration
-	progressLast     time.Time
-	foldersTree      *resources.FolderTree
-	userInfo         map[string]repository.CommitSignature
+	progress *jobs.JobProgressRecorder
+
+	userInfo   map[string]repository.CommitSignature
+	folderTree *resources.FolderTree
 
 	prefix         string // from options (now clean+safe)
 	ref            string // from options (only git)
 	keepIdentifier bool
 	withHistory    bool
-
-	jobStatus *provisioning.JobStatus
-	summary   map[string]*provisioning.JobResourceSummary
 }
 
 func newExportJob(ctx context.Context,
 	target repository.Repository,
 	options provisioning.ExportJobOptions,
 	client *resources.DynamicClient,
-	progress jobs.ProgressFn,
+	progress *jobs.JobProgressRecorder,
 ) *exportJob {
 	prefix := options.Prefix
 	if prefix != "" {
 		prefix = safepath.Clean(prefix)
 	}
 	return &exportJob{
-		namespace:        target.Config().Namespace,
-		target:           target,
-		client:           client,
-		logger:           logging.FromContext(ctx),
-		progress:         progress,
-		progressLast:     time.Now(),
-		progressInterval: time.Second * 5,
-
+		namespace:      target.Config().Namespace,
+		target:         target,
+		client:         client,
+		logger:         logging.FromContext(ctx),
+		progress:       progress,
 		prefix:         prefix,
 		ref:            options.Branch,
 		keepIdentifier: options.Identifier,
 		withHistory:    options.History,
-
-		jobStatus: &provisioning.JobStatus{
-			State: provisioning.JobStateWorking,
-		},
-		summary: make(map[string]*provisioning.JobResourceSummary),
+		folderTree:     resources.NewEmptyFolderTree(),
 	}
 }
 
-// Send progress messages to any listeners
-func (r *exportJob) maybeNotify(ctx context.Context) {
-	if time.Since(r.progressLast) > r.progressInterval {
-		r.progressLast = time.Now()
-		err := r.progress(ctx, *r.jobStatus)
-		if err != nil {
-			r.logger.Warn("unable to send progress", "err", err)
-		}
-	}
-}
-
-// Register summary information for a group/resource
-func (r *exportJob) getSummary(gr schema.GroupResource) *provisioning.JobResourceSummary {
-	summary, ok := r.summary[gr.String()]
-	if !ok {
-		summary = &provisioning.JobResourceSummary{
-			Group:    gr.Group,
-			Resource: gr.Resource,
-		}
-		r.summary[gr.String()] = summary
-		r.jobStatus.Summary = append(r.jobStatus.Summary, summary)
-	}
-	return summary
-}
-
-func (r *exportJob) add(ctx context.Context, summary *provisioning.JobResourceSummary, obj *unstructured.Unstructured) error {
-	if err := ctx.Err(); err != nil {
-		return err
-	}
-	r.maybeNotify(ctx)
-
-	item, err := utils.MetaAccessor(obj)
-	if err != nil {
-		return err
-	}
-
-	// Message from annotations
-	commitMessage := item.GetMessage()
-	if commitMessage == "" {
-		g := item.GetGeneration()
-		if g > 0 {
-			commitMessage = fmt.Sprintf("Generation: %d", g)
-		} else {
-			commitMessage = "exported from grafana"
-		}
-	}
-
-	name := item.GetName()
-	repoName := item.GetRepositoryName()
-	if repoName == r.target.Config().GetName() {
-		r.logger.Info("skip dashboard since it is already in repository", "dashboard", name)
-		return nil
-	}
-
-	title := item.FindTitle("")
-	if title == "" {
-		title = name
-	}
-	folder := item.GetFolder()
-
-	// Add the author in context (if available)
-	ctx = r.withAuthorSignature(ctx, item)
-
-	// Get the absolute path of the folder
-	fid, ok := r.foldersTree.DirPath(folder, "")
-	if !ok {
-		fid = resources.Folder{
-			Path: "__folder_not_found/" + slugify.Slugify(folder),
-		}
-		r.logger.Error("folder of item was not in tree of repository")
-	}
-
-	// Clear the metadata
-	delete(obj.Object, "metadata")
-
-	if r.keepIdentifier {
-		item.SetName(name) // keep the identifier in the metadata
-	}
-
-	body, err := json.MarshalIndent(obj.Object, "", "  ")
-	if err != nil {
-		return fmt.Errorf("failed to marshal dashboard %s: %w", name, err)
-	}
-
-	fileName := slugify.Slugify(title) + ".json"
-	if fid.Path != "" {
-		fileName, err = safepath.Join(fid.Path, fileName)
-		if err != nil {
-			return fmt.Errorf("error adding file path %s: %w", title, err)
-		}
-	}
-	if r.prefix != "" {
-		fileName, err = safepath.Join(r.prefix, fileName)
-		if err != nil {
-			return fmt.Errorf("error adding path prefix %s: %w", r.prefix, err)
-		}
-	}
-
-	// Write the file
-	err = r.target.Write(ctx, fileName, r.ref, body, commitMessage)
-	if err != nil {
-		summary.Error++
-		r.logger.Error("failed to write a file in repository", "error", err)
-		if len(summary.Errors) < 20 {
-			summary.Errors = append(summary.Errors, fmt.Sprintf("error writing: %s", fileName))
-		}
-	} else {
-		summary.Write++
-	}
-
-	return nil
-}
-
 func (r *exportJob) withAuthorSignature(ctx context.Context, item utils.GrafanaMetaAccessor) context.Context {
 	if r.userInfo == nil {
 		return ctx
@@ -209,5 +78,6 @@ func (r *exportJob) withAuthorSignature(ctx context.Context, item utils.GrafanaM
 	} else {
 		sig.When = item.GetCreationTimestamp().Time
 	}
+
 	return repository.WithAuthorSignature(ctx, sig)
 }

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

@@ -2,28 +2,29 @@ package export
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 
 	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"
 	dashboards "github.com/grafana/grafana/pkg/apis/dashboard"
-	provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
+	"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"
+	"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/unified/parquet"
 	"github.com/grafana/grafana/pkg/storage/unified/resource"
 )
 
-var (
-	_ resource.BatchResourceWriter = (*resourceReader)(nil)
-)
+var _ resource.BatchResourceWriter = (*resourceReader)(nil)
 
 type resourceReader struct {
-	job     *exportJob
-	summary *provisioning.JobResourceSummary
-	logger  logging.Logger
+	job *exportJob
 }
 
 // Close implements resource.BatchResourceWriter.
@@ -41,16 +42,17 @@ func (f *resourceReader) Write(ctx context.Context, key *resource.ResourceKey, v
 	item := &unstructured.Unstructured{}
 	err := item.UnmarshalJSON(value)
 	if err != nil {
-		return err
+		// TODO: should we fail the entire execution?
+		return fmt.Errorf("failed to unmarshal unstructured: %w", err)
 	}
-	err = f.job.add(ctx, f.summary, item)
-	if err != nil {
-		f.logger.Warn("error adding from legacy", "name", key.Name, "err", err)
-		f.summary.Errors = append(f.summary.Errors, fmt.Sprintf("%s: %s", key.Name, err.Error()))
-		if len(f.summary.Errors) > 50 {
-			return err
+
+	if result := f.job.write(ctx, item); result.Error != nil {
+		f.job.progress.Record(ctx, result)
+		if len(f.job.progress.Errors()) > 20 {
+			return fmt.Errorf("stopping execution due to too many errors")
 		}
 	}
+
 	return nil
 }
 
@@ -62,32 +64,30 @@ func (r *exportJob) loadResources(ctx context.Context) error {
 	}}
 
 	for _, kind := range kinds {
-		r.jobStatus.Message = "Exporting " + kind.Resource + "..."
+		r.progress.SetMessage(fmt.Sprintf("exporting %s resource", kind.Resource))
 		if r.legacy != nil {
+			r.progress.SetMessage(fmt.Sprintf("migrate %s resource", kind.Resource))
 			gr := kind.GroupResource()
-			reader := &resourceReader{
-				summary: r.getSummary(gr),
-				job:     r,
-				logger:  r.logger,
-			}
 			opts := legacy.MigrateOptions{
 				Namespace:   r.namespace,
 				WithHistory: r.withHistory,
 				Resources:   []schema.GroupResource{gr},
-				Store:       parquet.NewBatchResourceWriterClient(reader),
+				Store:       parquet.NewBatchResourceWriterClient(&resourceReader{job: r}),
 				OnlyCount:   true, // first get the count
 			}
 			stats, err := r.legacy.Migrate(ctx, opts)
 			if err != nil {
 				return fmt.Errorf("unable to count legacy items %w", err)
 			}
+
+			// FIXME: explain why we calculate it in this way
 			if len(stats.Summary) > 0 {
 				count := stats.Summary[0].Count
 				history := stats.Summary[0].History
 				if history > count {
 					count = history // the number of items we will process
 				}
-				reader.summary.Total = count
+				r.progress.SetTotal(int(count))
 			}
 
 			opts.OnlyCount = false // this time actually write
@@ -97,6 +97,7 @@ func (r *exportJob) loadResources(ctx context.Context) error {
 			}
 		}
 
+		r.progress.SetMessage(fmt.Sprintf("reading %s resource", kind.Resource))
 		if err := r.loadResourcesFromAPIServer(ctx, kind); err != nil {
 			return fmt.Errorf("error loading %s %w", kind.Resource, err)
 		}
@@ -105,11 +106,9 @@ func (r *exportJob) loadResources(ctx context.Context) error {
 }
 
 func (r *exportJob) loadResourcesFromAPIServer(ctx context.Context, kind schema.GroupVersionResource) error {
-	r.maybeNotify(ctx)
 	client := r.client.Resource(kind)
-	summary := r.getSummary(kind.GroupResource())
 
-	continueToken := ""
+	var continueToken string
 	for {
 		list, err := client.List(ctx, metav1.ListOptions{Limit: 100, Continue: continueToken})
 		if err != nil {
@@ -117,8 +116,9 @@ func (r *exportJob) loadResourcesFromAPIServer(ctx context.Context, kind schema.
 		}
 
 		for _, item := range list.Items {
-			if err = r.add(ctx, summary, &item); err != nil {
-				return fmt.Errorf("error adding value: %w", err)
+			r.progress.Record(ctx, r.write(ctx, &item))
+			if len(r.progress.Errors()) > 20 {
+				return fmt.Errorf("stopping execution due to too many errors")
 			}
 		}
 
@@ -130,3 +130,99 @@ func (r *exportJob) loadResourcesFromAPIServer(ctx context.Context, kind schema.
 
 	return nil
 }
+
+func (r *exportJob) write(ctx context.Context, obj *unstructured.Unstructured) jobs.JobResourceResult {
+	gvk := obj.GroupVersionKind()
+	result := jobs.JobResourceResult{
+		Name:     obj.GetName(),
+		Resource: gvk.Kind,
+		Group:    gvk.Group,
+		Action:   repository.FileActionCreated,
+	}
+
+	if err := ctx.Err(); err != nil {
+		result.Error = fmt.Errorf("context error: %w", err)
+		return result
+	}
+
+	meta, err := utils.MetaAccessor(obj)
+	if err != nil {
+		result.Error = fmt.Errorf("extract meta accessor: %w", err)
+		return result
+	}
+
+	// Message from annotations
+	commitMessage := meta.GetMessage()
+	if commitMessage == "" {
+		g := meta.GetGeneration()
+		if g > 0 {
+			commitMessage = fmt.Sprintf("Generation: %d", g)
+		} else {
+			commitMessage = "exported from grafana"
+		}
+	}
+
+	name := meta.GetName()
+	repoName := meta.GetRepositoryName()
+	if repoName == r.target.Config().GetName() {
+		result.Action = repository.FileActionIgnored
+		return result
+	}
+
+	title := meta.FindTitle("")
+	if title == "" {
+		title = name
+	}
+	folder := meta.GetFolder()
+
+	// Add the author in context (if available)
+	ctx = r.withAuthorSignature(ctx, meta)
+
+	// Get the absolute path of the folder
+	fid, ok := r.folderTree.DirPath(folder, "")
+	if !ok {
+		// FIXME: Shouldn't this fail instead?
+		fid = resources.Folder{
+			Path: "__folder_not_found/" + slugify.Slugify(folder),
+		}
+		r.logger.Error("folder of item was not in tree of repository")
+	}
+
+	result.Path = fid.Path
+
+	// Clear the metadata
+	delete(obj.Object, "metadata")
+
+	if r.keepIdentifier {
+		meta.SetName(name) // keep the identifier in the metadata
+	}
+
+	body, err := json.MarshalIndent(obj.Object, "", "  ")
+	if err != nil {
+		result.Error = fmt.Errorf("failed to marshal dashboard: %w", err)
+		return result
+	}
+
+	fileName := slugify.Slugify(title) + ".json"
+	if fid.Path != "" {
+		fileName, err = safepath.Join(fid.Path, fileName)
+		if err != nil {
+			result.Error = fmt.Errorf("error adding file path: %w", err)
+			return result
+		}
+	}
+	if r.prefix != "" {
+		fileName, err = safepath.Join(r.prefix, fileName)
+		if err != nil {
+			result.Error = fmt.Errorf("error adding path prefix: %w", err)
+			return result
+		}
+	}
+
+	err = r.target.Write(ctx, fileName, r.ref, body, commitMessage)
+	if err != nil {
+		result.Error = fmt.Errorf("failed to write file: %w", err)
+	}
+
+	return result
+}

pkg/registry/apis/provisioning/jobs/export/users.go

@@ -14,10 +14,6 @@ import (
 )
 
 func (r *exportJob) loadUsers(ctx context.Context) error {
-	status := r.jobStatus
-	status.Message = "reading user info"
-	r.maybeNotify(ctx)
-
 	client := r.client.Resource(schema.GroupVersionResource{
 		Group:    iam.GROUP,
 		Version:  iam.VERSION,
@@ -36,6 +32,7 @@ func (r *exportJob) loadUsers(ctx context.Context) error {
 	r.userInfo = make(map[string]repository.CommitSignature)
 	for _, item := range rawList.Items {
 		sig := repository.CommitSignature{}
+		// FIXME: should we improve logging here?
 		sig.Name, ok, err = unstructured.NestedString(item.Object, "spec", "login")
 		if !ok || err != nil {
 			continue

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

@@ -51,58 +51,69 @@ func (r *ExportWorker) IsSupported(ctx context.Context, job provisioning.Job) bo
 }
 
 // Process will start a job
-func (r *ExportWorker) Process(ctx context.Context, repo repository.Repository, job provisioning.Job, progress jobs.ProgressFn) (*provisioning.JobStatus, error) {
+func (r *ExportWorker) Process(ctx context.Context, repo repository.Repository, job provisioning.Job, progressFn jobs.ProgressFn) (*provisioning.JobStatus, error) {
 	if repo.Config().Spec.ReadOnly {
 		return &provisioning.JobStatus{
-			State:  provisioning.JobStateError,
-			Errors: []string{"Exporting to a read only repository is not supported"},
+			State:   provisioning.JobStateError,
+			Message: "Exporting to a read only repository is not supported",
 		}, nil
 	}
 
 	options := job.Spec.Export
 	if options == nil {
 		return &provisioning.JobStatus{
-			State:  provisioning.JobStateError,
-			Errors: []string{"Export job missing export settings"},
+			State:   provisioning.JobStateError,
+			Message: "Export job missing export settings",
 		}, nil
 	}
 
-	var err error
-	var buffered *gogit.GoGitRepo
+	progress := jobs.NewJobProgressRecorder(progressFn)
+	var (
+		err      error
+		buffered *gogit.GoGitRepo
+	)
+
 	if repo.Config().Spec.GitHub != nil {
+		progress.SetMessage("clone target")
 		buffered, err = gogit.Clone(ctx, repo.Config(), gogit.GoGitCloneOptions{
 			Root:                   r.clonedir,
 			SingleCommitBeforePush: !options.History,
 		}, r.secrets, os.Stdout)
 		if err != nil {
 			return &provisioning.JobStatus{
-				State:  provisioning.JobStateError,
-				Errors: []string{"Unable to clone target", err.Error()},
+				State:   provisioning.JobStateError,
+				Message: "Unable to clone target",
+				Errors:  []string{err.Error()},
 			}, nil
 		}
 
 		// New empty branch (same on main???)
 		if options.Branch != "" {
+			progress.SetMessage("create empty branch")
 			_, err := buffered.NewEmptyBranch(ctx, options.Branch)
 			if err != nil {
 				return &provisioning.JobStatus{
-					State:  provisioning.JobStateError,
-					Errors: []string{"Unable to create empty branch", err.Error()},
+					State:   provisioning.JobStateError,
+					Message: "Unable to create empty branch",
+					Errors:  []string{err.Error()},
 				}, nil
 			}
 		}
+
 		repo = buffered     // send all writes to the buffered repo
 		options.Branch = "" // :( the branch is now baked into the repo
 	}
 
 	dynamicClient, _, err := r.clients.New(repo.Config().Namespace)
 	if err != nil {
+		// TODO: how do we really want to return errors?
 		return nil, fmt.Errorf("error getting client %w", err)
 	}
 
 	worker := newExportJob(ctx, repo, *options, dynamicClient, progress)
 
 	if options.History {
+		progress.SetMessage("load users")
 		err = worker.loadUsers(ctx)
 		if err != nil {
 			return nil, fmt.Errorf("error loading users %w", err)
@@ -115,28 +126,27 @@ func (r *ExportWorker) Process(ctx context.Context, repo repository.Repository,
 	}
 
 	// Load and write all folders
+	progress.SetMessage("start folder export")
 	err = worker.loadFolders(ctx)
 	if err != nil {
-		return worker.jobStatus, err
+		// TODO: handle better
+		return progress.Complete(ctx, err), err
 	}
 
+	progress.SetMessage("start resource export")
 	err = worker.loadResources(ctx)
 	if err != nil {
-		return worker.jobStatus, err
+		// TODO: handle better
+		return progress.Complete(ctx, err), err
 	}
 
-	status := worker.jobStatus
-	if buffered != nil && status.State != provisioning.JobStateError {
-		status.Message = "pushing changes..."
-		worker.maybeNotify(ctx) // force notify?
-		err = buffered.Push(ctx, os.Stdout)
-		status.Message = ""
+	// TODO: handle the errors properly
+	if buffered != nil {
+		progress.SetMessage("push changes")
+		if err := buffered.Push(ctx, os.Stdout); err != nil {
+			return progress.Complete(ctx, err), err
+		}
 	}
 
-	// Add summary info to response
-	if !status.State.Finished() && err == nil {
-		status.State = provisioning.JobStateSuccess
-		status.Message = ""
-	}
-	return status, err
+	return progress.Complete(ctx, nil), err
 }

pkg/registry/apis/provisioning/jobs/progress.go

@@ -24,8 +24,7 @@ func MaybeNotifyProgress(threshold time.Duration, fn ProgressFn) ProgressFn {
 	}
 }
 
-// FIXME: ProgressRecorder should be moved to jobs package and initialized in the queue
-
+// FIXME: ProgressRecorder should be initialized in the queue
 type JobResourceResult struct {
 	Name     string
 	Resource string
@@ -36,32 +35,34 @@ type JobResourceResult struct {
 }
 
 type JobProgressRecorder struct {
-	total      int
-	ref        string
-	message    string
-	results    []JobResourceResult
-	errors     []string
-	progressFn ProgressFn
+	total       int
+	ref         string
+	message     string
+	resultCount int
+	errors      []string
+	progressFn  ProgressFn
+	summaries   map[string]*provisioning.JobResourceSummary
 }
 
 func NewJobProgressRecorder(progressFn ProgressFn) *JobProgressRecorder {
 	return &JobProgressRecorder{
 		progressFn: MaybeNotifyProgress(15*time.Second, progressFn),
+		summaries:  make(map[string]*provisioning.JobResourceSummary),
 	}
 }
 
 func (r *JobProgressRecorder) Record(ctx context.Context, result JobResourceResult) {
-	if r.results == nil {
-		r.results = make([]JobResourceResult, 0)
-	}
-	r.results = append(r.results, result)
+	r.resultCount++
 
 	if result.Error != nil {
 		logger := logging.FromContext(ctx)
 		logger.Error("job resource operation failed", "err", result.Error, "path", result.Path, "resource", result.Resource, "group", result.Group, "action", result.Action, "name", result.Name)
-		r.errors = append(r.errors, result.Error.Error())
+		if len(r.errors) < 20 {
+			r.errors = append(r.errors, result.Error.Error())
+		}
 	}
 
+	r.updateSummary(result)
 	r.notify(ctx)
 }
 
@@ -90,71 +91,53 @@ func (r *JobProgressRecorder) Errors() []string {
 }
 
 func (r *JobProgressRecorder) summary() []*provisioning.JobResourceSummary {
-	if len(r.results) == 0 {
+	if len(r.summaries) == 0 {
 		return nil
 	}
 
-	// Group results by resource+group
-	groupedResults := make(map[string][]JobResourceResult)
-	for _, result := range r.results {
-		key := result.Resource + ":" + result.Group
-		groupedResults[key] = append(groupedResults[key], result)
-	}
-
-	summaries := make([]*provisioning.JobResourceSummary, 0)
-	for _, results := range groupedResults {
-		if len(results) == 0 {
-			continue
-		}
-
-		// Count actions
-		actions := make(map[repository.FileAction]int64)
-		var errors []string
-		for _, result := range results {
-			if result.Error != nil {
-				errors = append(errors, result.Error.Error())
-			} else {
-				actions[result.Action]++
-			}
-		}
-
-		// Create summary for this group
-
-		// Default to unknown if resource or group is empty
-		resource := results[0].Resource
-		if resource == "" {
-			resource = "unknown"
-		}
-
-		group := results[0].Group
-		if group == "" {
-			group = "unknown"
-		}
-
-		summary := &provisioning.JobResourceSummary{
-			Resource: resource,
-			Group:    group,
-			Delete:   actions[repository.FileActionDeleted],
-			Update:   actions[repository.FileActionUpdated],
-			Create:   actions[repository.FileActionCreated],
-			Write:    actions[repository.FileActionCreated] + actions[repository.FileActionUpdated],
-			Error:    int64(len(errors)),
-			Noop:     actions[repository.FileActionIgnored],
-			Errors:   errors,
-		}
-
+	summaries := make([]*provisioning.JobResourceSummary, 0, len(r.summaries))
+	for _, summary := range r.summaries {
 		summaries = append(summaries, summary)
 	}
 
 	return summaries
 }
 
+func (r *JobProgressRecorder) updateSummary(result JobResourceResult) {
+	key := result.Resource + ":" + result.Group
+	summary, exists := r.summaries[key]
+	if !exists {
+		summary = &provisioning.JobResourceSummary{
+			Resource: result.Resource,
+			Group:    result.Group,
+		}
+		r.summaries[key] = summary
+	}
+
+	if result.Error != nil {
+		summary.Errors = append(summary.Errors, result.Error.Error())
+		summary.Error++
+	} else {
+		switch result.Action {
+		case repository.FileActionDeleted:
+			summary.Delete++
+		case repository.FileActionUpdated:
+			summary.Update++
+		case repository.FileActionCreated:
+			summary.Create++
+		case repository.FileActionIgnored:
+			summary.Noop++
+		}
+		summary.Write = summary.Create + summary.Update
+	}
+}
+
 func (r *JobProgressRecorder) progress() float64 {
 	if r.total == 0 {
 		return 0
 	}
 
-	return float64(r.total - len(r.results)/r.total*100)
+	return float64(r.resultCount) / float64(r.total) * 100
 }
 
 func (r *JobProgressRecorder) notify(ctx context.Context) {

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

@@ -2,6 +2,7 @@ package resources
 
 import (
 	"context"
+	"fmt"
 	"path"
 	"sort"
 	"strings"
@@ -101,7 +102,7 @@ func NewEmptyFolderTree() *FolderTree {
 func (t *FolderTree) AddUnstructured(item *unstructured.Unstructured, skipRepo string) error {
 	meta, err := utils.MetaAccessor(item)
 	if err != nil {
-		return err
+		return fmt.Errorf("extract meta accessor: %w", err)
 	}
 	if meta.GetRepositoryName() == skipRepo {
 		return nil // skip it... already in tree?