#2304 Utilize ConcurrentQueue{T} for managing service versions during polling by the PollKube discovery provider (#2335)

* CA1816: Call GC.SuppressFinalize correctly
* Enable testing for the PollKube provider
* Use ConcurrentQueue<T> to handle service versions while polling
* PollKube unit tests
* PollKube acceptance testing

Author: raman-m

src/Ocelot.Provider.Kubernetes/KubernetesProviderFactory.cs

@@ -31,10 +31,10 @@ private static IServiceDiscoveryProvider CreateProvider(IServiceProvider provide
             return new WatchKube(configuration, factory, kubeClient, serviceBuilder, Scheduler.Default);
         }
 
-        var defaultK8sProvider = new Kube(configuration, factory, kubeClient, serviceBuilder);
+        var kubeProvider = new Kube(configuration, factory, kubeClient, serviceBuilder);
 
         return PollKube.Equals(config.Type, StringComparison.OrdinalIgnoreCase)
-            ? new PollKube(config.PollingInterval, factory, defaultK8sProvider)
-            : defaultK8sProvider;
+            ? new PollKube(config.PollingInterval, factory, kubeProvider)
+            : kubeProvider;
     }
 }

src/Ocelot.Provider.Kubernetes/PollKube.cs

@@ -1,29 +1,30 @@
 using Ocelot.Logging;
 using Ocelot.Values;
+using System.Collections.Concurrent;
 
 namespace Ocelot.Provider.Kubernetes;
 
 public class PollKube : IServiceDiscoveryProvider, IDisposable
 {
     private readonly IOcelotLogger _logger;
-    private readonly IServiceDiscoveryProvider _kubeServiceDiscoveryProvider;
-    private readonly Timer _timer;
-    
+    private readonly IServiceDiscoveryProvider _discoveryProvider;
+    private readonly ConcurrentQueue<List<Service>> _queue = new();
+
+    private Timer _timer;
     private bool _polling;
-    private List<Service> _services;
 
-    public PollKube(int pollingInterval, IOcelotLoggerFactory factory, IServiceDiscoveryProvider kubeServiceDiscoveryProvider)
+    public PollKube(int pollingInterval, IOcelotLoggerFactory factory, IServiceDiscoveryProvider kubeProvider)
     {
         _logger = factory.CreateLogger<PollKube>();
-        _kubeServiceDiscoveryProvider = kubeServiceDiscoveryProvider;
-        _services = new List<Service>();
-
-        _timer = new Timer(OnTimerCallbackAsync, null, pollingInterval, pollingInterval);
+        _discoveryProvider = kubeProvider;
+        _timer = new(OnTimerCallbackAsync, null, pollingInterval, pollingInterval);
     }
 
     private async void OnTimerCallbackAsync(object state)
     {
-        if (_polling)
+        // Avoid polling if already in progress due to a slow completion of the Poll task,
+        // and ensure no more than three versions of services remain in the queue.
+        if (_polling || _queue.Count > 3)
         {
             return;
         }
@@ -33,18 +34,54 @@ private async void OnTimerCallbackAsync(object state)
         _polling = false;
     }
 
-    public Task<List<Service>> GetAsync()
+    public async Task<List<Service>> GetAsync()
     {
-        return Task.FromResult(_services);
+        // First cold request must call the provider
+        if (_queue.IsEmpty)
+        {
+            return await Poll();
+        }
+        else if (_polling && _queue.TryPeek(out var oldVersion))
+        {
+            return oldVersion;
+        }
+
+        // For services with multiple versions, remove outdated versions and retain only the latest one
+        while (!_polling && _queue.Count > 1 && _queue.TryDequeue(out _))
+        {
+        }
+
+        _queue.TryPeek(out var latestVersion);
+        return latestVersion;
     }
 
-    private async Task Poll()
+    protected virtual async Task<List<Service>> Poll()
     {
-        _services = await _kubeServiceDiscoveryProvider.GetAsync();
+        _polling = true;
+        try
+        {
+            var services = await _discoveryProvider.GetAsync();
+            _queue.Enqueue(services);
+            return services;
+        }
+        finally
+        {
+            _polling = false;
+        }
     }
 
     public void Dispose()
     {
-        _timer.Dispose();
+        Dispose(true);
+        GC.SuppressFinalize(this);
+    }
+
+    protected virtual void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            _timer.Dispose();
+            _timer = null;
+        }
     }
 }

src/Ocelot/LoadBalancer/Balancers/RoundRobin.cs

@@ -34,7 +34,7 @@ public RoundRobin(Func<Task<List<Service>>> services, string serviceName)
 
     public virtual async Task<Response<ServiceHostAndPort>> LeaseAsync(HttpContext httpContext)
     {
-        var services = await _servicesDelegate.Invoke() ?? new List<Service>();
+        var services = await _servicesDelegate.Invoke() ?? new(0);
         if (services.Count == 0)
         {
             return new ErrorResponse<ServiceHostAndPort>(new ServicesAreEmptyError($"There were no services in {Type} for '{_serviceName}' during {nameof(LeaseAsync)} operation!"));

test/Ocelot.AcceptanceTests/ServiceDiscovery/KubernetesServiceDiscoveryTests.cs

@@ -9,6 +9,7 @@
 using Ocelot.Configuration;
 using Ocelot.Configuration.File;
 using Ocelot.DependencyInjection;
+using Ocelot.Infrastructure.Extensions;
 using Ocelot.LoadBalancer.Balancers;
 using Ocelot.Logging;
 using Ocelot.Provider.Kubernetes;
@@ -40,7 +41,7 @@ public KubernetesServiceDiscoveryTests()
 
     [Theory]
     [InlineData(nameof(Kube))]
-    /* [InlineData(nameof(PollKube))] TODO Fails now. Bug 2304? -> https://github.com/ThreeMammals/Ocelot/issues/2304 */
+    [InlineData(nameof(PollKube))] // Bug 2304 -> https://github.com/ThreeMammals/Ocelot/issues/2304
     [InlineData(nameof(WatchKube))]
     public void ShouldReturnServicesFromK8s(string discoveryType)
     {
@@ -108,27 +109,31 @@ public void ShouldReturnServicesByPortNameAsDownstreamScheme(string downstreamSc
 
     [Theory]
     [Trait("Bug", "2110")]
-    [InlineData(1, 30)]
-    [InlineData(2, 50)]
-    [InlineData(3, 50)]
-    [InlineData(4, 50)]
-    [InlineData(5, 50)]
-    [InlineData(6, 99)]
-    [InlineData(7, 99)]
-    [InlineData(8, 99)]
-    [InlineData(9, 999)]
-    [InlineData(10, 999)]
-    public void ShouldHighlyLoadOnStableKubeProvider_WithRoundRobinLoadBalancing(int totalServices, int totalRequests)
+    [InlineData(1, 30, null)]
+    [InlineData(2, 50, null)]
+    [InlineData(3, 50, null)]
+    [InlineData(4, 50, null)]
+    [InlineData(5, 50, null)]
+    [InlineData(6, 99, null)]
+    [InlineData(7, 99, null)]
+    [InlineData(8, 99, null)]
+    [InlineData(9, 999, null)]
+    [InlineData(10, 999, nameof(Kube))]
+    [InlineData(10, 999, nameof(PollKube))]
+    [InlineData(10, 999, nameof(WatchKube))]
+    public void ShouldHighlyLoadOnStableKubeProvider_WithRoundRobinLoadBalancing(int totalServices, int totalRequests, string discoveryType)
     {
         // Skip in MacOS because the test is very unstable
         if (RuntimeInformation.IsOSPlatform(OSPlatform.OSX)) // the test is stable in Linux and Windows only
             return;
 
-        const int ZeroGeneration = 0;
-        var (endpoints, servicePorts) = GivenServiceDiscoveryAndLoadBalancing(totalServices);
+        int zeroGeneration = 0, k8sCount = totalRequests;
+        var (endpoints, servicePorts) = GivenServiceDiscoveryAndLoadBalancing(totalServices, discoveryType ?? nameof(Kube));
         GivenThereIsAFakeKubernetesProvider(endpoints); // stable, services will not be removed from the list
+        if (discoveryType == nameof(WatchKube))
+            k8sCount = GivenWatchReceivedEvent(); // 1
 
-        HighlyLoadOnKubeProviderAndRoundRobinBalancer(totalRequests, ZeroGeneration);
+        HighlyLoadOnKubeProviderAndRoundRobinBalancer(totalRequests, zeroGeneration, k8sCount);
 
         int bottom = totalRequests / totalServices,
             top = totalRequests - (bottom * totalServices) + bottom;
@@ -138,28 +143,37 @@ public void ShouldHighlyLoadOnStableKubeProvider_WithRoundRobinLoadBalancing(int
 
     [Theory]
     [Trait("Bug", "2110")]
-    [InlineData(5, 50, 1)]
-    [InlineData(5, 50, 2)]
-    [InlineData(5, 50, 3)]
-    [InlineData(5, 50, 4)]
-    public void ShouldHighlyLoadOnUnstableKubeProvider_WithRoundRobinLoadBalancing(int totalServices, int totalRequests, int k8sGeneration)
+    [InlineData(5, 50, 1, null)]
+    [InlineData(5, 50, 2, null)]
+    [InlineData(5, 50, 3, null)]
+    [InlineData(5, 50, 4, nameof(Kube))]
+    [InlineData(5, 50, 4, nameof(PollKube))]
+    [InlineData(5, 50, 4, nameof(WatchKube))]
+    public void ShouldHighlyLoadOnUnstableKubeProvider_WithRoundRobinLoadBalancing(int totalServices, int totalRequests, int k8sGeneration, string discoveryType)
     {
         // Skip in MacOS because the test is very unstable
         if (RuntimeInformation.IsOSPlatform(OSPlatform.OSX)) // the test is stable in Linux and Windows only
             return;
 
-        int failPerThreads = (totalRequests / k8sGeneration) - 1; // k8sGeneration means number of offline services
-        var (endpoints, servicePorts) = GivenServiceDiscoveryAndLoadBalancing(totalServices);
+        int failPerThreads = (totalRequests / k8sGeneration) - 1, // k8sGeneration means number of offline services
+            k8sCount = totalRequests;
+        var (endpoints, servicePorts) = GivenServiceDiscoveryAndLoadBalancing(totalServices, discoveryType ?? nameof(Kube));
         GivenThereIsAFakeKubernetesProvider(endpoints, false, k8sGeneration, failPerThreads); // false means unstable, k8sGeneration services will be removed from the list
+        if (discoveryType == nameof(WatchKube))
+        {
+            k8sCount = GivenWatchReceivedEvent(); // 1
+            k8sGeneration = 0;
+        }
 
-        HighlyLoadOnKubeProviderAndRoundRobinBalancer(totalRequests, k8sGeneration);
+        HighlyLoadOnKubeProviderAndRoundRobinBalancer(totalRequests, k8sGeneration, k8sCount);
 
         ThenAllServicesCalledOptimisticAmountOfTimes(_roundRobinAnalyzer); // with unstable checkings
         ThenServiceCountersShouldMatchLeasingCounters(_roundRobinAnalyzer, servicePorts, totalRequests);
     }
 
     [Theory]
     [InlineData(nameof(Kube))]
+    [InlineData(nameof(PollKube))] // Bug 2304 -> https://github.com/ThreeMammals/Ocelot/issues/2304
     [InlineData(nameof(WatchKube))]
     [Trait("Feat", "2256")]
     public void ShouldReturnServicesFromK8s_AddKubernetesWithNullConfigureOptions(string discoveryType)
@@ -238,7 +252,7 @@ public void ShouldReturnServicesFromK8s_OneWatchRequestUpdatesServicesInfo()
     [Trait("Feat", "2319")]
     [Trait("PR", "2324")] // https://github.com/ThreeMammals/Ocelot/pull/2324
     [InlineData(nameof(Kube))]
-    /* [InlineData(nameof(PollKube))] // Bug 2304 -> https://github.com/ThreeMammals/Ocelot/issues/2304 */
+    [InlineData(nameof(PollKube))] // Bug 2304 -> https://github.com/ThreeMammals/Ocelot/issues/2304
     [InlineData(nameof(WatchKube))]
     public void ShouldApplyGlobalLoadBalancerOptions_ForAllDynamicRoutes(string discoveryType)
     {
@@ -260,7 +274,11 @@ static void ConfigureDynamicRouting(FileConfiguration configuration)
         var upstreamPath = $"/{ServiceNamespace()}.{ServiceName()}/";
         WhenIGetUrlOnTheApiGatewayConcurrently(upstreamPath, 50);
 
-        _k8sCounter.ShouldBe(discoveryType == nameof(WatchKube) ? 1 : 50);
+        if (discoveryType == nameof(PollKube))
+            _k8sCounter.ShouldBeGreaterThanOrEqualTo(50); // can be 50, 51 and sometimes 52
+        else
+            _k8sCounter.ShouldBe(discoveryType == nameof(WatchKube) ? 1 : 50);
+
         _k8sServiceGeneration.ShouldBe(0);
         ThenAllStatusCodesShouldBe(HttpStatusCode.OK);
         ThenAllServicesShouldHaveBeenCalledTimes(50);
@@ -293,21 +311,21 @@ private void AddKubernetesWithNullConfigureOptions(IServiceCollection services)
         downstreams.ForEach(ds => GivenSubsetAddress(ds, subset));
         var endpoints = GivenEndpoints(subset, serviceName); // totalServices service instances with different ports
         var route = GivenRouteWithServiceName(serviceName, loadBalancerType); // !!!
-        var configuration = GivenKubeConfiguration(route, discoveryType);
+        var configuration = GivenKubeConfiguration(route, discoveryType.IfEmpty(nameof(Kube)));
         configure?.Invoke(configuration);
         GivenMultipleServiceInstancesAreRunning(downstreamUrls, downstreamResponses);
         GivenThereIsAConfiguration(configuration);
         GivenOcelotIsRunning(services ?? WithKubernetesAndRoundRobin);
         return (endpoints, servicePorts);
     }
 
-    private void HighlyLoadOnKubeProviderAndRoundRobinBalancer(int totalRequests, int k8sGenerationNo)
+    private void HighlyLoadOnKubeProviderAndRoundRobinBalancer(int totalRequests, int k8sGenerationNo, int? k8sCount = null)
     {
         // Act
         WhenIGetUrlOnTheApiGatewayConcurrently("/", totalRequests); // load by X parallel requests
 
         // Assert
-        _k8sCounter.ShouldBeGreaterThanOrEqualTo(totalRequests); // integration endpoint called times
+        _k8sCounter.ShouldBeGreaterThanOrEqualTo(k8sCount ?? totalRequests); // integration endpoint called times
         _k8sServiceGeneration.ShouldBe(k8sGenerationNo);
         ThenAllStatusCodesShouldBe(HttpStatusCode.OK);
         ThenAllServicesShouldHaveBeenCalledTimes(totalRequests);
@@ -373,7 +391,7 @@ private FileConfiguration GivenKubeConfiguration(FileRoute route, string type, s
             Host = u.Host,
             Port = u.Port,
             Type = type,
-            PollingInterval = 0,
+            PollingInterval = 3, // 3ms is very fast polling, make sense for PollKube provider only
             Namespace = ServiceNamespace(),
             Token = token ?? "Test",
         };
@@ -442,7 +460,7 @@ private void GivenThereIsAFakeKubernetesProvider(ResourceEventV1<EndpointsV1>[]
         handler.GivenThereIsAServiceRunningOn(_kubernetesUrl, (c) => GivenHandleWatchRequest(c, events, namespaces, serviceName));
     }
 
-    private void GivenWatchReceivedEvent() => _k8sWatchResetEvent.Set();
+    private int GivenWatchReceivedEvent() => _k8sWatchResetEvent.Set() ? 1 : 0;
     private static Task GivenDelay(int milliseconds) => Task.Delay(TimeSpan.FromMilliseconds(milliseconds));
 
     private async Task GivenHandleWatchRequest(HttpContext context,