Implement multiplexer proxy.

cmd/yurthub/app/config/config.go

@@ -29,6 +29,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	apiserver "k8s.io/apiserver/pkg/server"
 	"k8s.io/apiserver/pkg/server/dynamiccertificates"
@@ -54,11 +55,26 @@ import (
 	"github.com/openyurtio/openyurt/pkg/yurthub/filter/manager"
 	"github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/meta"
 	"github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/serializer"
+	"github.com/openyurtio/openyurt/pkg/yurthub/multiplexer"
+	"github.com/openyurtio/openyurt/pkg/yurthub/multiplexer/storage"
 	"github.com/openyurtio/openyurt/pkg/yurthub/network"
 	"github.com/openyurtio/openyurt/pkg/yurthub/storage/disk"
 	"github.com/openyurtio/openyurt/pkg/yurthub/util"
 )
 
+var DefaultMultiplexerResources = []schema.GroupVersionResource{
+	{
+		Group:    "",
+		Version:  "v1",
+		Resource: "services",
+	},
+	{
+		Group:    "discovery.k8s.io",
+		Version:  "v1",
+		Resource: "endpointslices",
+	},
+}
+
 // YurtHubConfiguration represents configuration of yurthub
 type YurtHubConfiguration struct {
 	LBMode                          string
@@ -101,6 +117,9 @@ type YurtHubConfiguration struct {
 	CoordinatorClient               kubernetes.Interface
 	LeaderElection                  componentbaseconfig.LeaderElectionConfiguration
 	HostControlPlaneAddr            string // ip:port
+	PostStartHooks                  map[string]func() error
+	MultiplexerCacheManager         multiplexer.MultiplexerManager
+	MultiplexerResources            []schema.GroupVersionResource
 }
 
 // Complete converts *options.YurtHubOptions to *YurtHubConfiguration
@@ -176,6 +195,8 @@ func Complete(options *options.YurtHubOptions) (*YurtHubConfiguration, error) {
 		CoordinatorStorageAddr:    options.CoordinatorStorageAddr,
 		LeaderElection:            options.LeaderElection,
 		HostControlPlaneAddr:      options.HostControlPlaneAddr,
+		MultiplexerResources:      DefaultMultiplexerResources,
+		MultiplexerCacheManager:   newMultiplexerCacheManager(options),
 	}
 
 	// if yurthub is in local mode, certMgr and networkMgr are no need to start
@@ -403,3 +424,17 @@ func prepareServerServing(options *options.YurtHubOptions, certMgr certificate.Y
 
 	return nil
 }
+
+func newMultiplexerCacheManager(options *options.YurtHubOptions) multiplexer.MultiplexerManager {
+	config := newRestConfig(options.YurtHubProxyHost, options.YurtHubProxyPort)
+	rsm := storage.NewStorageManager(config)
+
+	return multiplexer.NewRequestsMultiplexerManager(rsm)
+}
+
+func newRestConfig(host string, port int) *rest.Config {
+	return &rest.Config{
+		Host:      fmt.Sprintf("http://%s:%d", host, port),
+		UserAgent: util.MultiplexerProxyClientUserAgent,
+	}
+}

pkg/yurthub/filter/interfaces.go

@@ -19,10 +19,13 @@ package filter
 import (
 	"io"
 	"net/http"
+	"strings"
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
+
+	yurtutil "github.com/openyurtio/openyurt/pkg/util"
 )
 
 type NodesInPoolGetter func(poolName string) ([]string, error)
@@ -59,4 +62,35 @@ type ObjectFilter interface {
 	Filter(obj runtime.Object, stopCh <-chan struct{}) runtime.Object
 }
 
+type FilterManager interface {
+	FindResponseFilter(req *http.Request) (ResponseFilter, bool)
+	FindObjectFilters(req *http.Request) ObjectFilter
+}
+
 type NodeGetter func(name string) (*v1.Node, error)
+
+type UnionObjectFilter []ObjectFilter
+
+func (chain UnionObjectFilter) Name() string {
+	var names []string
+	for i := range chain {
+		names = append(names, chain[i].Name())
+	}
+	return strings.Join(names, ",")
+}
+
+func (chain UnionObjectFilter) SupportedResourceAndVerbs() map[string]sets.Set[string] {
+	// do nothing
+	return map[string]sets.Set[string]{}
+}
+
+func (chain UnionObjectFilter) Filter(obj runtime.Object, stopCh <-chan struct{}) runtime.Object {
+	for i := range chain {
+		obj = chain[i].Filter(obj, stopCh)
+		if yurtutil.IsNil(obj) {
+			break
+		}
+	}
+
+	return obj
+}

pkg/yurthub/filter/manager/manager.go

@@ -111,3 +111,20 @@ func (m *Manager) FindResponseFilter(req *http.Request) (filter.ResponseFilter,
 
 	return nil, false
 }
+
+func (m *Manager) FindObjectFilters(req *http.Request) filter.ObjectFilter {
+	objectFilters := make([]filter.ObjectFilter, 0)
+	approved, filterNames := m.Approver.Approve(req)
+	if !approved {
+		return nil
+	}
+
+	for i := range filterNames {
+		if objectFilter, ok := m.nameToObjectFilter[filterNames[i]]; ok {
+			objectFilters = append(objectFilters, objectFilter)
+		}
+	}
+
+	filters := filter.UnionObjectFilter(objectFilters)
+	return filters
+}

pkg/yurthub/filter/servicetopology/filter.go

@@ -20,7 +20,7 @@ import (
 	"context"
 
 	v1 "k8s.io/api/core/v1"
-	discovery "k8s.io/api/discovery/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	discoveryV1beta1 "k8s.io/api/discovery/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -130,7 +130,7 @@ func (stf *serviceTopologyFilter) Filter(obj runtime.Object, stopCh <-chan struc
 	}
 
 	switch v := obj.(type) {
-	case *v1.Endpoints, *discoveryV1beta1.EndpointSlice, *discovery.EndpointSlice:
+	case *v1.Endpoints, *discoveryV1beta1.EndpointSlice, *discoveryv1.EndpointSlice:
 		return stf.serviceTopologyHandler(v)
 	default:
 		return obj
@@ -164,9 +164,9 @@ func (stf *serviceTopologyFilter) resolveServiceTopologyType(obj runtime.Object)
 	case *discoveryV1beta1.EndpointSlice:
 		svcNamespace = v.Namespace
 		svcName = v.Labels[discoveryV1beta1.LabelServiceName]
-	case *discovery.EndpointSlice:
+	case *discoveryv1.EndpointSlice:
 		svcNamespace = v.Namespace
-		svcName = v.Labels[discovery.LabelServiceName]
+		svcName = v.Labels[discoveryv1.LabelServiceName]
 	case *v1.Endpoints:
 		svcNamespace = v.Namespace
 		svcName = v.Name
@@ -190,7 +190,7 @@ func (stf *serviceTopologyFilter) nodeTopologyHandler(obj runtime.Object) runtim
 	switch v := obj.(type) {
 	case *discoveryV1beta1.EndpointSlice:
 		return reassembleV1beta1EndpointSlice(v, stf.nodeName, nil)
-	case *discovery.EndpointSlice:
+	case *discoveryv1.EndpointSlice:
 		return reassembleEndpointSlice(v, stf.nodeName, nil)
 	case *v1.Endpoints:
 		return reassembleEndpoints(v, stf.nodeName, nil)
@@ -215,7 +215,7 @@ func (stf *serviceTopologyFilter) nodePoolTopologyHandler(obj runtime.Object) ru
 	switch v := obj.(type) {
 	case *discoveryV1beta1.EndpointSlice:
 		return reassembleV1beta1EndpointSlice(v, "", nodes)
-	case *discovery.EndpointSlice:
+	case *discoveryv1.EndpointSlice:
 		return reassembleEndpointSlice(v, "", nodes)
 	case *v1.Endpoints:
 		return reassembleEndpoints(v, "", nodes)
@@ -252,13 +252,13 @@ func reassembleV1beta1EndpointSlice(endpointSlice *discoveryV1beta1.EndpointSlic
 }
 
 // reassembleEndpointSlice will discard endpoints that are not on the same node/nodePool for v1.EndpointSlice
-func reassembleEndpointSlice(endpointSlice *discovery.EndpointSlice, nodeName string, nodes []string) *discovery.EndpointSlice {
+func reassembleEndpointSlice(endpointSlice *discoveryv1.EndpointSlice, nodeName string, nodes []string) *discoveryv1.EndpointSlice {
 	if len(nodeName) != 0 && len(nodes) != 0 {
 		klog.Warningf("reassembleEndpointSlice: nodeName(%s) and nodePool can not be set at the same time", nodeName)
 		return endpointSlice
 	}
 
-	var newEps []discovery.Endpoint
+	var newEps []discoveryv1.Endpoint
 	for i := range endpointSlice.Endpoints {
 		if len(nodeName) != 0 {
 			if *endpointSlice.Endpoints[i].NodeName == nodeName {

pkg/yurthub/multiplexer/cache.go

@@ -0,0 +1,73 @@
+/*
+Copyright 2024 The OpenYurt Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package multiplexer
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/watch"
+	kstorage "k8s.io/apiserver/pkg/storage"
+	"k8s.io/apiserver/pkg/storage/cacher"
+	"k8s.io/client-go/kubernetes/scheme"
+)
+
+type Interface interface {
+	Watch(ctx context.Context, key string, opts kstorage.ListOptions) (watch.Interface, error)
+	GetList(ctx context.Context, key string, opts kstorage.ListOptions, listObj runtime.Object) error
+}
+
+type ResourceCacheConfig struct {
+	KeyFunc      func(runtime.Object) (string, error)
+	NewFunc      func() runtime.Object
+	NewListFunc  func() runtime.Object
+	GetAttrsFunc kstorage.AttrFunc
+}
+
+func NewResourceCache(
+	s kstorage.Interface,
+	resource *schema.GroupVersionResource,
+	config *ResourceCacheConfig) (Interface, func(), error) {
+
+	cacheConfig := cacher.Config{
+		Storage:       s,
+		Versioner:     kstorage.APIObjectVersioner{},
+		GroupResource: resource.GroupResource(),
+		KeyFunc:       config.KeyFunc,
+		NewFunc:       config.NewFunc,
+		NewListFunc:   config.NewListFunc,
+		GetAttrsFunc:  config.GetAttrsFunc,
+		Codec:         scheme.Codecs.LegacyCodec(resource.GroupVersion()),
+	}
+
+	cacher, err := cacher.NewCacherFromConfig(cacheConfig)
+	if err != nil {
+		return nil, func() {}, fmt.Errorf("failed to new cacher from config, error: %v", err)
+	}
+
+	var once sync.Once
+	destroyFunc := func() {
+		once.Do(func() {
+			cacher.Stop()
+		})
+	}
+
+	return cacher, destroyFunc, nil
+}