fix(watcher): reconnect after server or client timeout (#780)

This fixes #739.
This closes #771.

Allows the resource watcher to retry the connection until the
cancellation token requests a stop. The watcher caches the
received entities and checks for their keys in a concurrent
dictionary. Recurring "added" events after reconnection
should not trigger a reconciliation.

Author: buehler

examples/Operator/Controller/V1TestEntityController.cs

@@ -10,18 +10,13 @@
 namespace Operator.Controller;
 
 [EntityRbac(typeof(V1TestEntity), Verbs = RbacVerb.All)]
-public class V1TestEntityController(ILogger<V1TestEntityController> logger,
-        EntityRequeue<V1TestEntity> requeue,
-        EventPublisher eventPublisher)
+public class V1TestEntityController(ILogger<V1TestEntityController> logger)
     : IEntityController<V1TestEntity>
 {
-    public async Task ReconcileAsync(V1TestEntity entity, CancellationToken cancellationToken)
+    public Task ReconcileAsync(V1TestEntity entity, CancellationToken cancellationToken)
     {
         logger.LogInformation("Reconciling entity {Entity}.", entity);
-
-        await eventPublisher(entity, "RECONCILED", "Entity was reconciled.");
-
-        requeue(entity, TimeSpan.FromSeconds(5));
+        return Task.CompletedTask;
     }
 
     public Task DeletedAsync(V1TestEntity entity, CancellationToken cancellationToken)

src/KubeOps.KubernetesClient/IKubernetesClient.cs

@@ -40,7 +40,7 @@ public interface IKubernetesClient : IDisposable
     /// <item>
     /// <description>
     ///     The fallback secret file if running on the cluster
-    ///     (/var/run/secrets/Kubernetes.io/serviceaccount/namespace)
+    ///     (<c>/var/run/secrets/Kubernetes.io/serviceaccount/namespace</c>)
     /// </description>
     /// </item>
     /// <item>
@@ -129,7 +129,7 @@ IList<TEntity> List<TEntity>(
         => List<TEntity>(@namespace, labelSelectors.ToExpression());
 
     /// <summary>
-    /// Create or Update a entity. This first fetches the entity from the Kubernetes API
+    /// Create or Update an entity. This first fetches the entity from the Kubernetes API
     /// and if it does exist, updates the entity. Otherwise, the entity is created.
     /// </summary>
     /// <typeparam name="TEntity">The type of the Kubernetes entity.</typeparam>
@@ -252,7 +252,7 @@ IEnumerable<TEntity> Create<TEntity>(params TEntity[] entities)
         => entities.Select(Create);
 
     /// <summary>
-    /// Update the given entity on the Kubernetes API.
+    /// Update (replace) the given entity on the Kubernetes API.
     /// </summary>
     /// <typeparam name="TEntity">The type of the Kubernetes entity.</typeparam>
     /// <param name="entity">The entity instance.</param>
@@ -262,10 +262,10 @@ Task<TEntity> UpdateAsync<TEntity>(TEntity entity, CancellationToken cancellatio
         where TEntity : IKubernetesObject<V1ObjectMeta>;
 
     /// <summary>
-    /// Update a list of entities on the Kubernetes API.
+    /// Update (replace) a list of entities on the Kubernetes API.
     /// </summary>
     /// <typeparam name="TEntity">The type of the Kubernetes entity.</typeparam>
-    /// <param name="entities">An enumerable of entities.</param>
+    /// <param name="entities">Enumerable of entities.</param>
     /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
     /// <returns>The updated instances of the entities.</returns>
     async Task<IEnumerable<TEntity>> UpdateAsync<TEntity>(
@@ -275,10 +275,10 @@ async Task<IEnumerable<TEntity>> UpdateAsync<TEntity>(
         => await Task.WhenAll(entities.Select(entity => UpdateAsync(entity, cancellationToken)));
 
     /// <summary>
-    /// Update a list of entities on the Kubernetes API.
+    /// Update (replace) a list of entities on the Kubernetes API.
     /// </summary>
     /// <typeparam name="TEntity">The type of the Kubernetes entity.</typeparam>
-    /// <param name="entities">An enumerable of entities.</param>
+    /// <param name="entities">Enumerable of entities.</param>
     /// <remarks>
     /// This is invoking the API without any cancellation support. In order to pass a <see cref="CancellationToken"/>,
     /// you need to use the <see cref="UpdateAsync{TEntity}(IEnumerable{TEntity},CancellationToken)"/> overload.
@@ -396,9 +396,9 @@ void Delete<TEntity>(string name, string? @namespace = null)
         => DeleteAsync<TEntity>(name, @namespace).GetAwaiter().GetResult();
 
     /// <summary>
-    /// Create a entity watcher on the Kubernetes API.
+    /// Create an entity watcher on the Kubernetes API.
     /// The entity watcher fires events for entity-events on
-    /// Kubernetes (events: <see cref="WatchEventType"/>.
+    /// Kubernetes (events: <see cref="WatchEventType"/>).
     /// </summary>
     /// <typeparam name="TEntity">The type of the Kubernetes entity.</typeparam>
     /// <param name="onEvent">Action that is called when an event occurs.</param>
@@ -409,19 +409,24 @@ void Delete<TEntity>(string name, string? @namespace = null)
     /// If the namespace is omitted, all entities on the cluster are watched.
     /// </param>
     /// <param name="timeout">The timeout which the watcher has (after this timeout, the server will close the connection).</param>
+    /// <param name="allowWatchBookmarks">
+    /// Parameter to tell the server to send BOOKMARK events. However, if the server has no implementation or
+    /// configuration for bookmarks, this flag is ignored.
+    /// </param>
     /// <param name="resourceVersion">
     /// When specified with a watch call, shows changes that occur after that particular version of a resource.
     /// Defaults to changes from the beginning of history.
     /// </param>
     /// <param name="cancellationToken">Cancellation-Token.</param>
     /// <param name="labelSelectors">A list of label-selectors to apply to the search.</param>
-    /// <returns>A entity watcher for the given entity.</returns>
+    /// <returns>An entity watcher for the given entity.</returns>
     Watcher<TEntity> Watch<TEntity>(
         Action<WatchEventType, TEntity> onEvent,
         Action<Exception>? onError = null,
         Action? onClose = null,
         string? @namespace = null,
         TimeSpan? timeout = null,
+        bool? allowWatchBookmarks = null,
         string? resourceVersion = null,
         CancellationToken cancellationToken = default,
         params LabelSelector[] labelSelectors)
@@ -432,14 +437,15 @@ Watcher<TEntity> Watch<TEntity>(
             onClose,
             @namespace,
             timeout,
+            allowWatchBookmarks,
             resourceVersion,
             labelSelectors.ToExpression(),
             cancellationToken);
 
     /// <summary>
-    /// Create a entity watcher on the Kubernetes API.
+    /// Create an entity watcher on the Kubernetes API.
     /// The entity watcher fires events for entity-events on
-    /// Kubernetes (events: <see cref="WatchEventType"/>.
+    /// Kubernetes (events: <see cref="WatchEventType"/>).
     /// </summary>
     /// <typeparam name="TEntity">The type of the Kubernetes entity.</typeparam>
     /// <param name="onEvent">Action that is called when an event occurs.</param>
@@ -450,19 +456,24 @@ Watcher<TEntity> Watch<TEntity>(
     /// If the namespace is omitted, all entities on the cluster are watched.
     /// </param>
     /// <param name="timeout">The timeout which the watcher has (after this timeout, the server will close the connection).</param>
+    /// <param name="allowWatchBookmarks">
+    /// Parameter to tell the server to send BOOKMARK events. However, if the server has no implementation or
+    /// configuration for bookmarks, this flag is ignored.
+    /// </param>
     /// <param name="resourceVersion">
     /// When specified with a watch call, shows changes that occur after that particular version of a resource.
     /// Defaults to changes from the beginning of history.
     /// </param>
     /// <param name="labelSelector">A string, representing an optional label selector for filtering watched objects.</param>
     /// <param name="cancellationToken">Cancellation-Token.</param>
-    /// <returns>A entity watcher for the given entity.</returns>
+    /// <returns>An entity watcher for the given entity.</returns>
     Watcher<TEntity> Watch<TEntity>(
         Action<WatchEventType, TEntity> onEvent,
         Action<Exception>? onError = null,
         Action? onClose = null,
         string? @namespace = null,
         TimeSpan? timeout = null,
+        bool? allowWatchBookmarks = null,
         string? resourceVersion = null,
         string? labelSelector = null,
         CancellationToken cancellationToken = default)
@@ -480,13 +491,18 @@ Watcher<TEntity> Watch<TEntity>(
     /// Defaults to changes from the beginning of history.
     /// </param>
     /// <param name="labelSelector">A string, representing an optional label selector for filtering watched objects.</param>
+    /// <param name="allowWatchBookmarks">
+    /// Parameter to tell the server to send BOOKMARK events. However, if the server has no implementation or
+    /// configuration for bookmarks, this flag is ignored.
+    /// </param>
     /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
     /// <typeparam name="TEntity">The type of the Kubernetes entity.</typeparam>
     /// <returns>An asynchronous enumerable that finishes once <paramref name="cancellationToken"/> is cancelled.</returns>
     IAsyncEnumerable<(WatchEventType Type, TEntity Entity)> WatchAsync<TEntity>(
         string? @namespace = null,
         string? resourceVersion = null,
         string? labelSelector = null,
+        bool? allowWatchBookmarks = null,
         CancellationToken cancellationToken = default)
         where TEntity : IKubernetesObject<V1ObjectMeta>;
 }