KubernetesCatalogWatch endpoint based

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<List<String>> catalogServicesState = new AtomicReference<>();
+	private final KubernetesClient kubernetesClient;
+	private final AtomicReference<List<String>> 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<String> previousState = catalogServicesState.get();
+			List<String> previousState = catalogEndpointsState.get();
 
-			List<String> services = kubernetesDiscoveryClient.getServices();
+			//not all pods participate in the service discovery. only those that have endpoints.
+			List<Endpoints> endpoints = kubernetesClient.endpoints().list().getItems();
+			List<String> 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);
 		}
 	}
+
+
 }

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);
 	}
 }

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<Endpoints, EndpointsList, DoneableEndpoints, Resource<Endpoints, DoneableEndpoints>> endpointsOperation;
+
+	@Captor
+	private ArgumentCaptor<HeartbeatEvent> heartbeatEventArgumentCaptor;
+
 	@InjectMocks
 	private KubernetesCatalogWatch underTest;
 
@@ -37,10 +53,11 @@ public class KubernetesCatalogWatchTest {
 	}
 
 	@Test
-	public void testRandomOrder() throws Exception {
-		final List<String> services = Arrays.asList("api", "api", "other");
-		final List<String> 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<String> expectedPodsList = Arrays.asList("api-pod", "other-pod");
+		assertEquals(expectedPodsList, event.getValue());
+	}
+
+
+	private EndpointsList createEndpointsListByServiceName(String... serviceNames) {
+		List<Endpoints> 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<EndpointSubset> createSubsetsByPodName(String... names) {
+		EndpointSubset endpointSubset = new EndpointSubset();
+		endpointSubset.setAddresses(createEndpointAddressByPodNames(names));
+		return Collections.singletonList(endpointSubset);
+	}
+
+	private List<EndpointAddress> 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());
+	}
+
+
 }