[PROVISIONING] Add Delete Policy to clean up resources on repository deletion (#99457)
* Add unsync mode to spec * Implement unsync modes * Eliminate one mode and use `deletePolicy` instead * Betterer changes * Fix leftover * Add TODO for repository resources * Revert frontend change for now * Revert remaining frontend changes * Revert betterer * Revert betterer
This commit is contained in:
1 parent
5cdaf77bc9
commit
2254a85b03
8 files changed
+166
-45
No files matched your search
@@ -84,6 +84,9 @@ 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"`
|
||||
|
||||
@@ -104,6 +107,15 @@ 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"`
|
||||
|
||||
@@ -987,6 +987,14 @@ 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\"`",
|
||||
@@ -1022,7 +1030,7 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositorySpec(ref common.ReferenceCa
|
||||
},
|
||||
},
|
||||
},
|
||||
Required: []string{"title", "editing", "type"},
|
||||
Required: []string{"title", "editing", "deletePolicy", "type"},
|
||||
},
|
||||
},
|
||||
Dependencies: []string{
|
||||
|
||||
@@ -11,15 +11,16 @@ 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"`
|
||||
Type *provisioningv0alpha1.RepositoryType `json:"type,omitempty"`
|
||||
Linting *bool `json:"linting,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"`
|
||||
DeletePolicy *provisioningv0alpha1.DeletePolicy `json:"deletePolicy,omitempty"`
|
||||
Type *provisioningv0alpha1.RepositoryType `json:"type,omitempty"`
|
||||
Linting *bool `json:"linting,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,6 +61,14 @@ 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.
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"path"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
@@ -16,36 +17,38 @@ import (
|
||||
|
||||
"github.com/grafana/grafana-app-sdk/logging"
|
||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
dashboards "github.com/grafana/grafana/pkg/apis/dashboard"
|
||||
dashboardsv2alpha1 "github.com/grafana/grafana/pkg/apis/dashboard/v2alpha1"
|
||||
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"
|
||||
)
|
||||
|
||||
type UnsyncOptions struct {
|
||||
KeepDashboards bool
|
||||
}
|
||||
|
||||
type Syncer struct {
|
||||
client *resources.DynamicClient
|
||||
parser *resources.Parser
|
||||
lister resources.ResourceLister
|
||||
folders dynamic.ResourceInterface
|
||||
repository repository.Repository
|
||||
}
|
||||
|
||||
func NewSyncer(
|
||||
repo repository.Repository,
|
||||
lister resources.ResourceLister,
|
||||
parser *resources.Parser,
|
||||
) (*Syncer, error) {
|
||||
dynamicClient := parser.Client()
|
||||
folders := dynamicClient.Resource(schema.GroupVersionResource{
|
||||
Group: "folder.grafana.app",
|
||||
Version: "v0alpha1",
|
||||
Resource: "folders",
|
||||
Group: folders.GROUP,
|
||||
Version: folders.VERSION,
|
||||
Resource: folders.RESOURCE,
|
||||
})
|
||||
|
||||
return &Syncer{
|
||||
parser: parser,
|
||||
client: dynamicClient,
|
||||
lister: lister,
|
||||
folders: folders,
|
||||
repository: repo,
|
||||
}, nil
|
||||
@@ -101,34 +104,101 @@ func (r *Syncer) Sync(ctx context.Context) (string, error) {
|
||||
return latest, nil
|
||||
}
|
||||
|
||||
func (r *Syncer) Unsync(ctx context.Context, opts UnsyncOptions) error {
|
||||
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)
|
||||
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")
|
||||
|
||||
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)
|
||||
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 (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)
|
||||
}
|
||||
|
||||
// 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, "/"))
|
||||
}
|
||||
})
|
||||
|
||||
logger := logging.FromContext(ctx)
|
||||
logger.Info("deleting resources", "count", len(list.Items))
|
||||
for _, item := range list.Items {
|
||||
// HACK: until we know how to get the version
|
||||
var version string
|
||||
switch item.Resource {
|
||||
case folders.RESOURCE:
|
||||
version = folders.VERSION
|
||||
case dashboards.DASHBOARD_RESOURCE:
|
||||
version = dashboardsv2alpha1.VERSION
|
||||
default:
|
||||
return fmt.Errorf("unknown resource api version: %s", item.Resource)
|
||||
}
|
||||
|
||||
meta, err := utils.MetaAccessor(obj)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create meta accessor from folder object: %w", err)
|
||||
client := r.client.Resource(schema.GroupVersionResource{
|
||||
Group: item.Group,
|
||||
Version: version,
|
||||
Resource: item.Resource,
|
||||
})
|
||||
|
||||
logger := logger.With("group", item.Group, "resource", item.Resource, "name", item.Name, "version", version)
|
||||
if err := client.Delete(ctx, item.Name, metav1.DeleteOptions{}); err != nil {
|
||||
logger.Error("failed to delete resource", "error", err)
|
||||
return fmt.Errorf("delete resource: %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")
|
||||
logger.Info("resource deleted")
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
@@ -104,7 +104,7 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job, progress
|
||||
return nil, fmt.Errorf("update repository status: %w", err)
|
||||
}
|
||||
|
||||
syncer, err := NewSyncer(repo, parser)
|
||||
syncer, err := NewSyncer(repo, g.lister, parser)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error creating replicator")
|
||||
}
|
||||
|
||||
@@ -321,6 +321,10 @@ 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
|
||||
}
|
||||
|
||||
if r.Spec.Type == provisioning.GitHubRepositoryType {
|
||||
if r.Spec.GitHub == nil {
|
||||
return fmt.Errorf("github configuration is required")
|
||||
@@ -475,6 +479,7 @@ func (b *ProvisioningAPIBuilder) GetPostStartHooks() (map[string]genericapiserve
|
||||
c.ProvisioningV0alpha1(),
|
||||
repoInformer,
|
||||
b, // repoGetter
|
||||
b.lister,
|
||||
b.parsers,
|
||||
b.identities,
|
||||
b.tester,
|
||||
|
||||
@@ -46,10 +46,11 @@ type queueItem struct {
|
||||
|
||||
// RepositoryController controls how and when CRD is established.
|
||||
type RepositoryController struct {
|
||||
client client.ProvisioningV0alpha1Interface
|
||||
repoLister listers.RepositoryLister
|
||||
repoSynced cache.InformerSynced
|
||||
parsers *resources.ParserFactory
|
||||
client client.ProvisioningV0alpha1Interface
|
||||
resourceLister resources.ResourceLister
|
||||
repoLister listers.RepositoryLister
|
||||
repoSynced cache.InformerSynced
|
||||
parsers *resources.ParserFactory
|
||||
|
||||
jobs jobs.JobQueue
|
||||
|
||||
@@ -71,15 +72,17 @@ func NewRepositoryController(
|
||||
provisioningClient client.ProvisioningV0alpha1Interface,
|
||||
repoInformer informer.RepositoryInformer,
|
||||
repoGetter RepoGetter,
|
||||
resourceLister resources.ResourceLister,
|
||||
parsers *resources.ParserFactory,
|
||||
identities auth.BackgroundIdentityService,
|
||||
tester *RepositoryTester,
|
||||
jobs jobs.JobQueue,
|
||||
) (*RepositoryController, error) {
|
||||
rc := &RepositoryController{
|
||||
client: provisioningClient,
|
||||
repoLister: repoInformer.Lister(),
|
||||
repoSynced: repoInformer.Informer().HasSynced,
|
||||
client: provisioningClient,
|
||||
resourceLister: resourceLister,
|
||||
repoLister: repoInformer.Lister(),
|
||||
repoSynced: repoInformer.Informer().HasSynced,
|
||||
queue: workqueue.NewTypedRateLimitingQueueWithConfig(
|
||||
workqueue.DefaultTypedControllerRateLimiter[*queueItem](),
|
||||
workqueue.TypedRateLimitingQueueConfig[*queueItem]{
|
||||
@@ -190,6 +193,8 @@ func (rc *RepositoryController) processNextWorkItem(ctx context.Context) bool {
|
||||
|
||||
item.attempts++
|
||||
logger = logger.With("error", err, "attempts", item.attempts)
|
||||
logger.Error("RepositoryController failed to process key")
|
||||
|
||||
if item.attempts >= maxAttempts {
|
||||
logger.Error("RepositoryController failed too many times")
|
||||
rc.queue.Forget(item)
|
||||
@@ -246,12 +251,18 @@ func (rc *RepositoryController) process(item *queueItem) error {
|
||||
return fmt.Errorf("failed to get parser for %s: %w", repo.Config().Name, err)
|
||||
}
|
||||
|
||||
replicator, err := jobs.NewSyncer(repo, parser)
|
||||
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, jobs.UnsyncOptions{}); err != nil {
|
||||
if err := replicator.Unsync(ctx); err != nil {
|
||||
return fmt.Errorf("unsync repository: %w", err)
|
||||
}
|
||||
|
||||
|
||||
@@ -173,6 +173,12 @@ 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"))
|
||||
|
||||
Reference in new issue
Block a user