From ceb192ca3e420ad691de396177260ce34641e00f Mon Sep 17 00:00:00 2001 From: Tim Ysewyn Date: Thu, 11 Mar 2021 23:05:59 +0100 Subject: [PATCH] Use the "primary-port-name" label to discover ports (#749) * Use the "primary-port-name" label to discover ports * Fixed warning message * Fixed code style issue * Use the "primary-port-name" label to discover ports * Do not change previous behaviour! * Fixed possibility of NPE * Updated documentation * Added fallbacks to https and http named ports * Extracted port discovery logic into a separate function * Fixed code style issue * Fixed another code style issue --- docs/src/main/asciidoc/discovery-client.adoc | 14 + .../KubernetesInformerDiscoveryClient.java | 96 +++++-- ...ubernetesInformerDiscoveryClientTests.java | 204 +++++++++++++- .../discovery/KubernetesDiscoveryClient.java | 66 +++-- .../KubernetesDiscoveryClientTest.java | 255 +++++++++++++++++- 5 files changed, 571 insertions(+), 64 deletions(-) diff --git a/docs/src/main/asciidoc/discovery-client.adoc b/docs/src/main/asciidoc/discovery-client.adoc index 49b12b6c..2a56f3a6 100644 --- a/docs/src/main/asciidoc/discovery-client.adoc +++ b/docs/src/main/asciidoc/discovery-client.adoc @@ -74,6 +74,20 @@ spring.cloud.kubernetes.discovery.include-not-ready-addresses=true NOTE: This might be useful when discovering services for monitoring purposes, and would enable inspecting the `/health` endpoint of not-ready service instances. ==== +If your service exposes multiple ports, you will need to specify which port the `DiscoveryClient` should use. +The `DiscoveryClient` will choose the port using the following logic. + +1. If the service has a label `primary-port-name` it will use the port with the name specified in the label's value. +2. If no label is present, then the port name specified in `spring.cloud.kubernetes.discovery.primary-port-name` will be used. +3. If neither of the above are specified it will use the port named `https`. +4. If none of the above conditions are met it will use the port named `http`. +5. As a last resort it wil pick the first port in the list of ports. + +WARNING: The last option may result in non-deterministic behaviour. +Please make sure to configure your service and/or application accordingly. + +By default all of the ports and their names will be added to the metadata of the `ServiceInstance`. + If, for any reason, you need to disable the `DiscoveryClient`, you can set the following property in `application.properties`: ==== diff --git a/spring-cloud-kubernetes-client-discovery/src/main/java/org/springframework/cloud/kubernetes/client/discovery/KubernetesInformerDiscoveryClient.java b/spring-cloud-kubernetes-client-discovery/src/main/java/org/springframework/cloud/kubernetes/client/discovery/KubernetesInformerDiscoveryClient.java index 878c43ee..7136a32f 100644 --- a/spring-cloud-kubernetes-client-discovery/src/main/java/org/springframework/cloud/kubernetes/client/discovery/KubernetesInformerDiscoveryClient.java +++ b/spring-cloud-kubernetes-client-discovery/src/main/java/org/springframework/cloud/kubernetes/client/discovery/KubernetesInformerDiscoveryClient.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -44,9 +45,17 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; +/** + * @author Min Kim + * @author Ryan Baxter + * @author Tim Yysewyn + */ public class KubernetesInformerDiscoveryClient implements DiscoveryClient, InitializingBean { private static final Log log = LogFactory.getLog(KubernetesInformerDiscoveryClient.class); + private static final String PRIMARY_PORT_NAME_LABEL_KEY = "primary-port-name"; + private static final String HTTPS_PORT_NAME = "https"; + private static final String HTTP_PORT_NAME = "http"; private final SharedInformerFactory sharedInformerFactory; @@ -98,7 +107,7 @@ public class KubernetesInformerDiscoveryClient implements DiscoveryClient, Initi Map svcMetadata = new HashMap<>(); if (this.properties.getMetadata() != null) { if (this.properties.getMetadata().isAddLabels()) { - if (service.getMetadata().getLabels() != null) { + if (service.getMetadata() != null && service.getMetadata().getLabels() != null) { String labelPrefix = this.properties.getMetadata().getLabelsPrefix() != null ? this.properties.getMetadata().getLabelsPrefix() : ""; service.getMetadata().getLabels().entrySet().stream() @@ -107,7 +116,7 @@ public class KubernetesInformerDiscoveryClient implements DiscoveryClient, Initi } } if (this.properties.getMetadata().isAddAnnotations()) { - if (service.getMetadata().getAnnotations() != null) { + if (service.getMetadata() != null && service.getMetadata().getAnnotations() != null) { String annotationPrefix = this.properties.getMetadata().getAnnotationsPrefix() != null ? this.properties.getMetadata().getAnnotationsPrefix() : ""; service.getMetadata().getAnnotations().entrySet().stream() @@ -123,36 +132,75 @@ public class KubernetesInformerDiscoveryClient implements DiscoveryClient, Initi // no available endpoints in the cluster return new ArrayList<>(); } - return ep.getSubsets().stream().flatMap(subset -> { - Map metadata = new HashMap<>(svcMetadata); - if (this.properties.getMetadata() != null && this.properties.getMetadata().isAddPorts()) { - subset.getPorts().stream().forEach(p -> metadata.put(p.getName(), Integer.toString(p.getPort()))); - } - V1EndpointPort port = subset.getPorts() != null && subset.getPorts().size() == 1 ? subset.getPorts().get(0) - : subset.getPorts().stream() - .filter(p -> p.getName().equalsIgnoreCase(this.properties.getPrimaryPortName())).findFirst() - .orElseThrow(IllegalStateException::new); - List addresses = subset.getAddresses(); - if (addresses == null) { - addresses = new ArrayList<>(); - } - if (this.properties.isIncludeNotReadyAddresses() - && !CollectionUtils.isEmpty(subset.getNotReadyAddresses())) { - addresses.addAll(subset.getNotReadyAddresses()); - } - return addresses.stream() - .map(addr -> new KubernetesServiceInstance( - addr.getTargetRef() != null ? addr.getTargetRef().getUid() : "", serviceId, addr.getIp(), - port.getPort(), metadata, false)); + Optional discoveredPrimaryPortName = Optional.empty(); + if (service.getMetadata() != null && service.getMetadata().getLabels() != null) { + discoveredPrimaryPortName = Optional + .ofNullable(service.getMetadata().getLabels().get(PRIMARY_PORT_NAME_LABEL_KEY)); + } + final String primaryPortName = discoveredPrimaryPortName.orElse(this.properties.getPrimaryPortName()); + + return ep.getSubsets().stream() + .filter(subset -> subset.getPorts() != null && subset.getPorts().size() > 0) // safeguard + .flatMap(subset -> { + Map metadata = new HashMap<>(svcMetadata); + List endpointPorts = subset.getPorts(); + if (this.properties.getMetadata() != null && this.properties.getMetadata().isAddPorts()) { + endpointPorts.forEach(p -> metadata.put(p.getName(), Integer.toString(p.getPort()))); + } + List addresses = subset.getAddresses(); + if (addresses == null) { + addresses = new ArrayList<>(); + } + if (this.properties.isIncludeNotReadyAddresses() + && !CollectionUtils.isEmpty(subset.getNotReadyAddresses())) { + addresses.addAll(subset.getNotReadyAddresses()); + } + + final int port = findEndpointPort(endpointPorts, primaryPortName, serviceId); + return addresses.stream() + .map(addr -> new KubernetesServiceInstance( + addr.getTargetRef() != null ? addr.getTargetRef().getUid() : "", serviceId, addr.getIp(), + port, metadata, false)); }).collect(Collectors.toList()); } + private int findEndpointPort(List endpointPorts, String primaryPortName, String serviceId) { + if (endpointPorts.size() == 1) { + return endpointPorts.get(0).getPort(); + } + else { + Map ports = endpointPorts.stream() + .filter(p -> StringUtils.hasText(p.getName())) + .collect(Collectors.toMap(V1EndpointPort::getName, V1EndpointPort::getPort)); + // This oneliner is looking for a port with a name equal to the primary port name specified in the service label + // or in spring.cloud.kubernetes.discovery.primary-port-name, equal to https, or equal to http. + // In case no port has been found return -1 to log a warning and fall back to the first port in the list. + int discoveredPort = ports.getOrDefault(primaryPortName, ports.getOrDefault(HTTPS_PORT_NAME, ports.getOrDefault(HTTP_PORT_NAME, -1))); + + if (discoveredPort == -1) { + if (StringUtils.hasText(primaryPortName)) { + log.warn("Could not find a port named '" + primaryPortName + "', 'https', or 'http' for service '" + serviceId + "'."); + } + else { + log.warn("Could not find a port named 'https' or 'http' for service '" + serviceId + "'."); + } + log.warn("Make sure that either the primary-port-name label has been added to the service, or that spring.cloud.kubernetes.discovery.primary-port-name has been configured."); + log.warn("Alternatively name the primary port 'https' or 'http'"); + log.warn("An incorrect configuration may result in non-deterministic behaviour."); + discoveredPort = endpointPorts.get(0).getPort(); + } + return discoveredPort; + } + } + @Override public List getServices() { List services = this.properties.isAllNamespaces() ? this.serviceLister.list() : this.serviceLister.namespace(this.namespace).list(); - return services.stream().map(s -> s.getMetadata().getName()).collect(Collectors.toList()); + return services.stream() + .filter(s -> s.getMetadata() != null) //safeguard + .map(s -> s.getMetadata().getName()).collect(Collectors.toList()); } @Override diff --git a/spring-cloud-kubernetes-client-discovery/src/test/java/org/springframework/cloud/kubernetes/client/discovery/KubernetesInformerDiscoveryClientTests.java b/spring-cloud-kubernetes-client-discovery/src/test/java/org/springframework/cloud/kubernetes/client/discovery/KubernetesInformerDiscoveryClientTests.java index 9cb6d04c..01285502 100644 --- a/spring-cloud-kubernetes-client-discovery/src/test/java/org/springframework/cloud/kubernetes/client/discovery/KubernetesInformerDiscoveryClientTests.java +++ b/spring-cloud-kubernetes-client-discovery/src/test/java/org/springframework/cloud/kubernetes/client/discovery/KubernetesInformerDiscoveryClientTests.java @@ -64,15 +64,34 @@ public class KubernetesInformerDiscoveryClientTests { .addSubsetsItem(new V1EndpointSubset().addPortsItem(new V1EndpointPort().port(8080)) .addAddressesItem(new V1EndpointAddress().ip("2.2.2.2"))); - private static final V1Service testServiceWithoutReadyAddresses = new V1Service() - .metadata(new V1ObjectMeta().name("test-svc-without-ready-addresses").namespace("namespace1")) - .spec(new V1ServiceSpec().loadBalancerIP("1.1.1.1")).status(new V1ServiceStatus()); - - private static final V1Endpoints testEndpointsWithoutReadyAddresses = new V1Endpoints() - .metadata(new V1ObjectMeta().name("test-svc-without-ready-addresses").namespace("namespace1")) + private static final V1Endpoints testEndpointWithoutReadyAddresses = new V1Endpoints() + .metadata(new V1ObjectMeta().name("test-svc-1").namespace("namespace1")) .addSubsetsItem(new V1EndpointSubset().addPortsItem(new V1EndpointPort().port(8080)) .addNotReadyAddressesItem(new V1EndpointAddress().ip("2.2.2.2"))); + private static final V1Endpoints testEndpointWithoutPorts = new V1Endpoints() + .metadata(new V1ObjectMeta().name("test-svc-1").namespace("namespace1")) + .addSubsetsItem(new V1EndpointSubset() + .addAddressesItem(new V1EndpointAddress().ip("1.1.1.1"))); + + private static final V1Endpoints testEndpointWithMultiplePorts = new V1Endpoints() + .metadata(new V1ObjectMeta().name("test-svc-1").namespace("namespace1")) + .addSubsetsItem(new V1EndpointSubset().addPortsItem(new V1EndpointPort().name("http").port(80)) + .addPortsItem(new V1EndpointPort().name("https").port(443)) + .addAddressesItem(new V1EndpointAddress().ip("1.1.1.1"))); + + private static final V1Endpoints testEndpointWithMultiplePortsWithoutHttps = new V1Endpoints() + .metadata(new V1ObjectMeta().name("test-svc-1").namespace("namespace1")) + .addSubsetsItem(new V1EndpointSubset().addPortsItem(new V1EndpointPort().name("http").port(80)) + .addPortsItem(new V1EndpointPort().name("tcp").port(443)) + .addAddressesItem(new V1EndpointAddress().ip("1.1.1.1"))); + + private static final V1Endpoints testEndpointWithMultiplePortsWithoutSupportedPortNames = new V1Endpoints() + .metadata(new V1ObjectMeta().name("test-svc-1").namespace("namespace1")) + .addSubsetsItem(new V1EndpointSubset().addPortsItem(new V1EndpointPort().name("tcp1").port(80)) + .addPortsItem(new V1EndpointPort().name("tcp2").port(443)) + .addAddressesItem(new V1EndpointAddress().ip("1.1.1.1"))); + @Test public void testDiscoveryGetServicesAllNamespaceShouldWork() { Lister serviceLister = setupServiceLister(testService1, testService2); @@ -116,6 +135,7 @@ public class KubernetesInformerDiscoveryClientTests { .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "2.2.2.2", 8080, new HashMap<>(), false)); verify(kubernetesDiscoveryProperties, times(2)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); } @Test @@ -131,27 +151,29 @@ public class KubernetesInformerDiscoveryClientTests { assertThat(discoveryClient.getInstances("test-svc-1")) .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "2.2.2.2", 8080, new HashMap<>(), false)); verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); } @Test public void testDiscoveryGetInstanceWithoutReadyAddressesShouldWork() { - Lister serviceLister = setupServiceLister(testServiceWithoutReadyAddresses); - Lister endpointsLister = setupEndpointsLister(testEndpointsWithoutReadyAddresses); + Lister serviceLister = setupServiceLister(testService1); + Lister endpointsLister = setupEndpointsLister(testEndpointWithoutReadyAddresses); when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); - assertThat(discoveryClient.getInstances("test-svc-without-ready-addresses")).isEmpty(); + assertThat(discoveryClient.getInstances("test-svc-1")).isEmpty(); verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); verify(kubernetesDiscoveryProperties, times(1)).isIncludeNotReadyAddresses(); } @Test public void testDiscoveryGetInstanceWithNotReadyAddressesIncludedShouldWork() { - Lister serviceLister = setupServiceLister(testServiceWithoutReadyAddresses); - Lister endpointsLister = setupEndpointsLister(testEndpointsWithoutReadyAddresses); + Lister serviceLister = setupServiceLister(testService1); + Lister endpointsLister = setupEndpointsLister(testEndpointWithoutReadyAddresses); when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); when(kubernetesDiscoveryProperties.isIncludeNotReadyAddresses()).thenReturn(true); @@ -159,13 +181,169 @@ public class KubernetesInformerDiscoveryClientTests { KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); - assertThat(discoveryClient.getInstances("test-svc-without-ready-addresses")) - .containsOnly(new KubernetesServiceInstance("", "test-svc-without-ready-addresses", "2.2.2.2", 8080, + assertThat(discoveryClient.getInstances("test-svc-1")) + .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "2.2.2.2", 8080, new HashMap<>(), false)); verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); verify(kubernetesDiscoveryProperties, times(1)).isIncludeNotReadyAddresses(); } + @Test + public void instanceWithoutEndpointsShouldBeSkipped() { + Lister serviceLister = setupServiceLister(testService1); + Lister endpointsLister = setupEndpointsLister(); + + when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); + + KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", + sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); + + assertThat(discoveryClient.getInstances("test-svc-1")).isEmpty(); + verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + } + + @Test + public void instanceWithoutPortsShouldBeSkipped() { + Lister serviceLister = setupServiceLister(testService1); + Lister endpointsLister = setupEndpointsLister(testEndpointWithoutPorts); + + when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); + + KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", + sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); + + assertThat(discoveryClient.getInstances("test-svc-1")).isEmpty(); + verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + } + + @Test + public void instanceWithMultiplePortsAndPrimaryPortNameConfiguredWithLabelShouldWork() { + Lister serviceLister = setupServiceLister(testService1 + .metadata(new V1ObjectMeta().name("test-svc-1").namespace("namespace1") + .putLabelsItem("primary-port-name", "https"))); + Lister endpointsLister = setupEndpointsLister(testEndpointWithMultiplePorts); + + when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); + + KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", + sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); + + assertThat(discoveryClient.getInstances("test-svc-1")) + .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "1.1.1.1", 443, + new HashMap<>(), false)); + verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); + verify(kubernetesDiscoveryProperties, times(1)).isIncludeNotReadyAddresses(); + } + + @Test + public void instanceWithMultiplePortsAndMisconfiguredPrimaryPortNameInLabelShouldReturnFirstPortAndLogWarning() { + Lister serviceLister = setupServiceLister(testService1 + .metadata(new V1ObjectMeta().name("test-svc-1").namespace("namespace1") + .putLabelsItem("primary-port-name", "oops"))); + Lister endpointsLister = setupEndpointsLister(testEndpointWithMultiplePortsWithoutSupportedPortNames); + + when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); + + KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", + sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); + + assertThat(discoveryClient.getInstances("test-svc-1")) + .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "1.1.1.1", 80, + new HashMap<>(), false)); + verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); + } + + @Test + public void instanceWithMultiplePortsAndGenericPrimaryPortNameConfiguredShouldWork() { + Lister serviceLister = setupServiceLister(testService1); + Lister endpointsLister = setupEndpointsLister(testEndpointWithMultiplePorts); + + when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); + when(kubernetesDiscoveryProperties.getPrimaryPortName()).thenReturn("https"); + + KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", + sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); + + assertThat(discoveryClient.getInstances("test-svc-1")) + .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "1.1.1.1", 443, + new HashMap<>(), false)); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); + verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).isIncludeNotReadyAddresses(); + } + + @Test + public void instanceWithMultiplePortsAndMisconfiguredGenericPrimaryPortNameShouldReturnFirstPortAndLogWarning() { + Lister serviceLister = setupServiceLister(testService1); + Lister endpointsLister = setupEndpointsLister(testEndpointWithMultiplePortsWithoutSupportedPortNames); + + when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); + when(kubernetesDiscoveryProperties.getPrimaryPortName()).thenReturn("oops"); + + KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", + sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); + + assertThat(discoveryClient.getInstances("test-svc-1")) + .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "1.1.1.1", 80, + new HashMap<>(), false)); + verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); + } + + @Test + public void instanceWithMultiplePortsAndWithoutPrimaryPortNameSpecifiedShouldFallBackToHttpsPort() { + Lister serviceLister = setupServiceLister(testService1); + Lister endpointsLister = setupEndpointsLister(testEndpointWithMultiplePorts); + + when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); + + KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", + sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); + + assertThat(discoveryClient.getInstances("test-svc-1")) + .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "1.1.1.1", 443, + new HashMap<>(), false)); + verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); + } + + @Test + public void instanceWithMultiplePortsAndWithoutPrimaryPortNameSpecifiedOrHttpsPortShouldFallBackToHttpPort() { + Lister serviceLister = setupServiceLister(testService1); + Lister endpointsLister = setupEndpointsLister(testEndpointWithMultiplePortsWithoutHttps); + + when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); + + KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", + sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); + + assertThat(discoveryClient.getInstances("test-svc-1")) + .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "1.1.1.1", 80, + new HashMap<>(), false)); + verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); + } + + @Test + public void instanceWithMultiplePortsAndWithoutAnyConfigurationShouldPickTheFirstPort() { + Lister serviceLister = setupServiceLister(testService1); + Lister endpointsLister = setupEndpointsLister(testEndpointWithMultiplePortsWithoutSupportedPortNames); + + when(kubernetesDiscoveryProperties.isAllNamespaces()).thenReturn(false); + + KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1", + sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties); + + assertThat(discoveryClient.getInstances("test-svc-1")) + .containsOnly(new KubernetesServiceInstance("", "test-svc-1", "1.1.1.1", 80, + new HashMap<>(), false)); + verify(kubernetesDiscoveryProperties, times(1)).isAllNamespaces(); + verify(kubernetesDiscoveryProperties, times(1)).getPrimaryPortName(); + } + private Lister setupServiceLister(V1Service... services) { Cache serviceCache = new Cache<>(); Lister serviceLister = new Lister<>(serviceCache); diff --git a/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClient.java b/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClient.java index 8741e1ff..52a72725 100644 --- a/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClient.java +++ b/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClient.java @@ -47,13 +47,17 @@ import static java.util.stream.Collectors.toMap; import static org.springframework.cloud.kubernetes.commons.discovery.KubernetesServiceInstance.NAMESPACE_METADATA_KEY; /** - * Kubeneretes implementation of {@link DiscoveryClient}. + * Kubernetes implementation of {@link DiscoveryClient}. * * @author Ioannis Canellos + * @author Tim Ysewyn */ public class KubernetesDiscoveryClient implements DiscoveryClient { private static final Log log = LogFactory.getLog(KubernetesDiscoveryClient.class); + private static final String PRIMARY_PORT_NAME_LABEL_KEY = "primary-port-name"; + private static final String HTTPS_PORT_NAME = "https"; + private static final String HTTP_PORT_NAME = "http"; private final KubernetesDiscoveryProperties properties; @@ -104,7 +108,7 @@ public class KubernetesDiscoveryClient implements DiscoveryClient { Assert.notNull(serviceId, "[Assertion failed] - the object argument must not be null"); List subsetsNS = this.getEndPointsList(serviceId).stream() - .map(endpoints -> getSubsetsFromEndpoints(endpoints)).collect(Collectors.toList()); + .map(this::getSubsetsFromEndpoints).collect(Collectors.toList()); List instances = new ArrayList<>(); if (!subsetsNS.isEmpty()) { @@ -133,13 +137,19 @@ public class KubernetesDiscoveryClient implements DiscoveryClient { final Map serviceMetadata = this.getServiceMetadata(service); KubernetesDiscoveryProperties.Metadata metadataProps = this.properties.getMetadata(); + String primaryPortName = this.properties.getPrimaryPortName(); + Map labels = service.getMetadata().getLabels(); + if (labels != null && labels.containsKey(PRIMARY_PORT_NAME_LABEL_KEY)) { + primaryPortName = labels.get(PRIMARY_PORT_NAME_LABEL_KEY); + } + for (EndpointSubset s : subsets) { // Extend the service metadata map with per-endpoint port information (if // requested) Map endpointMetadata = new HashMap<>(serviceMetadata); if (metadataProps.isAddPorts()) { Map ports = s.getPorts().stream() - .filter(port -> !StringUtils.isEmpty(port.getName())) + .filter(port -> StringUtils.hasText(port.getName())) .collect(toMap(EndpointPort::getName, port -> Integer.toString(port.getPort()))); Map portMetadata = getMapWithPrefixedKeys(ports, metadataProps.getPortsPrefix()); if (log.isDebugEnabled()) { @@ -157,23 +167,22 @@ public class KubernetesDiscoveryClient implements DiscoveryClient { if (this.properties.isIncludeNotReadyAddresses() && !CollectionUtils.isEmpty(s.getNotReadyAddresses())) { if (addresses == null) { - addresses = new ArrayList(); + addresses = new ArrayList<>(); } addresses.addAll(s.getNotReadyAddresses()); } for (EndpointAddress endpointAddress : addresses) { + int endpointPort = findEndpointPort(s, serviceId, primaryPortName); String instanceId = null; if (endpointAddress.getTargetRef() != null) { instanceId = endpointAddress.getTargetRef().getUid(); } - - EndpointPort endpointPort = findEndpointPort(s); instances.add(new KubernetesServiceInstance(instanceId, serviceId, endpointAddress.getIp(), - endpointPort.getPort(), endpointMetadata, - this.servicePortSecureResolver.resolve(new ServicePortSecureResolver.Input( - endpointPort.getPort(), service.getMetadata().getName(), - service.getMetadata().getLabels(), service.getMetadata().getAnnotations())))); + endpointPort, endpointMetadata, + this.servicePortSecureResolver.resolve(new ServicePortSecureResolver.Input( + endpointPort, service.getMetadata().getName(), + service.getMetadata().getLabels(), service.getMetadata().getAnnotations())))); } } } @@ -204,23 +213,34 @@ public class KubernetesDiscoveryClient implements DiscoveryClient { return serviceMetadata; } - private EndpointPort findEndpointPort(EndpointSubset s) { - List ports = s.getPorts(); - EndpointPort endpointPort; - if (ports.size() == 1) { - endpointPort = ports.get(0); + private int findEndpointPort(EndpointSubset s, String serviceId, String primaryPortName) { + List endpointPorts = s.getPorts(); + if (endpointPorts.size() == 1) { + return endpointPorts.get(0).getPort(); } else { - Predicate portPredicate; - if (!StringUtils.isEmpty(properties.getPrimaryPortName())) { - portPredicate = port -> properties.getPrimaryPortName().equalsIgnoreCase(port.getName()); + Map ports = endpointPorts.stream() + .filter(p -> StringUtils.hasText(p.getName())) + .collect(Collectors.toMap(EndpointPort::getName, EndpointPort::getPort)); + // This oneliner is looking for a port with a name equal to the primary port name specified in the service label + // or in spring.cloud.kubernetes.discovery.primary-port-name, equal to https, or equal to http. + // In case no port has been found return -1 to log a warning and fall back to the first port in the list. + int discoveredPort = ports.getOrDefault(primaryPortName, ports.getOrDefault(HTTPS_PORT_NAME, ports.getOrDefault(HTTP_PORT_NAME, -1))); + + if (discoveredPort == -1) { + if (StringUtils.hasText(primaryPortName)) { + log.warn("Could not find a port named '" + primaryPortName + "', 'https', or 'http' for service '" + serviceId + "'."); + } + else { + log.warn("Could not find a port named 'https' or 'http' for service '" + serviceId + "'."); + } + log.warn("Make sure that either the primary-port-name label has been added to the service, or that spring.cloud.kubernetes.discovery.primary-port-name has been configured."); + log.warn("Alternatively name the primary port 'https' or 'http'"); + log.warn("An incorrect configuration may result in non-deterministic behaviour."); + discoveredPort = endpointPorts.get(0).getPort(); } - else { - portPredicate = port -> true; - } - endpointPort = ports.stream().filter(portPredicate).findAny().orElseThrow(IllegalStateException::new); + return discoveredPort; } - return endpointPort; } private EndpointSubsetNS getSubsetsFromEndpoints(Endpoints endpoints) { diff --git a/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClientTest.java b/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClientTest.java index d7e8f73b..e6d85735 100644 --- a/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClientTest.java +++ b/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClientTest.java @@ -17,6 +17,7 @@ package org.springframework.cloud.kubernetes.fabric8.discovery; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -63,7 +64,7 @@ public class KubernetesDiscoveryClientTest { @Test public void getInstancesShouldBeAbleToHandleEndpointsSingleAddress() { - Map labels = new HashMap(); + Map labels = new HashMap<>(); labels.put("l", "v"); Endpoints endPoint = new EndpointsBuilder().withNewMetadata().withName("endpoint").withNamespace("test") @@ -110,7 +111,7 @@ public class KubernetesDiscoveryClientTest { @Test public void getInstancesShouldBeAbleToHandleEndpointsSingleAddressAndMultiplePorts() { - Map labels = new HashMap(); + Map labels = new HashMap<>(); labels.put("l2", "v2"); Endpoints endPoint1 = new EndpointsBuilder().withNewMetadata().withName("endpoint").withNamespace("test") @@ -152,7 +153,7 @@ public class KubernetesDiscoveryClientTest { @Test public void getEndPointsListTest() { - Map labels = new HashMap(); + Map labels = new HashMap<>(); labels.put("l", "v"); Endpoints endPoint = new EndpointsBuilder().withNewMetadata().withName("endpoint").withNamespace("test") @@ -184,7 +185,7 @@ public class KubernetesDiscoveryClientTest { @Test public void getInstancesShouldBeAbleToHandleEndpointsMultipleAddresses() { - Map labels = new HashMap(); + Map labels = new HashMap<>(); labels.put("l1", "v1"); Endpoints endPoint = new EndpointsBuilder().withNewMetadata().withName("endpoint").withNamespace("test") @@ -356,4 +357,250 @@ public class KubernetesDiscoveryClientTest { assertThat(instances).filteredOn(s -> s.getInstanceId().equals("70")).hasSize(1); } + @Test + public void instanceWithoutPortsShouldBeSkipped() { + Endpoints endPoint = new EndpointsBuilder().withNewMetadata().withName("endpoint1").withNamespace("test") + .withLabels(Collections.emptyMap()).endMetadata().build(); + + List endpointsList = new ArrayList<>(); + endpointsList.add(endPoint); + + EndpointsList endpoints = new EndpointsList(); + endpoints.setItems(endpointsList); + + mockServer.expect().get() + .withPath( + "/api/v1/namespaces/test/endpoints?fieldSelector=metadata.name%3Dendpoint1") + .andReturn(200, endpoints).once(); + + final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + + final KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, + KubernetesClient::services, new ServicePortSecureResolver(properties)); + + final List instances = discoveryClient.getInstances("endpoint1"); + + assertThat(instances).isEmpty(); + } + + @Test + public void getInstancesShouldBeAbleToHandleEndpointsSingleAddressAndMultiplePortsUsingPrimaryPortNameLabel() { + Map labels = new HashMap<>(); + labels.put("primary-port-name", "https"); + + Endpoints endPoint1 = new EndpointsBuilder().withNewMetadata().withName("endpoint2").withNamespace("test") + .withLabels(labels).endMetadata().addNewSubset().addNewAddress().withIp("ip1").withNewTargetRef() + .withUid("80").endTargetRef().endAddress().addNewPort("http", "https", 443, "TCP") + .addNewPort("http", "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?fieldSelector=metadata.name%3Dendpoint2") + .andReturn(200, endpoints).once(); + + Service service = new ServiceBuilder().withNewMetadata().withName("endpoint2").withNamespace("test") + .withLabels(labels).withAnnotations(labels).endMetadata().build(); + mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint2").andReturn(200, service) + .always(); + + final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + + final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, + KubernetesClient::services, new ServicePortSecureResolver(properties)); + + final List instances = discoveryClient.getInstances("endpoint2"); + + assertThat(instances).hasSize(1).filteredOn(s -> s.getHost().equals("ip1") && s.isSecure()).hasSize(1) + .filteredOn(s -> s.getInstanceId().equals("80")).hasSize(1).filteredOn(s -> 443 == s.getPort()) + .hasSize(1); + } + + @Test + public void instanceWithMultiplePortsAndMisconfiguredPrimaryPortNameInLabelWithoutFallbackShouldLogWarning() { + Map labels = new HashMap<>(); + labels.put("primary-port-name", "oops"); + + Endpoints endPoint1 = new EndpointsBuilder().withNewMetadata().withName("endpoint3").withNamespace("test") + .withLabels(labels).endMetadata().addNewSubset().addNewAddress().withIp("ip1").withNewTargetRef() + .withUid("90").endTargetRef().endAddress().addNewPort("http", "https1", 443, "TCP") + .addNewPort("http", "https2", 8443, "TCP") + .addNewPort("http", "http1", 80, "TCP") + .addNewPort("http", "http2", 8080, "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?fieldSelector=metadata.name%3Dendpoint3") + .andReturn(200, endpoints).once(); + + Service service = new ServiceBuilder().withNewMetadata().withName("endpoint3").withNamespace("test") + .withLabels(labels).withAnnotations(labels).endMetadata().build(); + mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint3").andReturn(200, service) + .always(); + + final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + + final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, + KubernetesClient::services, new ServicePortSecureResolver(properties)); + + final List instances = discoveryClient.getInstances("endpoint3"); + + assertThat(instances).hasSize(1).filteredOn(s -> s.getHost().equals("ip1") && s.isSecure()).hasSize(1) + .filteredOn(s -> s.getInstanceId().equals("90")).hasSize(1).filteredOn(s -> 443 == s.getPort()) + .hasSize(1); + } + + @Test + public void instanceWithMultiplePortsAndMisconfiguredGenericPrimaryPortNameWithoutFallbackShouldLogWarning() { + Map labels = new HashMap<>(); + + Endpoints endPoint1 = new EndpointsBuilder().withNewMetadata().withName("endpoint4").withNamespace("test") + .withLabels(labels).endMetadata().addNewSubset().addNewAddress().withIp("ip1").withNewTargetRef() + .withUid("100").endTargetRef().endAddress().addNewPort("http", "https1", 443, "TCP") + .addNewPort("http", "https2", 8443, "TCP") + .addNewPort("http", "http1", 80, "TCP") + .addNewPort("http", "http2", 8080, "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?fieldSelector=metadata.name%3Dendpoint4") + .andReturn(200, endpoints).once(); + + Service service = new ServiceBuilder().withNewMetadata().withName("endpoint4").withNamespace("test") + .withLabels(labels).withAnnotations(labels).endMetadata().build(); + + mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint4").andReturn(200, service) + .always(); + + final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + properties.setPrimaryPortName("oops"); + + final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, + KubernetesClient::services, new ServicePortSecureResolver(properties)); + + final List instances = discoveryClient.getInstances("endpoint4"); + + assertThat(instances).hasSize(1).filteredOn(s -> s.getHost().equals("ip1") && s.isSecure()).hasSize(1) + .filteredOn(s -> s.getInstanceId().equals("100")).hasSize(1).filteredOn(s -> 443 == s.getPort()) + .hasSize(1); + } + + @Test + public void instanceWithMultiplePortsAndWithoutPrimaryPortNameSpecifiedShouldFallBackToHttps() { + Map labels = new HashMap<>(); + + Endpoints endPoint1 = new EndpointsBuilder().withNewMetadata().withName("endpoint5").withNamespace("test") + .withLabels(labels).endMetadata().addNewSubset().addNewAddress().withIp("ip1").withNewTargetRef() + .withUid("110").endTargetRef().endAddress().addNewPort("http", "https", 443, "TCP") + .addNewPort("http", "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?fieldSelector=metadata.name%3Dendpoint5") + .andReturn(200, endpoints).once(); + + Service service = new ServiceBuilder().withNewMetadata().withName("endpoint5").withNamespace("test") + .withLabels(labels).withAnnotations(labels).endMetadata().build(); + mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint5").andReturn(200, service) + .always(); + + final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + + final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, + KubernetesClient::services, new ServicePortSecureResolver(properties)); + + final List instances = discoveryClient.getInstances("endpoint5"); + + assertThat(instances).hasSize(1).filteredOn(s -> s.getHost().equals("ip1") && s.isSecure()).hasSize(1) + .filteredOn(s -> s.getInstanceId().equals("110")).hasSize(1).filteredOn(s -> 443 == s.getPort()) + .hasSize(1); + } + + @Test + public void instanceWithMultiplePortsAndWithoutPrimaryPortNameSpecifiedOrHttpsPortShouldFallBackToHttp() { + Map labels = new HashMap<>(); + + Endpoints endPoint1 = new EndpointsBuilder().withNewMetadata().withName("endpoint5").withNamespace("test") + .withLabels(labels).endMetadata().addNewSubset().addNewAddress().withIp("ip1").withNewTargetRef() + .withUid("120").endTargetRef().endAddress().addNewPort("http", "https1", 443, "TCP") + .addNewPort("http", "https2", 8443, "TCP") + .addNewPort("http", "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?fieldSelector=metadata.name%3Dendpoint5") + .andReturn(200, endpoints).once(); + + Service service = new ServiceBuilder().withNewMetadata().withName("endpoint5").withNamespace("test") + .withLabels(labels).withAnnotations(labels).endMetadata().build(); + mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint5").andReturn(200, service) + .always(); + + final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + + final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, + KubernetesClient::services, new ServicePortSecureResolver(properties)); + + final List instances = discoveryClient.getInstances("endpoint5"); + + assertThat(instances).hasSize(1).filteredOn(s -> s.getHost().equals("ip1") && !s.isSecure()).hasSize(1) + .filteredOn(s -> s.getInstanceId().equals("120")).hasSize(1).filteredOn(s -> 80 == s.getPort()) + .hasSize(1); + } + + @Test + public void instanceWithMultiplePortsAndWithoutPrimaryPortNameSpecifiedShouldLogWarning() { + Map labels = new HashMap<>(); + + Endpoints endPoint1 = new EndpointsBuilder().withNewMetadata().withName("endpoint5").withNamespace("test") + .withLabels(labels).endMetadata().addNewSubset().addNewAddress().withIp("ip1").withNewTargetRef() + .withUid("130").endTargetRef().endAddress().addNewPort("http", "https", 443, "TCP") + .addNewPort("http", "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?fieldSelector=metadata.name%3Dendpoint5") + .andReturn(200, endpoints).once(); + + Service service = new ServiceBuilder().withNewMetadata().withName("endpoint5").withNamespace("test") + .withLabels(labels).withAnnotations(labels).endMetadata().build(); + mockServer.expect().get().withPath("/api/v1/namespaces/test/services/endpoint5").andReturn(200, service) + .always(); + + final KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(); + + final DiscoveryClient discoveryClient = new KubernetesDiscoveryClient(mockClient, properties, + KubernetesClient::services, new ServicePortSecureResolver(properties)); + + final List instances = discoveryClient.getInstances("endpoint5"); + + // We're returning the first discovered port to not change previous behaviour + assertThat(instances).hasSize(1).filteredOn(s -> s.getHost().equals("ip1") && s.isSecure()).hasSize(1) + .filteredOn(s -> s.getInstanceId().equals("130")).hasSize(1).filteredOn(s -> 443 == s.getPort()) + .hasSize(1); + } + }