test: Improved ResourceWatcher restart unit test (#527)

#482 #522 #523 #526

---------

Co-authored-by: Christoph Bühler <[email protected]>

Author: tomasfabian

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

@@ -1,4 +1,5 @@
-using System.Reactive.Linq;
+using System.Reactive.Concurrency;
+using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using System.Runtime.Serialization;
 using System.Text.Json;
@@ -39,14 +40,16 @@ public ResourceWatcher(
         _settings = settings;
         _reconnectSubscription =
             _reconnectHandler
-                .Select(Observable.Timer)
+                .Select(backoff => Observable.Timer(backoff, TimeBasedScheduler))
                 .Switch()
                 .Retry()
                 .Subscribe(async _ => await WatchResource(), error => _logger.LogError(error, $"There was an error while restarting the resource watcher {typeof(TEntity)}"));
     }
 
     public IObservable<WatchEvent> WatchEvents => _watchEvents;
 
+    internal IScheduler TimeBasedScheduler { get; set; } = DefaultScheduler.Instance;
+
     private TimeSpan DefaultBackoff => _settings.ErrorBackoffStrategy(1);
 
     public Task StartAsync()

tests/KubeOps.Test/KubeOps.Test.csproj

@@ -2,6 +2,10 @@
 
   <Import Project="..\..\config\CommonTests.targets" />
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Reactive.Testing" Version="5.0.0" />
+  </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\..\src\KubeOps.Testing\KubeOps.Testing.csproj" />
     <ProjectReference Include="..\..\src\KubeOps\KubeOps.csproj" />

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

@@ -1,12 +1,13 @@
-using FluentAssertions;
+using System.Reactive.Linq;
+using FluentAssertions;
 using k8s;
 using k8s.Models;
 using KubeOps.KubernetesClient;
 using KubeOps.Operator;
 using KubeOps.Operator.DevOps;
 using KubeOps.Operator.Kubernetes;
-using KubeOps.Testing;
 using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Reactive.Testing;
 using Moq;
 using Prometheus;
 using Xunit;
@@ -23,24 +24,238 @@ public class TestResource : IKubernetesObject<V1ObjectMeta>
         public V1ObjectMeta Metadata { get; set; } = null!;
     }
 
-    private readonly IKubernetesClient _client = new MockKubernetesClient();
+    private readonly Mock<IKubernetesClient> _client = new();
     private readonly Mock<IResourceWatcherMetrics<TestResource>> _metrics = new();
 
+    [Fact]
+    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();
+
+        _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);
+
+        var testScheduler = new TestScheduler();
+        resourceWatcher.TimeBasedScheduler = testScheduler;
+
+        await resourceWatcher.StartAsync();
+
+        onError?.Invoke(new Exception());
+
+        var backoff = settings.ErrorBackoffStrategy(1);
+
+        testScheduler.AdvanceBy(backoff.Add(TimeSpan.FromSeconds(1)).Ticks);
+
+        VerifyWatch(2);
+    }
+
+    [Fact]
+    public async Task Should_Not_Throw_Overflow_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();
+
+        _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);
+
+        var testScheduler = new TestScheduler();
+        resourceWatcher.TimeBasedScheduler = testScheduler;
+
+        await resourceWatcher.StartAsync();
+
+        const int numberOfRetries = 40;
+        for (int reconnectAttempts = 1; reconnectAttempts <= numberOfRetries; reconnectAttempts++)
+        {
+            onError?.Invoke(new Exception());
+
+            var backoff = settings.ErrorBackoffStrategy(reconnectAttempts > 39 ? 39 : reconnectAttempts);
+            if (backoff.TotalSeconds > settings.WatcherMaxRetrySeconds)
+            {
+                backoff = TimeSpan.FromSeconds(settings.WatcherMaxRetrySeconds);
+            }
+
+            testScheduler.AdvanceBy(backoff.Add(TimeSpan.FromSeconds(1)).Ticks);
+        }
+
+        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();
+
         _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, new NullLogger<ResourceWatcher<TestResource>>(), _metrics.Object, settings);
+        using var resourceWatcher = new ResourceWatcher<TestResource>(_client.Object, new NullLogger<ResourceWatcher<TestResource>>(), _metrics.Object, settings);
 
         await resourceWatcher.StartAsync();
 
         await resourceWatcher.StopAsync();
 
         await resourceWatcher.StartAsync();
 
+        var kubernetesException = new KubernetesException(new V1Status());
+
+        onError?.Invoke(kubernetesException);
+
         resourceWatcher.WatchEvents.Should().NotBeNull();
+
+        VerifyWatch(2);
+    }
+
+    [Fact]
+    public async Task Should_Publish_On_Watcher_Event()
+    {
+        var settings = new OperatorSettings();
+
+        Action<WatchEventType, TestResource> onWatcherEvent = 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?>(
+                (_, onWatcherEventArg, _, _, _, _, _) => { onWatcherEvent = onWatcherEventArg; })
+            .Returns(Task.FromResult(CreateFakeWatcher()))
+            .Verifiable();
+
+        _metrics.Setup(c => c.Running).Returns(Mock.Of<IGauge>());
+        _metrics.Setup(c => c.WatchedEvents).Returns(Mock.Of<ICounter>());
+
+        using var resourceWatcher = new ResourceWatcher<TestResource>(
+            _client.Object,
+            new NullLogger<ResourceWatcher<TestResource>>(),
+            _metrics.Object,
+            settings);
+
+        var watchEvents = resourceWatcher.WatchEvents.Replay(1);
+        watchEvents.Connect();
+
+        await resourceWatcher.StartAsync();
+
+        var resource = new TestResource
+        {
+            Metadata = new()
+        };
+
+        onWatcherEvent(WatchEventType.Added, resource);
+
+        var watchEvent = await watchEvents.FirstAsync();
+
+        watchEvent.Type.Should().Be(WatchEventType.Added);
+        watchEvent.Resource.Should().BeEquivalentTo(resource);
+    }
+
+    [Fact]
+    public async Task Should_Restart_Watcher_On_Close()
+    {
+        var settings = new OperatorSettings();
+
+        Action? onClose = 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?>(
+                (_, _, _, onCloseArg, _, _, _) =>
+                {
+                    onClose = onCloseArg;
+                })
+            .Returns(Task.FromResult(CreateFakeWatcher()))
+            .Verifiable();
+
+        _metrics.Setup(c => c.Running).Returns(Mock.Of<IGauge>());
+        _metrics.Setup(c => c.WatcherClosed).Returns(Mock.Of<ICounter>());
+
+        using var resourceWatcher = new ResourceWatcher<TestResource>(_client.Object, new NullLogger<ResourceWatcher<TestResource>>(), _metrics.Object, settings);
+
+        await resourceWatcher.StartAsync();
+
+        onClose?.Invoke();
+
+        VerifyWatch(2);
+    }
+
+    private void VerifyWatch(int callTime)
+    {
+        _client.Verify(
+            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?>()), Times.Exactly(callTime));
+    }
+
+    private static Watcher<TestResource> CreateFakeWatcher()
+    {
+        return new Watcher<TestResource>(
+            () => Task.FromResult(new StreamReader(new MemoryStream())),
+            (_, __) => { },
+            _ => { });
     }
 }