Rate limit site info events

lib/node.go

@@ -0,0 +1,127 @@
+/*
+Copyright AppsCode Inc. and Contributors
+
+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 lib
+
+import (
+	"fmt"
+
+	api "go.bytebuilders.dev/audit/api/v1"
+
+	"gomodules.xyz/counter/hourly"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+var siteEventCounter hourly.Int32
+
+type SiteInfoPublisher struct {
+	p           *EventPublisher
+	createEvent EventCreator
+}
+
+var _ cache.ResourceEventHandler = &SiteInfoPublisher{}
+
+func (p *SiteInfoPublisher) OnAdd(o interface{}, isInInitialList bool) {
+	obj, ok := o.(client.Object)
+	if !ok {
+		return
+	}
+
+	ev, err := p.createEvent(obj)
+	if err != nil {
+		klog.V(5).InfoS("failed to create event data", "error", err)
+		return
+	}
+
+	if err = p.p.Publish(ev, api.EventCreated); err != nil {
+		klog.V(5).InfoS("error while publishing event", "error", err)
+		return
+	}
+
+	siteEventCounter.Inc()
+}
+
+func (p *SiteInfoPublisher) OnUpdate(oldObj, newObj interface{}) {
+	uOld, ok := oldObj.(client.Object)
+	if !ok {
+		return
+	}
+	uNew, ok := newObj.(client.Object)
+	if !ok {
+		return
+	}
+
+	alreadySentHourly := siteEventCounter.Get() > 0
+	if alreadySentHourly &&
+		uOld.GetUID() == uNew.GetUID() && uOld.GetGeneration() == uNew.GetGeneration() {
+		if klog.V(8).Enabled() {
+			klog.V(8).InfoS("skipping update event",
+				"gvk", uNew.GetObjectKind().GroupVersionKind(),
+				"namespace", uNew.GetNamespace(),
+				"name", uNew.GetName(),
+			)
+		}
+		return
+	}
+
+	ev, err := p.createEvent(uNew)
+	if err != nil {
+		klog.V(5).InfoS("failed to create event data", "error", err)
+		return
+	}
+
+	if err = p.p.Publish(ev, api.EventUpdated); err != nil {
+		klog.V(5).InfoS("failed to publish event", "error", err)
+		return
+	}
+
+	siteEventCounter.Inc()
+}
+
+func (p *SiteInfoPublisher) OnDelete(obj interface{}) {
+	var object client.Object
+	var ok bool
+	if object, ok = obj.(client.Object); !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			klog.V(5).Info("error decoding object, invalid type")
+			return
+		}
+		object, ok = tombstone.Obj.(client.Object)
+		if !ok {
+			utilruntime.HandleError(fmt.Errorf("error decoding object tombstone, invalid type"))
+			return
+		}
+		klog.V(5).Infof("Recovered deleted object '%v' from tombstone", tombstone.Obj.(metav1.Object).GetName())
+	}
+
+	ev, err := p.createEvent(object)
+	if err != nil {
+		klog.V(5).InfoS("failed to create event data", "error", err)
+		return
+	}
+
+	if err := p.p.Publish(ev, api.EventDeleted); err != nil {
+		klog.V(5).InfoS("failed to publish event", "error", err)
+		return
+	}
+
+	siteEventCounter.Inc()
+}

lib/publisher.go

@@ -29,13 +29,10 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2/event"
 	"github.com/nats-io/nats.go"
 	"go.bytebuilders.dev/license-verifier/info"
-	"gomodules.xyz/counter/hourly"
 	"gomodules.xyz/sync"
 	core "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -52,6 +49,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
+const eventInterval = 1 * time.Hour
+
 // Informer - informer allows you interact with the underlying informer.
 type Informer interface {
 	// AddEventHandlerWithResyncPeriod adds an event handler to the shared informer using the
@@ -193,7 +192,7 @@ func (p *EventPublisher) ForGVK(informer Informer, gvk schema.GroupVersionKind)
 			return ev, nil
 		},
 	}
-	_, _ = informer.AddEventHandlerWithResyncPeriod(h, 1*time.Hour)
+	_, _ = informer.AddEventHandlerWithResyncPeriod(h, eventInterval)
 }
 
 type funcNodeLister func() ([]*core.Node, error)
@@ -240,7 +239,7 @@ func (p *EventPublisher) setupSiteInfoPublisher(cfg *rest.Config, kc kubernetes.
 		p.si.Product = new(auditorapi.ProductInfo)
 	}
 
-	_, err = nodeInformer.AddEventHandler(&ResourceEventPublisher{
+	_, err = nodeInformer.AddEventHandlerWithResyncPeriod(&SiteInfoPublisher{
 		p: p,
 		createEvent: func(_ client.Object) (*api.Event, error) {
 			cmeta, err := clusterid.ClusterMetadata(kc.CoreV1().Namespaces())
@@ -277,7 +276,7 @@ func (p *EventPublisher) setupSiteInfoPublisher(cfg *rest.Config, kc kubernetes.
 			}
 			return ev, nil
 		},
-	})
+	}, eventInterval)
 	return err
 }
 
@@ -300,116 +299,3 @@ func (p *EventPublisher) SetupWithManager(ctx context.Context, mgr manager.Manag
 	}
 	return p.SetupWithManagerForKind(ctx, mgr, gvk)
 }
-
-type ResourceEventPublisher struct {
-	p           *EventPublisher
-	createEvent EventCreator
-
-	counters map[kmapi.OID]int32
-	mu       gosync.Mutex
-}
-
-var _ cache.ResourceEventHandler = &ResourceEventPublisher{}
-
-func (p *ResourceEventPublisher) OnAdd(o interface{}, isInInitialList bool) {
-	obj, ok := o.(client.Object)
-	if !ok {
-		return
-	}
-
-	ev, err := p.createEvent(obj)
-	if err != nil {
-		klog.V(5).InfoS("failed to create event data", "error", err)
-		return
-	}
-
-	if err = p.p.Publish(ev, api.EventCreated); err != nil {
-		klog.V(5).InfoS("error while publishing event", "error", err)
-		return
-	}
-
-	p.mu.Lock()
-	c := p.counters[kmapi.NewObjectID(obj).OID()]
-	hourly.Inc32(&c)
-	p.mu.Unlock()
-}
-
-func (p *ResourceEventPublisher) OnUpdate(oldObj, newObj interface{}) {
-	uOld, ok := oldObj.(client.Object)
-	if !ok {
-		return
-	}
-	uNew, ok := newObj.(client.Object)
-	if !ok {
-		return
-	}
-
-	alreadySentHourly := false
-	oid := kmapi.NewObjectID(uNew).OID()
-	p.mu.Lock()
-	c := p.counters[oid]
-	alreadySentHourly = hourly.Get32(&c) > 0
-	p.mu.Unlock()
-
-	if alreadySentHourly &&
-		uOld.GetUID() == uNew.GetUID() && uOld.GetGeneration() == uNew.GetGeneration() {
-		if klog.V(8).Enabled() {
-			klog.V(8).InfoS("skipping update event",
-				"gvk", uNew.GetObjectKind().GroupVersionKind(),
-				"namespace", uNew.GetNamespace(),
-				"name", uNew.GetName(),
-			)
-		}
-		return
-	}
-
-	ev, err := p.createEvent(uNew)
-	if err != nil {
-		klog.V(5).InfoS("failed to create event data", "error", err)
-		return
-	}
-
-	if err = p.p.Publish(ev, api.EventUpdated); err != nil {
-		klog.V(5).InfoS("failed to publish event", "error", err)
-		return
-	}
-
-	p.mu.Lock()
-	c = p.counters[oid]
-	hourly.Inc32(&c)
-	p.mu.Unlock()
-}
-
-func (p *ResourceEventPublisher) OnDelete(obj interface{}) {
-	var object client.Object
-	var ok bool
-	if object, ok = obj.(client.Object); !ok {
-		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
-		if !ok {
-			klog.V(5).Info("error decoding object, invalid type")
-			return
-		}
-		object, ok = tombstone.Obj.(client.Object)
-		if !ok {
-			utilruntime.HandleError(fmt.Errorf("error decoding object tombstone, invalid type"))
-			return
-		}
-		klog.V(5).Infof("Recovered deleted object '%v' from tombstone", tombstone.Obj.(metav1.Object).GetName())
-	}
-
-	ev, err := p.createEvent(object)
-	if err != nil {
-		klog.V(5).InfoS("failed to create event data", "error", err)
-		return
-	}
-
-	if err := p.p.Publish(ev, api.EventDeleted); err != nil {
-		klog.V(5).InfoS("failed to publish event", "error", err)
-		return
-	}
-
-	p.mu.Lock()
-	c := p.counters[kmapi.NewObjectID(object).OID()]
-	hourly.Inc32(&c)
-	p.mu.Unlock()
-}