From e78f3b176b005d1451c310394c1c18d6c7c4c2a6 Mon Sep 17 00:00:00 2001 From: Haytham Mohamed Date: Wed, 10 Jun 2020 11:13:55 -0500 Subject: [PATCH 1/3] discovered services filtered by labels --- .../discovery/KubernetesDiscoveryClient.java | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClient.java b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClient.java index 8d05b49e..4e259e9d 100644 --- a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClient.java +++ b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClient.java @@ -17,7 +17,6 @@ package org.springframework.cloud.kubernetes.discovery; import java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -102,13 +101,7 @@ public class KubernetesDiscoveryClient implements DiscoveryClient { Assert.notNull(serviceId, "[Assertion failed] - the object argument must not be null"); - List endpointsList = this.properties.isAllNamespaces() - ? this.client.endpoints().inAnyNamespace() - .withField("metadata.name", serviceId).list().getItems() - : Collections - .singletonList(this.client.endpoints().withName(serviceId).get()); - - List subsetsNS = endpointsList.stream() + List subsetsNS = this.getEndPointsList(serviceId).stream() .map(endpoints -> getSubsetsFromEndpoints(endpoints)) .collect(Collectors.toList()); @@ -122,6 +115,15 @@ public class KubernetesDiscoveryClient implements DiscoveryClient { return instances; } + private List getEndPointsList(String serviceId) { + return this.properties.isAllNamespaces() + ? this.client.endpoints().inAnyNamespace() + .withField("metadata.name", serviceId) + .withLabels(properties.getServiceLabels()).list().getItems() + : this.client.endpoints().withField("metadata.name", serviceId) + .withLabels(properties.getServiceLabels()).list().getItems(); + } + private List getNamespaceServiceInstances(EndpointSubsetNS es, String serviceId) { String namespace = es.getNamespace(); From 9df02f137decc230e00daebf2f8a9d4c40ca3780 Mon Sep 17 00:00:00 2001 From: Haytham Mohamed Date: Wed, 24 Jun 2020 22:42:20 -0500 Subject: [PATCH 2/3] add tests --- .../discovery/KubernetesDiscoveryClient.java | 2 +- ...etesDiscoveryClientFilterMetadataTest.java | 21 +- .../KubernetesDiscoveryClientTest.java | 312 +++++++++++------- 3 files changed, 218 insertions(+), 117 deletions(-) diff --git a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClient.java b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClient.java index 4e259e9d..ac8eb75d 100644 --- a/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClient.java +++ b/spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClient.java @@ -115,7 +115,7 @@ public class KubernetesDiscoveryClient implements DiscoveryClient { return instances; } - private List getEndPointsList(String serviceId) { + public List getEndPointsList(String serviceId) { return this.properties.isAllNamespaces() ? this.client.endpoints().inAnyNamespace() .withField("metadata.name", serviceId) diff --git a/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientFilterMetadataTest.java b/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientFilterMetadataTest.java index ae22dda2..c001699b 100644 --- a/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientFilterMetadataTest.java +++ b/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientFilterMetadataTest.java @@ -16,6 +16,7 @@ package org.springframework.cloud.kubernetes.discovery; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -34,6 +35,9 @@ import io.fabric8.kubernetes.api.model.ServiceList; import io.fabric8.kubernetes.api.model.ServicePort; import io.fabric8.kubernetes.api.model.ServicePortBuilder; import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.Watch; +import io.fabric8.kubernetes.client.Watcher; +import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable; import io.fabric8.kubernetes.client.dsl.MixedOperation; import io.fabric8.kubernetes.client.dsl.Resource; import io.fabric8.kubernetes.client.dsl.ServiceResource; @@ -49,7 +53,9 @@ import org.springframework.cloud.client.ServiceInstance; import static java.util.stream.Collectors.toList; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.entry; +import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.when; @RunWith(MockitoJUnitRunner.class) @@ -79,6 +85,9 @@ public class KubernetesDiscoveryClientFilterMetadataTest { @Mock private Resource endpointsResource; + @Mock + FilterWatchListDeletable> filter; + @InjectMocks private KubernetesDiscoveryClient underTest; @@ -360,10 +369,16 @@ public class KubernetesDiscoveryClientFilterMetadataTest { .addNewSubset().addAllToPorts(getEndpointPorts(ports)).addNewAddress() .endAddress().endSubset().build(); - when(this.endpointsResource.get()).thenReturn(endpoints); - when(this.endpointsOperation.withName(serviceId)) - .thenReturn(this.endpointsResource); when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); + + EndpointsList endpointsList = new EndpointsList(null, + Collections.singletonList(endpoints), null, null); + when(filter.list()).thenReturn(endpointsList); + when(filter.withLabels(anyMap())).thenReturn(filter); + + when(this.kubernetesClient.endpoints().withField(eq("metadata.name"), + eq(serviceId))).thenReturn(filter); + } private List getServicePorts(Map ports) { diff --git a/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientTest.java b/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientTest.java index 8eac4a77..fe05ee20 100644 --- a/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientTest.java +++ b/spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientTest.java @@ -19,6 +19,7 @@ package org.springframework.cloud.kubernetes.discovery; import java.util.ArrayList; import java.util.HashMap; import java.util.List; +import java.util.Map; import io.fabric8.kubernetes.api.model.Endpoints; import io.fabric8.kubernetes.api.model.EndpointsBuilder; @@ -61,106 +62,44 @@ public class KubernetesDiscoveryClientTest { } @Test - public void getInstancesShouldBeAbleToHandleEndpointsFromMultipleNamespaces() { - Endpoints endPoints1 = new EndpointsBuilder().withNewMetadata() - .withName("endpoint").withNamespace("test").endMetadata().addNewSubset() - .addNewAddress().withIp("ip1").withNewTargetRef().withUid("uid1") - .endTargetRef().endAddress().addNewPort("http", 80, "TCP").endSubset() - .build(); + public void getInstancesShouldBeAbleToHandleEndpointsSingleAddress() { + Map labels = new HashMap(); + labels.put("l", "v"); - Endpoints endpoints2 = new EndpointsBuilder().withNewMetadata() - .withName("endpoint").withNamespace("test2").endMetadata().addNewSubset() - .addNewAddress().withIp("ip2").withNewTargetRef().withUid("uid2") + Endpoints endPoint = new EndpointsBuilder().withNewMetadata().withName("endpoint") + .withNamespace("test").withLabels(labels).endMetadata().addNewSubset() + .addNewAddress().withIp("ip1").withNewTargetRef().withUid("10") .endTargetRef().endAddress().addNewPort("http", 80, "TCP").endSubset() .build(); List endpointsList = new ArrayList<>(); - endpointsList.add(endPoints1); - endpointsList.add(endpoints2); + endpointsList.add(endPoint); EndpointsList endpoints = new EndpointsList(); endpoints.setItems(endpointsList); + mockServer.expect().get().withPath( + "/api/v1/namespaces/test/endpoints?labelSelector=l%3Dv&fieldSelector=metadata.name%3Dendpoint") + .andReturn(200, endpoints).once(); + mockServer.expect().get() .withPath("/api/v1/endpoints?fieldSelector=metadata.name%3Dendpoint") .andReturn(200, endpoints).once(); - mockServer.expect().get().withPath("/api/v1/namespaces/test/endpoints/endpoint") - .andReturn(200, endPoints1).once(); + mockServer.expect().get().withPath( + "/api/v1/namespaces/test/endpoints?fieldSelector=metadata.name%3Dendpoint") + .andReturn(200, endpoints).once(); - mockServer.expect().get().withPath("/api/v1/namespaces/test2/endpoints/endpoint") - .andReturn(200, endpoints2).once(); - - Service service1 = new ServiceBuilder().withNewMetadata().withName("endpoint") - .withNamespace("test").withLabels(new HashMap() { - { - put("l", "v"); - } - }).endMetadata().build(); - - Service service2 = new ServiceBuilder().withNewMetadata().withName("endpoint") - .withNamespace("test2").withLabels(new HashMap() { - { - put("l", "v"); - } - }).endMetadata().build(); - - List servicesList = new ArrayList<>(); - servicesList.add(service1); - servicesList.add(service2); - - ServiceList services = new ServiceList(); - services.setItems(servicesList); - - mockServer.expect().get() - .withPath("/api/v1/services?fieldSelector=metadata.name%3Dendpoint") - .andReturn(200, services).once(); + Service service = new ServiceBuilder().withNewMetadata().withName("endpoint") + .withNamespace("test").withLabels(labels).endMetadata().build(); mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint") - .andReturn(200, service1).always(); - - mockServer.expect().get().withPath("/api/v1/namespaces/test2/services/endpoint") - .andReturn(200, service2).always(); - - final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); - properties.setAllNamespaces(true); - - final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, - properties, KubernetesClient::services, - new DefaultIsServicePortSecureResolver(properties)); - - final List instances = discoveryClient.getInstances("endpoint"); - - assertThat(instances).hasSize(2); - assertThat(instances).filteredOn(s -> s.getHost().equals("ip1") && !s.isSecure()) - .hasSize(1); - assertThat(instances).filteredOn(s -> s.getHost().equals("ip2") && !s.isSecure()) - .hasSize(1); - assertThat(instances).filteredOn(s -> s.getInstanceId().equals("uid1")) - .hasSize(1); - assertThat(instances).filteredOn(s -> s.getInstanceId().equals("uid2")) - .hasSize(1); - } - - @Test - public void getInstancesShouldBeAbleToHandleEndpointsSingleAddress() { - mockServer.expect().get().withPath("/api/v1/namespaces/test/endpoints/endpoint") - .andReturn(200, new EndpointsBuilder().withNewMetadata() - .withName("endpoint").endMetadata().addNewSubset().addNewAddress() - .withIp("ip1").withNewTargetRef().withUid("uid1").endTargetRef() - .endAddress().addNewPort("http", 80, "TCP").endSubset().build()) - .once(); - - mockServer.expect().get().withPath("/api/v1/services/endpoint") - .andReturn(200, new ServiceBuilder().withNewMetadata() - .withName("endpoint").withLabels(new HashMap() { - { - put("l", "v"); - } - }).endMetadata().build()) - .always(); + .andReturn(200, service).always(); final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + properties.setServiceLabels(labels); + properties.getMetadata().setAddLabels(false); + properties.getMetadata().setAddAnnotations(false); final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, KubernetesClient::services, @@ -170,30 +109,45 @@ public class KubernetesDiscoveryClientTest { assertThat(instances).hasSize(1) .filteredOn(s -> s.getHost().equals("ip1") && !s.isSecure()).hasSize(1) - .filteredOn(s -> s.getInstanceId().equals("uid1")).hasSize(1); + .filteredOn(s -> s.getInstanceId().equals("10")).hasSize(1); } @Test public void getInstancesShouldBeAbleToHandleEndpointsSingleAddressAndMultiplePorts() { - mockServer.expect().get().withPath("/api/v1/namespaces/test/endpoints/endpoint") - .andReturn(200, new EndpointsBuilder().withNewMetadata() - .withName("endpoint").endMetadata().addNewSubset().addNewAddress() - .withIp("ip1").withNewTargetRef().withUid("uid").endTargetRef() - .endAddress().addNewPort("mgmt", 9000, "TCP") - .addNewPort("http", 80, "TCP").endSubset().build()) - .once(); + Map labels = new HashMap(); + labels.put("l2", "v2"); + + Endpoints endPoint1 = new EndpointsBuilder().withNewMetadata() + .withName("endpoint").withNamespace("test").withLabels(labels) + .endMetadata().addNewSubset().addNewAddress().withIp("ip1") + .withNewTargetRef().withUid("20").endTargetRef().endAddress() + .addNewPort("mgmt", 900, "TCP").addNewPort("http", 80, "TCP").endSubset() + .build(); + + List endpointsList = new ArrayList<>(); + endpointsList.add(endPoint1); + + EndpointsList endpoints = new EndpointsList(); + endpoints.setItems(endpointsList); + + mockServer.expect().get().withPath( + "/api/v1/namespaces/test/endpoints?labelSelector=l2%3Dv2&fieldSelector=metadata.name%3Dendpoint") + .andReturn(200, endpoints).once(); + + mockServer.expect().get().withPath( + "/api/v1/namespaces/test/endpoints?fieldSelector=metadata.name%3Dendpoint") + .andReturn(200, endpoints).once(); + + Service service = new ServiceBuilder().withNewMetadata().withName("endpoint") + .withNamespace("test").withLabels(labels).withAnnotations(labels) + .endMetadata().build(); mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint") - .andReturn(200, new ServiceBuilder().withNewMetadata() - .withName("endpoint").withLabels(new HashMap() { - { - put("l", "v"); - } - }).endMetadata().build()) - .always(); + .andReturn(200, service).always(); final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); properties.setPrimaryPortName("http"); + final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, KubernetesClient::services, new DefaultIsServicePortSecureResolver(properties)); @@ -202,29 +156,81 @@ public class KubernetesDiscoveryClientTest { assertThat(instances).hasSize(1) .filteredOn(s -> s.getHost().equals("ip1") && !s.isSecure()).hasSize(1) - .filteredOn(s -> s.getInstanceId().equals("uid")).hasSize(1) + .filteredOn(s -> s.getInstanceId().equals("20")).hasSize(1) .filteredOn(s -> 80 == s.getPort()).hasSize(1); } @Test - public void getInstancesShouldBeAbleToHandleEndpointsMultipleAddresses() { - mockServer.expect().get().withPath("/api/v1/namespaces/test/endpoints/endpoint") - .andReturn(200, new EndpointsBuilder().withNewMetadata() - .withName("endpoint").endMetadata().addNewSubset().addNewAddress() - .withIp("ip1").endAddress().addNewAddress().withIp("ip2") - .endAddress().addNewPort("https", 443, "TCP").endSubset().build()) - .once(); + public void getEndPointsListTest() { + Map labels = new HashMap(); + labels.put("l", "v"); - mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint") - .andReturn(200, new ServiceBuilder().withNewMetadata() - .withName("endpoint").withLabels(new HashMap() { - { - put("l", "v"); - } - }).endMetadata().build()) - .always(); + Endpoints endPoint = new EndpointsBuilder().withNewMetadata().withName("endpoint") + .withNamespace("test").withLabels(labels).endMetadata().addNewSubset() + .addNewAddress().withIp("ip1").withNewTargetRef().withUid("30") + .endTargetRef().endAddress().addNewPort("http", 80, "TCP").endSubset() + .build(); + + List endpointsList = new ArrayList<>(); + endpointsList.add(endPoint); + + EndpointsList endpoints = new EndpointsList(); + endpoints.setItems(endpointsList); + + mockServer.expect().get().withPath( + "/api/v1/namespaces/test/endpoints?labelSelector=l%3Dv&fieldSelector=metadata.name%3Dendpoint") + .andReturn(200, endpoints).once(); final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + properties.setServiceLabels(labels); + + final KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient( + mockClient, properties, KubernetesClient::services, + new DefaultIsServicePortSecureResolver(properties)); + + final List result_endpoints = discoveryClient + .getEndPointsList("endpoint"); + + assertThat(result_endpoints).hasSize(1); + } + + @Test + public void getInstancesShouldBeAbleToHandleEndpointsMultipleAddresses() { + Map labels = new HashMap(); + labels.put("l1", "v1"); + + Endpoints endPoint = new EndpointsBuilder().withNewMetadata().withName("endpoint") + .withNamespace("test").withLabels(labels).endMetadata().addNewSubset() + .addNewAddress().withIp("ip1").withNewTargetRef().withUid("40") + .endTargetRef().endAddress().addNewAddress().withIp("ip2") + .withNewTargetRef().withUid("50").endTargetRef().endAddress() + .addNewPort("https", 443, "TCP").endSubset().build(); + + List endpointsList = new ArrayList<>(); + endpointsList.add(endPoint); + + EndpointsList endpoints = new EndpointsList(); + endpoints.setItems(endpointsList); + + mockServer.expect().get().withPath( + "/api/v1/namespaces/test/endpoints?labelSelector=l1%3Dv1&fieldSelector=metadata.name%3Dendpoint") + .andReturn(200, endpoints).once(); + + mockServer.expect().get().withPath( + "/api/v1/namespaces/test/endpoints?fieldSelector=metadata.name%3Dendpoint") + .andReturn(200, endpoints).once(); + + Service service = new ServiceBuilder().withNewMetadata().withName("endpoint") + .withNamespace("test").withLabels(labels).endMetadata().build(); + + mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint") + .andReturn(200, service).always(); + + final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + properties.setServiceLabels(labels); + properties.getMetadata().setAddAnnotations(false); + properties.getMetadata().setAddLabels(false); + final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, KubernetesClient::services, new DefaultIsServicePortSecureResolver(properties)); @@ -295,4 +301,84 @@ public class KubernetesDiscoveryClientTest { assertThat(services).containsOnly("s1", "s2"); } + @Test + public void getInstancesShouldBeAbleToHandleEndpointsFromMultipleNamespaces() { + Endpoints endPoints1 = new EndpointsBuilder().withNewMetadata() + .withName("endpoint").withNamespace("test").endMetadata().addNewSubset() + .addNewAddress().withIp("ip1").withNewTargetRef().withUid("60") + .endTargetRef().endAddress().addNewPort("http", 80, "TCP").endSubset() + .build(); + + Endpoints endpoints2 = new EndpointsBuilder().withNewMetadata() + .withName("endpoint").withNamespace("test2").endMetadata().addNewSubset() + .addNewAddress().withIp("ip2").withNewTargetRef().withUid("70") + .endTargetRef().endAddress().addNewPort("http", 80, "TCP").endSubset() + .build(); + + List endpointsList = new ArrayList<>(); + endpointsList.add(endPoints1); + endpointsList.add(endpoints2); + + EndpointsList endpoints = new EndpointsList(); + endpoints.setItems(endpointsList); + + mockServer.expect().get() + .withPath("/api/v1/endpoints?fieldSelector=metadata.name%3Dendpoint") + .andReturn(200, endpoints).once(); + + mockServer.expect().get().withPath("/api/v1/namespaces/test/endpoints/endpoint") + .andReturn(200, endPoints1).once(); + + mockServer.expect().get().withPath("/api/v1/namespaces/test2/endpoints/endpoint") + .andReturn(200, endpoints2).once(); + + Service service1 = new ServiceBuilder().withNewMetadata().withName("endpoint") + .withNamespace("test").withLabels(new HashMap() { + { + put("l", "v"); + } + }).endMetadata().build(); + + Service service2 = new ServiceBuilder().withNewMetadata().withName("endpoint") + .withNamespace("test2").withLabels(new HashMap() { + { + put("l", "v"); + } + }).endMetadata().build(); + + List servicesList = new ArrayList<>(); + servicesList.add(service1); + servicesList.add(service2); + + ServiceList services = new ServiceList(); + services.setItems(servicesList); + + mockServer.expect().get() + .withPath("/api/v1/services?fieldSelector=metadata.name%3Dendpoint") + .andReturn(200, services).once(); + + mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint") + .andReturn(200, service1).always(); + + mockServer.expect().get().withPath("/api/v1/namespaces/test2/services/endpoint") + .andReturn(200, service2).always(); + + final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + properties.setAllNamespaces(true); + + final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, + properties, KubernetesClient::services, + new DefaultIsServicePortSecureResolver(properties)); + + final List instances = discoveryClient.getInstances("endpoint"); + + assertThat(instances).hasSize(2); + assertThat(instances).filteredOn(s -> s.getHost().equals("ip1") && !s.isSecure()) + .hasSize(1); + assertThat(instances).filteredOn(s -> s.getHost().equals("ip2") && !s.isSecure()) + .hasSize(1); + assertThat(instances).filteredOn(s -> s.getInstanceId().equals("60")).hasSize(1); + assertThat(instances).filteredOn(s -> s.getInstanceId().equals("70")).hasSize(1); + } + } From 0951e892cebd3fbc9f5b0d651dffe0cfb6e46b7a Mon Sep 17 00:00:00 2001 From: Haytham Mohamed Date: Thu, 9 Jul 2020 18:16:37 -0500 Subject: [PATCH 3/3] 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