added support for interceptors

.github/workflows/e2e.yaml

@@ -16,7 +16,7 @@ concurrency:
   cancel-in-progress: true
 
 env:
-  VCLUSTER_VERSION: v0.20.0-alpha.2
+  VCLUSTER_VERSION: v0.20.0-alpha.4
   VCLUSTER_SUFFIX: vcluster
   VCLUSTER_NAME: vcluster
   VCLUSTER_NAMESPACE: vcluster

e2e/plugin/plugin.go

@@ -192,4 +192,14 @@ var _ = ginkgo.Describe("Plugin test", func() {
 			WithTimeout(pollingDurationLong).
 			Should(gomega.BeTrue())
 	})
+
+	ginkgo.It("check the interceptor", func() {
+		// wait for secret to become synced
+		vPod := &corev1.Pod{}
+		err := f.VclusterCRClient.Get(f.Context, types.NamespacedName{Name: "stuff", Namespace: "test"}, vPod)
+		framework.ExpectNoError(err)
+
+		// check if secret is synced correctly
+		framework.ExpectEqual(vPod.Name, "definitelynotstuff")
+	})
 })

e2e/test_plugin/main.go

@@ -31,6 +31,9 @@ func main() {
 	plugin.MustRegister(syncers.NewMyDeploymentSyncer(ctx))
 	plugin.MustRegister(syncers.NewCarSyncer(ctx))
 	plugin.MustRegister(syncers.NewImportSecrets(ctx))
+	plugin.MustRegister(syncers.DummyInterceptor{})
+
+	klog.Info("finished registering the plugins")
 	plugin.MustStart()
 }
 

e2e/test_plugin/syncers/interceptor.go

@@ -0,0 +1,55 @@
+package syncers
+
+import (
+	"net/http"
+
+	"github.com/loft-sh/vcluster-sdk/plugin"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/apiserver/pkg/endpoints/handlers/negotiation"
+	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+
+	v2 "github.com/loft-sh/vcluster/pkg/plugin/v2"
+	corev1 "k8s.io/api/core/v1"
+)
+
+var _ plugin.Interceptor = DummyInterceptor{}
+
+type DummyInterceptor struct {
+}
+
+func (d DummyInterceptor) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	scheme := runtime.NewScheme()
+	clientgoscheme.AddToScheme(scheme)
+
+	s := serializer.NewCodecFactory(scheme)
+	responsewriters.WriteObjectNegotiated(
+		s,
+		negotiation.DefaultEndpointRestrictions,
+		schema.GroupVersion{
+			Group:   "",
+			Version: "v1"},
+		w,
+		r,
+		200,
+		&corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "definitelynotstuff"}},
+		false)
+}
+
+func (d DummyInterceptor) Name() string {
+	return "testinterceptor"
+}
+
+func (d DummyInterceptor) InterceptionRules() []v2.InterceptorRule {
+	return []v2.InterceptorRule{
+		{
+			APIGroups:     []string{"*"},
+			Resources:     []string{"pods"},
+			ResourceNames: []string{"*"},
+			Verbs:         []string{"get"},
+		},
+	}
+}

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/ghodss/yaml v1.0.0
 	github.com/hashicorp/go-plugin v1.6.0
 	github.com/loft-sh/log v0.0.0-20230824104949-bd516c25712a
-	github.com/loft-sh/vcluster v0.20.0-alpha.2.0.20240403130844-8bb987ed97b4
+	github.com/loft-sh/vcluster v0.20.0-alpha.3.0.20240409111424-27cde82f6544
 	github.com/onsi/ginkgo/v2 v2.14.0
 	github.com/onsi/gomega v1.30.0
 	github.com/pkg/errors v0.9.1

go.sum

@@ -630,6 +630,10 @@ github.com/loft-sh/utils v0.0.29 h1:P/MObccXToAZy2QoJSQDJ+OJx1qHitpFHEVj3QBSNJs=
 github.com/loft-sh/utils v0.0.29/go.mod h1:9hlX9cGpWHg3mNi/oBlv3X4ePGDMK66k8MbOZGFMDTI=
 github.com/loft-sh/vcluster v0.20.0-alpha.2.0.20240403130844-8bb987ed97b4 h1:D486kUE5ZQSx2DDQQ5kf/gowKZG/dNw7As4vg9WR7dw=
 github.com/loft-sh/vcluster v0.20.0-alpha.2.0.20240403130844-8bb987ed97b4/go.mod h1:Ty7km/e/U7wVF9kIqROOR51/XezgHEtS/za0QVPhKkU=
+github.com/loft-sh/vcluster v0.20.0-alpha.3.0.20240409074019-5b5358d2f54d h1:nTQLvvJS4b9tEBMk4s2yQkxnW/9MW5rZl3eCDqeKarE=
+github.com/loft-sh/vcluster v0.20.0-alpha.3.0.20240409074019-5b5358d2f54d/go.mod h1:Ty7km/e/U7wVF9kIqROOR51/XezgHEtS/za0QVPhKkU=
+github.com/loft-sh/vcluster v0.20.0-alpha.3.0.20240409111424-27cde82f6544 h1:/UxPl3HXlCwT8whH57NanvcInguAWgubal2QxvSGhoE=
+github.com/loft-sh/vcluster v0.20.0-alpha.3.0.20240409111424-27cde82f6544/go.mod h1:Ty7km/e/U7wVF9kIqROOR51/XezgHEtS/za0QVPhKkU=
 github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/magiconair/properties v1.8.5/go.mod h1:y3VJvCyxH9uVvJTWEGAELF3aiYNyPKd5NZ3oSwXrF60=
 github.com/mailru/easyjson v0.0.0-20160728113105-d5b7844b561a/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=

plugin/manager.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net/http"
 	"os"
+	"strconv"
 	"sync"
 
 	"github.com/ghodss/yaml"
@@ -22,6 +24,7 @@ import (
 	"github.com/loft-sh/vcluster/pkg/util/clienthelper"
 	contextutil "github.com/loft-sh/vcluster/pkg/util/context"
 	"github.com/pkg/errors"
+	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
@@ -30,7 +33,9 @@ import (
 )
 
 func newManager() Manager {
-	return &manager{}
+	return &manager{
+		interceptorsHandlers: make(map[string]http.Handler),
+	}
 }
 
 type manager struct {
@@ -47,6 +52,10 @@ type manager struct {
 
 	syncers []syncertypes.Base
 
+	interceptorsHandlers map[string]http.Handler
+	interceptors         []Interceptor
+	interceptorsPort     int
+
 	proConfig v2.InitConfigPro
 
 	options Options
@@ -106,6 +115,7 @@ func (m *manager) InitWithOptions(options Options) (*synccontext.RegisterContext
 	if err != nil {
 		return nil, fmt.Errorf("error decoding init config %s: %w", initRequest.Config, err)
 	}
+	m.interceptorsPort = initConfig.Port
 
 	// try to change working dir
 	if initConfig.WorkingDir != "" {
@@ -177,7 +187,15 @@ func (m *manager) Register(syncer syncertypes.Base) error {
 	m.m.Lock()
 	defer m.m.Unlock()
 
-	m.syncers = append(m.syncers, syncer)
+	if int, ok := syncer.(Interceptor); ok {
+		if _, ok := m.interceptorsHandlers[int.Name()]; ok {
+			return fmt.Errorf("could not add the interceptor %s because its name is already in use", int.Name())
+		}
+		m.interceptorsHandlers[int.Name()] = int
+		m.interceptors = append(m.interceptors, int)
+	} else {
+		m.syncers = append(m.syncers, syncer)
+	}
 	return nil
 }
 
@@ -191,6 +209,24 @@ func (m *manager) Start() error {
 	return nil
 }
 
+func (m *manager) startInterceptors() error {
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		handlerName := r.Header.Get("VCluster-Plugin-Handler-Name")
+		if handlerName == "" {
+			responsewriters.InternalError(w, r, errors.New("header VCluster-Plugin-Handler-Name wasn't set"))
+			return
+		}
+		interceptorHandler, ok := m.interceptorsHandlers[handlerName]
+		if !ok {
+			responsewriters.InternalError(w, r, errors.New("header VCluster-Plugin-Handler-Name had no match"))
+			return
+		}
+		interceptorHandler.ServeHTTP(w, r)
+	})
+
+	return http.ListenAndServe("127.0.0.1:"+strconv.Itoa(m.interceptorsPort), handler)
+}
+
 func (m *manager) start() error {
 	m.m.Lock()
 	defer m.m.Unlock()
@@ -206,9 +242,23 @@ func (m *manager) start() error {
 		return fmt.Errorf("find all hooks: %w", err)
 	}
 
+	// find the interceptors
+	interceptors := m.findAllInterceptors()
+
 	// signal we are ready
-	m.pluginServer.SetReady(hooks)
+	m.pluginServer.SetReady(hooks, interceptors, m.interceptorsPort)
 
+	if len(m.interceptors) > 0 {
+		go func() {
+			// we need to start them regardless of being the leader, since the traffic is
+			// directed to all replicas
+			err := m.startInterceptors()
+			if err != nil {
+				klog.Error(err, "error while running the http interceptors:")
+				os.Exit(1)
+			}
+		}()
+	}
 	// wait until we are leader to continue
 	<-m.pluginServer.IsLeader()
 
@@ -305,6 +355,11 @@ func (m *manager) start() error {
 	return nil
 }
 
+func (m *manager) findAllInterceptors() []Interceptor {
+	klog.Info("len of m.interceptor is : ", len(m.interceptors))
+	return m.interceptors
+}
+
 func (m *manager) findAllHooks() (map[types.VersionKindType][]ClientHook, error) {
 	// gather all hooks
 	hooks := map[types.VersionKindType][]ClientHook{}

plugin/server.go

@@ -21,7 +21,7 @@ type server interface {
 	Serve()
 
 	// SetReady signals the plugin server the plugin is ready to start
-	SetReady(hooks map[types.VersionKindType][]ClientHook)
+	SetReady(hooks map[types.VersionKindType][]ClientHook, interceptors []Interceptor, port int)
 
 	// Initialized retrieves the initialize request
 	Initialized() <-chan *pluginv2.Initialize_Request
@@ -43,7 +43,9 @@ func newPluginServer() (server, error) {
 type pluginServer struct {
 	pluginv2.UnimplementedPluginServer
 
-	hooks map[types.VersionKindType][]ClientHook
+	hooks            map[types.VersionKindType][]ClientHook
+	interceptors     []Interceptor
+	interceptorsPort int
 
 	initialized chan *pluginv2.Initialize_Request
 	isReady     chan struct{}
@@ -88,8 +90,10 @@ func (p *pluginServer) IsLeader() <-chan struct{} {
 	return p.isLeader
 }
 
-func (p *pluginServer) SetReady(hooks map[types.VersionKindType][]ClientHook) {
+func (p *pluginServer) SetReady(hooks map[types.VersionKindType][]ClientHook, interceptors []Interceptor, port int) {
 	p.hooks = hooks
+	p.interceptors = interceptors
+	p.interceptorsPort = port
 	close(p.isReady)
 }
 
@@ -216,9 +220,11 @@ func (p *pluginServer) GetPluginConfig(context.Context, *pluginv2.GetPluginConfi
 		return nil, err
 	}
 
+	interceptorConfig := p.getInterceptorConfig()
 	// build plugin config
 	pluginConfig := &v2.PluginConfig{
-		ClientHooks: clientHooks,
+		ClientHooks:  clientHooks,
+		Interceptors: interceptorConfig,
 	}
 
 	// marshal plugin config
@@ -265,6 +271,15 @@ func (p *pluginServer) getClientHooks() ([]*v2.ClientHook, error) {
 	return registeredHooks, nil
 }
 
+func (p *pluginServer) getInterceptorConfig() map[string][]v2.InterceptorRule {
+	res := make(map[string][]v2.InterceptorRule)
+	for _, interceptor := range p.interceptors {
+		res[interceptor.Name()] = interceptor.InterceptionRules()
+	}
+
+	return res
+}
+
 var _ plugin.Plugin = &pluginServer{}
 
 // Server always returns an error; we're only implementing the GRPCPlugin

plugin/types.go

@@ -2,6 +2,7 @@ package plugin
 
 import (
 	"context"
+	"net/http"
 
 	synccontext "github.com/loft-sh/vcluster/pkg/controllers/syncer/context"
 	v2 "github.com/loft-sh/vcluster/pkg/plugin/v2"
@@ -58,6 +59,16 @@ type ClientHook interface {
 	Resource() client.Object
 }
 
+type Interceptor interface {
+	syncertypes.Base
+
+	// Handler is the handler that will handle the requests delegated by the syncer
+	http.Handler
+
+	// InterceptionRules returns an rbac style struct which defines what to intercept
+	InterceptionRules() []v2.InterceptorRule
+}
+
 type MutateCreateVirtual interface {
 	MutateCreateVirtual(ctx context.Context, obj client.Object) (client.Object, error)
 }