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

addressing related fix to #472

Browse Source
This commit is contained in:
Haytham Mohamed
2020-07-09 18:16:37 -05:00
parent 9df02f137d
commit 0951e892ce
3 changed files with 163 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatch.java
View File

@@ -45,12 +45,16 @@ public class KubernetesCatalogWatch implements ApplicationEventPublisherAware {
private final KubernetesClient kubernetesClient;
private final KubernetesDiscoveryProperties properties;
private final AtomicReference<List<String>> 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> endpoints = this.kubernetesClient.endpoints().list()
.getItems();
List<Endpoints> endpoints = this.properties.isAllNamespaces()
? this.kubernetesClient.endpoints().inAnyNamespace()
.withLabels(properties.getServiceLabels()).list().getItems()
: this.kubernetesClient.endpoints()
.withLabels(properties.getServiceLabels()).list().getItems();
List<String> endpointsPodNames = endpoints.stream().map(Endpoints::getSubsets)
.filter(Objects::nonNull).flatMap(Collection::stream)
.map(EndpointSubset::getAddresses).filter(Objects::nonNull)

5
spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesCatalogWatchAutoConfiguration.java
View File

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

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

@@ -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<String> 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