Repository: kumahq/kuma
Pull request #1493 (2021-01-29 13:26:11)

feat(kuma-cp) XDS extensions and new matcher strategy
### Summary

Introducing Hooks, a way to extend Kuma with XDS modification. You can create a new project, have Kuma as a dependency, build a Bootstrap plugin that plugs in XDS hook.

Right now I created one Hook that operates on ResourceSet which allows essentially to do everything you want with XDS resources (add, modify, remove).
The ResourceSetHook may have a slight performance hit, because you need to iterate over all resources of one type. Let's say you want to modify inbound listener, you need to iterate over all listeners. I don't think this is noticeable performance hit, but I left a room to introduce more specific Hooks for builders if needed.

I also introduced a new matching strategy. Instead of matching DataplanePolicy on any inbound and applying on a all resources of the dataplane, you can now match DataplanePolicy on individual inbounds.

### Documentation

- [X] No docs, internal changes



Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 3c044bd7c..02a80d1fa 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -105,7 +105,7 @@ jobs:
     environment:
       GO_VERSION: *go_version
       GO111MODULE: "on"
-      CLANG_FORMAT_PATH: clang-format-12
+      CLANG_FORMAT_PATH: clang-format-13
       # if GOPATH is not set, `golang-ci` fails with an obscure message
       # "ERRO Running error: context loading failed: failed to load program with go/packages: could not determine GOARCH and Go compiler"
       GOPATH: /root/.go-kuma-go
@@ -215,7 +215,7 @@ jobs:
     # we need a large resource class to satisfy the needs of golangci-lint run under `make check`
     resource_class: large
     environment:
-      CLANG_FORMAT_PATH: clang-format-12
+      CLANG_FORMAT_PATH: clang-format-13
     steps:
     - checkout
     - restore_cache:
@@ -243,7 +243,7 @@ jobs:
 
          wget -O - https://apt.llvm.org/llvm-snapshot.gpg.key|apt-key add -
 
-         apt update && apt install -y clang-format-12
+         apt update && apt install -y clang-format-13
     - run:
         name: "Run code generators (go generate, protoc, ...) and code checks (go fmt, go vet, ...)"
         command: make check
diff --git a/app/kuma-dp/cmd/context.go b/app/kuma-dp/cmd/context.go
new file mode 100644
index 000000000..4d62f575a
--- /dev/null
+++ b/app/kuma-dp/cmd/context.go
@@ -0,0 +1,27 @@
+package cmd
+
+import (
+	"crypto/tls"
+	"net/http"
+	"time"
+
+	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/envoy"
+	"github.com/kumahq/kuma/pkg/core/runtime/component"
+	leader_memory "github.com/kumahq/kuma/pkg/plugins/leader/memory"
+)
+
+// RootContext contains variables, functions and components that can be overridden when extending kuma-dp or running the test.
+type RootContext struct {
+	ComponentManager   component.Manager
+	BootstrapGenerator envoy.BootstrapConfigFactoryFunc
+}
+
+func DefaultRootContext() *RootContext {
+	return &RootContext{
+		ComponentManager: component.NewManager(leader_memory.NewNeverLeaderElector()),
+		BootstrapGenerator: envoy.NewRemoteBootstrapGenerator(&http.Client{
+			Timeout:   10 * time.Second,
+			Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}},
+		}),
+	}
+}
diff --git a/app/kuma-dp/cmd/root.go b/app/kuma-dp/cmd/root.go
index e622aa8dc..7447640f7 100644
--- a/app/kuma-dp/cmd/root.go
+++ b/app/kuma-dp/cmd/root.go
@@ -15,8 +15,8 @@ var (
 	dataplaneLog = core.Log.WithName("kuma-dp")
 )
 
-// newRootCmd represents the base command when called without any subcommands.
-func newRootCmd() *cobra.Command {
+// NewRootCmd represents the base command when called without any subcommands.
+func NewRootCmd(rootCtx *RootContext) *cobra.Command {
 	args := struct {
 		logLevel string
 	}{}
@@ -41,13 +41,13 @@ func newRootCmd() *cobra.Command {
 	// root flags
 	cmd.PersistentFlags().StringVar(&args.logLevel, "log-level", kuma_log.InfoLevel.String(), kuma_cmd.UsageOptions("log level", kuma_log.OffLevel, kuma_log.InfoLevel, kuma_log.DebugLevel))
 	// sub-commands
-	cmd.AddCommand(newRunCmd())
+	cmd.AddCommand(newRunCmd(rootCtx))
 	cmd.AddCommand(version.NewVersionCmd())
 	return cmd
 }
 
 func DefaultRootCmd() *cobra.Command {
-	return newRootCmd()
+	return NewRootCmd(DefaultRootContext())
 }
 
 // Execute adds all child commands to the root command and sets flags appropriately.
diff --git a/app/kuma-dp/cmd/run.go b/app/kuma-dp/cmd/run.go
index fb679b96b..0b26780ac 100644
--- a/app/kuma-dp/cmd/run.go
+++ b/app/kuma-dp/cmd/run.go
@@ -1,12 +1,9 @@
 package cmd
 
 import (
-	"crypto/tls"
 	"io/ioutil"
-	"net/http"
 	"os"
 	"path/filepath"
-	"time"
 
 	kumadp_config "github.com/kumahq/kuma/app/kuma-dp/pkg/config"
 
@@ -19,23 +16,13 @@ import (
 	kuma_dp "github.com/kumahq/kuma/pkg/config/app/kuma-dp"
 	config_types "github.com/kumahq/kuma/pkg/config/types"
 	"github.com/kumahq/kuma/pkg/core"
-	"github.com/kumahq/kuma/pkg/core/runtime/component"
-	leader_memory "github.com/kumahq/kuma/pkg/plugins/leader/memory"
 	util_net "github.com/kumahq/kuma/pkg/util/net"
 	kuma_version "github.com/kumahq/kuma/pkg/version"
 )
 
-var (
-	runLog = dataplaneLog.WithName("run")
-	// overridable by tests
-	bootstrapGenerator = envoy.NewRemoteBootstrapGenerator(&http.Client{
-		Timeout:   10 * time.Second,
-		Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}},
-	},
-	)
-)
+var runLog = dataplaneLog.WithName("run")
 
-func newRunCmd() *cobra.Command {
+func newRunCmd(rootCtx *RootContext) *cobra.Command {
 	cfg := kuma_dp.DefaultConfig()
 	cmd := &cobra.Command{
 		Use:   "run",
@@ -116,7 +103,7 @@ func newRunCmd() *cobra.Command {
 
 			dataplane, err := envoy.New(envoy.Opts{
 				Config:    cfg,
-				Generator: bootstrapGenerator,
+				Generator: rootCtx.BootstrapGenerator,
 				Dataplane: dp,
 				Stdout:    cmd.OutOrStdout(),
 				Stderr:    cmd.OutOrStderr(),
@@ -127,13 +114,12 @@ func newRunCmd() *cobra.Command {
 			}
 			server := accesslogs.NewAccessLogServer(cfg.Dataplane)
 
-			componentMgr := component.NewManager(leader_memory.NewNeverLeaderElector())
-			if err := componentMgr.Add(server, dataplane); err != nil {
+			if err := rootCtx.ComponentManager.Add(server, dataplane); err != nil {
 				return err
 			}
 
 			runLog.Info("starting Kuma DP", "version", kuma_version.Build.Version)
-			if err := componentMgr.Start(shouldQuit); err != nil {
+			if err := rootCtx.ComponentManager.Start(shouldQuit); err != nil {
 				runLog.Error(err, "error while running Kuma DP")
 				return err
 			}
diff --git a/pkg/core/bootstrap/bootstrap.go b/pkg/core/bootstrap/bootstrap.go
index 9b8f2dc0d..4a920e23d 100644
--- a/pkg/core/bootstrap/bootstrap.go
+++ b/pkg/core/bootstrap/bootstrap.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/kumahq/kuma/pkg/api-server/customization"
 	"github.com/kumahq/kuma/pkg/core/managers/apis/zone"
+	xds_hooks "github.com/kumahq/kuma/pkg/xds/hooks"
 
 	"github.com/pkg/errors"
 
@@ -90,6 +91,7 @@ func buildRuntime(cfg kuma_cp.Config, closeCh <-chan struct{}) (core_runtime.Run
 
 	builder.WithLookupIP(lookup.CachedLookupIP(net.LookupIP, cfg.General.DNSCacheTTL))
 	builder.WithAPIManager(customization.NewAPIList())
+	builder.WithXDSHooks(&xds_hooks.Hooks{})
 
 	if err := initializeAfterBootstrap(cfg, builder); err != nil {
 		return nil, err
diff --git a/pkg/core/policy/dataplane_matcher.go b/pkg/core/policy/dataplane_matcher.go
index c04ede3c4..60a79e086 100644
--- a/pkg/core/policy/dataplane_matcher.go
+++ b/pkg/core/policy/dataplane_matcher.go
@@ -49,6 +49,56 @@ func SelectDataplanePolicy(dataplane *mesh.DataplaneResource, policies []Datapla
 	return bestPolicy
 }
 
+// SelectInboundDataplanePolicies given a Dataplane definition and a list of DataplanePolicy returns the "best matching" DataplanePolicy for each inbound separately.
+// A DataplanePolicy for an inbound is considered a match if the inbound matches all the tags listed in selector of the DataplanePolicy.
+// Every matching DataplanePolicy gets a rank (score) defined as a maximum number of tags in a matching selector.
+// DataplanePolicy with an empty list of selectors is considered a match with a rank (score) of 0.
+// DataplanePolicy with an empty selector (one that has no tags) is considered a match with a rank (score) of 0.
+// In case if there are multiple DataplanePolicies with the same rank (score), the policy created last is chosen.
+func SelectInboundDataplanePolicies(dataplane *mesh.DataplaneResource, policies []DataplanePolicy) InboundDataplanePolicyMap {
+	sort.Stable(DataplanePolicyByName(policies)) // sort to avoid flakiness
+
+	match := InboundDataplanePolicyMap{}
+
+	for _, inbound := range dataplane.Spec.Networking.GetInbound() {
+		var bestPolicy DataplanePolicy
+		var bestRank mesh_proto.TagSelectorRank
+		sameRankCreatedLater := func(policy DataplanePolicy, rank mesh_proto.TagSelectorRank) bool {
+			return rank.CompareTo(bestRank) == 0 && policy.GetMeta().GetCreationTime().After(bestPolicy.GetMeta().GetCreationTime())
+		}
+		for _, policy := range policies {
+			if 0 == len(policy.Selectors()) { // match everything
+				if bestPolicy == nil || sameRankCreatedLater(policy, mesh_proto.TagSelectorRank{}) {
+					bestPolicy = policy
+				}
+				continue
+			}
+			for _, selector := range policy.Selectors() {
+				if 0 == len(selector.Match) { // match everything
+					if bestPolicy == nil || sameRankCreatedLater(policy, mesh_proto.TagSelectorRank{}) {
+						bestPolicy = policy
+					}
+					continue
+				}
+				tagSelector := mesh_proto.TagSelector(selector.Match)
+				if inbound.MatchTags(tagSelector) {
+					rank := tagSelector.Rank()
+					if rank.CompareTo(bestRank) > 0 || sameRankCreatedLater(policy, rank) {
+						bestRank = rank
+						bestPolicy = policy
+					}
+				}
+			}
+		}
+		if bestPolicy != nil {
+			iface := dataplane.Spec.GetNetworking().ToInboundInterface(inbound)
+			match[iface] = bestPolicy
+		}
+	}
+
+	return match
+}
+
 type DataplanePolicyByName []DataplanePolicy
 
 func (a DataplanePolicyByName) Len() int      { return len(a) }
diff --git a/pkg/core/policy/types.go b/pkg/core/policy/types.go
index 5d227ecca..0146d295e 100644
--- a/pkg/core/policy/types.go
+++ b/pkg/core/policy/types.go
@@ -23,3 +23,5 @@ type DataplanePolicy interface {
 	core_model.Resource
 	Selectors() []*mesh_proto.Selector
 }
+
+type InboundDataplanePolicyMap map[mesh_proto.InboundInterface]DataplanePolicy
diff --git a/pkg/core/runtime/builder.go b/pkg/core/runtime/builder.go
index bf1de0d22..848ed55f2 100644
--- a/pkg/core/runtime/builder.go
+++ b/pkg/core/runtime/builder.go
@@ -6,6 +6,7 @@ import (
 	"os"
 
 	api_server "github.com/kumahq/kuma/pkg/api-server/customization"
+	xds_hooks "github.com/kumahq/kuma/pkg/xds/hooks"
 
 	"github.com/pkg/errors"
 
@@ -42,6 +43,7 @@ type BuilderContext interface {
 	Metrics() metrics.Metrics
 	EventReaderFactory() events.ListenerFactory
 	APIManager() api_server.APIManager
+	XDSHooks() *xds_hooks.Hooks
 }
 
 var _ BuilderContext = &Builder{}
@@ -65,6 +67,7 @@ type Builder struct {
 	metrics  metrics.Metrics
 	erf      events.ListenerFactory
 	apim     api_server.APIManager
+	xdsh     *xds_hooks.Hooks
 	closeCh  <-chan struct{}
 	*runtimeInfo
 }
@@ -176,6 +179,11 @@ func (b *Builder) WithAPIManager(apim api_server.APIManager) *Builder {
 	return b
 }
 
+func (b *Builder) WithXDSHooks(xdsh *xds_hooks.Hooks) *Builder {
+	b.xdsh = xdsh
+	return b
+}
+
 func (b *Builder) Build() (Runtime, error) {
 	if b.cm == nil {
 		return nil, errors.Errorf("ComponentManager has not been configured")
@@ -213,6 +221,9 @@ func (b *Builder) Build() (Runtime, error) {
 	if b.apim == nil {
 		return nil, errors.Errorf("APIManager has not been configured")
 	}
+	if b.xdsh == nil {
+		return nil, errors.Errorf("XDSHooks has not been configured")
+	}
 	return &runtime{
 		RuntimeInfo: b.runtimeInfo,
 		RuntimeContext: &runtimeContext{
@@ -231,6 +242,7 @@ func (b *Builder) Build() (Runtime, error) {
 			metrics:  b.metrics,
 			erf:      b.erf,
 			apim:     b.apim,
+			xdsh:     b.xdsh,
 		},
 		Manager: b.cm,
 	}, nil
@@ -284,10 +296,12 @@ func (b *Builder) Metrics() metrics.Metrics {
 func (b *Builder) EventReaderFactory() events.ListenerFactory {
 	return b.erf
 }
-
 func (b *Builder) APIManager() api_server.APIManager {
 	return b.apim
 }
+func (b *Builder) XDSHooks() *xds_hooks.Hooks {
+	return b.xdsh
+}
 func (b *Builder) CloseCh() <-chan struct{} {
 	return b.closeCh
 }
diff --git a/pkg/core/runtime/runtime.go b/pkg/core/runtime/runtime.go
index 1a1557132..7a88b4170 100644
--- a/pkg/core/runtime/runtime.go
+++ b/pkg/core/runtime/runtime.go
@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	api_server "github.com/kumahq/kuma/pkg/api-server/customization"
+	xds_hooks "github.com/kumahq/kuma/pkg/xds/hooks"
 
 	"github.com/kumahq/kuma/pkg/core/datasource"
 	"github.com/kumahq/kuma/pkg/dns/resolver"
@@ -54,6 +55,7 @@ type RuntimeContext interface {
 	Metrics() metrics.Metrics
 	EventReaderFactory() events.ListenerFactory
 	APIInstaller() api_server.APIInstaller
+	XDSHooks() *xds_hooks.Hooks
 }
 
 var _ Runtime = &runtime{}
@@ -108,6 +110,7 @@ type runtimeContext struct {
 	metrics  metrics.Metrics
 	erf      events.ListenerFactory
 	apim     api_server.APIInstaller
+	xdsh     *xds_hooks.Hooks
 }
 
 func (rc *runtimeContext) Metrics() metrics.Metrics {
@@ -169,6 +172,11 @@ func (rc *runtimeContext) LeaderInfo() component.LeaderInfo {
 func (rc *runtimeContext) LookupIP() lookup.LookupIPFunc {
 	return rc.lif
 }
+
 func (rc *runtimeContext) APIInstaller() api_server.APIInstaller {
 	return rc.apim
 }
+
+func (rc *runtimeContext) XDSHooks() *xds_hooks.Hooks {
+	return rc.xdsh
+}
diff --git a/pkg/xds/hooks/hooks.go b/pkg/xds/hooks/hooks.go
new file mode 100644
index 000000000..1f2377a27
--- /dev/null
+++ b/pkg/xds/hooks/hooks.go
@@ -0,0 +1,13 @@
+package hooks
+
+type Hooks struct {
+	resourceSetHooks []ResourceSetHook
+}
+
+func (h *Hooks) AddResourceSetHook(hook ResourceSetHook) {
+	h.resourceSetHooks = append(h.resourceSetHooks, hook)
+}
+
+func (h *Hooks) ResourceSetHooks() []ResourceSetHook {
+	return h.resourceSetHooks
+}
diff --git a/pkg/xds/hooks/resource_set.go b/pkg/xds/hooks/resource_set.go
new file mode 100644
index 000000000..d611959ad
--- /dev/null
+++ b/pkg/xds/hooks/resource_set.go
@@ -0,0 +1,14 @@
+package hooks
+
+import (
+	core_xds "github.com/kumahq/kuma/pkg/core/xds"
+	xds_context "github.com/kumahq/kuma/pkg/xds/context"
+)
+
+// ResourceSetHook is a way to modify XDS resources generated by Kuma
+// Since resourceSet is an argument, you can add new, remove or modify the existing resources
+// To support V2 and V3, search the resource set for typeUrls for proper version
+// If you want to for example modify only inbound listeners, search for the resource origin of OriginInbound
+type ResourceSetHook interface {
+	Modify(resourceSet *core_xds.ResourceSet, ctx xds_context.Context, proxy *core_xds.Proxy) error
+}
diff --git a/pkg/xds/server/components.go b/pkg/xds/server/components.go
index 200367f62..7e47476ab 100644
--- a/pkg/xds/server/components.go
+++ b/pkg/xds/server/components.go
@@ -20,12 +20,13 @@ import (
 )
 
 var (
-	meshResources = meshResourceTypes(map[core_model.ResourceType]bool{
+	// HashMeshExcludedResources defines Mesh-scoped resources that are not used in XDS therefore when counting hash mesh we can skip them
+	HashMeshExcludedResources = map[core_model.ResourceType]bool{
 		core_mesh.DataplaneInsightType:  true,
 		core_mesh.DataplaneOverviewType: true,
 		core_mesh.ServiceInsightType:    true,
 		core_system.ConfigType:          true,
-	})
+	}
 )
 
 func meshResourceTypes(exclude map[core_model.ResourceType]bool) []core_model.ResourceType {
@@ -53,7 +54,7 @@ func RegisterXDS(rt core_runtime.Runtime, server *grpc.Server) error {
 	if err != nil {
 		return err
 	}
-	meshSnapshotCache, err := mesh.NewCache(rt.ReadOnlyResourceManager(), rt.Config().Store.Cache.ExpirationTime, meshResources, rt.LookupIP(), rt.Metrics())
+	meshSnapshotCache, err := mesh.NewCache(rt.ReadOnlyResourceManager(), rt.Config().Store.Cache.ExpirationTime, meshResourceTypes(HashMeshExcludedResources), rt.LookupIP(), rt.Metrics())
 	if err != nil {
 		return err
 	}
diff --git a/pkg/xds/server/v2/components.go b/pkg/xds/server/v2/components.go
index b9c75326c..b66345042 100644
--- a/pkg/xds/server/v2/components.go
+++ b/pkg/xds/server/v2/components.go
@@ -73,6 +73,7 @@ func RegisterXDS(
 func DefaultReconciler(rt core_runtime.Runtime, xdsContext v2.XdsContext) xds_sync.SnapshotReconciler {
 	return &reconciler{
 		&templateSnapshotGenerator{
+			ResourceSetHooks: rt.XDSHooks().ResourceSetHooks(),
 			ProxyTemplateResolver: &xds_template.SimpleProxyTemplateResolver{
 				ReadOnlyResourceManager: rt.ReadOnlyResourceManager(),
 				DefaultProxyTemplate:    xds_template.DefaultProxyTemplate,
@@ -85,6 +86,7 @@ func DefaultReconciler(rt core_runtime.Runtime, xdsContext v2.XdsContext) xds_sy
 func DefaultIngressReconciler(rt core_runtime.Runtime, xdsContext v2.XdsContext) xds_sync.SnapshotReconciler {
 	return &reconciler{
 		generator: &templateSnapshotGenerator{
+			ResourceSetHooks: rt.XDSHooks().ResourceSetHooks(),
 			ProxyTemplateResolver: &xds_template.StaticProxyTemplateResolver{
 				Template: xds_template.IngressProxyTemplate,
 			},
diff --git a/pkg/xds/server/v2/reconcile.go b/pkg/xds/server/v2/reconcile.go
index 829155195..fe495dd58 100644
--- a/pkg/xds/server/v2/reconcile.go
+++ b/pkg/xds/server/v2/reconcile.go
@@ -2,11 +2,13 @@ package v2
 
 import (
 	"github.com/golang/protobuf/proto"
+	"github.com/pkg/errors"
 
 	"github.com/kumahq/kuma/pkg/core"
 	model "github.com/kumahq/kuma/pkg/core/xds"
 	xds_context "github.com/kumahq/kuma/pkg/xds/context"
 	"github.com/kumahq/kuma/pkg/xds/generator"
+	xds_hooks "github.com/kumahq/kuma/pkg/xds/hooks"
 	xds_sync "github.com/kumahq/kuma/pkg/xds/sync"
 	xds_template "github.com/kumahq/kuma/pkg/xds/template"
 
@@ -93,6 +95,7 @@ type snapshotGenerator interface {
 
 type templateSnapshotGenerator struct {
 	ProxyTemplateResolver xds_template.ProxyTemplateResolver
+	ResourceSetHooks      []xds_hooks.ResourceSetHook
 }
 
 func (s *templateSnapshotGenerator) GenerateSnapshot(ctx xds_context.Context, proxy *model.Proxy) (envoy_cache.Snapshot, error) {
@@ -105,6 +108,11 @@ func (s *templateSnapshotGenerator) GenerateSnapshot(ctx xds_context.Context, pr
 		reconcileLog.Error(err, "failed to generate a snapshot", "proxy", proxy, "template", template)
 		return envoy_cache.Snapshot{}, err
 	}
+	for _, hook := range s.ResourceSetHooks {
+		if err := hook.Modify(rs, ctx, proxy); err != nil {
+			return envoy_cache.Snapshot{}, errors.Wrapf(err, "could not apply hook %T", hook)
+		}
+	}
 
 	version := "" // empty value is a sign to other components to generate the version automatically
 	out := envoy_cache.Snapshot{
diff --git a/pkg/xds/server/v3/components.go b/pkg/xds/server/v3/components.go
index 857f3d9f6..46e5e4df9 100644
--- a/pkg/xds/server/v3/components.go
+++ b/pkg/xds/server/v3/components.go
@@ -73,6 +73,7 @@ func RegisterXDS(
 func DefaultReconciler(rt core_runtime.Runtime, xdsContext v3.XdsContext) xds_sync.SnapshotReconciler {
 	return &reconciler{
 		&templateSnapshotGenerator{
+			ResourceSetHooks: rt.XDSHooks().ResourceSetHooks(),
 			ProxyTemplateResolver: &xds_template.SimpleProxyTemplateResolver{
 				ReadOnlyResourceManager: rt.ReadOnlyResourceManager(),
 				DefaultProxyTemplate:    xds_template.DefaultProxyTemplate,
@@ -85,6 +86,7 @@ func DefaultReconciler(rt core_runtime.Runtime, xdsContext v3.XdsContext) xds_sy
 func DefaultIngressReconciler(rt core_runtime.Runtime, xdsContext v3.XdsContext) xds_sync.SnapshotReconciler {
 	return &reconciler{
 		generator: &templateSnapshotGenerator{
+			ResourceSetHooks: rt.XDSHooks().ResourceSetHooks(),
 			ProxyTemplateResolver: &xds_template.StaticProxyTemplateResolver{
 				Template: xds_template.IngressProxyTemplate,
 			},
diff --git a/pkg/xds/server/v3/reconcile.go b/pkg/xds/server/v3/reconcile.go
index 47cb5c33b..5c7ad7254 100644
--- a/pkg/xds/server/v3/reconcile.go
+++ b/pkg/xds/server/v3/reconcile.go
@@ -2,11 +2,13 @@ package v3
 
 import (
 	"github.com/golang/protobuf/proto"
+	"github.com/pkg/errors"
 
 	"github.com/kumahq/kuma/pkg/core"
 	model "github.com/kumahq/kuma/pkg/core/xds"
 	xds_context "github.com/kumahq/kuma/pkg/xds/context"
 	"github.com/kumahq/kuma/pkg/xds/generator"
+	xds_hooks "github.com/kumahq/kuma/pkg/xds/hooks"
 	xds_sync "github.com/kumahq/kuma/pkg/xds/sync"
 	xds_template "github.com/kumahq/kuma/pkg/xds/template"
 
@@ -93,6 +95,7 @@ type snapshotGenerator interface {
 
 type templateSnapshotGenerator struct {
 	ProxyTemplateResolver xds_template.ProxyTemplateResolver
+	ResourceSetHooks      []xds_hooks.ResourceSetHook
 }
 
 func (s *templateSnapshotGenerator) GenerateSnapshot(ctx xds_context.Context, proxy *model.Proxy) (envoy_cache.Snapshot, error) {
@@ -105,6 +108,11 @@ func (s *templateSnapshotGenerator) GenerateSnapshot(ctx xds_context.Context, pr
 		reconcileLog.Error(err, "failed to generate a snapshot", "proxy", proxy, "template", template)
 		return envoy_cache.Snapshot{}, err
 	}
+	for _, hook := range s.ResourceSetHooks {
+		if err := hook.Modify(rs, ctx, proxy); err != nil {
+			return envoy_cache.Snapshot{}, errors.Wrapf(err, "could not apply hook %T", hook)
+		}
+	}
 
 	version := "" // empty value is a sign to other components to generate the version automatically
 	out := envoy_cache.Snapshot{


Test patch:
diff --git a/app/kuma-dp/cmd/root_test.go b/app/kuma-dp/cmd/root_test.go
index 2e9839fb4..6de0e9bf1 100644
--- a/app/kuma-dp/cmd/root_test.go
+++ b/app/kuma-dp/cmd/root_test.go
@@ -8,7 +8,7 @@ import (
 var _ = Describe("root", func() {
 	It("should be possible to run `kuma-dp` without a sub-command", func() {
 		// given
-		cmd := newRootCmd()
+		cmd := NewRootCmd(DefaultRootContext())
 		cmd.SetArgs([]string{})
 		// when
 		err := cmd.Execute()
diff --git a/app/kuma-dp/cmd/run_test.go b/app/kuma-dp/cmd/run_test.go
index c5dcbd687..90afe22f2 100644
--- a/app/kuma-dp/cmd/run_test.go
+++ b/app/kuma-dp/cmd/run_test.go
@@ -29,20 +29,12 @@ import (
 var _ = Describe("run", func() {
 
 	var backupSetupSignalHandler func() <-chan struct{}
-	var backupBootstrapGenerator envoy.BootstrapConfigFactoryFunc
 
 	BeforeEach(func() {
 		backupSetupSignalHandler = core.SetupSignalHandler
-		backupBootstrapGenerator = bootstrapGenerator
-		bootstrapGenerator = func(_ string, cfg kumadp.Config, _ *rest.Resource, _ types.BootstrapVersion, version envoy.EnvoyVersion) ([]byte, types.BootstrapVersion, error) {
-			respBytes, err := ioutil.ReadFile(filepath.Join("testdata", "bootstrap-config.golden.yaml"))
-			Expect(err).ToNot(HaveOccurred())
-			return respBytes, "", nil
-		}
 	})
 	AfterEach(func() {
 		core.SetupSignalHandler = backupSetupSignalHandler
-		bootstrapGenerator = backupBootstrapGenerator
 	})
 
 	var stopCh chan struct{}
@@ -113,7 +105,13 @@ var _ = Describe("run", func() {
 			}
 
 			// given
-			cmd := newRootCmd()
+			rootCtx := DefaultRootContext()
+			rootCtx.BootstrapGenerator = func(_ string, cfg kumadp.Config, _ *rest.Resource, _ types.BootstrapVersion, version envoy.EnvoyVersion) ([]byte, types.BootstrapVersion, error) {
+				respBytes, err := ioutil.ReadFile(filepath.Join("testdata", "bootstrap-config.golden.yaml"))
+				Expect(err).ToNot(HaveOccurred())
+				return respBytes, "", nil
+			}
+			cmd := NewRootCmd(rootCtx)
 			cmd.SetArgs(append([]string{"run"}, given.args...))
 			cmd.SetOut(&bytes.Buffer{})
 			cmd.SetErr(&bytes.Buffer{})
@@ -305,7 +303,7 @@ var _ = Describe("run", func() {
 		defer l.Close()
 
 		// given
-		cmd := newRootCmd()
+		cmd := NewRootCmd(DefaultRootContext())
 		cmd.SetArgs([]string{
 			"run",
 			"--cp-address", "http://localhost:1234",
diff --git a/pkg/test/runtime/runtime.go b/pkg/test/runtime/runtime.go
index bd9ec05b8..3e7973396 100644
--- a/pkg/test/runtime/runtime.go
+++ b/pkg/test/runtime/runtime.go
@@ -4,6 +4,7 @@ import (
 	"net"
 
 	"github.com/kumahq/kuma/pkg/api-server/customization"
+	xds_hooks "github.com/kumahq/kuma/pkg/xds/hooks"
 
 	kuma_cp "github.com/kumahq/kuma/pkg/config/app/kuma-cp"
 	config_manager "github.com/kumahq/kuma/pkg/core/config/manager"
@@ -71,6 +72,7 @@ func BuilderFor(cfg kuma_cp.Config) (*core_runtime.Builder, error) {
 	builder.WithLookupIP(net.LookupIP)
 	builder.WithEventReaderFactory(events.NewEventBus())
 	builder.WithAPIManager(customization.NewAPIList())
+	builder.WithXDSHooks(&xds_hooks.Hooks{})
 
 	_ = initializeConfigManager(cfg, builder)
 	_ = initializeDNSResolver(cfg, builder)