From de7db6e709b92770e7d410965b5c1db6faaecea4 Mon Sep 17 00:00:00 2001 From: Gytis Trikleris Date: Fri, 1 Jun 2018 11:54:21 +0200 Subject: [PATCH] Initial LeaderInitiator --- .../cloud/kubernetes/leader/Leader.java | 41 ++++ .../leader/LeaderConfiguration.java | 126 ++++++++++ .../kubernetes/leader/LeaderInitiator.java | 227 ++++++++++++++++++ 3 files changed, 394 insertions(+) create mode 100644 spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/Leader.java create mode 100644 spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderConfiguration.java create mode 100644 spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderInitiator.java diff --git a/spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/Leader.java b/spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/Leader.java new file mode 100644 index 00000000..fdf19aea --- /dev/null +++ b/spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/Leader.java @@ -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 Gytis Trikleris + */ +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; + } + +} diff --git a/spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderConfiguration.java b/spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderConfiguration.java new file mode 100644 index 00000000..0f646a7f --- /dev/null +++ b/spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderConfiguration.java @@ -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 Gytis Trikleris + */ +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 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 getLabels() { + return labels; + } + + public void setLabels(Map 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; + } +} diff --git a/spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderInitiator.java b/spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderInitiator.java new file mode 100644 index 00000000..079b6cd9 --- /dev/null +++ b/spring-cloud-kubernetes-lock/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderInitiator.java @@ -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 Gytis Trikleris + */ +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 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())); + } + +}