Skip to content

Commit bb2b67c

Browse files
adityathebemoshloop
adityathebe
authored and
moshloop
committed
chore: use priority queue from commons
1 parent afab95c commit bb2b67c

File tree

5 files changed

+49
-52
lines changed

5 files changed

+49
-52
lines changed

‎go.mod

+2-9
Original file line numberDiff line numberDiff line change
@@ -47,10 +47,9 @@ require (
4747
github.com/aws/aws-sdk-go-v2/service/sts v1.30.5
4848
github.com/aws/aws-sdk-go-v2/service/support v1.24.3
4949
github.com/aws/smithy-go v1.22.1
50-
github.com/emirpasic/gods v1.18.1
5150
github.com/evanphx/json-patch v5.7.0+incompatible
5251
github.com/flanksource/artifacts v1.0.14
53-
github.com/flanksource/commons v1.32.1
52+
github.com/flanksource/commons v1.34.0
5453
github.com/flanksource/duty v1.0.769
5554
github.com/flanksource/is-healthy v1.0.53
5655
github.com/flanksource/ketall v1.1.7
@@ -117,10 +116,8 @@ require (
117116
github.com/asecurityteam/rolling v2.0.4+incompatible // indirect
118117
github.com/aws/aws-sdk-go-v2/config v1.27.29 // indirect
119118
github.com/aws/aws-sdk-go-v2/credentials v1.17.29 // indirect
120-
github.com/bahlo/generic-list-go v0.2.0 // indirect
121119
github.com/beorn7/perks v1.0.1 // indirect
122120
github.com/bmatcuk/doublestar/v4 v4.7.1 // indirect
123-
github.com/buger/jsonparser v1.1.1 // indirect
124121
github.com/cenkalti/backoff/v4 v4.2.1 // indirect
125122
github.com/cert-manager/cert-manager v1.9.0 // indirect
126123
github.com/cespare/xxhash/v2 v2.3.0 // indirect
@@ -129,6 +126,7 @@ require (
129126
github.com/distribution/reference v0.5.0 // indirect
130127
github.com/eko/gocache/lib/v4 v4.1.6 // indirect
131128
github.com/eko/gocache/store/go_cache/v4 v4.2.2 // indirect
129+
github.com/emirpasic/gods v1.18.1 // indirect
132130
github.com/emirpasic/gods/v2 v2.0.0-alpha // indirect
133131
github.com/evanphx/json-patch/v5 v5.7.0 // indirect
134132
github.com/exaring/otelpgx v0.6.2 // indirect
@@ -162,7 +160,6 @@ require (
162160
github.com/hashicorp/hcl/v2 v2.21.0 // indirect
163161
github.com/henvic/httpretty v0.1.3 // indirect
164162
github.com/hirochachacha/go-smb2 v1.1.0 // indirect
165-
github.com/invopop/jsonschema v0.12.0 // indirect
166163
github.com/itchyny/gojq v0.12.16 // indirect
167164
github.com/itchyny/timefmt-go v0.1.6 // indirect
168165
github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 // indirect
@@ -207,11 +204,7 @@ require (
207204
github.com/vadimi/go-http-ntlm v1.0.3 // indirect
208205
github.com/vadimi/go-http-ntlm/v2 v2.4.1 // indirect
209206
github.com/vadimi/go-ntlm v1.2.1 // indirect
210-
github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect
211207
github.com/xanzy/ssh-agent v0.3.3 // indirect
212-
github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect
213-
github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
214-
github.com/xeipuuv/gojsonschema v1.2.0 // indirect
215208
github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect
216209
github.com/yuin/gopher-lua v1.1.1 // indirect
217210
github.com/yusufpapurcu/wmi v1.2.4 // indirect

‎go.sum

+2-16
Original file line numberDiff line numberDiff line change
@@ -833,8 +833,6 @@ github.com/aws/aws-sdk-go-v2/service/support v1.24.3 h1:Bbesu6YZvEYACyydELMwUTYY
833833
github.com/aws/aws-sdk-go-v2/service/support v1.24.3/go.mod h1:NvXUhACskXZ2tiXzECpC/97xKzyY7/Wcc1ug5rla7kY=
834834
github.com/aws/smithy-go v1.22.1 h1:/HPHZQ0g7f4eUeK6HKglFz8uwVfZKgoI25rb/J+dnro=
835835
github.com/aws/smithy-go v1.22.1/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg=
836-
github.com/bahlo/generic-list-go v0.2.0 h1:5sz/EEAK+ls5wF+NeqDpk5+iNdMDXrh3z3nPnH1Wvgk=
837-
github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xWbdbCW3pNTGyYg=
838836
github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
839837
github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
840838
github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
@@ -853,8 +851,6 @@ github.com/boombuler/barcode v1.0.0/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl
853851
github.com/boombuler/barcode v1.0.1/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
854852
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
855853
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
856-
github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs=
857-
github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0=
858854
github.com/bwesterb/go-ristretto v1.2.3/go.mod h1:fUIoIZaG73pV5biE2Blr2xEzDoMj7NFEuV9ekS419A0=
859855
github.com/cactus/go-statsd-client/statsd v0.0.0-20200423205355-cb0885a1018c/go.mod h1:l/bIBLeOl9eX+wxJAzxS4TveKRtAqlyDpHjhkfO0MEI=
860856
github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
@@ -972,8 +968,8 @@ github.com/fergusstrange/embedded-postgres v1.25.0 h1:sa+k2Ycrtz40eCRPOzI7Ry7Ttk
972968
github.com/fergusstrange/embedded-postgres v1.25.0/go.mod h1:t/MLs0h9ukYM6FSt99R7InCHs1nW0ordoVCcnzmpTYw=
973969
github.com/flanksource/artifacts v1.0.14 h1:Vv70bccsae0MwGaf/uSPp34J5V1/PyKfct9z5JYCTJU=
974970
github.com/flanksource/artifacts v1.0.14/go.mod h1:qHVCnQu5k50aWNJ5UhpcAKEl7pAzqUrFFKGSm147G70=
975-
github.com/flanksource/commons v1.32.1 h1:380cRHbbjGoxHO8DbiqTrBz7XYw/qUjDIwKDcCNZqmU=
976-
github.com/flanksource/commons v1.32.1/go.mod h1:gO/d401JGFx10+6/9V+PXiGAAVUwxrcLgkke1tGyyNU=
971+
github.com/flanksource/commons v1.34.0 h1:cT9VYWMJDE/KSoPa71UUr1pl764MWBVI1PmrhFSc7B8=
972+
github.com/flanksource/commons v1.34.0/go.mod h1:gO/d401JGFx10+6/9V+PXiGAAVUwxrcLgkke1tGyyNU=
977973
github.com/flanksource/duty v1.0.769 h1:5EOBis382RhFLNkP2hVhYFAnXUdxRXm3GdCUgjYi+hY=
978974
github.com/flanksource/duty v1.0.769/go.mod h1:sZY2NytdenrkqXoMD6Gn2C8xH6dm5HsqOeE0p74Z2VE=
979975
github.com/flanksource/gomplate/v3 v3.20.4/go.mod h1:27BNWhzzSjDed1z8YShO6W+z6G9oZXuxfNFGd/iGSdc=
@@ -1320,8 +1316,6 @@ github.com/imdario/mergo v0.3.16/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+h
13201316
github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
13211317
github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
13221318
github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
1323-
github.com/invopop/jsonschema v0.12.0 h1:6ovsNSuvn9wEQVOyc72aycBMVQFKz7cPdMJn10CvzRI=
1324-
github.com/invopop/jsonschema v0.12.0/go.mod h1:ffZ5Km5SWWRAIN6wbDXItl95euhFz2uON45H2qjYt+0=
13251319
github.com/itchyny/gojq v0.12.13/go.mod h1:JzwzAqenfhrPUuwbmEz3nu3JQmFLlQTQMUcOdnu/Sf4=
13261320
github.com/itchyny/gojq v0.12.16 h1:yLfgLxhIr/6sJNVmYfQjTIv0jGctu6/DgDoivmxTr7g=
13271321
github.com/itchyny/gojq v0.12.16/go.mod h1:6abHbdC2uB9ogMS38XsErnfqJ94UlngIJGlRAIj4jTM=
@@ -1766,16 +1760,8 @@ github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6Kllzaw
17661760
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
17671761
github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQD0Loo=
17681762
github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
1769-
github.com/wk8/go-ordered-map/v2 v2.1.8 h1:5h/BUHu93oj4gIdvHHHGsScSTMijfx5PeYkE/fJgbpc=
1770-
github.com/wk8/go-ordered-map/v2 v2.1.8/go.mod h1:5nJHM5DyteebpVlHnWMV0rPz6Zp7+xBAnxjb1X5vnTw=
17711763
github.com/xanzy/ssh-agent v0.3.3 h1:+/15pJfg/RsTxqYcX6fHqOXZwwMP+2VyYWJeWM2qQFM=
17721764
github.com/xanzy/ssh-agent v0.3.3/go.mod h1:6dzNDKs0J9rVPHPhaGCukekBHKqfl+L3KghI1Bc68Uw=
1773-
github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f h1:J9EGpcZtP0E/raorCMxlFGSTBrsSlaDGf3jU/qvAE2c=
1774-
github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
1775-
github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHovont7NscjpAxXsDA8S8BMYve8Y5+7cuRE7R0=
1776-
github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
1777-
github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74=
1778-
github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
17791765
github.com/xhit/go-str2duration v1.2.0/go.mod h1:3cPSlfZlUHVlneIVfePFWcJZsuwf+P1v2SRTV4cUmp4=
17801766
github.com/xhit/go-str2duration/v2 v2.1.0/go.mod h1:ohY8p+0f07DiV6Em5LKB0s2YpLtXVyJfNt1+BlmyAsU=
17811767
github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 h1:nIPpBwaJSVYIxUFsDv3M8ofmx9yWTog9BfvIu0q41lo=

‎scrapers/incremental.go

+3-7
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ import (
99
"k8s.io/apimachinery/pkg/runtime"
1010
"k8s.io/apimachinery/pkg/types"
1111

12-
pq "github.com/emirpasic/gods/queues/priorityqueue"
12+
"github.com/flanksource/commons/collections"
1313
"github.com/flanksource/config-db/api"
1414
v1 "github.com/flanksource/config-db/api/v1"
1515
"github.com/flanksource/config-db/db"
@@ -32,7 +32,7 @@ func consumeKubernetesWatchJobKey(id string) string {
3232

3333
// ConsumeKubernetesWatchJobFunc returns a job that consumes kubernetes objects received from shared informers
3434
// for the given config of the scrapeconfig.
35-
func ConsumeKubernetesWatchJobFunc(sc api.ScrapeContext, config v1.Kubernetes, queue *pq.Queue) *job.Job {
35+
func ConsumeKubernetesWatchJobFunc(sc api.ScrapeContext, config v1.Kubernetes, queue *collections.Queue[*kubernetes.QueueItem]) *job.Job {
3636
return &job.Job{
3737
Name: "ConsumeKubernetesWatch",
3838
Context: sc.DutyContext().WithObject(sc.ScrapeConfig().ObjectMeta),
@@ -64,7 +64,7 @@ func ConsumeKubernetesWatchJobFunc(sc api.ScrapeContext, config v1.Kubernetes, q
6464
)
6565

6666
for {
67-
val, more := queue.Dequeue()
67+
queueItem, more := queue.Dequeue()
6868
if !more {
6969
break
7070
}
@@ -75,10 +75,6 @@ func ConsumeKubernetesWatchJobFunc(sc api.ScrapeContext, config v1.Kubernetes, q
7575
break
7676
}
7777

78-
queueItem, ok := val.(*kubernetes.QueueItem)
79-
if !ok {
80-
return fmt.Errorf("unexpected item in the priority queue: %T", val)
81-
}
8278
obj := queueItem.Obj
8379
queuedTime[string(obj.GetUID())] = queueItem.Timestamp
8480

‎scrapers/kubernetes/informers.go

+26-14
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ import (
77
"sync"
88
"time"
99

10-
pq "github.com/emirpasic/gods/queues/priorityqueue"
10+
"github.com/flanksource/commons/collections"
1111
"github.com/flanksource/commons/logger"
1212
"github.com/flanksource/config-db/api"
1313
v1 "github.com/flanksource/config-db/api/v1"
@@ -34,10 +34,24 @@ var (
3434
)
3535

3636
// WatchResources watches Kubernetes resources with shared informers
37-
func WatchResources(ctx api.ScrapeContext, config v1.Kubernetes) (*pq.Queue, error) {
38-
priorityQueue := pq.NewWith(pqComparator)
37+
func WatchResources(ctx api.ScrapeContext, config v1.Kubernetes) (*collections.Queue[*QueueItem], error) {
38+
priorityQueue, err := collections.NewQueue(collections.QueueOpts[*QueueItem]{
39+
Metrics: collections.MetricsOpts[*QueueItem]{
40+
Name: "shared_informer",
41+
Labels: map[string]any{
42+
"scraper_id": ctx.ScraperID(),
43+
},
44+
},
45+
Comparator: pqComparator,
46+
Equals: queueItemIsEqual,
47+
Dedupe: true,
48+
})
49+
if err != nil {
50+
return nil, fmt.Errorf("failed to create queue: %w", err)
51+
}
52+
3953
if loaded, ok := WatchQueue.LoadOrStore(config.Hash(), priorityQueue); ok {
40-
priorityQueue = loaded.(*pq.Queue)
54+
priorityQueue = loaded.(*collections.Queue[*QueueItem])
4155
}
4256

4357
if config.Kubeconfig != nil {
@@ -86,7 +100,7 @@ type SharedInformerManager struct {
86100

87101
type DeleteObjHandler func(ctx context.Context, id string) error
88102

89-
func (t *SharedInformerManager) Register(ctx api.ScrapeContext, watchResource v1.KubernetesResourceToWatch, queue *pq.Queue) error {
103+
func (t *SharedInformerManager) Register(ctx api.ScrapeContext, watchResource v1.KubernetesResourceToWatch, queue *collections.Queue[*QueueItem]) error {
90104
registrationTime := time.Now()
91105

92106
apiVersion, kind := watchResource.ApiVersion, watchResource.Kind
@@ -385,21 +399,19 @@ func NewQueueItem(obj *unstructured.Unstructured, operation QueueItemOperation)
385399
}
386400
}
387401

388-
func pqComparator(a, b any) int {
389-
if a == nil || b == nil {
390-
return 0
391-
}
392-
393-
qa := a.(*QueueItem)
394-
qb := b.(*QueueItem)
402+
func queueItemIsEqual(qa, qb *QueueItem) bool {
403+
return qa.Obj.GetUID() == qb.Obj.GetUID()
404+
}
395405

406+
func pqComparator(qa, qb *QueueItem) int {
396407
if qa.Obj.GetUID() == qb.Obj.GetUID() {
397408
resourceVersionA, ok, _ := unstructured.NestedString(qa.Obj.Object, "metadata", "resourceVersion")
398409
if ok {
399410
resourceVersionB, _, _ := unstructured.NestedString(qb.Obj.Object, "metadata", "resourceVersion")
400411

401-
// Because of the way we are deduping, we want the earlier version in front of the queue.
402-
return strings.Compare(resourceVersionA, resourceVersionB)
412+
// Because of the way we are deduping, we want the latest version in front of the queue.
413+
// the later versions are discarded.
414+
return strings.Compare(resourceVersionB, resourceVersionA)
403415
}
404416
}
405417

‎scrapers/kubernetes/informers_test.go

+16-6
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,8 @@ import (
66
"testing"
77
"time"
88

9-
"github.com/emirpasic/gods/queues/priorityqueue"
9+
"github.com/flanksource/commons/collections"
10+
"github.com/flanksource/commons/hash"
1011
"github.com/google/uuid"
1112
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
1213
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -166,27 +167,35 @@ func TestPqComparator(t *testing.T) {
166167
Obj: getUnstructuredWithResourceVersion("Pod", "a", "2c6a2f24-0199-435d-83a6-bd3f6d18d06d", "4", now.Add(-1*time.Hour)),
167168
},
168169
},
169-
expected: []string{"a-1", "a-2", "a-3", "a-4"},
170+
expected: []string{"a-4"},
170171
},
171172
}
172173

173174
for _, tt := range tests {
174175
t.Run(tt.name, func(t *testing.T) {
175-
q := priorityqueue.NewWith(pqComparator)
176+
q, err := collections.NewQueue(collections.QueueOpts[*QueueItem]{
177+
Metrics: collections.MetricsOpts[*QueueItem]{
178+
Name: fmt.Sprintf("m_%s", hash.Sha256Hex(tt.name)[:10]),
179+
},
180+
Comparator: pqComparator,
181+
Equals: queueItemIsEqual,
182+
Dedupe: true,
183+
})
184+
if err != nil {
185+
t.Fatalf("failed to create queue: %v", err)
186+
}
176187

177188
for _, item := range tt.Items {
178189
q.Enqueue(&item)
179190
}
180191

181192
var result []string
182193
for {
183-
v, ok := q.Dequeue()
194+
item, ok := q.Dequeue()
184195
if !ok {
185196
break
186197
}
187198

188-
item := v.(*QueueItem)
189-
190199
resourceVersion, ok, _ := unstructured.NestedString(item.Obj.Object, "metadata", "resourceVersion")
191200
if ok {
192201
result = append(result, fmt.Sprintf("%s-%s", item.Obj.GetName(), resourceVersion))
@@ -207,6 +216,7 @@ func getUnstructuredEvent(kind, name string, creationTimestamp, recreationTimest
207216
Object: map[string]any{
208217
"kind": kind,
209218
"metadata": map[string]any{
219+
"uid": uuid.NewString(),
210220
"name": name,
211221
"creationTimestamp": creationTimestamp.Format(time.RFC3339),
212222
"managedFields": []any{

0 commit comments

Comments
 (0)