From 5febe5e44089d8ab8d3853b55505a79c5182b09e Mon Sep 17 00:00:00 2001 From: Oleg Vyukov Date: Thu, 5 Jul 2018 12:44:27 +0300 Subject: [PATCH] KubernetesCatalogWatch endpoint based --- .../discovery/KubernetesCatalogWatch.java | 34 ++++-- ...netesDiscoveryClientAutoConfiguration.java | 2 +- .../discovery/KubernetesCatalogWatchTest.java | 108 ++++++++++++++++-- 3 files changed, 125 insertions(+), 19 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 43a1870f..8eb89eb4 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 @@ -16,6 +16,8 @@ package org.springframework.cloud.kubernetes.discovery; +import io.fabric8.kubernetes.api.model.Endpoints; +import io.fabric8.kubernetes.client.KubernetesClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.cloud.client.discovery.event.HeartbeatEvent; @@ -25,6 +27,7 @@ import org.springframework.scheduling.annotation.Scheduled; import java.util.List; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; /** @@ -34,12 +37,13 @@ public class KubernetesCatalogWatch implements ApplicationEventPublisherAware { private static final Logger logger = LoggerFactory.getLogger(KubernetesCatalogWatch.class); - private final KubernetesDiscoveryClient kubernetesDiscoveryClient; - private final AtomicReference> catalogServicesState = new AtomicReference<>(); + private final KubernetesClient kubernetesClient; + private final AtomicReference> catalogEndpointsState = new AtomicReference<>(); private ApplicationEventPublisher publisher; - public KubernetesCatalogWatch(KubernetesDiscoveryClient kubernetesDiscoveryClient) { - this.kubernetesDiscoveryClient = kubernetesDiscoveryClient; + + public KubernetesCatalogWatch(KubernetesClient kubernetesClient) { + this.kubernetesClient = kubernetesClient; } @Override @@ -50,19 +54,27 @@ public class KubernetesCatalogWatch implements ApplicationEventPublisherAware { @Scheduled(fixedDelayString = "${spring.cloud.kubernetes.discovery.catalogServicesWatchDelay:30000}") public void catalogServicesWatch() { try { - List previousState = catalogServicesState.get(); + List previousState = catalogEndpointsState.get(); - List services = kubernetesDiscoveryClient.getServices(); + //not all pods participate in the service discovery. only those that have endpoints. + List endpoints = kubernetesClient.endpoints().list().getItems(); + List endpointsPodNames = + endpoints.stream() + .flatMap(endpoint -> endpoint.getSubsets().stream()) + .flatMap(subset -> subset.getAddresses().stream()) + .map(endpointAddress -> endpointAddress.getTargetRef().getName()) // pod name unique in namespace + .sorted(String::compareTo).collect(Collectors.toList()); - services.sort(String::compareTo); - catalogServicesState.set(services); + catalogEndpointsState.set(endpointsPodNames); - if (!services.equals(previousState)) { - logger.trace("Received services update from kubernetesDiscoveryClient: {}", services); - publisher.publishEvent(new HeartbeatEvent(this, services)); + if (!endpointsPodNames.equals(previousState)) { + logger.trace("Received endpoints update from kubernetesClient: {}", endpointsPodNames); + publisher.publishEvent(new HeartbeatEvent(this, endpointsPodNames)); } } catch (Exception e) { logger.error("Error watching Kubernetes Services", e); } } + + } diff --git a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientAutoConfiguration.java b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientAutoConfiguration.java index 7295daeb..10d38581 100644 --- a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientAutoConfiguration.java +++ b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientAutoConfiguration.java @@ -39,7 +39,7 @@ public class KubernetesDiscoveryClientAutoConfiguration { @Bean @ConditionalOnMissingBean @ConditionalOnProperty(name = "spring.cloud.kubernetes.discovery.catalog-services-watch.enabled", matchIfMissing = true) - public KubernetesCatalogWatch kubernetesCatalogWatch(KubernetesDiscoveryClient client) { + public KubernetesCatalogWatch kubernetesCatalogWatch(KubernetesClient client) { return new KubernetesCatalogWatch(client); } } 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 4dfe0b30..41e7dde2 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 @@ -1,8 +1,14 @@ package org.springframework.cloud.kubernetes.discovery; +import io.fabric8.kubernetes.api.model.*; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.dsl.Resource; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; @@ -10,10 +16,14 @@ import org.springframework.cloud.client.discovery.event.HeartbeatEvent; import org.springframework.context.ApplicationEventPublisher; import java.util.Arrays; +import java.util.Collections; import java.util.List; -import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; -import static org.junit.Assert.*; +import static java.util.Arrays.stream; +import static org.hamcrest.CoreMatchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; import static org.mockito.Mockito.*; /** @@ -23,11 +33,17 @@ import static org.mockito.Mockito.*; public class KubernetesCatalogWatchTest { @Mock - private KubernetesDiscoveryClient kubernetesDiscoveryClient; + private KubernetesClient kubernetesClient; @Mock private ApplicationEventPublisher applicationEventPublisher; + @Mock + private MixedOperation> endpointsOperation; + + @Captor + private ArgumentCaptor heartbeatEventArgumentCaptor; + @InjectMocks private KubernetesCatalogWatch underTest; @@ -37,10 +53,11 @@ public class KubernetesCatalogWatchTest { } @Test - public void testRandomOrder() throws Exception { - final List services = Arrays.asList("api", "api", "other"); - final List shuffleServices = Arrays.asList("api", "other", "api"); - when(kubernetesDiscoveryClient.getServices()).thenReturn(services); + public void testRandomOrderChangePods() throws Exception { + when(endpointsOperation.list()) + .thenReturn(createSingleEndpointEndpointListByPodName("api-pod", "other-pod")) + .thenReturn(createSingleEndpointEndpointListByPodName("other-pod", "api-pod")); + when(kubernetesClient.endpoints()).thenReturn(endpointsOperation); underTest.catalogServicesWatch(); // second execution on shuffleServices @@ -48,4 +65,81 @@ public class KubernetesCatalogWatchTest { verify(applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); } + + @Test + public void testRandomOrderChangeServices() throws Exception { + when(endpointsOperation.list()) + .thenReturn(createEndpointsListByServiceName("api-service", "other-service")) + .thenReturn(createEndpointsListByServiceName("other-service", "api-service")); + when(kubernetesClient.endpoints()).thenReturn(endpointsOperation); + + underTest.catalogServicesWatch(); + // second execution on shuffleServices + underTest.catalogServicesWatch(); + + verify(applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); + } + + @Test + public void testEventBody() throws Exception { + when(endpointsOperation.list()) + .thenReturn(createSingleEndpointEndpointListByPodName("api-pod", "other-pod")); + when(kubernetesClient.endpoints()).thenReturn(endpointsOperation); + + underTest.catalogServicesWatch(); + + verify(applicationEventPublisher).publishEvent(heartbeatEventArgumentCaptor.capture()); + + HeartbeatEvent event = heartbeatEventArgumentCaptor.getValue(); + assertThat(event.getValue(), instanceOf(List.class)); + + List expectedPodsList = Arrays.asList("api-pod", "other-pod"); + assertEquals(expectedPodsList, event.getValue()); + } + + + private EndpointsList createEndpointsListByServiceName(String... serviceNames) { + List endpoints = stream(serviceNames) + .map(s -> createEndpointsByPodName(s + "-singlePodUniqueId")) + .collect(Collectors.toList()); + + EndpointsList endpointsList = new EndpointsList(); + endpointsList.setItems(endpoints); + return endpointsList; + } + + private EndpointsList createSingleEndpointEndpointListByPodName(String... podNames) { + Endpoints endpoints = new Endpoints(); + endpoints.setSubsets(createSubsetsByPodName(podNames)); + + EndpointsList endpointsList = new EndpointsList(); + endpointsList.setItems(Collections.singletonList(endpoints)); + return endpointsList; + } + + + private Endpoints createEndpointsByPodName(String podName) { + Endpoints endpoints = new Endpoints(); + endpoints.setSubsets(createSubsetsByPodName(podName)); + return endpoints; + } + + + private List createSubsetsByPodName(String... names) { + EndpointSubset endpointSubset = new EndpointSubset(); + endpointSubset.setAddresses(createEndpointAddressByPodNames(names)); + return Collections.singletonList(endpointSubset); + } + + private List createEndpointAddressByPodNames(String[] names) { + return stream(names).map(name -> { + ObjectReference podRef = new ObjectReference(); + podRef.setName(name); + EndpointAddress endpointAddress = new EndpointAddress(); + endpointAddress.setTargetRef(podRef); + return endpointAddress; + }).collect(Collectors.toList()); + } + + }