From e2e93bed67707bd977f2a1a7d49f8641cedabd1c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roberto=20Jim=C3=A9nez=20S=C3=A1nchez?= Date: Fri, 14 Feb 2025 11:07:34 +0100 Subject: [PATCH] [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 --- .../apis/provisioning/jobs/export/folders.go | 86 +++++----- .../apis/provisioning/jobs/export/job.go | 154 ++---------------- .../provisioning/jobs/export/resources.go | 152 +++++++++++++---- .../apis/provisioning/jobs/export/users.go | 5 +- .../apis/provisioning/jobs/export/worker.go | 60 ++++--- .../apis/provisioning/jobs/progress.go | 111 ++++++------- .../apis/provisioning/resources/tree.go | 3 +- 7 files changed, 268 insertions(+), 303 deletions(-) diff --git a/pkg/registry/apis/provisioning/jobs/export/folders.go b/pkg/registry/apis/provisioning/jobs/export/folders.go index c47e667e130..fea92b25f23 100644 --- a/pkg/registry/apis/provisioning/jobs/export/folders.go +++ b/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 } diff --git a/pkg/registry/apis/provisioning/jobs/export/job.go b/pkg/registry/apis/provisioning/jobs/export/job.go index abe1e2cad54..b36236e1690 100644 --- a/pkg/registry/apis/provisioning/jobs/export/job.go +++ b/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) } diff --git a/pkg/registry/apis/provisioning/jobs/export/resources.go b/pkg/registry/apis/provisioning/jobs/export/resources.go index a332b6ece41..9c182ccfd48 100644 --- a/pkg/registry/apis/provisioning/jobs/export/resources.go +++ b/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 +} diff --git a/pkg/registry/apis/provisioning/jobs/export/users.go b/pkg/registry/apis/provisioning/jobs/export/users.go index 32548a172c4..5e717f5404b 100644 --- a/pkg/registry/apis/provisioning/jobs/export/users.go +++ b/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 diff --git a/pkg/registry/apis/provisioning/jobs/export/worker.go b/pkg/registry/apis/provisioning/jobs/export/worker.go index bc27197b907..4e991136cd0 100644 --- a/pkg/registry/apis/provisioning/jobs/export/worker.go +++ b/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 } diff --git a/pkg/registry/apis/provisioning/jobs/progress.go b/pkg/registry/apis/provisioning/jobs/progress.go index 3ba89f5335d..c6194ea8a6c 100644 --- a/pkg/registry/apis/provisioning/jobs/progress.go +++ b/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) { diff --git a/pkg/registry/apis/provisioning/resources/tree.go b/pkg/registry/apis/provisioning/resources/tree.go index 11e48fe4b15..b76c0886c77 100644 --- a/pkg/registry/apis/provisioning/resources/tree.go +++ b/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?