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

* Added 'WatchKube' discovery provider
* Error handling, first results awaiting, unit tests
* acceptance test, description id docs
* increase first results timeout
* remove BDDfy
* fixes after rebase
* minor fixes
* added tests
* adjust tests
* add ObservableExtensions tests
* Code review by @raman-m
* Is unit coverage 100% ?
* Converting the internal constants into public static properties
* Update docs
---------
Co-authored-by: VIMPELCOM_MAIN\NKuksov <nkuksov@beeline.ru>
Co-authored-by: Raman Maksimchuk <dotnet044@gmail.com>

Author: 3 people

.config/dotnet-tools.json

@@ -13,6 +13,13 @@
       "commands": [
         "csmacnz.Coveralls"
       ]
+    },
+    "dotnet-reportgenerator-globaltool": {
+      "version": "5.4.7",
+      "commands": [
+        "reportgenerator"
+      ],
+      "rollForward": false
     }
   }
 }

docs/features/kubernetes.rst

@@ -179,8 +179,21 @@ The following examples show how to set up a route that will work in Kubernetes.
 The most important thing is the ``ServiceName`` which is made up of the Kubernetes service name.
 We also need to set up the ``ServiceDiscoveryProvider`` in ``GlobalConfiguration``.
 
+Regarding global and route configurations, if your downstream service resides in a different namespace, you can override the global setting at the route level by specifying a ``ServiceNamespace``.
+
+.. code-block:: json
+
+  "Routes": [
+    {
+      "ServiceName": "my-service",
+      "ServiceNamespace": "my-namespace"
+    }
+  ]
+
+.. _k8s-kube-provider:
+
 ``Kube`` provider
-^^^^^^^^^^^^^^^^^
+-----------------
 
 The example here shows a typical configuration:
 
@@ -203,7 +216,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.
@@ -215,7 +228,7 @@ Service deployment in ``Dev`` namespace, and discovery provider type is ``Kube``
 .. _k8s-pollkube-provider:
 
 ``PollKube`` provider
-^^^^^^^^^^^^^^^^^^^^^
+---------------------
 
 You use Ocelot to poll Kubernetes for latest service information rather than per request.
 If you want to poll Kubernetes for the latest services rather than per request (default behaviour) then you need to set the following configuration:
@@ -236,23 +249,81 @@ 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>`_. 
 
-Global vs Route levels
-^^^^^^^^^^^^^^^^^^^^^^
+.. _k8s-watchkube-provider:
 
-If your downstream service resides in a different namespace, you can override the global setting at the route-level by specifying a ``ServiceNamespace``:
+``WatchKube`` provider [#f3]_
+-----------------------------
+.. _Kubernetes API: https://kubernetes.io/docs/reference/using-api/
+.. _watch requests: https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes
+
+With this configuration, `Kubernetes API`_ "`watch requests`_" are used to fetch service configuration.
+Essentially, it establishes one streamed HTTP connection with the `Kubernetes API`_ per downstream service.
+Changes streamed through this connection will be used to update the list of available endpoints.
 
 .. code-block:: json
 
-  "Routes": [
-    {
-      "ServiceName": "my-service",
-      "ServiceNamespace": "my-namespace"
-    }
-  ]
+  "ServiceDiscoveryProvider": {
+    "Namespace": "dev",
+    "Type": "WatchKube"
+  }
+
+The provider has an implicit configuration for fine-tuned watching, which are available and can only be initialized in C# code.
+
+* ``WatchKube.FirstResultsFetchingTimeoutSeconds``: `This <https://github.com/search?q=repo%3AThreeMammals%2FOcelot%20FirstResultsFetchingTimeoutSeconds&type=code>`_ is the default number of seconds to wait after Ocelot starts, following the provider's creation, to fetch the first result from the Kubernetes endpoint. :sup:`1`
+* ``WatchKube.FailedSubscriptionRetrySeconds``: `This <https://github.com/search?q=repo%3AThreeMammals%2FOcelot%20FailedSubscriptionRetrySeconds&type=code>`__ is the default number of seconds to wait before scheduling the next retry for the subscription operation. :sup:`1`
+
+.. _break3: http://break.do
+
+  **Note 1**: For both ``static int`` properties, the default value is 1 (one) second. The constraint ensures that the assigned value is greater than or equal to 1 (one). Therefore, the minimum value is 1 (one) second.
+
+  **Note 2**: The ``WatchKube`` provider is specifically designed for high-load Ocelot vs. Kubernetes environments with high RPS ratios.
+  To better understand which type is suitable for your needs, we have added a table :ref:`k8s-comparing-providers`.
+  
+.. _k8s-comparing-providers:
+
+Comparing providers
+-------------------
+This table explains the most important indicators that may influence Ocelot vs. Kubernetes deployment or DevOps strategy.
+The evolution path of all providers follows: ``Kube`` -> ``PollKube`` -> ``WatchKube``, with ``WatchKube`` being the most advanced provider.
+
+.. list-table::
+  :widths: 34 22 22 22
+  :header-rows: 1
+
+  * - *Indicators \\ Providers*
+    - :ref:`Kube <k8s-kube-provider>`
+    - :ref:`PollKube <k8s-pollkube-provider>`
+    - :ref:`WatchKube <k8s-watchkube-provider>`
+  * - Extra latency
+    - One hop per route
+    - \-
+    - \-
+  * - Speed of response to endpoints changes
+    - High
+    - Low :sup:`1`
+    - High
+  * - Pressure on `Kubernetes API`_
+    - High
+    - Low :sup:`1`
+    - Low
+  * - Ocelot load (estimated) :sup:`2`
+    - < 1000 RPS
+    - > 1000 RPS
+    - > 5000 RPS
+  * - Ocelot deployment :sup:`3`
+    - Single instance
+    - Multiple instances
+    - Cluster of instances
+
+.. _break4: http://break.do
+
+  | :sup:`1` Depends on the ``PollingInterval`` option.
+  | :sup:`2` Please consider this a rough load estimation, as our team has not provided any tests or benchmarks.
+  | :sup:`3` The term "instance" refers to an Ocelot instance, not a Kubernetes one.
 
 .. _k8s-downstream-scheme-vs-port-names:
 
-Downstream Scheme vs Port Names [#f3]_
+Downstream Scheme vs Port Names [#f4]_
 --------------------------------------
 
 Kubernetes configuration permits the definition of multiple ports with names for each address of an endpoint subset.
@@ -286,7 +357,7 @@ you must define ``DownstreamScheme`` to enable the provider to recognize the des
     }
   ]
 
-.. _break3: http://break.do
+.. _break5: http://break.do
 
   **Note**: In the absence of a specified ``DownstreamScheme`` (which is the default behavior), the ``Kube`` provider will select **the first available port** from the ``EndpointSubsetV1.Ports`` collection.
   Consequently, if the port name is not designated, the default downstream scheme utilized will be ``http``.
@@ -295,13 +366,16 @@ you must define ``DownstreamScheme`` to enable the provider to recognize the des
 
 .. [#f1] The :doc:`../features/kubernetes` feature was requested as part of issue `345`_ to add support for `Kubernetes <https://kubernetes.io/>`_ :doc:`../features/servicediscovery` provider, and released in version `13.4.1`_ 
 .. [#f2] The :ref:`k8s-addkubernetes-action-method` was requested as part of issue `2255`_ (PR `2257`_), and released in version `24.0`_
-.. [#f3] The :ref:`k8s-downstream-scheme-vs-port-names` feature was requested as part of issue `1967`_ and released in version `23.3`_
+.. [#f3] The :ref:`k8s-watchkube-provider` was discussed in thread `2168`_ and released in version `24.1`_
+.. [#f4] The :ref:`k8s-downstream-scheme-vs-port-names` feature was requested as part of issue `1967`_ and released in version `23.3`_
 
 .. _345: https://github.com/ThreeMammals/Ocelot/issues/345
 .. _1134: https://github.com/ThreeMammals/Ocelot/pull/1134
 .. _1967: https://github.com/ThreeMammals/Ocelot/issues/1967
+.. _2168: https://github.com/ThreeMammals/Ocelot/discussions/2168
 .. _2255: https://github.com/ThreeMammals/Ocelot/issues/2255
 .. _2257: https://github.com/ThreeMammals/Ocelot/pull/2257
 .. _13.4.1: https://github.com/ThreeMammals/Ocelot/releases/tag/13.4.1
 .. _23.3: https://github.com/ThreeMammals/Ocelot/releases/tag/23.3.0
 .. _24.0: https://github.com/ThreeMammals/Ocelot/releases/tag/24.0.0
+.. _24.1: https://github.com/ThreeMammals/Ocelot/releases/tag/24.1.0

src/Ocelot.Provider.Kubernetes/EndPointClientV1.cs

@@ -7,27 +7,41 @@ 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)
     {
-        if (string.IsNullOrEmpty(serviceName))
-        {
-            throw new ArgumentNullException(nameof(serviceName));
-        }
+        ArgumentException.ThrowIfNullOrEmpty(serviceName);
 
-        var request = Collection
-            .WithTemplateParameters(new
-            {
-                Namespace = kubeNamespace ?? KubeClient.DefaultNamespace,
-                ServiceName = serviceName,
-            });
+        var request = EndpointsRequest.WithTemplateParameters(new
+        {
+            Namespace = kubeNamespace ?? KubeClient.DefaultNamespace,
+            ServiceName = serviceName,
+        });
 
         return Http.GetAsync(request, cancellationToken)
-            .ReadContentAsObjectV1Async<EndpointsV1>(operationDescription: $"get {nameof(EndpointsV1)}");
+            .ReadContentAsObjectV1Async<EndpointsV1>(operationDescription: $"{nameof(GetAsync)} {nameof(EndpointsV1)}");
+    }
+
+    public IObservable<IResourceEventV1<EndpointsV1>> Watch(string serviceName, string kubeNamespace, CancellationToken cancellationToken = default)
+    {
+        ArgumentException.ThrowIfNullOrEmpty(serviceName);
+
+        var request = EndpointsWatchRequest.WithTemplateParameters(new
+        {
+            ServiceName = serviceName,
+            Namespace = kubeNamespace ?? KubeClient.DefaultNamespace,
+        });
+
+        return ObserveEvents<EndpointsV1>(request,
+            $"{nameof(Watch)} {nameof(EndpointsV1)} for '{serviceName}' in the 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,13 +2,15 @@
 using Ocelot.Configuration;
 using Ocelot.Logging;
 using Ocelot.Provider.Kubernetes.Interfaces;
+using System.Reactive.Concurrency;
 
 namespace Ocelot.Provider.Kubernetes;
 
 public static class KubernetesProviderFactory // TODO : IServiceDiscoveryProviderFactory
 {
     /// <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;
 
@@ -17,14 +19,18 @@ private static IServiceDiscoveryProvider CreateProvider(IServiceProvider provide
         var factory = provider.GetService<IOcelotLoggerFactory>();
         var kubeClient = provider.GetService<IKubeApiClient>();
         var serviceBuilder = provider.GetService<IKubeServiceBuilder>();
-
         var configuration = new KubeRegistryConfiguration
         {
             KeyOfServiceInK8s = route.ServiceName,
             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,19 @@
+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)
+        => 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

@@ -48,5 +48,6 @@
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\Ocelot\Ocelot.csproj" />
+    <InternalsVisibleTo Include="Ocelot.UnitTests" />
   </ItemGroup>
 </Project>