feat(leader election): support leader election via V1Lease.

This closes #22. The leader election utilizes V1Lease objects
in the namespace that the operator is running in. If no object
exists, the instance is elected as leader. Later, the config
determines the interval in which the check occurs. When a leader
does not update the "RenewTime", another instance can take leadership.

Author: Christoph Bühler

src/KubeOps/Operator/Builder/OperatorBuilder.cs

@@ -6,12 +6,14 @@
 using KubeOps.Operator.Controller;
 using KubeOps.Operator.DevOps;
 using KubeOps.Operator.Finalizer;
+using KubeOps.Operator.Leader;
 using KubeOps.Operator.Queue;
 using KubeOps.Operator.Serialization;
 using KubeOps.Operator.Services;
 using KubeOps.Operator.Watcher;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Diagnostics.HealthChecks;
+using Microsoft.Rest.Serialization;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Converters;
 using Newtonsoft.Json.Serialization;
@@ -91,23 +93,22 @@ internal IOperatorBuilder AddOperatorBase(OperatorSettings settings)
             // support lazy service resolution
             Services.AddTransient(typeof(Lazy<>), typeof(LazyService<>));
 
-            Services.AddTransient(
-                _ => new JsonSerializerSettings
-                {
-                    ContractResolver = new NamingConvention(),
-                    Converters = new List<JsonConverter>
-                    {
-                        new StringEnumConverter { NamingStrategy = new CamelCaseNamingStrategy() },
-                    },
-                });
-            JsonConvert.DefaultSettings = () => new JsonSerializerSettings
+            var jsonSettings = new JsonSerializerSettings
             {
+                DateFormatHandling = DateFormatHandling.IsoDateFormat,
+                DateTimeZoneHandling = DateTimeZoneHandling.Utc,
+                NullValueHandling = NullValueHandling.Ignore,
+                ReferenceLoopHandling = ReferenceLoopHandling.Serialize,
                 ContractResolver = new NamingConvention(),
                 Converters = new List<JsonConverter>
                 {
                     new StringEnumConverter { NamingStrategy = new CamelCaseNamingStrategy() },
+                    new Iso8601TimeSpanConverter(),
                 },
+                DateFormatString = "yyyy'-'MM'-'dd'T'HH':'mm':'ss.ffffffK",
             };
+            Services.AddTransient(_ => jsonSettings);
+            JsonConvert.DefaultSettings = () => jsonSettings;
 
             Services.AddTransient(
                 _ => new SerializerBuilder()
@@ -126,15 +127,32 @@ internal IOperatorBuilder AddOperatorBase(OperatorSettings settings)
                 {
                     SerializationSettings =
                     {
+                        Formatting = Formatting.Indented,
+                        DateFormatHandling = DateFormatHandling.IsoDateFormat,
+                        DateTimeZoneHandling = DateTimeZoneHandling.Utc,
+                        NullValueHandling = NullValueHandling.Ignore,
+                        ReferenceLoopHandling = ReferenceLoopHandling.Serialize,
                         ContractResolver = new NamingConvention(),
                         Converters = new List<JsonConverter>
-                            { new StringEnumConverter { NamingStrategy = new CamelCaseNamingStrategy() } },
+                        {
+                            new StringEnumConverter { NamingStrategy = new CamelCaseNamingStrategy() },
+                            new Iso8601TimeSpanConverter(),
+                        },
+                        DateFormatString = "yyyy'-'MM'-'dd'T'HH':'mm':'ss.ffffffK",
                     },
                     DeserializationSettings =
                     {
+                        DateFormatHandling = DateFormatHandling.IsoDateFormat,
+                        DateTimeZoneHandling = DateTimeZoneHandling.Utc,
+                        NullValueHandling = NullValueHandling.Ignore,
+                        ReferenceLoopHandling = ReferenceLoopHandling.Serialize,
                         ContractResolver = new NamingConvention(),
                         Converters = new List<JsonConverter>
-                            { new StringEnumConverter { NamingStrategy = new CamelCaseNamingStrategy() } },
+                        {
+                            new StringEnumConverter { NamingStrategy = new CamelCaseNamingStrategy() },
+                            new Iso8601TimeSpanConverter(),
+                        },
+                        DateFormatString = "yyyy'-'MM'-'dd'T'HH':'mm':'ss.ffffffK",
                     },
                 });
 
@@ -151,6 +169,10 @@ internal IOperatorBuilder AddOperatorBase(OperatorSettings settings)
             // Add the default controller liveness check.
             AddHealthCheck<ControllerLivenessCheck>();
 
+            // Support for leader election via V1Leases.
+            Services.AddHostedService<LeaderElector>();
+            Services.AddSingleton<ILeaderElection, LeaderElection>();
+
             return this;
         }
     }

src/KubeOps/Operator/Client/IKubernetesClient.cs

@@ -40,6 +40,10 @@ public interface IKubernetesClient
         /// <returns>A string containing the current namespace (or a fallback of it).</returns>
         Task<string> GetCurrentNamespace(string downwardApiEnvName = "POD_NAMESPACE");
 
+        /// <summary>
+        /// Fetch and return the actual kubernetes <see cref="VersionInfo"/> (aka. Server Version).
+        /// </summary>
+        /// <returns>The <see cref="VersionInfo"/> of the current server.</returns>
         Task<VersionInfo> GetServerVersion();
 
         Task<TResource?> Get<TResource>(string name, string? @namespace = null)

src/KubeOps/Operator/Controller/ResourceControllerBase.cs

@@ -7,6 +7,7 @@
 using k8s.Models;
 using KubeOps.Operator.Client;
 using KubeOps.Operator.Finalizer;
+using KubeOps.Operator.Leader;
 using KubeOps.Operator.Queue;
 using KubeOps.Operator.Services;
 using Microsoft.Extensions.Logging;
@@ -39,22 +40,24 @@ protected ResourceControllerBase(IResourceServices<TEntity> services)
 
         protected IKubernetesClient Client => _services.Client;
 
-        public async Task StartAsync(CancellationToken cancellationToken)
+        public Task StartAsync(CancellationToken cancellationToken)
         {
             _logger.LogInformation(@"Startup CRD Controller for ""{resource}"".", typeof(TEntity));
-
-            _services.EventQueue.ResourceEvent += OnResourceEvent;
-            await _services.EventQueue.Start();
             _running = true;
+            _services.LeaderElection.LeadershipChange += LeadershipChanged;
+            LeadershipChanged(null, _services.LeaderElection.State);
+
+            return Task.CompletedTask;
         }
 
         public Task StopAsync(CancellationToken cancellationToken)
         {
             _logger.LogInformation(@"Shutdown CRD Controller for ""{resource}"".", typeof(TEntity));
 
-            _services.EventQueue.Stop();
-            _services.EventQueue.ResourceEvent -= OnResourceEvent;
+            LeadershipChanged(null, LeaderState.None);
+            _services.LeaderElection.LeadershipChange -= LeadershipChanged;
             _running = false;
+
             return Task.CompletedTask;
         }
 
@@ -116,6 +119,23 @@ protected Task AttachFinalizer<TFinalizer>(TEntity resource)
             return finalizer.Register(resource);
         }
 
+        private async void LeadershipChanged(object? sender, LeaderState state)
+        {
+            if (state == LeaderState.Leader)
+            {
+                _logger.LogInformation("This instance was elected as leader, starting event queue.");
+                _services.EventQueue.ResourceEvent += OnResourceEvent;
+                await _services.EventQueue.Start();
+
+                return;
+            }
+
+            _logger.LogInformation(
+                "This instance has either resigned from leadership, was elected candidate or is shutting down. Stopping event queue.");
+            await _services.EventQueue.Stop();
+            _services.EventQueue.ResourceEvent -= OnResourceEvent;
+        }
+
         private async void OnResourceEvent(object? _, (ResourceEventType Type, TEntity Resource) args)
         {
             var (type, resource) = args;

src/KubeOps/Operator/Leader/ILeaderElection.cs

@@ -0,0 +1,13 @@
+using System;
+
+namespace KubeOps.Operator.Leader
+{
+    public interface ILeaderElection
+    {
+        event EventHandler<LeaderState>? LeadershipChange;
+
+        LeaderState State { get; }
+
+        internal void LeadershipChanged(LeaderState state);
+    }
+}

src/KubeOps/Operator/Leader/LeaderElection.cs

@@ -0,0 +1,31 @@
+using System;
+using Microsoft.Extensions.Logging;
+
+namespace KubeOps.Operator.Leader
+{
+    internal class LeaderElection : ILeaderElection
+    {
+        private readonly ILogger<LeaderElection> _logger;
+
+        public LeaderElection(ILogger<LeaderElection> logger)
+        {
+            _logger = logger;
+        }
+
+        public event EventHandler<LeaderState>? LeadershipChange;
+
+        public LeaderState State { get; private set; } = LeaderState.None;
+
+        void ILeaderElection.LeadershipChanged(LeaderState state)
+        {
+            if (State == state)
+            {
+                return;
+            }
+
+            _logger.LogDebug("Leadership state changed to: {state}.", state);
+            State = state;
+            LeadershipChange?.Invoke(this, state);
+        }
+    }
+}