Initial LeaderInitiator
This commit is contained in:
Gytis Trikleris
committed by
Ioannis Canellos
Ioannis Canellos
parent
085d2cd5c3
commit
de7db6e709
@@ -0,0 +1,41 @@
|
||||
/*
|
||||
* Copyright 2018 Red Hat, Inc, and individual contributors.
|
||||
*
|
||||
* 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
|
||||
*
|
||||
* http://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.leader;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:gytis@redhat.com">Gytis Trikleris</a>
|
||||
*/
|
||||
public class Leader {
|
||||
|
||||
private final String role;
|
||||
|
||||
private final String id;
|
||||
|
||||
public Leader(String role, String id) {
|
||||
this.role = role;
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String getRole() {
|
||||
return role;
|
||||
}
|
||||
|
||||
public String getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
/*
|
||||
* Copyright 2018 Red Hat, Inc, and individual contributors.
|
||||
*
|
||||
* 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
|
||||
*
|
||||
* http://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.leader;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:gytis@redhat.com">Gytis Trikleris</a>
|
||||
*/
|
||||
public class LeaderConfiguration {
|
||||
|
||||
private static final String DEFAULT_LEADER_ID_PREFIX = "leader.id.";
|
||||
|
||||
private static final boolean DEFAULT_AUTO_STARTUP = true;
|
||||
|
||||
private static final String DEFAULT_CONFIG_MAP_NAME = "leaders";
|
||||
|
||||
private static final long DEFAULT_LEASE_DURATION = 30000;
|
||||
|
||||
private static final long DEFAULT_RETRY_PERIOD = 5000;
|
||||
|
||||
private static final double DEFAULT_JITTER_FACTOR = 1.2;
|
||||
|
||||
private boolean autoStartup = DEFAULT_AUTO_STARTUP;
|
||||
|
||||
private String namespace;
|
||||
|
||||
private String configMapName = DEFAULT_CONFIG_MAP_NAME;
|
||||
|
||||
private Map<String, String> labels = Collections.emptyMap();
|
||||
|
||||
private String leaderIdPrefix = DEFAULT_LEADER_ID_PREFIX;
|
||||
|
||||
private long leaseDuration = DEFAULT_LEASE_DURATION;
|
||||
|
||||
private long retryPeriod = DEFAULT_RETRY_PERIOD;
|
||||
|
||||
private double jitterFactor = DEFAULT_JITTER_FACTOR;
|
||||
|
||||
public boolean isAutoStartup() {
|
||||
return autoStartup;
|
||||
}
|
||||
|
||||
public void setAutoStartup(boolean autoStartup) {
|
||||
this.autoStartup = autoStartup;
|
||||
}
|
||||
|
||||
public String getNamespace() {
|
||||
return namespace;
|
||||
}
|
||||
|
||||
public String getNamespace(String defaultValue) {
|
||||
if (namespace == null || namespace.isEmpty()) {
|
||||
return defaultValue;
|
||||
}
|
||||
|
||||
return namespace;
|
||||
}
|
||||
|
||||
public void setNamespace(String namespace) {
|
||||
this.namespace = namespace;
|
||||
}
|
||||
|
||||
public String getConfigMapName() {
|
||||
return configMapName;
|
||||
}
|
||||
|
||||
public void setConfigMapName(String configMapName) {
|
||||
this.configMapName = configMapName;
|
||||
}
|
||||
|
||||
public Map<String, String> getLabels() {
|
||||
return labels;
|
||||
}
|
||||
|
||||
public void setLabels(Map<String, String> labels) {
|
||||
this.labels = Collections.unmodifiableMap(labels);
|
||||
}
|
||||
|
||||
public String getLeaderIdPrefix() {
|
||||
return leaderIdPrefix;
|
||||
}
|
||||
|
||||
public void setLeaderIdPrefix(String leaderIdPrefix) {
|
||||
this.leaderIdPrefix = leaderIdPrefix;
|
||||
}
|
||||
|
||||
public long getLeaseDuration() {
|
||||
return leaseDuration;
|
||||
}
|
||||
|
||||
public void setLeaseDuration(long leaseDuration) {
|
||||
this.leaseDuration = leaseDuration;
|
||||
}
|
||||
|
||||
public long getRetryPeriod() {
|
||||
return retryPeriod;
|
||||
}
|
||||
|
||||
public void setRetryPeriod(long retryPeriod) {
|
||||
this.retryPeriod = retryPeriod;
|
||||
}
|
||||
|
||||
public double getJitterFactor() {
|
||||
return jitterFactor;
|
||||
}
|
||||
|
||||
public void setJitterFactor(double jitterFactor) {
|
||||
this.jitterFactor = jitterFactor;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,227 @@
|
||||
/*
|
||||
* Copyright 2018 Red Hat, Inc, and individual contributors.
|
||||
*
|
||||
* 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
|
||||
*
|
||||
* http://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.leader;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
import io.fabric8.kubernetes.api.model.ConfigMap;
|
||||
import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
|
||||
import io.fabric8.kubernetes.api.model.ObjectMeta;
|
||||
import io.fabric8.kubernetes.api.model.Pod;
|
||||
import io.fabric8.kubernetes.client.KubernetesClient;
|
||||
import org.springframework.context.SmartLifecycle;
|
||||
import org.springframework.integration.leader.Candidate;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:gytis@redhat.com">Gytis Trikleris</a>
|
||||
*/
|
||||
public class LeaderInitiator implements SmartLifecycle {
|
||||
|
||||
private final Lock lock = new ReentrantLock();
|
||||
|
||||
private final KubernetesClient kubernetesClient;
|
||||
|
||||
private final Candidate candidate;
|
||||
|
||||
private final LeaderConfiguration leaderConfiguration;
|
||||
|
||||
private ScheduledExecutorService scheduledExecutorService;
|
||||
|
||||
private String currentLeaderId; // TODO remove once events are implemented
|
||||
|
||||
public LeaderInitiator(KubernetesClient kubernetesClient, Candidate candidate,
|
||||
LeaderConfiguration leaderConfiguration) {
|
||||
this.kubernetesClient = kubernetesClient;
|
||||
this.candidate = candidate;
|
||||
this.leaderConfiguration = leaderConfiguration;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return leaderConfiguration.isAutoStartup();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
lock.lock();
|
||||
try {
|
||||
if (!isRunning()) {
|
||||
scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
|
||||
scheduledExecutorService.execute(this::update);
|
||||
}
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
lock.lock();
|
||||
try {
|
||||
if (isRunning()) {
|
||||
scheduledExecutorService.shutdown();
|
||||
scheduledExecutorService = null;
|
||||
}
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop(Runnable runnable) {
|
||||
stop();
|
||||
runnable.run();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return scheduledExecutorService != null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getPhase() {
|
||||
return 0; // TODO implement
|
||||
}
|
||||
|
||||
public String getCurrentLeaderId() {
|
||||
return currentLeaderId;
|
||||
}
|
||||
|
||||
private void update() {
|
||||
ConfigMap configMap = getConfigMap();
|
||||
Leader leader = getLeader(configMap);
|
||||
|
||||
if (leader == null) {
|
||||
System.out.println("Currently there is no leader, trying to become one");
|
||||
try {
|
||||
takeLeadership(configMap);
|
||||
currentLeaderId = candidate.getId();
|
||||
scheduleUpdate(leaderConfiguration.getLeaseDuration());
|
||||
} catch (Exception e) {
|
||||
// Leadership takeover failed, try again later
|
||||
System.out.println("Leadership takeover failed: " + e.getMessage());
|
||||
scheduleUpdate(leaderConfiguration.getRetryPeriod());
|
||||
}
|
||||
} else if (!isValidLeader(leader)) {
|
||||
System.out.println("Old leader is not valid any more, try to take over");
|
||||
try {
|
||||
takeLeadership(configMap);
|
||||
currentLeaderId = candidate.getId();
|
||||
scheduleUpdate(leaderConfiguration.getLeaseDuration()); // Is this needed?
|
||||
} catch (Exception e) {
|
||||
// Leadership takeover failed, try again later
|
||||
System.out.println("Leadership takeover failed: " + e.getMessage());
|
||||
scheduleUpdate(leaderConfiguration.getRetryPeriod());
|
||||
}
|
||||
} else if (!isCandidateALeader(leader)) {
|
||||
currentLeaderId = leader.getId();
|
||||
System.out.println(currentLeaderId + " is a leader, check in later");
|
||||
scheduleUpdate(leaderConfiguration.getLeaseDuration());
|
||||
} else {
|
||||
System.out.println("I am a leader, check in later");
|
||||
currentLeaderId = candidate.getId();
|
||||
scheduleUpdate(leaderConfiguration.getLeaseDuration()); // Is this needed?
|
||||
}
|
||||
}
|
||||
|
||||
private void takeLeadership(ConfigMap oldConfigMap) {
|
||||
String leaderIdKey = leaderConfiguration.getLeaderIdPrefix() + candidate.getRole();
|
||||
|
||||
if (oldConfigMap == null) {
|
||||
ConfigMap newConfigMap = new ConfigMapBuilder().withNewMetadata()
|
||||
.withName(leaderConfiguration.getConfigMapName())
|
||||
.addToLabels("provider", "spring-cloud-kubernetes")
|
||||
.addToLabels("kind", "locks")
|
||||
.endMetadata()
|
||||
.addToData(leaderIdKey, candidate.getId())
|
||||
.build();
|
||||
|
||||
kubernetesClient.configMaps()
|
||||
.inNamespace(leaderConfiguration.getNamespace(kubernetesClient.getNamespace()))
|
||||
.create(newConfigMap);
|
||||
} else {
|
||||
ConfigMap newConfigMap = new ConfigMapBuilder(oldConfigMap)
|
||||
.addToData(leaderIdKey, candidate.getId())
|
||||
.build();
|
||||
|
||||
kubernetesClient.configMaps()
|
||||
.inNamespace(leaderConfiguration.getNamespace(kubernetesClient.getNamespace()))
|
||||
.withName(leaderConfiguration.getConfigMapName())
|
||||
.lockResourceVersion(oldConfigMap.getMetadata().getResourceVersion())
|
||||
.replace(newConfigMap);
|
||||
}
|
||||
|
||||
System.out.println(candidate.getId() + " is now a leader");
|
||||
}
|
||||
|
||||
private ConfigMap getConfigMap() {
|
||||
try {
|
||||
return kubernetesClient.configMaps()
|
||||
.inNamespace(leaderConfiguration.getNamespace(kubernetesClient.getNamespace()))
|
||||
.withName(leaderConfiguration.getConfigMapName())
|
||||
.get();
|
||||
} catch (Exception e) {
|
||||
System.out.println("Failed to get a ConfigMap: " + e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private Leader getLeader(ConfigMap configMap) {
|
||||
if (configMap == null || configMap.getData() == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
Map<String, String> data = configMap.getData();
|
||||
String leaderIdKey = leaderConfiguration.getLeaderIdPrefix() + candidate.getRole();
|
||||
String leaderId = data.get(leaderIdKey);
|
||||
if (leaderId == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return new Leader(candidate.getRole(), leaderId);
|
||||
}
|
||||
|
||||
private void scheduleUpdate(long waitPeriod) {
|
||||
scheduledExecutorService.schedule(this::update, jitter(waitPeriod), TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
private long jitter(long num) {
|
||||
return (long) (num * (1 + Math.random() * (leaderConfiguration.getJitterFactor() - 1)));
|
||||
}
|
||||
|
||||
private boolean isCandidateALeader(Leader leader) {
|
||||
return candidate.getId().equals(leader.getId());
|
||||
}
|
||||
|
||||
private boolean isValidLeader(Leader leader) {
|
||||
return kubernetesClient.pods()
|
||||
.inNamespace(leaderConfiguration.getNamespace(kubernetesClient.getNamespace()))
|
||||
.withLabels(leaderConfiguration.getLabels())
|
||||
.list()
|
||||
.getItems()
|
||||
.stream()
|
||||
.map(Pod::getMetadata)
|
||||
.map(ObjectMeta::getName)
|
||||
.anyMatch(name -> name.equals(leader.getId()));
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user