feat: add exponential backoff to restart watcher

Author: Christoph Bühler

src/KubeOps/KubeOps.csproj

@@ -25,7 +25,7 @@
 
     <ItemGroup>
         <PackageReference Include="CompareNETObjects" Version="4.66.0" />
-        <PackageReference Include="KubernetesClient" Version="2.0.18" />
+        <PackageReference Include="KubernetesClient" Version="2.0.21" />
         <PackageReference Include="McMaster.Extensions.CommandLineUtils" Version="3.0.0" />
         <PackageReference Include="McMaster.Extensions.Hosting.CommandLine" Version="3.0.0" />
         <PackageReference Include="YamlDotNet" Version="8.1.1" />

src/KubeOps/Operator/Watcher/EntityWatcher.cs

@@ -7,16 +7,25 @@
 using KubeOps.Operator.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Timer = System.Timers.Timer;
 
 namespace KubeOps.Operator.Watcher
 {
     internal class EntityWatcher<TEntity> : IDisposable
         where TEntity : IKubernetesObject<V1ObjectMeta>
     {
+        private const double MaxRetrySeconds = 64;
+
+        private int _errorCount = 0;
+
         private readonly ILogger<EntityWatcher<TEntity>> _logger;
+        private readonly Random _rnd = new Random();
         private CancellationTokenSource? _cancellation;
         private Watcher<TEntity>? _watcher;
 
+        private Timer? _reconnectTimer;
+        private Timer? _resetErrCountTimer;
+
         private readonly Lazy<IKubernetesClient> _client =
             new Lazy<IKubernetesClient>(() => DependencyInjector.Services.GetRequiredService<IKubernetesClient>());
 
@@ -51,6 +60,8 @@ public void Dispose()
                 WatcherEvent -= (EventHandler<(WatchEventType type, TEntity resource)>) handler;
             }
 
+            _reconnectTimer?.Dispose();
+            _resetErrCountTimer?.Dispose();
             _cancellation?.Dispose();
             _watcher?.Dispose();
             _logger.LogTrace(@"Disposed resource watcher for type ""{type}"".", typeof(TEntity));
@@ -71,6 +82,19 @@ private async Task WatchResource()
                 }
             }
 
+            _resetErrCountTimer = new Timer(TimeSpan.FromSeconds(10).TotalMilliseconds);
+            _resetErrCountTimer.Elapsed += (_, __) =>
+            {
+                _logger.LogTrace("Reset error count in resource watcher.");
+                _errorCount = 0;
+                _resetErrCountTimer.Dispose();
+                _resetErrCountTimer = null;
+                _reconnectTimer?.Stop();
+                _reconnectTimer?.Dispose();
+                _reconnectTimer = null;
+            };
+            _resetErrCountTimer.Start();
+
             _cancellation = new CancellationTokenSource();
             // TODO: namespaced resources
             _watcher = await _client.Value.Watch<TEntity>(
@@ -117,11 +141,19 @@ private void OnWatcherEvent(WatchEventType type, TEntity resource)
         private void OnException(Exception e)
         {
             _logger.LogError(e, @"There was an error while watching the resource ""{resource}"".", typeof(TEntity));
-            // _logger.LogInformation("Trying to reconnect.");
-            // RestartWatcher();
             _cancellation?.Cancel();
             _watcher?.Dispose();
             _watcher = null;
+
+            _logger.LogInformation("Trying to reconnect with exponential backoff.");
+            _resetErrCountTimer?.Stop();
+            _resetErrCountTimer?.Dispose();
+            _resetErrCountTimer = null;
+            _reconnectTimer?.Stop();
+            _reconnectTimer?.Dispose();
+            _reconnectTimer = new Timer(ExponentialBackoff(++_errorCount).TotalMilliseconds);
+            _reconnectTimer.Elapsed += (_, __) => RestartWatcher();
+            _reconnectTimer.Start();
         }
 
         private void OnClose()
@@ -132,5 +164,9 @@ private void OnClose()
                 RestartWatcher();
             }
         }
+
+        private TimeSpan ExponentialBackoff(int retryCount) => TimeSpan
+            .FromSeconds(Math.Min(Math.Pow(2, retryCount), MaxRetrySeconds))
+            .Add(TimeSpan.FromMilliseconds(_rnd.Next(0, 1000)));
     }
 }

tests/KubeOps.Test/KubeOps.Test.csproj

@@ -7,10 +7,10 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.5.0"/>
-        <PackageReference Include="xunit" Version="2.4.0"/>
-        <PackageReference Include="xunit.runner.visualstudio" Version="2.4.0"/>
-        <PackageReference Include="coverlet.collector" Version="1.2.0"/>
+        <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.6.1" />
+        <PackageReference Include="xunit" Version="2.4.1" />
+        <PackageReference Include="xunit.runner.visualstudio" Version="2.4.1" />
+        <PackageReference Include="coverlet.collector" Version="1.2.1" />
     </ItemGroup>
 
 </Project>

tests/KubeOps.TestOperator/Entities/TestEntity.cs

@@ -13,7 +13,7 @@ public class TestEntityStatus
         public string Status { get; set; } = string.Empty;
     }
 
-    [KubernetesEntity(Group = "testing", ApiVersion = "v1")]
+    [KubernetesEntity(Group = "testing.dev", ApiVersion = "v1", PluralName = "testentities")]
     public class TestEntity : CustomKubernetesEntity<TestEntitySpec, TestEntityStatus>
     {
     }

tests/KubeOps.TestOperator/KubeOps.TestOperator.csproj

@@ -9,4 +9,10 @@
       <ProjectReference Include="..\..\src\KubeOps\KubeOps.csproj" />
     </ItemGroup>
 
+    <ItemGroup>
+      <Content Include="appsettings.json">
+        <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+      </Content>
+    </ItemGroup>
+
 </Project>

tests/KubeOps.TestOperator/Program.cs

@@ -1,12 +1,13 @@
-using KubeOps.Operator;
+using System.Threading.Tasks;
+using KubeOps.Operator;
 using KubeOps.TestOperator.Controller;
 using KubeOps.TestOperator.Entities;
 
 namespace KubeOps.TestOperator
 {
     public static class Program
     {
-        public static void Main(string[] args) => new KubernetesOperator()
+        public static Task<int> Main(string[] args) => new KubernetesOperator()
             .ConfigureServices(services => { services.AddResourceController<TestController, TestEntity>(); })
             .Run(args);
     }

tests/KubeOps.TestOperator/appsettings.json

@@ -0,0 +1,9 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Debug",
+      "System": "Information",
+      "Microsoft": "Information"
+    }
+  }
+}