diff --git a/docs/modules/ROOT/pages/leader-election.adoc b/docs/modules/ROOT/pages/leader-election.adoc index bc0d920a77..42730e6c54 100644 --- a/docs/modules/ROOT/pages/leader-election.adoc +++ b/docs/modules/ROOT/pages/leader-election.adoc @@ -25,3 +25,89 @@ To specify the name of the configmap used for leader election use the following ---- spring.cloud.kubernetes.leader.config-map-name=leader ---- + + +''' + +There is another way you can configure leader election, and it comes with native support in the fabric8 library (k8s native client support is not yet implemented). In the long run, this will be the default way to configure leader election, while the previous one will be dropped. You can treat this one much like the JDKs "preview" features. + +To be able to use it, you need to set the property: + +[source] +---- +spring.cloud.kubernetes.leader.election.enabled=true +---- + +Unlike the old implementation, this one will use either the `Lease` _or_ `ConfigMap` as the lock, depending on your cluster version. You can force using configMap still, even if leases are supported, via : + +[source] +---- +spring.cloud.kubernetes.leader.election.use-config-map-as-lock=true +---- + +The name of that `Lease` or `ConfigMap` can be defined using the property (default value is `spring-k8s-leader-election-lock`): + +[source] +---- +spring.cloud.kubernetes.leader.election.lockName=other-name +---- + +The namespace where the lock is created (`default` being set if no explicit one exists) can be set also: + +[source] +---- +spring.cloud.kubernetes.leader.election.lockNamespace=other-namespace +---- + +Before the leader election process kicks in, you can wait until the pod is ready (via the readiness check). This is enabled by default, but you can disable it if needed: + +[source] +---- +spring.cloud.kubernetes.leader.election.waitForPodReady=false +---- + +Like with the old implementation, we will publish events by default, but this can be disabled: + +[source] +---- +spring.cloud.kubernetes.leader.election.publishEvents=false +---- + +There are a few parameters that control how the leader election process will happen. To explain them, we need to look at the high-level implementation of this process. All the candidate pods try to become the leader, or they try to _acquire_ the lock. If the lock is already taken, they will continue to retry to acquire it every `spring.cloud.kubernetes.leader.election.retryPeriod` (value is specified as `java.time.Duration`, and by default it is 2 seconds). + +If the lock is not taken, current pod becomes the leader. It does so by inserting a so-called "record" into the lock (`Lease` or `ConfigMap`). Among the things that the "record" contains, is the `leaseDuration` (that you can specify via `spring.cloud.kubernetes.leader.election.leaseDuration`; by default it is 15 seconds and is of type `java.time.Duration`). This acts like a TTL on the lock: no other candidate can acquire the lock, unless this period has expired (from the last renewal time). + +Once a certain pod establishes itself as the leader (by acquiring the lock), it will continuously (every `spring.cloud.kubernetes.leader.election.retryPeriod`) try to renew its lease, or in other words: it will try to extend its leadership. When a renewal happens, the "record" that is stored inside the lock, is updated. For example, `renewTime` is updated inside the record, to denote when the last renewal happened. (You can always peek inside these fields by using `kubectl describe lease...` for example). + +Renewal must happen within a certain interval, specified by `spring.cloud.kubernetes.leader.election.renewDeadline`. By default, it is equal to 10 seconds, and it means that the leader pod has a maximum of 10 seconds to renew its leadership. If that does not happen, this pod loses its leadership and leader election starts again. Because other pods try to become leaders every 2 seconds (by default), it could mean that the pod that just lost leadership, will become leader again. If you want other pods to have a higher chance of becoming leaders, you can set the property (specified in seconds, by default it is 0) : + +[source] +---- +spring.cloud.kubernetes.leader.election.wait-after-renewal-failure=3 +---- + +This will mean that the pod (that could not renew its lease) and lost leadership, will wait this many seconds, before trying to become leader again. + +Let's try to explain these settings based on an example: there are two pods that participate in leader election. For simplicity let's call them `podA` and `podB`. They both start at the same time: `12:00:00`, but `podA` establishes itself as the leader. This means that every two seconds (`retryPeriod`), `podB` will try to become the new leader. So at `12:00:02`, then at `12:00:04` and so on, it will basically ask : "Can I become the leader?". In our simplified example, the answer to that question can be answered based on `podA` activity. + +After `podA` has become the leader, at every 2 seconds, it will try to "extend" or _renew_ its leadership. So at `12:00:02`, then at `12:00:04` and so on, `podA` goes to the lock and updates its record to reflect that it is still the leader. Between the last successful renewal and the next one, it has exactly 10 seconds (`renewalDeadline`). If it fails to renew its leadership (there is a connection problem or a big GC pause, etc.) within those 10 seconds, it stops leading and `podB` can acquire the leadership now. When `podA` stops being a leader in a graceful way, the lock record is "cleared", basically meaning that `podB` can acquire leadership immediately. + +A different story happens when `podA` dies with an OutOfMemory for example, without being able to gracefully update lock record and this is when `leaseDuration` argument matters. The easiest way to explain is via an example: + +`podA` has renewed its leadership at `12:00:04`, but at `12:00:05` it has been killed by the OOMKiller. At `12:00:06`, `podB` will try to become the leader. It will check if "now" (`12:00:06`) is _after_ last renewal + lease duration, essentially it will check: + +[source] +---- +12:00:06 > (12:00:04 + 00:00:10) +---- + +The condition is not fulfilled, so it can't become the leader. Same result will be at `12:00:08`, `12:00:10` and so on, until `12:00:16` and this is where the TTL (`leaseDuration`) of the lock will expire and `podB` can acquire it. As such, a lower value of `leaseDuration` will mean a faster acquiring of leadership by other pods. + +You might have to give proper RBAC to be able to use this functionality, for example: + +[source] +---- + - apiGroups: [ "coordination.k8s.io" ] + resources: [ "leases", "configmaps" ] + verbs: [ "get", "update", "create", "patch"] +---- diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/LeaderUtils.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/LeaderUtils.java index 12e7180857..27dcdcc2d3 100644 --- a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/LeaderUtils.java +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/LeaderUtils.java @@ -16,18 +16,56 @@ 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"; @@ -35,6 +73,27 @@ private LeaderUtils() { } + /** + * ideally, should always be present. If not, downward api must enable this one. + */ + public static Optional 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)) { diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/CachedSingleThreadScheduler.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/CachedSingleThreadScheduler.java new file mode 100644 index 0000000000..32b141c6ef --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/CachedSingleThreadScheduler.java @@ -0,0 +1,119 @@ +/* + * 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.util.concurrent.Executors; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; + +import jakarta.annotation.Nonnull; +import org.apache.commons.logging.LogFactory; + +import org.springframework.core.log.LogAccessor; + +/** + * This is taken from fabric8 with some changes (we need it, so it could be placed 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 + */ +public final class CachedSingleThreadScheduler { + + private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(CachedSingleThreadScheduler.class)); + + private final ReentrantLock lock = new ReentrantLock(); + + private final long ttlMillis; + + private final String name; + + private ScheduledThreadPoolExecutor executor; + + public CachedSingleThreadScheduler(String name, long ttlMillis) { + this.ttlMillis = ttlMillis; + this.name = name; + } + + public ScheduledFuture scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) { + try { + lock.lock(); + this.startExecutor(); + LOG.debug(() -> "Scheduling command to run in : " + name); + return this.executor.scheduleWithFixedDelay(command, initialDelay, delay, unit); + } + finally { + lock.unlock(); + } + } + + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + try { + lock.lock(); + this.startExecutor(); + LOG.debug(() -> "Scheduling command to run in : " + name); + return this.executor.schedule(command, delay, unit); + } + finally { + lock.unlock(); + } + } + + private void startExecutor() { + if (this.executor == null) { + this.executor = new ScheduledThreadPoolExecutor(1, threadFactory()); + this.executor.setRemoveOnCancelPolicy(true); + this.executor.scheduleWithFixedDelay(this::shutdownCheck, this.ttlMillis, this.ttlMillis, + TimeUnit.MILLISECONDS); + } + + } + + private void shutdownCheck() { + try { + lock.lock(); + if (this.executor.getQueue().isEmpty()) { + LOG.debug(() -> "Shutting down executor : " + name); + this.executor.shutdownNow(); + this.executor = null; + } + } + finally { + lock.unlock(); + } + + } + + private ThreadFactory threadFactory() { + return new ThreadFactory() { + final ThreadFactory threadFactory = Executors.defaultThreadFactory(); + + @Override + public Thread newThread(@Nonnull Runnable runnable) { + Thread thread = threadFactory.newThread(runnable); + thread.setName("fabric8-leader-election" + "-" + thread.getName()); + thread.setDaemon(true); + return thread; + } + }; + } + +} diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/ConditionalOnLeaderElectionDisabled.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/ConditionalOnLeaderElectionDisabled.java new file mode 100644 index 0000000000..6fce5c9a6e --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/ConditionalOnLeaderElectionDisabled.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 { + + } + + } + +} diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/ConditionalOnLeaderElectionEnabled.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/ConditionalOnLeaderElectionEnabled.java new file mode 100644 index 0000000000..a58a68e553 --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/ConditionalOnLeaderElectionEnabled.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: + * spring.cloud.kubernetes.leader.election.enabled. + * + * @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 { + +} diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionCallbacks.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionCallbacks.java new file mode 100644 index 0000000000..dc919afd21 --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionCallbacks.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(() -> 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 onNewLeaderCallback(ApplicationEventPublisher applicationEventPublisher, + LeaderElectionProperties properties) { + return holderIdentity -> { + LOG.info(() -> "id : " + holderIdentity + " is the new leader"); + if (properties.publishEvents()) { + applicationEventPublisher.publishEvent(new NewLeaderEvent(holderIdentity)); + } + }; + } + +} diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionProperties.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionProperties.java new file mode 100644 index 0000000000..f8b2465e48 --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionProperties.java @@ -0,0 +1,90 @@ +/* + * 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.time.Duration; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.bind.DefaultValue; + +import static org.springframework.cloud.kubernetes.commons.leader.LeaderUtils.LEADER_ELECTION_PROPERTY_PREFIX; + +/** + * 
+ * waitForPodReady: should we wait for the readiness of the pod,
+ *      before we even trigger the leader election process.
+ * publishEvents: should we publish events (ApplicationEvent)
+ *      when the state of leaders changes.
+ * leaseDuration: TTL of the lease. No other leader candidate
+ *      can acquire the lease unless this one expires.
+ * lockNamespace: where to create the "lock"
+ *      (this is either a lease or a config map)
+ * lockName: the name of the lease or configmap
+ * renewDeadline: once the lock is acquired,
+ *      and we are the current leader, we try to "extend" the lease.
+ *      We must extend it within this timeline.
+ * retryPeriod: how often to retry when trying to get
+ *      the lock to become the leader. In our current code,
+ *      this is what we use in LeaderInitiator::start,
+ *      more exactly in the scheduleAtFixRate
+ *
+ *
+ * First, we try to acquire the lock (lock is either a configmap or a lease)
+ * and by "acquire" I mean write to it (or its annotations for a configmap).
+ * Whoever writes first (all others will get a 409) becomes the leader.
+ * All leader candidates that are not leaders will continue to spin forever
+ * until they get a chance to become one. They retry every 'retryPeriod'.
+ * The current leader, after it establishes itself as one,
+ * will spin forever too, but will try to extend its leadership.
+ * It extends that by updating the entries in the lease,
+ * specifically the one we care about is: renewTime.
+ * This one is updated every 'retryPeriod'. For example,
+ * every 2 seconds (retryPeriod), it will update its 'renewTime' with "now".
+ *
+ * All other, non-leaders are spinning and check a few things in each cycle:
+ * "Am I the leader?" If the answer is no, they go below:
+ * "Can I become the leader?" This is answered by looking at:
+ * now().isAfter(leaderElectionRecord.getRenewTime()
+ *           .plus(leaderElectionConfig.getLeaseDuration()))
+ * So they can only try to acquire the leadership if 'leaseDuration'
+ * (basically a TTL) + renewTime (when was the last renewal) has expired.
+ * This means that no one will be able to even try to acquire the lock
+ * until that leaseDuration expires. When the pod is killed or dies
+ * unexpectedly (OOM, for example), all non-leaders will wait until
+ * leaseDuration expires.
+ *
+ * In case of a graceful shutdown (we call CompletableFuture::cancel on the fabric8 instances),
+ * there is code that fabric8 will trigger to "reset" the lease:
+ * they will set the renewTime to "now" and leaseDuration to 1 second.
+ *
+ * + * @author wind57 + */ +// @formatter:off +@ConfigurationProperties(LEADER_ELECTION_PROPERTY_PREFIX) +public record LeaderElectionProperties( + @DefaultValue("true") boolean waitForPodReady, + @DefaultValue("true") boolean publishEvents, + @DefaultValue("15s") Duration leaseDuration, + @DefaultValue("default") String lockNamespace, + @DefaultValue("spring-k8s-leader-election-lock") String lockName, + @DefaultValue("10s") Duration renewDeadline, + @DefaultValue("2s") Duration retryPeriod, + @DefaultValue("0s") Duration waitAfterRenewalFailure, + @DefaultValue("false") boolean useConfigMapAsLock) { +// @formatter:on +} diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/PodReadyRunner.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/PodReadyRunner.java new file mode 100644 index 0000000000..c98197728d --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/PodReadyRunner.java @@ -0,0 +1,104 @@ +/* + * 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.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; + +import org.springframework.core.log.LogAccessor; + +/** + * @author wind57 + */ +public final class PodReadyRunner { + + private final String candidateIdentity; + + private final String candidateNamespace; + + public PodReadyRunner(String candidateIdentity, String candidateNamespace) { + this.candidateIdentity = candidateIdentity; + this.candidateNamespace = candidateNamespace; + } + + // how often the inner runnable runs, or how much is the scheduler kept alive + private static final long TTL_MILLIS = 100; + + private static final LogAccessor LOG = new LogAccessor(PodReadyRunner.class); + + private final CachedSingleThreadScheduler podReadyScheduler = new CachedSingleThreadScheduler("podReadyExecutor", + TTL_MILLIS); + + public CompletableFuture podReady(BooleanSupplier podReadySupplier) { + + CompletableFuture podReadyFuture = new CompletableFuture<>(); + + ScheduledFuture future = podReadyScheduler.scheduleWithFixedDelay(() -> { + + if (podReadyFuture.isDone()) { + LOG.info(() -> "pod readiness is known, not running another cycle"); + return; + } + + try { + if (podReadySupplier.getAsBoolean()) { + LOG.info( + () -> "Pod : " + candidateIdentity + " in namespace : " + candidateNamespace + " is ready"); + podReadyFuture.complete(null); + } + else { + LOG.debug(() -> "Pod : " + candidateIdentity + " in namespace : " + candidateNamespace + + " is not ready, will retry in one second"); + } + } + catch (Exception e) { + LOG.error(() -> "exception waiting for pod : " + e.getMessage()); + LOG.error(() -> "leader election for : " + candidateIdentity + " was not successful"); + podReadyFuture.completeExceptionally(e); + } + + }, 1, 1, TimeUnit.SECONDS); + + // cancel the future, thus shutting down the executor + podReadyFuture.whenComplete((ok, nok) -> { + if (nok != null) { + if (podReadyFuture.isCancelled()) { + // something triggered us externally by calling + // CompletableFuture::cancel, + // need to shut down the readiness check + LOG.debug(() -> "canceling scheduled future because completable future was cancelled"); + } + else { + LOG.debug(() -> "canceling scheduled future because readiness failed"); + } + } + else { + LOG.debug(() -> "canceling scheduled future because readiness succeeded"); + } + + // no matter the outcome, we cancel the future and thus shut down the + // executor that runs it. + future.cancel(true); + }); + + return podReadyFuture; + + } + +} diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/events/NewLeaderEvent.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/events/NewLeaderEvent.java new file mode 100644 index 0000000000..727dce8ba4 --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/events/NewLeaderEvent.java @@ -0,0 +1,34 @@ +/* + * 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.events; + +import org.springframework.context.ApplicationEvent; + +public final class NewLeaderEvent extends ApplicationEvent { + + private final String holderIdentity; + + public NewLeaderEvent(Object source) { + super(source); + holderIdentity = (String) source; + } + + public String holderIdentity() { + return holderIdentity; + } + +} diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/events/StartLeadingEvent.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/events/StartLeadingEvent.java new file mode 100644 index 0000000000..2d580d7013 --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/events/StartLeadingEvent.java @@ -0,0 +1,37 @@ +/* + * 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.events; + +import org.springframework.context.ApplicationEvent; + +/** + * @author wind57 + */ +public final class StartLeadingEvent extends ApplicationEvent { + + private final String holderIdentity; + + public StartLeadingEvent(Object source) { + super(source); + holderIdentity = (String) source; + } + + public String holderIdentity() { + return holderIdentity; + } + +} diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/events/StopLeadingEvent.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/events/StopLeadingEvent.java new file mode 100644 index 0000000000..ee0bc1abbe --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/leader/election/events/StopLeadingEvent.java @@ -0,0 +1,37 @@ +/* + * 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.events; + +import org.springframework.context.ApplicationEvent; + +/** + * @author wind57 + */ +public final class StopLeadingEvent extends ApplicationEvent { + + private final String holderIdentity; + + public StopLeadingEvent(Object source) { + super(source); + holderIdentity = (String) source; + } + + public String holderIdentity() { + return holderIdentity; + } + +} diff --git a/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/LeaderUtilsTests.java b/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/LeaderUtilsTests.java index 74fb4997b9..b3bad5be3b 100644 --- a/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/LeaderUtilsTests.java +++ b/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/LeaderUtilsTests.java @@ -18,6 +18,7 @@ import java.net.InetAddress; import java.net.UnknownHostException; +import java.util.Optional; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; @@ -62,4 +63,23 @@ void hostNameReadFromApiCall() throws UnknownHostException { inet4AddressMockedStatic.close(); } + @Test + void podNamespaceMissing() { + MockedStatic envReaderMockedStatic = Mockito.mockStatic(EnvReader.class); + // envReaderMockedStatic.when(() -> EnvReader.getEnv("")).thenReturn(""); + Optional podNamespace = LeaderUtils.podNamespace(); + Assertions.assertThat(podNamespace.isEmpty()).isTrue(); + envReaderMockedStatic.close(); + } + + @Test + void podNamespacePresent() { + MockedStatic envReaderMockedStatic = Mockito.mockStatic(EnvReader.class); + envReaderMockedStatic.when(() -> EnvReader.getEnv("POD_NAMESPACE")).thenReturn("podNamespace"); + Optional podNamespace = LeaderUtils.podNamespace(); + Assertions.assertThat(podNamespace.isPresent()).isTrue(); + Assertions.assertThat(podNamespace.get()).isEqualTo("podNamespace"); + envReaderMockedStatic.close(); + } + } diff --git a/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/CachedSingleThreadSchedulerTest.java b/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/CachedSingleThreadSchedulerTest.java new file mode 100644 index 0000000000..b099900913 --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/CachedSingleThreadSchedulerTest.java @@ -0,0 +1,200 @@ +/* + * 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.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; + +import org.assertj.core.api.Assertions; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; + +/** + * @author wind57 + */ +@ExtendWith(OutputCaptureExtension.class) +@SpringBootTest(properties = { "spring.cloud.config.enabled=false", + "logging.level.org.springframework.cloud.kubernetes.commons.leader.election=debug" }) +class CachedSingleThreadSchedulerTest { + + /** + * 
+	 *     - pod readiness passes after two attempts
+	 *     - we check that the executor is shutdown after readiness passes
+	 *
+ */ + @Test + void readinessPasses(CapturedOutput output) throws Exception { + + AtomicInteger counter = new AtomicInteger(); + + BooleanSupplier supplier = () -> { + if (counter.get() == 2) { + return true; + } + else { + counter.incrementAndGet(); + } + return false; + }; + + PodReadyRunner readyRunner = new PodReadyRunner("my-pod", "my-namespace"); + CompletableFuture ready = readyRunner.podReady(supplier); + ready.get(); + + String out = output.getOut(); + Assertions.assertThat(out).contains("Scheduling command to run in : podReadyExecutor"); + Assertions.assertThat(out) + .contains("Pod : my-pod in namespace : " + "my-namespace is not ready, will retry in one second"); + Assertions.assertThat(out).contains("Pod : my-pod in namespace : " + "my-namespace is ready"); + + // executor is shutting down + Awaitility.await() + .pollInterval(Duration.ofMillis(1)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut().contains("Shutting down executor : podReadyExecutor")); + + Awaitility.await() + .pollInterval(Duration.ofMillis(1)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut().contains("canceling scheduled future because readiness succeeded")); + } + + /** + * 
+	 *     - pod readiness fails after one attempt
+	 *     - we check that the executor is shutdown after that
+	 *
+ */ + @Test + void readinessFails(CapturedOutput output) throws Exception { + + AtomicInteger counter = new AtomicInteger(); + + BooleanSupplier supplier = () -> { + if (counter.get() == 1) { + throw new RuntimeException("just because"); + } + else { + counter.incrementAndGet(); + } + return false; + }; + + PodReadyRunner readyRunner = new PodReadyRunner("my-pod", "my-namespace"); + CompletableFuture ready = readyRunner.podReady(supplier); + + ExecutorService readyCheckExecutor = Executors.newSingleThreadExecutor(); + + boolean[] caught = new boolean[1]; + // just like Fabric8LeaderElectionInitiator does it + // ready.get is called in a different executor + readyCheckExecutor.submit(() -> { + try { + ready.get(); + } + catch (Exception e) { + caught[0] = true; + throw new RuntimeException(e); + } + }); + + // pod readiness is started + Awaitility.await() + .pollInterval(Duration.ofMillis(200)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut().contains("Scheduling command to run in : podReadyExecutor")); + + // pod readiness progresses + Awaitility.await() + .pollInterval(Duration.ofMillis(200)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut() + .contains("Pod : my-pod in namespace : " + "my-namespace is not ready, will retry in one second")); + + // executor is shutting down + Awaitility.await() + .pollInterval(Duration.ofMillis(1)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut().contains("Shutting down executor : podReadyExecutor")); + + Awaitility.await() + .pollInterval(Duration.ofMillis(1)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut().contains("canceling scheduled future because readiness failed")); + + Assertions.assertThat(caught[0]).isTrue(); + } + + /** + * 
+	 *     - pod readiness is not established
+	 *     - we cancel the future
+	 *     - we check that the executor is shutdown after that
+	 *
+ */ + @Test + void readinessCanceled(CapturedOutput output) throws Exception { + + BooleanSupplier supplier = () -> false; + + PodReadyRunner readyRunner = new PodReadyRunner("my-pod", "my-namespace"); + CompletableFuture ready = readyRunner.podReady(supplier); + + // sleep a few cycles of pod readiness check + Thread.sleep(2_000); + + // cancel must end the readiness check + ready.cancel(true); + + // pod readiness is started + Awaitility.await() + .pollInterval(Duration.ofMillis(200)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut().contains("Scheduling command to run in : podReadyExecutor")); + + // pod readiness progresses + Awaitility.await() + .pollInterval(Duration.ofMillis(200)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut() + .contains("Pod : my-pod in namespace : " + "my-namespace is not ready, will retry in one second")); + + // executor is shutting down + Awaitility.await() + .pollInterval(Duration.ofMillis(1)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut().contains("Shutting down executor : podReadyExecutor")); + + Awaitility.await() + .pollInterval(Duration.ofMillis(1)) + .atMost(Duration.ofSeconds(10)) + .until(() -> output.getOut() + .contains("canceling scheduled future because completable future was cancelled")); + + } + +} diff --git a/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionPropertiesTests.java b/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionPropertiesTests.java new file mode 100644 index 0000000000..291cef6aa5 --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/LeaderElectionPropertiesTests.java @@ -0,0 +1,83 @@ +/* + * 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.time.Duration; + +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.context.annotation.Configuration; + +/** + * @author wind57 + */ +class LeaderElectionPropertiesTests { + + @Test + void testDefaults() { + new ApplicationContextRunner().withUserConfiguration(Config.class).run(context -> { + LeaderElectionProperties properties = context.getBean(LeaderElectionProperties.class); + Assertions.assertThat(properties).isNotNull(); + Assertions.assertThat(properties.publishEvents()).isTrue(); + Assertions.assertThat(properties.waitForPodReady()).isTrue(); + Assertions.assertThat(properties.leaseDuration()).isEqualTo(Duration.ofSeconds(15)); + Assertions.assertThat(properties.lockNamespace()).isEqualTo("default"); + Assertions.assertThat(properties.lockName()).isEqualTo("spring-k8s-leader-election-lock"); + Assertions.assertThat(properties.renewDeadline()).isEqualTo(Duration.ofSeconds(10)); + Assertions.assertThat(properties.retryPeriod()).isEqualTo(Duration.ofSeconds(2)); + Assertions.assertThat(properties.waitAfterRenewalFailure()).isEqualTo(Duration.ofSeconds(0)); + Assertions.assertThat(properties.useConfigMapAsLock()).isFalse(); + }); + } + + @Test + void testNonDefaults() { + new ApplicationContextRunner().withUserConfiguration(Config.class) + .withPropertyValues("spring.cloud.kubernetes.leader.election.wait-for-pod-ready=false", + "spring.cloud.kubernetes.leader.election.publish-events=false", + "spring.cloud.kubernetes.leader.election.lease-duration=10s", + "spring.cloud.kubernetes.leader.election.lock-namespace=lock-namespace", + "spring.cloud.kubernetes.leader.election.lock-name=lock-name", + "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") + .run(context -> { + LeaderElectionProperties properties = context.getBean(LeaderElectionProperties.class); + Assertions.assertThat(properties).isNotNull(); + Assertions.assertThat(properties.waitForPodReady()).isFalse(); + Assertions.assertThat(properties.publishEvents()).isFalse(); + Assertions.assertThat(properties.leaseDuration()).isEqualTo(Duration.ofSeconds(10)); + Assertions.assertThat(properties.lockNamespace()).isEqualTo("lock-namespace"); + Assertions.assertThat(properties.lockName()).isEqualTo("lock-name"); + Assertions.assertThat(properties.renewDeadline()).isEqualTo(Duration.ofDays(2)); + Assertions.assertThat(properties.retryPeriod()).isEqualTo(Duration.ofMinutes(3)); + Assertions.assertThat(properties.waitAfterRenewalFailure()).isEqualTo(Duration.ofMinutes(13)); + Assertions.assertThat(properties.useConfigMapAsLock()).isTrue(); + }); + } + + @EnableConfigurationProperties(LeaderElectionProperties.class) + @Configuration + static class Config { + + } + +} diff --git a/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/PodReadyRunnerTests.java b/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/PodReadyRunnerTests.java new file mode 100644 index 0000000000..559e2e35cd --- /dev/null +++ b/spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/leader/election/PodReadyRunnerTests.java @@ -0,0 +1,237 @@ +/* + * 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.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * @author wind57 + */ +@SpringBootTest(properties = { "logging.level.org.springframework.cloud.kubernetes.commons.leader.election=debug", + "spring.cloud.config.enabled=false" }) +@ExtendWith(OutputCaptureExtension.class) +class PodReadyRunnerTests { + + private final PodReadyRunner podReadyRunner = new PodReadyRunner("identity", "namespace"); + + /** + * 
+	 *     - readiness passes from the first cycle
+	 *     - assert that proper logging is in place
+	 *     - assert that executor is getting shutdown
+	 *
+ */ + @Test + void readinessOKFromTheFirstCycle(CapturedOutput output) throws Exception { + BooleanSupplier readinessSupplier = () -> true; + CompletableFuture readinessFuture = podReadyRunner.podReady(readinessSupplier); + readinessFuture.get(); + + assertThat(output.getOut()).contains("Pod : identity in namespace : namespace is ready"); + assertThat(output.getOut()).contains("canceling scheduled future because readiness succeeded"); + + await().atMost(Duration.ofSeconds(3)) + .pollInterval(Duration.ofMillis(200)) + .until(() -> output.getOut().contains("Shutting down executor : podReadyExecutor")); + } + + /** + * 
+	 *     - readiness passes from the second cycle
+	 *
+ */ + @Test + void readinessOKFromTheSecondCycle(CapturedOutput output) throws Exception { + AtomicInteger counter = new AtomicInteger(0); + BooleanSupplier readinessSupplier = () -> { + if (counter.get() == 0) { + counter.incrementAndGet(); + return false; + } + return true; + }; + CompletableFuture readinessFuture = podReadyRunner.podReady(readinessSupplier); + readinessFuture.get(); + + assertThat(output.getOut()) + .contains("Pod : identity in namespace : namespace is not ready, will retry in one second"); + assertThat(output.getOut()).contains("Pod : identity in namespace : namespace is ready"); + assertThat(output.getOut()).contains("canceling scheduled future because readiness succeeded"); + + await().atMost(Duration.ofSeconds(3)) + .pollInterval(Duration.ofMillis(200)) + .until(() -> output.getOut().contains("Shutting down executor : podReadyExecutor")); + } + + /** + * 
+	 *     - readiness throws an Exception in the second cycle
+	 *
+ */ + @Test + void readinessFailsOnTheSecondCycle(CapturedOutput output) { + AtomicInteger counter = new AtomicInteger(0); + BooleanSupplier readinessSupplier = () -> { + if (counter.get() == 0) { + counter.incrementAndGet(); + return false; + } + throw new RuntimeException("fail on the second cycle"); + }; + CompletableFuture readinessFuture = podReadyRunner.podReady(readinessSupplier); + boolean caught = false; + try { + readinessFuture.get(); + } + catch (Exception e) { + caught = true; + assertThat(output.getOut()) + .contains("Pod : identity in namespace : namespace is not ready, will retry in one second"); + assertThat(output.getOut()).contains("exception waiting for pod : fail on the second cycle"); + assertThat(output.getOut()).contains("leader election for : identity was not successful"); + assertThat(output.getOut()).contains("canceling scheduled future because readiness failed"); + + await().atMost(Duration.ofSeconds(3)) + .pollInterval(Duration.ofMillis(200)) + .until(() -> output.getOut().contains("Shutting down executor : podReadyExecutor")); + } + assertThat(caught).isTrue(); + } + + /** + * 
+	 *     - readiness throws an Exception in the second cycle
+	 *     - we chain one more thenApply and test it, just like
+	 *       Fabric8LeaderElectionInitiator does it.
+	 *
+ */ + @Test + void readinessFailsOnTheSecondCycleAttachNewPipeline(CapturedOutput output) { + AtomicInteger counter = new AtomicInteger(0); + BooleanSupplier readinessSupplier = () -> { + if (counter.get() == 0) { + counter.incrementAndGet(); + return false; + } + throw new RuntimeException("fail on the second cycle"); + }; + CompletableFuture podReadyFuture = podReadyRunner.podReady(readinessSupplier); + + CompletableFuture ready = podReadyFuture.whenComplete((ok, error) -> { + if (error != null) { + System.out.println("readiness failed and we caught that"); + } + else { + System.out.println("readiness succeeded"); + } + }); + + boolean caught = false; + try { + ready.get(); + } + catch (Exception e) { + caught = true; + assertThat(output.getOut()) + .contains("Pod : identity in namespace : namespace is not ready, will retry in one second"); + assertThat(output.getOut()).contains("exception waiting for pod : fail on the second cycle"); + assertThat(output.getOut()).contains("leader election for : identity was not successful"); + assertThat(output.getOut()).contains("readiness failed and we caught that"); + assertThat(output.getOut()).contains("canceling scheduled future because readiness failed"); + + await().atMost(Duration.ofSeconds(3)) + .pollInterval(Duration.ofMillis(200)) + .until(() -> output.getOut().contains("Shutting down executor : podReadyExecutor")); + } + assertThat(caught).isTrue(); + } + + /** + * 
+	 *     - readiness is canceled
+	 *     - we chain one more thenApply and test it, just like
+	 *       Fabric8LeaderElectionInitiator does it.
+	 *
+	 *     - this simulates when we issue cancel of the podReadyFuture from
+	 *     - pre-destroy code.
+	 *
+ */ + @Test + void readinessCanceledOnTheSecondCycleAttachNewPipeline(CapturedOutput output) throws Exception { + BooleanSupplier readinessSupplier = () -> false; + + CompletableFuture podReadyFuture = podReadyRunner.podReady(readinessSupplier); + + CompletableFuture ready = podReadyFuture.whenComplete((ok, error) -> { + if (error != null) { + System.out.println("readiness failed and we caught that"); + } + else { + System.out.println("readiness succeeded"); + } + }); + + // sleep a few cycles of pod readiness check + Thread.sleep(2_000); + + ScheduledExecutorService cancelScheduler = null; + + boolean caught = false; + // cancel podReady future in a different thread + cancelScheduler = Executors.newScheduledThreadPool(1); + cancelScheduler.scheduleWithFixedDelay(() -> podReadyFuture.cancel(true), 1, 1, TimeUnit.SECONDS); + + try { + ready.get(); + } + catch (Exception e) { + caught = true; + assertThat(output.getOut()) + .contains("Pod : identity in namespace : namespace is not ready, will retry in one second"); + // this is a cancel of the future, not an exception per se + assertThat(output.getOut()).doesNotContain("leader election for : identity was not successful"); + assertThat(output.getOut()).contains("readiness failed and we caught that"); + + assertThat(output.getOut()).contains("canceling scheduled future because completable future was cancelled"); + assertThat(output.getOut()).doesNotContain("canceling scheduled future because readiness failed"); + + await().atMost(Duration.ofSeconds(3)) + .pollInterval(Duration.ofMillis(200)) + .until(() -> output.getOut().contains("Shutting down executor : podReadyExecutor")); + } + assertThat(caught).isTrue(); + cancelScheduler.shutdownNow(); + + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8LeaderAutoConfiguration.java b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8LeaderAutoConfiguration.java index bb04ed4275..01af2413a1 100644 --- a/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8LeaderAutoConfiguration.java +++ b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8LeaderAutoConfiguration.java @@ -30,6 +30,7 @@ import org.springframework.cloud.kubernetes.commons.leader.LeaderInitiator; import org.springframework.cloud.kubernetes.commons.leader.LeaderProperties; import org.springframework.cloud.kubernetes.commons.leader.LeaderUtils; +import org.springframework.cloud.kubernetes.commons.leader.election.ConditionalOnLeaderElectionDisabled; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -45,6 +46,7 @@ @EnableConfigurationProperties(LeaderProperties.class) @ConditionalOnBean(KubernetesClient.class) @ConditionalOnProperty(value = "spring.cloud.kubernetes.leader.enabled", matchIfMissing = true) +@ConditionalOnLeaderElectionDisabled public class Fabric8LeaderAutoConfiguration { /* diff --git a/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionAutoConfiguration.java b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionAutoConfiguration.java new file mode 100644 index 0000000000..0bec6c4ecc --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionAutoConfiguration.java @@ -0,0 +1,128 @@ +/* + * 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 io.fabric8.kubernetes.api.model.APIResource; +import io.fabric8.kubernetes.api.model.APIResourceList; +import io.fabric8.kubernetes.api.model.GroupVersionForDiscovery; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfig; +import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfigBuilder; +import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.ConfigMapLock; +import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.LeaseLock; +import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.Lock; + +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.boot.health.autoconfigure.contributor.ConditionalOnEnabledHealthIndicator; +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; +import static org.springframework.cloud.kubernetes.commons.leader.LeaderUtils.LEASE; + +/** + * @author wind57 + */ +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(LeaderElectionProperties.class) +@ConditionalOnBean(KubernetesClient.class) +@ConditionalOnLeaderElectionEnabled +@ConditionalOnCloudPlatform(CloudPlatform.KUBERNETES) +@AutoConfigureAfter(Fabric8LeaderElectionCallbacksAutoConfiguration.class) +class Fabric8LeaderElectionAutoConfiguration { + + private static final String COORDINATION_VERSION_GROUP = COORDINATION_GROUP + "/" + COORDINATION_VERSION; + + private static final LogAccessor LOG = new LogAccessor(Fabric8LeaderElectionAutoConfiguration.class); + + @Bean + @ConditionalOnClass(InfoContributor.class) + @ConditionalOnEnabledHealthIndicator("leader.election") + Fabric8LeaderElectionInfoContributor leaderElectionInfoContributor(String holderIdentity, + LeaderElectionConfig leaderElectionConfig, KubernetesClient fabric8KubernetesClient) { + return new Fabric8LeaderElectionInfoContributor(holderIdentity, leaderElectionConfig, fabric8KubernetesClient); + } + + @Bean + @ConditionalOnMissingBean + Fabric8LeaderElectionInitiator fabric8LeaderElectionInitiator(String holderIdentity, String podNamespace, + KubernetesClient fabric8KubernetesClient, LeaderElectionConfig fabric8LeaderElectionConfig, + LeaderElectionProperties leaderElectionProperties) { + return new Fabric8LeaderElectionInitiator(holderIdentity, podNamespace, fabric8KubernetesClient, + fabric8LeaderElectionConfig, leaderElectionProperties); + } + + @Bean + @ConditionalOnMissingBean + LeaderElectionConfig fabric8LeaderElectionConfig(LeaderElectionProperties properties, Lock lock, + Fabric8LeaderElectionCallbacks fabric8LeaderElectionCallbacks) { + return new LeaderElectionConfigBuilder() + .withReleaseOnCancel() + .withName("Spring k8s leader election") + .withLeaseDuration(properties.leaseDuration()) + .withLock(lock) + .withRenewDeadline(properties.renewDeadline()) + .withRetryPeriod(properties.retryPeriod()) + .withLeaderCallbacks(fabric8LeaderElectionCallbacks) + .build(); + } + + @Bean + @ConditionalOnMissingBean + Lock lock(KubernetesClient fabric8KubernetesClient, LeaderElectionProperties properties, String holderIdentity) { + boolean leaseSupported = fabric8KubernetesClient.getApiGroups() + .getGroups() + .stream() + .flatMap(x -> x.getVersions().stream()) + .map(GroupVersionForDiscovery::getGroupVersion) + .filter(COORDINATION_VERSION_GROUP::equals) + .findFirst() + .map(fabric8KubernetesClient::getApiResources) + .map(APIResourceList::getResources) + .map(x -> x.stream().map(APIResource::getKind)) + .flatMap(x -> x.filter(y -> y.equals(LEASE)).findFirst()) + .isPresent(); + + 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(), holderIdentity); + } + else { + LOG.info(() -> "will use lease as the lock for leader election"); + return new LeaseLock(properties.lockNamespace(), properties.lockName(), holderIdentity); + } + } + else { + LOG.info(() -> "will use configmap as the lock for leader election"); + return new ConfigMapLock(properties.lockNamespace(), properties.lockName(), holderIdentity); + } + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionCallbacks.java b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionCallbacks.java new file mode 100644 index 0000000000..910d7fb3a7 --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionCallbacks.java @@ -0,0 +1,32 @@ +/* + * Copyright 2013-2024 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.function.Consumer; + +import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks; + +/** + * @author wind57 + */ +final class Fabric8LeaderElectionCallbacks extends LeaderCallbacks { + + Fabric8LeaderElectionCallbacks(Runnable onStartLeading, Runnable onStopLeading, Consumer onNewLeader) { + super(onStartLeading, onStopLeading, onNewLeader); + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionCallbacksAutoConfiguration.java b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionCallbacksAutoConfiguration.java new file mode 100644 index 0000000000..7e1faddede --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionCallbacksAutoConfiguration.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.fabric8.leader.election; + +import java.util.function.Consumer; + +import io.fabric8.kubernetes.client.KubernetesClient; + +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.commons.KubernetesCommonsAutoConfiguration; +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.cloud.kubernetes.fabric8.Fabric8AutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(LeaderElectionProperties.class) +@ConditionalOnBean(KubernetesClient.class) +@ConditionalOnCloudPlatform(CloudPlatform.KUBERNETES) +@ConditionalOnLeaderElectionEnabled +@AutoConfigureAfter({ Fabric8AutoConfiguration.class, KubernetesCommonsAutoConfiguration.class }) +final class Fabric8LeaderElectionCallbacksAutoConfiguration extends LeaderElectionCallbacks { + + @Bean + @ConditionalOnMissingBean + Fabric8LeaderElectionCallbacks fabric8LeaderElectionCallbacks(Runnable onStartLeadingCallback, + Runnable onStopLeadingCallback, Consumer onNewLeaderCallback) { + return new Fabric8LeaderElectionCallbacks(onStartLeadingCallback, onStopLeadingCallback, onNewLeaderCallback); + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInfoContributor.java b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInfoContributor.java new file mode 100644 index 0000000000..706a2fbb4d --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInfoContributor.java @@ -0,0 +1,60 @@ +/* + * 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.HashMap; +import java.util.Map; +import java.util.Optional; + +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfig; + +import org.springframework.boot.actuate.info.Info; +import org.springframework.boot.actuate.info.InfoContributor; + +/** + * @author wind57 + */ +final class Fabric8LeaderElectionInfoContributor implements InfoContributor { + + private final String holderIdentity; + + private final LeaderElectionConfig leaderElectionConfig; + + private final KubernetesClient fabric8KubernetesClient; + + Fabric8LeaderElectionInfoContributor(String holderIdentity, LeaderElectionConfig leaderElectionConfig, + KubernetesClient fabric8KubernetesClient) { + this.holderIdentity = holderIdentity; + this.leaderElectionConfig = leaderElectionConfig; + this.fabric8KubernetesClient = fabric8KubernetesClient; + } + + @Override + public void contribute(Info.Builder builder) { + Map details = new HashMap<>(); + Optional.ofNullable(leaderElectionConfig.getLock().get(fabric8KubernetesClient)) + .ifPresentOrElse(leaderRecord -> { + boolean isLeader = holderIdentity.equals(leaderRecord.getHolderIdentity()); + details.put("leaderId", holderIdentity); + details.put("isLeader", isLeader); + }, () -> details.put("leaderId", "Unknown")); + + builder.withDetail("leaderElection", details); + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInitiator.java b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInitiator.java new file mode 100644 index 0000000000..af2ad8a5ca --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInitiator.java @@ -0,0 +1,208 @@ +/* +* Copyright 2013-2024 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.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.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfig; +import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector; +import io.fabric8.kubernetes.client.readiness.Readiness; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; + +import org.springframework.cloud.kubernetes.commons.leader.election.LeaderElectionProperties; +import org.springframework.cloud.kubernetes.commons.leader.election.PodReadyRunner; +import org.springframework.core.log.LogAccessor; + +import static java.util.concurrent.Executors.newSingleThreadExecutor; + +/** + * @author wind57 + */ + final class Fabric8LeaderElectionInitiator { + + private static final LogAccessor LOG = new LogAccessor(Fabric8LeaderElectionInitiator.class); + + private final PodReadyRunner podReadyRunner; + + private final String candidateIdentity; + + private final KubernetesClient fabric8KubernetesClient; + + private final LeaderElectionConfig leaderElectionConfig; + + private final LeaderElectionProperties leaderElectionProperties; + + private final boolean waitForPodReady; + + private final ExecutorService podReadyWaitingExecutor; + + private final BooleanSupplier podReadySupplier; + + private volatile CompletableFuture podReadyFuture; + + private volatile boolean destroyCalled = false; + + private volatile CompletableFuture leaderFuture; + + Fabric8LeaderElectionInitiator(String candidateIdentity, String candidateNamespace, + KubernetesClient fabric8KubernetesClient, LeaderElectionConfig leaderElectionConfig, + LeaderElectionProperties leaderElectionProperties) { + this.candidateIdentity = candidateIdentity; + this.fabric8KubernetesClient = fabric8KubernetesClient; + this.leaderElectionConfig = leaderElectionConfig; + this.leaderElectionProperties = leaderElectionProperties; + this.waitForPodReady = leaderElectionProperties.waitForPodReady(); + + this.podReadyWaitingExecutor = newSingleThreadExecutor(runnable -> + new Thread(runnable, "Fabric8LeaderElectionInitiator-" + candidateIdentity)); + + this.podReadySupplier = () -> { + Pod pod = fabric8KubernetesClient.pods().inNamespace(candidateNamespace).withName(candidateIdentity).get(); + return Readiness.isPodReady(pod); + }; + + this.podReadyRunner = new PodReadyRunner(candidateIdentity, candidateNamespace); + } + + /** + * 
+	 * 	We first try to see if we need to wait for the pod to be ready
+	 * 	before starting the leader election process.
+	 *
+ * + */ + @PostConstruct + void postConstruct() { + LOG.info(() -> "starting leader initiator : " + candidateIdentity); + + // wait until the pod is ready + if (waitForPodReady) { + LOG.info(() -> "will wait until pod " + candidateIdentity + " is ready"); + podReadyFuture = podReadyRunner.podReady(podReadySupplier); + } + else { + podReadyFuture = CompletableFuture.completedFuture(null); + } + + // wait in a different thread until the pod is ready + // and don't block the main application from starting + podReadyWaitingExecutor.submit(() -> { + if (waitForPodReady) { + + // if 'ready' is already completed at this point, thread will run this, + // otherwise it will attach the pipeline and move on to 'blockReadinessCheck' + CompletableFuture ready = podReadyFuture.whenComplete((ok, error) -> { + if (error != null) { + LOG.error(() -> "readiness failed for : " + candidateIdentity); + LOG.error(() -> "leader election for : " + candidateIdentity + " will not start"); + } + else { + LOG.info(() -> candidateIdentity + " is ready"); + startLeaderElection(); + } + }); + + blockReadinessCheck(ready); + + } + else { + startLeaderElection(); + } + }); + + } + + @PreDestroy + void preDestroy() { + destroyCalled = true; + LOG.info(() -> "preDestroy called on the leader initiator : " + candidateIdentity); + + if (podReadyFuture != null && !podReadyFuture.isDone()) { + // if the task is not running, this has no effect. + // if the task is running, calling this will also make sure + // that the caching executor will shut down too. + podReadyFuture.cancel(true); + } + + if (leaderFuture != null) { + LOG.info(() -> "leaderFuture will be canceled for : " + candidateIdentity); + // needed to release the lock, in case we are holding it. + // fabric8 internally expects this one to be called + leaderFuture.cancel(true); + } + podReadyWaitingExecutor.shutdownNow(); + } + + private void startLeaderElection() { + leaderFuture = leaderElector(leaderElectionConfig, fabric8KubernetesClient).start(); + leaderFuture.whenComplete((ok, error) -> { + + if (ok != null) { + LOG.info(() -> "leaderFuture finished normally, will re-start it for : " + candidateIdentity); + startLeaderElection(); + return; + } + + if (error instanceof CancellationException) { + if (!destroyCalled) { + LOG.warn(() -> "renewal failed for : " + candidateIdentity + ", will re-start it after : " + + leaderElectionProperties.waitAfterRenewalFailure().toSeconds() + " seconds"); + sleep(); + startLeaderElection(); + } + } + else { + LOG.warn(() -> "leader election is over for : " + candidateIdentity); + } + + try { + leaderFuture.get(); + } catch (Exception e) { + LOG.warn(() -> "leader election failed for : " + candidateIdentity + ". Trying to recover..."); + } + }); + } + + 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); + } + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/spring-cloud-kubernetes-fabric8-leader/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index d612cac6dc..c94b99ab84 100644 --- a/spring-cloud-kubernetes-fabric8-leader/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/spring-cloud-kubernetes-fabric8-leader/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1 +1,3 @@ org.springframework.cloud.kubernetes.fabric8.leader.Fabric8LeaderAutoConfiguration +org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionCallbacksAutoConfiguration +org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionAutoConfiguration diff --git a/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderApp.java b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderApp.java new file mode 100644 index 0000000000..f51818e22f --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderApp.java @@ -0,0 +1,64 @@ +/* + * Copyright 2013-2024 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 io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import io.fabric8.kubernetes.client.dsl.Resource; +import io.fabric8.kubernetes.client.dsl.internal.BaseOperation; +import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.Lock; +import org.mockito.Mockito; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Primary; + +@Configuration +public class Fabric8LeaderApp { + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Bean + KubernetesClient kubernetesClient() { + KubernetesClient client = Mockito.mock(KubernetesClient.class); + Mockito.when(client.getNamespace()).thenReturn("a"); + + MixedOperation mixedOperation = Mockito.mock(MixedOperation.class); + Mockito.when(client.configMaps()).thenReturn(mixedOperation); + + PodResource podResource = Mockito.mock(PodResource.class); + Mockito.when(podResource.isReady()).thenReturn(true); + + Mockito.when(client.pods()).thenReturn(mixedOperation); + Mockito.when(mixedOperation.withName(Mockito.anyString())).thenReturn(podResource); + + Resource resource = Mockito.mock(Resource.class); + + BaseOperation baseOperation = Mockito.mock(BaseOperation.class); + Mockito.when(baseOperation.withName("leaders")).thenReturn(resource); + + Mockito.when(mixedOperation.inNamespace("a")).thenReturn(baseOperation); + return client; + } + + @Bean + @Primary + Lock lock() { + return Mockito.mock(Lock.class); + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderAutoConfigurationTests.java b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderAutoConfigurationTests.java new file mode 100644 index 0000000000..6979c2d1d3 --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderAutoConfigurationTests.java @@ -0,0 +1,116 @@ +/* + * 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.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.cloud.kubernetes.fabric8.leader.Fabric8LeaderAutoConfiguration; +import org.springframework.cloud.kubernetes.fabric8.leader.Fabric8PodReadinessWatcher; + +/** + * tests that ensure 'spring.cloud.kubernetes.leader.election' enabled correct + * auto-configurations, when it is enabled/disabled. + * + * @author wind57 + */ +class Fabric8LeaderAutoConfigurationTests { + + /** + * 
+	 *     - spring.cloud.kubernetes.leader.election is not present
+	 *
+	 *     As such:
+	 *
+	 *     - Fabric8LeaderAutoConfiguration must be picked up
+	 *     - Fabric8LeaderElectionAutoConfiguration must not be picked up
+	 *
+ */ + @Test + void leaderElectionAnnotationMissing() { + new ApplicationContextRunner().withUserConfiguration(Fabric8LeaderApp.class) + .withConfiguration(AutoConfigurations.of(Fabric8LeaderAutoConfiguration.class, + Fabric8LeaderElectionAutoConfiguration.class, + Fabric8LeaderElectionCallbacksAutoConfiguration.class)) + .run(context -> { + + // this one comes from Fabric8LeaderElectionAutoConfiguration + Assertions.assertThat(context).doesNotHaveBean(Fabric8LeaderElectionInitiator.class); + + // this one comes from Fabric8LeaderAutoConfiguration + Assertions.assertThat(context).hasSingleBean(Fabric8PodReadinessWatcher.class); + }); + } + + /** + * 
+	 *     - spring.cloud.kubernetes.leader.election = false
+	 *
+	 *     As such:
+	 *
+	 *     - Fabric8LeaderAutoConfiguration must be picked up
+	 *     - Fabric8LeaderElectionAutoConfiguration must not be picked up
+	 *
+ */ + @Test + void leaderElectionAnnotationPresentEqualToFalse() { + new ApplicationContextRunner().withUserConfiguration(Fabric8LeaderApp.class) + .withConfiguration(AutoConfigurations.of(Fabric8LeaderAutoConfiguration.class, + Fabric8LeaderElectionAutoConfiguration.class, + Fabric8LeaderElectionCallbacksAutoConfiguration.class)) + .withPropertyValues("spring.cloud.kubernetes.leader.election.enabled=false") + .run(context -> { + + // this one comes from Fabric8LeaderElectionAutoConfiguration + Assertions.assertThat(context).doesNotHaveBean(Fabric8LeaderElectionInitiator.class); + + // this one comes from Fabric8LeaderAutoConfiguration + Assertions.assertThat(context).hasSingleBean(Fabric8PodReadinessWatcher.class); + }); + } + + /** + * 
+	 *     - spring.cloud.kubernetes.leader.election = false
+	 *
+	 *     As such:
+	 *
+	 *     - Fabric8LeaderAutoConfiguration must not be picked up
+	 *     - Fabric8LeaderElectionAutoConfiguration must be picked up
+	 *
+ */ + @Test + void leaderElectionAnnotationPresentEqualToTrue() { + new ApplicationContextRunner().withUserConfiguration(Fabric8LeaderApp.class) + .withConfiguration(AutoConfigurations.of(Fabric8LeaderAutoConfiguration.class, + Fabric8LeaderElectionAutoConfiguration.class, + Fabric8LeaderElectionCallbacksAutoConfiguration.class)) + .withPropertyValues("spring.cloud.kubernetes.leader.election.enabled=true", + "spring.main.cloud-platform=kubernetes") + .run(context -> { + + // this one comes from Fabric8LeaderElectionAutoConfiguration + Assertions.assertThat(context).hasSingleBean(Fabric8LeaderElectionInitiator.class); + + // this one comes from Fabric8LeaderAutoConfiguration + Assertions.assertThat(context).doesNotHaveBean(Fabric8PodReadinessWatcher.class); + }); + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInfoContributorIsLeaderTest.java b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInfoContributorIsLeaderTest.java new file mode 100644 index 0000000000..1cb0b7525b --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInfoContributorIsLeaderTest.java @@ -0,0 +1,150 @@ +/* + * Copyright 2013-2024 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.time.ZonedDateTime; + +import io.fabric8.kubernetes.api.model.APIGroupList; +import io.fabric8.kubernetes.api.model.APIGroupListBuilder; +import io.fabric8.kubernetes.api.model.APIResourceBuilder; +import io.fabric8.kubernetes.api.model.APIResourceListBuilder; +import io.fabric8.kubernetes.api.model.GroupVersionForDiscoveryBuilder; +import io.fabric8.kubernetes.api.model.coordination.v1.Lease; +import io.fabric8.kubernetes.api.model.coordination.v1.LeaseBuilder; +import io.fabric8.kubernetes.api.model.coordination.v1.LeaseSpecBuilder; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.Resource; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.test.web.server.LocalManagementPort; +import org.springframework.boot.webtestclient.AutoConfigureWebTestClient; +import org.springframework.cloud.kubernetes.commons.leader.LeaderUtils; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Primary; +import org.springframework.http.MediaType; +import org.springframework.test.web.reactive.server.WebTestClient; + +/** + * @author wind57 + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = { "spring.main.cloud-platform=KUBERNETES", "management.endpoints.web.exposure.include=info", + "management.endpoint.info.show-details=always", "management.info.kubernetes.enabled=true", + "spring.cloud.kubernetes.leader.election.enabled=true" }) +@AutoConfigureWebTestClient +class Fabric8LeaderElectionInfoContributorIsLeaderTest { + + private static final String HOLDER_IDENTITY = "leader"; + + @LocalManagementPort + private int port; + + @Autowired + private WebTestClient webClient; + + private static MockedStatic leaderUtilsMockedStatic; + + @BeforeAll + static void beforeAll() { + leaderUtilsMockedStatic = Mockito.mockStatic(LeaderUtils.class); + leaderUtilsMockedStatic.when(LeaderUtils::hostName).thenReturn(HOLDER_IDENTITY); + } + + @AfterAll + static void afterAll() { + leaderUtilsMockedStatic.close(); + } + + @Test + void infoEndpointIsLeaderTest() { + webClient.get() + .uri("http://localhost:{port}/actuator/info", port) + .accept(MediaType.APPLICATION_JSON) + .exchange() + .expectStatus() + .isOk() + .expectBody() + .jsonPath("leaderElection.isLeader") + .isEqualTo(true) + .jsonPath("leaderElection.leaderId") + .isEqualTo(HOLDER_IDENTITY); + } + + @TestConfiguration + static class Configuration { + + @Bean + @Primary + KubernetesClient mockKubernetesClient() { + KubernetesClient client = Mockito.mock(KubernetesClient.class); + mockForLeaseSupport(client); + mockForLeaderSupport(client); + return client; + } + + private void mockForLeaseSupport(KubernetesClient client) { + Mockito.when(client.getApiResources("coordination.k8s.io/v1")) + .thenReturn( + new APIResourceListBuilder().withResources(new APIResourceBuilder().withKind("Lease").build()) + .build()); + + APIGroupList apiGroupList = new APIGroupListBuilder().addNewGroup() + .withVersions(new GroupVersionForDiscoveryBuilder().withGroupVersion("coordination.k8s.io/v1").build()) + .endGroup() + .build(); + + Mockito.when(client.getApiGroups()).thenReturn(apiGroupList); + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + private void mockForLeaderSupport(KubernetesClient client) { + + Lease lease = new LeaseBuilder().withNewMetadata() + .withName("spring-k8s-leader-election-lock") + .endMetadata() + .withSpec(new LeaseSpecBuilder().withHolderIdentity(HOLDER_IDENTITY) + .withLeaseDurationSeconds(1) + .withAcquireTime(ZonedDateTime.now()) + .withRenewTime(ZonedDateTime.now()) + .withLeaseTransitions(1) + .build()) + .build(); + + MixedOperation mixedOperation = Mockito.mock(MixedOperation.class); + Mockito.when(client.resources(Lease.class)).thenReturn(mixedOperation); + + Resource resource = Mockito.mock(Resource.class); + Mockito.when(resource.get()).thenReturn(lease); + + NonNamespaceOperation nonNamespaceOperation = Mockito.mock(NonNamespaceOperation.class); + Mockito.when(mixedOperation.inNamespace("default")).thenReturn(nonNamespaceOperation); + Mockito.when(nonNamespaceOperation.withName("spring-k8s-leader-election-lock")).thenReturn(resource); + + } + + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInfoContributorIsNotLeaderTest.java b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInfoContributorIsNotLeaderTest.java new file mode 100644 index 0000000000..069d5037f9 --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInfoContributorIsNotLeaderTest.java @@ -0,0 +1,150 @@ +/* + * 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.time.ZonedDateTime; + +import io.fabric8.kubernetes.api.model.APIGroupList; +import io.fabric8.kubernetes.api.model.APIGroupListBuilder; +import io.fabric8.kubernetes.api.model.APIResourceBuilder; +import io.fabric8.kubernetes.api.model.APIResourceListBuilder; +import io.fabric8.kubernetes.api.model.GroupVersionForDiscoveryBuilder; +import io.fabric8.kubernetes.api.model.coordination.v1.Lease; +import io.fabric8.kubernetes.api.model.coordination.v1.LeaseBuilder; +import io.fabric8.kubernetes.api.model.coordination.v1.LeaseSpecBuilder; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.Resource; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.test.web.server.LocalManagementPort; +import org.springframework.boot.webtestclient.AutoConfigureWebTestClient; +import org.springframework.cloud.kubernetes.commons.leader.LeaderUtils; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Primary; +import org.springframework.http.MediaType; +import org.springframework.test.web.reactive.server.WebTestClient; + +/** + * @author wind57 + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = { "spring.main.cloud-platform=KUBERNETES", "management.endpoints.web.exposure.include=info", + "management.endpoint.info.show-details=always", "management.info.kubernetes.enabled=true", + "spring.cloud.kubernetes.leader.election.enabled=true" }) +@AutoConfigureWebTestClient +class Fabric8LeaderElectionInfoContributorIsNotLeaderTest { + + private static final String HOLDER_IDENTITY = "leader"; + + @LocalManagementPort + private int port; + + @Autowired + private WebTestClient webClient; + + private static MockedStatic leaderUtilsMockedStatic; + + @BeforeAll + static void beforeAll() { + leaderUtilsMockedStatic = Mockito.mockStatic(LeaderUtils.class); + leaderUtilsMockedStatic.when(LeaderUtils::hostName).thenReturn("non-" + HOLDER_IDENTITY); + } + + @AfterAll + static void afterAll() { + leaderUtilsMockedStatic.close(); + } + + @Test + void infoEndpointIsNotLeaderTest() { + webClient.get() + .uri("http://localhost:{port}/actuator/info", port) + .accept(MediaType.APPLICATION_JSON) + .exchange() + .expectStatus() + .isOk() + .expectBody() + .jsonPath("leaderElection.isLeader") + .isEqualTo(false) + .jsonPath("leaderElection.leaderId") + .isEqualTo("non-" + HOLDER_IDENTITY); + } + + @TestConfiguration + static class Configuration { + + @Bean + @Primary + KubernetesClient mockKubernetesClient() { + KubernetesClient client = Mockito.mock(KubernetesClient.class); + mockForLeaseSupport(client); + mockForLeaderSupport(client); + return client; + } + + private void mockForLeaseSupport(KubernetesClient client) { + Mockito.when(client.getApiResources("coordination.k8s.io/v1")) + .thenReturn( + new APIResourceListBuilder().withResources(new APIResourceBuilder().withKind("Lease").build()) + .build()); + + APIGroupList apiGroupList = new APIGroupListBuilder().addNewGroup() + .withVersions(new GroupVersionForDiscoveryBuilder().withGroupVersion("coordination.k8s.io/v1").build()) + .endGroup() + .build(); + + Mockito.when(client.getApiGroups()).thenReturn(apiGroupList); + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + private void mockForLeaderSupport(KubernetesClient client) { + + Lease lease = new LeaseBuilder().withNewMetadata() + .withName("spring-k8s-leader-election-lock") + .endMetadata() + .withSpec(new LeaseSpecBuilder().withHolderIdentity(HOLDER_IDENTITY) + .withLeaseDurationSeconds(1) + .withAcquireTime(ZonedDateTime.now()) + .withRenewTime(ZonedDateTime.now()) + .withLeaseTransitions(1) + .build()) + .build(); + + MixedOperation mixedOperation = Mockito.mock(MixedOperation.class); + Mockito.when(client.resources(Lease.class)).thenReturn(mixedOperation); + + Resource resource = Mockito.mock(Resource.class); + Mockito.when(resource.get()).thenReturn(lease); + + NonNamespaceOperation nonNamespaceOperation = Mockito.mock(NonNamespaceOperation.class); + Mockito.when(mixedOperation.inNamespace("default")).thenReturn(nonNamespaceOperation); + Mockito.when(nonNamespaceOperation.withName("spring-k8s-leader-election-lock")).thenReturn(resource); + + } + + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderOldAndNewImplementationTests.java b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderOldAndNewImplementationTests.java new file mode 100644 index 0000000000..e2e152ff44 --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderOldAndNewImplementationTests.java @@ -0,0 +1,244 @@ +/* + * 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 io.fabric8.kubernetes.api.model.APIGroupList; +import io.fabric8.kubernetes.api.model.APIGroupListBuilder; +import io.fabric8.kubernetes.api.model.APIResourceBuilder; +import io.fabric8.kubernetes.api.model.APIResourceListBuilder; +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.GroupVersionForDiscoveryBuilder; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import io.fabric8.kubernetes.client.dsl.Resource; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.cloud.kubernetes.commons.KubernetesCommonsAutoConfiguration; +import org.springframework.cloud.kubernetes.fabric8.Fabric8AutoConfiguration; +import org.springframework.cloud.kubernetes.fabric8.leader.Fabric8LeaderAutoConfiguration; +import org.springframework.context.annotation.Bean; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests that prove that previous and new leader implementation works based on the flags + * we set. + * + * @author wind57 + */ +class Fabric8LeaderOldAndNewImplementationTests { + + private ApplicationContextRunner applicationContextRunner; + + /** + * 
+	 *     - 'spring.cloud.kubernetes.leader.enabled'           is not set
+	 *     - 'spring.cloud.kubernetes.leader.election.enabled'  is not set
+	 *
+	 *     As such :
+	 *
+	 *     - 'Fabric8LeaderAutoConfiguration'                   is active
+	 *     - 'Fabric8LeaderElectionAutoConfiguration'           is not active
+	 *
+ */ + @Test + void noFlagsSet() { + setup("spring.main.cloud-platform=KUBERNETES"); + applicationContextRunner.run(context -> { + assertThat(context).hasSingleBean(Fabric8LeaderAutoConfiguration.class); + assertThat(context).doesNotHaveBean(Fabric8LeaderElectionAutoConfiguration.class); + }); + } + + /** + * 
+	 *     - 'spring.cloud.kubernetes.leader.enabled'          =  true
+	 *     - 'spring.cloud.kubernetes.leader.election.enabled'    is not set
+	 *
+	 *     As such :
+	 *
+	 *     - 'Fabric8LeaderAutoConfiguration'                   is active
+	 *     - 'Fabric8LeaderElectionAutoConfiguration'           is not active
+	 *
+ */ + @Test + void oldImplementationEnabled() { + setup("spring.main.cloud-platform=KUBERNETES", "spring.cloud.kubernetes.leader.enabled=true"); + applicationContextRunner.run(context -> { + assertThat(context).hasSingleBean(Fabric8LeaderAutoConfiguration.class); + assertThat(context).doesNotHaveBean(Fabric8LeaderElectionAutoConfiguration.class); + }); + } + + /** + * 
+	 *     - 'spring.cloud.kubernetes.leader.enabled'          = false
+	 *     - 'spring.cloud.kubernetes.leader.election.enabled'   is not set
+	 *
+	 *     As such :
+	 *
+	 *     - 'Fabric8LeaderAutoConfiguration'                   is not active
+	 *     - 'Fabric8LeaderElectionAutoConfiguration'           is not active
+	 *
+ */ + @Test + void oldImplementationDisabled() { + setup("spring.main.cloud-platform=KUBERNETES", "spring.cloud.kubernetes.leader.enabled=false"); + applicationContextRunner.run(context -> { + assertThat(context).doesNotHaveBean(Fabric8LeaderAutoConfiguration.class); + assertThat(context).doesNotHaveBean(Fabric8LeaderElectionAutoConfiguration.class); + }); + } + + /** + * 
+	 *     - 'spring.cloud.kubernetes.leader.enabled'            is not set
+	 *     - 'spring.cloud.kubernetes.leader.election.enabled' = false
+	 *
+	 *     As such :
+	 *
+	 *     - 'Fabric8LeaderAutoConfiguration'                   is active
+	 *     - 'Fabric8LeaderElectionAutoConfiguration'           is not active
+	 *
+ */ + @Test + void newImplementationDisabled() { + setup("spring.main.cloud-platform=KUBERNETES", "spring.cloud.kubernetes.leader.election.enabled=false"); + applicationContextRunner.run(context -> { + assertThat(context).hasSingleBean(Fabric8LeaderAutoConfiguration.class); + assertThat(context).doesNotHaveBean(Fabric8LeaderElectionAutoConfiguration.class); + }); + } + + /** + * 
+	 *     - 'spring.cloud.kubernetes.leader.enabled'            is not set
+	 *     - 'spring.cloud.kubernetes.leader.election.enabled' = true
+	 *
+	 *     As such :
+	 *
+	 *     - 'Fabric8LeaderAutoConfiguration'                   is not active
+	 *     - 'Fabric8LeaderElectionAutoConfiguration'           is active
+	 *
+ */ + @Test + void newImplementationEnabled() { + setup("spring.main.cloud-platform=KUBERNETES", "spring.cloud.kubernetes.leader.election.enabled=true"); + applicationContextRunner.run(context -> { + assertThat(context).doesNotHaveBean(Fabric8LeaderAutoConfiguration.class); + assertThat(context).hasSingleBean(Fabric8LeaderElectionAutoConfiguration.class); + }); + } + + /** + * 
+	 *     - 'spring.cloud.kubernetes.leader.enabled'          = false
+	 *     - 'spring.cloud.kubernetes.leader.election.enabled' = false
+	 *
+	 *     As such :
+	 *
+	 *     - 'Fabric8LeaderAutoConfiguration'                   is not active
+	 *     - 'Fabric8LeaderElectionAutoConfiguration'           is not active
+	 *
+ */ + @Test + void bothDisabled() { + setup("spring.main.cloud-platform=KUBERNETES", "spring.cloud.kubernetes.leader.enabled=false", + "spring.cloud.kubernetes.leader.election.enabled=false"); + applicationContextRunner.run(context -> { + assertThat(context).doesNotHaveBean(Fabric8LeaderAutoConfiguration.class); + assertThat(context).doesNotHaveBean(Fabric8LeaderElectionAutoConfiguration.class); + }); + } + + /** + * 
+	 *     - 'spring.cloud.kubernetes.leader.enabled'          = true
+	 *     - 'spring.cloud.kubernetes.leader.election.enabled' = true
+	 *
+	 *     As such :
+	 *
+	 *     - 'Fabric8LeaderAutoConfiguration'                   is not active
+	 *     - 'Fabric8LeaderElectionAutoConfiguration'           is active
+	 *
+	 *     You can't enable both of them, only the new one will work.
+	 *
+ */ + @Test + void bothEnabled() { + setup("spring.main.cloud-platform=KUBERNETES", "spring.cloud.kubernetes.leader.enabled=true", + "spring.cloud.kubernetes.leader.election.enabled=true"); + applicationContextRunner.run(context -> { + assertThat(context).doesNotHaveBean(Fabric8LeaderAutoConfiguration.class); + assertThat(context).hasSingleBean(Fabric8LeaderElectionAutoConfiguration.class); + }); + } + + private void setup(String... properties) { + applicationContextRunner = new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(Fabric8LeaderElectionCallbacksAutoConfiguration.class, + Fabric8AutoConfiguration.class, KubernetesCommonsAutoConfiguration.class, + Fabric8LeaderElectionAutoConfiguration.class, Fabric8LeaderAutoConfiguration.class)) + .withUserConfiguration(Fabric8LeaderOldAndNewImplementationTests.Configuration.class) + .withPropertyValues(properties); + } + + @TestConfiguration + static class Configuration { + + @Bean + @SuppressWarnings({ "rawtypes", "unchecked" }) + KubernetesClient mockKubernetesClient() { + KubernetesClient client = Mockito.mock(KubernetesClient.class); + + Mockito.when(client.getNamespace()).thenReturn("namespace"); + + MixedOperation mixedOperation = Mockito.mock(MixedOperation.class); + NonNamespaceOperation nonNamespaceOperation = Mockito.mock(NonNamespaceOperation.class); + Mockito.when(client.configMaps()).thenReturn(mixedOperation); + + Mockito.when(mixedOperation.inNamespace(Mockito.anyString())).thenReturn(nonNamespaceOperation); + Resource configMapResource = Mockito.mock(Resource.class); + Mockito.when(nonNamespaceOperation.withName(Mockito.anyString())).thenReturn(configMapResource); + + Mockito.when(client.pods()).thenReturn(mixedOperation); + PodResource podResource = Mockito.mock(PodResource.class); + Mockito.when(mixedOperation.withName(Mockito.anyString())).thenReturn(podResource); + + Mockito.when(client.getApiResources("coordination.k8s.io/v1")) + .thenReturn( + new APIResourceListBuilder().withResources(new APIResourceBuilder().withKind("Lease").build()) + .build()); + + APIGroupList apiGroupList = new APIGroupListBuilder().addNewGroup() + .withVersions(new GroupVersionForDiscoveryBuilder().withGroupVersion("coordination.k8s.io/v1").build()) + .endGroup() + .build(); + + Mockito.when(client.getApiGroups()).thenReturn(apiGroupList); + return client; + } + + } + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/it/Fabric8LeaderElectionSimpleITTest.java b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/it/Fabric8LeaderElectionSimpleITTest.java new file mode 100644 index 0000000000..83cc9053c0 --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/it/Fabric8LeaderElectionSimpleITTest.java @@ -0,0 +1,158 @@ +/* + * 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.it; + +import java.time.Duration; + +import io.fabric8.kubernetes.client.Config; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientBuilder; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.springframework.cloud.kubernetes.commons.leader.LeaderUtils; +import org.springframework.cloud.kubernetes.commons.leader.election.LeaderElectionProperties; +import org.testcontainers.k3s.K3sContainer; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import org.springframework.cloud.kubernetes.integration.tests.commons.Commons; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Primary; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * A very simple test where we are the sole participant in the leader + * election and everything goes fine from start to end. It's a happy path + * scenario test. + * + * @author wind57 + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = { "spring.main.cloud-platform=KUBERNETES", "spring.cloud.kubernetes.leader.election.enabled=true", + "spring.cloud.kubernetes.leader.election.wait-for-pod-ready=false" }) +@ExtendWith(OutputCaptureExtension.class) +class Fabric8LeaderElectionSimpleITTest { + + private static K3sContainer container; + + private static final MockedStatic LEADER_UTILS_MOCKED_STATIC = Mockito.mockStatic(LeaderUtils.class); + + @Autowired + private KubernetesClient kubernetesClient; + + @BeforeAll + static void beforeAll() { + container = Commons.container(); + container.start(); + + LEADER_UTILS_MOCKED_STATIC.when(LeaderUtils::hostName).thenReturn("simple-it"); + } + + @AfterAll + static void afterAll() { + container.stop(); + } + + @Test + void test(CapturedOutput output) { + + // we have become the leader + Awaitility.await() + .atMost(Duration.ofSeconds(60)) + .pollInterval(Duration.ofSeconds(1)) + .until(() -> output.getOut().contains("simple-it 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 : simple-it"); + + // 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 (simple-it)'"); + + // 4. we are the leader (comes from our code) + assertThat(output.getOut()).contains("Leader changed from null to simple-it"); + + // 5. wait until a renewal happens (comes from fabric code) + // this one means that we have extended our leadership + Awaitility.await() + .atMost(Duration.ofSeconds(15)) + .pollInterval(Duration.ofSeconds(1)) + .until(() -> output.getOut().contains( + "Attempting to renew leader lease 'LeaseLock: default - spring-k8s-leader-election-lock (simple-it)'")); + + + +// +// Lease lockLease = kubernetesClient.leases() +// .inNamespace("default") +// .withName("spring-k8s-leader-election-lock") +// .get(); +// ZonedDateTime currentAcquiredTime = lockLease.getSpec().getAcquireTime(); +// Assertions.assertThat(currentAcquiredTime).isNotNull(); +// Assertions.assertThat(lockLease.getSpec().getLeaseDurationSeconds()).isEqualTo(15); +// Assertions.assertThat(lockLease.getSpec().getLeaseTransitions()).isEqualTo(0); +// +// ZonedDateTime currentRenewalTime = lockLease.getSpec().getRenewTime(); +// Assertions.assertThat(currentRenewalTime).isNotNull(); +// +// // renew happened, we renew by default on every two seconds +// Awaitility.await() +// .pollInterval(Duration.ofSeconds(1)) +// .atMost(Duration.ofSeconds(4)) +// .until(() -> !(currentRenewalTime.equals(kubernetesClient.leases() +// .inNamespace("default") +// .withName("spring-k8s-leader-election-lock") +// .get() +// .getSpec() +// .getRenewTime()))) + + + } + + @TestConfiguration + static class LocalConfiguration { + + @Bean + @Primary + KubernetesClient client() { + String kubeConfigYaml = container.getKubeConfigYaml(); + Config config = Config.fromKubeconfig(kubeConfigYaml); + return new KubernetesClientBuilder().withConfig(config).build(); + } + + } + + // test with pod ready + // simulate that we lose leadership, must re-try + +} diff --git a/spring-cloud-kubernetes-fabric8-leader/src/test/resources/logback-test.xml b/spring-cloud-kubernetes-fabric8-leader/src/test/resources/logback-test.xml new file mode 100644 index 0000000000..55654605fa --- /dev/null +++ b/spring-cloud-kubernetes-fabric8-leader/src/test/resources/logback-test.xml @@ -0,0 +1,6 @@ + + + + + + diff --git a/spring-cloud-kubernetes-integration-tests/spring-cloud-kubernetes-k8s-client-kafka-configmap-reload/kafka-configmap-test-app/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/multiple/apps/ConfigurationWatcherBusKafkaIT.java b/spring-cloud-kubernetes-integration-tests/spring-cloud-kubernetes-k8s-client-kafka-configmap-reload/kafka-configmap-test-app/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/multiple/apps/ConfigurationWatcherBusKafkaIT.java index f80e5c3313..aae56e7519 100644 --- a/spring-cloud-kubernetes-integration-tests/spring-cloud-kubernetes-k8s-client-kafka-configmap-reload/kafka-configmap-test-app/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/multiple/apps/ConfigurationWatcherBusKafkaIT.java +++ b/spring-cloud-kubernetes-integration-tests/spring-cloud-kubernetes-k8s-client-kafka-configmap-reload/kafka-configmap-test-app/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/multiple/apps/ConfigurationWatcherBusKafkaIT.java @@ -72,8 +72,6 @@ static void beforeAll() throws Exception { Commons.loadSpringCloudKubernetesImage(CONFIG_WATCHER_APP_IMAGE, K3S); Images.loadKafka(K3S); - - util = new Util(K3S); util.setUp(NAMESPACE); } diff --git a/spring-cloud-kubernetes-test-support/src/main/java/org/springframework/cloud/kubernetes/integration/tests/commons/FixedPortsK3sContainer.java b/spring-cloud-kubernetes-test-support/src/main/java/org/springframework/cloud/kubernetes/integration/tests/commons/FixedPortsK3sContainer.java index 88096eb2d8..2535eae8c7 100644 --- a/spring-cloud-kubernetes-test-support/src/main/java/org/springframework/cloud/kubernetes/integration/tests/commons/FixedPortsK3sContainer.java +++ b/spring-cloud-kubernetes-test-support/src/main/java/org/springframework/cloud/kubernetes/integration/tests/commons/FixedPortsK3sContainer.java @@ -37,7 +37,7 @@ final class FixedPortsK3sContainer extends K3sContainer { /** * Test containers exposed ports. */ - private static final int[] EXPOSED_PORTS = new int[] { 80, 6443, 8080, 8888, 9092, 32321 }; + private static final int[] EXPOSED_PORTS = new int[] { 80, 6443, 8080, 8888, 9092, 32321, 32322 }; /** * Rancher version to use for test-containers. diff --git a/spring-cloud-kubernetes-test-support/src/main/resources/setup/role.yaml b/spring-cloud-kubernetes-test-support/src/main/resources/setup/role.yaml index 4a397a6920..a159e008fc 100644 --- a/spring-cloud-kubernetes-test-support/src/main/resources/setup/role.yaml +++ b/spring-cloud-kubernetes-test-support/src/main/resources/setup/role.yaml @@ -4,6 +4,6 @@ metadata: namespace: default name: namespace-reader rules: - - apiGroups: ["", "extensions", "apps", "discovery.k8s.io"] - resources: ["configmaps", "pods", "services", "endpoints", "secrets", "endpointslices"] - verbs: ["get", "list", "watch"] + - apiGroups: ["", "extensions", "apps", "discovery.k8s.io", "coordination.k8s.io"] + resources: ["configmaps", "pods", "services", "endpoints", "secrets", "endpointslices", "leases"] + verbs: ["get", "list", "watch", "create", "update", "patch"]