feat: Resource Event De-duplication

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/Operator/Builder/OperatorBuilder.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/Caching/IResourceCache.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/ResourceCache{TEntity}.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";