SpringCloud/spring-cloud-kubernetes
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Unify endpoints calls and resolution (#1241)

Browse Source
This commit is contained in:
erabii
2023-02-24 16:07:21 +02:00
committed by GitHub
parent 4560341ff5
commit d172f1d8ab
5 changed files with 90 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/Fabric8EndpointsCatalogWatch.java
View File

@@ -16,7 +16,6 @@
package org.springframework.cloud.kubernetes.fabric8.discovery;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.function.Function;
@@ -26,12 +25,10 @@ import io.fabric8.kubernetes.api.model.EndpointAddress;
import io.fabric8.kubernetes.api.model.EndpointSubset;
import io.fabric8.kubernetes.api.model.Endpoints;
import io.fabric8.kubernetes.api.model.ObjectReference;
import io.fabric8.kubernetes.client.KubernetesClient;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.kubernetes.commons.discovery.EndpointNameAndNamespace;
import org.springframework.cloud.kubernetes.fabric8.Fabric8Utils;
import org.springframework.core.log.LogAccessor;
import static org.springframework.cloud.kubernetes.fabric8.discovery.KubernetesDiscoveryClientUtils.endpoints;
/**
* Implementation that is based on Endpoints.
@@ -41,30 +38,10 @@ import org.springframework.core.log.LogAccessor;
final class Fabric8EndpointsCatalogWatch
implements Function<Fabric8CatalogWatchContext, List<EndpointNameAndNamespace>> {
private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(Fabric8EndpointsCatalogWatch.class));
@Override
public List<EndpointNameAndNamespace> apply(Fabric8CatalogWatchContext context) {
List<Endpoints> endpoints;
KubernetesClient client = context.kubernetesClient();
if (context.properties().allNamespaces()) {
LOG.debug(() -> "discovering endpoints in all namespaces");
endpoints = client.endpoints().inAnyNamespace().withLabels(context.properties().serviceLabels()).list()
.getItems();
}
else if (!context.properties().namespaces().isEmpty()) {
LOG.debug(() -> "discovering endpoints in " + context.properties().namespaces());
List<Endpoints> inner = new ArrayList<>(context.properties().namespaces().size());
context.properties().namespaces().forEach(namespace -> inner.addAll(endpoints(context, namespace, client)));
endpoints = inner;
}
else {
String namespace = Fabric8Utils.getApplicationNamespace(context.kubernetesClient(), null, "catalog-watcher",
context.namespaceProvider());
LOG.debug(() -> "discovering endpoints in namespace : " + namespace);
endpoints = endpoints(context, namespace, client);
}
List<Endpoints> endpoints = endpoints(context.properties(), context.kubernetesClient(),
context.namespaceProvider(), "catalog-watcher", null);
/**
* <pre>
@@ -83,9 +60,4 @@ final class Fabric8EndpointsCatalogWatch
return Fabric8CatalogWatchContext.state(references);
}
private List<Endpoints> endpoints(Fabric8CatalogWatchContext context, String namespace, KubernetesClient client) {
return client.endpoints().inNamespace(namespace).withLabels(context.properties().serviceLabels()).list()
.getItems();
}
}

20
spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClient.java
View File

@@ -33,7 +33,6 @@ import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
import org.springframework.cloud.kubernetes.fabric8.Fabric8Utils;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.Environment;
import org.springframework.core.log.LogAccessor;
@@ -115,24 +114,7 @@ public class KubernetesDiscoveryClient implements DiscoveryClient, EnvironmentAw
}
public List<Endpoints> getEndPointsList(String serviceId) {
if (properties.allNamespaces()) {
LOG.debug(() -> "searching for endpoints in all namespaces");
return endpoints(client.endpoints().inAnyNamespace().withNewFilter(), properties, serviceId);
}
else if (properties.namespaces().isEmpty()) {
String namespace = Fabric8Utils.getApplicationNamespace(client, null, "discovery", namespaceProvider);
LOG.debug(() -> "searching for endpoints in namespace : " + namespace);
return endpoints(client.endpoints().inNamespace(namespace).withNewFilter(), properties, serviceId);
}
else {
LOG.debug(() -> "searching for endpoints in namespaces : " + properties.namespaces());
List<Endpoints> endpoints = new ArrayList<>();
for (String namespace : properties.namespaces()) {
endpoints.addAll(
endpoints(client.endpoints().inNamespace(namespace).withNewFilter(), properties, serviceId));
}
return endpoints;
}
return endpoints(properties, client, namespaceProvider, "fabric8-discovery", serviceId);
}
private List<ServiceInstance> getNamespaceServiceInstances(EndpointSubsetNS es, String serviceId) {

50
spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClientUtils.java
View File

@@ -30,14 +30,18 @@ import io.fabric8.kubernetes.api.model.Endpoints;
import io.fabric8.kubernetes.api.model.EndpointsList;
import io.fabric8.kubernetes.api.model.ObjectReference;
import io.fabric8.kubernetes.api.model.Service;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.dsl.FilterNested;
import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
import io.fabric8.kubernetes.client.dsl.Resource;
import jakarta.annotation.Nullable;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
import org.springframework.cloud.kubernetes.commons.discovery.DefaultKubernetesServiceInstance;
import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
import org.springframework.cloud.kubernetes.fabric8.Fabric8Utils;
import org.springframework.core.log.LogAccessor;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
@@ -163,11 +167,49 @@ final class KubernetesDiscoveryClientUtils {
return serviceMetadata;
}
static List<Endpoints> endpoints(
static List<Endpoints> endpoints(KubernetesDiscoveryProperties properties, KubernetesClient client,
KubernetesNamespaceProvider namespaceProvider, String target, @Nullable String serviceName) {
List<Endpoints> endpoints;
if (properties.allNamespaces()) {
LOG.debug(() -> "discovering endpoints in all namespaces");
endpoints = filteredEndpoints(client.endpoints().inAnyNamespace().withNewFilter(), properties, serviceName);
}
else if (!properties.namespaces().isEmpty()) {
LOG.debug(() -> "discovering endpoints in namespaces : " + properties.namespaces());
List<Endpoints> inner = new ArrayList<>(properties.namespaces().size());
properties.namespaces().forEach(namespace -> inner.addAll(filteredEndpoints(
client.endpoints().inNamespace(namespace).withNewFilter(), properties, serviceName)));
endpoints = inner;
}
else {
String namespace = Fabric8Utils.getApplicationNamespace(client, null, target, namespaceProvider);
LOG.debug(() -> "discovering endpoints in namespace : " + namespace);
endpoints = filteredEndpoints(client.endpoints().inNamespace(namespace).withNewFilter(), properties,
serviceName);
}
return endpoints;
}
/**
* serviceName can be null, in which case the filter for "metadata.name" will not be
* applied.
*/
static List<Endpoints> filteredEndpoints(
FilterNested<FilterWatchListDeletable<Endpoints, EndpointsList, Resource<Endpoints>>> filterNested,
KubernetesDiscoveryProperties properties, String serviceId) {
return filterNested.withField("metadata.name", serviceId).withLabels(properties.serviceLabels()).endFilter()
.list().getItems();
KubernetesDiscoveryProperties properties, @Nullable String serviceName) {
FilterNested<FilterWatchListDeletable<Endpoints, EndpointsList, Resource<Endpoints>>> partial = filterNested
.withLabels(properties.serviceLabels());
if (serviceName != null) {
partial = partial.withField("metadata.name", serviceName);
}
return partial.endFilter().list().getItems();
}
static List<EndpointAddress> addresses(EndpointSubset endpointSubset, KubernetesDiscoveryProperties properties) {

34
spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatchTest.java
View File

@@ -28,6 +28,7 @@ import io.fabric8.kubernetes.api.model.Endpoints;
import io.fabric8.kubernetes.api.model.EndpointsList;
import io.fabric8.kubernetes.api.model.ObjectReference;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.dsl.FilterNested;
import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
import io.fabric8.kubernetes.client.dsl.MixedOperation;
import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
@@ -73,6 +74,9 @@ class KubernetesCatalogWatchTest {
private static final FilterWatchListDeletable<Endpoints, EndpointsList, Resource<Endpoints>> FILTER_WATCH_LIST_DELETABLE = Mockito
.mock(FilterWatchListDeletable.class);
private static final FilterNested<FilterWatchListDeletable<Endpoints, EndpointsList, Resource<Endpoints>>> FILTER_NESTED = Mockito
.mock(FilterNested.class);
private static final ArgumentCaptor<HeartbeatEvent> HEARTBEAT_EVENT_ARGUMENT_CAPTOR = ArgumentCaptor
.forClass(HeartbeatEvent.class);
@@ -103,7 +107,8 @@ class KubernetesCatalogWatchTest {
createInAllNamespaceWatcher();
when(MIXED_OPERATION.list()).thenReturn(createSingleEndpointEndpointListByPodName("api-pod", "other-pod"))
when(FILTER_WATCH_LIST_DELETABLE.list())
.thenReturn(createSingleEndpointEndpointListByPodName("api-pod", "other-pod"))
.thenReturn(createSingleEndpointEndpointListByPodName("other-pod", "api-pod"));
kubernetesCatalogWatch.catalogServicesWatch();
@@ -134,7 +139,8 @@ class KubernetesCatalogWatchTest {
createInAllNamespaceWatcher();
when(MIXED_OPERATION.list()).thenReturn(createEndpointsListByServiceName("api-service", "other-service"))
when(FILTER_WATCH_LIST_DELETABLE.list())
.thenReturn(createEndpointsListByServiceName("api-service", "other-service"))
.thenReturn(createEndpointsListByServiceName("other-service", "api-service"));
kubernetesCatalogWatch.catalogServicesWatch();
@@ -169,7 +175,7 @@ class KubernetesCatalogWatchTest {
createInAllNamespaceWatcher();
when(MIXED_OPERATION.list())
when(FILTER_WATCH_LIST_DELETABLE.list())
.thenReturn(createSingleEndpointListWithNamespace("default", "api-pod", "other-pod"));
kubernetesCatalogWatch.catalogServicesWatch();
@@ -207,7 +213,7 @@ class KubernetesCatalogWatchTest {
EndpointsList endpoints = createSingleEndpointEndpointListWithoutSubsets();
when(MIXED_OPERATION.list()).thenReturn(endpoints);
when(FILTER_WATCH_LIST_DELETABLE.list()).thenReturn(endpoints);
kubernetesCatalogWatch.catalogServicesWatch();
// second execution on shuffleServices
@@ -241,7 +247,7 @@ class KubernetesCatalogWatchTest {
EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod");
endpoints.getItems().get(0).getSubsets().get(0).setAddresses(null);
when(MIXED_OPERATION.list()).thenReturn(endpoints);
when(FILTER_WATCH_LIST_DELETABLE.list()).thenReturn(endpoints);
kubernetesCatalogWatch.catalogServicesWatch();
// second execution on shuffleServices
@@ -275,7 +281,7 @@ class KubernetesCatalogWatchTest {
EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod");
endpoints.getItems().get(0).getSubsets().get(0).getAddresses().get(0).setTargetRef(null);
when(MIXED_OPERATION.list()).thenReturn(endpoints);
when(FILTER_WATCH_LIST_DELETABLE.list()).thenReturn(endpoints);
kubernetesCatalogWatch.catalogServicesWatch();
// second execution on shuffleServices
@@ -360,10 +366,6 @@ class KubernetesCatalogWatchTest {
private void createInAllNamespaceWatcher() {
when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
when(MIXED_OPERATION.inAnyNamespace()).thenReturn(MIXED_OPERATION);
when(MIXED_OPERATION.withLabels(Map.of())).thenReturn(MIXED_OPERATION);
// all-namespaces = true
KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(true, true, Set.of(), true, 60,
false, "", Set.of(), Map.of(), "", null, 0, false);
@@ -371,6 +373,14 @@ class KubernetesCatalogWatchTest {
kubernetesCatalogWatch = new KubernetesCatalogWatch(CLIENT, properties, namespaceProvider);
kubernetesCatalogWatch.setApplicationEventPublisher(APPLICATION_EVENT_PUBLISHER);
kubernetesCatalogWatch.postConstruct();
when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
when(MIXED_OPERATION.inAnyNamespace()).thenReturn(MIXED_OPERATION);
when(MIXED_OPERATION.withLabels(Map.of())).thenReturn(MIXED_OPERATION);
when(MIXED_OPERATION.withNewFilter()).thenReturn(FILTER_NESTED);
when(FILTER_NESTED.withLabels(Map.of())).thenReturn(FILTER_NESTED);
when(FILTER_NESTED.endFilter()).thenReturn(FILTER_WATCH_LIST_DELETABLE);
}
private void createInSpecificNamespaceWatcher() {
@@ -386,7 +396,9 @@ class KubernetesCatalogWatchTest {
when(namespaceProvider.getNamespace()).thenReturn("catalog-watcher-namespace");
when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
when(MIXED_OPERATION.inNamespace("catalog-watcher-namespace")).thenReturn(NON_NAMESPACE_OPERATION);
when(NON_NAMESPACE_OPERATION.withLabels(Map.of())).thenReturn(FILTER_WATCH_LIST_DELETABLE);
when(NON_NAMESPACE_OPERATION.withNewFilter()).thenReturn(FILTER_NESTED);
when(FILTER_NESTED.withLabels(Map.of())).thenReturn(FILTER_NESTED);
when(FILTER_NESTED.endFilter()).thenReturn(FILTER_WATCH_LIST_DELETABLE);
}
}

33
spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesDiscoveryClientTests.java
View File

@@ -80,7 +80,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("serviceId");
Assertions.assertEquals(result.size(), 0);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in all namespaces"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in all namespaces"));
}
/**
@@ -105,7 +105,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("blue-service");
Assertions.assertEquals(result.size(), 1);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in all namespaces"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in all namespaces"));
}
/**
@@ -130,7 +130,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("blue-service");
Assertions.assertEquals(result.size(), 1);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in all namespaces"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in all namespaces"));
}
/**
@@ -155,7 +155,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("blue-service");
Assertions.assertEquals(result.size(), 0);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in all namespaces"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in all namespaces"));
}
/**
@@ -181,7 +181,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("service-one");
Assertions.assertEquals(result.size(), 1);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in all namespaces"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in all namespaces"));
}
/**
@@ -210,7 +210,7 @@ class KubernetesDiscoveryClientTests {
Assertions.assertEquals(
result.stream().map(Endpoints::getMetadata).map(ObjectMeta::getNamespace).sorted().toList(),
List.of("a", "b"));
Assertions.assertTrue(output.getOut().contains("searching for endpoints in all namespaces"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in all namespaces"));
}
/**
@@ -231,7 +231,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("serviceId");
Assertions.assertEquals(result.size(), 0);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in namespace : test"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespace : test"));
}
/**
@@ -256,7 +256,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("blue-service");
Assertions.assertEquals(result.size(), 1);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in namespace : test"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespace : test"));
}
/**
@@ -281,7 +281,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("blue-service");
Assertions.assertEquals(result.size(), 1);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in namespace : test"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespace : test"));
}
/**
@@ -306,7 +306,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("blue-service");
Assertions.assertEquals(result.size(), 0);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in namespace : test"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespace : test"));
}
/**
@@ -332,7 +332,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("service-one");
Assertions.assertEquals(result.size(), 1);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in namespace : test"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespace : test"));
}
/**
@@ -361,7 +361,7 @@ class KubernetesDiscoveryClientTests {
Assertions.assertEquals(
result.stream().map(Endpoints::getMetadata).map(ObjectMeta::getNamespace).sorted().toList(),
List.of("test"));
Assertions.assertTrue(output.getOut().contains("searching for endpoints in namespace : test"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespace : test"));
}
/**
@@ -382,7 +382,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("serviceId");
Assertions.assertEquals(result.size(), 0);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in namespaces : [test]"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespaces : [test]"));
}
/**
@@ -407,7 +407,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("blue-service");
Assertions.assertEquals(result.size(), 1);
Assertions.assertTrue(output.getOut().contains("searching for endpoints in namespaces : [test]"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespaces : [test]"));
}
/**
@@ -435,7 +435,7 @@ class KubernetesDiscoveryClientTests {
List<Endpoints> result = discoveryClient.getEndPointsList("blue-service");
Assertions.assertEquals(result.size(), 1);
Assertions.assertEquals(result.get(0).getMetadata().getLabels(), Map.of("color", "blue", "shape", "round"));
Assertions.assertTrue(output.getOut().contains("searching for endpoints in namespaces : [a]"));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespaces : [a]"));
}
/**
@@ -464,8 +464,7 @@ class KubernetesDiscoveryClientTests {
KubernetesDiscoveryClient discoveryClient = new KubernetesDiscoveryClient(client, properties, null, null, null);
List<Endpoints> result = discoveryClient.getEndPointsList("blue-service");
Assertions.assertEquals(result.size(), 2);
Assertions
.assertTrue(output.getOut().contains("searching for endpoints in namespaces : " + namespacesAsString));
Assertions.assertTrue(output.getOut().contains("discovering endpoints in namespaces : " + namespacesAsString));
}
private void createEndpoints(String namespace, String name, Map<String, String> labels) {