Fix controller memory leak

Soluto · Mar 29, 2021 · 3cbc22d · 3cbc22d
1 parent b045f14
commit 3cbc22d
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 12 deletions.
diff --git a/src/crd-controller/HostedServices/V1Alpha2Controller.cs b/src/crd-controller/HostedServices/V1Alpha2Controller.cs
@@ -53,6 +53,7 @@ private IDisposable ObserveKamusSecret(CancellationToken token)
                     ApiVersion,
                     "kamussecrets",
                     token)
+                .Take(1)
                 .SelectMany(x =>
                     Observable.FromAsync(async () => await HandleEvent(x.Item1, x.Item2))
                 )

diff --git a/src/crd-controller/utils/KubernetesExtensions.cs b/src/crd-controller/utils/KubernetesExtensions.cs
@@ -2,6 +2,7 @@
 using System.Reactive.Linq;
 using System.Threading;
 using k8s;
+using Microsoft.VisualBasic;
 
 namespace CustomResourceDescriptorController.utils
 {
@@ -13,22 +14,24 @@ public static class KubernetesExtensions
             string version,
             string plural,
             CancellationToken cancellationToken
-            ) where TCRD : class
+        ) where TCRD : class
         {
+            Watcher<TCRD> watcher = null;
             return Observable.FromAsync(async () =>
-            {
-                var subject = new System.Reactive.Subjects.Subject<(WatchEventType, TCRD)>();
-                var path = $"apis/{group}/{version}/watch/{plural}";
-                await kubernetes.WatchObjectAsync<TCRD>(path,
-                    timeoutSeconds: int.MaxValue,
-                    onEvent: (@type, @event) => subject.OnNext((@type, @event)),
-                    onError: e => subject.OnError(e),
-                    onClosed: () => subject.OnCompleted(), cancellationToken: cancellationToken);
-                return subject;
-            })
+                {
+                    var subject = new System.Reactive.Subjects.Subject<(WatchEventType, TCRD)>();
+                    var path = $"apis/{group}/{version}/watch/{plural}";
+                    watcher = await kubernetes.WatchObjectAsync<TCRD>(path,
+                        timeoutSeconds: int.MaxValue,
+                        onEvent: (@type, @event) => subject.OnNext((@type, @event)),
+                        onError: e => subject.OnError(e),
+                        onClosed: () => subject.OnCompleted(), cancellationToken: cancellationToken);
+                    return subject;
+                })
                 .SelectMany(x => x)
                 .Select(t => (t.Item1, t.Item2 as TCRD))
-                .Where(t => t.Item2 != null);
+                .Where(t => t.Item2 != null)
+                .Finally(() => watcher?.Dispose());
         }
     }
 }