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.