6717161144

Committed 01 Nov 2023 07:58AM UTC coverage: 22.807%. First build

Build # 6717161144

Build Type

Pull #174

github

Committed by

web-flow

Commit Message

Merge df31ddbcc into 9d530c589

Pull Request Pull Request #174: feat(scan): deduplicate events sent to be processed

Run Details

244 of 244 new or added lines in 6 files covered. (100.0%)

876 of 3841 relevant lines covered (22.81%)

1.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.11

/continuousscanning/cooldownqueue.go

package continuousscanning

import (
        "context"
        "time"

        lru "github.com/hashicorp/golang-lru/v2/expirable"
        "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
        "k8s.io/apimachinery/pkg/watch"

        "github.com/kubescape/go-logger"
        "github.com/kubescape/go-logger/helpers"
)

const (
        // Default size for the cooldown queue
        DefaultQueueSize = 512
        // Default TTL for events put in the queue
        DefaultTTL = 5 * time.Second
)

// cooldownQueue is a queue that lets clients put events into it with a cooldown
//
// When a client puts an event into a queue, it forwards the event to its
// output channel and starts a cooldown for this event. If a client attempts to
// put the same event into the queue while the cooldown is running, the queue
// will silently drop the event. When the cooldown resets and a client puts the
// same event into the queue, it will be forwarded to the output channel
type cooldownQueue struct {
        seenEvents *lru.LRU[string, bool]
        // inner channel for producing events
        innerChan chan watch.Event
        // public channel for reading events
        ResultChan <-chan watch.Event
}

// NewCooldownQueue returns a new Cooldown Queue
func NewCooldownQueue(size int, cooldown time.Duration) *cooldownQueue {
        lru := lru.NewLRU[string, bool](size, nil, cooldown)
        c := make(chan watch.Event)

        return &cooldownQueue{
                seenEvents: lru,
                innerChan:  c,
                ResultChan: c,
        }

}

func makeEventKey(e watch.Event) string {
        object, ok := e.Object.(*unstructured.Unstructured)
        if !ok {
                return ""
        }

        eventKey := string(e.Type) + "-" + string(object.GetUID())
        return eventKey
}

// Enqueue enqueues an event in the Cooldown Queue
func (q *cooldownQueue) Enqueue(ctx context.Context, e watch.Event) {
        eventKey := makeEventKey(e)
        logger.L().Debug("Adding event to queue", helpers.String("eventKey", eventKey))

        _, exists := q.seenEvents.Get(eventKey)
        if exists {
                logger.L().Debug("key exists, dropping event", helpers.Interface("eventKey", eventKey))
                return
        }

        go func() {
                logger.L().Debug("pushing event", helpers.Interface("eventKey", eventKey))
                q.innerChan <- e
                logger.L().Debug("pushed event", helpers.Interface("event", eventKey))
        }()
        q.seenEvents.Add(eventKey, true)
}

func (q *cooldownQueue) Stop(ctx context.Context) {
        close(q.innerChan)
}

1	package continuousscanning
2
3	import (
4	"context"
5	"time"
6
7	lru "github.com/hashicorp/golang-lru/v2/expirable"
8	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
9	"k8s.io/apimachinery/pkg/watch"
10
11	"github.com/kubescape/go-logger"
12	"github.com/kubescape/go-logger/helpers"
13	)
14
15	const (
16	// Default size for the cooldown queue
17	DefaultQueueSize = 512
18	// Default TTL for events put in the queue
19	DefaultTTL = 5 * time.Second
20	)
21
22	// cooldownQueue is a queue that lets clients put events into it with a cooldown
23	//
24	// When a client puts an event into a queue, it forwards the event to its
25	// output channel and starts a cooldown for this event. If a client attempts to
26	// put the same event into the queue while the cooldown is running, the queue
27	// will silently drop the event. When the cooldown resets and a client puts the
28	// same event into the queue, it will be forwarded to the output channel
29	type cooldownQueue struct {
30	seenEvents *lru.LRU[string, bool]
31	// inner channel for producing events
32	innerChan chan watch.Event
33	// public channel for reading events
34	ResultChan <-chan watch.Event
35	}
36
37	// NewCooldownQueue returns a new Cooldown Queue
38	func NewCooldownQueue(size int, cooldown time.Duration) *cooldownQueue {	2✔
39	lru := lru.NewLRU[string, bool](size, nil, cooldown)	2✔
40	c := make(chan watch.Event)	2✔
41		2✔
42	return &cooldownQueue{	2✔
43	seenEvents: lru,	2✔
44	innerChan: c,	2✔
45	ResultChan: c,	2✔
46	}	2✔
47		2✔
48	}	2✔
49
50	func makeEventKey(e watch.Event) string {	4✔
51	object, ok := e.Object.(*unstructured.Unstructured)	4✔
52	if !ok {	4✔
53	return ""	×
54	}	×
55
56	eventKey := string(e.Type) + "-" + string(object.GetUID())	4✔
57	return eventKey	4✔
58	}
59
60	// Enqueue enqueues an event in the Cooldown Queue
61	func (q *cooldownQueue) Enqueue(ctx context.Context, e watch.Event) {	4✔
62	eventKey := makeEventKey(e)	4✔
63	logger.L().Debug("Adding event to queue", helpers.String("eventKey", eventKey))	4✔
64		4✔
65	_, exists := q.seenEvents.Get(eventKey)	4✔
66	if exists {	4✔
67	logger.L().Debug("key exists, dropping event", helpers.Interface("eventKey", eventKey))	×
68	return	×
69	}	×
70
71	go func() {	8✔
72	logger.L().Debug("pushing event", helpers.Interface("eventKey", eventKey))	4✔
73	q.innerChan <- e	4✔
74	logger.L().Debug("pushed event", helpers.Interface("event", eventKey))	4✔
75	}()	4✔
76	q.seenEvents.Add(eventKey, true)	4✔
77	}
78
79	func (q *cooldownQueue) Stop(ctx context.Context) {	2✔
80	close(q.innerChan)	2✔
81	}	2✔

kubescape / operator / 6717161144

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous