diff --git a/pkg/apimachinery/apis/common/v0alpha1/unstructured.go b/pkg/apimachinery/apis/common/v0alpha1/unstructured.go index e779c5fafd3..5d1f452850c 100644 --- a/pkg/apimachinery/apis/common/v0alpha1/unstructured.go +++ b/pkg/apimachinery/apis/common/v0alpha1/unstructured.go @@ -63,6 +63,18 @@ func (u *Unstructured) DeepCopy() *Unstructured { } out := new(Unstructured) *out = *u + + // If the core value includes an 'int' the runtime will fail + defer func() { + if r := recover(); r != nil { + raw, err := u.MarshalJSON() + if err != nil { + panic("unable to marshal object") + } + _ = out.UnmarshalJSON(raw) + } + }() + out.Object = runtime.DeepCopyJSON(u.Object) return out } diff --git a/pkg/apis/provisioning/v0alpha1/jobs.go b/pkg/apis/provisioning/v0alpha1/jobs.go index 7e2172f343e..0c3c6902573 100644 --- a/pkg/apis/provisioning/v0alpha1/jobs.go +++ b/pkg/apis/provisioning/v0alpha1/jobs.go @@ -119,6 +119,17 @@ type JobStatus struct { Summary []JobResourceSummary `json:"summary,omitempty"` } +// Convert a JOB to a +func (in *JobStatus) ToSyncStatus(jobId string) SyncStatus { + return SyncStatus{ + JobID: jobId, + State: in.State, + Started: in.Started, + Finished: in.Finished, + Message: in.Errors, + } +} + type JobResourceSummary struct { Group string `json:"group,omitempty"` Resource string `json:"resource,omitempty"` diff --git a/pkg/apis/provisioning/v0alpha1/types.go b/pkg/apis/provisioning/v0alpha1/types.go index 4c34c7735cc..29d4cca570b 100644 --- a/pkg/apis/provisioning/v0alpha1/types.go +++ b/pkg/apis/provisioning/v0alpha1/types.go @@ -81,9 +81,6 @@ type RepositorySpec struct { // Edit options within the repository Editing EditingOptions `json:"editing"` - // DeletePolicy options within the repository - DeletePolicy DeletePolicy `json:"deletePolicy"` - // The repository type. When selected oneOf the values below should be non-nil Type RepositoryType `json:"type"` @@ -101,15 +98,6 @@ type RepositorySpec struct { GitHub *GitHubRepositoryConfig `json:"github,omitempty"` } -type DeletePolicy string - -const ( - // DeletePolityRetain keep all provisioned resources on repository deletion - DeletePolityRetain DeletePolicy = "retain" - // DeletePolityClean delete all provisioned resources on repository deletion - DeletePolityClean DeletePolicy = "clean" -) - type EditingOptions struct { // End users can create new files in the remote file system Create bool `json:"create"` diff --git a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go index b5f96dea04b..afc66aafc84 100644 --- a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go +++ b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go @@ -987,14 +987,6 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositorySpec(ref common.ReferenceCa Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.EditingOptions"), }, }, - "deletePolicy": { - SchemaProps: spec.SchemaProps{ - Description: "DeletePolicy options within the repository", - Default: "", - Type: []string{"string"}, - Format: "", - }, - }, "type": { SchemaProps: spec.SchemaProps{ Description: "The repository type. When selected oneOf the values below should be non-nil\n\nPossible enum values:\n - `\"github\"`\n - `\"local\"`\n - `\"s3\"`", @@ -1023,7 +1015,7 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositorySpec(ref common.ReferenceCa }, }, }, - Required: []string{"title", "editing", "deletePolicy", "type"}, + Required: []string{"title", "editing", "type"}, }, }, Dependencies: []string{ diff --git a/pkg/generated/applyconfiguration/provisioning/v0alpha1/repositoryspec.go b/pkg/generated/applyconfiguration/provisioning/v0alpha1/repositoryspec.go index da6d24aa4a6..9e5f2041be6 100644 --- a/pkg/generated/applyconfiguration/provisioning/v0alpha1/repositoryspec.go +++ b/pkg/generated/applyconfiguration/provisioning/v0alpha1/repositoryspec.go @@ -11,15 +11,14 @@ import ( // RepositorySpecApplyConfiguration represents a declarative configuration of the RepositorySpec type for use // with apply. type RepositorySpecApplyConfiguration struct { - Title *string `json:"title,omitempty"` - Description *string `json:"description,omitempty"` - Folder *string `json:"folder,omitempty"` - Editing *EditingOptionsApplyConfiguration `json:"editing,omitempty"` - DeletePolicy *provisioningv0alpha1.DeletePolicy `json:"deletePolicy,omitempty"` - Type *provisioningv0alpha1.RepositoryType `json:"type,omitempty"` - Local *LocalRepositoryConfigApplyConfiguration `json:"local,omitempty"` - S3 *S3RepositoryConfigApplyConfiguration `json:"s3,omitempty"` - GitHub *GitHubRepositoryConfigApplyConfiguration `json:"github,omitempty"` + Title *string `json:"title,omitempty"` + Description *string `json:"description,omitempty"` + Folder *string `json:"folder,omitempty"` + Editing *EditingOptionsApplyConfiguration `json:"editing,omitempty"` + Type *provisioningv0alpha1.RepositoryType `json:"type,omitempty"` + Local *LocalRepositoryConfigApplyConfiguration `json:"local,omitempty"` + S3 *S3RepositoryConfigApplyConfiguration `json:"s3,omitempty"` + GitHub *GitHubRepositoryConfigApplyConfiguration `json:"github,omitempty"` } // RepositorySpecApplyConfiguration constructs a declarative configuration of the RepositorySpec type for use with @@ -60,14 +59,6 @@ func (b *RepositorySpecApplyConfiguration) WithEditing(value *EditingOptionsAppl return b } -// WithDeletePolicy sets the DeletePolicy 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 DeletePolicy field is set to the value of the last call. -func (b *RepositorySpecApplyConfiguration) WithDeletePolicy(value provisioningv0alpha1.DeletePolicy) *RepositorySpecApplyConfiguration { - b.DeletePolicy = &value - return b -} - // WithType sets the Type 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 Type field is set to the value of the last call. diff --git a/pkg/registry/apis/provisioning/finalizers.go b/pkg/registry/apis/provisioning/finalizers.go new file mode 100644 index 00000000000..dff3fa1d16a --- /dev/null +++ b/pkg/registry/apis/provisioning/finalizers.go @@ -0,0 +1,151 @@ +package provisioning + +import ( + "sort" + "strings" + + "golang.org/x/net/context" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/dynamic" + + "github.com/grafana/grafana-app-sdk/logging" + "github.com/grafana/grafana/pkg/apimachinery/utils" + dashboards "github.com/grafana/grafana/pkg/apis/dashboard" + folders "github.com/grafana/grafana/pkg/apis/folder/v0alpha1" + provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" +) + +// Remove everything this repo created +const finalizer_REMOVE_ORPHAN_RESOURCE = "remove-orphan-resources" + +// Remove the metadata for anything this repo created +const finalizer_RELEASE_ORPHAN_RESOURCE = "release-orphan-resources" + +// Calls the "OnDelete" function for resource +const finalizer_CLEANUP_FINALIZER = "cleanup" + +type finalizer struct { + lister resources.ResourceLister + client *resources.ClientFactory +} + +func (f *finalizer) process(ctx context.Context, + repo repository.Repository, + finalizers []string, +) error { + logger := logging.FromContext(ctx) + + for _, finalizer := range finalizers { + switch finalizer { + case finalizer_CLEANUP_FINALIZER: + // NOTE: the controller loop will never get run unless a finalizer is set + hooks, ok := repo.(repository.RepositoryHooks) + if ok { + if err := hooks.OnDelete(ctx); err != nil { + logger.Warn("Error running deletion hooks", "err", err) + } + } + + case finalizer_RELEASE_ORPHAN_RESOURCE: + err := f.processExistingItems(ctx, repo.Config(), + func(client dynamic.ResourceInterface, item *provisioning.ResourceListItem) error { + _, err := client.Patch(ctx, item.Name, types.JSONPatchType, []byte(`[ + {"op": "remove", "path": "/metadata/annotations/`+utils.AnnoKeyRepoName+`" }, + {"op": "remove", "path": "/metadata/annotations/`+utils.AnnoKeyRepoPath+`" }, + {"op": "remove", "path": "/metadata/annotations/`+utils.AnnoKeyRepoHash+`" } + ]`), v1.PatchOptions{}) + return err + }) + if err != nil { + return err + } + + case finalizer_REMOVE_ORPHAN_RESOURCE: + err := f.processExistingItems(ctx, repo.Config(), + func(client dynamic.ResourceInterface, item *provisioning.ResourceListItem) error { + return client.Delete(ctx, item.Name, v1.DeleteOptions{}) + }) + if err != nil { + return err + } + + default: + logger.Warn("skipping unknown finalizer", "finalizer", finalizer) + } + } + return nil +} + +// internal iterator to walk the existing items +func (f *finalizer) processExistingItems( + ctx context.Context, + repo *provisioning.Repository, + cb func(client dynamic.ResourceInterface, item *provisioning.ResourceListItem) error, +) error { + logger := logging.FromContext(ctx) + client, _, err := f.client.New(repo.Namespace) + if err != nil { + return err + } + + items, err := f.lister.List(ctx, repo.Namespace, repo.Name) + if err != nil { + logger.Warn("error listing resources", "error", err) + return err + } + + // Safe deletion order + sortResourceListForDeletion(items) + count := 0 + errors := 0 + + for _, item := range items.Items { + // HACK: we need to find a better way to know the API version + var version string + switch item.Group { + case folders.GROUP: + version = folders.VERSION + case dashboards.GROUP: + version = "v0alpha1" // the constant is internal + default: + version = "v0alpha1" + } + + res := client.Resource(schema.GroupVersionResource{ + Group: item.Group, + Resource: item.Resource, + Version: version, + }) + + err = cb(res, &item) + if err != nil { + logger.Warn("error processing item", "name", item.Name, "error", err) + errors++ + } else { + count++ + } + } + logger.Info("processed orphan items", "items", count, "errors", errors) + return nil +} + +func sortResourceListForDeletion(list *provisioning.ResourceList) { + // FIXME: this code should be simplified once unified storage folders support recursive deletion + // Sort by the following logic: + // - Put folders at the end so that we empty them first. + // - Sort folders by depth so that we remove the deepest first + sort.Slice(list.Items, func(i, j int) bool { + switch { + case list.Items[i].Group != folders.RESOURCE: + return true + case list.Items[j].Group != folders.RESOURCE: + return false + default: + return len(strings.Split(list.Items[i].Path, "/")) > len(strings.Split(list.Items[j].Path, "/")) + } + }) +} diff --git a/pkg/registry/apis/provisioning/jobs/queue.go b/pkg/registry/apis/provisioning/jobs/queue.go index 331d944db25..71d9b55cdf7 100644 --- a/pkg/registry/apis/provisioning/jobs/queue.go +++ b/pkg/registry/apis/provisioning/jobs/queue.go @@ -149,7 +149,14 @@ func (s *jobStore) drainPending() { State: provisioning.JobStateError, Errors: []string{err.Error()}, } - } else if status.State == "" { + } + if status == nil { + status = &provisioning.JobStatus{ + State: provisioning.JobStateError, + Errors: []string{"no status response from worker"}, + } + } + if status.State == "" { status.State = provisioning.JobStateSuccess } logger.Debug("job processing finished", "status", status.State) @@ -192,6 +199,19 @@ func (s *jobStore) Next(ctx context.Context) *provisioning.Job { return nil } +// Get the status for a job +func (s *jobStore) Status(ctx context.Context, namespace string, name string) *provisioning.JobStatus { + s.mutex.Lock() + defer s.mutex.Unlock() + + for _, job := range s.jobs { + if job.Name == name && job.Namespace == namespace { + return job.Status.DeepCopy() + } + } + return nil +} + // Complete implements JobQueue. func (s *jobStore) Update(ctx context.Context, namespace string, name string, status provisioning.JobStatus) error { s.mutex.Lock() diff --git a/pkg/registry/apis/provisioning/jobs/syncer.go b/pkg/registry/apis/provisioning/jobs/syncer.go index 3e217217dde..34bf0e49a0e 100644 --- a/pkg/registry/apis/provisioning/jobs/syncer.go +++ b/pkg/registry/apis/provisioning/jobs/syncer.go @@ -137,53 +137,6 @@ func (r *Syncer) cleanUnnecessaryResources(ctx context.Context, tree []repositor return nil } -func (r *Syncer) Unsync(ctx context.Context) error { - cfg := r.repository.Config() - - logger := logging.FromContext(ctx) - logger = logger.With("repository", cfg.Name, "namespace", cfg.GetNamespace(), "folder", cfg.Spec.Folder, "mode", cfg.Spec.DeletePolicy) - logger.Info("start repository unsync") - defer logger.Info("end repository unsync") - - switch cfg.Spec.DeletePolicy { - case provisioning.DeletePolityRetain: - logger.Info("keep all resources") - - // TODO: remove repository info from all resources in the list - - if r.repository.Config().Spec.Folder != "" { - obj, err := r.folders.Get(ctx, r.repository.Config().Spec.Folder, metav1.GetOptions{}) - if err != nil { - return fmt.Errorf("get folder to unsync: %w", err) - } - - meta, err := utils.MetaAccessor(obj) - if err != nil { - return fmt.Errorf("create meta accessor from folder object: %w", err) - } - - meta.SetRepositoryInfo(nil) - obj, err = r.folders.Update(ctx, obj, metav1.UpdateOptions{}) - if err != nil { - return fmt.Errorf("remove repo info from folder: %w", err) - } - - logger.Info("removed repo info from folder", "object", obj) - } else { - logger.Info("skip repo info removal as it's root folder") - } - case provisioning.DeletePolityClean: - logger.Info("remove all provisioned resources") - if err := r.deleteAllProvisionedResources(ctx); err != nil { - return fmt.Errorf("delete all resources: %w", err) - } - default: - return fmt.Errorf("invalid unsync mode: %s", cfg.Spec.DeletePolicy) - } - - return nil -} - func sortResourceListForDeletion(list *provisioning.ResourceList) { // FIXME: this code should be simplified once unified storage folders support recursive deletion // Sort by the following logic: @@ -201,29 +154,6 @@ func sortResourceListForDeletion(list *provisioning.ResourceList) { }) } -func (r *Syncer) deleteAllProvisionedResources(ctx context.Context) error { - list, err := r.lister.List(ctx, r.client.GetNamespace(), r.repository.Config().Name) - if err != nil { - return fmt.Errorf("list resources to delete: %w", err) - } - - sortResourceListForDeletion(list) - - logger := logging.FromContext(ctx) - logger.Info("deleting resources", "count", len(list.Items)) - for _, item := range list.Items { - logger := logger.With("group", item.Group, "resource", item.Resource, "name", item.Name) - - if err := r.deleteListResource(ctx, item); err != nil { - logger.Error("failed to delete resource", "error", err) - return fmt.Errorf("delete resource: %w", err) - } - logger.Info("resource deleted") - } - - return nil -} - func (r *Syncer) deleteListResource(ctx context.Context, item provisioning.ResourceListItem) error { // HACK: we need to find a better way to know the API version var version string diff --git a/pkg/registry/apis/provisioning/jobs/types.go b/pkg/registry/apis/provisioning/jobs/types.go index 13be180ea82..40e0c471a35 100644 --- a/pkg/registry/apis/provisioning/jobs/types.go +++ b/pkg/registry/apis/provisioning/jobs/types.go @@ -22,6 +22,9 @@ type JobQueue interface { // Get the next job we should process Next(ctx context.Context) *provisioning.Job + // Get the status for a job + Status(ctx context.Context, namespace string, name string) *provisioning.JobStatus + // Update the status on a given job // This is only valid if current job is not finished Update(ctx context.Context, namespace string, name string, status provisioning.JobStatus) error diff --git a/pkg/registry/apis/provisioning/jobs/worker.go b/pkg/registry/apis/provisioning/jobs/worker.go index cb3db8f31bd..5d61c2da2e5 100644 --- a/pkg/registry/apis/provisioning/jobs/worker.go +++ b/pkg/registry/apis/provisioning/jobs/worker.go @@ -2,12 +2,14 @@ package jobs import ( "context" + "encoding/json" "fmt" "net/url" "time" apierrors "k8s.io/apimachinery/pkg/api/errors" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" "k8s.io/apiserver/pkg/endpoints/request" "github.com/grafana/grafana-app-sdk/logging" @@ -89,63 +91,8 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job, progress switch job.Spec.Action { case provisioning.JobActionSync: - started := time.Now() + return g.doSync(ctx, repo, job, parser, progress) - // Update the status to indicate that we are working on it - cfg := repo.Config().DeepCopy() - status := &provisioning.SyncStatus{ - 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) - } - - syncer, err := NewSyncer(repo, g.lister, parser) - if err != nil { - return nil, fmt.Errorf("error creating replicator") - } - - var complete bool - if job.Spec.Sync != nil && job.Spec.Sync.Complete { - complete = job.Spec.Sync.Complete - } - - // Sync the repository - ref, syncError := syncer.Sync(ctx, complete) - status = &provisioning.SyncStatus{ - State: provisioning.JobStateSuccess, - JobID: job.GetName(), - Hash: ref, - Started: started.UnixMilli(), - Finished: time.Now().UnixMilli(), - Message: []string{}, - } - - if syncError != nil { - status.State = provisioning.JobStateError - status.Message = append(status.Message, syncError.Error()) - } - - // Update the resource stats - stats, err := g.lister.Stats(ctx, cfg.Namespace, cfg.Name) - if err != nil { - logger.Warn("unable to read stats", "error", err) - } else if stats != nil { - cfg.Status.Stats = stats.Items - } - - cfg.Status.Sync = *status - if _, err := g.client.Repositories(cfg.GetNamespace()).UpdateStatus(ctx, cfg, 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) if !ok { @@ -200,6 +147,87 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job, progress default: return nil, fmt.Errorf("unknown job action: %s", job.Spec.Action) } +} + +func (g *JobWorker) doSync(ctx context.Context, + repo repository.Repository, + job provisioning.Job, + parser *resources.Parser, + _ func(provisioning.JobStatus) error, +) (*provisioning.JobStatus, error) { + var err error + cfg := repo.Config() + + logger := logging.FromContext(ctx).With("job", job.GetName(), "namespace", job.GetNamespace()) + status := job.Status.ToSyncStatus(job.Name) + patch, err := json.Marshal(map[string]any{ + "status": map[string]any{ + "sync": status, + }, + }) + if err != nil { + return nil, err + } + + cfg, err = g.client.Repositories(cfg.Namespace). + Patch(ctx, cfg.Name, types.MergePatchType, patch, v1.PatchOptions{}, "status") + if err != nil { + logger.Warn("unable to update repo with job status", "err", err) + } + + syncer, err := NewSyncer(repo, g.lister, parser) + if err != nil { + return nil, fmt.Errorf("error creating replicator") + } + + var complete bool + if job.Spec.Sync != nil && job.Spec.Sync.Complete { + complete = job.Spec.Sync.Complete + } + + // Sync the repository + ref, syncError := syncer.Sync(ctx, complete) + status = job.Status.ToSyncStatus(job.Name) + status.Hash = ref + status.Finished = time.Now().UnixMilli() + if syncError != nil { + status.State = provisioning.JobStateError + status.Message = []string{ + "error running sync", + syncError.Error(), + } + } else { + status.State = provisioning.JobStateSuccess + } + + // Update the resource stats + stats, err := g.lister.Stats(ctx, cfg.Namespace, cfg.Name) + if err != nil { + logger.Warn("unable to read stats", "error", err) + } + if stats == nil { + stats = &provisioning.ResourceStats{} + } + + patch, err = json.Marshal(map[string]any{ + "status": map[string]any{ + "sync": status, + "stats": stats.Items, + }, + }) + if err != nil { + return nil, err + } + + _, err = g.client.Repositories(cfg.Namespace). + Patch(ctx, cfg.Name, types.MergePatchType, patch, v1.PatchOptions{}, "status") + if err != nil { + logger.Warn("unable to update repo with job status", "err", err) + } + + if syncError != nil { + return nil, syncError + } return &provisioning.JobStatus{ State: provisioning.JobStateSuccess, diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 7ce8b05441d..6f1573e013c 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -321,8 +321,12 @@ func (b *ProvisioningAPIBuilder) Mutate(ctx context.Context, a admission.Attribu return fmt.Errorf("expected repository configuration") } - if r.Spec.DeletePolicy == "" { - r.Spec.DeletePolicy = provisioning.DeletePolityClean + // This is called on every update, so be careful to only add the finalizer for create + if len(r.Finalizers) == 0 && a.GetOperation() == admission.Create { + r.Finalizers = []string{ + finalizer_REMOVE_ORPHAN_RESOURCE, + finalizer_CLEANUP_FINALIZER, + } } if r.Spec.Type == provisioning.GitHubRepositoryType { @@ -482,6 +486,7 @@ func (b *ProvisioningAPIBuilder) GetPostStartHooks() (map[string]genericapiserve b.lister, b.parsers, b.identities, + b.lister, b.tester, b.jobs, ) diff --git a/pkg/registry/apis/provisioning/repository_controller.go b/pkg/registry/apis/provisioning/repository_controller.go index c6cac71ff11..bf320e94944 100644 --- a/pkg/registry/apis/provisioning/repository_controller.go +++ b/pkg/registry/apis/provisioning/repository_controller.go @@ -8,6 +8,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" @@ -27,19 +28,10 @@ import ( const loggerName = "provisioning-repository-controller" -type operation int - -const ( - operationCreate operation = iota - operationUpdate - operationDelete -) - const maxAttempts = 3 type queueItem struct { key string - op operation obj interface{} attempts int } @@ -51,8 +43,10 @@ type RepositoryController struct { repoLister listers.RepositoryLister repoSynced cache.InformerSynced parsers *resources.ParserFactory + logger logging.Logger - jobs jobs.JobQueue + jobs jobs.JobQueue + finalizer *finalizer // Converts config to instance repoGetter RepoGetter @@ -61,7 +55,7 @@ type RepositoryController struct { // To allow injection for testing. processFn func(item *queueItem) error - enqueueRepository func(op operation, obj any) + enqueueRepository func(obj any) keyFunc func(obj any) (string, error) queue workqueue.TypedRateLimitingInterface[*queueItem] @@ -75,6 +69,7 @@ func NewRepositoryController( resourceLister resources.ResourceLister, parsers *resources.ParserFactory, identities auth.BackgroundIdentityService, + lister resources.ResourceLister, tester *RepositoryTester, jobs jobs.JobQueue, ) (*RepositoryController, error) { @@ -92,14 +87,20 @@ func NewRepositoryController( repoGetter: repoGetter, parsers: parsers, identities: identities, - tester: tester, - jobs: jobs, + finalizer: &finalizer{ + lister: lister, + client: parsers.Client, + }, + tester: tester, + jobs: jobs, + logger: logging.DefaultLogger.With("logger", loggerName), } _, err := repoInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ - AddFunc: rc.addRepository, - UpdateFunc: rc.updateRepository, - DeleteFunc: rc.deleteRepository, + AddFunc: rc.enqueue, + UpdateFunc: func(oldObj, newObj interface{}) { + rc.enqueue(newObj) + }, }) if err != nil { return nil, err @@ -125,7 +126,7 @@ func (rc *RepositoryController) Run(ctx context.Context, workerCount int) { defer utilruntime.HandleCrash() defer rc.queue.ShutDown() - logger := logging.FromContext(ctx).With("logger", loggerName) + logger := rc.logger ctx = logging.Context(ctx, logger) logger.Info("Starting RepositoryController") defer logger.Info("Shutting down RepositoryController") @@ -149,29 +150,17 @@ func (rc *RepositoryController) runWorker(ctx context.Context) { } } -func (rc *RepositoryController) enqueue(op operation, obj interface{}) { +func (rc *RepositoryController) enqueue(obj interface{}) { key, err := rc.keyFunc(obj) if err != nil { utilruntime.HandleError(fmt.Errorf("couldn't get key for object: %v", err)) return } - item := queueItem{key: key, obj: obj, op: op} + item := queueItem{key: key, obj: obj} rc.queue.Add(&item) } -func (rc *RepositoryController) addRepository(obj interface{}) { - rc.enqueueRepository(operationCreate, obj) -} - -func (rc *RepositoryController) updateRepository(oldObj, newObj interface{}) { - rc.enqueueRepository(operationUpdate, newObj) -} - -func (rc *RepositoryController) deleteRepository(obj interface{}) { - rc.enqueueRepository(operationDelete, obj) -} - // processNextWorkItem deals with one key off the queue. // It returns false when it's time to quit. func (rc *RepositoryController) processNextWorkItem(ctx context.Context) bool { @@ -215,61 +204,16 @@ func (rc *RepositoryController) processNextWorkItem(ctx context.Context) bool { return true } -// process is the business logic of the controller. +//nolint:gocyclo func (rc *RepositoryController) process(item *queueItem) error { - ctx := context.Background() - - logger := logging.FromContext(ctx).With("logger", loggerName, "key", item.key) - ctx = logging.Context(ctx, logger) + logger := rc.logger.With("key", item.key) namespace, name, err := cache.SplitMetaNamespaceKey(item.key) if err != nil { return err } - if item.op == operationDelete { - logger.Info("handle repository deletion") - cfg, ok := item.obj.(*provisioning.Repository) - if !ok { - return errors.New("object is not a repository") - } - - repo, err := rc.repoGetter.AsRepository(ctx, cfg) - if err != nil { - return fmt.Errorf("unable to create repository from object: %w", err) - } - - if hooks, ok := repo.(repository.RepositoryHooks); !ok { - err := hooks.OnDelete(ctx) - if err != nil { - return fmt.Errorf("handle repository delete: %w", err) - } - } - - parser, err := rc.parsers.GetParser(ctx, repo) - if err != nil { - return fmt.Errorf("failed to get parser for %s: %w", repo.Config().Name, err) - } - - id, err := rc.identities.WorkerIdentity(ctx, namespace) - if err != nil { - return err - } - ctx = identity.WithRequester(ctx, id) - - replicator, err := jobs.NewSyncer(repo, rc.resourceLister, parser) - if err != nil { - return fmt.Errorf("error creating replicator") - } - - if err := replicator.Unsync(ctx); err != nil { - return fmt.Errorf("unsync repository: %w", err) - } - - return nil - } - - cachedRepo, err := rc.repoLister.Repositories(namespace).Get(name) + obj, err := rc.repoLister.Repositories(namespace).Get(name) switch { case apierrors.IsNotFound(err): return errors.New("repository not found in cache") @@ -277,52 +221,78 @@ func (rc *RepositoryController) process(item *queueItem) error { return err } - id, err := rc.identities.WorkerIdentity(ctx, cachedRepo.Namespace) - if err != nil { - return err - } - ctx = identity.WithRequester(ctx, id) - logger = logger.With("repository", cachedRepo.Name, "namespace", cachedRepo.Namespace) - ctx = logging.Context(ctx, logger) - - repo, err := rc.repoGetter.AsRepository(ctx, cachedRepo) - if err != nil { - return fmt.Errorf("unable to create repository from configuration: %w", err) - } - - hasSpecChanged := cachedRepo.Generation != cachedRepo.Status.ObservedGeneration - if !hasSpecChanged { + healthAge := time.Since(time.UnixMilli(obj.Status.Health.Checked)) + hasSpecChanged := obj.Generation != obj.Status.ObservedGeneration + if !hasSpecChanged && obj.DeletionTimestamp == nil && (healthAge < time.Hour*4) { logger.Info("repository spec unchanged") return nil } - logger.Info("repository spec changed", "previous", cachedRepo.Status.ObservedGeneration, "current", cachedRepo.Generation) + ctx := context.Background() + id, err := rc.identities.WorkerIdentity(ctx, namespace) + if err != nil { + return err + } + ctx = identity.WithRequester(ctx, id) + logger = logger.WithContext(ctx) - var status *provisioning.RepositoryStatus - hooks, ok := repo.(repository.RepositoryHooks) - if ok { - if cachedRepo.Status.ObservedGeneration > 0 { - logger.Info("handle repository update") - status, err = hooks.OnUpdate(ctx) + repo, err := rc.repoGetter.AsRepository(ctx, obj) + if err != nil { + return fmt.Errorf("unable to create repository from configuration: %w", err) + } + + // Safe to edit the repository from here + obj = obj.DeepCopy() + hooks, _ := repo.(repository.RepositoryHooks) + status := &obj.Status + var sync *provisioning.SyncOptions + + switch { + // Delete (note this switch does not fallthrough to the health check) + case obj.DeletionTimestamp != nil: + logger.Info("handle repository delete") + + // Process any finalizers + if len(obj.Finalizers) > 0 { + err = rc.finalizer.process(ctx, repo, obj.Finalizers) if err != nil { - return fmt.Errorf("handle repository update: %w", err) + return fmt.Errorf("error running finalizers %w", err) } - } else { - logger.Info("handle repository init") + + // remove the finalizers + _, err = rc.client.Repositories(obj.GetNamespace()). + Patch(ctx, obj.Name, types.JSONPatchType, []byte(`[ + { "op": "remove", "path": "/metadata/finalizers" } + ]`), v1.PatchOptions{ + FieldManager: "repository-controller", + }) + } + return err // delete will be called again + + // Create + case obj.Status.ObservedGeneration < 1: + logger.Info("handle repository create") + if hooks != nil { status, err = hooks.OnCreate(ctx) - if err != nil { - return fmt.Errorf("handle repository create: %w", err) - } + } + sync = &provisioning.SyncOptions{Complete: true} + + // Update + default: + logger.Info("handle repository update") + if hooks != nil { + status, err = hooks.OnUpdate(ctx) + } + sync = &provisioning.SyncOptions{ + Complete: hasSpecChanged, } } - if status == nil { - status = cachedRepo.Status.DeepCopy() + if err != nil { + return err } - status.ObservedGeneration = cachedRepo.Generation - if time.Since(time.UnixMilli(cachedRepo.Status.Health.Checked)) < 200*time.Millisecond { - logger.Info("skipping health check as it was recently checked") - } else { + status.ObservedGeneration = obj.Generation + if obj.DeletionTimestamp == nil && healthAge > 200*time.Millisecond { res, err := rc.tester.TestRepository(ctx, repo) if err != nil { res = &provisioning.TestResults{ @@ -334,44 +304,65 @@ func (rc *RepositoryController) process(item *queueItem) error { } } + status.Health = provisioning.HealthStatus{ + Healthy: res.Success, + Checked: time.Now().UnixMilli(), + Message: res.Errors, + } if !res.Success { logger.Error("repository is unhealthy", "errors", res.Errors) } - - status.Health = provisioning.HealthStatus{ - Checked: status.Health.Checked, - Healthy: res.Success, - Message: res.Errors, - } } - cfg := cachedRepo.DeepCopy() - cfg.Status = *status - if _, err := rc.client.Repositories(cachedRepo.GetNamespace()). - UpdateStatus(ctx, cfg, v1.UpdateOptions{}); err != nil { + // Queue a sync job + hash := status.Sync.Hash + if status.Sync.State == provisioning.JobStateWorking { + res := rc.jobs.Status(ctx, obj.Namespace, status.Sync.JobID) + if res == nil { + logger.Warn("error getting job status", "job", status.Sync.JobID) + res = &provisioning.JobStatus{ + State: provisioning.JobStateError, + Errors: []string{"unable to find status for job: " + status.Sync.JobID}, + } + } + status.Sync = res.ToSyncStatus(status.Sync.JobID) + } + + // Maybe add a new sync job + if status.Health.Healthy && sync != nil && status.Sync.State != provisioning.JobStateWorking { + job, err := rc.jobs.Add(ctx, &provisioning.Job{ + ObjectMeta: v1.ObjectMeta{ + Namespace: obj.Namespace, + }, + Spec: provisioning.JobSpec{ + Repository: obj.GetName(), + Action: provisioning.JobActionSync, + Sync: sync, + }, + }) + if err != nil { + logger.Error("error adding sync job", "error", err) + status.Sync = provisioning.SyncStatus{ + State: provisioning.JobStateError, + Started: time.Now().UnixMilli(), + Message: []string{ + "Error starting sync job", + err.Error(), + }, + } + } else { + status.Sync = job.Status.ToSyncStatus(job.Name) + logger.Info("sync job triggered", "job", job.Name) + } + } + status.Sync.Hash = hash + obj.Status = *status + + // Write the updated status (careful not to trigger inf loop) + if _, err := rc.client.Repositories(obj.GetNamespace()). + UpdateStatus(ctx, obj, v1.UpdateOptions{}); err != nil { 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{ - Repository: cachedRepo.GetName(), - Action: provisioning.JobActionSync, - Sync: &provisioning.SyncOptions{ - // We force on configuration changes - Complete: true, - }, - }, - }) - if err != nil { - return fmt.Errorf("trigger sync job: %w", err) - } - logger.Info("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 fe4ba6d704b..230e39092fa 100644 --- a/pkg/registry/apis/provisioning/test.go +++ b/pkg/registry/apis/provisioning/test.go @@ -173,12 +173,6 @@ func ValidateRepository(repo repository.Repository) field.ErrorList { cfg.Spec.GitHub, "Local config only valid when type is local")) } - switch cfg.Spec.DeletePolicy { - case provisioning.DeletePolityRetain, provisioning.DeletePolityClean: - default: - list = append(list, field.Invalid(field.NewPath("spec", "deletePolicy"), cfg.Spec.DeletePolicy, "Invalid unsync mode")) - } - if cfg.Spec.Type != provisioning.GitHubRepositoryType && cfg.Spec.GitHub != nil { list = append(list, field.Invalid(field.NewPath("spec", "github"), cfg.Spec.GitHub, "Github config only valid when type is github")) diff --git a/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json b/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json index d562c86aafe..6681d5f6886 100644 --- a/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json +++ b/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json @@ -4089,15 +4089,9 @@ "required": [ "title", "editing", - "deletePolicy", "type" ], "properties": { - "deletePolicy": { - "description": "DeletePolicy options within the repository", - "type": "string", - "default": "" - }, "description": { "description": "Describe the feature toggle", "type": "string" diff --git a/public/app/features/provisioning/api/endpoints.gen.ts b/public/app/features/provisioning/api/endpoints.gen.ts index 341efed4a04..db6d3b39797 100644 --- a/public/app/features/provisioning/api/endpoints.gen.ts +++ b/public/app/features/provisioning/api/endpoints.gen.ts @@ -731,8 +731,6 @@ export type ComGithubGrafanaGrafanaPkgApisProvisioningV0Alpha1S3RepositoryConfig region?: string; }; export type ComGithubGrafanaGrafanaPkgApisProvisioningV0Alpha1RepositorySpec = { - /** DeletePolicy options within the repository */ - deletePolicy: string; /** Describe the feature toggle */ description?: string; /** Edit options within the repository */ diff --git a/public/app/features/provisioning/api/endpoints.ts b/public/app/features/provisioning/api/endpoints.ts index 303b9734f06..58f7b4cdbac 100644 --- a/public/app/features/provisioning/api/endpoints.ts +++ b/public/app/features/provisioning/api/endpoints.ts @@ -236,19 +236,16 @@ const injectedRtkApi = api overrideExisting: false, }); export { injectedRtkApi as generatedAPI }; -export type ListJobResponse = JobList; -export type ListJobArg = { -}; -export type GetJobResponse = Job; +export type ListJobResponse = JobList; +export type ListJobArg = {}; +export type GetJobResponse = Job; export type GetJobArg = { name: string; pretty?: string; }; -export type ListRepositoryResponse = - RepositoryList; -export type ListRepositoryArg = { -}; -export type CreateRepositoryResponse = | Repository; +export type ListRepositoryResponse = RepositoryList; +export type ListRepositoryArg = {}; +export type CreateRepositoryResponse = Repository; export type CreateRepositoryArg = { pretty?: string; dryRun?: string; @@ -256,12 +253,12 @@ export type CreateRepositoryArg = { fieldValidation?: string; body: Repository; }; -export type GetRepositoryResponse = Repository; +export type GetRepositoryResponse = Repository; export type GetRepositoryArg = { name: string; pretty?: string; }; -export type ReplaceRepositoryResponse = | Repository; +export type ReplaceRepositoryResponse = Repository; export type ReplaceRepositoryArg = { name: string; pretty?: string; @@ -270,7 +267,7 @@ export type ReplaceRepositoryArg = { fieldValidation?: string; body: Repository; }; -export type DeleteRepositoryResponse = | Status; +export type DeleteRepositoryResponse = Status; export type DeleteRepositoryArg = { name: string; pretty?: string; @@ -281,7 +278,7 @@ export type DeleteRepositoryArg = { propagationPolicy?: string; body: DeleteOptions; }; -export type CreateRepositoryExportResponse = Job; +export type CreateRepositoryExportResponse = Job; export type CreateRepositoryExportArg = { name: string; body: { @@ -291,7 +288,7 @@ export type CreateRepositoryExportArg = { prefix?: string; }; }; -export type GetRepositoryFilesResponse = { +export type GetRepositoryFilesResponse = { apiVersion?: string; items?: any[]; kind?: string; @@ -301,15 +298,13 @@ export type GetRepositoryFilesArg = { name: string; ref?: string; }; -export type GetRepositoryFilesWithPathResponse = - ResourceWrapper; +export type GetRepositoryFilesWithPathResponse = ResourceWrapper; export type GetRepositoryFilesWithPathArg = { name: string; path: string; ref?: string; }; -export type ReplaceRepositoryFilesWithPathResponse = - ResourceWrapper; +export type ReplaceRepositoryFilesWithPathResponse = ResourceWrapper; export type ReplaceRepositoryFilesWithPathArg = { name: string; path: string; @@ -319,8 +314,7 @@ export type ReplaceRepositoryFilesWithPathArg = { [key: string]: any; }; }; -export type CreateRepositoryFilesWithPathResponse = - ResourceWrapper; +export type CreateRepositoryFilesWithPathResponse = ResourceWrapper; export type CreateRepositoryFilesWithPathArg = { name: string; path: string; @@ -330,37 +324,34 @@ export type CreateRepositoryFilesWithPathArg = { [key: string]: any; }; }; -export type DeleteRepositoryFilesWithPathResponse = - ResourceWrapper; +export type DeleteRepositoryFilesWithPathResponse = ResourceWrapper; export type DeleteRepositoryFilesWithPathArg = { name: string; path: string; ref?: string; message?: string; }; -export type GetRepositoryHistoryResponse = string; +export type GetRepositoryHistoryResponse = string; export type GetRepositoryHistoryArg = { name: string; ref?: string; }; -export type GetRepositoryHistoryWithPathResponse = string; +export type GetRepositoryHistoryWithPathResponse = string; export type GetRepositoryHistoryWithPathArg = { name: string; path: string; ref?: string; }; -export type GetRepositoryResourcesResponse = - ResourceList; +export type GetRepositoryResourcesResponse = ResourceList; export type GetRepositoryResourcesArg = { name: string; }; -export type GetRepositoryStatusResponse = - Repository; +export type GetRepositoryStatusResponse = Repository; export type GetRepositoryStatusArg = { name: string; pretty?: string; }; -export type ReplaceRepositoryStatusResponse = | Repository; +export type ReplaceRepositoryStatusResponse = Repository; export type ReplaceRepositoryStatusArg = { name: string; pretty?: string; @@ -369,15 +360,14 @@ export type ReplaceRepositoryStatusArg = { fieldValidation?: string; body: Repository; }; -export type CreateRepositorySyncResponse = Job; +export type CreateRepositorySyncResponse = Job; export type CreateRepositorySyncArg = { name: string; body: { complete?: boolean; }; }; -export type CreateRepositoryTestResponse = - TestResults; +export type CreateRepositoryTestResponse = TestResults; export type CreateRepositoryTestArg = { name: string; body: { @@ -388,24 +378,21 @@ export type CreateRepositoryTestArg = { status?: any; }; }; -export type GetRepositoryWebhookResponse = - WebhookResponse; +export type GetRepositoryWebhookResponse = WebhookResponse; export type GetRepositoryWebhookArg = { name: string; }; -export type CreateRepositoryWebhookResponse = - WebhookResponse; +export type CreateRepositoryWebhookResponse = WebhookResponse; export type CreateRepositoryWebhookArg = { name: string; }; -export type GetResourceStatsResponse = { +export type GetResourceStatsResponse = { apiVersion?: string; items?: any[]; kind?: string; metadata?: any; }; -export type GetResourceStatsArg = { -}; +export type GetResourceStatsArg = {}; export type Time = string; export type FieldsV1 = object; export type ManagedFieldsEntry = { @@ -526,7 +513,6 @@ export type S3RepositoryConfig = { region?: string; }; export type RepositorySpec = { - deletePolicy: string; description?: string; editing: EditingOptions; folder?: string; diff --git a/scripts/generate-rtk-apis.ts b/scripts/generate-rtk-apis.ts index fca66e438f0..da0f0c68e4c 100644 --- a/scripts/generate-rtk-apis.ts +++ b/scripts/generate-rtk-apis.ts @@ -38,7 +38,7 @@ const config: ConfigFile = { }, '../public/app/features/provisioning/api/endpoints.gen.ts': { apiFile: '../public/app/features/provisioning/api/baseAPI.ts', - schemaFile: '../openapi/provisioning.grafana.app-v0alpha1.json', + schemaFile: '../pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json', apiImport: 'baseAPI', filterEndpoints, argSuffix: 'Arg',