fixes(buehler#739): No retry is performed when the ResourceWatcher fail to watch a resource

Author: duke-bartholomew

.gitignore

@@ -33,3 +33,5 @@ coverage.info
 
 # Docs
 _site
+
+.idea/

src/KubeOps.KubernetesClient/ExceptionExtensions.cs

@@ -0,0 +1,28 @@
+using System.Text.RegularExpressions;
+
+namespace KubeOps.Operator.Watcher;
+
+/// <summary>
+/// Method extensions for the <see cref="Exception"/> class.
+/// </summary>
+public static class ExceptionExtensions
+{
+    /// <summary>
+    /// Walk through all collected Exceptions (base exception and all inner exceptions) LINQ style.
+    /// </summary>
+    public static IEnumerable<Exception> All(this Exception self)
+    {
+        if (self == null)
+        {
+            throw new ArgumentNullException(nameof(self));
+        }
+
+        var cause = self;
+        do
+        {
+            yield return cause;
+            cause = ReferenceEquals(cause, cause.InnerException) ? null : cause.InnerException;
+        }
+        while (cause != null && !ReferenceEquals(cause, self));
+    }
+}

src/KubeOps.KubernetesClient/IKubernetesClient.cs

@@ -87,7 +87,7 @@ Task<string> GetCurrentNamespaceAsync(
     /// <param name="labelSelector">A string, representing an optional label selector for filtering fetched objects.</param>
     /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
     /// <returns>A list of Kubernetes entities.</returns>
-    Task<IList<TEntity>> ListAsync<TEntity>(
+    Task<(string? Version, IList<TEntity> Items)> ListAsync<TEntity>(
         string? @namespace = null,
         string? labelSelector = null,
         CancellationToken cancellationToken = default)
@@ -106,7 +106,7 @@ Task<IList<TEntity>> ListAsync<TEntity>(
     /// </param>
     /// <param name="labelSelectors">A list of label-selectors to apply to the search.</param>
     /// <returns>A list of Kubernetes entities.</returns>
-    Task<IList<TEntity>> ListAsync<TEntity>(
+    Task<(string? Version, IList<TEntity> Items)> ListAsync<TEntity>(
         string? @namespace = null,
         params LabelSelector[] labelSelectors)
         where TEntity : IKubernetesObject<V1ObjectMeta>
@@ -116,13 +116,13 @@ Task<IList<TEntity>> ListAsync<TEntity>(
     }
 
     /// <inheritdoc cref="ListAsync{TEntity}(string?,string?,CancellationToken)"/>
-    IList<TEntity> List<TEntity>(
+    (string? Version, IList<TEntity> Items) List<TEntity>(
         string? @namespace = null,
         string? labelSelector = null)
         where TEntity : IKubernetesObject<V1ObjectMeta>;
 
     /// <inheritdoc cref="ListAsync{TEntity}(string?,LabelSelector[])"/>
-    IList<TEntity> List<TEntity>(
+    (string? Version, IList<TEntity> Items) List<TEntity>(
         string? @namespace = null,
         params LabelSelector[] labelSelectors)
         where TEntity : IKubernetesObject<V1ObjectMeta>
@@ -489,4 +489,12 @@ Watcher<TEntity> Watch<TEntity>(
         string? labelSelector = null,
         CancellationToken cancellationToken = default)
         where TEntity : IKubernetesObject<V1ObjectMeta>;
+
+    public Task WatchSafeAsync<TEntity>(
+        Func<WatchEventType, TEntity?, CancellationToken, Task> eventTask,
+        string? @namespace = null,
+        string? resourceVersion = null,
+        string? labelSelector = null,
+        CancellationToken cancellationToken = default)
+        where TEntity : IKubernetesObject<V1ObjectMeta>;
 }

src/KubeOps.KubernetesClient/KubernetesClient.cs

@@ -1,13 +1,16 @@
 using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Net;
+using System.Net.Sockets;
 using System.Runtime.CompilerServices;
+using System.Text.RegularExpressions;
 
 using k8s;
 using k8s.Autorest;
 using k8s.Models;
 
 using KubeOps.Abstractions.Entities;
+using KubeOps.Operator.Watcher;
 using KubeOps.Transpiler;
 
 namespace KubeOps.KubernetesClient;
@@ -19,6 +22,22 @@ public class KubernetesClient : IKubernetesClient
     private const string DefaultNamespace = "default";
 
     private static readonly ConcurrentDictionary<Type, EntityMetadata> MetadataCache = new();
+    private static List<int?> ResourceFailureCodes = ((int?[])[(int)HttpStatusCode.GatewayTimeout, (int)HttpStatusCode.Gone]).ToList();
+
+    /// <summary>
+    /// HACK to ge the last applicable resourceVersion from the exception.
+    /// </summary>
+    /// <example>
+    ///  "too old resource version: 512122628 (544688086)".
+    /// </example>
+    private static string? ResourceVersionFromException(Exception? ex)
+    {
+        if (ex?.Message is null) return null;
+
+        var pattern = @"^\s*too old resource version.*\(([a-zA-Z0-9_-]+)\)\s*$";
+        var match = Regex.Match(ex.Message, pattern);
+        return (match.Groups.Count > 1) ? match.Groups[1].Value : null;
+    }
 
     private readonly KubernetesClientConfiguration _clientConfig;
     private readonly IKubernetes _client;
@@ -29,18 +48,14 @@ public class KubernetesClient : IKubernetesClient
     /// The client will use the default configuration.
     /// </summary>
     public KubernetesClient()
-        : this(KubernetesClientConfiguration.BuildDefaultConfig())
-    {
-    }
+        : this(KubernetesClientConfiguration.BuildDefaultConfig()) { }
 
     /// <summary>
     /// Create a new Kubernetes client for the given entity with a custom client configuration.
     /// </summary>
     /// <param name="clientConfig">The config for the underlying Kubernetes client.</param>
     public KubernetesClient(KubernetesClientConfiguration clientConfig)
-        : this(clientConfig, new Kubernetes(clientConfig))
-    {
-    }
+        : this(clientConfig, new Kubernetes(clientConfig)) { }
 
     /// <summary>
     /// Create a new Kubernetes client for the given entity with a custom client configuration and client.
@@ -180,7 +195,7 @@ public string GetCurrentNamespace(string downwardApiEnvName = "POD_NAMESPACE")
     }
 
     /// <inheritdoc />
-    public async Task<IList<TEntity>> ListAsync<TEntity>(
+    public async Task<(string? Version, IList<TEntity> Items)> ListAsync<TEntity>(
         string? @namespace = null,
         string? labelSelector = null,
         CancellationToken cancellationToken = default)
@@ -189,7 +204,7 @@ public async Task<IList<TEntity>> ListAsync<TEntity>(
         ThrowIfDisposed();
 
         var metadata = GetMetadata<TEntity>();
-        return (@namespace switch
+        var result = @namespace switch
         {
             null => await _client.CustomObjects.ListClusterCustomObjectAsync<EntityList<TEntity>>(
                 metadata.Group ?? string.Empty,
@@ -204,17 +219,20 @@ public async Task<IList<TEntity>> ListAsync<TEntity>(
                 metadata.PluralName,
                 labelSelector: labelSelector,
                 cancellationToken: cancellationToken),
-        }).Items;
+        };
+
+        return (result.Metadata.ResourceVersion, result.Items);
     }
 
     /// <inheritdoc />
-    public IList<TEntity> List<TEntity>(string? @namespace = null, string? labelSelector = null)
+    public (string? Version, IList<TEntity> Items) List<TEntity>(string? @namespace = null,
+        string? labelSelector = null)
         where TEntity : IKubernetesObject<V1ObjectMeta>
     {
         ThrowIfDisposed();
 
         var metadata = GetMetadata<TEntity>();
-        return (@namespace switch
+        var result = @namespace switch
         {
             null => _client.CustomObjects.ListClusterCustomObject<EntityList<TEntity>>(
                 metadata.Group ?? string.Empty,
@@ -227,7 +245,9 @@ public IList<TEntity> List<TEntity>(string? @namespace = null, string? labelSele
                 @namespace,
                 metadata.PluralName,
                 labelSelector: labelSelector),
-        }).Items;
+        };
+
+        return (result.Metadata.ResourceVersion, result.Items);
     }
 
     /// <inheritdoc />
@@ -339,6 +359,51 @@ public async Task DeleteAsync<TEntity>(
         }
     }
 
+    /// <summary>
+    /// 
+    /// </summary>
+    /// <param name="onEvent"></param>
+    /// <param name="namespace"></param>
+    /// <param name="resourceVersion"></param>
+    /// <param name="labelSelector"></param>
+    /// <param name="cancellationToken"></param>
+    /// <typeparam name="TEntity"></typeparam>
+    public async Task WatchSafeAsync<TEntity>(
+        Func<WatchEventType, TEntity?, CancellationToken, Task> eventTask,
+        string? @namespace = null,
+        string? resourceVersion = null,
+        string? labelSelector = null,
+        CancellationToken cancellationToken = default)
+        where TEntity : IKubernetesObject<V1ObjectMeta>
+    {
+        var currentVersion = resourceVersion;
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            try
+            {
+                await foreach (var (typ, e) in WatchAsync<TEntity>(@namespace, currentVersion, labelSelector, cancellationToken))
+                {
+                    currentVersion = e.ResourceVersion();
+                    await eventTask(typ, e, cancellationToken);
+                }
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                // OK, end the watch
+            }
+            catch (KubernetesException cause) when (ResourceFailureCodes.Contains(cause.Status.Code))
+            {
+                currentVersion = ResourceVersionFromException(cause);
+                if (currentVersion == null) break; // bail out of watch
+            }
+            catch (Exception cause) when (cause.All().Any(e => e.Message.Contains("server reset the stream")
+                                                               || e is SocketException { ErrorCode: 104 }))
+            {
+                await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
+            }
+        }
+    }
+
     /// <inheritdoc />
     public Watcher<TEntity> Watch<TEntity>(
         Action<WatchEventType, TEntity> onEvent,

src/KubeOps.Operator/Watcher/BackoffPolicy.cs

@@ -0,0 +1,45 @@
+namespace KubeOps.Operator.Watcher;
+
+/// <summary>
+/// Simple exponential backoff logic.
+/// </summary>
+public class BackoffPolicy(CancellationToken stoppingToken, Func<int, TimeSpan> policy)
+{
+    private int _retries = 0;
+
+    /// <summary>
+    /// Default exponential backoff algorithm
+    /// </summary>
+    public static Func<int, TimeSpan> ExponentialWithJitter(int maxExp = 5, int jitterMillis = 1000)
+        => retries => TimeSpan.FromSeconds(Math.Pow(2, Math.Clamp(retries, 0, maxExp)))
+            .Add(TimeSpan.FromMilliseconds(new Random().Next(0, jitterMillis)));
+
+    /// <summary>
+    /// Clear all counters.
+    /// </summary>
+    public void Clear()
+    {
+        _retries = 0;
+    }
+
+    /// <summary>
+    /// Adds a delay.
+    /// </summary>
+    /// <param name="ex"><see cref="Exception"/>.</param>
+    /// <returns><see cref="Task"/>.</returns>
+    public async Task WaitOnException(Exception ex)
+    {
+        try
+        {
+            _retries++;
+            await Task.Delay(WaitTime(), stoppingToken);
+        }
+        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+        {
+            // Do nothing
+        }
+    }
+
+    private TimeSpan WaitTime()
+        => policy(_retries);
+}