From 0951e892cebd3fbc9f5b0d651dffe0cfb6e46b7a Mon Sep 17 00:00:00 2001 From: Haytham Mohamed Date: Thu, 9 Jul 2020 18:16:37 -0500 Subject: [PATCH] addressing related fix to #472 --- .../discovery/KubernetesCatalogWatch.java | 13 +- ...bernetesCatalogWatchAutoConfiguration.java | 5 +- .../discovery/KubernetesCatalogWatchTest.java | 150 ++++++++++++++++++ 3 files changed, 163 insertions(+), 5 deletions(-) diff --git a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatch.java b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatch.java index 03511b14..416c5371 100644 --- a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatch.java +++ b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatch.java @@ -45,12 +45,16 @@ public class KubernetesCatalogWatch implements ApplicationEventPublisherAware { private final KubernetesClient kubernetesClient; + private final KubernetesDiscoveryProperties properties; + private final AtomicReference> catalogEndpointsState = new AtomicReference<>(); private ApplicationEventPublisher publisher; - public KubernetesCatalogWatch(KubernetesClient kubernetesClient) { + public KubernetesCatalogWatch(KubernetesClient kubernetesClient, + KubernetesDiscoveryProperties properties) { this.kubernetesClient = kubernetesClient; + this.properties = properties; } @Override @@ -66,8 +70,11 @@ public class KubernetesCatalogWatch implements ApplicationEventPublisherAware { // not all pods participate in the service discovery. only those that have // endpoints. - List endpoints = this.kubernetesClient.endpoints().list() - .getItems(); + List endpoints = this.properties.isAllNamespaces() + ? this.kubernetesClient.endpoints().inAnyNamespace() + .withLabels(properties.getServiceLabels()).list().getItems() + : this.kubernetesClient.endpoints() + .withLabels(properties.getServiceLabels()).list().getItems(); List endpointsPodNames = endpoints.stream().map(Endpoints::getSubsets) .filter(Objects::nonNull).flatMap(Collection::stream) .map(EndpointSubset::getAddresses).filter(Objects::nonNull) diff --git a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatchAutoConfiguration.java b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatchAutoConfiguration.java index 7fd8b229..c0cb2a5f 100644 --- a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatchAutoConfiguration.java +++ b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatchAutoConfiguration.java @@ -42,8 +42,9 @@ public class KubernetesCatalogWatchAutoConfiguration { @ConditionalOnProperty( name = "spring.cloud.kubernetes.discovery.catalog-services-watch.enabled", matchIfMissing = true) - public KubernetesCatalogWatch kubernetesCatalogWatch(KubernetesClient client) { - return new KubernetesCatalogWatch(client); + public KubernetesCatalogWatch kubernetesCatalogWatch(KubernetesClient client, + KubernetesDiscoveryProperties properties) { + return new KubernetesCatalogWatch(client, properties); } } diff --git a/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatchTest.java b/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatchTest.java index 8102e9e6..081dc2f1 100644 --- a/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatchTest.java +++ b/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatchTest.java @@ -44,6 +44,7 @@ import org.springframework.context.ApplicationEventPublisher; import static java.util.Arrays.stream; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.Mockito.any; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -54,6 +55,9 @@ import static org.mockito.Mockito.when; @RunWith(MockitoJUnitRunner.class) public class KubernetesCatalogWatchTest { + @Mock + private KubernetesDiscoveryProperties properties; + @Mock private KubernetesClient kubernetesClient; @@ -81,7 +85,31 @@ public class KubernetesCatalogWatchTest { createSingleEndpointEndpointListByPodName("api-pod", "other-pod")) .thenReturn(createSingleEndpointEndpointListByPodName("other-pod", "api-pod")); + when(this.properties.isAllNamespaces()).thenReturn(false); when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); + + this.underTest.catalogServicesWatch(); + // second execution on shuffleServices + this.underTest.catalogServicesWatch(); + + verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); + } + + @Test + public void testRandomOrderChangePodsAllNamespaces() throws Exception { + when(this.endpointsOperation.list()) + .thenReturn( + createSingleEndpointEndpointListByPodName("api-pod", "other-pod")) + .thenReturn(createSingleEndpointEndpointListByPodName("other-pod", + "api-pod")); + when(this.properties.isAllNamespaces()).thenReturn(true); + when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace()) + .thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); this.underTest.catalogServicesWatch(); // second execution on shuffleServices @@ -97,7 +125,31 @@ public class KubernetesCatalogWatchTest { createEndpointsListByServiceName("api-service", "other-service")) .thenReturn( createEndpointsListByServiceName("other-service", "api-service")); + when(this.properties.isAllNamespaces()).thenReturn(false); when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); + + this.underTest.catalogServicesWatch(); + // second execution on shuffleServices + this.underTest.catalogServicesWatch(); + + verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); + } + + @Test + public void testRandomOrderChangeServicesAllNamespaces() throws Exception { + when(this.endpointsOperation.list()) + .thenReturn( + createEndpointsListByServiceName("api-service", "other-service")) + .thenReturn( + createEndpointsListByServiceName("other-service", "api-service")); + when(this.properties.isAllNamespaces()).thenReturn(true); + when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace()) + .thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); this.underTest.catalogServicesWatch(); // second execution on shuffleServices @@ -110,7 +162,33 @@ public class KubernetesCatalogWatchTest { public void testEventBody() throws Exception { when(this.endpointsOperation.list()).thenReturn( createSingleEndpointEndpointListByPodName("api-pod", "other-pod")); + when(this.properties.isAllNamespaces()).thenReturn(false); when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); + + this.underTest.catalogServicesWatch(); + + verify(this.applicationEventPublisher) + .publishEvent(this.heartbeatEventArgumentCaptor.capture()); + + HeartbeatEvent event = this.heartbeatEventArgumentCaptor.getValue(); + assertThat(event.getValue()).isInstanceOf(List.class); + + List expectedPodsList = Arrays.asList("api-pod", "other-pod"); + assertThat(event.getValue()).isEqualTo(expectedPodsList); + } + + @Test + public void testEventBodyAllNamespaces() throws Exception { + when(this.endpointsOperation.list()).thenReturn( + createSingleEndpointEndpointListByPodName("api-pod", "other-pod")); + when(this.properties.isAllNamespaces()).thenReturn(true); + when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace()) + .thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); this.underTest.catalogServicesWatch(); @@ -129,8 +207,31 @@ public class KubernetesCatalogWatchTest { EndpointsList endpoints = createSingleEndpointEndpointListWithoutSubsets(); + when(this.properties.isAllNamespaces()).thenReturn(false); when(this.endpointsOperation.list()).thenReturn(endpoints); when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); + + this.underTest.catalogServicesWatch(); + // second execution on shuffleServices + this.underTest.catalogServicesWatch(); + + verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); + } + + @Test + public void testEndpointsWithoutSubsetsAllNamespaces() { + + EndpointsList endpoints = createSingleEndpointEndpointListWithoutSubsets(); + + when(this.properties.isAllNamespaces()).thenReturn(true); + when(this.endpointsOperation.list()).thenReturn(endpoints); + when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace()) + .thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); this.underTest.catalogServicesWatch(); // second execution on shuffleServices @@ -145,8 +246,32 @@ public class KubernetesCatalogWatchTest { EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod"); endpoints.getItems().get(0).getSubsets().get(0).setAddresses(null); + when(this.properties.isAllNamespaces()).thenReturn(false); when(this.endpointsOperation.list()).thenReturn(endpoints); when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); + + this.underTest.catalogServicesWatch(); + // second execution on shuffleServices + this.underTest.catalogServicesWatch(); + + verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); + } + + @Test + public void testEndpointsWithoutAddressesAllNamespaces() { + + EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod"); + endpoints.getItems().get(0).getSubsets().get(0).setAddresses(null); + + when(this.properties.isAllNamespaces()).thenReturn(true); + when(this.endpointsOperation.list()).thenReturn(endpoints); + when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace()) + .thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); this.underTest.catalogServicesWatch(); // second execution on shuffleServices @@ -162,8 +287,33 @@ public class KubernetesCatalogWatchTest { endpoints.getItems().get(0).getSubsets().get(0).getAddresses().get(0) .setTargetRef(null); + when(this.properties.isAllNamespaces()).thenReturn(false); when(this.endpointsOperation.list()).thenReturn(endpoints); when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); + + this.underTest.catalogServicesWatch(); + // second execution on shuffleServices + this.underTest.catalogServicesWatch(); + + verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); + } + + @Test + public void testEndpointsWithoutTargetRefsAllNamespaces() { + + EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod"); + endpoints.getItems().get(0).getSubsets().get(0).getAddresses().get(0) + .setTargetRef(null); + + when(this.properties.isAllNamespaces()).thenReturn(true); + when(this.endpointsOperation.list()).thenReturn(endpoints); + when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace()) + .thenReturn(this.endpointsOperation); + when(this.kubernetesClient.endpoints().inAnyNamespace().withLabels(anyMap())) + .thenReturn(this.endpointsOperation); this.underTest.catalogServicesWatch(); // second execution on shuffleServices