Refactor lock to leader

spring-cloud-kubernetes-leader/pom.xml

@@ -0,0 +1,79 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Copyright (C) 2018 to the original 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
+
+            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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-kubernetes</artifactId>
+		<version>0.3.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<artifactId>spring-cloud-kubernetes-leader</artifactId>
+	<name>Spring Cloud Kubernetes :: Leader</name>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-kubernetes-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.jboss.arquillian.junit</groupId>
+			<artifactId>arquillian-junit-standalone</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.arquillian.cube</groupId>
+			<artifactId>arquillian-cube-kubernetes</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.arquillian.cube</groupId>
+			<artifactId>arquillian-cube-requirement</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.assertj</groupId>
+			<artifactId>assertj-core</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-core</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-failsafe-plugin</artifactId>
+				<version>${maven-failsafe-plugin.version}</version>
+			</plugin>
+		</plugins>
+	</build>
+</project>

spring-cloud-kubernetes-leader/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 <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;
+	}
+
+}

spring-cloud-kubernetes-leader/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 <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;
+	}
+}

spring-cloud-kubernetes-leader/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 <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()));
+	}
+
+}