Fix infinite loop and record more information (#98185)

* Fix infinite loop and record more information

* Regenerate code

* Use static client in job worker

* Fix condition to log that a repository is unhealthy

* Submit sync job after we are done with the status update

* Do not set generation in test endpoint

pkg/apis/provisioning/v0alpha1/types.go

@@ -126,10 +126,13 @@ type RepositoryStatus struct {
 	// Initialized is true when the repository has been initialized
 	Initialized bool `json:"initialized"`
 
+	// The generation of the spec last time reconciliation ran
+	ObservedGeneration int64 `json:"observedGeneration"`
+
 	// This will get updated with the current health status (and updated periodically)
 	Health HealthStatus `json:"health"`
 
-	// Sync information
+	// Sync information with the last sync information
 	Sync SyncStatus `json:"sync"`
 
 	// Webhook Information (if applicable)
@@ -140,12 +143,9 @@ type HealthStatus struct {
 	// When not healthy, requests will not be executed
 	Healthy bool `json:"healthy"`
 
-	// When the sync job started
+	// When the health was checked last time
 	Checked int64 `json:"checked,omitempty"`
 
-	// The generation (spec changed) that triggered this health check
-	Generation int64 `json:"generation,omitempty"`
-
 	// Summary messages (will be shown to users)
 	Message []string `json:"message,omitempty"`
 }

pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go

@@ -284,14 +284,7 @@ func schema_pkg_apis_provisioning_v0alpha1_HealthStatus(ref common.ReferenceCall
 					},
 					"checked": {
 						SchemaProps: spec.SchemaProps{
-							Description: "When the sync job started",
-							Type:        []string{"integer"},
-							Format:      "int64",
-						},
-					},
-					"generation": {
-						SchemaProps: spec.SchemaProps{
-							Description: "The generation (spec changed) that triggered this health check",
+							Description: "When the health was checked last time",
 							Type:        []string{"integer"},
 							Format:      "int64",
 						},
@@ -892,6 +885,14 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositoryStatus(ref common.Reference
 							Format:      "",
 						},
 					},
+					"observedGeneration": {
+						SchemaProps: spec.SchemaProps{
+							Description: "The generation of the spec last time reconciliation ran",
+							Default:     0,
+							Type:        []string{"integer"},
+							Format:      "int64",
+						},
+					},
 					"health": {
 						SchemaProps: spec.SchemaProps{
 							Description: "This will get updated with the current health status (and updated periodically)",
@@ -901,7 +902,7 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositoryStatus(ref common.Reference
 					},
 					"sync": {
 						SchemaProps: spec.SchemaProps{
-							Description: "Sync information",
+							Description: "Sync information with the last sync information",
 							Default:     map[string]interface{}{},
 							Ref:         ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.SyncStatus"),
 						},
@@ -913,7 +914,7 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositoryStatus(ref common.Reference
 						},
 					},
 				},
-				Required: []string{"initialized", "health", "sync", "webhook"},
+				Required: []string{"initialized", "observedGeneration", "health", "sync", "webhook"},
 			},
 		},
 		Dependencies: []string{

pkg/generated/applyconfiguration/provisioning/v0alpha1/healthstatus.go

@@ -7,10 +7,9 @@ package v0alpha1
 // HealthStatusApplyConfiguration represents a declarative configuration of the HealthStatus type for use
 // with apply.
 type HealthStatusApplyConfiguration struct {
-	Healthy    *bool    `json:"healthy,omitempty"`
-	Checked    *int64   `json:"checked,omitempty"`
-	Generation *int64   `json:"generation,omitempty"`
-	Message    []string `json:"message,omitempty"`
+	Healthy *bool    `json:"healthy,omitempty"`
+	Checked *int64   `json:"checked,omitempty"`
+	Message []string `json:"message,omitempty"`
 }
 
 // HealthStatusApplyConfiguration constructs a declarative configuration of the HealthStatus type for use with
@@ -35,14 +34,6 @@ func (b *HealthStatusApplyConfiguration) WithChecked(value int64) *HealthStatusA
 	return b
 }
 
-// WithGeneration sets the Generation field in the declarative configuration to the given value
-// and returns the receiver, so that objects can be built by chaining "With" function invocations.
-// If called multiple times, the Generation field is set to the value of the last call.
-func (b *HealthStatusApplyConfiguration) WithGeneration(value int64) *HealthStatusApplyConfiguration {
-	b.Generation = &value
-	return b
-}
-
 // WithMessage adds the given value to the Message field in the declarative configuration
 // and returns the receiver, so that objects can be build by chaining "With" function invocations.
 // If called multiple times, values provided by each call will be appended to the Message field.

pkg/generated/applyconfiguration/provisioning/v0alpha1/repositorystatus.go

@@ -7,10 +7,11 @@ package v0alpha1
 // RepositoryStatusApplyConfiguration represents a declarative configuration of the RepositoryStatus type for use
 // with apply.
 type RepositoryStatusApplyConfiguration struct {
-	Initialized *bool                            `json:"initialized,omitempty"`
-	Health      *HealthStatusApplyConfiguration  `json:"health,omitempty"`
-	Sync        *SyncStatusApplyConfiguration    `json:"sync,omitempty"`
-	Webhook     *WebhookStatusApplyConfiguration `json:"webhook,omitempty"`
+	Initialized        *bool                            `json:"initialized,omitempty"`
+	ObservedGeneration *int64                           `json:"observedGeneration,omitempty"`
+	Health             *HealthStatusApplyConfiguration  `json:"health,omitempty"`
+	Sync               *SyncStatusApplyConfiguration    `json:"sync,omitempty"`
+	Webhook            *WebhookStatusApplyConfiguration `json:"webhook,omitempty"`
 }
 
 // RepositoryStatusApplyConfiguration constructs a declarative configuration of the RepositoryStatus type for use with
@@ -27,6 +28,14 @@ func (b *RepositoryStatusApplyConfiguration) WithInitialized(value bool) *Reposi
 	return b
 }
 
+// WithObservedGeneration sets the ObservedGeneration field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the ObservedGeneration field is set to the value of the last call.
+func (b *RepositoryStatusApplyConfiguration) WithObservedGeneration(value int64) *RepositoryStatusApplyConfiguration {
+	b.ObservedGeneration = &value
+	return b
+}
+
 // WithHealth sets the Health field in the declarative configuration to the given value
 // and returns the receiver, so that objects can be built by chaining "With" function invocations.
 // If called multiple times, the Health field is set to the value of the last call.

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

@@ -142,8 +142,8 @@ func (s *jobStore) drainPending() {
 			logger.DebugContext(ctx, "job processing finished", "status", status.State)
 		}
 
-		status.Started = started.UnixNano()
-		status.Finished = time.Now().UnixNano()
+		status.Started = started.UnixMilli()
+		status.Finished = time.Now().UnixMilli()
 
 		err = s.Complete(ctx, job.Namespace, job.Name, *status)
 		if err != nil {

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

@@ -2,17 +2,17 @@ package jobs
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"log/slog"
 	"net/url"
+	"time"
 
 	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"
 	provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
+	client "github.com/grafana/grafana/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/auth"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
@@ -27,6 +27,7 @@ type RepoGetter interface {
 var _ Worker = (*JobWorker)(nil)
 
 type JobWorker struct {
+	client      client.ProvisioningV0alpha1Interface
 	getter      RepoGetter
 	parsers     *resources.ParserFactory
 	identities  auth.BackgroundIdentityService
@@ -39,6 +40,7 @@ type JobWorker struct {
 func NewJobWorker(
 	getter RepoGetter,
 	parsers *resources.ParserFactory,
+	client client.ProvisioningV0alpha1Interface,
 	identities auth.BackgroundIdentityService,
 	logger *slog.Logger,
 	render rendering.Service,
@@ -47,6 +49,7 @@ func NewJobWorker(
 ) *JobWorker {
 	return &JobWorker{
 		getter:      getter,
+		client:      client,
 		parsers:     parsers,
 		identities:  identities,
 		logger:      logger,
@@ -90,14 +93,29 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job) (*provisi
 
 	switch job.Spec.Action {
 	case provisioning.JobActionSync:
-		ref, syncError := replicator.Sync(ctx)
+		started := time.Now()
+
+		// Update the status to indicate that we are working on it
+		cfg := repo.Config().DeepCopy()
 		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?
+			State:   provisioning.JobStateWorking,
+			JobID:   job.GetName(),
+			Started: started.UnixMilli(),
+		}
+		cfg.Status.Sync = *status
+		cfg, err := g.client.Repositories(cfg.GetNamespace()).UpdateStatus(ctx, cfg, v1.UpdateOptions{})
+		if err != nil {
+			return nil, fmt.Errorf("update repository status: %w", err)
+		}
+
+		// Sync the repository
+		ref, syncError := replicator.Sync(ctx)
+		status = &provisioning.SyncStatus{
+			State:    provisioning.JobStateFinished,
+			JobID:    job.GetName(),
 			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
+			Started:  started.UnixMilli(),
+			Finished: time.Now().UnixMilli(),
 			Message:  []string{},
 		}
 
@@ -106,18 +124,8 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job) (*provisi
 			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 {
+		if _, err := g.client.Repositories(cfg.GetNamespace()).UpdateStatus(ctx, cfg, v1.UpdateOptions{}); err != nil {
 			return nil, fmt.Errorf("update repository status: %w", err)
 		}
 

pkg/registry/apis/provisioning/register.go

@@ -177,16 +177,6 @@ func (b *ProvisioningAPIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserv
 		return fmt.Errorf("failed to create repository storage: %w", err)
 	}
 
-	b.jobs.Register(jobs.NewJobWorker(
-		b,
-		b.parsers,
-		b.identities,
-		b.logger.With("worker", "github"),
-		b.render,
-		b.blobstore,
-		b.urlProvider,
-	))
-
 	repositoryStatusStorage := grafanaregistry.NewRegistryStatusStore(opts.Scheme, repositoryStorage)
 	b.getter = repositoryStorage
 
@@ -403,6 +393,17 @@ func (b *ProvisioningAPIBuilder) GetPostStartHooks() (map[string]genericapiserve
 				logger:        slog.Default().With("logger", "provisioning-repository-tester"),
 			}
 
+			b.jobs.Register(jobs.NewJobWorker(
+				b,
+				b.parsers,
+				c.ProvisioningV0alpha1(),
+				b.identities,
+				b.logger.With("worker", "github"),
+				b.render,
+				b.blobstore,
+				b.urlProvider,
+			))
+
 			repoController, err := NewRepositoryController(
 				c.ProvisioningV0alpha1(),
 				repoInformer,

pkg/registry/apis/provisioning/repository_controller.go

@@ -226,63 +226,27 @@ func (rc *RepositoryController) process(item *queueItem) error {
 		return fmt.Errorf("unable to create repository from configuration: %w", err)
 	}
 
-	// Did the spec change
-	now := time.Now().UnixMilli()
-	generationChanged := cachedRepo.Generation != cachedRepo.Status.Health.Generation
-	elapsed := time.Duration(now-cachedRepo.Status.Health.Checked) * time.Millisecond
-	if elapsed < time.Millisecond*200 {
-		// avoids possible inf loop!!!
-		return nil
-	}
-	if elapsed < time.Second*30 && !generationChanged {
-		// We checked status recently! and the generation has not changed
+	hasSpecChanged := cachedRepo.Generation != cachedRepo.Status.ObservedGeneration
+	if !hasSpecChanged {
+		logger.InfoContext(ctx, "repository spec unchanged")
 		return nil
 	}
 
-	res, err := rc.tester.TestRepository(ctx, repo)
-	if err != nil {
-		res = &provisioning.TestResults{
-			Success: false,
-			Errors: []string{
-				"error running test repository",
-				err.Error(),
-			},
-		}
-	}
+	logger.InfoContext(ctx, "repository spec changed", "previous", cachedRepo.Status.ObservedGeneration, "current", cachedRepo.Generation)
 
 	var status *provisioning.RepositoryStatus
-	if res.Success {
-		if cachedRepo.Status.Initialized {
-			logger.InfoContext(ctx, "handle repository update")
-			status, err = repo.OnUpdate(ctx, logger)
-			if err != nil {
-				rc.logger.Error("OnUpdate", "error", err)
-			}
-		} else {
-			logger.InfoContext(ctx, "handle repository init")
-			status, err = repo.OnCreate(ctx, logger)
-			if err != nil {
-				rc.logger.Error("OnCreate", "error", err)
-			}
-		}
-
-		job, err := rc.jobs.Add(ctx, &provisioning.Job{
-			ObjectMeta: v1.ObjectMeta{
-				Namespace: cachedRepo.Namespace,
-				Labels: map[string]string{
-					"repository": cachedRepo.Name,
-				},
-			},
-			Spec: provisioning.JobSpec{
-				Action: provisioning.JobActionSync,
-			},
-		})
+	if cachedRepo.Status.Initialized {
+		logger.InfoContext(ctx, "handle repository update")
+		status, err = repo.OnUpdate(ctx, logger)
 		if err != nil {
-			return fmt.Errorf("trigger sync job: %w", err)
+			rc.logger.Error("OnUpdate", "error", err)
 		}
-		logger.InfoContext(ctx, "sync job triggered", "job", job.Name)
 	} else {
-		logger.ErrorContext(ctx, "repository is unhealthy", "errors", res.Errors)
+		logger.InfoContext(ctx, "handle repository init")
+		status, err = repo.OnCreate(ctx, logger)
+		if err != nil {
+			rc.logger.Error("OnCreate", "error", err)
+		}
 	}
 
 	if status == nil {
@@ -290,12 +254,31 @@ func (rc *RepositoryController) process(item *queueItem) error {
 	} else {
 		status.Initialized = true
 	}
+	status.ObservedGeneration = cachedRepo.Generation
 
-	status.Health = provisioning.HealthStatus{
-		Checked:    status.Health.Checked,
-		Generation: status.Health.Generation,
-		Healthy:    res.Success,
-		Message:    res.Errors,
+	if time.Since(time.UnixMilli(cachedRepo.Status.Health.Checked)) < 200*time.Millisecond {
+		logger.InfoContext(ctx, "skipping health check as it was recently checked")
+	} else {
+		res, err := rc.tester.TestRepository(ctx, repo)
+		if err != nil {
+			res = &provisioning.TestResults{
+				Success: false,
+				Errors: []string{
+					"error running test repository",
+					err.Error(),
+				},
+			}
+		}
+
+		if !res.Success {
+			logger.ErrorContext(ctx, "repository is unhealthy", "errors", res.Errors)
+		}
+
+		status.Health = provisioning.HealthStatus{
+			Checked: status.Health.Checked,
+			Healthy: res.Success,
+			Message: res.Errors,
+		}
 	}
 
 	cfg := cachedRepo.DeepCopy()
@@ -305,5 +288,21 @@ func (rc *RepositoryController) process(item *queueItem) error {
 		return fmt.Errorf("update status: %w", err)
 	}
 
+	job, err := rc.jobs.Add(ctx, &provisioning.Job{
+		ObjectMeta: v1.ObjectMeta{
+			Namespace: cachedRepo.Namespace,
+			Labels: map[string]string{
+				"repository": cachedRepo.Name,
+			},
+		},
+		Spec: provisioning.JobSpec{
+			Action: provisioning.JobActionSync,
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("trigger sync job: %w", err)
+	}
+	logger.InfoContext(ctx, "sync job triggered", "job", job.Name)
+
 	return nil
 }

pkg/registry/apis/provisioning/test.go

@@ -147,10 +147,9 @@ func (t *RepositoryTester) UpdateHealthStatus(ctx context.Context, cfg *provisio
 
 	repo := cfg.DeepCopy()
 	repo.Status.Health = provisioning.HealthStatus{
-		Healthy:    res.Success,
-		Checked:    time.Now().UnixMilli(),
-		Generation: cfg.Generation,
-		Message:    res.Errors,
+		Healthy: res.Success,
+		Checked: time.Now().UnixMilli(),
+		Message: res.Errors,
 	}
 
 	_, err := t.client.Repositories(repo.GetNamespace()).