feat(operator): add better error handling and reconnection logic

buehler · Oct 6, 2023 · 62db9ad · 62db9ad
1 parent 003950a
commit 62db9ad
Show file tree

Hide file tree

Showing 8 changed files with 27 additions and 131 deletions.
diff --git a/_old/src/KubeOps/Operator/Errors/BackoffStrategies.cs b/_old/src/KubeOps/Operator/Errors/BackoffStrategies.cs
diff --git a/_old/src/KubeOps/Operator/Errors/CrdConversionException.cs b/_old/src/KubeOps/Operator/Errors/CrdConversionException.cs
diff --git a/_old/src/KubeOps/Operator/Errors/CrdPropertyTypeException.cs b/_old/src/KubeOps/Operator/Errors/CrdPropertyTypeException.cs
diff --git a/examples/Operator/Controller/V1SecondEntityController.cs b/examples/Operator/Controller/V1SecondEntityController.cs
diff --git a/examples/Operator/Entities/V1SecondEntity.cs b/examples/Operator/Entities/V1SecondEntity.cs
diff --git a/examples/Operator/Finalizer/FinalizerTwo.cs b/examples/Operator/Finalizer/FinalizerTwo.cs
diff --git a/examples/Operator/todos.txt b/examples/Operator/todos.txt
@@ -1,4 +1,3 @@
 todo:
-- error handling
 - web: webhooks
 - docs
diff --git a/src/KubeOps.Operator/Watcher/ResourceWatcher{TEntity}.cs b/src/KubeOps.Operator/Watcher/ResourceWatcher{TEntity}.cs
@@ -29,6 +29,7 @@ internal class ResourceWatcher<TEntity> : IHostedService
     private readonly ConcurrentDictionary<string, long> _entityCache = new();
     private readonly Lazy<List<FinalizerRegistration>> _finalizers;
     private bool _stopped;
+    private uint _watcherReconnectRetries;
 
     private Watcher<TEntity>? _watcher;
 
@@ -81,21 +82,14 @@ private void WatchResource()
             }
         }
 
+        _logger.LogDebug("""Create watcher for entity of type "{type}".""", typeof(TEntity));
         _watcher = _client.Watch(OnEvent, OnError, OnClosed, @namespace: _settings.Namespace);
     }
 
     private void StopWatching()
     {
         _watcher?.Dispose();
-    }
-
-    private void OnClosed()
-    {
-        _logger.LogDebug("The server closed the connection.");
-        if (!_stopped)
-        {
-            WatchResource();
-        }
+        _watcher = null;
     }
 
     private async void OnEntityRequeue(object? sender, (string Name, string? Namespace) queued)
@@ -116,7 +110,7 @@ private async void OnEntityRequeue(object? sender, (string Name, string? Namespa
         await ReconcileModification(entity);
     }
 
-    private void OnError(Exception e)
+    private async void OnError(Exception e)
     {
         switch (e)
         {
@@ -138,11 +132,34 @@ e.InnerException is EndOfStreamException &&
         }
 
         _logger.LogError(e, """There was an error while watching the resource "{resource}".""", typeof(TEntity));
+        StopWatching();
+        _watcherReconnectRetries++;
+
+        var delay = TimeSpan
+            .FromSeconds(Math.Pow(2, Math.Clamp(_watcherReconnectRetries, 0, 5)))
+            .Add(TimeSpan.FromMilliseconds(new Random().Next(0, 1000)));
+        _logger.LogWarning(
+            "There were {retries} errors / retries in the watcher. Wait {seconds}s before next attempt to connect.",
+            _watcherReconnectRetries,
+            delay.TotalSeconds);
+        await Task.Delay(delay);
+
         WatchResource();
     }
 
+    private void OnClosed()
+    {
+        _logger.LogDebug("The watcher was closed.");
+        if (!_stopped && _watcherReconnectRetries == 0)
+        {
+            WatchResource();
+        }
+    }
+
     private async void OnEvent(WatchEventType type, TEntity entity)
     {
+        _watcherReconnectRetries = 0;
+
         _logger.LogTrace(
             """Received watch event "{eventType}" for "{kind}/{name}".""",
             type,