K8s client leader election

pom.xml

@@ -131,6 +131,7 @@
 		<module>spring-cloud-starter-kubernetes-fabric8-loadbalancer</module>
 		<module>spring-cloud-kubernetes-discovery</module>
 		<module>spring-cloud-starter-kubernetes-discoveryclient</module>
+		<module>spring-cloud-kubernetes-client-leader</module>
 	</modules>
 
 	<dependencyManagement>

spring-cloud-kubernetes-client-leader/pom.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-kubernetes</artifactId>
+		<version>5.0.1-SNAPSHOT</version>
+	</parent>
+
+	<artifactId>spring-cloud-kubernetes-client-leader</artifactId>
+	<name>K8s Client Spring Cloud Kubernetes :: Leader</name>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-kubernetes-client-autoconfig</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-configuration-processor</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-kubernetes-test-support</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-webtestclient</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.wiremock</groupId>
+			<artifactId>wiremock-standalone</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+</project>

spring-cloud-kubernetes-client-leader/src/main/java/org/springframework/cloud/kubernetes/client/leader/election/KubernetesClientLeaderElectionAutoConfiguration.java

@@ -0,0 +1,146 @@
+/*
+ * 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.client.leader.election;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+
+import io.kubernetes.client.extended.leaderelection.LeaderElectionConfig;
+import io.kubernetes.client.extended.leaderelection.Lock;
+import io.kubernetes.client.extended.leaderelection.resourcelock.ConfigMapLock;
+import io.kubernetes.client.extended.leaderelection.resourcelock.LeaseLock;
+import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.apis.CustomObjectsApi;
+import io.kubernetes.client.openapi.models.V1APIResource;
+import io.kubernetes.client.openapi.models.V1Pod;
+
+import org.springframework.boot.actuate.autoconfigure.info.ConditionalOnEnabledInfoContributor;
+import org.springframework.boot.actuate.info.InfoContributor;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnCloudPlatform;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.cloud.CloudPlatform;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.kubernetes.commons.leader.election.ConditionalOnLeaderElectionEnabled;
+import org.springframework.cloud.kubernetes.commons.leader.election.LeaderElectionProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.log.LogAccessor;
+
+import static org.springframework.cloud.kubernetes.commons.leader.LeaderUtils.COORDINATION_GROUP;
+import static org.springframework.cloud.kubernetes.commons.leader.LeaderUtils.COORDINATION_VERSION;
+
+/**
+ * @author wind57
+ */
+@Configuration(proxyBeanMethods = false)
+@EnableConfigurationProperties(LeaderElectionProperties.class)
+@ConditionalOnBean(ApiClient.class)
+@ConditionalOnLeaderElectionEnabled
+@ConditionalOnCloudPlatform(CloudPlatform.KUBERNETES)
+@AutoConfigureAfter(KubernetesClientLeaderElectionCallbacksAutoConfiguration.class)
+class KubernetesClientLeaderElectionAutoConfiguration {
+
+	private static final LogAccessor LOG = new LogAccessor(KubernetesClientLeaderElectionAutoConfiguration.class);
+
+	@Bean
+	@ConditionalOnClass(InfoContributor.class)
+	@ConditionalOnEnabledInfoContributor("leader.election")
+	KubernetesClientLeaderElectionInfoContributor kubernetesClientLeaderElectionInfoContributor(
+			String candidateIdentity, LeaderElectionConfig leaderElectionConfig) {
+		return new KubernetesClientLeaderElectionInfoContributor(candidateIdentity, leaderElectionConfig);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	KubernetesClientLeaderElectionInitiator kubernetesClientLeaderElectionInitiator(String candidateIdentity,
+			String podNamespace, LeaderElectionConfig leaderElectionConfig,
+			LeaderElectionProperties leaderElectionProperties, BooleanSupplier podReadySupplier,
+			KubernetesClientLeaderElectionCallbacks callbacks) {
+		return new KubernetesClientLeaderElectionInitiator(candidateIdentity, podNamespace, leaderElectionConfig,
+				leaderElectionProperties, podReadySupplier, callbacks);
+	}
+
+	@Bean
+	BooleanSupplier kubernetesClientPodReadySupplier(CoreV1Api coreV1Api, String candidateIdentity,
+			String podNamespace) {
+		return () -> {
+			try {
+				V1Pod pod = coreV1Api.readNamespacedPod(candidateIdentity, podNamespace).execute();
+				return isPodReady(pod);
+			}
+			catch (ApiException e) {
+				throw new RuntimeException(e);
+			}
+		};
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	LeaderElectionConfig kubernetesClientLeaderElectionConfig(LeaderElectionProperties properties, Lock lock) {
+		return new LeaderElectionConfig(lock, properties.leaseDuration(), properties.renewDeadline(),
+				properties.retryPeriod());
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	Lock kubernetesClientLeaderElectionLock(ApiClient apiClient, LeaderElectionProperties properties,
+			String candidateIdentity) {
+
+		CustomObjectsApi customObjectsApi = new CustomObjectsApi(apiClient);
+		boolean leaseSupported;
+		try {
+			List<V1APIResource> resources = customObjectsApi.getAPIResources(COORDINATION_GROUP, COORDINATION_VERSION)
+				.execute()
+				.getResources();
+
+			leaseSupported = resources.stream().map(V1APIResource::getKind).anyMatch("Lease"::equals);
+		}
+		catch (ApiException e) {
+			throw new RuntimeException(e);
+		}
+
+		if (leaseSupported) {
+			if (properties.useConfigMapAsLock()) {
+				LOG.info(() -> "leases are supported on the cluster, but config map will be used "
+						+ "(because 'spring.cloud.kubernetes.leader.election.use-config-map-as-lock=true')");
+				return new ConfigMapLock(properties.lockNamespace(), properties.lockName(), candidateIdentity);
+			}
+			else {
+				LOG.info(() -> "will use lease as the lock for leader election");
+				return new LeaseLock(properties.lockNamespace(), properties.lockName(), candidateIdentity, apiClient);
+			}
+		}
+		else {
+			LOG.info(() -> "will use configmap as the lock for leader election");
+			return new ConfigMapLock(properties.lockNamespace(), properties.lockName(), candidateIdentity, apiClient);
+		}
+	}
+
+	private boolean isPodReady(V1Pod pod) {
+		return pod != null && pod.getStatus() != null && pod.getStatus().getConditions() != null
+				&& pod.getStatus()
+					.getConditions()
+					.stream()
+					.anyMatch(c -> "Ready".equals(c.getType()) && "True".equals(c.getStatus()));
+	}
+
+}

spring-cloud-kubernetes-client-leader/src/main/java/org/springframework/cloud/kubernetes/client/leader/election/KubernetesClientLeaderElectionCallbacks.java

@@ -0,0 +1,26 @@
+/*
+ * 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.client.leader.election;
+
+import java.util.function.Consumer;
+
+/**
+ * @author wind57
+ */
+record KubernetesClientLeaderElectionCallbacks(Runnable onStartLeadingCallback, Runnable onStopLeadingCallback,
+		Consumer<String> onNewLeaderCallback) {
+}

spring-cloud-kubernetes-client-leader/src/main/java/org/springframework/cloud/kubernetes/client/leader/election/KubernetesClientLeaderElectionCallbacksAutoConfiguration.java

@@ -0,0 +1,55 @@
+/*
+ * 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.client.leader.election;
+
+import java.util.function.Consumer;
+
+import io.kubernetes.client.openapi.ApiClient;
+
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnCloudPlatform;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.cloud.CloudPlatform;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.kubernetes.client.KubernetesClientAutoConfiguration;
+import org.springframework.cloud.kubernetes.commons.leader.election.ConditionalOnLeaderElectionEnabled;
+import org.springframework.cloud.kubernetes.commons.leader.election.LeaderElectionCallbacks;
+import org.springframework.cloud.kubernetes.commons.leader.election.LeaderElectionProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author wind57
+ */
+@Configuration(proxyBeanMethods = false)
+@EnableConfigurationProperties(LeaderElectionProperties.class)
+@ConditionalOnBean(ApiClient.class)
+@ConditionalOnCloudPlatform(CloudPlatform.KUBERNETES)
+@ConditionalOnLeaderElectionEnabled
+@AutoConfigureAfter({ KubernetesClientAutoConfiguration.class })
+final class KubernetesClientLeaderElectionCallbacksAutoConfiguration extends LeaderElectionCallbacks {
+
+	@Bean
+	@ConditionalOnMissingBean
+	KubernetesClientLeaderElectionCallbacks kubernetesClientLeaderElectionCallbacks(Runnable onStartLeadingCallback,
+			Runnable onStopLeadingCallback, Consumer<String> onNewLeaderCallback) {
+		return new KubernetesClientLeaderElectionCallbacks(onStartLeadingCallback, onStopLeadingCallback,
+				onNewLeaderCallback);
+	}
+
+}

spring-cloud-kubernetes-client-leader/src/main/java/org/springframework/cloud/kubernetes/client/leader/election/KubernetesClientLeaderElectionInfoContributor.java

@@ -0,0 +1,63 @@
+/*
+ * 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.client.leader.election;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import io.kubernetes.client.extended.leaderelection.LeaderElectionConfig;
+import io.kubernetes.client.openapi.ApiException;
+
+import org.springframework.boot.actuate.info.Info;
+import org.springframework.boot.actuate.info.InfoContributor;
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * @author wind57
+ */
+final class KubernetesClientLeaderElectionInfoContributor implements InfoContributor {
+
+	private static final LogAccessor LOG = new LogAccessor(KubernetesClientLeaderElectionInfoContributor.class);
+
+	private final String candidateIdentity;
+
+	private final LeaderElectionConfig leaderElectionConfig;
+
+	KubernetesClientLeaderElectionInfoContributor(String candidateIdentity, LeaderElectionConfig leaderElectionConfig) {
+		this.candidateIdentity = candidateIdentity;
+		this.leaderElectionConfig = leaderElectionConfig;
+	}
+
+	@Override
+	public void contribute(Info.Builder builder) {
+		Map<String, Object> details = new HashMap<>();
+		try {
+			Optional.ofNullable(leaderElectionConfig.getLock().get()).ifPresentOrElse(leaderRecord -> {
+				boolean isLeader = candidateIdentity.equals(leaderRecord.getHolderIdentity());
+				details.put("leaderId", candidateIdentity);
+				details.put("isLeader", isLeader);
+			}, () -> details.put("leaderId", "Unknown"));
+		}
+		catch (ApiException e) {
+			LOG.error(e, "error in leader election info contributor");
+		}
+
+		builder.withDetail("leaderElection", details);
+	}
+
+}