feat: add FusionCache for resource watcher to enable L1 and L2 cachin…

…g (hybrid cache)

- Integrated FusionCache for robust caching in resource watchers.
- Enhanced default configuration with extensible settings in `OperatorSettings`.
- Improved concurrency handling using `SemaphoreSlim` for entity events.
- Updated tests and dependencies to reflect caching changes.

src/KubeOps.Abstractions/Builder/OperatorSettings.cs

@@ -1,11 +1,13 @@
 using System.Text.RegularExpressions;
 
+using ZiggyCreatures.Caching.Fusion;
+
 namespace KubeOps.Abstractions.Builder;
 
 /// <summary>
 /// Operator settings.
 /// </summary>
-public sealed class OperatorSettings
+public sealed partial class OperatorSettings
 {
     private const string DefaultOperatorName = "KubernetesOperator";
     private const string NonCharReplacement = "-";
@@ -15,7 +17,7 @@ public sealed class OperatorSettings
     /// Defaults to "kubernetesoperator" when not set.
     /// </summary>
     public string Name { get; set; } =
-        new Regex(@"(\W|_)", RegexOptions.CultureInvariant).Replace(
+        OperatorNameRegex().Replace(
                 DefaultOperatorName,
                 NonCharReplacement)
             .ToLowerInvariant();
@@ -59,4 +61,14 @@ public sealed class OperatorSettings
     /// The wait timeout if the lease cannot be acquired.
     /// </summary>
     public TimeSpan LeaderElectionRetryPeriod { get; set; } = TimeSpan.FromSeconds(2);
+
+    /// <summary>
+    /// Allows configuration of the FusionCache settings for resource watcher entity caching.
+    /// This property is optional and can be used to customize caching behavior for resource watcher entities.
+    /// If not set, a default cache configuration is applied.
+    /// </summary>
+    public Action<FusionCacheOptions>? ConfigureResourceWatcherEntityCache { get; set; }
+
+    [GeneratedRegex(@"(\W|_)", RegexOptions.CultureInvariant)]
+    private static partial Regex OperatorNameRegex();
 }

src/KubeOps.Abstractions/KubeOps.Abstractions.csproj

@@ -16,6 +16,7 @@
     <ItemGroup>
         <PackageReference Include="KubernetesClient" Version="16.0.7" />
         <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.6"/>
+        <PackageReference Include="ZiggyCreatures.FusionCache" Version="2.3.0" />
     </ItemGroup>
 
 </Project>

src/KubeOps.Operator/Builder/OperatorBuilder.cs

@@ -21,6 +21,8 @@
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 
+using ZiggyCreatures.Caching.Fusion;
+
 namespace KubeOps.Operator.Builder;
 
 internal sealed class OperatorBuilder : IOperatorBuilder
@@ -36,6 +38,13 @@ public OperatorBuilder(IServiceCollection services, OperatorSettings settings)
 
     public IServiceCollection Services { get; }
 
+    private static Action<FusionCacheOptions> DefaultCacheConfiguration
+        => options =>
+        {
+            options.DefaultEntryOptions
+                .SetDuration(Timeout.InfiniteTimeSpan);
+        };
+
     public IOperatorBuilder AddController<TImplementation, TEntity>()
         where TImplementation : class, IEntityController<TEntity>
         where TEntity : IKubernetesObject<V1ObjectMeta>
@@ -111,6 +120,12 @@ private void AddOperatorBase()
         Services.AddSingleton(_settings);
         Services.AddSingleton(new ActivitySource(_settings.Name));
 
+        // add and configure resource watcher entity cache
+        Services
+            .AddFusionCache(CacheConstants.CacheNames.ResourceWatcher)
+            .WithOptions(
+                options => (_settings.ConfigureResourceWatcherEntityCache ?? DefaultCacheConfiguration).Invoke(options));
+
         // Add the default configuration and the client separately. This allows external users to override either
         // just the config (e.g. for integration tests) or to replace the whole client, e.g. with a mock.
         // We also add the k8s.IKubernetes as a singleton service, in order to allow to access internal services

src/KubeOps.Operator/Constants/CacheConstants.cs

@@ -0,0 +1,19 @@
+namespace KubeOps.Operator.Constants;
+
+/// <summary>
+/// Provides constant values used for caching purposes within the operator.
+/// </summary>
+public static class CacheConstants
+{
+    /// <summary>
+    /// Contains constant values representing names used within the operator's caching mechanisms.
+    /// </summary>
+    public static class CacheNames
+    {
+        /// <summary>
+        /// Represents a constant string used as a name for the resource watcher
+        /// in the operator's caching mechanisms.
+        /// </summary>
+        public const string ResourceWatcher = "ResourceWatcher";
+    }
+}

src/KubeOps.Operator/KubeOps.Operator.csproj

@@ -17,6 +17,7 @@
 
     <ItemGroup>
         <PackageReference Include="Microsoft.Extensions.Hosting" Version="9.0.6"/>
+        <PackageReference Include="ZiggyCreatures.FusionCache" Version="2.3.0" />
     </ItemGroup>
 
     <ItemGroup>

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

@@ -12,6 +12,8 @@
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 
+using ZiggyCreatures.Caching.Fusion;
+
 namespace KubeOps.Operator.Watcher;
 
 internal sealed class LeaderAwareResourceWatcher<TEntity>(
@@ -21,6 +23,7 @@ internal sealed class LeaderAwareResourceWatcher<TEntity>(
     TimedEntityQueue<TEntity> queue,
     OperatorSettings settings,
     IEntityLabelSelector<TEntity> labelSelector,
+    IFusionCacheProvider cacheProvider,
     IKubernetesClient client,
     IHostApplicationLifetime hostApplicationLifetime,
     LeaderElector elector)
@@ -31,6 +34,7 @@ internal sealed class LeaderAwareResourceWatcher<TEntity>(
         queue,
         settings,
         labelSelector,
+        cacheProvider,
         client)
     where TEntity : IKubernetesObject<V1ObjectMeta>
 {

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

@@ -12,13 +12,16 @@
 using KubeOps.Abstractions.Entities;
 using KubeOps.Abstractions.Finalizer;
 using KubeOps.KubernetesClient;
+using KubeOps.Operator.Constants;
 using KubeOps.Operator.Logging;
 using KubeOps.Operator.Queue;
 
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 
+using ZiggyCreatures.Caching.Fusion;
+
 namespace KubeOps.Operator.Watcher;
 
 public class ResourceWatcher<TEntity>(
@@ -28,12 +31,14 @@ public class ResourceWatcher<TEntity>(
     TimedEntityQueue<TEntity> requeue,
     OperatorSettings settings,
     IEntityLabelSelector<TEntity> labelSelector,
+    IFusionCacheProvider cacheProvider,
     IKubernetesClient client)
     : IHostedService, IAsyncDisposable, IDisposable
     where TEntity : IKubernetesObject<V1ObjectMeta>
 {
-    private readonly ConcurrentDictionary<string, long> _entityCache = new();
+    private readonly ConcurrentDictionary<string, SemaphoreSlim> _entityLocks = new();
 
+    private readonly IFusionCache _entityCache = cacheProvider.GetCache(CacheConstants.CacheNames.ResourceWatcher);
     private CancellationTokenSource _cancellationTokenSource = new();
     private uint _watcherReconnectRetries;
     private Task? _eventWatcher;
@@ -132,42 +137,68 @@ static async ValueTask CastAndDispose(IDisposable resource)
 
     protected virtual async Task OnEventAsync(WatchEventType type, TEntity entity, CancellationToken cancellationToken)
     {
+        SemaphoreSlim? semaphore;
+
         switch (type)
         {
             case WatchEventType.Added:
-                if (_entityCache.TryAdd(entity.Uid(), entity.Generation() ?? 0))
+                semaphore = _entityLocks.GetOrAdd(entity.Uid(), _ => new(1, 1));
+                await semaphore.WaitAsync(cancellationToken);
+
+                try
                 {
-                    // Only perform reconciliation if the entity was not already in the cache.
-                    await ReconcileModificationAsync(entity, cancellationToken);
+                    var cachedGeneration = await _entityCache.TryGetAsync<long?>(entity.Uid(), token: cancellationToken);
+
+                    if (!cachedGeneration.HasValue)
+                    {
+                        // Only perform reconciliation if the entity was not already in the cache.
+                        await _entityCache.SetAsync(entity.Uid(), entity.Generation() ?? 0, token: cancellationToken);
+                        await ReconcileModificationAsync(entity, cancellationToken);
+                    }
+                    else
+                    {
+                        logger.LogDebug(
+                            """Received ADDED event for entity "{Kind}/{Name}" which was already in the cache. Skip event.""",
+                            entity.Kind,
+                            entity.Name());
+                    }
                 }
-                else
+                finally
                 {
-                    logger.LogDebug(
-                        """Received ADDED event for entity "{Kind}/{Name}" which was already in the cache. Skip event.""",
-                        entity.Kind,
-                        entity.Name());
+                    semaphore.Release();
                 }
 
                 break;
             case WatchEventType.Modified:
                 switch (entity)
                 {
                     case { Metadata.DeletionTimestamp: null }:
-                        _entityCache.TryGetValue(entity.Uid(), out var cachedGeneration);
+                        semaphore = _entityLocks.GetOrAdd(entity.Uid(), _ => new(1, 1));
+                        await semaphore.WaitAsync(cancellationToken);
 
-                        // Check if entity spec has changed through "Generation" value increment. Skip reconcile if not changed.
-                        if (entity.Generation() <= cachedGeneration)
+                        try
                         {
-                            logger.LogDebug(
-                                """Entity "{Kind}/{Name}" modification did not modify generation. Skip event.""",
-                                entity.Kind,
-                                entity.Name());
-                            return;
+                            var cachedGeneration = await _entityCache.TryGetAsync<long?>(entity.Uid(), token: cancellationToken);
+
+                            // Check if entity spec has changed through "Generation" value increment. Skip reconcile if not changed.
+                            if (cachedGeneration.HasValue && cachedGeneration >= entity.Generation())
+                            {
+                                logger.LogDebug(
+                                    """Entity "{Kind}/{Name}" modification did not modify generation. Skip event.""",
+                                    entity.Kind,
+                                    entity.Name());
+                                return;
+                            }
+
+                            // update cached generation since generation now changed
+                            await _entityCache.SetAsync(entity.Uid(), entity.Generation() ?? 1, token: cancellationToken);
+                            await ReconcileModificationAsync(entity, cancellationToken);
+                        }
+                        finally
+                        {
+                            semaphore.Release();
                         }
 
-                        // update cached generation since generation now changed
-                        _entityCache.TryUpdate(entity.Uid(), entity.Generation() ?? 1, cachedGeneration);
-                        await ReconcileModificationAsync(entity, cancellationToken);
                         break;
                     case { Metadata: { DeletionTimestamp: not null, Finalizers.Count: > 0 } }:
                         await ReconcileFinalizersSequentialAsync(entity, cancellationToken);
@@ -311,7 +342,7 @@ e.InnerException is EndOfStreamException &&
     private async Task ReconcileDeletionAsync(TEntity entity, CancellationToken cancellationToken)
     {
         requeue.Remove(entity);
-        _entityCache.TryRemove(entity.Uid(), out _);
+        await _entityCache.RemoveAsync(entity.Uid(), token: cancellationToken);
 
         await using var scope = provider.CreateAsyncScope();
         var controller = scope.ServiceProvider.GetRequiredService<IEntityController<TEntity>>();

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

@@ -7,13 +7,16 @@
 using KubeOps.Abstractions.Builder;
 using KubeOps.Abstractions.Entities;
 using KubeOps.KubernetesClient;
+using KubeOps.Operator.Constants;
 using KubeOps.Operator.Queue;
 using KubeOps.Operator.Watcher;
 
 using Microsoft.Extensions.Logging;
 
 using Moq;
 
+using ZiggyCreatures.Caching.Fusion;
+
 namespace KubeOps.Operator.Test.Watcher;
 
 public sealed class ResourceWatcherTest
@@ -28,13 +31,27 @@ 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 cache = Mock.Of<IFusionCache>();
+        var cacheProvider = Mock.Of<IFusionCacheProvider>();
         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>(activitySource, logger, serviceProvider, timedEntityQueue, operatorSettings, labelSelector, kubernetesClient);
+        Mock.Get(cacheProvider)
+            .Setup(cp => cp.GetCache(It.Is<string>(s => s == CacheConstants.CacheNames.ResourceWatcher)))
+            .Returns(() => cache);
+
+        var resourceWatcher = new ResourceWatcher<V1Pod>(
+            activitySource,
+            logger,
+            serviceProvider,
+            timedEntityQueue,
+            operatorSettings,
+            labelSelector,
+            cacheProvider,
+            kubernetesClient);
 
         // Act.
         // Start and stop the watcher.