feat: Resource Watcher with Labels (#866)

This feature extends the ResourceWatcher functionality with support for
watching only resources which match the label restrictions.

---------

Co-authored-by: Razvan Grama <[email protected]>

Author: rgrama

src/KubeOps.Abstractions/Builder/IOperatorBuilder.cs

@@ -2,6 +2,7 @@
 using k8s.Models;
 
 using KubeOps.Abstractions.Controller;
+using KubeOps.Abstractions.Entities;
 using KubeOps.Abstractions.Finalizer;
 
 using Microsoft.Extensions.DependencyInjection;
@@ -29,6 +30,19 @@ IOperatorBuilder AddController<TImplementation, TEntity>()
         where TImplementation : class, IEntityController<TEntity>
         where TEntity : IKubernetesObject<V1ObjectMeta>;
 
+    /// <summary>
+    /// Add a controller implementation for a specific entity to the operator.
+    /// The metadata for the entity must be added as well.
+    /// </summary>
+    /// <typeparam name="TImplementation">Implementation type of the controller.</typeparam>
+    /// <typeparam name="TEntity">Entity type.</typeparam>
+    /// <typeparam name="TLabelSelector">Label Selector type.</typeparam>
+    /// <returns>The builder for chaining.</returns>
+    IOperatorBuilder AddController<TImplementation, TEntity, TLabelSelector>()
+        where TImplementation : class, IEntityController<TEntity>
+        where TEntity : IKubernetesObject<V1ObjectMeta>
+        where TLabelSelector : class, IEntityLabelSelector<TEntity>;
+
     /// <summary>
     /// Add a finalizer implementation for a specific entity.
     /// This adds the implementation as a transient service and registers

src/KubeOps.Abstractions/Entities/DefaultEntityLabelSelector{TEntity}.cs

@@ -0,0 +1,10 @@
+using k8s;
+using k8s.Models;
+
+namespace KubeOps.Abstractions.Entities;
+
+public class DefaultEntityLabelSelector<TEntity> : IEntityLabelSelector<TEntity>
+    where TEntity : IKubernetesObject<V1ObjectMeta>
+{
+    public ValueTask<string?> GetLabelSelectorAsync(CancellationToken cancellationToken) => ValueTask.FromResult<string?>(null);
+}

src/KubeOps.Abstractions/Entities/IEntityLabelSelector{TEntity}.cs

@@ -0,0 +1,15 @@
+using k8s;
+using k8s.Models;
+
+namespace KubeOps.Abstractions.Entities;
+
+// This is the same pattern used by Microsoft on ILogger<T>.
+// An alternative would be to use a KeyedSingleton when registering this however that's only valid from .NET 8 and above.
+// Other methods are far less elegant
+#pragma warning disable S2326
+public interface IEntityLabelSelector<TEntity>
+    where TEntity : IKubernetesObject<V1ObjectMeta>
+{
+    ValueTask<string?> GetLabelSelectorAsync(CancellationToken cancellationToken);
+}
+#pragma warning restore S2326

src/KubeOps.Operator/Builder/OperatorBuilder.cs

@@ -3,6 +3,7 @@
 
 using KubeOps.Abstractions.Builder;
 using KubeOps.Abstractions.Controller;
+using KubeOps.Abstractions.Entities;
 using KubeOps.Abstractions.Events;
 using KubeOps.Abstractions.Finalizer;
 using KubeOps.Abstractions.Queue;
@@ -54,6 +55,31 @@ public IOperatorBuilder AddController<TImplementation, TEntity>()
         return this;
     }
 
+    public IOperatorBuilder AddController<TImplementation, TEntity, TLabelSelector>()
+        where TImplementation : class, IEntityController<TEntity>
+        where TEntity : IKubernetesObject<V1ObjectMeta>
+        where TLabelSelector : class, IEntityLabelSelector<TEntity>
+    {
+        Services.AddHostedService<EntityRequeueBackgroundService<TEntity>>();
+        Services.TryAddScoped<IEntityController<TEntity>, TImplementation>();
+        Services.TryAddSingleton(new TimedEntityQueue<TEntity>());
+        Services.TryAddTransient<IEntityRequeueFactory, KubeOpsEntityRequeueFactory>();
+        Services.TryAddTransient<EntityRequeue<TEntity>>(services =>
+            services.GetRequiredService<IEntityRequeueFactory>().Create<TEntity>());
+        Services.TryAddSingleton<IEntityLabelSelector<TEntity>, TLabelSelector>();
+
+        if (_settings.EnableLeaderElection)
+        {
+            Services.AddHostedService<LeaderAwareResourceWatcher<TEntity>>();
+        }
+        else
+        {
+            Services.AddHostedService<ResourceWatcher<TEntity>>();
+        }
+
+        return this;
+    }
+
     public IOperatorBuilder AddFinalizer<TImplementation, TEntity>(string identifier)
         where TImplementation : class, IEntityFinalizer<TEntity>
         where TEntity : IKubernetesObject<V1ObjectMeta>
@@ -91,6 +117,8 @@ private void AddOperatorBase()
         Services.TryAddTransient<EventPublisher>(
             services => services.GetRequiredService<IEventPublisherFactory>().Create());
 
+        Services.AddSingleton(typeof(IEntityLabelSelector<>), typeof(DefaultEntityLabelSelector<>));
+
         if (_settings.EnableLeaderElection)
         {
             Services.AddLeaderElection();

src/KubeOps.Operator/Watcher/LeaderAwareResourceWatcher{TEntity}.cs

@@ -3,6 +3,7 @@
 using k8s.Models;
 
 using KubeOps.Abstractions.Builder;
+using KubeOps.Abstractions.Entities;
 using KubeOps.KubernetesClient;
 using KubeOps.Operator.Queue;
 
@@ -16,10 +17,11 @@ internal sealed class LeaderAwareResourceWatcher<TEntity>(
     IServiceProvider provider,
     TimedEntityQueue<TEntity> queue,
     OperatorSettings settings,
+    IEntityLabelSelector<TEntity> labelSelector,
     IKubernetesClient client,
     IHostApplicationLifetime hostApplicationLifetime,
     LeaderElector elector)
-    : ResourceWatcher<TEntity>(logger, provider, queue, settings, client)
+    : ResourceWatcher<TEntity>(logger, provider, queue, settings, labelSelector, client)
     where TEntity : IKubernetesObject<V1ObjectMeta>
 {
     private CancellationTokenSource _cts = new();

src/KubeOps.Operator/Watcher/ResourceWatcher{TEntity}.cs

@@ -8,6 +8,7 @@
 
 using KubeOps.Abstractions.Builder;
 using KubeOps.Abstractions.Controller;
+using KubeOps.Abstractions.Entities;
 using KubeOps.Abstractions.Finalizer;
 using KubeOps.KubernetesClient;
 using KubeOps.Operator.Queue;
@@ -23,6 +24,7 @@ internal class ResourceWatcher<TEntity>(
     IServiceProvider provider,
     TimedEntityQueue<TEntity> requeue,
     OperatorSettings settings,
+    IEntityLabelSelector<TEntity> labelSelector,
     IKubernetesClient client)
     : IHostedService, IAsyncDisposable, IDisposable
     where TEntity : IKubernetesObject<V1ObjectMeta>
@@ -139,6 +141,7 @@ private async Task WatchClientEventsAsync(CancellationToken stoppingToken)
                 await foreach ((WatchEventType type, TEntity entity) in client.WatchAsync<TEntity>(
                                    settings.Namespace,
                                    resourceVersion: currentVersion,
+                                   labelSelector: await labelSelector.GetLabelSelectorAsync(stoppingToken),
                                    allowWatchBookmarks: true,
                                    cancellationToken: stoppingToken))
                 {

test/KubeOps.Operator.Test/Builder/OperatorBuilder.Test.cs

@@ -2,16 +2,19 @@
 
 using KubeOps.Abstractions.Builder;
 using KubeOps.Abstractions.Controller;
+using KubeOps.Abstractions.Entities;
 using KubeOps.Abstractions.Events;
 using KubeOps.Abstractions.Finalizer;
 using KubeOps.Abstractions.Queue;
+using KubeOps.KubernetesClient.LabelSelectors;
 using KubeOps.Operator.Builder;
 using KubeOps.Operator.Finalizer;
 using KubeOps.Operator.Queue;
 using KubeOps.Operator.Test.TestEntities;
 using KubeOps.Operator.Watcher;
 
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Hosting;
 
 namespace KubeOps.Operator.Test.Builder;
@@ -29,6 +32,29 @@ public void Should_Add_Default_Resources()
         _builder.Services.Should().Contain(s =>
             s.ServiceType == typeof(EventPublisher) &&
             s.Lifetime == ServiceLifetime.Transient);
+        _builder.Services.Should().Contain(s =>
+            s.ServiceType == typeof(IEntityLabelSelector<>) &&
+            s.ImplementationType == typeof(DefaultEntityLabelSelector<>) &&
+            s.Lifetime == ServiceLifetime.Singleton);
+    }
+
+    [Fact]
+    public void Should_Use_Specific_EntityLabelSelector_Implementation()
+    {
+        // Arrange
+        var services = new ServiceCollection();
+
+        // Register the default and specific implementations
+        services.AddSingleton(typeof(IEntityLabelSelector<>), typeof(DefaultEntityLabelSelector<>));
+        services.TryAddSingleton<IEntityLabelSelector<V1OperatorIntegrationTestEntity>, TestLabelSelector>();
+
+        var serviceProvider = services.BuildServiceProvider();
+
+        // Act
+        var resolvedService = serviceProvider.GetRequiredService<IEntityLabelSelector<V1OperatorIntegrationTestEntity>>();
+
+        // Assert
+        Assert.IsType<TestLabelSelector>(resolvedService);
     }
 
     [Fact]
@@ -52,6 +78,31 @@ public void Should_Add_Controller_Resources()
             s.Lifetime == ServiceLifetime.Transient);
     }
 
+    [Fact]
+    public void Should_Add_Controller_Resources_With_Label_Selector()
+    {
+        _builder.AddController<TestController, V1OperatorIntegrationTestEntity, TestLabelSelector>();
+
+        _builder.Services.Should().Contain(s =>
+            s.ServiceType == typeof(IEntityController<V1OperatorIntegrationTestEntity>) &&
+            s.ImplementationType == typeof(TestController) &&
+            s.Lifetime == ServiceLifetime.Scoped);
+        _builder.Services.Should().Contain(s =>
+            s.ServiceType == typeof(IHostedService) &&
+            s.ImplementationType == typeof(ResourceWatcher<V1OperatorIntegrationTestEntity>) &&
+            s.Lifetime == ServiceLifetime.Singleton);
+        _builder.Services.Should().Contain(s =>
+            s.ServiceType == typeof(TimedEntityQueue<V1OperatorIntegrationTestEntity>) &&
+            s.Lifetime == ServiceLifetime.Singleton);
+        _builder.Services.Should().Contain(s =>
+            s.ServiceType == typeof(EntityRequeue<V1OperatorIntegrationTestEntity>) &&
+            s.Lifetime == ServiceLifetime.Transient);
+        _builder.Services.Should().Contain(s =>
+            s.ServiceType == typeof(IEntityLabelSelector<V1OperatorIntegrationTestEntity>) &&
+            s.ImplementationType == typeof(TestLabelSelector) &&
+            s.Lifetime == ServiceLifetime.Singleton);
+    }
+
     [Fact]
     public void Should_Add_Finalizer_Resources()
     {
@@ -105,4 +156,17 @@ private class TestFinalizer : IEntityFinalizer<V1OperatorIntegrationTestEntity>
         public Task FinalizeAsync(V1OperatorIntegrationTestEntity entity, CancellationToken cancellationToken) =>
             Task.CompletedTask;
     }
+
+    private class TestLabelSelector : IEntityLabelSelector<V1OperatorIntegrationTestEntity>
+    {
+        public ValueTask<string?> GetLabelSelectorAsync(CancellationToken cancellationToken)
+        {
+            var labelSelectors = new LabelSelector[]
+            {
+                new EqualsSelector("label", "value")
+            };
+
+            return ValueTask.FromResult<string?>(labelSelectors.ToExpression());
+        }
+    }
 }

test/KubeOps.Operator.Test/Watcher/ResourceWatcher{TEntity}.Test.cs

@@ -4,6 +4,7 @@
 using k8s.Models;
 
 using KubeOps.Abstractions.Builder;
+using KubeOps.Abstractions.Entities;
 using KubeOps.KubernetesClient;
 using KubeOps.Operator.Queue;
 using KubeOps.Operator.Watcher;
@@ -25,12 +26,13 @@ public async Task Restarting_Watcher_Should_Trigger_New_Watch()
         var timedEntityQueue = new TimedEntityQueue<V1Pod>();
         var operatorSettings = new OperatorSettings() { Namespace = "unit-test" };
         var kubernetesClient = Mock.Of<IKubernetesClient>();
+        var labelSelector = new DefaultEntityLabelSelector<V1Pod>();
 
         Mock.Get(kubernetesClient)
             .Setup(client => client.WatchAsync<V1Pod>("unit-test", null, null, true, It.IsAny<CancellationToken>()))
             .Returns<string?, string?, string?, bool?, CancellationToken>((_, _, _, _, cancellationToken) => WaitForCancellationAsync<(WatchEventType, V1Pod)>(cancellationToken));
 
-        var resourceWatcher = new ResourceWatcher<V1Pod>(logger, serviceProvider, timedEntityQueue, operatorSettings, kubernetesClient);
+        var resourceWatcher = new ResourceWatcher<V1Pod>(logger, serviceProvider, timedEntityQueue, operatorSettings, labelSelector, kubernetesClient);
 
         // Act.
         // Start and stop the watcher.