ResourceWatcher reconnection fixes

[tomasfabian]

Removed disposal of _reconnectHandler in ResourceWatcher which causes ObjectDisposedException after leader reelections and the watcher stops observing the resource. #522

The situation happens after a consecutive call to StartAsync after StopAsync:

await _watcher.StartAsync();

await _watcher.StopAsync();

await _watcher.StartAsync();

The _reconnectHandler should be disposed only in the if statement of the Disposing method:

private void Disposing(bool fromStop)
{
        if (!fromStop)
        {
            _watchEvents.Dispose();
            _reconnectHandler.Dispose();
        }

        // _reconnectHandler.Dispose(); // this causes an error later on
        //...
}

Thx

[tomasfabian]

I think that the subscription from the ctor should be also moved to StartAsync since it is disposed in StopAsync:

        _reconnectSubscription =
            _reconnectHandler
                .Select(Observable.Timer)
                .Switch()
                .Retry()
                .Subscribe(async _ => await WatchResource(), error => _logger.LogError(error, $"There was an error while restarting the resource watcher {typeof(TEntity)}"));

The reconnect subscription should be either recreated on each StartAsync or not disposed during CloseAsync in case it should be active during the whole lifetime of the object.

Let me know please which one is the preferred solution and I can do it, too. Thx.

EDIT: For now I moved the disposal of _reconnectSubscription into the if statement (will not be executed for CloseAsync).

[buehler]

Hey @tomasfabian
I guess not dispose it during closing is fine. It is not that big of a memory overhead and as long as one solution (or rather "fix") is in place we should be fine.

Would you mind to add a test case for this scenario? It should be fairly easy to add a test where you call the sequence of methods as mentioned in the issue and this PR and see if the await method throws an error.

Thank you for the contribution!

[tomasfabian]

Hello @buehler,
ResourceWatcher<TEntity> is not prepared for unit testing. I'm not able to easily mock the ResourceWatcherMetrics<TEntity> ATM. It would require to wrap also the Prometheus dependencies such as Gauge.Child, because they have constructors with an internal modifier etc.

What do you think / suggest?

[tomasfabian]

I can introduce a new interface IResourceWatcherMetrics<TEntity> would you accept it as part of this PR, please?

public interface IResourceWatcherMetrics<TEntity> : IResourceWatcherMetrics
    where TEntity : IKubernetesObject<V1ObjectMeta>
{
}

public interface IResourceWatcherMetrics
{
    IGauge Running { get; }
    // Gauge.Child Running { get; }

    ICounter WatchedEvents { get; }
    // Counter.Child WatchedEvents { get; }

    ICounter WatcherExceptions { get; }
    // Counter.Child WatcherExceptions { get; }

    ICounter WatcherClosed { get; }
    // Counter.Child WatcherClosed { get; }
}

internal class ResourceWatcherMetrics<TEntity> : IResourceWatcherMetrics<TEntity>
    where TEntity : IKubernetesObject<V1ObjectMeta>
{
  // ...
}

And then in the OperatorBuilder I have to change the DI registration:

Services.AddSingleton(typeof(IResourceWatcherMetrics<>), typeof(ResourceWatcherMetrics<>));
// Services.AddSingleton(typeof(ResourceWatcherMetrics<>));

Thx

[buehler]

Ahh I see, yeah, you're right. No, then we'll leave it this way. If there is another issue with this matter, we can introduce an internal interface for this and add tests.