fix: dont dispose the resource watcher when serialization exceptions happen (#534)

This PR prevents the disposal of the resource watcher during JSON
serialization exceptions.

Fixes #477.

Author: tomasfabian

src/KubeOps/Operator/Kubernetes/ResourceWatcher{TEntity}.cs

@@ -155,6 +155,17 @@ private async void RestartWatcher()
 
     private void OnException(Exception e)
     {
+        switch (e)
+        {
+            case SerializationException when
+                e.InnerException is JsonException &&
+                e.InnerException.Message.Contains("The input does not contain any JSON tokens"):
+                _logger.LogDebug(
+                    @"The watcher received an empty response for resource ""{resource}"".",
+                    typeof(TEntity));
+                return;
+        }
+
         var backoff = DefaultBackoff;
 
         try
@@ -174,13 +185,6 @@ private void OnException(Exception e)
                         typeof(TEntity));
                     WatchResource().ConfigureAwait(false);
                     return;
-                case SerializationException when
-                    e.InnerException is JsonException &&
-                    e.InnerException.Message.Contains("The input does not contain any JSON tokens"):
-                    _logger.LogDebug(
-                        @"The watcher received an empty response for resource ""{resource}"".",
-                        typeof(TEntity));
-                    return;
             }
 
             ++_reconnectAttempts;

tests/KubeOps.Test/Operator/Kubernetes/ResourceWatcher{TEntity}.Test.cs

@@ -1,4 +1,6 @@
 using System.Reactive.Linq;
+using System.Runtime.Serialization;
+using System.Text.Json;
 using FluentAssertions;
 using k8s;
 using k8s.Models;
@@ -32,22 +34,7 @@ public async Task Should_Restart_Watcher_On_Exception()
     {
         var settings = new OperatorSettings();
 
-        Action<Exception>? onError = null;
-
-        _client.Setup(
-                c => c.Watch(
-                    It.IsAny<TimeSpan>(),
-                    It.IsAny<Action<WatchEventType, TestResource>>(),
-                    It.IsAny<Action<Exception>?>(),
-                    It.IsAny<Action>(),
-                    null,
-                    It.IsAny<CancellationToken>(),
-                    It.IsAny<string?>()))
-            .Callback<TimeSpan, Action<WatchEventType, TestResource>, Action<Exception>?, Action?, string?,
-                CancellationToken, string?>(
-                (_, _, onErrorArg, _, _, _, _) => { onError = onErrorArg; })
-            .Returns(Task.FromResult(CreateFakeWatcher()))
-            .Verifiable();
+        Func<Action<Exception>?> getOnError = SetupKubernetesClientWatch();
 
         _metrics.Setup(c => c.Running).Returns(Mock.Of<IGauge>());
         _metrics.Setup(c => c.WatcherExceptions).Returns(Mock.Of<ICounter>());
@@ -63,7 +50,7 @@ public async Task Should_Restart_Watcher_On_Exception()
 
         await resourceWatcher.StartAsync();
 
-        onError?.Invoke(new Exception());
+        getOnError()?.Invoke(new Exception());
 
         var backoff = settings.ErrorBackoffStrategy(1);
 
@@ -72,11 +59,8 @@ public async Task Should_Restart_Watcher_On_Exception()
         VerifyWatch(2);
     }
 
-    [Fact]
-    public async Task Should_Not_Throw_Overflow_Exception()
+    private Func<Action<Exception>?> SetupKubernetesClientWatch()
     {
-        var settings = new OperatorSettings();
-
         Action<Exception>? onError = null;
 
         _client.Setup(
@@ -88,12 +72,25 @@ public async Task Should_Not_Throw_Overflow_Exception()
                     null,
                     It.IsAny<CancellationToken>(),
                     It.IsAny<string?>()))
-            .Callback<TimeSpan, Action<WatchEventType, TestResource>, Action<Exception>?, Action?, string?,
-                CancellationToken, string?>(
-                (_, _, onErrorArg, _, _, _, _) => { onError = onErrorArg; })
+            .Callback<TimeSpan, Action<WatchEventType, TestResource>, Action<Exception>?, Action?, string?, CancellationToken, string?>(
+                (_, _, onErrorArg, _, _, _, _) =>
+                {
+                    onError = onErrorArg;
+                })
             .Returns(Task.FromResult(CreateFakeWatcher()))
             .Verifiable();
 
+
+        return () => onError;
+    }
+
+    [Fact]
+    public async Task Should_Not_Throw_Overflow_Exception()
+    {
+        var settings = new OperatorSettings();
+
+        Func<Action<Exception>?> getOnError = SetupKubernetesClientWatch();
+
         _metrics.Setup(c => c.Running).Returns(Mock.Of<IGauge>());
         _metrics.Setup(c => c.WatcherExceptions).Returns(Mock.Of<ICounter>());
 
@@ -111,7 +108,7 @@ public async Task Should_Not_Throw_Overflow_Exception()
         const int numberOfRetries = 40;
         for (int reconnectAttempts = 1; reconnectAttempts <= numberOfRetries; reconnectAttempts++)
         {
-            onError?.Invoke(new Exception());
+            getOnError()?.Invoke(new Exception());
 
             var backoff = settings.ErrorBackoffStrategy(reconnectAttempts > 39 ? 39 : reconnectAttempts);
             if (backoff.TotalSeconds > settings.WatcherMaxRetrySeconds)
@@ -122,24 +119,15 @@ public async Task Should_Not_Throw_Overflow_Exception()
             testScheduler.AdvanceBy(backoff.Add(TimeSpan.FromSeconds(1)).Ticks);
         }
 
-        VerifyWatch(numberOfRetries+1);
+        VerifyWatch(numberOfRetries + 1);
     }
 
     [Fact]
     public async Task Should_Not_Dispose_Reconnect_Subject_Or_Throw_Exception_After_Restarts()
     {
         var settings = new OperatorSettings();
 
-        Action<Exception>? onError = null;
-
-        _client.Setup(c => c.Watch(It.IsAny<TimeSpan>(), It.IsAny<Action<WatchEventType, TestResource>>(), It.IsAny<Action<Exception>?>(), It.IsAny<Action>(), null, It.IsAny<CancellationToken>(), It.IsAny<string?>()))
-            .Callback<TimeSpan, Action<WatchEventType, TestResource>, Action<Exception>?, Action?, string?, CancellationToken, string?>(
-                (_, _, onErrorArg, _, _, _, _) =>
-                {
-                    onError = onErrorArg;
-                })
-            .Returns(Task.FromResult(CreateFakeWatcher()))
-            .Verifiable();
+        Func<Action<Exception>?> getOnError = SetupKubernetesClientWatch();
 
         _metrics.Setup(c => c.Running).Returns(Mock.Of<IGauge>());
         _metrics.Setup(c => c.WatcherExceptions).Returns(Mock.Of<ICounter>());
@@ -154,13 +142,62 @@ public async Task Should_Not_Dispose_Reconnect_Subject_Or_Throw_Exception_After_
 
         var kubernetesException = new KubernetesException(new V1Status());
 
-        onError?.Invoke(kubernetesException);
+        getOnError()?.Invoke(kubernetesException);
 
         resourceWatcher.WatchEvents.Should().NotBeNull();
 
         VerifyWatch(2);
     }
 
+    [Fact]
+    public async Task Should_Not_Restart_On_Serialization_Exception()
+    {
+        var settings = new OperatorSettings();
+
+        Func<Action<Exception>?> getOnError = SetupKubernetesClientWatch();
+
+        _metrics.Setup(c => c.Running).Returns(Mock.Of<IGauge>());
+        _metrics.Setup(c => c.WatcherExceptions).Returns(Mock.Of<ICounter>());
+
+        using var resourceWatcher = new ResourceWatcher<TestResource>(_client.Object, new NullLogger<ResourceWatcher<TestResource>>(), _metrics.Object, settings);
+
+        await resourceWatcher.StartAsync();
+
+        var serializationException = new SerializationException(string.Empty, new JsonException("The input does not contain any JSON tokens"));
+
+        getOnError()?.Invoke(serializationException);
+
+        resourceWatcher.WatchEvents.Should().NotBeNull();
+
+        VerifyWatch(1);
+    }
+
+    [Fact]
+    public async Task Should_Be_Restarted_After_TaskCanceledException_IOException()
+    {
+        var settings = new OperatorSettings();
+
+        Func<Action<Exception>?> getOnError = SetupKubernetesClientWatch();
+
+        SetupResourceWatcherMetrics();
+
+        using var resourceWatcher = new ResourceWatcher<TestResource>(_client.Object, new NullLogger<ResourceWatcher<TestResource>>(), _metrics.Object, settings);
+
+        await resourceWatcher.StartAsync();
+
+        var serializationException = new TaskCanceledException(string.Empty, new IOException($@"Either the server or the client did close the connection on watcher for resource ""{typeof(TestResource)}"". Restart."));
+
+        getOnError()?.Invoke(serializationException);
+
+        VerifyWatch(2);
+    }
+
+    private void SetupResourceWatcherMetrics()
+    {
+        _metrics.Setup(c => c.Running).Returns(Mock.Of<IGauge>());
+        _metrics.Setup(c => c.WatcherExceptions).Returns(Mock.Of<ICounter>());
+    }
+
     [Fact]
     public async Task Should_Publish_On_Watcher_Event()
     {