mirrors/kyverno
mirrors/kyverno
1
0
Fork 0
mirror of https://github.com/kyverno/kyverno.git synced 2025-03-31 03:45:17 +00:00
Code Activity

Reduce RCR Throttling (#1545)

Browse source 
* buffer report change requests

Signed-off-by: Shuting Zhao <shutting06@gmail.com>

* fix clusterReportChangeRequest

Signed-off-by: Shuting Zhao <shutting06@gmail.com>

* further reduce RCRs in background scan

Signed-off-by: Shuting Zhao <shutting06@gmail.com>
This commit is contained in:
shuting 2021-02-07 19:46:50 -08:00 • committed by GitHub
parent 6b3ab3fe23
commit bd44dbff41
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 271 additions and 94 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
Makefile
View file

@ -64,6 +64,10 @@ docker-publish-kyverno: docker-build-kyverno docker-tag-repo-kyverno docker-pu
docker-build-kyverno:
@docker build -f $(PWD)/$(KYVERNO_PATH)/Dockerfile -t $(REPO)/$(KYVERNO_IMAGE):$(IMAGE_TAG) . --build-arg LD_FLAGS=$(LD_FLAGS)
docker-build-local-kyverno:
CGO_ENABLED=0 GOOS=linux go build -o $(PWD)/$(KYVERNO_PATH)/kyverno -ldflags=$(LD_FLAGS) $(PWD)/$(KYVERNO_PATH)/main.go
@docker build -f $(PWD)/$(KYVERNO_PATH)/localDockerfile -t $(REPO)/$(KYVERNO_IMAGE):$(IMAGE_TAG) $(PWD)/$(KYVERNO_PATH)
docker-tag-repo-kyverno:
@echo "docker tag $(REPO)/$(KYVERNO_IMAGE):$(IMAGE_TAG) $(REPO)/$(KYVERNO_IMAGE):latest"
@docker tag $(REPO)/$(KYVERNO_IMAGE):$(IMAGE_TAG) $(REPO)/$(KYVERNO_IMAGE):latest

3
cmd/kyverno/localDockerfile Normal file
View file

@ -0,0 +1,3 @@
FROM scratch
ADD kyverno /kyverno
ENTRYPOINT ["/kyverno"]

1
go.mod
View file

@ -26,6 +26,7 @@ require (
github.com/onsi/ginkgo v1.11.0
github.com/onsi/gomega v1.8.1
github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6
github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/pkg/errors v0.9.1
github.com/sirupsen/logrus v1.6.0 // indirect
github.com/spf13/cobra v1.0.0

3
go.sum
View file

@ -540,6 +540,9 @@ github.com/onsi/gomega v1.8.1/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoT
github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6 h1:lNCW6THrCKBiJBpz8kbVGjC7MgdCGKwuvBgc7LoD6sw=
github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI=
github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
github.com/patrickmn/go-cache v1.0.0 h1:3gD5McaYs9CxjyK5AXGcq8gdeCARtd/9gJDUvVeaZ0Y=
github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/paulmach/orb v0.1.3/go.mod h1:VFlX/8C+IQ1p6FTRRKzKoOPJnvEtA5G0Veuqwbu//Vk=
github.com/pborman/getopt v0.0.0-20180729010549-6fdd0a2c7117/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=
github.com/pborman/uuid v1.2.0 h1:J7Q5mO4ysT1dv8hyrUGHb9+ooztCXu1D8MY8DZYsu3g=

7
pkg/policyreport/builder.go
View file

@ -13,13 +13,13 @@ import (
"github.com/kyverno/kyverno/pkg/engine/response"
"github.com/kyverno/kyverno/pkg/engine/utils"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
)
const (
clusterreportchangerequest string = "clusterreportchangerequest"
resourceLabelNamespace string = "kyverno.io/resource.namespace"
deletedLabelResource string = "kyverno.io/delete.resource.name"
deletedLabelResourceKind string = "kyverno.io/delete.resource.kind"
@ -72,7 +72,7 @@ type requestBuilder struct {
}
// NewBuilder ...
func NewBuilder(cpolLister kyvernolister.ClusterPolicyLister, polLister kyvernolister.PolicyLister) *requestBuilder {
func NewBuilder(cpolLister kyvernolister.ClusterPolicyLister, polLister kyvernolister.PolicyLister) Builder {
return &requestBuilder{cpolLister: cpolLister, polLister: polLister}
}
@ -123,6 +123,7 @@ func (builder *requestBuilder) build(info Info) (req *unstructured.Unstructured,
}
}
req.SetCreationTimestamp(metav1.Now())
return req, nil
}
@ -152,7 +153,7 @@ func set(obj *unstructured.Unstructured, info Info) {
obj.SetAPIVersion(request.SchemeGroupVersion.Group + "/" + request.SchemeGroupVersion.Version)
if info.Namespace == "" {
obj.SetGenerateName(clusterreportchangerequest + "-")
obj.SetGenerateName("crcr-")
obj.SetKind("ClusterReportChangeRequest")
} else {
obj.SetGenerateName("rcr-")

240
pkg/policyreport/changerequestcreator.go Normal file
View file

@ -0,0 +1,240 @@
package policyreport
import (
"crypto/rand"
"math/big"
"reflect"
"sync"
"time"
"github.com/go-logr/logr"
"github.com/kyverno/kyverno/pkg/config"
dclient "github.com/kyverno/kyverno/pkg/dclient"
cache "github.com/patrickmn/go-cache"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
)
// creator is an interface that buffers report change requests
// merges and creates requests every tickerInterval
type creator interface {
add(request *unstructured.Unstructured)
create(request *unstructured.Unstructured) error
run(stopChan <-chan struct{})
}
type changeRequestCreator struct {
dclient *dclient.Client
// addCache preserves requests that are to be added to report
RCRCache *cache.Cache
CRCRCache *cache.Cache
// removeCache preserves requests that are to be removed from report
// removeCache *cache.Cache
mutex sync.RWMutex
queue []string
tickerInterval time.Duration
log logr.Logger
}
func newChangeRequestCreator(client *dclient.Client, tickerInterval time.Duration, log logr.Logger) creator {
return &changeRequestCreator{
dclient: client,
RCRCache: cache.New(0, 24*time.Hour),
CRCRCache: cache.New(0, 24*time.Hour),
queue: []string{},
tickerInterval: tickerInterval,
log: log,
}
}
func (c *changeRequestCreator) add(request *unstructured.Unstructured) {
uid, _ := rand.Int(rand.Reader, big.NewInt(100000))
switch request.GetKind() {
case "ClusterReportChangeRequest":
c.CRCRCache.Add(uid.String(), request, cache.NoExpiration)
case "ReportChangeRequest":
c.RCRCache.Add(uid.String(), request, cache.NoExpiration)
default:
return
}
c.mutex.Lock()
c.queue = append(c.queue, uid.String())
c.mutex.Unlock()
}
func (c *changeRequestCreator) create(request *unstructured.Unstructured) error {
ns := ""
if request.GetKind() == "ReportChangeRequest" {
ns = config.KyvernoNamespace
}
_, err := c.dclient.CreateResource(request.GetAPIVersion(), request.GetKind(), ns, request, false)
return err
}
func (c *changeRequestCreator) run(stopChan <-chan struct{}) {
ticker := time.NewTicker(c.tickerInterval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
requests, size := c.mergeRequests()
for _, request := range requests {
if err := c.create(request); err != nil {
c.log.Error(err, "failed to create report change request", "req", request.Object)
}
}
c.cleanupQueue(size)
case <-stopChan:
return
}
}
}
func (c *changeRequestCreator) cleanupQueue(size int) {
c.mutex.Lock()
defer c.mutex.Unlock()
for i := 0; i < size; i++ {
uid := c.queue[i]
c.CRCRCache.Delete(uid)
c.RCRCache.Delete(uid)
}
c.queue = c.queue[size:]
}
// mergeRequests merges all current cached requests
// it blocks writing to the cache
func (c *changeRequestCreator) mergeRequests() (results []*unstructured.Unstructured, size int) {
c.mutex.Lock()
defer c.mutex.Unlock()
mergedCRCR := &unstructured.Unstructured{}
mergedRCR := make(map[string]*unstructured.Unstructured, 0)
size = len(c.queue)
for _, uid := range c.queue {
if unstr, ok := c.CRCRCache.Get(uid); ok {
if crcr, ok := unstr.(*unstructured.Unstructured); ok {
if isDeleteRequest(crcr) {
if !reflect.DeepEqual(mergedCRCR, &unstructured.Unstructured{}) {
results = append(results, mergedCRCR)
mergedCRCR = &unstructured.Unstructured{}
}
results = append(results, crcr)
} else {
if reflect.DeepEqual(mergedCRCR, &unstructured.Unstructured{}) {
mergedCRCR = crcr
continue
}
if ok := merge(mergedCRCR, crcr); !ok {
results = append(results, mergedCRCR)
mergedCRCR = crcr
}
}
}
continue
}
if unstr, ok := c.RCRCache.Get(uid); ok {
if rcr, ok := unstr.(*unstructured.Unstructured); ok {
resourceNS := rcr.GetLabels()[resourceLabelNamespace]
mergedNamespacedRCR, ok := mergedRCR[resourceNS]
if !ok {
mergedNamespacedRCR = &unstructured.Unstructured{}
}
if isDeleteRequest(rcr) {
if !reflect.DeepEqual(mergedNamespacedRCR, &unstructured.Unstructured{}) {
results = append(results, mergedNamespacedRCR)
mergedRCR[resourceNS] = &unstructured.Unstructured{}
}
results = append(results, rcr)
} else {
if reflect.DeepEqual(mergedNamespacedRCR, &unstructured.Unstructured{}) {
mergedRCR[resourceNS] = rcr
continue
}
if ok := merge(mergedNamespacedRCR, rcr); !ok {
results = append(results, mergedNamespacedRCR)
mergedRCR[resourceNS] = rcr
} else {
mergedRCR[resourceNS] = mergedNamespacedRCR
}
}
}
}
}
if !reflect.DeepEqual(mergedCRCR, &unstructured.Unstructured{}) {
results = append(results, mergedCRCR)
}
for _, mergedNamespacedRCR := range mergedRCR {
if !reflect.DeepEqual(mergedNamespacedRCR, &unstructured.Unstructured{}) {
results = append(results, mergedNamespacedRCR)
}
}
return
}
// merge merges elements from a source object into a
// destination object if they share the same namespace label
func merge(dst, src *unstructured.Unstructured) bool {
dstNS := dst.GetLabels()[resourceLabelNamespace]
srcNS := src.GetLabels()[resourceLabelNamespace]
if dstNS != srcNS {
return false
}
if dstResults, ok, _ := unstructured.NestedSlice(dst.UnstructuredContent(), "results"); ok {
if srcResults, ok, _ := unstructured.NestedSlice(src.UnstructuredContent(), "results"); ok {
dstResults = append(dstResults, srcResults...)
if err := unstructured.SetNestedSlice(dst.UnstructuredContent(), dstResults, "results"); err == nil {
addSummary(dst, src)
return true
}
}
}
return false
}
func addSummary(dst, src *unstructured.Unstructured) {
if dstSum, ok, _ := unstructured.NestedMap(dst.UnstructuredContent(), "summary"); ok {
if srcSum, ok, _ := unstructured.NestedMap(src.UnstructuredContent(), "summary"); ok {
for key, dstVal := range dstSum {
if dstValInt, ok := dstVal.(int64); ok {
if srcVal, ok := srcSum[key].(int64); ok {
dstSum[key] = dstValInt + srcVal
}
}
}
}
unstructured.SetNestedMap(dst.UnstructuredContent(), dstSum, "summary")
}
}
func isDeleteRequest(request *unstructured.Unstructured) bool {
deleteLabels := []string{deletedLabelPolicy, deletedLabelRule, deletedLabelResource, deletedLabelResourceKind}
labels := request.GetLabels()
for _, l := range deleteLabels {
if _, ok := labels[l]; ok {
return true
}
}
return false
}

93
pkg/policyreport/reportrequest.go
View file

@ -10,21 +10,14 @@ import (
"github.com/go-logr/logr"
kyverno "github.com/kyverno/kyverno/pkg/api/kyverno/v1"
changerequest "github.com/kyverno/kyverno/pkg/api/kyverno/v1alpha1"
policyreportclient "github.com/kyverno/kyverno/pkg/client/clientset/versioned"
kyvernoinformer "github.com/kyverno/kyverno/pkg/client/informers/externalversions/kyverno/v1"
requestinformer "github.com/kyverno/kyverno/pkg/client/informers/externalversions/kyverno/v1alpha1"
kyvernolister "github.com/kyverno/kyverno/pkg/client/listers/kyverno/v1"
requestlister "github.com/kyverno/kyverno/pkg/client/listers/kyverno/v1alpha1"
"github.com/kyverno/kyverno/pkg/config"
client "github.com/kyverno/kyverno/pkg/dclient"
dclient "github.com/kyverno/kyverno/pkg/dclient"
"github.com/kyverno/kyverno/pkg/engine/response"
"github.com/kyverno/kyverno/pkg/policystatus"
apierrors "k8s.io/apimachinery/pkg/api/errors"
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
unstructured "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/tools/cache"
@ -62,6 +55,9 @@ type Generator struct {
queue workqueue.RateLimitingInterface
dataStore *dataStore
requestCreator creator
log logr.Logger
}
@ -86,6 +82,7 @@ func NewReportChangeRequestGenerator(client *policyreportclient.Clientset,
polListerSynced: polInformer.Informer().HasSynced,
queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), workQueueName),
dataStore: newDataStore(),
requestCreator: newChangeRequestCreator(dclient, 3*time.Second, log.WithName("requestCreator")),
log: log,
}
@ -191,6 +188,8 @@ func (gen *Generator) Run(workers int, stopCh <-chan struct{}) {
go wait.Until(gen.runWorker, time.Second, stopCh)
}
go gen.requestCreator.run(stopCh)
<-stopCh
}
@ -264,91 +263,17 @@ func (gen *Generator) processNextWorkItem() bool {
func (gen *Generator) syncHandler(info Info) error {
builder := NewBuilder(gen.cpolLister, gen.polLister)
rcrUnstructured, err := builder.build(info)
reportReq, err := builder.build(info)
if err != nil {
return fmt.Errorf("unable to build reportChangeRequest: %v", err)
}
if rcrUnstructured == nil {
if reportReq == nil {
return nil
}
gen.log.V(4).Info("reconcile report change request", "key", info.ToKey())
return gen.sync(rcrUnstructured, info)
}
func (gen *Generator) sync(reportReq *unstructured.Unstructured, info Info) error {
logger := gen.log.WithName("sync report change request")
defer logger.V(4).Info("successfully reconciled report change request", "kind", reportReq.GetKind(), "key", info.ToKey())
reportReq.SetCreationTimestamp(v1.Now())
if reportReq.GetKind() == "ClusterReportChangeRequest" {
return gen.syncClusterReportChangeRequest(reportReq, logger)
}
return gen.syncReportChangeRequest(reportReq, logger)
}
func (gen *Generator) syncClusterReportChangeRequest(reportReq *unstructured.Unstructured, logger logr.Logger) error {
old, err := gen.clusterReportChangeRequestLister.Get(reportReq.GetName())
if err != nil {
if apierrors.IsNotFound(err) {
if _, err = gen.dclient.CreateResource(reportReq.GetAPIVersion(), reportReq.GetKind(), "", reportReq, false); err != nil {
return fmt.Errorf("failed to create clusterReportChangeRequest: %v", err)
}
gen.requestCreator.add(reportReq)
return nil
}
return fmt.Errorf("unable to get %s: %v", reportReq.GetKind(), err)
}
return updateReportChangeRequest(gen.dclient, old, reportReq, logger)
}
func (gen *Generator) syncReportChangeRequest(reportReq *unstructured.Unstructured, logger logr.Logger) error {
old, err := gen.reportChangeRequestLister.ReportChangeRequests(config.KyvernoNamespace).Get(reportReq.GetName())
if err != nil {
if apierrors.IsNotFound(err) {
if _, err = gen.dclient.CreateResource(reportReq.GetAPIVersion(), reportReq.GetKind(), config.KyvernoNamespace, reportReq, false); err != nil {
return fmt.Errorf("failed to create ReportChangeRequest: %v", err)
}
return nil
}
return fmt.Errorf("unable to get existing reportChangeRequest %v", err)
}
return updateReportChangeRequest(gen.dclient, old, reportReq, logger)
}
func updateReportChangeRequest(dClient *client.Client, old interface{}, new *unstructured.Unstructured, log logr.Logger) (err error) {
oldUnstructured := make(map[string]interface{})
if oldTyped, ok := old.(*changerequest.ReportChangeRequest); ok {
if oldUnstructured, err = runtime.DefaultUnstructuredConverter.ToUnstructured(oldTyped); err != nil {
return fmt.Errorf("unable to convert reportChangeRequest: %v", err)
}
new.SetResourceVersion(oldTyped.GetResourceVersion())
new.SetUID(oldTyped.GetUID())
} else {
oldTyped := old.(*changerequest.ClusterReportChangeRequest)
if oldUnstructured, err = runtime.DefaultUnstructuredConverter.ToUnstructured(oldTyped); err != nil {
return fmt.Errorf("unable to convert clusterReportChangeRequest: %v", err)
}
new.SetUID(oldTyped.GetUID())
new.SetResourceVersion(oldTyped.GetResourceVersion())
}
if !hasResultsChanged(oldUnstructured, new.UnstructuredContent()) {
log.V(4).Info("unchanged report request", "name", new.GetName())
return nil
}
if _, err = dClient.UpdateResource(new.GetAPIVersion(), new.GetKind(), config.KyvernoNamespace, new, false); err != nil {
return fmt.Errorf("failed to update report request: %v", err)
}
log.V(4).Info("successfully updated report request", "kind", new.GetKind(), "name", new.GetName())
return
}
func hasResultsChanged(old, new map[string]interface{}) bool {