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

@@ -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)

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{

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.