SpringCloud/spring-cloud-kubernetes
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fabric leader clean up 4 (#1644)

Browse Source
This commit is contained in:
erabii
2024-04-24 21:52:01 +03:00
committed by GitHub
parent af02c5fd16
commit 1c819d00b1
3 changed files with 60 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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();
}
}
}

16
spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8LeaderAutoConfiguration.java
View File

@@ -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,

72
spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/Fabric8PodReadinessWatcher.java
View File

@@ -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<Pod> {
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();
}
});
}
}