Skip to content

Commit 5e60256

Browse files
slashbenslashben
authored
Merge pull request #54 from kubescape/hotfix/watcher
Revert "Merge pull request #53 from kubescape/bugfix/memory-leak"
2 parents 0220117 + a561cbe commit 5e60256

File tree

4 files changed

+67
-87
lines changed

4 files changed

+67
-87
lines changed

‎Makefile

+1-1
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ IMAGE_NAME ?= kapprofiler:latest
1414

1515

1616
$(BINARY_NAME): $(GOFILES) go.mod go.sum Makefile
17-
CGO_ENABLED=0 go build -o $(BINARY_NAME) -v
17+
CGO_ENABLED=1 go build -o $(BINARY_NAME) -v
1818

1919
test:
2020
$(GOTEST_SUDO_PREFIX) $(GOTEST) -v ./...

‎pkg/collector/types.go

+2-9
Original file line numberDiff line numberDiff line change
@@ -130,15 +130,8 @@ func (a DnsCalls) Equals(b DnsCalls) bool {
130130
if len(a.Addresses) != len(b.Addresses) {
131131
return false
132132
}
133-
for _, addr := range a.Addresses {
134-
found := false
135-
for _, baddr := range b.Addresses {
136-
if addr == baddr {
137-
found = true
138-
break
139-
}
140-
}
141-
if !found {
133+
for i, addr := range a.Addresses {
134+
if addr != b.Addresses[i] {
142135
return false
143136
}
144137
}

‎pkg/controller/controller.go

+8-6
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@ func (c *Controller) StartController() {
6363
},
6464
UpdateFunc: func(obj *unstructured.Unstructured) {
6565
c.handleApplicationProfile(obj)
66+
6667
},
6768
DeleteFunc: func(obj *unstructured.Unstructured) {
6869
c.handleApplicationProfile(obj)
@@ -76,6 +77,7 @@ func (c *Controller) StartController() {
7677

7778
// Set the watcher
7879
c.watcher = appProfileWatcher
80+
7981
}
8082

8183
// Stop the AppProfile controller
@@ -110,7 +112,7 @@ func (c *Controller) handleApplicationProfile(applicationProfileUnstructured *un
110112
if err != nil {
111113
return
112114
}
113-
deploymentApplicationProfile := collector.ApplicationProfile{
115+
deploymentApplicationProfile := &collector.ApplicationProfile{
114116
TypeMeta: metav1.TypeMeta{
115117
Kind: collector.ApplicationProfileKind,
116118
APIVersion: collector.ApplicationProfileApiVersion,
@@ -123,7 +125,7 @@ func (c *Controller) handleApplicationProfile(applicationProfileUnstructured *un
123125
Containers: applicationProfile.Spec.Containers,
124126
},
125127
}
126-
deploymentApplicationProfileRaw, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&deploymentApplicationProfile)
128+
deploymentApplicationProfileRaw, err := runtime.DefaultUnstructuredConverter.ToUnstructured(deploymentApplicationProfile)
127129
if err != nil {
128130
return
129131
}
@@ -143,7 +145,7 @@ func (c *Controller) handleApplicationProfile(applicationProfileUnstructured *un
143145
return
144146
}
145147

146-
deploymentApplicationProfile := collector.ApplicationProfile{}
148+
deploymentApplicationProfile := &collector.ApplicationProfile{}
147149
deploymentApplicationProfile.Labels = applicationProfile.GetLabels()
148150
deploymentApplicationProfile.Spec.Containers = applicationProfile.Spec.Containers
149151
deploymentApplicationProfileRaw, _ := json.Marshal(deploymentApplicationProfile)
@@ -355,7 +357,7 @@ func (c *Controller) handleApplicationProfile(applicationProfileUnstructured *un
355357
// Fetch ApplicationProfile of the controller
356358
existingApplicationProfile, err := c.dynamicClient.Resource(collector.AppProfileGvr).Namespace(pod.Namespace).Get(context.TODO(), applicationProfileNameForController, metav1.GetOptions{})
357359
if err != nil { // ApplicationProfile of controller doesn't exist so create a new one
358-
controllerApplicationProfile := collector.ApplicationProfile{
360+
controllerApplicationProfile := &collector.ApplicationProfile{
359361
TypeMeta: metav1.TypeMeta{
360362
Kind: collector.ApplicationProfileKind,
361363
APIVersion: collector.ApplicationProfileApiVersion,
@@ -368,7 +370,7 @@ func (c *Controller) handleApplicationProfile(applicationProfileUnstructured *un
368370
Containers: containers,
369371
},
370372
}
371-
controllerApplicationProfileRaw, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&controllerApplicationProfile)
373+
controllerApplicationProfileRaw, err := runtime.DefaultUnstructuredConverter.ToUnstructured(controllerApplicationProfile)
372374
if err != nil {
373375
log.Printf("Error converting ApplicationProfile of controller %v", err)
374376
return
@@ -384,7 +386,7 @@ func (c *Controller) handleApplicationProfile(applicationProfileUnstructured *un
384386
// Don't update the application profile
385387
return
386388
}
387-
controllerApplicationProfile := collector.ApplicationProfile{}
389+
controllerApplicationProfile := &collector.ApplicationProfile{}
388390
controllerApplicationProfile.Labels = applicationProfileUnstructured.GetLabels()
389391
controllerApplicationProfile.Spec.Containers = containers
390392
controllerApplicationProfileRaw, _ := json.Marshal(controllerApplicationProfile)

‎pkg/watcher/watcher.go

+56-71
Original file line numberDiff line numberDiff line change
@@ -90,84 +90,69 @@ func (w *Watcher) Start(notifyF WatchNotifyFunctions, gvr schema.GroupVersionRes
9090
}
9191
w.watcher = watcher
9292
w.running = true
93-
94-
// Use a context to gracefully stop the goroutine
95-
ctx, cancel := context.WithCancel(context.Background())
96-
defer cancel()
97-
9893
go func() {
9994
// Watch for events
100-
defer func() {
101-
// Ensure the watcher is closed when the goroutine exits
102-
watcher.Stop()
103-
}()
10495

10596
for {
106-
select {
107-
case event, ok := <-watcher.ResultChan():
108-
if !ok {
109-
if w.running {
110-
// Need to restart the watcher: wait a bit and restart
111-
time.Sleep(5 * time.Second)
112-
listOptions.ResourceVersion = resourceVersion
113-
newWatcher, err := w.client.Resource(gvr).Namespace("").Watch(context.Background(), listOptions)
114-
if err != nil {
115-
log.Printf("watcher restart error: %v", err)
116-
return
117-
}
118-
w.watcher = newWatcher
119-
// Restart the loop
120-
continue
121-
} else {
122-
// Stop the watcher
123-
return
97+
event, ok := <-watcher.ResultChan()
98+
if !ok {
99+
if w.running {
100+
// Need to restart the watcher: wait a bit and restart
101+
time.Sleep(5 * time.Second)
102+
listOptions.ResourceVersion = resourceVersion
103+
w.watcher, err = w.client.Resource(gvr).Namespace("").Watch(context.Background(), listOptions)
104+
if err != nil {
105+
log.Printf("watcher restart error: %v", err)
124106
}
107+
// Restart the loop
108+
continue
109+
} else {
110+
// Stop the watcher
111+
return
125112
}
126-
127-
switch event.Type {
128-
case watch.Added:
129-
// Convert the object to unstructured
130-
addedObject := event.Object.(*unstructured.Unstructured)
131-
if addedObject == nil {
132-
log.Printf("watcher error: addedObject is nil")
133-
continue
134-
}
135-
// Update the resourceVersion
136-
if addedObject.GetResourceVersion() > resourceVersion {
137-
resourceVersion = addedObject.GetResourceVersion()
138-
}
139-
notifyF.AddFunc(addedObject)
140-
case watch.Modified:
141-
// Convert the object to unstructured
142-
modifiedObject := event.Object.(*unstructured.Unstructured)
143-
if modifiedObject == nil {
144-
log.Printf("watcher error: modifiedObject is nil")
145-
continue
146-
}
147-
// Update the resourceVersion
148-
if modifiedObject.GetResourceVersion() > resourceVersion {
149-
resourceVersion = modifiedObject.GetResourceVersion()
150-
}
151-
notifyF.UpdateFunc(modifiedObject)
152-
case watch.Deleted:
153-
// Convert the object to unstructured
154-
deletedObject := event.Object.(*unstructured.Unstructured)
155-
if deletedObject == nil {
156-
log.Printf("watcher error: deletedObject is nil")
157-
continue
158-
}
159-
// Update the resourceVersion
160-
if deletedObject.GetResourceVersion() > resourceVersion {
161-
resourceVersion = deletedObject.GetResourceVersion()
162-
}
163-
notifyF.DeleteFunc(deletedObject)
164-
case watch.Error:
165-
log.Printf("watcher error: %v", event.Object)
113+
}
114+
switch event.Type {
115+
case watch.Added:
116+
// Convert the object to unstructured
117+
addedObject := event.Object.(*unstructured.Unstructured)
118+
if addedObject == nil {
119+
log.Printf("watcher error: addedObject is nil")
120+
continue
166121
}
167-
168-
case <-ctx.Done():
169-
// Exit the goroutine when the context is canceled
170-
return
122+
// Update the resourceVersion
123+
if addedObject.GetResourceVersion() > resourceVersion {
124+
resourceVersion = addedObject.GetResourceVersion()
125+
}
126+
notifyF.AddFunc(addedObject)
127+
addedObject = nil // Make sure the item is scraped by the GC
128+
case watch.Modified:
129+
// Convert the object to unstructured
130+
modifiedObject := event.Object.(*unstructured.Unstructured)
131+
if modifiedObject == nil {
132+
log.Printf("watcher error: modifiedObject is nil")
133+
continue
134+
}
135+
// Update the resourceVersion
136+
if modifiedObject.GetResourceVersion() > resourceVersion {
137+
resourceVersion = modifiedObject.GetResourceVersion()
138+
}
139+
notifyF.UpdateFunc(modifiedObject)
140+
modifiedObject = nil // Make sure the item is scraped by the GC
141+
case watch.Deleted:
142+
// Convert the object to unstructured
143+
deletedObject := event.Object.(*unstructured.Unstructured)
144+
if deletedObject == nil {
145+
log.Printf("watcher error: deletedObject is nil")
146+
continue
147+
}
148+
// Update the resourceVersion
149+
if deletedObject.GetResourceVersion() > resourceVersion {
150+
resourceVersion = deletedObject.GetResourceVersion()
151+
}
152+
notifyF.DeleteFunc(deletedObject)
153+
deletedObject = nil // Make sure the item is scraped by the GC
154+
case watch.Error:
155+
log.Printf("watcher error: %v", event.Object)
171156
}
172157
}
173158
}()

0 commit comments

Comments
 (0)