Provisioning: Support finalizers in our controller (#99532)

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
 }

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"`

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"`

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{

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.

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, "/"))
+		}
+	})
+}

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

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

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

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,

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

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
 }

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

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"

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 */

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;

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',