feat(operator): reworked entity requeue logic

BREAKING CHANGE: controllers do not have
return values anymore. To requeue an entity,
use the `EntityRequeue<_>` delegate. When
an entity is requeued, the reconcile loop
is called after the timeout. If - during this
timeout - the entity is modified or deleted,
the timeout will be cancelled.

Author: Christoph Bühler

examples/Operator/Controller/V1TestEntityController.cs

@@ -1,5 +1,6 @@
 using KubeOps.Abstractions.Controller;
 using KubeOps.Abstractions.Finalizer;
+using KubeOps.Abstractions.Queue;
 using KubeOps.Abstractions.Rbac;
 using KubeOps.KubernetesClient;
 
@@ -15,32 +16,26 @@ public class V1TestEntityController : IEntityController<V1TestEntity>
 {
     private readonly ILogger<V1TestEntityController> _logger;
     private readonly IKubernetesClient<V1TestEntity> _client;
-    private readonly EntityFinalizerAttacher<FinalizerOne, V1TestEntity> _finalizer1;
-    private readonly EntityFinalizerAttacher<FinalizerTwo, V1TestEntity> _finalizer2;
+    private readonly EntityRequeue<V1TestEntity> _requeue;
 
     public V1TestEntityController(
         ILogger<V1TestEntityController> logger,
         IKubernetesClient<V1TestEntity> client,
-        EntityFinalizerAttacher<FinalizerOne, V1TestEntity> finalizer1,
-        EntityFinalizerAttacher<FinalizerTwo, V1TestEntity> finalizer2)
+        EntityRequeue<V1TestEntity> requeue)
     {
         _logger = logger;
         _client = client;
-        _finalizer1 = finalizer1;
-        _finalizer2 = finalizer2;
+        _requeue = requeue;
     }
 
     public async Task ReconcileAsync(V1TestEntity entity)
     {
         _logger.LogInformation("Reconciling entity {Entity}.", entity);
 
-        entity = await _finalizer1(entity);
-        entity = await _finalizer2(entity);
-
-        entity.Status.Status = "Reconciling";
-        entity = await _client.UpdateStatusAsync(entity);
         entity.Status.Status = "Reconciled";
         await _client.UpdateStatusAsync(entity);
+
+        _requeue(entity, TimeSpan.FromSeconds(5));
     }
 
     public Task DeletedAsync(V1TestEntity entity)

examples/Operator/todos.txt

@@ -1,7 +1,6 @@
 todo:
 - events
 - leadership election
-- requeue
 - build targets
 - other CLI commands
 - error handling

src/KubeOps.Abstractions/Controller/IEntityController{TEntity}.cs

@@ -4,9 +4,35 @@
 namespace KubeOps.Abstractions.Controller;
 
 /// <summary>
-/// Generic entity controller interface.
+/// Generic entity controller. The controller manages the reconcile loop
+/// for a given entity type.
 /// </summary>
 /// <typeparam name="TEntity">The type of the Kubernetes entity.</typeparam>
+/// <example>
+/// Simple example controller that just logs the entity.
+/// <code>
+/// public class V1TestEntityController : IEntityController&lt;V1TestEntity&gt;
+/// {
+///     private readonly ILogger&lt;V1TestEntityController&gt; _logger;
+///
+///     public V1TestEntityController(
+///         ILogger&lt;V1TestEntityController&gt; logger)
+///     {
+///         _logger = logger;
+///     }
+///
+///     public async Task ReconcileAsync(V1TestEntity entity)
+///     {
+///         _logger.LogInformation("Reconciling entity {Entity}.", entity);
+///     }
+///
+///     public async Task DeletedAsync(V1TestEntity entity)
+///     {
+///         _logger.LogInformation("Deleting entity {Entity}.", entity);
+///     }
+/// }
+/// </code>
+/// </example>
 public interface IEntityController<in TEntity>
     where TEntity : IKubernetesObject<V1ObjectMeta>
 {

src/KubeOps.Abstractions/Finalizer/EntityFinalizerAttacher.cs

@@ -14,12 +14,18 @@ namespace KubeOps.Abstractions.Finalizer;
 /// modification and Kubernetes client interactions, since the resource version
 /// is updated each time the entity is modified.
 /// </para>
+/// <para>
+/// Note that the operator needs RBAC access to modify the list of
+/// finalizers on the entity.
+/// </para>
 /// </summary>
 /// <typeparam name="TImplementation">The type of the entity finalizer.</typeparam>
 /// <typeparam name="TEntity">The type of the Kubernetes entity.</typeparam>
 /// <param name="entity">The instance of the entity, that the finalizer is attached if needed.</param>
 /// <returns>A <see cref="Task"/> that resolves when the finalizer was attached.</returns>
 /// <example>
+/// Use the finalizer delegate to attach the "FinalizerOne" to the entity as soon
+/// as the entity gets reconciled.
 /// <code>
 /// [EntityRbac(typeof(V1TestEntity), Verbs = RbacVerb.All)]
 /// public class V1TestEntityController : IEntityController&lt;V1TestEntity&gt;

src/KubeOps.Abstractions/Finalizer/IEntityFinalizer{TEntity}.cs

@@ -15,5 +15,10 @@ public interface IEntityFinalizer<in TEntity>
     /// </summary>
     /// <param name="entity">The kubernetes entity that needs to be finalized.</param>
     /// <returns>A task that resolves when the operation is done.</returns>
+#if NETSTANDARD2_0
     Task FinalizeAsync(TEntity entity);
+#else
+    Task FinalizeAsync(TEntity entity) =>
+        Task.CompletedTask;
+#endif
 }

src/KubeOps.Abstractions/Queue/EntityRequeue.cs

@@ -0,0 +1,47 @@
+using k8s;
+using k8s.Models;
+
+namespace KubeOps.Abstractions.Queue;
+
+/// <summary>
+/// <para>Injectable delegate for requeueing entities.</para>
+/// <para>
+/// Use this delegate when you need to pro-actively reconcile an entity after a
+/// certain amount of time. This is useful if you want to check your entities
+/// periodically.
+/// </para>
+/// <para>
+/// After the timeout is reached, the entity is fetched
+/// from the API and passed to the controller for reconciliation.
+/// If the entity was deleted in the meantime, the controller will not be called.
+/// </para>
+/// <para>
+/// If the entity gets modified while the timeout is running, the timer
+/// is canceled and restarted, if another requeue is requested.
+/// </para>
+/// </summary>
+/// <typeparam name="TEntity">The type of the entity.</typeparam>
+/// <param name="entity">The instance of the entity that should be requeued.</param>
+/// <param name="requeueIn">The time to wait before another reconcile loop is fired.</param>
+/// <example>
+/// Use the requeue delegate to repeatedly reconcile an entity after 5 seconds.
+/// <code>
+/// [EntityRbac(typeof(V1TestEntity), Verbs = RbacVerb.All)]
+/// public class V1TestEntityController : IEntityController&lt;V1TestEntity&gt;
+/// {
+///     private readonly EntityRequeue&lt;V1TestEntity&gt; _requeue;
+///
+///     public V1TestEntityController(EntityRequeue&lt;V1TestEntity&gt; requeue)
+///     {
+///         _requeue = requeue;
+///     }
+///
+///     public async Task ReconcileAsync(V1TestEntity entity)
+///     {
+///         _requeue(entity, TimeSpan.FromSeconds(5));
+///     }
+/// }
+/// </code>
+/// </example>
+public delegate void EntityRequeue<TEntity>(TEntity entity, TimeSpan requeueIn)
+    where TEntity : IKubernetesObject<V1ObjectMeta>;

src/KubeOps.Abstractions/Rbac/EntityRbacAttribute.cs

@@ -5,7 +5,12 @@
 /// Attach this attribute to a controller with the type reference to
 /// a custom entity to define rbac needs for this given type(s).
 /// </summary>
-/// <example>[EntityRbac(typeof(V1TestEntity), Verbs = RbacVerb.All)].</example>
+/// <example>
+/// Allow the operator "ALL" access to the V1TestEntity.
+/// <code>
+/// [EntityRbac(typeof(V1TestEntity), Verbs = RbacVerb.All)]
+/// </code>
+/// </example>
 [AttributeUsage(AttributeTargets.Class, AllowMultiple = true)]
 public class EntityRbacAttribute : RbacAttribute
 {

src/KubeOps.Abstractions/Rbac/GenericRbacAttribute.cs

@@ -21,7 +21,7 @@ public class GenericRbacAttribute : RbacAttribute
 #if NETSTANDARD
     public string[] Groups { get; set; } = { };
 #else
-    public string[] Groups { get; init; } = { };
+    public string[] Groups { get; init; } = Array.Empty<string>();
 #endif
 
     /// <summary>
@@ -34,7 +34,7 @@ public class GenericRbacAttribute : RbacAttribute
 #if NETSTANDARD
     public string[] Resources { get; set; } = { };
 #else
-    public string[] Resources { get; init; } = { };
+    public string[] Resources { get; init; } = Array.Empty<string>();
 #endif
 
     /// <summary>
@@ -43,7 +43,7 @@ public class GenericRbacAttribute : RbacAttribute
 #if NETSTANDARD
     public string[] Urls { get; set; } = { };
 #else
-    public string[] Urls { get; init; } = { };
+    public string[] Urls { get; init; } = Array.Empty<string>();
 #endif
 
     /// <summary>

src/KubeOps.Operator/Builder/OperatorBuilder.cs

@@ -5,8 +5,10 @@
 using KubeOps.Abstractions.Controller;
 using KubeOps.Abstractions.Entities;
 using KubeOps.Abstractions.Finalizer;
+using KubeOps.Abstractions.Queue;
 using KubeOps.KubernetesClient;
 using KubeOps.Operator.Finalizer;
+using KubeOps.Operator.Queue;
 using KubeOps.Operator.Watcher;
 
 using Microsoft.Extensions.DependencyInjection;
@@ -36,6 +38,20 @@ public IOperatorBuilder AddController<TImplementation, TEntity>()
     {
         Services.AddScoped<IEntityController<TEntity>, TImplementation>();
         Services.AddHostedService<ResourceWatcher<TEntity>>();
+        Services.AddSingleton(new TimedEntityQueue<TEntity>());
+        Services.AddTransient<EntityRequeue<TEntity>>(services => (entity, timespan) =>
+        {
+            var logger = services.GetService<ILogger<EntityRequeue<TEntity>>>();
+            var queue = services.GetRequiredService<TimedEntityQueue<TEntity>>();
+
+            logger?.LogTrace(
+                """Requeue entity "{kind}/{name}" in {milliseconds}ms.""",
+                entity.Kind,
+                entity.Name(),
+                timespan.TotalMilliseconds);
+
+            queue.Enqueue(entity, timespan);
+        });
 
         return this;
     }

src/KubeOps.Operator/Queue/TimedEntityQueue.cs

@@ -0,0 +1,63 @@
+using System.Collections.Concurrent;
+
+using k8s;
+using k8s.Models;
+
+using Timer = System.Timers.Timer;
+
+namespace KubeOps.Operator.Queue;
+
+internal class TimedEntityQueue<TEntity>
+    where TEntity : IKubernetesObject<V1ObjectMeta>
+{
+    private readonly ConcurrentDictionary<string, (string Name, string? Namespace, Timer Timer)> _queue = new();
+
+    public event EventHandler<(string Name, string? Namespace)>? RequeueRequested;
+
+    internal int Count => _queue.Count;
+
+    public void Clear()
+    {
+        foreach (var (_, _, timer) in _queue.Values)
+        {
+            timer.Stop();
+        }
+
+        _queue.Clear();
+    }
+
+    public void Enqueue(TEntity entity, TimeSpan requeueIn)
+    {
+        var (_, _, timer) =
+            _queue.AddOrUpdate(
+                entity.Uid(),
+                (entity.Name(), entity.Namespace(), new Timer(requeueIn.TotalMilliseconds)),
+                (_, e) =>
+                {
+                    e.Timer.Stop();
+                    e.Timer.Dispose();
+                    return (e.Name, e.Namespace, new Timer(requeueIn.TotalMilliseconds));
+                });
+
+        timer.Elapsed += (_, _) =>
+        {
+            if (!_queue.TryRemove(entity.Metadata.Uid, out var e))
+            {
+                return;
+            }
+
+            e.Timer.Stop();
+            e.Timer.Dispose();
+            RequeueRequested?.Invoke(this, (e.Name, e.Namespace));
+        };
+        timer.Start();
+    }
+
+    public void RemoveIfQueued(TEntity entity)
+    {
+        if (_queue.TryRemove(entity.Uid(), out var entry))
+        {
+            entry.Timer.Stop();
+        }
+    }
+}