feature: add data filtering framework

cmd/yurthub/app/config/config.go

@@ -21,13 +21,23 @@ import (
 	"net"
 	"net/url"
 	"strings"
+	"time"
 
 	"github.com/openyurtio/openyurt/cmd/yurthub/app/options"
 	"github.com/openyurtio/openyurt/pkg/projectinfo"
 	"github.com/openyurtio/openyurt/pkg/yurthub/cachemanager"
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter"
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter/masterservice"
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter/servicetopology"
 	"github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/serializer"
 	"github.com/openyurtio/openyurt/pkg/yurthub/storage/factory"
+	yurtclientset "github.com/openyurtio/yurt-app-manager-api/pkg/yurtappmanager/client/clientset/versioned"
+	yurtinformers "github.com/openyurtio/yurt-app-manager-api/pkg/yurtappmanager/client/informers/externalversions"
 
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog"
 )
 
@@ -55,6 +65,10 @@ type YurtHubConfiguration struct {
 	HubAgentDummyIfName         string
 	StorageWrapper              cachemanager.StorageWrapper
 	SerializerManager           *serializer.SerializerManager
+	MutatedMasterServiceAddr    string
+	Filters                     *filter.Filters
+	SharedFactory               informers.SharedInformerFactory
+	YurtSharedFactory           yurtinformers.SharedInformerFactory
 }
 
 // Complete converts *options.YurtHubOptions to *YurtHubConfiguration
@@ -75,6 +89,28 @@ func Complete(options *options.YurtHubOptions) (*YurtHubConfiguration, error) {
 	hubServerAddr := net.JoinHostPort(options.YurtHubHost, options.YurtHubPort)
 	proxyServerAddr := net.JoinHostPort(options.YurtHubHost, options.YurtHubProxyPort)
 	proxyServerDummyAddr := net.JoinHostPort(options.HubAgentDummyIfIP, options.YurtHubProxyPort)
+
+	sharedFactory, yurtSharedFactory, err := createSharedInformers(fmt.Sprintf("http://%s", proxyServerAddr))
+	if err != nil {
+		return nil, err
+	}
+
+	var filters *filter.Filters
+	var mutatedMasterServiceAddr string
+	if options.EnableResourceFilter {
+		filters = filter.NewFilters(options.DisabledResourceFilters)
+		registerAllFilters(filters)
+
+		mutatedMasterServiceAddr = us[0].Host
+		if options.AccessServerThroughHub {
+			if options.EnableDummyIf {
+				mutatedMasterServiceAddr = proxyServerDummyAddr
+			} else {
+				mutatedMasterServiceAddr = proxyServerAddr
+			}
+		}
+	}
+
 	cfg := &YurtHubConfiguration{
 		LBMode:                      options.LBMode,
 		RemoteServers:               us,
@@ -98,6 +134,10 @@ func Complete(options *options.YurtHubOptions) (*YurtHubConfiguration, error) {
 		HubAgentDummyIfName:         options.HubAgentDummyIfName,
 		StorageWrapper:              storageWrapper,
 		SerializerManager:           serializerManager,
+		MutatedMasterServiceAddr:    mutatedMasterServiceAddr,
+		Filters:                     filters,
+		SharedFactory:               sharedFactory,
+		YurtSharedFactory:           yurtSharedFactory,
 	}
 
 	return cfg, nil
@@ -129,3 +169,32 @@ func parseRemoteServers(serverAddr string) ([]*url.URL, error) {
 
 	return us, nil
 }
+
+// createSharedInformers create sharedInformers from the given proxyAddr.
+func createSharedInformers(proxyAddr string) (informers.SharedInformerFactory, yurtinformers.SharedInformerFactory, error) {
+	var kubeConfig *rest.Config
+	var err error
+	kubeConfig, err = clientcmd.BuildConfigFromFlags(proxyAddr, "")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	client, err := kubernetes.NewForConfig(kubeConfig)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	yurtClient, err := yurtclientset.NewForConfig(kubeConfig)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return informers.NewSharedInformerFactory(client, 24*time.Hour), yurtinformers.NewSharedInformerFactory(yurtClient, 24*time.Hour), nil
+}
+
+// registerAllFilters by order, the front registered filter will be
+// called before the later registered ones.
+func registerAllFilters(filters *filter.Filters) {
+	servicetopology.Register(filters)
+	masterservice.Register(filters)
+}

cmd/yurthub/app/options/options.go

@@ -57,6 +57,9 @@ type YurtHubOptions struct {
 	HubAgentDummyIfIP         string
 	HubAgentDummyIfName       string
 	DiskCachePath             string
+	AccessServerThroughHub    bool
+	EnableResourceFilter      bool
+	DisabledResourceFilters   []string
 }
 
 // NewYurtHubOptions creates a new YurtHubOptions with a default config.
@@ -81,8 +84,10 @@ func NewYurtHubOptions() *YurtHubOptions {
 		HubAgentDummyIfIP:         "169.254.2.1",
 		HubAgentDummyIfName:       fmt.Sprintf("%s-dummy0", projectinfo.GetHubName()),
 		DiskCachePath:             disk.CacheBaseDir,
+		AccessServerThroughHub:    false,
+		EnableResourceFilter:      true,
+		DisabledResourceFilters:   make([]string, 0),
 	}
-
 	return o
 }
 
@@ -136,6 +141,9 @@ func (o *YurtHubOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&o.HubAgentDummyIfIP, "dummy-if-ip", o.HubAgentDummyIfIP, "the ip address of dummy interface that used for container connect hub agent(exclusive ips: 169.254.31.0/24, 169.254.1.1/32)")
 	fs.StringVar(&o.HubAgentDummyIfName, "dummy-if-name", o.HubAgentDummyIfName, "the name of dummy interface that is used for hub agent")
 	fs.StringVar(&o.DiskCachePath, "disk-cache-path", o.DiskCachePath, "the path for kubernetes to storage metadata")
+	fs.BoolVar(&o.AccessServerThroughHub, "access-server-through-hub", o.AccessServerThroughHub, "enable pods access kube-apiserver through yurthub or not")
+	fs.BoolVar(&o.EnableResourceFilter, "enable-resource-filter", o.EnableResourceFilter, "enable to filter response that comes back from reverse proxy")
+	fs.StringSliceVar(&o.DisabledResourceFilters, "disabled-resource-filters", o.DisabledResourceFilters, "disable resource filters to handle response")
 }
 
 // verifyDummyIP verify the specified ip is valid or not

cmd/yurthub/app/start.go

@@ -26,6 +26,9 @@ import (
 	"github.com/openyurtio/openyurt/pkg/yurthub/certificate"
 	"github.com/openyurtio/openyurt/pkg/yurthub/certificate/hubself"
 	"github.com/openyurtio/openyurt/pkg/yurthub/certificate/kubelet"
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter"
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter/initializer"
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter/servicetopology"
 	"github.com/openyurtio/openyurt/pkg/yurthub/gc"
 	"github.com/openyurtio/openyurt/pkg/yurthub/healthchecker"
 	"github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/rest"
@@ -136,8 +139,16 @@ func Run(cfg *config.YurtHubConfiguration, stopCh <-chan struct{}) error {
 	gcMgr.Run()
 	trace++
 
+	klog.Infof("%d. new filter chain for mutating response body", trace)
+	filterChain, err := createFilterChain(cfg)
+	if err != nil {
+		klog.Errorf("could not new filter chain, %v", err)
+		return err
+	}
+	trace++
+
 	klog.Infof("%d. new reverse proxy handler for remote servers", trace)
-	yurtProxyHandler, err := proxy.NewYurtReverseProxyHandler(cfg, cacheMgr, transportManager, healthChecker, certManager, stopCh)
+	yurtProxyHandler, err := proxy.NewYurtReverseProxyHandler(cfg, cacheMgr, transportManager, healthChecker, certManager, filterChain, stopCh)
 	if err != nil {
 		klog.Errorf("could not create reverse proxy handler, %v", err)
 		return err
@@ -156,15 +167,31 @@ func Run(cfg *config.YurtHubConfiguration, stopCh <-chan struct{}) error {
 		klog.Infof("%d. new %s server and begin to serve, dummy proxy server: %s", trace, projectinfo.GetHubName(), cfg.YurtHubProxyServerDummyAddr)
 	}
 
+	// start shared informers here
+	if filterChain != nil && cfg.Filters.Enabled(servicetopology.FilterName) {
+		cfg.SharedFactory.Start(stopCh)
+		cfg.YurtSharedFactory.Start(stopCh)
+	}
+
 	klog.Infof("%d. new %s server and begin to serve, proxy server: %s, hub server: %s", trace, projectinfo.GetHubName(), cfg.YurtHubProxyServerAddr, cfg.YurtHubServerAddr)
 	s, err := server.NewYurtHubServer(cfg, certManager, yurtProxyHandler)
 	if err != nil {
 		klog.Errorf("could not create hub server, %v", err)
 		return err
 	}
 	s.Run()
-
 	klog.Infof("hub agent exited")
 	<-stopCh
 	return nil
 }
+
+func createFilterChain(cfg *config.YurtHubConfiguration) (filter.Interface, error) {
+	if cfg.Filters == nil {
+		return nil, nil
+	}
+
+	genericInitializer := initializer.New(cfg.SharedFactory, cfg.YurtSharedFactory, cfg.SerializerManager, cfg.StorageWrapper, cfg.NodeName, cfg.MutatedMasterServiceAddr)
+	initializerChain := filter.FilterInitializers{}
+	initializerChain = append(initializerChain, genericInitializer)
+	return cfg.Filters.NewFromFilters(initializerChain)
+}

go.mod

@@ -5,36 +5,28 @@ go 1.13
 require (
 	github.com/NYTimes/gziphandler v1.1.1 // indirect
 	github.com/aliyun/alibaba-cloud-sdk-go v1.61.355
-	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/daviddengcn/go-colortext v0.0.0-20160507010035-511bcaf42ccd
 	github.com/docker/docker v17.12.0-ce-rc1.0.20200531234253-77e06fda0c94+incompatible // indirect
 	github.com/emicklei/go-restful v2.12.0+incompatible // indirect
-	github.com/evanphx/json-patch v4.5.0+incompatible // indirect
 	github.com/go-openapi/spec v0.19.8 // indirect
 	github.com/google/uuid v1.1.1
-	github.com/googleapis/gnostic v0.3.1 // indirect
 	github.com/gorilla/mux v1.7.4
-	github.com/hashicorp/golang-lru v0.5.4 // indirect
-	github.com/imdario/mergo v0.3.9 // indirect
-	github.com/json-iterator/go v1.1.10 // indirect
 	github.com/onsi/ginkgo v1.13.0
 	github.com/onsi/gomega v1.10.1
 	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/openyurtio/yurt-app-manager-api v0.18.8
 	github.com/prometheus/client_golang v1.7.1
-	github.com/prometheus/procfs v0.0.11 // indirect
 	github.com/spf13/cobra v1.0.0
 	github.com/spf13/pflag v1.0.5
 	github.com/vishvananda/netlink v1.0.0
 	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d // indirect
-	golang.org/x/text v0.3.3 // indirect
 	golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1 // indirect
 	google.golang.org/grpc v1.27.0
-	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 	gopkg.in/square/go-jose.v2 v2.5.1 // indirect
-	k8s.io/api v0.18.8
-	k8s.io/apimachinery v0.18.8
+	k8s.io/api v0.19.7
+	k8s.io/apimachinery v0.19.7
 	k8s.io/apiserver v0.18.8
-	k8s.io/client-go v0.18.8
+	k8s.io/client-go v0.19.2
 	k8s.io/cluster-bootstrap v0.0.0
 	k8s.io/component-base v0.18.8
 	k8s.io/klog v1.0.0

go.sum

@@ -170,6 +170,7 @@ github.com/go-lintpack/lintpack v0.5.2/go.mod h1:NwZuYi2nUHho8XEIZ6SIxihrnPoqBTD
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logr/logr v0.1.0 h1:M1Tv3VzNlEHg6uyACnRdtrploV2P7wZqH8BoQMtz0cg=
 github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
+github.com/go-logr/zapr v0.1.0/go.mod h1:tabnROwaDl0UNxkVeFRbY8bwB37GwRv0P8lg6aAiEnk=
 github.com/go-ole/go-ole v1.2.1/go.mod h1:7FAglXiTm7HKlQRDeOQ6ZNUHidzCWXuZWq/1dTyBNF8=
 github.com/go-openapi/analysis v0.0.0-20180825180245-b006789cd277/go.mod h1:k70tL6pCuVxPJOHXQ+wIac1FUrvNkHolPie/cLEU6hI=
 github.com/go-openapi/analysis v0.17.0/go.mod h1:IowGgpVeD0vNm45So8nr+IcQ3pxVtpRoBWb8PVZO0ik=
@@ -491,6 +492,8 @@ github.com/opencontainers/runtime-spec v1.0.0/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/
 github.com/opencontainers/selinux v1.3.1-0.20190929122143-5215b1806f52/go.mod h1:+BLncwf63G4dgOzykXAxcmnFlUaOlkDdmw/CqsW6pjs=
 github.com/openyurtio/apiserver-network-proxy v1.18.8 h1:xXqaP8DAOvCHD7DNIqtBOhuWxCnwULLc1PqOMoJ7UeI=
 github.com/openyurtio/apiserver-network-proxy v1.18.8/go.mod h1:X5Au3jBNIgYL2uK0IHeNGnZqlUlVSCFQhi/npPgkKRg=
+github.com/openyurtio/yurt-app-manager-api v0.18.8 h1:VyA1hZMu61aRttqWNKGXp5ct6pNELGct4NCm6xkqEmc=
+github.com/openyurtio/yurt-app-manager-api v0.18.8/go.mod h1:brjCZScD7rNyjD+FVpuqLrXPZFLKGdFP801KMARUtbc=
 github.com/pborman/uuid v1.2.0 h1:J7Q5mO4ysT1dv8hyrUGHb9+ooztCXu1D8MY8DZYsu3g=
 github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
 github.com/pelletier/go-toml v1.1.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
@@ -785,6 +788,7 @@ golang.org/x/tools v0.0.0-20190920225731-5eefd052ad72/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gomodules.xyz/jsonpatch/v2 v2.0.1/go.mod h1:IhYNNY4jnS53ZnfE4PAmpKtDpTCj1JFXc+3mwe7XcUU=
 gonum.org/v1/gonum v0.0.0-20180816165407-929014505bf4/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
 gonum.org/v1/gonum v0.0.0-20190331200053-3d26580ed485/go.mod h1:2ltnJ7xHfj0zHS40VVPYEAAMTa3ZGguvHGBSJeRWqE0=
 gonum.org/v1/gonum v0.6.2/go.mod h1:9mxDZsDKxgMAuccQkewq682L+0eCu4dCN2yonUJTCLU=
@@ -918,6 +922,8 @@ mvdan.cc/unparam v0.0.0-20190209190245-fbb59629db34/go.mod h1:H6SUd1XjIs+qQCyskX
 rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.7 h1:uuHDyjllyzRyCIvvn0OBjiRB0SgBZGqHNYAmjR7fO50=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.7/go.mod h1:PHgbrJT7lCHcxMU+mDHEm+nx46H4zuuHZkDP6icnhu0=
+sigs.k8s.io/controller-runtime v0.5.7 h1:QcB8YQTyMshLLspHiqAkKHO74PgmUAUmTDhol4VccOw=
+sigs.k8s.io/controller-runtime v0.5.7/go.mod h1:KjjGQrdWFaSTHwB5A5VDmX9sMLlvkXjVazxVbfOI3a8=
 sigs.k8s.io/kustomize v2.0.3+incompatible/go.mod h1:MkjgH3RdOWrievjo6c9T245dYlB5QeXV4WCbnt/PEpU=
 sigs.k8s.io/structured-merge-diff/v3 v3.0.0-20200116222232-67a7b8c61874/go.mod h1:PlARxl6Hbt/+BC80dRLi1qAmnMqwqDg62YvvVkZjemw=
 sigs.k8s.io/structured-merge-diff/v3 v3.0.0 h1:dOmIZBMfhcHS09XZkMyUgkq5trg3/jRyJYFZUiaOp8E=

pkg/yurthub/cachemanager/cache_agent.go

@@ -31,6 +31,7 @@ var (
 		"flanneld",
 		"coredns",
 		projectinfo.GetAgentName(),
+		projectinfo.GetHubName(),
 	}
 	cacheAgentsKey = "_internal/cache-manager/cache-agent.conf"
 	sepForAgent    = ","

pkg/yurthub/cachemanager/cache_manager.go

@@ -243,7 +243,6 @@ func (cm *cacheManager) saveWatchObject(ctx context.Context, info *apirequest.Re
 		klog.Errorf("failed to create serializer in saveWatchObject, %s", util.ReqInfoString(info))
 		return fmt.Errorf("failed to create serializer in saveWatchObject, %s", util.ReqInfoString(info))
 	}
-
 	accessor := meta.NewAccessor()
 
 	d, err := s.WatchDecoder(r)

pkg/yurthub/filter/approver.go

@@ -0,0 +1,41 @@
+/*
+Copyright 2021 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 filter
+
+import "k8s.io/apimachinery/pkg/util/sets"
+
+type Approver struct {
+	comp       string
+	resource   string
+	operations sets.String
+}
+
+func NewApprover(comp, resource string, verbs ...string) *Approver {
+	return &Approver{
+		comp:       comp,
+		resource:   resource,
+		operations: sets.NewString(verbs...),
+	}
+}
+
+func (a *Approver) Approve(comp, resource, verb string) bool {
+	if a.comp != comp || a.resource != resource {
+		return false
+	}
+
+	return a.operations.Has(verb)
+}

pkg/yurthub/filter/chain.go

@@ -0,0 +1,61 @@
+/*
+Copyright 2021 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 filter
+
+import (
+	"io"
+	"net/http"
+
+	apirequest "k8s.io/apiserver/pkg/endpoints/request"
+
+	"github.com/openyurtio/openyurt/pkg/yurthub/util"
+)
+
+type filterChain []Interface
+
+func (fc filterChain) Approve(comp, resource, verb string) bool {
+	for _, f := range fc {
+		if f.Approve(comp, resource, verb) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (fc filterChain) Filter(req *http.Request, rc io.ReadCloser, stopCh <-chan struct{}) (int, io.ReadCloser, error) {
+	ctx := req.Context()
+	comp, ok := util.ClientComponentFrom(ctx)
+	if !ok {
+		return 0, rc, nil
+	}
+
+	info, ok := apirequest.RequestInfoFrom(ctx)
+	if !ok {
+		return 0, rc, nil
+	}
+
+	for _, f := range fc {
+		if !f.Approve(comp, info.Resource, info.Verb) {
+			continue
+		}
+
+		return f.Filter(req, rc, stopCh)
+	}
+
+	return 0, rc, nil
+}