leader election (#537)

* init leader leaderelection

* fix format

* remove unused import

* add config map

* fix space

* add multi lock

* lease lock

* cp case LeaderElection

* more test cases

* port all testcases from java

* fix space

* fix default timeout

* try to fix flasky  gh action by reducing renew deadline

* try debug failed gh action

Author: tg123

src/KubernetesClient/LeaderElection/ILock.cs

@@ -0,0 +1,46 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace k8s.LeaderElection
+{
+    /// <summary>
+    /// ILock offers a common interface for locking on arbitrary resources used in leader election. The Interface is used to hide the details on specific implementations in order to allow them to change over time.
+    /// </summary>
+    public interface ILock
+    {
+        /// <summary>
+        /// Get returns the LeaderElectionRecord
+        /// </summary>
+        /// <param name="cancellationToken">token to cancel the task</param>
+        /// <returns>the record</returns>
+        Task<LeaderElectionRecord> GetAsync(CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Create attempts to create a LeaderElectionRecord
+        /// </summary>
+        /// <param name="record">record to create</param>
+        /// <param name="cancellationToken">token to cancel the task</param>
+        /// <returns>true if created</returns>
+        Task<bool> CreateAsync(LeaderElectionRecord record, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Update will update and existing LeaderElectionRecord
+        /// </summary>
+        /// <param name="record">record to create</param>
+        /// <param name="cancellationToken">token to cancel the task</param>
+        /// <returns>true if updated</returns>
+        Task<bool> UpdateAsync(LeaderElectionRecord record, CancellationToken cancellationToken = default);
+
+
+        /// <summary>
+        /// the locks Identity
+        /// </summary>
+        string Identity { get; }
+
+        /// <summary>
+        /// Describe is used to convert details on current resource lock into a string
+        /// </summary>
+        /// <returns>resource lock description</returns>
+        string Describe();
+    }
+}

src/KubernetesClient/LeaderElection/LeaderElectionConfig.cs

@@ -0,0 +1,20 @@
+using System;
+
+namespace k8s.LeaderElection
+{
+    public class LeaderElectionConfig
+    {
+        public ILock Lock { get; set; }
+
+        public TimeSpan LeaseDuration { get; set; } = TimeSpan.FromSeconds(15);
+
+        public TimeSpan RenewDeadline { get; set; } = TimeSpan.FromSeconds(10);
+
+        public TimeSpan RetryPeriod { get; set; } = TimeSpan.FromSeconds(2);
+
+        public LeaderElectionConfig(ILock @lock)
+        {
+            Lock = @lock;
+        }
+    }
+}

src/KubernetesClient/LeaderElection/LeaderElectionRecord.cs

@@ -0,0 +1,74 @@
+using System;
+
+namespace k8s.LeaderElection
+{
+    /// <summary>
+    /// LeaderElectionRecord is the record that is stored in the leader election annotation.
+    /// This information should be used for observational purposes only and could be replaced with a random string (e.g. UUID) with only slight modification of this code.
+    /// </summary>
+    public class LeaderElectionRecord
+    {
+        /// <summary>
+        /// the ID that owns the lease. If empty, no one owns this lease and all callers may acquire.
+        /// </summary>
+        public string HolderIdentity { get; set; }
+
+        /// <summary>
+        /// LeaseDuration in seconds
+        /// </summary>
+        public int LeaseDurationSeconds { get; set; }
+
+        /// <summary>
+        /// acquire time
+        /// </summary>
+        // public DateTimeOffset? AcquireTime { get; set; }
+        public DateTime? AcquireTime { get; set; }
+
+        /// <summary>
+        /// renew time
+        /// </summary>
+        // public DateTimeOffset? RenewTime { get; set; }
+        public DateTime? RenewTime { get; set; }
+
+        /// <summary>
+        /// leader transitions
+        /// </summary>
+        public int LeaderTransitions { get; set; }
+
+        protected bool Equals(LeaderElectionRecord other)
+        {
+            return HolderIdentity == other?.HolderIdentity && Nullable.Equals(AcquireTime, other.AcquireTime) && Nullable.Equals(RenewTime, other.RenewTime);
+        }
+
+        public override bool Equals(object obj)
+        {
+            if (ReferenceEquals(null, obj))
+            {
+                return false;
+            }
+
+            if (ReferenceEquals(this, obj))
+            {
+                return true;
+            }
+
+            if (obj.GetType() != this.GetType())
+            {
+                return false;
+            }
+
+            return Equals((LeaderElectionRecord)obj);
+        }
+
+        public override int GetHashCode()
+        {
+            unchecked
+            {
+                var hashCode = (HolderIdentity != null ? HolderIdentity.GetHashCode() : 0);
+                hashCode = (hashCode * 397) ^ AcquireTime.GetHashCode();
+                hashCode = (hashCode * 397) ^ RenewTime.GetHashCode();
+                return hashCode;
+            }
+        }
+    }
+}

src/KubernetesClient/LeaderElection/LeaderElector.cs

@@ -0,0 +1,248 @@
+using System;
+using System.Net;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Rest;
+
+namespace k8s.LeaderElection
+{
+    public class LeaderElector : IDisposable
+    {
+        private const double JitterFactor = 1.2;
+
+        private readonly LeaderElectionConfig config;
+
+        /// <summary>
+        /// OnStartedLeading is called when a LeaderElector client starts leading
+        /// </summary>
+        public event Action OnStartedLeading;
+
+        /// <summary>
+        /// OnStoppedLeading is called when a LeaderElector client stops leading
+        /// </summary>
+        public event Action OnStoppedLeading;
+
+        /// <summary>
+        /// OnNewLeader is called when the client observes a leader that is
+        /// not the previously observed leader. This includes the first observed
+        /// leader when the client starts.
+        /// </summary>
+        public event Action<string> OnNewLeader;
+
+        private volatile LeaderElectionRecord observedRecord;
+        private DateTimeOffset observedTime = DateTimeOffset.MinValue;
+        private string reportedLeader;
+
+        public LeaderElector(LeaderElectionConfig config)
+        {
+            this.config = config;
+        }
+
+        public bool IsLeader()
+        {
+            return observedRecord?.HolderIdentity != null && observedRecord?.HolderIdentity == config.Lock.Identity;
+        }
+
+        public string GetLeader()
+        {
+            return observedRecord?.HolderIdentity;
+        }
+
+        public async Task RunAsync(CancellationToken cancellationToken = default)
+        {
+            await AcquireAsync(cancellationToken).ConfigureAwait(false);
+
+            try
+            {
+                OnStartedLeading?.Invoke();
+
+                // renew loop
+                for (; ; )
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+                    var acq = Task.Run(async () =>
+                    {
+                        try
+                        {
+                            while (!await TryAcquireOrRenew(cancellationToken).ConfigureAwait(false))
+                            {
+                                await Task.Delay(config.RetryPeriod, cancellationToken).ConfigureAwait(false);
+                                MaybeReportTransition();
+                            }
+                        }
+                        catch
+                        {
+                            // ignore
+                            return false;
+                        }
+
+                        return true;
+                    });
+
+
+                    if (await Task.WhenAny(acq, Task.Delay(config.RenewDeadline, cancellationToken))
+                        .ConfigureAwait(false) == acq)
+                    {
+                        var succ = await acq.ConfigureAwait(false);
+
+                        if (succ)
+                        {
+                            await Task.Delay(config.RetryPeriod, cancellationToken).ConfigureAwait(false);
+                            // retry
+                            continue;
+                        }
+
+                        // renew failed
+                    }
+
+                    // timeout
+                    break;
+                }
+            }
+            finally
+            {
+                OnStoppedLeading?.Invoke();
+            }
+        }
+
+        private async Task<bool> TryAcquireOrRenew(CancellationToken cancellationToken)
+        {
+            var l = config.Lock;
+            var leaderElectionRecord = new LeaderElectionRecord()
+            {
+                HolderIdentity = l.Identity,
+                LeaseDurationSeconds = config.LeaseDuration.Seconds,
+                AcquireTime = DateTime.UtcNow,
+                RenewTime = DateTime.UtcNow,
+                LeaderTransitions = 0,
+            };
+
+            // 1. obtain or create the ElectionRecord
+
+            LeaderElectionRecord oldLeaderElectionRecord = null;
+            try
+            {
+                oldLeaderElectionRecord = await l.GetAsync(cancellationToken).ConfigureAwait(false);
+            }
+            catch (HttpOperationException e)
+            {
+                if (e.Response.StatusCode != HttpStatusCode.NotFound)
+                {
+                    return false;
+                }
+            }
+
+            if (oldLeaderElectionRecord?.AcquireTime == null ||
+                oldLeaderElectionRecord?.RenewTime == null ||
+                oldLeaderElectionRecord?.HolderIdentity == null)
+            {
+                var created = await l.CreateAsync(leaderElectionRecord, cancellationToken).ConfigureAwait(false);
+                if (created)
+                {
+                    observedRecord = leaderElectionRecord;
+                    observedTime = DateTimeOffset.Now;
+                    return true;
+                }
+
+                return false;
+            }
+
+
+            // 2. Record obtained, check the Identity & Time
+            if (!Equals(observedRecord, oldLeaderElectionRecord))
+            {
+                observedRecord = oldLeaderElectionRecord;
+                observedTime = DateTimeOffset.Now;
+            }
+
+            if (!string.IsNullOrEmpty(oldLeaderElectionRecord.HolderIdentity)
+                && observedTime + config.LeaseDuration > DateTimeOffset.Now
+                && !IsLeader())
+            {
+                // lock is held by %v and has not yet expired", oldLeaderElectionRecord.HolderIdentity
+                return false;
+            }
+
+            // 3. We're going to try to update. The leaderElectionRecord is set to it's default
+            // here. Let's correct it before updating.
+            if (IsLeader())
+            {
+                leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime;
+                leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions;
+            }
+            else
+            {
+                leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1;
+            }
+
+            var updated = await l.UpdateAsync(leaderElectionRecord, cancellationToken).ConfigureAwait(false);
+            if (!updated)
+            {
+                return false;
+            }
+
+            observedRecord = leaderElectionRecord;
+            observedTime = DateTimeOffset.Now;
+
+            return true;
+        }
+
+        private async Task AcquireAsync(CancellationToken cancellationToken)
+        {
+            for (; ; )
+            {
+                try
+                {
+                    var delay = config.RetryPeriod.Milliseconds;
+                    var acq = TryAcquireOrRenew(cancellationToken);
+
+                    if (await Task.WhenAny(acq, Task.Delay(delay, cancellationToken))
+                        .ConfigureAwait(false) == acq)
+                    {
+                        if (await acq.ConfigureAwait(false))
+                        {
+                            return;
+                        }
+                    }
+
+                    delay = (int)(delay * JitterFactor);
+                }
+                finally
+                {
+                    MaybeReportTransition();
+                }
+            }
+        }
+
+        private void MaybeReportTransition()
+        {
+            if (observedRecord == null)
+            {
+                return;
+            }
+
+            if (observedRecord.HolderIdentity == reportedLeader)
+            {
+                return;
+            }
+
+            reportedLeader = observedRecord.HolderIdentity;
+
+            OnNewLeader?.Invoke(reportedLeader);
+        }
+
+        protected virtual void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+            }
+        }
+
+        /// <inheritdoc/>
+        public void Dispose()
+        {
+            Dispose(true);
+            GC.SuppressFinalize(this);
+        }
+    }
+}