#2168 A brand-new WatchKube provider for Kubernetes service discovery

docs/features/kubernetes.rst

@@ -203,7 +203,7 @@ The example here shows a typical configuration:
     }
   }
 
-Service deployment in ``Dev`` namespace, and discovery provider type is ``Kube``, you also can set :ref:`k8s-pollkube-provider` type.
+Service deployment in ``Dev`` namespace, and discovery provider type is ``Kube``, you also can set :ref:`k8s-pollkube-provider` or :ref:`k8s-watchkube-provider` type.
 
   **Note 1**: ``Scheme``, ``Host``, ``Port``, and ``Token`` are not used if ``usePodServiceAccount`` is true when `KubeClient`_ is created from a pod service account.
   Please refer to the :ref:`k8s-install` section for technical details.
@@ -236,6 +236,22 @@ The polling interval is in milliseconds and tells Ocelot how often to call Kuber
   We doubt it will matter for most people and polling may give a tiny performance improvement over calling Kubernetes per request.
   There is no way for Ocelot to work these out for you, except perhaps through a `discussion <https://github.com/ThreeMammals/Ocelot/discussions>`_. 
 
+.. _k8s-watchkube-provider:
+
+WatchKube provider
+^^^^^^^^^^^^^^^^^^
+
+This option utilizes Kubernetes API `watch requests <https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes>`_ for fetching service configuration.
+Essentially it means that there will be one streamed http connection with kube-api per downstream service.
+Changes streamed by this connection will be used for updating available endpoints list.
+
+.. code-block:: json
+
+  "ServiceDiscoveryProvider": {
+    "Namespace": "dev",
+    "Type": "WatchKube"
+  }
+
 Global vs Route levels
 ^^^^^^^^^^^^^^^^^^^^^^
 

src/Ocelot.Provider.Kubernetes/EndPointClientV1.cs

@@ -7,27 +7,49 @@ namespace Ocelot.Provider.Kubernetes;
 
 public class EndPointClientV1 : KubeResourceClient, IEndPointClient
 {
-    private static readonly HttpRequest Collection = KubeRequest.Create("api/v1/namespaces/{Namespace}/endpoints/{ServiceName}");
+    private static readonly HttpRequest EndpointsRequest =
+        KubeRequest.Create("api/v1/namespaces/{Namespace}/endpoints/{ServiceName}");
+
+    private static readonly HttpRequest EndpointsWatchRequest =
+        KubeRequest.Create("api/v1/watch/namespaces/{Namespace}/endpoints/{ServiceName}");
 
     public EndPointClientV1(IKubeApiClient client) : base(client)
     {
     }
 
-    public Task<EndpointsV1> GetAsync(string serviceName, string kubeNamespace = null, CancellationToken cancellationToken = default)
+    public Task<EndpointsV1> GetAsync(string serviceName, string kubeNamespace = null,
+        CancellationToken cancellationToken = default)
     {
         if (string.IsNullOrEmpty(serviceName))
         {
             throw new ArgumentNullException(nameof(serviceName));
         }
 
-        var request = Collection
+        var request = EndpointsRequest
             .WithTemplateParameters(new
             {
-                Namespace = kubeNamespace ?? KubeClient.DefaultNamespace,
-                ServiceName = serviceName,
+                Namespace = kubeNamespace ?? KubeClient.DefaultNamespace, ServiceName = serviceName,
             });
 
         return Http.GetAsync(request, cancellationToken)
             .ReadContentAsObjectV1Async<EndpointsV1>(operationDescription: $"get {nameof(EndpointsV1)}");
     }
+
+    public IObservable<IResourceEventV1<EndpointsV1>> Watch(string serviceName, string kubeNamespace,
+        CancellationToken cancellationToken = default)
+    {
+        if (string.IsNullOrEmpty(serviceName))
+        {
+            throw new ArgumentNullException(nameof(serviceName));
+        }
+
+        return ObserveEvents<EndpointsV1>(
+            EndpointsWatchRequest.WithTemplateParameters(new
+            {
+                ServiceName = serviceName,
+                Namespace = kubeNamespace ?? KubeClient.DefaultNamespace,
+            }),
+            "watch v1/Endpoints '" + serviceName + "' in namespace " +
+            (kubeNamespace ?? KubeClient.DefaultNamespace));
+    }
 }

src/Ocelot.Provider.Kubernetes/Interfaces/IEndPointClient.cs

@@ -6,4 +6,6 @@ namespace Ocelot.Provider.Kubernetes.Interfaces;
 public interface IEndPointClient : IKubeResourceClient
 {
     Task<EndpointsV1> GetAsync(string serviceName, string kubeNamespace = null, CancellationToken cancellationToken = default);
+
+    IObservable<IResourceEventV1<EndpointsV1>> Watch(string serviceName, string kubeNamespace = null, CancellationToken cancellationToken = default);
 }

src/Ocelot.Provider.Kubernetes/Kube.cs

@@ -56,7 +56,7 @@ private async Task<EndpointsV1> GetEndpoint()
         try
         {
             return await _kubeApi
-                .ResourceClient<IEndPointClient>(client => new EndPointClientV1(client))
+                .EndpointsV1()
                 .GetAsync(_configuration.KeyOfServiceInK8s, _configuration.KubeNamespace);
         }
         catch (KubeApiException ex)

src/Ocelot.Provider.Kubernetes/KubeApiClientExtensions.cs

@@ -0,0 +1,9 @@
+using Ocelot.Provider.Kubernetes.Interfaces;
+
+namespace Ocelot.Provider.Kubernetes;
+
+public static class KubeApiClientExtensions
+{
+    public static IEndPointClient EndpointsV1(this IKubeApiClient client)
+        => client.ResourceClient<IEndPointClient>(x => new EndPointClientV1(x));
+}

src/Ocelot.Provider.Kubernetes/KubernetesProviderFactory.cs

@@ -2,6 +2,7 @@
 using Ocelot.Configuration;
 using Ocelot.Logging;
 using Ocelot.Provider.Kubernetes.Interfaces;
+using System.Reactive.Concurrency;
 
 namespace Ocelot.Provider.Kubernetes;
 
@@ -10,9 +11,12 @@ public static class KubernetesProviderFactory // TODO : IServiceDiscoveryProvide
     /// <summary>String constant used for provider type definition.</summary>
     public const string PollKube = nameof(Kubernetes.PollKube);
 
+    public const string WatchKube = nameof(Kubernetes.WatchKube);
+
     public static ServiceDiscoveryFinderDelegate Get { get; } = CreateProvider;
 
-    private static IServiceDiscoveryProvider CreateProvider(IServiceProvider provider, ServiceProviderConfiguration config, DownstreamRoute route)
+    private static IServiceDiscoveryProvider CreateProvider(IServiceProvider provider,
+        ServiceProviderConfiguration config, DownstreamRoute route)
     {
         var factory = provider.GetService<IOcelotLoggerFactory>();
         var kubeClient = provider.GetService<IKubeApiClient>();
@@ -21,10 +25,15 @@ private static IServiceDiscoveryProvider CreateProvider(IServiceProvider provide
         var configuration = new KubeRegistryConfiguration
         {
             KeyOfServiceInK8s = route.ServiceName,
-            KubeNamespace = string.IsNullOrEmpty(route.ServiceNamespace) ? config.Namespace : route.ServiceNamespace,
+            KubeNamespace = string.IsNullOrEmpty(route.ServiceNamespace)
+                ? config.Namespace
+                : route.ServiceNamespace,
             Scheme = route.DownstreamScheme,
         };
 
+        if (WatchKube.Equals(config.Type, StringComparison.OrdinalIgnoreCase))
+            return new WatchKube(configuration, factory, kubeClient, serviceBuilder, Scheduler.Default);
+
         var defaultK8sProvider = new Kube(configuration, factory, kubeClient, serviceBuilder);
 
         return PollKube.Equals(config.Type, StringComparison.OrdinalIgnoreCase)

src/Ocelot.Provider.Kubernetes/ObservableExtensions.cs

@@ -0,0 +1,26 @@
+using System.Reactive.Concurrency;
+using System.Reactive.Linq;
+
+namespace Ocelot.Provider.Kubernetes;
+
+public static class ObservableExtensions
+{
+    public static IObservable<TSource> RetryAfter<TSource>(this IObservable<TSource> source,
+        TimeSpan dueTime,
+        IScheduler scheduler)
+    {
+        return RepeatInfinite(source, dueTime, scheduler).Catch();
+    }
+
+    private static IEnumerable<IObservable<TSource>> RepeatInfinite<TSource>(IObservable<TSource> source,
+        TimeSpan dueTime,
+        IScheduler scheduler)
+    {
+        yield return source;
+
+        while (true)
+        {
+            yield return source.DelaySubscription(dueTime, scheduler);
+        }
+    }
+}

src/Ocelot.Provider.Kubernetes/Ocelot.Provider.Kubernetes.csproj

@@ -49,4 +49,7 @@
   <ItemGroup>
     <ProjectReference Include="..\Ocelot\Ocelot.csproj" />
   </ItemGroup>
-</Project>
+  <ItemGroup>
+    <InternalsVisibleTo Include="Ocelot.UnitTests" />
+  </ItemGroup>
+</Project>

src/Ocelot.Provider.Kubernetes/WatchKube.cs

@@ -0,0 +1,88 @@
+using KubeClient.Models;
+using Ocelot.Logging;
+using Ocelot.Provider.Kubernetes.Interfaces;
+using Ocelot.Values;
+using System.Reactive.Concurrency;
+using System.Reactive.Linq;
+
+namespace Ocelot.Provider.Kubernetes;
+
+public class WatchKube : IServiceDiscoveryProvider, IDisposable
+{
+    internal const int FailedSubscriptionRetrySeconds = 5;
+    internal const int FirstResultsFetchingTimeoutSeconds = 3;
+
+    private readonly KubeRegistryConfiguration _configuration;
+    private readonly IOcelotLogger _logger;
+    private readonly IKubeApiClient _kubeApi;
+    private readonly IKubeServiceBuilder _serviceBuilder;
+    private readonly IScheduler _scheduler;
+
+    private readonly IDisposable _subscription;
+    private readonly TaskCompletionSource _firstResultsCompletionSource;
+
+    private List<Service> _services = new();
+
+    public WatchKube(
+        KubeRegistryConfiguration configuration,
+        IOcelotLoggerFactory factory,
+        IKubeApiClient kubeApi,
+        IKubeServiceBuilder serviceBuilder,
+        IScheduler scheduler)
+    {
+        _configuration = configuration;
+        _logger = factory.CreateLogger<WatchKube>();
+        _kubeApi = kubeApi;
+        _serviceBuilder = serviceBuilder;
+        _scheduler = scheduler;
+
+        _firstResultsCompletionSource = new TaskCompletionSource();
+        SetFirstResultsCompletedAfterDelay();
+        _subscription = CreateSubscription();
+    }
+
+    public virtual async Task<List<Service>> GetAsync()
+    {
+        // wait for first results fetching
+        await _firstResultsCompletionSource.Task;
+
+        if (_services is not { Count: > 0 })
+        {
+            _logger.LogWarning(() => GetMessage("Subscription to service endpoints gave no results!"));
+        }
+
+        return _services;
+    }
+
+    private void SetFirstResultsCompletedAfterDelay() => Observable
+        .Timer(TimeSpan.FromSeconds(FirstResultsFetchingTimeoutSeconds), _scheduler)
+        .Subscribe(_ => _firstResultsCompletionSource.TrySetResult());
+
+    private IDisposable CreateSubscription() =>
+        _kubeApi
+            .EndpointsV1()
+            .Watch(_configuration.KeyOfServiceInK8s, _configuration.KubeNamespace)
+            .Do(_ => { }, ex => _logger.LogError(() => GetMessage("Endpoints subscription error occured."), ex))
+            .RetryAfter(TimeSpan.FromSeconds(FailedSubscriptionRetrySeconds), _scheduler)
+            .Subscribe(
+                onNext: endpointEvent =>
+                {
+                    _services = endpointEvent.EventType switch
+                    {
+                        ResourceEventType.Deleted or ResourceEventType.Error => new(),
+                        _ when (endpointEvent.Resource?.Subsets?.Count ?? 0) == 0 => new(),
+                        _ => _serviceBuilder.BuildServices(_configuration, endpointEvent.Resource).ToList(),
+                    };
+                    _firstResultsCompletionSource.TrySetResult();
+                },
+                onCompleted: () =>
+                {
+                    // called only when subscription canceled in Dispose
+                    _logger.LogInformation(() => GetMessage("Subscription to service endpoints completed"));
+                });
+
+    private string GetMessage(string message)
+        => $"{nameof(WatchKube)} provider. Namespace:{_configuration.KubeNamespace}, Service:{_configuration.KeyOfServiceInK8s}; {message}";
+
+    public void Dispose() => _subscription.Dispose();
+}

test/Ocelot.AcceptanceTests/ConcurrentSteps.cs

@@ -171,8 +171,19 @@ private async Task GetParallelResponse(string url, int threadIndex)
 
     public void ThenAllStatusCodesShouldBe(HttpStatusCode expected)
         => _responses.ShouldAllBe(response => response.Value.StatusCode == expected);
+
     public void ThenAllResponseBodiesShouldBe(string expectedBody)
-        => _responses.ShouldAllBe(response => response.Value.Content.ReadAsStringAsync().Result == expectedBody);
+    {
+        foreach (var r in _responses)
+        {
+            var content = r.Value.Content.ReadAsStringAsync().Result;
+            content = content?.Contains(':') == true
+                ? content.Split(':')[1] // remove counter for body comparison
+                : "0";
+
+            content.ShouldBe(expectedBody);
+        }
+    }
 
     protected string CalledTimesMessage()
         => $"All values are [{string.Join(',', _counters)}]";