Merge pull request #630 from Xabaril/k8s-operator-cluster-and-namespaced

[K8s operator] Cluster and Namespaced services support
Xabaril · Sep 11, 2020 · f5d1b8a · f5d1b8a
2 parents 0005c30 + 0e35c7c
commit f5d1b8a
Show file tree

Hide file tree

Showing 12 changed files with 196 additions and 17 deletions.
diff --git a/build/dependencies.props b/build/dependencies.props
@@ -78,7 +78,7 @@
     <MicrosoftAzureKeyVault>4.1.0</MicrosoftAzureKeyVault>
     <DogStatsDCSharpClient>3.3.0</DogStatsDCSharpClient>
     <MicrosoftAzureDevices>1.18.1</MicrosoftAzureDevices>
-    <KubernetesClient>1.6.10</KubernetesClient>
+    <KubernetesClient>2.0.29</KubernetesClient>
     <SolrNetCore>1.0.19</SolrNetCore>
     <IBMMQDotnetClient>9.1.4</IBMMQDotnetClient>
     <PomeloEntityFrameworkCoreMySql>3.1.2</PomeloEntityFrameworkCoreMySql>

diff --git a/deploy/operator/crd/healthcheck-crd.yaml b/deploy/operator/crd/healthcheck-crd.yaml
@@ -23,6 +23,11 @@ spec:
           properties:
             name:
               type: string
+            scope:
+             type: string
+             enum:
+               - Cluster
+               - Namespaced
             serviceType:
               type: string
               enum:
@@ -99,4 +104,5 @@ spec:
 
           required:
             - name
+            - scope
             - servicesLabel
diff --git a/doc/k8s-operator.md b/doc/k8s-operator.md
@@ -38,8 +38,21 @@ The [HealthCheck operator definition](https://github.com/Xabaril/AspNetCore.Diag
 | Field         | Description                                                                        |
 | ------------- | :--------------------------------------------------------------------------------- |
 | name          | Name of the healthcheck resource                                                   |
+| scope         | Cluster / Namespaced                                                               |
 | servicesLabel | The label the operator service watcher will use to detected healthchecks endpoints |
 
+
+### Scope definition (Cluster or Namespaced)
+
+The scope field (Cluster or Namespaced) is mandatory and will specify to the operator whether it should watch for healthchecks services in the
+same namespace where the UI resource is created or watch to all services in all namespaces.
+
+If you wan't to have different UI's for different namespaced services you should use **Namespaced**
+
+If you wan't to have a single UI that monitors all the healthchecks from the cluster you should use **Cluster**
+
+Note: The UI resources created by the operator (deployment, service, configmap, secret, etc) will always be created in the metadata specified namespace.
+
 ### Optional fields
 
 | Field                 | Description                                                          | Default                                              |
@@ -76,6 +89,8 @@ metadata:
   namespace: demo
 spec:
   name: healthchecks-ui
+  scope: Namespaced #The UI will be created at specified namespace (demo) and will watch healthchecks services in demo namespace only
+  #scope: Cluster The UI will be created at specified namespace (demo) but will watch healthcheck services across all namespaces
   servicesLabel: HealthChecks
   serviceType: LoadBalancer
   stylesheetContent: >

diff --git a/src/HealthChecks.UI.K8s.Operator/Constants.cs b/src/HealthChecks.UI.K8s.Operator/Constants.cs
@@ -29,6 +29,12 @@ internal class Operation
                 public const string Delete = "Delete";
                 public const string Patch = "Patch";
             }
+
+            internal class Scope
+            {
+                public const string Cluster = "Cluster";
+                public const string Namespaced = "Namespaced";
+            }
         }
     }
 }
diff --git a/src/HealthChecks.UI.K8s.Operator/Crd/HealthCheckResourceSpec.cs b/src/HealthChecks.UI.K8s.Operator/Crd/HealthCheckResourceSpec.cs
@@ -6,6 +6,7 @@ namespace HealthChecks.UI.K8s.Operator
     public class HealthCheckResourceSpec
     {
         public string Name { get; set; }
+        public string Scope { get; set; }
         public string PortNumber { get; set; }
         public string ServiceType { get; set; }
         public string UiPath { get; set; }

diff --git a/src/HealthChecks.UI.K8s.Operator/Diagnostics/OperatorDiagnostics.cs b/src/HealthChecks.UI.K8s.Operator/Diagnostics/OperatorDiagnostics.cs
@@ -39,7 +39,7 @@ public void OperatorThrow(Exception exception)
 
         public void ServiceWatcherThrow(Exception exception)
         {
-            _logger.LogError(exception, "The operator service watcher threw an unhandled exception");
+            _logger.LogError(exception.Message, "The operator service watcher threw an unhandled exception");
         }
 
         public void UiPathConfigured(string path, string value)

diff --git a/src/HealthChecks.UI.K8s.Operator/Handlers/NotificationHandler.cs b/src/HealthChecks.UI.K8s.Operator/Handlers/NotificationHandler.cs
@@ -0,0 +1,44 @@
+using k8s;
+using k8s.Models;
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Net.Http;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace HealthChecks.UI.K8s.Operator.Handlers
+{
+    public class NotificationHandler
+    {
+        private readonly IKubernetes _client;
+        private readonly IHttpClientFactory _httpClientFactory;
+        private readonly ILogger<K8sOperator> _logger;
+
+        public NotificationHandler(IKubernetes client, IHttpClientFactory httpClientFactory, ILogger<K8sOperator> logger)
+        {
+            _client = client ?? throw new ArgumentNullException(nameof(client));
+            _httpClientFactory = httpClientFactory ?? throw new ArgumentNullException(nameof(httpClientFactory));
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+        }
+        public async Task NotifyDiscoveredServiceAsync(WatchEventType type, V1Service service, HealthCheckResource resource)
+        {
+            var uiService = await _client.ListNamespacedOwnedServiceAsync(resource.Metadata.NamespaceProperty, resource.Metadata.Uid);
+            var secret = await _client.ListNamespacedOwnedSecretAsync(resource.Metadata.NamespaceProperty, resource.Metadata.Uid);
+
+            if (!service.Metadata.Labels.ContainsKey(resource.Spec.ServicesLabel))
+            {
+                type = WatchEventType.Deleted;
+            }
+
+            await HealthChecksPushService.PushNotification(
+                type,
+                resource,
+                uiService,
+                service,
+                secret,
+                _logger,
+                _httpClientFactory);
+        }
+    }
+}
diff --git a/src/HealthChecks.UI.K8s.Operator/Operator/ClusterServiceWatcher.cs b/src/HealthChecks.UI.K8s.Operator/Operator/ClusterServiceWatcher.cs
@@ -0,0 +1,65 @@
+using HealthChecks.UI.K8s.Operator.Diagnostics;
+using HealthChecks.UI.K8s.Operator.Handlers;
+using k8s;
+using k8s.Models;
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Net.Http;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace HealthChecks.UI.K8s.Operator.Operator
+{
+    internal class ClusterServiceWatcher
+    {
+        private readonly IKubernetes _client;
+        private readonly ILogger<K8sOperator> _logger;
+        private readonly OperatorDiagnostics _diagnostics;
+        private readonly NotificationHandler _notificationHandler;
+        private Watcher<V1Service> _watcher;
+        public ClusterServiceWatcher(
+          IKubernetes client,
+          ILogger<K8sOperator> logger,
+          OperatorDiagnostics diagnostics,
+          NotificationHandler notificationHandler,
+          IHttpClientFactory httpClientFactory)
+        {
+            _client = client ?? throw new ArgumentNullException(nameof(client));
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+            _diagnostics = diagnostics ?? throw new ArgumentNullException(nameof(diagnostics));
+            _notificationHandler = notificationHandler ?? throw new ArgumentNullException(nameof(notificationHandler));
+        }
+
+        internal Task Watch(HealthCheckResource resource, CancellationToken token)
+        {
+            var response = _client.ListServiceForAllNamespacesWithHttpMessagesAsync(
+                labelSelector: $"{resource.Spec.ServicesLabel}",
+                watch: true,
+                cancellationToken: token);
+
+            _watcher = response.Watch<V1Service, V1ServiceList>(
+                onEvent: async (type, item) => await _notificationHandler.NotifyDiscoveredServiceAsync(type, item, resource),
+                onError: e => _diagnostics.ServiceWatcherThrow(e)
+            );
+
+            _diagnostics.ServiceWatcherStarting("All");
+
+            return Task.CompletedTask;
+        }
+
+        internal void Stopwatch(HealthCheckResource resource)
+        {
+            Dispose();
+        }
+
+        public void Dispose()
+        {
+            if(_watcher != null && _watcher.Watching)
+            {
+                _watcher.Dispose();
+            }
+        }
+    }
+}
diff --git a/src/HealthChecks.UI.K8s.Operator/Operator/HealthChecksOperator.cs b/src/HealthChecks.UI.K8s.Operator/Operator/HealthChecksOperator.cs
@@ -1,12 +1,14 @@
 using HealthChecks.UI.K8s.Operator.Controller;
 using HealthChecks.UI.K8s.Operator.Diagnostics;
+using HealthChecks.UI.K8s.Operator.Operator;
 using k8s;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using System;
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
+using static HealthChecks.UI.K8s.Operator.Constants;
 
 namespace HealthChecks.UI.K8s.Operator
 {
@@ -15,7 +17,8 @@ internal class HealthChecksOperator : IHostedService
         private Watcher<HealthCheckResource> _watcher;
         private readonly IKubernetes _client;
         private readonly IHealthChecksController _controller;
-        private readonly HealthCheckServiceWatcher _serviceWatcher;
+        private readonly NamespacedServiceWatcher _serviceWatcher;
+        private readonly ClusterServiceWatcher _clusterServiceWatcher;
         private readonly OperatorDiagnostics _diagnostics;
         private readonly ILogger<K8sOperator> _logger;
         private readonly CancellationTokenSource _operatorCts = new CancellationTokenSource();
@@ -26,14 +29,16 @@ internal class HealthChecksOperator : IHostedService
         public HealthChecksOperator(
             IKubernetes client,
             IHealthChecksController controller,
-            HealthCheckServiceWatcher serviceWatcher,
+            NamespacedServiceWatcher serviceWatcher,
+            ClusterServiceWatcher clusterServiceWatcher,
             OperatorDiagnostics diagnostics,
             ILogger<K8sOperator> logger)
         {
 
             _client = client ?? throw new ArgumentNullException(nameof(client));
             _controller = controller ?? throw new ArgumentNullException(nameof(controller));
             _serviceWatcher = serviceWatcher ?? throw new ArgumentNullException(nameof(serviceWatcher));
+            _clusterServiceWatcher = clusterServiceWatcher ?? throw new ArgumentNullException(nameof(clusterServiceWatcher));
             _diagnostics = diagnostics ?? throw new ArgumentNullException(nameof(diagnostics));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
@@ -63,6 +68,7 @@ public Task StopAsync(CancellationToken cancellationToken)
             }
 
             _serviceWatcher.Dispose();
+            _clusterServiceWatcher.Dispose();
             _channel.Writer.Complete();
 
             return Task.CompletedTask;
@@ -111,13 +117,13 @@ private async Task OperatorListener()
                             {
                                 await _controller.DeployAsync(item.Resource);
                                 await WaitForAvailableReplicas(item.Resource);
-                                await _serviceWatcher.Watch(item.Resource, _operatorCts.Token);
+                                await StartServiceWatcher(item.Resource);
                             });
                         }
                         else if (item.EventType == WatchEventType.Deleted)
                         {
                             await _controller.DeleteDeploymentAsync(item.Resource);
-                            _serviceWatcher.Stopwatch(item.Resource);
+                            StopServiceWatcher(item.Resource);
                         }
                     }
                     catch (Exception ex)
@@ -128,6 +134,36 @@ private async Task OperatorListener()
             }
         }
 
+        private async Task StartServiceWatcher(HealthCheckResource resource)
+        {
+            Func<Task> startWatcher = async () => await _serviceWatcher.Watch(resource, _operatorCts.Token);
+            Func<Task> startClusterWatcher = async () => await _clusterServiceWatcher.Watch(resource, _operatorCts.Token);
+
+            var start = resource.Spec.Scope switch
+            {
+                Deployment.Scope.Namespaced => startWatcher,
+                Deployment.Scope.Cluster => startClusterWatcher,
+                _ => throw new ArgumentOutOfRangeException(nameof(resource.Spec.Scope))
+            };
+
+            await start();
+        }
+
+        private void StopServiceWatcher(HealthCheckResource resource)
+        {
+            Action stopWatcher = () => _serviceWatcher.Stopwatch(resource);
+            Action stopClusterWatcher = () => _clusterServiceWatcher.Stopwatch(resource);
+
+            var stop = resource.Spec.Scope switch
+            {
+                Deployment.Scope.Namespaced => stopWatcher,
+                Deployment.Scope.Cluster => stopClusterWatcher,
+                _ => throw new ArgumentOutOfRangeException(nameof(resource.Spec.Scope))
+            };
+
+            stop();
+        }
+
         private async Task WaitForAvailableReplicas(HealthCheckResource resource)
         {
             int retries = 1;

diff --git a/src/HealthChecks.UI.K8s.Operator/Operator/KubernetesAddressFactory.cs b/src/HealthChecks.UI.K8s.Operator/Operator/KubernetesAddressFactory.cs
@@ -10,7 +10,12 @@ public static string CreateAddress(V1Service service, HealthCheckResource resour
         {
             var defaultPort = int.Parse(resource.Spec.PortNumber ?? Constants.DefaultPort);
             var port = GetServicePort(service)?.Port ?? defaultPort;
-            var address = service.Spec.ClusterIP;
+            var address = GetLoadBalancerAddress(service);
+
+            if(string.IsNullOrEmpty(address))
+            {
+                address = service.Spec.ClusterIP;
+            }
 
             string healthScheme = resource.Spec.HealthChecksScheme;
 

diff --git a/...tor/Operator/HealthCheckServiceWatcher.cs → ...ator/Operator/NamespacedServiceWatcher.cs b/...tor/Operator/HealthCheckServiceWatcher.cs → ...ator/Operator/NamespacedServiceWatcher.cs
@@ -1,4 +1,5 @@
 using HealthChecks.UI.K8s.Operator.Diagnostics;
+using HealthChecks.UI.K8s.Operator.Handlers;
 using k8s;
 using k8s.Models;
 using Microsoft.Extensions.Logging;
@@ -11,24 +12,27 @@
 
 namespace HealthChecks.UI.K8s.Operator
 {
-    internal class HealthCheckServiceWatcher : IDisposable
+    internal class NamespacedServiceWatcher : IDisposable
     {
         private readonly IKubernetes _client;
         private readonly ILogger<K8sOperator> _logger;
         private readonly OperatorDiagnostics _diagnostics;
+        private readonly NotificationHandler _notificationHandler;
         private readonly IHttpClientFactory _httpClientFactory;
         private Dictionary<HealthCheckResource, Watcher<V1Service>> _watchers = new Dictionary<HealthCheckResource, Watcher<V1Service>>();
 
-        public HealthCheckServiceWatcher(
+        public NamespacedServiceWatcher(
             IKubernetes client,
             ILogger<K8sOperator> logger,
             OperatorDiagnostics diagnostics, 
+            NotificationHandler notificationHandler,
             IHttpClientFactory httpClientFactory)
         {
             _client = client ?? throw new ArgumentNullException(nameof(client));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
             _diagnostics = diagnostics ?? throw new ArgumentNullException(nameof(diagnostics));
             _httpClientFactory = httpClientFactory ?? throw new ArgumentNullException(nameof(httpClientFactory));
+            _notificationHandler = notificationHandler ?? throw new ArgumentNullException(nameof(notificationHandler));
         }
 
         internal Task Watch(HealthCheckResource resource, CancellationToken token)
@@ -44,7 +48,7 @@ internal Task Watch(HealthCheckResource resource, CancellationToken token)
                     cancellationToken: token);
 
                 var watcher = response.Watch<V1Service, V1ServiceList>(
-                    onEvent: async (type, item) => await OnServiceDiscoveredAsync(type, item, resource),
+                    onEvent: async (type, item) => await _notificationHandler.NotifyDiscoveredServiceAsync(type, item, resource),
                     onError: e => _diagnostics.ServiceWatcherThrow(e)
                 );
 
@@ -98,11 +102,5 @@ public void Dispose()
                 if (w != null && w.Watching) w.Dispose();
             });
         }
-
-        private class ServiceWatch
-        {
-            WatchEventType EventType { get; set; }
-            V1Service Service { get; set; }
-        }
     }
 }
diff --git a/src/HealthChecks.UI.K8s.Operator/Program.cs b/src/HealthChecks.UI.K8s.Operator/Program.cs
@@ -8,6 +8,7 @@
 using Serilog;
 using Serilog.Events;
 using HealthChecks.UI.K8s.Operator.Diagnostics;
+using HealthChecks.UI.K8s.Operator.Operator;
 
 namespace HealthChecks.UI.K8s.Operator
 {
@@ -49,7 +50,9 @@ public static IHostBuilder CreateHostBuilder(string[] args) =>
                 .AddSingleton<ServiceHandler>()
                 .AddSingleton<SecretHandler>()
                 .AddSingleton<ConfigMaphandler>()
-                .AddSingleton<HealthCheckServiceWatcher>();
+                .AddSingleton<NotificationHandler>()
+                .AddSingleton<NamespacedServiceWatcher>()
+                .AddSingleton<ClusterServiceWatcher>();
 
             }).ConfigureLogging((context, builder) =>
             {