fix(queue): delayed enqueues (requeue) will refetch the resource to avoid conflict

Author: Christoph Bühler

src/KubeOps/Operator/Client/KubernetesClient.cs

@@ -154,7 +154,7 @@ public async Task UpdateStatus<TStatus>(IStatus<TStatus> resource)
         {
             if (!(resource is IKubernetesObject<V1ObjectMeta> kubernetesObject))
             {
-                throw new ArgumentException("Resource is not a propper kubernetes object");
+                throw new ArgumentException("Resource is not a proper kubernetes object");
             }
 
             var crd = kubernetesObject.CreateResourceDefinition();
@@ -176,7 +176,10 @@ public async Task UpdateStatus<TStatus>(IStatus<TStatus> resource)
             if (result?.ToObject(resource.GetType()) is IKubernetesObject<V1ObjectMeta> parsed)
             {
                 kubernetesObject.Metadata.ResourceVersion = parsed.Metadata.ResourceVersion;
+                return;
             }
+
+            throw new ArgumentException("Could not parse result");
         }
 
         public Task Delete<TResource>(TResource resource)

src/KubeOps/Operator/Queue/ResourceEventQueue.cs

@@ -7,14 +7,15 @@
 using k8s;
 using k8s.Models;
 using KubeOps.Operator.Caching;
+using KubeOps.Operator.Client;
 using KubeOps.Operator.Errors;
 using KubeOps.Operator.Watcher;
 using Microsoft.Extensions.Logging;
 
 namespace KubeOps.Operator.Queue
 {
     internal class ResourceEventQueue<TEntity> : IResourceEventQueue<TEntity>
-        where TEntity : IKubernetesObject<V1ObjectMeta>
+        where TEntity : class, IKubernetesObject<V1ObjectMeta>
     {
         // TODO: Make configurable
         private const int QueueLimit = 512;
@@ -24,6 +25,7 @@ internal class ResourceEventQueue<TEntity> : IResourceEventQueue<TEntity>
 
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1);
         private readonly ILogger<ResourceEventQueue<TEntity>> _logger;
+        private readonly IKubernetesClient _client;
         private readonly IResourceCache<TEntity> _cache;
         private readonly IResourceWatcher<TEntity> _watcher;
 
@@ -39,10 +41,12 @@ internal class ResourceEventQueue<TEntity> : IResourceEventQueue<TEntity>
 
         public ResourceEventQueue(
             ILogger<ResourceEventQueue<TEntity>> logger,
+            IKubernetesClient client,
             IResourceCache<TEntity> cache,
             IResourceWatcher<TEntity> watcher)
         {
             _logger = logger;
+            _client = client;
             _cache = cache;
             _watcher = watcher;
         }
@@ -131,17 +135,20 @@ public async Task Enqueue(TEntity resource, TimeSpan? enqueueDelay = null)
                                 _semaphore.Release();
                             }
 
-                            var cachedResource = _cache.Get(delayedResource.Metadata.Uid);
-                            if (cachedResource == null)
+                            var newResource = await _client.Get<TEntity>(
+                                delayedResource.Metadata.Name,
+                                delayedResource.Metadata.NamespaceProperty);
+
+                            if (newResource == null)
                             {
                                 _logger.LogDebug(
-                                    @"Resource ""{kind}/{name}"" was not present in the cache anymore. Don't execute delayed timer.",
-                                    delayedResource.Kind,
-                                    delayedResource.Metadata.Name);
+                                    @"Resource ""{kind}/{name}"" for enqueued event was not present anymore.",
+                                    resource.Kind,
+                                    resource.Metadata.Name);
                                 return;
                             }
 
-                            await Enqueue(cachedResource);
+                            await Enqueue(newResource);
                         });
                     _delayedEnqueue.Add(resource.Metadata.Uid, timer);