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 dda97528..015de232 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 @@ -18,6 +18,7 @@ package org.springframework.cloud.kubernetes.commons.leader; import java.net.InetAddress; import java.net.UnknownHostException; +import java.util.concurrent.locks.ReentrantLock; import org.springframework.cloud.kubernetes.commons.EnvReader; import org.springframework.util.StringUtils; @@ -44,4 +45,15 @@ public final class LeaderUtils { } } + public static void guarded(ReentrantLock lock, Runnable runnable) { + try { + lock.lock(); + runnable.run(); + } + finally { + lock.unlock(); + } + + } + } 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 a8fd8147..a76730f1 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 @@ -79,6 +79,16 @@ public class Fabric8LeaderAutoConfiguration { return new LeaderInfoContributor(fabric8LeadershipController, candidate); } + /** + * watches the readiness of the pod. In case of a readiness change, it has to go + * through leader process again. + */ + @Bean + public Fabric8PodReadinessWatcher hostPodWatcher(Candidate candidate, KubernetesClient kubernetesClient, + Fabric8LeadershipController fabric8LeadershipController) { + return new Fabric8PodReadinessWatcher(candidate.getId(), kubernetesClient, fabric8LeadershipController); + } + @Bean public Fabric8LeadershipController leadershipController(Candidate candidate, LeaderProperties leaderProperties, LeaderEventPublisher leaderEventPublisher, KubernetesClient kubernetesClient) { @@ -91,12 +101,6 @@ public class Fabric8LeaderAutoConfiguration { return new Fabric8LeaderRecordWatcher(leaderProperties, fabric8LeadershipController, kubernetesClient); } - @Bean - public Fabric8PodReadinessWatcher hostPodWatcher(Candidate candidate, KubernetesClient kubernetesClient, - Fabric8LeadershipController fabric8LeadershipController) { - return new Fabric8PodReadinessWatcher(candidate.getId(), kubernetesClient, fabric8LeadershipController); - } - @Bean(destroyMethod = "stop") public LeaderInitiator leaderInitiator(LeaderProperties leaderProperties, Fabric8LeadershipController fabric8LeadershipController, diff --git a/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8PodReadinessWatcher.java b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8PodReadinessWatcher.java index 43cb1d87..86eaa913 100644 --- a/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8PodReadinessWatcher.java +++ b/spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8PodReadinessWatcher.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2019 the original author or authors. + * 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. @@ -16,6 +16,8 @@ package org.springframework.cloud.kubernetes.fabric8.leader; +import java.util.concurrent.locks.ReentrantLock; + import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.client.KubernetesClient; import io.fabric8.kubernetes.client.Watch; @@ -23,19 +25,21 @@ import io.fabric8.kubernetes.client.Watcher; import io.fabric8.kubernetes.client.WatcherException; import io.fabric8.kubernetes.client.dsl.PodResource; import io.fabric8.kubernetes.client.readiness.Readiness; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.apache.commons.logging.LogFactory; import org.springframework.cloud.kubernetes.commons.leader.PodReadinessWatcher; +import org.springframework.core.log.LogAccessor; + +import static org.springframework.cloud.kubernetes.commons.leader.LeaderUtils.guarded; /** * @author Gytis Trikleris */ public class Fabric8PodReadinessWatcher implements PodReadinessWatcher, Watcher { - private static final Logger LOGGER = LoggerFactory.getLogger(Fabric8PodReadinessWatcher.class); + private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(Fabric8PodReadinessWatcher.class)); - private final Object lock = new Object(); + private final ReentrantLock lock = new ReentrantLock(); private final String podName; @@ -43,9 +47,9 @@ public class Fabric8PodReadinessWatcher implements PodReadinessWatcher, Watcher< private final Fabric8LeadershipController fabric8LeadershipController; - private boolean previousState; + private volatile boolean previousState; - private Watch watch; + private volatile Watch watch; public Fabric8PodReadinessWatcher(String podName, KubernetesClient kubernetesClient, Fabric8LeadershipController fabric8LeadershipController) { @@ -56,54 +60,54 @@ public class Fabric8PodReadinessWatcher implements PodReadinessWatcher, Watcher< @Override public void start() { - if (this.watch == null) { - synchronized (this.lock) { - if (this.watch == null) { - LOGGER.debug("Starting pod readiness watcher for '{}'", this.podName); - PodResource podResource = this.kubernetesClient.pods().withName(this.podName); - this.previousState = podResource.isReady(); - this.watch = podResource.watch(this); + if (watch == null) { + guarded(lock, () -> { + if (watch == null) { + LOGGER.debug(() -> "Starting pod readiness watcher for :" + podName); + PodResource podResource = kubernetesClient.pods().withName(this.podName); + previousState = podResource.isReady(); + watch = podResource.watch(this); } - } + }); } } @Override public void stop() { - if (this.watch != null) { - synchronized (this.lock) { - if (this.watch != null) { - LOGGER.debug("Stopping pod readiness watcher for '{}'", this.podName); - this.watch.close(); - this.watch = null; + if (watch != null) { + guarded(lock, () -> { + if (watch != null) { + LOGGER.debug(() -> "Stopping pod readiness watcher for :" + podName); + watch.close(); + watch = null; } - } + }); } } @Override public void eventReceived(Action action, Pod pod) { boolean currentState = Readiness.isPodReady(pod); - if (this.previousState != currentState) { - synchronized (this.lock) { - if (this.previousState != currentState) { - LOGGER.debug("'{}' readiness status changed to '{}', triggering leadership update", this.podName, - currentState); - this.previousState = currentState; - this.fabric8LeadershipController.update(); + if (previousState != currentState) { + guarded(lock, () -> { + if (previousState != currentState) { + LOGGER.debug(() -> "readiness status changed for pod : " + podName + " to state: " + currentState + + ", triggering leadership update"); + previousState = currentState; + fabric8LeadershipController.update(); } - } + }); } } @Override public void onClose(WatcherException cause) { if (cause != null) { - synchronized (this.lock) { - LOGGER.warn("Watcher stopped unexpectedly, will restart", cause); - this.watch = null; + guarded(lock, () -> { + LOGGER.warn(() -> "Watcher stopped unexpectedly, will restart" + cause.getMessage()); + watch = null; start(); - } + }); } }