added tests

Signed-off-by: wind57 <[email protected]>

Author: wind57

spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionProperties.java

@@ -41,6 +41,10 @@
  *      the lock to become the leader. In our current code,
  *      this is what we use in LeaderInitiator::start,
  *      more exactly in the scheduleAtFixRate
+ * restartOnFailure: what to do when leader election future fails
+ *      with an Exception. Do we restart the leader election process
+ *      or let it fail and thus this instance never participates
+ *      in the leader election process.
  *
  *
  * First, we try to acquire the lock (lock is either a configmap or a lease)
@@ -84,7 +88,8 @@ public record LeaderElectionProperties(
 	@DefaultValue("spring-k8s-leader-election-lock") String lockName,
 	@DefaultValue("10s") Duration renewDeadline,
 	@DefaultValue("2s") Duration retryPeriod,
-	@DefaultValue("0s") Duration waitAfterRenewalFailure,
-	@DefaultValue("false") boolean useConfigMapAsLock) {
+	@DefaultValue("3s") Duration waitAfterRenewalFailure,
+	@DefaultValue("false") boolean useConfigMapAsLock,
+	@DefaultValue("true") boolean restartOnFailure) {
 // @formatter:on
 }

spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionPropertiesTests.java

@@ -44,6 +44,7 @@ void testDefaults() {
 			Assertions.assertThat(properties.retryPeriod()).isEqualTo(Duration.ofSeconds(2));
 			Assertions.assertThat(properties.waitAfterRenewalFailure()).isEqualTo(Duration.ofSeconds(0));
 			Assertions.assertThat(properties.useConfigMapAsLock()).isFalse();
+			Assertions.assertThat(properties.restartOnFailure()).isTrue();
 		});
 	}
 
@@ -58,7 +59,8 @@ void testNonDefaults() {
 					"spring.cloud.kubernetes.leader.election.renew-deadline=2d",
 					"spring.cloud.kubernetes.leader.election.retry-period=3m",
 					"spring.cloud.kubernetes.leader.election.wait-after-renewal-failure=13m",
-					"spring.cloud.kubernetes.leader.election.use-config-map-as-lock=true")
+					"spring.cloud.kubernetes.leader.election.use-config-map-as-lock=true",
+					"spring.cloud.kubernetes.leader.election.restart-on-failure=false")
 			.run(context -> {
 				LeaderElectionProperties properties = context.getBean(LeaderElectionProperties.class);
 				Assertions.assertThat(properties).isNotNull();
@@ -71,6 +73,7 @@ void testNonDefaults() {
 				Assertions.assertThat(properties.retryPeriod()).isEqualTo(Duration.ofMinutes(3));
 				Assertions.assertThat(properties.waitAfterRenewalFailure()).isEqualTo(Duration.ofMinutes(13));
 				Assertions.assertThat(properties.useConfigMapAsLock()).isTrue();
+				Assertions.assertThat(properties.restartOnFailure()).isFalse();
 			});
 	}
 

spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/HealthEventListenerConfiguration.java

@@ -38,7 +38,7 @@ HealthEventListener healthEventListener() {
 		return new HealthEventListener();
 	}
 
-	private static class HealthEventListener implements ApplicationListener<InstanceRegisteredEvent<?>> {
+	private static final class HealthEventListener implements ApplicationListener<InstanceRegisteredEvent<?>> {
 
 		@Override
 		public void onApplicationEvent(InstanceRegisteredEvent<?> event) {

spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInitiator.java

@@ -19,12 +19,10 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
 import java.util.function.BooleanSupplier;
 
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfig;
-import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
 
@@ -33,6 +31,10 @@
 import org.springframework.core.log.LogAccessor;
 
 import static java.util.concurrent.Executors.newSingleThreadExecutor;
+import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.attachStatusLoggerPipeline;
+import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.blockReadinessCheck;
+import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.leaderElector;
+import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.sleep;
 
 /**
  * @author wind57
@@ -104,7 +106,7 @@ void postConstruct() {
 		// and don't block the main application from starting
 		podReadyWaitingExecutor.execute(() -> {
 			if (waitForPodReady) {
-				CompletableFuture<?> ready = attachStatusLoggerPipeline();
+				CompletableFuture<?> ready = attachStatusLoggerPipeline(podReadyFuture, candidateIdentity);
 				blockReadinessCheck(ready);
 				startLeaderElection();
 			}
@@ -157,69 +159,37 @@ private void startLeaderElection() {
 					if (!destroyCalled) {
 						LOG.warn(() -> "renewal failed for  : " + candidateIdentity + ", will re-start it after : "
 								+ leaderElectionProperties.waitAfterRenewalFailure().toSeconds() + " seconds");
-						sleep();
-						startLeaderElection();
+						sleep(leaderElectionProperties);
+						podReadyWaitingExecutor.execute(this::startLeaderElection);
 					}
 				}
 				else {
 					LOG.warn(() -> "leader failed with : " + error.getMessage());
-					LOG.warn(() -> "leader election is over for : " + candidateIdentity);
+					if (leaderElectionProperties.restartOnFailure()) {
+						LOG.info(() -> "will restart leader election for : " + candidateIdentity);
+						sleep(leaderElectionProperties);
+						podReadyWaitingExecutor.execute(this::startLeaderElection);
+					}
+					else {
+						LOG.warn(() -> "leader election is over for : " + candidateIdentity);
+					}
 				}
 			}
 			else {
 				// ok is always null; since it does not represent anything, it just passes
 				// the state further
 				LOG.info(() -> "leaderFuture finished normally, will re-start it for : " + candidateIdentity);
-				startLeaderElection();
+				podReadyWaitingExecutor.execute(this::startLeaderElection);
 			}
 		});
 
 		try {
 			leaderFuture.get();
 		}
 		catch (Exception e) {
-			LOG.warn(() -> "leader election failed for : " + candidateIdentity + ". Trying to recover...");
+			LOG.warn(() -> "leader election failed for : " + candidateIdentity);
 		}
 
 	}
 
-	private LeaderElector leaderElector(LeaderElectionConfig config, KubernetesClient fabric8KubernetesClient) {
-		return fabric8KubernetesClient.leaderElector().withConfig(config).build();
-	}
-
-	private void sleep() {
-		try {
-			TimeUnit.SECONDS.sleep(leaderElectionProperties.waitAfterRenewalFailure().toSeconds());
-		}
-		catch (InterruptedException e) {
-			throw new RuntimeException(e);
-		}
-	}
-
-	private void blockReadinessCheck(CompletableFuture<?> ready) {
-		try {
-			ready.get();
-		}
-		catch (Exception e) {
-			throw new RuntimeException(e);
-		}
-	}
-
-	/**
-	 * 	if 'ready' is already completed at this point, thread will run this,
-	 * 	otherwise it will attach the pipeline and move on to
-	 * 	'blockReadinessCheck'
-	 */
-	private CompletableFuture<?> attachStatusLoggerPipeline() {
-		return podReadyFuture.whenComplete((ok, error) -> {
-			if (error != null) {
-				LOG.error(() -> "readiness failed for : " + candidateIdentity
-					+ ", leader election will not start");
-			}
-			else {
-				LOG.info(() -> candidateIdentity + " is ready");
-			}
-		});
-	}
-
 }

spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInitiatorUtil.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2013-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.kubernetes.fabric8.leader.election;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfig;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector;
+
+import org.springframework.cloud.kubernetes.commons.leader.election.LeaderElectionProperties;
+import org.springframework.core.log.LogAccessor;
+
+final class Fabric8LeaderElectionInitiatorUtil {
+
+	private static final LogAccessor LOG = new LogAccessor(Fabric8LeaderElectionInitiatorUtil.class);
+
+	private Fabric8LeaderElectionInitiatorUtil() {
+
+	}
+
+	/**
+	 * if 'ready' is already completed at this point, thread will run this, otherwise it
+	 * will attach the pipeline and move on to 'blockReadinessCheck'.
+	 */
+	static CompletableFuture<?> attachStatusLoggerPipeline(CompletableFuture<?> innerPodReadyFuture,
+			String candidateIdentity) {
+		return innerPodReadyFuture.whenComplete((ok, error) -> {
+			if (error != null) {
+				LOG.error(() -> "readiness failed for : " + candidateIdentity + ", leader election will not start");
+			}
+			else {
+				LOG.info(() -> candidateIdentity + " is ready");
+			}
+		});
+	}
+
+	static void sleep(LeaderElectionProperties leaderElectionProperties) {
+		try {
+			TimeUnit.SECONDS.sleep(leaderElectionProperties.waitAfterRenewalFailure().toSeconds());
+		}
+		catch (InterruptedException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	static LeaderElector leaderElector(LeaderElectionConfig config, KubernetesClient fabric8KubernetesClient) {
+		return fabric8KubernetesClient.leaderElector().withConfig(config).build();
+	}
+
+	static void blockReadinessCheck(CompletableFuture<?> ready) {
+		try {
+			ready.get();
+		}
+		catch (Exception e) {
+			LOG.error(e, () -> "block readiness check failed with : " + e.getMessage());
+		}
+	}
+
+}

spring-cloud-kubernetes-integration-tests/spring-cloud-kubernetes-fabric8-leader-election/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Assertions.java

@@ -32,35 +32,40 @@ private Assertions() {
 
 	}
 
-	static void assertAcquireAndRenew(CapturedOutput output, Supplier<Lease> leaseSupplier, String name) {
+	/**
+	 * lease was acquired and we renewed it, at least once.
+	 */
+	static void assertAcquireAndRenew(CapturedOutput output, Supplier<Lease> leaseSupplier, String candidateIdentity) {
 		// we have become the leader
-		awaitUntil(60, 100, () -> output.getOut().contains(name + " is the new leader"));
+		awaitUntil(60, 100, () -> output.getOut().contains(candidateIdentity + " is the new leader"));
 
 		// let's unwind some logs to see that the process is how we expect it to be
 
 		// 1. lease is used as the lock (comes from our code)
 		assertThat(output.getOut()).contains("will use lease as the lock for leader election");
 
 		// 2. we start leader initiator for our hostname (comes from our code)
-		assertThat(output.getOut()).contains("starting leader initiator : " + name);
+		assertThat(output.getOut()).contains("starting leader initiator : " + candidateIdentity);
 
 		// 3. we try to acquire the lease (comes from fabric8 code)
 		assertThat(output.getOut()).contains(
-			"Attempting to acquire leader lease 'LeaseLock: default - spring-k8s-leader-election-lock (" + name + ")'");
+			"Attempting to acquire leader lease 'LeaseLock: default - spring-k8s-leader-election-lock " +
+				"(" + candidateIdentity + ")'");
 
 		// 4. lease has been acquired
 		assertThat(output.getOut())
-			.contains("Acquired lease 'LeaseLock: default - spring-k8s-leader-election-lock (" + name + ")'");
+			.contains("Acquired lease 'LeaseLock: default - spring-k8s-leader-election-lock " +
+				"(" + candidateIdentity + ")'");
 
 		// 5. we are the leader (comes from fabric8 code)
-		assertThat(output.getOut()).contains("Leader changed from null to " + name);
+		assertThat(output.getOut()).contains("Leader changed from null to " + candidateIdentity);
 
 		// 6. wait until a renewal happens (comes from fabric code)
 		// this one means that we have extended our leadership
 		awaitUntil(30, 500,
 			() -> output.getOut()
 				.contains("Attempting to renew leader lease 'LeaseLock: "
-					+ "default - spring-k8s-leader-election-lock (" + name + ")'"));
+					+ "default - spring-k8s-leader-election-lock (" + candidateIdentity + ")'"));
 
 		Lease lease = leaseSupplier.get();
 
@@ -79,4 +84,5 @@ static void assertAcquireAndRenew(CapturedOutput output, Supplier<Lease> leaseSu
 		});
 	}
 
+
 }

spring-cloud-kubernetes-integration-tests/spring-cloud-kubernetes-fabric8-leader-election/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionCompletedExceptionallyAndRestartedIT.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2013-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.kubernetes.fabric8.leader.election;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.springframework.cloud.kubernetes.fabric8.leader.election.Assertions.assertAcquireAndRenew;
+import static org.springframework.cloud.kubernetes.integration.tests.commons.Awaitilities.awaitUntil;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * <pre>
+ *     - we acquire the leadership
+ * </pre>
+ *
+ * @author wind57
+ */
+@TestPropertySource(properties = { "spring.cloud.kubernetes.leader.election.wait-for-pod-ready=true",
+	"spring.cloud.kubernetes.leader.election.restart-on-failure=true", "readiness.passes=true" })
+class Fabric8LeaderElectionCompletedExceptionallyAndRestartedIT extends AbstractLeaderElection {
+
+	private static final String NAME = "acquired-then-fails";
+
+	@Autowired
+	private Fabric8LeaderElectionInitiator initiator;
+
+	@BeforeAll
+	static void beforeAll() {
+		AbstractLeaderElection.beforeAll(NAME);
+	}
+
+	@Test
+	void test(CapturedOutput output) {
+
+		assertAcquireAndRenew(output, this::getLease, NAME);
+
+		initiator.leaderFeature().completeExceptionally(new RuntimeException("we kill the leadership future"));
+
+		// from the callback
+		awaitUntil(5, 50, () -> output.getOut().contains("id : acquired-then-fails stopped being a leader"));
+
+		awaitUntil(5, 50, () -> output.getOut().contains("leader failed with : we kill the leadership future"));
+
+		awaitUntil(5, 50, () -> output.getOut().contains("leader election failed for : acquired-then-fails"));
+
+		int afterLeaderFailure = output.getOut().indexOf("leader election failed for : acquired-then-fails");
+
+		afterLeaderFailure(afterLeaderFailure, output);
+	}
+
+	private void afterLeaderFailure(int afterLeaderFailure, CapturedOutput output) {
+		awaitUntil(60, 100, () -> output.getOut().substring(afterLeaderFailure).
+			contains(NAME + " is the new leader"));
+
+		// lease has been re-acquired
+		assertThat(output.getOut().substring(afterLeaderFailure))
+			.contains("Acquired lease 'LeaseLock: default - spring-k8s-leader-election-lock " + "(" + NAME + ")'");
+
+		// renewal happens (comes from fabric code)
+		// this one means that we have extended our leadership
+		awaitUntil(30, 500,
+			() -> output.getOut().substring(afterLeaderFailure)
+				.contains("Attempting to renew leader lease 'LeaseLock: "
+					+ "default - spring-k8s-leader-election-lock (" + NAME + ")'"));
+	}
+
+}