Repository: kumahq/kuma
Pull request #1784 (2021-04-08 15:13:59)

chore(kuma-cp) add extension points to res manager and webhooks
### Summary

### Full changelog

* Add ability to add or override custom resource manager
* Add ability to add custom validators on K8S
* More graceful events stop

### Documentation

- [X] No docs, internal changes only



Patch:
diff --git a/pkg/core/resources/manager/customizable_manager.go b/pkg/core/resources/manager/customizable_manager.go
index 88e15416f..6f9ce0a58 100644
--- a/pkg/core/resources/manager/customizable_manager.go
+++ b/pkg/core/resources/manager/customizable_manager.go
@@ -7,43 +7,55 @@ import (
 	"github.com/kumahq/kuma/pkg/core/resources/store"
 )
 
-func NewCustomizableResourceManager(defaultManager ResourceManager, customManagers map[model.ResourceType]ResourceManager) ResourceManager {
+type CustomizableResourceManager interface {
+	ResourceManager
+	Customize(model.ResourceType, ResourceManager)
+	ResourceManager(model.ResourceType) ResourceManager
+}
+
+func NewCustomizableResourceManager(defaultManager ResourceManager, customManagers map[model.ResourceType]ResourceManager) CustomizableResourceManager {
 	return &customizableResourceManager{
 		defaultManager: defaultManager,
 		customManagers: customManagers,
 	}
 }
 
+var _ CustomizableResourceManager = &customizableResourceManager{}
+
 type customizableResourceManager struct {
 	defaultManager ResourceManager
 	customManagers map[model.ResourceType]ResourceManager
 }
 
+func (m *customizableResourceManager) Customize(resourceType model.ResourceType, manager ResourceManager) {
+	m.customManagers[resourceType] = manager
+}
+
 func (m *customizableResourceManager) Get(ctx context.Context, resource model.Resource, fs ...store.GetOptionsFunc) error {
-	return m.resourceManager(resource.GetType()).Get(ctx, resource, fs...)
+	return m.ResourceManager(resource.GetType()).Get(ctx, resource, fs...)
 }
 
 func (m *customizableResourceManager) List(ctx context.Context, list model.ResourceList, fs ...store.ListOptionsFunc) error {
-	return m.resourceManager(list.GetItemType()).List(ctx, list, fs...)
+	return m.ResourceManager(list.GetItemType()).List(ctx, list, fs...)
 }
 
 func (m *customizableResourceManager) Create(ctx context.Context, resource model.Resource, fs ...store.CreateOptionsFunc) error {
-	return m.resourceManager(resource.GetType()).Create(ctx, resource, fs...)
+	return m.ResourceManager(resource.GetType()).Create(ctx, resource, fs...)
 }
 
 func (m *customizableResourceManager) Delete(ctx context.Context, resource model.Resource, fs ...store.DeleteOptionsFunc) error {
-	return m.resourceManager(resource.GetType()).Delete(ctx, resource, fs...)
+	return m.ResourceManager(resource.GetType()).Delete(ctx, resource, fs...)
 }
 
 func (m *customizableResourceManager) DeleteAll(ctx context.Context, list model.ResourceList, fs ...store.DeleteAllOptionsFunc) error {
-	return m.resourceManager(list.GetItemType()).DeleteAll(ctx, list, fs...)
+	return m.ResourceManager(list.GetItemType()).DeleteAll(ctx, list, fs...)
 }
 
 func (m *customizableResourceManager) Update(ctx context.Context, resource model.Resource, fs ...store.UpdateOptionsFunc) error {
-	return m.resourceManager(resource.GetType()).Update(ctx, resource, fs...)
+	return m.ResourceManager(resource.GetType()).Update(ctx, resource, fs...)
 }
 
-func (m *customizableResourceManager) resourceManager(typ model.ResourceType) ResourceManager {
+func (m *customizableResourceManager) ResourceManager(typ model.ResourceType) ResourceManager {
 	if customManager, ok := m.customManagers[typ]; ok {
 		return customManager
 	}
diff --git a/pkg/core/runtime/builder.go b/pkg/core/runtime/builder.go
index e585fd88b..a83528637 100644
--- a/pkg/core/runtime/builder.go
+++ b/pkg/core/runtime/builder.go
@@ -37,7 +37,7 @@ type BuilderContext interface {
 	ResourceStore() core_store.ResourceStore
 	SecretStore() store.SecretStore
 	ConfigStore() core_store.ResourceStore
-	ResourceManager() core_manager.ResourceManager
+	ResourceManager() core_manager.CustomizableResourceManager
 	Config() kuma_cp.Config
 	DataSourceLoader() datasource.Loader
 	Extensions() context.Context
@@ -61,7 +61,7 @@ type Builder struct {
 	rs         core_store.ResourceStore
 	ss         store.SecretStore
 	cs         core_store.ResourceStore
-	rm         core_manager.ResourceManager
+	rm         core_manager.CustomizableResourceManager
 	rom        core_manager.ReadOnlyResourceManager
 	cam        core_ca.Managers
 	dsl        datasource.Loader
@@ -118,7 +118,7 @@ func (b *Builder) WithConfigStore(cs core_store.ResourceStore) *Builder {
 	return b
 }
 
-func (b *Builder) WithResourceManager(rm core_manager.ResourceManager) *Builder {
+func (b *Builder) WithResourceManager(rm core_manager.CustomizableResourceManager) *Builder {
 	b.rm = rm
 	return b
 }
@@ -297,7 +297,7 @@ func (b *Builder) SecretStore() store.SecretStore {
 func (b *Builder) ConfigStore() core_store.ResourceStore {
 	return b.cs
 }
-func (b *Builder) ResourceManager() core_manager.ResourceManager {
+func (b *Builder) ResourceManager() core_manager.CustomizableResourceManager {
 	return b.rm
 }
 func (b *Builder) ReadOnlyResourceManager() core_manager.ReadOnlyResourceManager {
diff --git a/pkg/events/eventbus.go b/pkg/events/eventbus.go
index 8b4a66f33..8b07fb4cc 100644
--- a/pkg/events/eventbus.go
+++ b/pkg/events/eventbus.go
@@ -54,6 +54,6 @@ func (k *reader) Recv(stop <-chan struct{}) (Event, error) {
 		}
 		return event, nil
 	case <-stop:
-		return nil, errors.New("stop channel was closed")
+		return nil, ListenerStoppedErr
 	}
 }
diff --git a/pkg/events/interfaces.go b/pkg/events/interfaces.go
index b0d2dea4f..1254559cf 100644
--- a/pkg/events/interfaces.go
+++ b/pkg/events/interfaces.go
@@ -1,6 +1,8 @@
 package events
 
 import (
+	"github.com/pkg/errors"
+
 	"github.com/kumahq/kuma/pkg/core/resources/model"
 )
 
@@ -20,6 +22,8 @@ type ResourceChangedEvent struct {
 	Key       model.ResourceKey
 }
 
+var ListenerStoppedErr = errors.New("listener closed")
+
 type Listener interface {
 	Recv(stop <-chan struct{}) (Event, error)
 }
diff --git a/pkg/insights/resyncer.go b/pkg/insights/resyncer.go
index 4cda019fc..18d848596 100644
--- a/pkg/insights/resyncer.go
+++ b/pkg/insights/resyncer.go
@@ -99,6 +99,9 @@ func (r *resyncer) Start(stop <-chan struct{}) error {
 	eventReader := r.eventFactory.New()
 	for {
 		event, err := eventReader.Recv(stop)
+		if err == events.ListenerStoppedErr {
+			return nil
+		}
 		if err != nil {
 			return err
 		}
diff --git a/pkg/kds/context/context.go b/pkg/kds/context/context.go
index 2d722048c..982a777f8 100644
--- a/pkg/kds/context/context.go
+++ b/pkg/kds/context/context.go
@@ -23,23 +23,26 @@ type Context struct {
 	GlobalProvidedFilter  reconcile.ResourceFilter
 	RemoteProvidedFilter  reconcile.ResourceFilter
 	// Configs contains the names of system.ConfigResource that will be transferred from Global to Remote
-	Configs []string
+	Configs map[string]bool
 }
 
 func DefaultContext(manager manager.ResourceManager, zone string) *Context {
+	configs := map[string]bool{
+		config_manager.ClusterIdConfigKey: true,
+	}
 	return &Context{
 		RemoteClientCtx:      context.Background(),
-		GlobalProvidedFilter: GlobalProvidedFilter(manager),
+		GlobalProvidedFilter: GlobalProvidedFilter(manager, configs),
 		RemoteProvidedFilter: RemoteProvidedFilter(zone),
-		Configs:              []string{config_manager.ClusterIdConfigKey},
+		Configs:              configs,
 	}
 }
 
 // GlobalProvidedFilter returns ResourceFilter which filters Resources provided by Global, specifically
 // excludes Dataplanes and Ingresses from 'clusterID' cluster
-func GlobalProvidedFilter(rm manager.ResourceManager) reconcile.ResourceFilter {
+func GlobalProvidedFilter(rm manager.ResourceManager, configs map[string]bool) reconcile.ResourceFilter {
 	return func(clusterID string, r model.Resource) bool {
-		if r.GetType() == system.ConfigType && r.GetMeta().GetName() != config_manager.ClusterIdConfigKey {
+		if r.GetType() == system.ConfigType && !configs[r.GetMeta().GetName()] {
 			return false
 		}
 		if r.GetType() != mesh.DataplaneType {
diff --git a/pkg/kds/remote/components.go b/pkg/kds/remote/components.go
index 31e8ebf33..6f7d23396 100644
--- a/pkg/kds/remote/components.go
+++ b/pkg/kds/remote/components.go
@@ -107,12 +107,7 @@ func Callbacks(rt core_runtime.Runtime, syncer sync_store.ResourceSyncer, k8sSto
 			}
 			if rs.GetItemType() == system.ConfigType {
 				return syncer.Sync(rs, sync_store.PrefilterBy(func(r model.Resource) bool {
-					for _, config := range rt.KDSContext().Configs {
-						if config == r.GetMeta().GetName() {
-							return true
-						}
-					}
-					return false
+					return rt.KDSContext().Configs[r.GetMeta().GetName()]
 				}))
 			}
 			return syncer.Sync(rs)
diff --git a/pkg/plugins/bootstrap/k8s/plugin.go b/pkg/plugins/bootstrap/k8s/plugin.go
index 1b451c5a4..d7c1c480a 100644
--- a/pkg/plugins/bootstrap/k8s/plugin.go
+++ b/pkg/plugins/bootstrap/k8s/plugin.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/kumahq/kuma/pkg/core"
 	kuma_kube_cache "github.com/kumahq/kuma/pkg/plugins/bootstrap/k8s/cache"
+	k8s_common "github.com/kumahq/kuma/pkg/plugins/common/k8s"
 
 	"github.com/kumahq/kuma/pkg/plugins/resources/k8s"
 
@@ -69,6 +70,7 @@ func (p *plugin) BeforeBootstrap(b *core_runtime.Builder, _ core_plugins.PluginC
 	} else {
 		b.WithExtensions(k8s_extensions.NewResourceConverterContext(b.Extensions(), k8s.NewSimpleConverter()))
 	}
+	b.WithExtensions(k8s_extensions.NewCompositeValidatorContext(b.Extensions(), &k8s_common.CompositeValidator{}))
 	return nil
 }
 
diff --git a/pkg/plugins/runtime/k8s/webhooks/composite_validator.go b/pkg/plugins/common/k8s/composite_validator.go
similarity index 98%
rename from pkg/plugins/runtime/k8s/webhooks/composite_validator.go
rename to pkg/plugins/common/k8s/composite_validator.go
index 93a88e545..0ec18bb4a 100644
--- a/pkg/plugins/runtime/k8s/webhooks/composite_validator.go
+++ b/pkg/plugins/common/k8s/composite_validator.go
@@ -1,4 +1,4 @@
-package webhooks
+package k8s
 
 import (
 	"context"
diff --git a/pkg/plugins/extensions/k8s/context.go b/pkg/plugins/extensions/k8s/context.go
index 8137a75e4..6ad14b9e3 100644
--- a/pkg/plugins/extensions/k8s/context.go
+++ b/pkg/plugins/extensions/k8s/context.go
@@ -45,3 +45,14 @@ func FromSecretClientContext(ctx context.Context) (client kube_client.Client, ok
 	client, ok = ctx.Value(secretClient{}).(kube_client.Client)
 	return
 }
+
+type compositeValidatorKey struct{}
+
+func NewCompositeValidatorContext(ctx context.Context, compositeValidator *k8s_common.CompositeValidator) context.Context {
+	return context.WithValue(ctx, compositeValidatorKey{}, compositeValidator)
+}
+
+func FromCompositeValidatorContext(ctx context.Context) (validator *k8s_common.CompositeValidator, ok bool) {
+	validator, ok = ctx.Value(compositeValidatorKey{}).(*k8s_common.CompositeValidator)
+	return
+}
diff --git a/pkg/plugins/runtime/k8s/plugin.go b/pkg/plugins/runtime/k8s/plugin.go
index 4c6518dde..09259bbfc 100644
--- a/pkg/plugins/runtime/k8s/plugin.go
+++ b/pkg/plugins/runtime/k8s/plugin.go
@@ -235,7 +235,10 @@ func generateDefaulterPath(gvk kube_schema.GroupVersionKind) string {
 }
 
 func addValidators(mgr kube_ctrl.Manager, rt core_runtime.Runtime, converter k8s_common.Converter) error {
-	composite := k8s_webhooks.CompositeValidator{}
+	composite, ok := k8s_extensions.FromCompositeValidatorContext(rt.Extensions())
+	if !ok {
+		return errors.Errorf("could not find composite validator in the extensions context")
+	}
 
 	handler := k8s_webhooks.NewValidatingWebhook(converter, core_registry.Global(), k8s_registry.Global(), rt.Config().Mode)
 	composite.AddValidator(handler)
diff --git a/pkg/plugins/runtime/k8s/webhooks/mesh_validator.go b/pkg/plugins/runtime/k8s/webhooks/mesh_validator.go
index c4f747dd6..17fb28927 100644
--- a/pkg/plugins/runtime/k8s/webhooks/mesh_validator.go
+++ b/pkg/plugins/runtime/k8s/webhooks/mesh_validator.go
@@ -17,7 +17,7 @@ import (
 	mesh_k8s "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/api/v1alpha1"
 )
 
-func NewMeshValidatorWebhook(validator managers_mesh.MeshValidator, converter k8s_common.Converter, resourceManager manager.ResourceManager) AdmissionValidator {
+func NewMeshValidatorWebhook(validator managers_mesh.MeshValidator, converter k8s_common.Converter, resourceManager manager.ResourceManager) k8s_common.AdmissionValidator {
 	return &MeshValidator{
 		validator:       validator,
 		converter:       converter,
diff --git a/pkg/plugins/runtime/k8s/webhooks/validation.go b/pkg/plugins/runtime/k8s/webhooks/validation.go
index 007ef4a85..93a415f95 100644
--- a/pkg/plugins/runtime/k8s/webhooks/validation.go
+++ b/pkg/plugins/runtime/k8s/webhooks/validation.go
@@ -23,7 +23,7 @@ import (
 	k8s_registry "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/pkg/registry"
 )
 
-func NewValidatingWebhook(converter k8s_common.Converter, coreRegistry core_registry.TypeRegistry, k8sRegistry k8s_registry.TypeRegistry, mode core.CpMode) AdmissionValidator {
+func NewValidatingWebhook(converter k8s_common.Converter, coreRegistry core_registry.TypeRegistry, k8sRegistry k8s_registry.TypeRegistry, mode core.CpMode) k8s_common.AdmissionValidator {
 	return &validatingHandler{
 		coreRegistry: coreRegistry,
 		k8sRegistry:  k8sRegistry,
diff --git a/pkg/plugins/runtime/k8s/webhooks/zone_validator.go b/pkg/plugins/runtime/k8s/webhooks/zone_validator.go
index c72b4742b..7a96385d3 100644
--- a/pkg/plugins/runtime/k8s/webhooks/zone_validator.go
+++ b/pkg/plugins/runtime/k8s/webhooks/zone_validator.go
@@ -8,10 +8,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
 	"github.com/kumahq/kuma/pkg/core/managers/apis/zone"
+	k8s_common "github.com/kumahq/kuma/pkg/plugins/common/k8s"
 	mesh_k8s "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/api/v1alpha1"
 )
 
-func NewZoneValidatorWebhook(validator zone.Validator) AdmissionValidator {
+func NewZoneValidatorWebhook(validator zone.Validator) k8s_common.AdmissionValidator {
 	return &ZoneValidator{
 		validator: validator,
 	}


Test patch:
diff --git a/pkg/plugins/runtime/k8s/webhooks/composite_validator_test.go b/pkg/plugins/common/k8s/composite_validator_test.go
similarity index 95%
rename from pkg/plugins/runtime/k8s/webhooks/composite_validator_test.go
rename to pkg/plugins/common/k8s/composite_validator_test.go
index e0997b9e5..1336582d1 100644
--- a/pkg/plugins/runtime/k8s/webhooks/composite_validator_test.go
+++ b/pkg/plugins/common/k8s/composite_validator_test.go
@@ -1,13 +1,13 @@
-package webhooks_test
+package k8s_test
 
 import (
 	"context"
 
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
+	k8s_common "github.com/kumahq/kuma/pkg/plugins/common/k8s"
 	mesh_k8s "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/api/v1alpha1"
 	k8s_registry "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/pkg/registry"
 	sample_k8s "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/test/api/sample/v1alpha1"
-	"github.com/kumahq/kuma/pkg/plugins/runtime/k8s/webhooks"
 	sample_proto "github.com/kumahq/kuma/pkg/test/apis/sample/v1alpha1"
 
 	admissionv1beta1 "k8s.io/api/admission/v1beta1"
@@ -24,7 +24,7 @@ import (
 type denyingValidator struct {
 }
 
-var _ webhooks.AdmissionValidator = &denyingValidator{}
+var _ k8s_common.AdmissionValidator = &denyingValidator{}
 
 func (d *denyingValidator) InjectDecoder(*admission.Decoder) error {
 	return nil
@@ -45,7 +45,7 @@ var _ = Describe("Composite Validator", func() {
 	var kubeTypes k8s_registry.TypeRegistry
 
 	BeforeEach(func() {
-		composite := webhooks.CompositeValidator{}
+		composite := k8s_common.CompositeValidator{}
 		composite.AddValidator(&denyingValidator{})
 
 		handler = composite.WebHook()
diff --git a/pkg/test/runtime/runtime.go b/pkg/test/runtime/runtime.go
index 630df7c9e..1410881ed 100644
--- a/pkg/test/runtime/runtime.go
+++ b/pkg/test/runtime/runtime.go
@@ -97,7 +97,7 @@ func initializeDNSResolver(cfg kuma_cp.Config, builder *core_runtime.Builder) er
 	return nil
 }
 
-func newResourceManager(builder *core_runtime.Builder) core_manager.ResourceManager {
+func newResourceManager(builder *core_runtime.Builder) core_manager.CustomizableResourceManager {
 	defaultManager := core_manager.NewResourceManager(builder.ResourceStore())
 	customManagers := map[core_model.ResourceType]core_manager.ResourceManager{}
 	customizableManager := core_manager.NewCustomizableResourceManager(defaultManager, customManagers)