leader to give up leadership lock on graceful shutdown

Author: himanshug

extended/src/main/java/io/kubernetes/client/extended/leaderelection/LeaderElector.java

@@ -262,21 +262,44 @@ private boolean tryAcquireOrRenew() {
         return false;
       }
 
+      if (log.isDebugEnabled()) {
+        log.debug("Lock not found, try to create it");
+      }
+
+      // No Lock resource exists, try to get leadership by creating it
       return createLock(lock, leaderElectionRecord);
     }
 
+    // alright, we have an existing lock resource
+    // 1. Is Lock Empty? --> try to get leadership by updating it
+    // 2. Am I the Leader? --> update info and renew lease by updating it
+    // 3. I am not the Leader?
+    // 3.1 is Lock expired? --> try to get leadership by updating it
+    // 3.2 Lock not expired? --> update info, try later
+
     if (oldLeaderElectionRecord == null
         || oldLeaderElectionRecord.getAcquireTime() == null
         || oldLeaderElectionRecord.getRenewTime() == null
         || oldLeaderElectionRecord.getHolderIdentity() == null) {
-      return createLock(lock, leaderElectionRecord);
+      // We found the lock resource with an empty LeaderElectionRecord, try to get leadership by updating it
+      if (log.isDebugEnabled()) {
+        log.debug("Update lock to get lease");
+      }
+
+      if (oldLeaderElectionRecord != null) {
+        // maintain the leaderTransitions
+        leaderElectionRecord.setLeaderTransitions(oldLeaderElectionRecord.getLeaderTransitions() + 1);
+      }
+
+      return updateLock(lock, leaderElectionRecord);
     }
 
-    // 2. Record obtained, check the Identity & Time
+    // 2. Record obtained with LeaderElectionRecord, check the Identity & Time
     if (!oldLeaderElectionRecord.equals(this.observedRecord)) {
       this.observedRecord = oldLeaderElectionRecord;
       this.observedTimeMilliSeconds = System.currentTimeMillis();
     }
+
     if (observedTimeMilliSeconds + config.getLeaseDuration().toMillis() > now.getTime()
         && !isLeader()) {
       if (log.isDebugEnabled()) {
@@ -296,26 +319,20 @@ private boolean tryAcquireOrRenew() {
       leaderElectionRecord.setLeaderTransitions(oldLeaderElectionRecord.getLeaderTransitions() + 1);
     }
 
-    // update the lock itself
     if (log.isDebugEnabled()) {
-      log.debug("Update lock acquire time to keep lease");
+      log.debug("Update lock to renew lease");
     }
-    boolean updateSuccess = config.getLock().update(leaderElectionRecord);
-    if (!updateSuccess) {
-      return false;
-    }
-    this.observedRecord = leaderElectionRecord;
-    this.observedTimeMilliSeconds = System.currentTimeMillis();
-    if (log.isDebugEnabled()) {
+
+    boolean renewalStatus = updateLock(lock, leaderElectionRecord);
+
+    if (renewalStatus && log.isDebugEnabled()) {
       log.debug("TryAcquireOrRenew return success");
     }
-    return true;
+
+    return renewalStatus;
   }
 
   private boolean createLock(Lock lock, LeaderElectionRecord leaderElectionRecord) {
-    if (log.isDebugEnabled()) {
-      log.debug("Lock not found, try to create it");
-    }
     boolean createSuccess = lock.create(leaderElectionRecord);
     if (!createSuccess) {
       return false;
@@ -325,6 +342,16 @@ private boolean createLock(Lock lock, LeaderElectionRecord leaderElectionRecord)
     return true;
   }
 
+  private boolean updateLock(Lock lock, LeaderElectionRecord leaderElectionRecord) {
+    boolean updateSuccess = lock.update(leaderElectionRecord);
+    if (!updateSuccess) {
+      return false;
+    }
+    this.observedRecord = leaderElectionRecord;
+    this.observedTimeMilliSeconds = System.currentTimeMillis();
+    return true;
+  }
+
   private boolean isLeader() {
     return this.config.getLock().identity().equals(this.observedRecord.getHolderIdentity());
   }
@@ -345,8 +372,49 @@ private void maybeReportTransition() {
 
   @Override
   public void close() {
+    log.info("Closing...");
     scheduledWorkers.shutdownNow();
     leaseWorkers.shutdownNow();
     hookWorkers.shutdownNow();
+
+    // If I am the leader, free the lock so that other candidates can take it immediately
+    if (observedRecord != null && isLeader()) {
+
+      // First ensure that all executors have stopped
+      try {
+        boolean isTerminated = scheduledWorkers.awaitTermination(config.getRetryPeriod().getSeconds(), TimeUnit.SECONDS);
+        if (!isTerminated) {
+          log.warn("scheduledWorkers executor termination didn't finish.");
+          return;
+        }
+
+        isTerminated = leaseWorkers.awaitTermination(config.getRetryPeriod().getSeconds(), TimeUnit.SECONDS);
+        if (!isTerminated) {
+          log.warn("leaseWorkers executor termination didn't finish.");
+          return;
+        }
+
+        isTerminated = hookWorkers.awaitTermination(config.getRetryPeriod().getSeconds(), TimeUnit.SECONDS);
+        if (!isTerminated) {
+          log.warn("hookWorkers executor termination didn't finish.");
+          return;
+        }
+      } catch (InterruptedException ex) {
+        log.warn("Failed to ensure executors termination.", ex);
+        return;
+      }
+
+      log.info("Giving up the lock....");
+      LeaderElectionRecord emptyRecord = new LeaderElectionRecord();
+      // maintain leaderTransitions count
+      emptyRecord.setLeaderTransitions(observedRecord.getLeaderTransitions());
+      boolean status = this.config.getLock().update(emptyRecord);
+
+      if (!status) {
+        log.warn("Failed to give up the lock.");
+      }
+    }
+
+    log.info("Closed");
   }
 }

extended/src/main/java/io/kubernetes/client/extended/leaderelection/resourcelock/ConfigMapLock.java

@@ -59,6 +59,8 @@ public ConfigMapLock(String namespace, String name, String identity, ApiClient a
   @Override
   public LeaderElectionRecord get() throws ApiException {
     V1ConfigMap configMap = coreV1Client.readNamespacedConfigMap(name, namespace, null, null, null);
+    configMapRefer.set(configMap);
+
     Map<String, String> annotations = configMap.getMetadata().getAnnotations();
     if (annotations == null || annotations.isEmpty()) {
       configMap.getMetadata().setAnnotations(new HashMap<>());
@@ -74,7 +76,7 @@ public LeaderElectionRecord get() throws ApiException {
             .getApiClient()
             .getJSON()
             .deserialize(recordRawStringContent, LeaderElectionRecord.class);
-    configMapRefer.set(configMap);
+
     return record;
   }
 

extended/src/main/java/io/kubernetes/client/extended/leaderelection/resourcelock/EndpointsLock.java

@@ -59,6 +59,8 @@ public EndpointsLock(String namespace, String name, String identity, ApiClient a
   @Override
   public LeaderElectionRecord get() throws ApiException {
     V1Endpoints endpoints = coreV1Client.readNamespacedEndpoints(name, namespace, null, null, null);
+    endpointsRefer.set(endpoints);
+
     Map<String, String> annotations = endpoints.getMetadata().getAnnotations();
     if (annotations == null || annotations.isEmpty()) {
       endpoints.getMetadata().setAnnotations(new HashMap<>());
@@ -74,7 +76,6 @@ public LeaderElectionRecord get() throws ApiException {
             .getApiClient()
             .getJSON()
             .deserialize(recordRawStringContent, LeaderElectionRecord.class);
-    endpointsRefer.set(endpoints);
     return record;
   }
 

extended/src/test/java/io/kubernetes/client/extended/leaderelection/LeaderElectionTest.java

@@ -17,6 +17,8 @@
 import static org.mockito.Mockito.*;
 
 import io.kubernetes.client.openapi.ApiException;
+
+import java.net.HttpURLConnection;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Date;
@@ -412,9 +414,12 @@ public MockResourceLock(String iden) {
     }
 
     @Override
-    public LeaderElectionRecord get() {
+    public LeaderElectionRecord get() throws ApiException {
       lock.lock();
       try {
+        if (leaderRecord == null) {
+            throw new ApiException("Record Not Found", HttpURLConnection.HTTP_NOT_FOUND, null, null);
+        }
         return leaderRecord;
       } finally {
         lock.unlock();