From 32e9adaaafc38b1b44ac6e8677dc4df3efe4b78b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roberto=20Jim=C3=A9nez=20S=C3=A1nchez?= Date: Wed, 18 Dec 2024 12:41:50 +0100 Subject: [PATCH] Move sync status update from replicator to worker (#98175) * Move sync status out of replicator * Comment out JobID as it triggers the infinite loop * Add started and finished timestamps --- pkg/registry/apis/provisioning/jobs/queue.go | 4 ++ pkg/registry/apis/provisioning/jobs/worker.go | 40 +++++++++++++++-- .../apis/provisioning/resources/replicator.go | 43 ++++--------------- 3 files changed, 50 insertions(+), 37 deletions(-) diff --git a/pkg/registry/apis/provisioning/jobs/queue.go b/pkg/registry/apis/provisioning/jobs/queue.go index a4c45606381..4af483c5cbf 100644 --- a/pkg/registry/apis/provisioning/jobs/queue.go +++ b/pkg/registry/apis/provisioning/jobs/queue.go @@ -119,6 +119,7 @@ func (s *jobStore) drainPending() { } logger := s.logger.With("job", job.GetName(), "namespace", job.GetNamespace()) + started := time.Now() var status *provisioning.JobStatus if s.worker == nil { status = &provisioning.JobStatus{ @@ -141,6 +142,9 @@ func (s *jobStore) drainPending() { logger.DebugContext(ctx, "job processing finished", "status", status.State) } + status.Started = started.UnixNano() + status.Finished = time.Now().UnixNano() + err = s.Complete(ctx, job.Namespace, job.Name, *status) if err != nil { logger.ErrorContext(ctx, "error running job", "error", err) diff --git a/pkg/registry/apis/provisioning/jobs/worker.go b/pkg/registry/apis/provisioning/jobs/worker.go index e41ffd06239..ddd6485413b 100644 --- a/pkg/registry/apis/provisioning/jobs/worker.go +++ b/pkg/registry/apis/provisioning/jobs/worker.go @@ -2,10 +2,13 @@ package jobs import ( "context" + "encoding/json" "fmt" "log/slog" "net/url" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apiserver/pkg/endpoints/request" "github.com/grafana/grafana/pkg/apimachinery/identity" @@ -87,9 +90,39 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job) (*provisi switch job.Spec.Action { case provisioning.JobActionSync: - err := replicator.Sync(ctx) - if err != nil { - return nil, err + ref, syncError := replicator.Sync(ctx) + status := &provisioning.SyncStatus{ + State: provisioning.JobStateFinished, + // FIXME: infinite loop if we set this in the controller + // JobID: job.GetName(), // TODO: Should we use the job id here? + Hash: ref, + Started: 0, // FIXME: infinite loop if we set this in the controller + Finished: 0, // FIXME: infinite loop if we set this in the controller + Message: []string{}, + } + + if syncError != nil { + status.State = provisioning.JobStateError + status.Message = append(status.Message, syncError.Error()) + } + + cfg := repo.Config().DeepCopy() + cfg.Status.Sync = *status + + // TODO: Can we use typed client for this? + client := parser.Client().Resource(provisioning.RepositoryResourceInfo.GroupVersionResource()) + unstructuredResource := &unstructured.Unstructured{} + jj, _ := json.Marshal(cfg) + if err := json.Unmarshal(jj, &unstructuredResource.Object); err != nil { + return nil, fmt.Errorf("error loading config json: %w", err) + } + + if _, err := client.UpdateStatus(ctx, unstructuredResource, v1.UpdateOptions{}); err != nil { + return nil, fmt.Errorf("update repository status: %w", err) + } + + if syncError != nil { + return nil, syncError } case provisioning.JobActionPullRequest: prRepo, ok := repo.(PullRequestRepo) @@ -124,6 +157,7 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job) (*provisi return nil, err } default: + return nil, fmt.Errorf("unknown job action: %s", job.Spec.Action) } return &provisioning.JobStatus{ diff --git a/pkg/registry/apis/provisioning/resources/replicator.go b/pkg/registry/apis/provisioning/resources/replicator.go index 9cf954ac5be..395e1f158b9 100644 --- a/pkg/registry/apis/provisioning/resources/replicator.go +++ b/pkg/registry/apis/provisioning/resources/replicator.go @@ -53,16 +53,14 @@ func NewReplicator( } // Sync replicates all files in the repository. -func (r *Replicator) Sync(ctx context.Context) error { +func (r *Replicator) Sync(ctx context.Context) (string, error) { // FIXME: how to handle the scenario in which folder changes? cfg := r.repository.Config() lastCommit := cfg.Status.Sync.Hash versionedRepo, isVersioned := r.repository.(repository.VersionedRepository) - // started := time.Now() - if err := r.ensureRepositoryFolderExists(ctx); err != nil { - return fmt.Errorf("ensure repository folder exists: %w", err) + return "", fmt.Errorf("ensure repository folder exists: %w", err) } var latest string @@ -70,60 +68,37 @@ func (r *Replicator) Sync(ctx context.Context) error { case !isVersioned: r.logger.InfoContext(ctx, "replicate tree unversioned repository") if err := r.replicateTree(ctx, ""); err != nil { - return fmt.Errorf("replicate tree: %w", err) + return "", fmt.Errorf("replicate tree: %w", err) } case lastCommit == "": var err error latest, err = versionedRepo.LatestRef(ctx, r.logger) if err != nil { - return fmt.Errorf("latest ref: %w", err) + return "", fmt.Errorf("latest ref: %w", err) } if err := r.replicateTree(ctx, latest); err != nil { - return fmt.Errorf("replicate tree: %w", err) + return latest, fmt.Errorf("replicate tree: %w", err) } r.logger.InfoContext(ctx, "initial replication for versioned repository", "latest", latest) default: var err error latest, err = versionedRepo.LatestRef(ctx, r.logger) if err != nil { - return fmt.Errorf("latest ref: %w", err) + return "", fmt.Errorf("latest ref: %w", err) } r.logger.InfoContext(ctx, "replicate changes for versioned repository", "last_commit", lastCommit, "latest", latest) changes, err := versionedRepo.CompareFiles(ctx, r.logger, lastCommit, latest) if err != nil { - return fmt.Errorf("compare files: %w", err) + return latest, fmt.Errorf("compare files: %w", err) } if err := r.replicateChanges(ctx, changes); err != nil { - return fmt.Errorf("replicate changes: %w", err) + return latest, fmt.Errorf("replicate changes: %w", err) } } - // TODO: move the sync status to the job worker - status := &provisioning.SyncStatus{ - // FIXME: these create infinite loop - // Started: started.Unix(), - // Finished: time.Now().Unix(), - Hash: latest, - } - - cfg.Status.Sync = *status - - // TODO: Can we use typed client for this? - client := r.client.Resource(provisioning.RepositoryResourceInfo.GroupVersionResource()) - unstructuredResource := &unstructured.Unstructured{} - jj, _ := json.Marshal(cfg) - err := json.Unmarshal(jj, &unstructuredResource.Object) - if err != nil { - return fmt.Errorf("error loading config json: %w", err) - } - - if _, err := client.UpdateStatus(ctx, unstructuredResource, metav1.UpdateOptions{}); err != nil { - return fmt.Errorf("update repository status: %w", err) - } - - return nil + return latest, nil } // replicateTree replicates all files in the repository.