Skip to content

Commit

Permalink
add data filtering framework
Browse files Browse the repository at this point in the history
1. support endpointslice filter for keeping service traffic in-bound of nodePool
2. support master service mutation for pod use InClusterConfig access kube-apiserver
  • Loading branch information
rambohe-ch committed Jul 26, 2021
1 parent 0bc28b7 commit d7083cd
Show file tree
Hide file tree
Showing 21 changed files with 1,242 additions and 9 deletions.
69 changes: 69 additions & 0 deletions cmd/yurthub/app/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)

Expand Down Expand Up @@ -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
Expand All @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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)
}
10 changes: 9 additions & 1 deletion cmd/yurthub/app/options/options.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
}

Expand Down Expand Up @@ -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
Expand Down
31 changes: 29 additions & 2 deletions cmd/yurthub/app/start.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand All @@ -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)
}
7 changes: 4 additions & 3 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ require (
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
Expand All @@ -31,10 +32,10 @@ require (
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
Expand Down
6 changes: 6 additions & 0 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand Down
1 change: 1 addition & 0 deletions pkg/yurthub/cachemanager/cache_agent.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ var (
"flanneld",
"coredns",
projectinfo.GetAgentName(),
projectinfo.GetHubName(),
}
cacheAgentsKey = "_internal/cache-manager/cache-agent.conf"
sepForAgent = ","
Expand Down
1 change: 0 additions & 1 deletion pkg/yurthub/cachemanager/cache_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
41 changes: 41 additions & 0 deletions pkg/yurthub/filter/approver.go
Original file line number Diff line number Diff line change
@@ -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)
}
61 changes: 61 additions & 0 deletions pkg/yurthub/filter/chain.go
Original file line number Diff line number Diff line change
@@ -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
}
Loading

0 comments on commit d7083cd

Please sign in to comment.