diff --git a/pkg/apis/provisioning/v0alpha1/types.go b/pkg/apis/provisioning/v0alpha1/types.go index a3eadf0c32e..2e6acabf2ec 100644 --- a/pkg/apis/provisioning/v0alpha1/types.go +++ b/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"` } diff --git a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go index 2f3dc8c0900..63cd82da586 100644 --- a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go +++ b/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{ diff --git a/pkg/generated/applyconfiguration/provisioning/v0alpha1/healthstatus.go b/pkg/generated/applyconfiguration/provisioning/v0alpha1/healthstatus.go index 3c8901bf98d..6d3b33cb318 100644 --- a/pkg/generated/applyconfiguration/provisioning/v0alpha1/healthstatus.go +++ b/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. diff --git a/pkg/generated/applyconfiguration/provisioning/v0alpha1/repositorystatus.go b/pkg/generated/applyconfiguration/provisioning/v0alpha1/repositorystatus.go index 07aa9945940..ea63efb90f3 100644 --- a/pkg/generated/applyconfiguration/provisioning/v0alpha1/repositorystatus.go +++ b/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. diff --git a/pkg/registry/apis/provisioning/jobs/queue.go b/pkg/registry/apis/provisioning/jobs/queue.go index 4af483c5cbf..6f70b18ad02 100644 --- a/pkg/registry/apis/provisioning/jobs/queue.go +++ b/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 { diff --git a/pkg/registry/apis/provisioning/jobs/worker.go b/pkg/registry/apis/provisioning/jobs/worker.go index ddd6485413b..90be7205059 100644 --- a/pkg/registry/apis/provisioning/jobs/worker.go +++ b/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) } diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 153affe210a..2414cf8e580 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/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, diff --git a/pkg/registry/apis/provisioning/repository_controller.go b/pkg/registry/apis/provisioning/repository_controller.go index 78039282cae..83a3f05fe24 100644 --- a/pkg/registry/apis/provisioning/repository_controller.go +++ b/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 } diff --git a/pkg/registry/apis/provisioning/test.go b/pkg/registry/apis/provisioning/test.go index 64776a2ef3a..3e8f7111943 100644 --- a/pkg/registry/apis/provisioning/test.go +++ b/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()).