KubernetesCatalogWatch implements ApplicationEventPublisherAware { @Scheduled(fixedDelayString = "${spring.cloud.kubernetes.discovery.catalogServicesWatchDelay:30000}") public void catalogServicesWatch() { try { List<String> previousState = this.catalogEndpointsState.get(); 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) .flatMap(Collection::stream).map(EndpointAddress::getTargetRef).filter(Objects::nonNull) .map(ObjectReference::getName) .sorted(String::compareTo).collect(Collectors.toList()); this.catalogEndpointsState.set(endpointsPodNames); if (!endpointsPodNames.equals(previousState)) { logger.trace("Received endpoints update from kubernetesClient: {}", endpointsPodNames); this.publisher.publishEvent(new HeartbeatEvent(this, endpointsPodNames)); } } catch (Exception e) { logger.error("Error watching Kubernetes Services", e); } } KubernetesCatalogWatch(KubernetesClient kubernetesClient, KubernetesDiscoveryProperties properties); @Override void setApplicationEventPublisher(ApplicationEventPublisher publisher); @Scheduled(fixedDelayString = "${spring.cloud.kubernetes.discovery.catalogServicesWatchDelay:30000}") void catalogServicesWatch();  }
@Test public void testEndpointsWithoutAddresses() { EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod"); endpoints.getItems().get(0).getSubsets().get(0).setAddresses(null); 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(); 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.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(); this.underTest.catalogServicesWatch(); verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); }
@Test public void testEndpointsWithoutTargetRefs() { EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod"); endpoints.getItems().get(0).getSubsets().get(0).getAddresses().get(0).setTargetRef(null); 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(); 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.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(); this.underTest.catalogServicesWatch(); verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); }
@Test public void testRandomOrderChangePods() throws Exception { when(this.endpointsOperation.list()) .thenReturn(createSingleEndpointEndpointListByPodName("api-pod", "other-pod")) .thenReturn(createSingleEndpointEndpointListByPodName("other-pod", "api-pod")); when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); when(this.kubernetesClient.endpoints().withLabels(anyMap())).thenReturn(this.endpointsOperation); this.underTest.catalogServicesWatch(); 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.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(); this.underTest.catalogServicesWatch(); verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); }
@Test public void testRandomOrderChangeServices() throws Exception { when(this.endpointsOperation.list()) .thenReturn(createEndpointsListByServiceName("api-service", "other-service")) .thenReturn(createEndpointsListByServiceName("other-service", "api-service")); when(this.kubernetesClient.endpoints()).thenReturn(this.endpointsOperation); when(this.kubernetesClient.endpoints().withLabels(anyMap())).thenReturn(this.endpointsOperation); this.underTest.catalogServicesWatch(); 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.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(); this.underTest.catalogServicesWatch(); verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); }
@Test public void testEventBody() throws Exception { when(this.endpointsOperation.list()) .thenReturn(createSingleEndpointEndpointListByPodName("api-pod", "other-pod")); 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.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(); 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 testEndpointsWithoutSubsets() { EndpointsList endpoints = createSingleEndpointEndpointListWithoutSubsets(); 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(); this.underTest.catalogServicesWatch(); verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); }
@Test public void testEndpointsWithoutSubsetsAllNamespaces() { EndpointsList endpoints = createSingleEndpointEndpointListWithoutSubsets(); 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(); this.underTest.catalogServicesWatch(); verify(this.applicationEventPublisher).publishEvent(any(HeartbeatEvent.class)); }