From dce324961c49cf4a581bfe916dd32d90de2d294e Mon Sep 17 00:00:00 2001 From: Gytis Trikleris Date: Wed, 22 Aug 2018 10:25:27 +0200 Subject: [PATCH] Synchronize and test leader watchers --- .../leader/LeaderRecordWatcher.java | 36 +++-- .../leader/PodReadinessWatcher.java | 46 ++++-- .../leader/LeaderRecordWatcherTest.java | 134 +++++++++++++++++ .../leader/PodReadinessWatcherTest.java | 138 ++++++++++++++++++ 4 files changed, 327 insertions(+), 27 deletions(-) create mode 100644 spring-cloud-kubernetes-leader/src/test/java/org/springframework/cloud/kubernetes/leader/LeaderRecordWatcherTest.java create mode 100644 spring-cloud-kubernetes-leader/src/test/java/org/springframework/cloud/kubernetes/leader/PodReadinessWatcherTest.java diff --git a/spring-cloud-kubernetes-leader/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderRecordWatcher.java b/spring-cloud-kubernetes-leader/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderRecordWatcher.java index 70fe86b8..41bdc63c 100644 --- a/spring-cloud-kubernetes-leader/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderRecordWatcher.java +++ b/spring-cloud-kubernetes-leader/src/main/java/org/springframework/cloud/kubernetes/leader/LeaderRecordWatcher.java @@ -32,6 +32,8 @@ public class LeaderRecordWatcher implements Watcher { private static final Logger LOGGER = LoggerFactory.getLogger(LeaderRecordWatcher.class); + private final Object lock = new Object(); + private final LeadershipController leadershipController; private final LeaderProperties leaderProperties; @@ -49,20 +51,28 @@ public class LeaderRecordWatcher implements Watcher { public void start() { if (watch == null) { - LOGGER.debug("Starting leader record watcher"); - watch = kubernetesClient - .configMaps() - .inNamespace(leaderProperties.getNamespace(kubernetesClient.getNamespace())) - .withName(leaderProperties.getConfigMapName()) - .watch(this); + synchronized (lock) { + if (watch == null) { + LOGGER.debug("Starting leader record watcher"); + watch = kubernetesClient + .configMaps() + .inNamespace(leaderProperties.getNamespace(kubernetesClient.getNamespace())) + .withName(leaderProperties.getConfigMapName()) + .watch(this); + } + } } } public void stop() { if (watch != null) { - LOGGER.debug("Stopping leader record watcher"); - watch.close(); - watch = null; + synchronized (lock) { + if (watch != null) { + LOGGER.debug("Stopping leader record watcher"); + watch.close(); + watch = null; + } + } } } @@ -78,9 +88,11 @@ public class LeaderRecordWatcher implements Watcher { @Override public void onClose(KubernetesClientException cause) { if (cause != null) { - LOGGER.warn("Watcher stopped unexpectedly, will restart", cause); - watch = null; - start(); + synchronized (lock) { + LOGGER.warn("Watcher stopped unexpectedly, will restart", cause); + watch = null; + start(); + } } } } diff --git a/spring-cloud-kubernetes-leader/src/main/java/org/springframework/cloud/kubernetes/leader/PodReadinessWatcher.java b/spring-cloud-kubernetes-leader/src/main/java/org/springframework/cloud/kubernetes/leader/PodReadinessWatcher.java index 0a04cf98..a6fd2724 100644 --- a/spring-cloud-kubernetes-leader/src/main/java/org/springframework/cloud/kubernetes/leader/PodReadinessWatcher.java +++ b/spring-cloud-kubernetes-leader/src/main/java/org/springframework/cloud/kubernetes/leader/PodReadinessWatcher.java @@ -35,6 +35,8 @@ public class PodReadinessWatcher implements Watcher { private static final Logger LOGGER = LoggerFactory.getLogger(PodReadinessWatcher.class); + private final Object lock = new Object(); + private final String podName; private final KubernetesClient kubernetesClient; @@ -54,20 +56,28 @@ public class PodReadinessWatcher implements Watcher { public void start() { if (watch == null) { - LOGGER.debug("Starting pod readiness watcher for '{}'", podName); - PodResource podResource = kubernetesClient - .pods() - .withName(podName); - previousState = podResource.isReady(); - watch = podResource.watch(this); + synchronized (lock) { + if (watch == null) { + LOGGER.debug("Starting pod readiness watcher for '{}'", podName); + PodResource podResource = kubernetesClient + .pods() + .withName(podName); + previousState = podResource.isReady(); + watch = podResource.watch(this); + } + } } } public void stop() { if (watch != null) { - LOGGER.debug("Stopping pod readiness watcher for '{}'", podName); - watch.close(); - watch = null; + synchronized (lock) { + if (watch != null) { + LOGGER.debug("Stopping pod readiness watcher for '{}'", podName); + watch.close(); + watch = null; + } + } } } @@ -75,18 +85,24 @@ public class PodReadinessWatcher implements Watcher { public void eventReceived(Action action, Pod pod) { boolean currentState = Readiness.isPodReady(pod); if (previousState != currentState) { - LOGGER.debug("'{}' readiness status changed to '{}', triggering leadership update", podName, currentState); - previousState = currentState; - leadershipController.update(); + synchronized (lock) { + if (previousState != currentState) { + LOGGER.debug("'{}' readiness status changed to '{}', triggering leadership update", podName, currentState); + previousState = currentState; + leadershipController.update(); + } + } } } @Override public void onClose(KubernetesClientException cause) { if (cause != null) { - LOGGER.warn("Watcher stopped unexpectedly, will restart", cause); - watch = null; - start(); + synchronized (lock) { + LOGGER.warn("Watcher stopped unexpectedly, will restart", cause); + watch = null; + start(); + } } } } diff --git a/spring-cloud-kubernetes-leader/src/test/java/org/springframework/cloud/kubernetes/leader/LeaderRecordWatcherTest.java b/spring-cloud-kubernetes-leader/src/test/java/org/springframework/cloud/kubernetes/leader/LeaderRecordWatcherTest.java new file mode 100644 index 00000000..85add96b --- /dev/null +++ b/spring-cloud-kubernetes-leader/src/test/java/org/springframework/cloud/kubernetes/leader/LeaderRecordWatcherTest.java @@ -0,0 +1,134 @@ +/* + * Copyright 2013-2018 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. + * 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 io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.ConfigMapList; +import io.fabric8.kubernetes.api.model.DoneableConfigMap; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.Watch; +import io.fabric8.kubernetes.client.Watcher; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.Resource; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** + * @author Gytis Trikleris + */ +@RunWith(MockitoJUnitRunner.class) +public class LeaderRecordWatcherTest { + + @Mock + private LeaderProperties mockLeaderProperties; + + @Mock + private LeadershipController mockLeadershipController; + + @Mock + private KubernetesClient mockKubernetesClient; + + @Mock + private MixedOperation> + mockConfigMapsOperation; + + @Mock + private NonNamespaceOperation> + mockInNamespaceOperation; + + @Mock + private Resource mockWithNameResource; + + @Mock + private Watch mockWatch; + + @Mock + private ConfigMap mockConfigMap; + + @Mock + private KubernetesClientException mockKubernetesClientException; + + private LeaderRecordWatcher watcher; + + @Before + public void before() { + watcher = new LeaderRecordWatcher(mockLeaderProperties, mockLeadershipController, mockKubernetesClient); + + given(mockKubernetesClient.configMaps()).willReturn(mockConfigMapsOperation); + given(mockConfigMapsOperation.inNamespace(null)).willReturn(mockInNamespaceOperation); + given(mockInNamespaceOperation.withName(null)).willReturn(mockWithNameResource); + given(mockWithNameResource.watch(watcher)).willReturn(mockWatch); + } + + @Test + public void shouldStartOnce() { + watcher.start(); + watcher.start(); + + verify(mockWithNameResource).watch(watcher); + } + + @Test + public void shouldStopOnce() { + watcher.start(); + watcher.stop(); + watcher.stop(); + + verify(mockWatch).close(); + } + + @Test + public void shouldHandleEvent() { + watcher.eventReceived(Watcher.Action.ADDED, mockConfigMap); + watcher.eventReceived(Watcher.Action.DELETED, mockConfigMap); + watcher.eventReceived(Watcher.Action.MODIFIED, mockConfigMap); + + verify(mockLeadershipController, times(3)).update(); + } + + @Test + public void shouldIgnoreErrorEvent() { + watcher.eventReceived(Watcher.Action.ERROR, mockConfigMap); + + verify(mockLeadershipController, times(0)).update(); + } + + @Test + public void shouldHandleClose() { + watcher.onClose(mockKubernetesClientException); + + verify(mockWithNameResource).watch(watcher); + } + + @Test + public void shouldIgnoreCloseWithoutCause() { + watcher.onClose(null); + + verify(mockWithNameResource, times(0)).watch(watcher); + } + +} diff --git a/spring-cloud-kubernetes-leader/src/test/java/org/springframework/cloud/kubernetes/leader/PodReadinessWatcherTest.java b/spring-cloud-kubernetes-leader/src/test/java/org/springframework/cloud/kubernetes/leader/PodReadinessWatcherTest.java new file mode 100644 index 00000000..34f577ef --- /dev/null +++ b/spring-cloud-kubernetes-leader/src/test/java/org/springframework/cloud/kubernetes/leader/PodReadinessWatcherTest.java @@ -0,0 +1,138 @@ +/* + * Copyright 2013-2018 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. + * 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 io.fabric8.kubernetes.api.model.DoneablePod; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.api.model.PodCondition; +import io.fabric8.kubernetes.api.model.PodList; +import io.fabric8.kubernetes.api.model.PodStatus; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.Watch; +import io.fabric8.kubernetes.client.Watcher; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** + * @author Gytis Trikleris + */ +@RunWith(MockitoJUnitRunner.class) +public class PodReadinessWatcherTest { + + private static final String POD_NAME = "test-pod"; + + @Mock + private LeadershipController mockLeadershipController; + + @Mock + private KubernetesClient mockKubernetesClient; + + @Mock + private MixedOperation> mockPodsOperation; + + @Mock + private PodResource mockPodResource; + + @Mock + private Pod mockPod; + + @Mock + private PodStatus mockPodStatus; + + @Mock + private Watch mockWatch; + + @Mock + private KubernetesClientException mockKubernetesClientException; + + private PodReadinessWatcher watcher; + + @Before + public void before() { + watcher = new PodReadinessWatcher(POD_NAME, mockKubernetesClient, mockLeadershipController); + + given(mockKubernetesClient.pods()).willReturn(mockPodsOperation); + given(mockPodsOperation.withName(POD_NAME)).willReturn(mockPodResource); + given(mockPodResource.watch(watcher)).willReturn(mockWatch); + } + + @Test + public void shouldStartOnce() { + watcher.start(); + watcher.start(); + + verify(mockPodResource).watch(watcher); + } + + @Test + public void shouldStopOnce() { + watcher.start(); + watcher.stop(); + watcher.stop(); + + verify(mockWatch).close(); + } + + @Test + public void shouldHandleEventWithStateChange() { + given(mockPodResource.isReady()).willReturn(true); + given(mockPod.getStatus()).willReturn(mockPodStatus); + + watcher.start(); + watcher.eventReceived(Watcher.Action.ADDED, mockPod); + + verify(mockLeadershipController).update(); + } + + @Test + public void shouldIgnoreEventIfStateDoesNotChange() { + given(mockPod.getStatus()).willReturn(mockPodStatus); + + watcher.start(); + watcher.eventReceived(Watcher.Action.ADDED, mockPod); + + verify(mockLeadershipController, times(0)).update(); + } + + @Test + public void shouldHandleClose() { + watcher.onClose(mockKubernetesClientException); + + verify(mockPodResource).watch(watcher); + } + + @Test + public void shouldIgnoreCloseWithoutCause() { + watcher.onClose(null); + + verify(mockPodResource, times(0)).watch(watcher); + } + +}