started work

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

Author: wind57

spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/DeleteMeSandbox.java

@@ -16,25 +16,84 @@
 
 package org.springframework.cloud.kubernetes.commons.leader;
 
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Optional;
 import java.util.concurrent.locks.ReentrantLock;
 
 import org.springframework.cloud.kubernetes.commons.EnvReader;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.util.StringUtils;
 
+import static org.springframework.cloud.kubernetes.commons.KubernetesClientProperties.SERVICE_ACCOUNT_NAMESPACE_PATH;
+
 /**
  * @author wind57
  */
 public final class LeaderUtils {
 
+	/**
+	 * Coordination group for leader election.
+	 */
+	public static final String COORDINATION_GROUP = "coordination.k8s.io";
+
+	/**
+	 * Coordination version for leader election.
+	 */
+	public static final String COORDINATION_VERSION = "v1";
+
+	/**
+	 * Lease constant.
+	 */
+	public static final String LEASE = "Lease";
+
+	/**
+	 * Prefix for all properties related to leader election.
+	 */
+	public static final String LEADER_ELECTION_PROPERTY_PREFIX = "spring.cloud.kubernetes.leader.election";
+
+	/**
+	 * Property that controls whether leader election is enabled.
+	 */
+	public static final String LEADER_ELECTION_ENABLED_PROPERTY = LEADER_ELECTION_PROPERTY_PREFIX + ".enabled";
+
+	private static final String POD_NAMESPACE = "POD_NAMESPACE";
+
+	private static final LogAccessor LOG = new LogAccessor(LeaderUtils.class);
+
 	// k8s environment variable responsible for host name
 	private static final String HOSTNAME = "HOSTNAME";
 
 	private LeaderUtils() {
 
 	}
 
+	/**
+	 * ideally, should always be present. If not, downward api must enable this one.
+	 */
+	public static Optional<String> podNamespace() {
+		Path serviceAccountPath = new File(SERVICE_ACCOUNT_NAMESPACE_PATH).toPath();
+		boolean serviceAccountNamespaceExists = Files.isRegularFile(serviceAccountPath);
+		if (serviceAccountNamespaceExists) {
+			try {
+				String namespace = new String(Files.readAllBytes(serviceAccountPath)).replace(System.lineSeparator(),
+						"");
+				LOG.info(() -> "read namespace : " + namespace + " from service account " + serviceAccountPath);
+				return Optional.of(namespace);
+			}
+			catch (IOException e) {
+				throw new UncheckedIOException(e);
+			}
+
+		}
+		return Optional.ofNullable(EnvReader.getEnv(POD_NAMESPACE));
+	}
+
 	public static String hostName() throws UnknownHostException {
 		String hostName = EnvReader.getEnv(HOSTNAME);
 		if (StringUtils.hasText(hostName)) {

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

@@ -14,9 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.kubernetes.commons;
-
-import jakarta.annotation.Nonnull;
+package org.springframework.cloud.kubernetes.commons.leader.election;
 
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledFuture;
@@ -25,9 +23,13 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantLock;
 
+import jakarta.annotation.Nonnull;
+
 /**
  * This is taken from fabric8 with some minor changes (we need it, so it could be placed
- * in the common package).
+ * in the common package). A single thread scheduler that will shutdown itself when there
+ * are no more jobs running inside it. When all ScheduledFuture::cancel are called, the
+ * queue of tasks will be empty and there is an internal runnable that checks that.
  *
  * @author wind57
  */
@@ -70,7 +72,7 @@ private void startExecutor() {
 			this.executor = new ScheduledThreadPoolExecutor(1, threadFactory());
 			this.executor.setRemoveOnCancelPolicy(true);
 			this.executor.scheduleWithFixedDelay(this::shutdownCheck, this.ttlMillis, this.ttlMillis,
-				TimeUnit.MILLISECONDS);
+					TimeUnit.MILLISECONDS);
 		}
 
 	}

spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/CachedSingleThreadScheduler.java renamed to spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/CachedSingleThreadScheduler.java

@@ -0,0 +1,52 @@
+/*
+ * 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.commons.leader.election;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.boot.autoconfigure.condition.NoneNestedConditions;
+import org.springframework.context.annotation.Conditional;
+
+/**
+ * @author wind57
+ */
+@Target({ ElementType.TYPE, ElementType.METHOD })
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@Conditional(ConditionalOnLeaderElectionDisabled.OnLeaderElectionDisabled.class)
+public @interface ConditionalOnLeaderElectionDisabled {
+
+	class OnLeaderElectionDisabled extends NoneNestedConditions {
+
+		OnLeaderElectionDisabled() {
+			super(ConfigurationPhase.REGISTER_BEAN);
+		}
+
+		@ConditionalOnLeaderElectionEnabled
+		static class OnLeaderElectionDisabledClass {
+
+		}
+
+	}
+
+}

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

@@ -0,0 +1,43 @@
+/*
+ * 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.commons.leader.election;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+import static org.springframework.cloud.kubernetes.commons.leader.LeaderUtils.LEADER_ELECTION_ENABLED_PROPERTY;
+
+/**
+ * Provides a more succinct conditional for:
+ * <code>spring.cloud.kubernetes.leader.election.enabled</code>.
+ *
+ * @author wind57
+ */
+@Target({ ElementType.TYPE, ElementType.METHOD })
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@ConditionalOnProperty(value = LEADER_ELECTION_ENABLED_PROPERTY, havingValue = "true", matchIfMissing = false)
+public @interface ConditionalOnLeaderElectionEnabled {
+
+}

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

@@ -0,0 +1,87 @@
+/*
+ * 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.commons.leader.election;
+
+import java.net.UnknownHostException;
+import java.util.function.Consumer;
+
+import org.springframework.cloud.kubernetes.commons.leader.LeaderUtils;
+import org.springframework.cloud.kubernetes.commons.leader.election.events.NewLeaderEvent;
+import org.springframework.cloud.kubernetes.commons.leader.election.events.StartLeadingEvent;
+import org.springframework.cloud.kubernetes.commons.leader.election.events.StopLeadingEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.annotation.Bean;
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * common leader election callbacks that are supposed to be used in both fabric8 and
+ * k8s-native clients.
+ *
+ * @author wind57
+ */
+public class LeaderElectionCallbacks {
+
+	private static final LogAccessor LOG = new LogAccessor(LeaderElectionCallbacks.class);
+
+	@Bean
+	public final String holderIdentity() throws UnknownHostException {
+		String podHostName = LeaderUtils.hostName();
+		LOG.debug(() -> "using pod hostname : " + podHostName);
+		return podHostName;
+	}
+
+	@Bean
+	public final String podNamespace() {
+		String podNamespace = LeaderUtils.podNamespace().orElse("default");
+		LOG.debug(() -> "using pod namespace : " + podNamespace);
+		return podNamespace;
+	}
+
+	@Bean
+	public final Runnable onStartLeadingCallback(ApplicationEventPublisher applicationEventPublisher,
+			String holderIdentity, LeaderElectionProperties properties) {
+		return () -> {
+			LOG.info(() -> "id : " + holderIdentity + " is now a leader");
+			if (properties.publishEvents()) {
+				applicationEventPublisher.publishEvent(new StartLeadingEvent(holderIdentity));
+			}
+		};
+	}
+
+	@Bean
+	public final Runnable onStopLeadingCallback(ApplicationEventPublisher applicationEventPublisher,
+			String holderIdentity, LeaderElectionProperties properties) {
+		return () -> {
+			LOG.info(() -> "id : " + holderIdentity + " stopped being a leader");
+			if (properties.publishEvents()) {
+				applicationEventPublisher.publishEvent(new StopLeadingEvent(holderIdentity));
+			}
+		};
+	}
+
+	@Bean
+	public final Consumer<String> onNewLeaderCallback(ApplicationEventPublisher applicationEventPublisher,
+			LeaderElectionProperties properties) {
+		return holderIdentity -> {
+			LOG.info(() -> "id : " + holderIdentity + " is the new leader");
+			if (properties.publishEvents()) {
+				applicationEventPublisher.publishEvent(new NewLeaderEvent(holderIdentity));
+			}
+		};
+	}
+
+}