feat: Resource Event De-duplication (#394)

When an event for a given resource is added to the queue, all enqueued events
(those currently waiting for their "Delay" to elapse) for that resource are
unsubscribed from and are thus ignored.

For example, if I were to queue a "NotModified" event for a resource with
a delay of 10 seconds, then 5 seconds later queue a "Deleted" event, the
"Deleted" event would process immediately, and the "NotModified" event would
be ignored and not processed.

This has no effect on events which are currently in the "UpdateResourceData"
or "HandleEvent" steps, as those are already past the de-duplication step.

Author: erin-allison

src/KubeOps.Testing/KubernetesOperatorFactory.cs

@@ -4,7 +4,6 @@
 using DotnetKubernetesClient;
 using k8s;
 using k8s.Models;
-using KubeOps.Operator.Builder;
 using KubeOps.Operator.Controller;
 using KubeOps.Operator.Kubernetes;
 using KubeOps.Operator.Leadership;
@@ -55,20 +54,24 @@ public void Run()
             var server = Server;
         }
 
-        public Task EnqueueEvent<TEntity>(ResourceEventType type, TEntity resource)
+        public Task EnqueueEvent<TEntity>(ResourceEventType type, TEntity resource, int attempt = 0, TimeSpan? delay = null)
             where TEntity : class, IKubernetesObject<V1ObjectMeta>
         {
-            var controller = GetMockController<TEntity>();
+            var queue = Services.GetService<IEventQueue<TEntity>>();
 
-            return controller?.EnqueueEvent(type, resource) ?? Task.CompletedTask;
+            queue?.EnqueueLocal(new ResourceEvent<TEntity>(type, resource, attempt, delay));
+
+            return Task.CompletedTask;
         }
 
         public Task EnqueueFinalization<TEntity>(TEntity resource)
             where TEntity : class, IKubernetesObject<V1ObjectMeta>
         {
-            var controller = GetMockController<TEntity>();
+            var queue = Services.GetService<IEventQueue<TEntity>>();
+
+            queue?.EnqueueLocal(new ResourceEvent<TEntity>(ResourceEventType.Finalizing, resource));
 
-            return controller?.EnqueueFinalization(resource) ?? Task.CompletedTask;
+            return Task.CompletedTask;
         }
 
         /// <summary>
@@ -90,37 +93,25 @@ protected override IHostBuilder CreateHostBuilder() =>
         protected override void ConfigureWebHost(IWebHostBuilder builder)
         {
             base.ConfigureWebHost(builder);
-            if (_solutionRelativeContentRoot != null)
-            {
-                builder.UseSolutionRelativeContentRoot(_solutionRelativeContentRoot);
-            }
-
             builder.ConfigureTestServices(
                 services =>
                 {
                     var elector = services.First(
                         d => d.ServiceType == typeof(IHostedService) && d.ImplementationType == typeof(LeaderElector));
                     services.Remove(elector);
 
-                    services.RemoveAll(typeof(IKubernetesClient));
+                    services.RemoveAll<IKubernetesClient>();
                     services.AddSingleton<IKubernetesClient, MockKubernetesClient>();
 
-                    services.RemoveAll<Func<IComponentRegistrar.ControllerRegistration, IManagedResourceController>>();
-                    services.AddSingleton(
-                        s => (Func<IComponentRegistrar.ControllerRegistration, IManagedResourceController>)(r =>
-                            (IManagedResourceController)ActivatorUtilities.CreateInstance(
-                                s,
-                                typeof(MockManagedResourceController<>).MakeGenericType(r.EntityType),
-                                r)));
+                    services.RemoveAll(typeof(IEventQueue<>));
+                    services.AddSingleton(typeof(IEventQueue<>), typeof(MockEventQueue<>));
                 });
+            if (_solutionRelativeContentRoot != null)
+            {
+                builder.UseSolutionRelativeContentRoot(_solutionRelativeContentRoot);
+            }
+
             builder.ConfigureLogging(logging => logging.ClearProviders());
         }
-
-        private MockManagedResourceController<TEntity>? GetMockController<TEntity>()
-            where TEntity : class, IKubernetesObject<V1ObjectMeta> =>
-            Services.GetRequiredService<IControllerInstanceBuilder>()
-                .BuildControllers<TEntity>()
-                .OfType<MockManagedResourceController<TEntity>>()
-                .FirstOrDefault();
     }
 }

src/KubeOps.Testing/MockEventQueue.cs

@@ -0,0 +1,36 @@
+using System;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using System.Threading.Tasks;
+using k8s;
+using k8s.Models;
+using KubeOps.Operator.Controller;
+
+namespace KubeOps.Testing;
+
+internal class MockEventQueue<TEntity> : IEventQueue<TEntity>
+    where TEntity : class, IKubernetesObject<V1ObjectMeta>
+{
+    private readonly Subject<ResourceEvent<TEntity>> _localEvents;
+
+    public MockEventQueue()
+    {
+        _localEvents = new Subject<ResourceEvent<TEntity>>();
+
+        Events = _localEvents;
+    }
+
+    public IObservable<ResourceEvent<TEntity>> Events { get; }
+
+    public Task StartAsync(Action<ResourceEvent<TEntity>> onWatcherEvent)
+    {
+        return Task.CompletedTask;
+    }
+
+    public Task StopAsync()
+    {
+        return Task.CompletedTask;
+    }
+
+    public void EnqueueLocal(ResourceEvent<TEntity> resourceEvent) => _localEvents.OnNext(resourceEvent);
+}

src/KubeOps.Testing/MockManagedResourceController{TResource}.cs

@@ -199,11 +199,12 @@ internal IOperatorBuilder AddOperatorBase(OperatorSettings settings)
 
         Services.AddTransient<EntitySerializer>();
 
-        Services.AddTransient<IKubernetesClient, KubernetesClient>();
-        Services.AddTransient<IEventManager, EventManager>();
+        Services.AddScoped<IKubernetesClient, KubernetesClient>();
+        Services.AddScoped<IEventManager, EventManager>();
 
-        Services.AddTransient(typeof(ResourceCache<>));
-        Services.AddTransient(typeof(ResourceWatcher<>));
+        Services.AddScoped(typeof(ResourceCache<>));
+        Services.AddScoped(typeof(ResourceWatcher<>));
+        Services.AddScoped(typeof(IEventQueue<>), typeof(EventQueue<>));
 
         // Support all the metrics
         Services.AddSingleton(typeof(ResourceWatcherMetrics<>));

src/KubeOps/Operator/Builder/OperatorBuilder.cs

@@ -0,0 +1,19 @@
+using System.Collections.Generic;
+using k8s;
+using k8s.Models;
+
+namespace KubeOps.Operator.Caching;
+
+internal interface IResourceCache<TEntity>
+    where TEntity : IKubernetesObject<V1ObjectMeta>
+{
+    TEntity Get(string id);
+
+    TEntity Upsert(TEntity resource, out CacheComparisonResult result);
+
+    void Fill(IEnumerable<TEntity> resources);
+
+    void Remove(TEntity resource);
+
+    void Clear();
+}

src/KubeOps/Operator/Caching/IResourceCache.cs

@@ -10,7 +10,7 @@
 
 namespace KubeOps.Operator.Caching;
 
-internal class ResourceCache<TEntity>
+internal class ResourceCache<TEntity> : IResourceCache<TEntity>
     where TEntity : IKubernetesObject<V1ObjectMeta>
 {
     private const string Finalizers = "Metadata.Finalizers";