Use endpoint slices for fabric8 catalog watcher (#1149)

Author: wind57

docs/src/main/asciidoc/discovery-client.adoc

@@ -134,8 +134,19 @@ should not rely on the details. Instead, they should see if there are difference
 The endpoints will be queried in either all namespaces (enabled via `spring.cloud.kubernetes.discovery.all-namespaces=true`), or
 we will use: xref:property-source-config.adoc#namespace-resolution[Namespace Resolution].
 
-
-
-
 In order to enable this functionality you need to add
 `@EnableScheduling` on a configuration class in your application.
+
+By default, we use the `Endpoints`(see https://kubernetes.io/docs/concepts/services-networking/service/#endpoints) API to find out the current state of services. There is another way though, via `EndpointSlices` (https://kubernetes.io/docs/concepts/services-networking/endpoint-slices/). Such support can be enabled via a property: `spring.cloud.kubernetes.discovery.use-endpoint-slices=true` (by default it is `false`). Of course, your cluster has to support it also. As a matter of fact, if you enable this property, but your cluster does not support it, we will fail starting the application. If you decide to enable such support, you also need proper Role/ClusterRole set-up. For example:
+
+```
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  namespace: default
+  name: namespace-reader
+rules:
+  - apiGroups: ["discovery.k8s.io"]
+    resources: ["endpointslices"]
+    verbs: ["get", "list", "watch"]
+```

spring-cloud-kubernetes-client-discovery/src/test/java/org/springframework/cloud/kubernetes/client/discovery/KubernetesInformerDiscoveryClientTests.java

@@ -124,7 +124,7 @@ public void testServiceWithUnsetPortNames() {
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, true,
 				Set.of(), true, 60, false, null, Set.of(), Map.of(), null,
-				KubernetesDiscoveryProperties.Metadata.DEFAULT, 0);
+				KubernetesDiscoveryProperties.Metadata.DEFAULT, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -157,7 +157,7 @@ public void testDiscoveryWithServiceLabels() {
 		labels.put("spring", "true");
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, true,
-				Set.of(), true, 60, false, null, Set.of(), labels, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), labels, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("",
 				sharedInformerFactory, serviceLister, null, null, null, kubernetesDiscoveryProperties);
@@ -176,7 +176,7 @@ public void testDiscoveryInstancesWithServiceLabels() {
 		labels.put("spring", "true");
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, true,
-				Set.of(), true, 60, false, null, Set.of(), labels, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), labels, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -192,7 +192,7 @@ public void testDiscoveryInstancesWithSecuredServiceByAnnotations() {
 		Lister<V1Service> serviceLister = setupServiceLister(testServiceSecuredAnnotation1);
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpoints1);
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, true,
-				Set.of(), true, 60, false, null, Set.of(), new HashMap<>(), null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), new HashMap<>(), null, null, 0, false);
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
 		assertThat(discoveryClient.getServices().toArray())
@@ -207,7 +207,7 @@ public void testDiscoveryInstancesWithSecuredServiceByLabels() {
 		Lister<V1Service> serviceLister = setupServiceLister(testServiceSecuredLabel1);
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpoints1);
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, true,
-				Set.of(), true, 60, false, null, Set.of(), new HashMap<>(), null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), new HashMap<>(), null, null, 0, false);
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
 
@@ -235,7 +235,7 @@ public void testDiscoveryGetInstanceAllNamespaceShouldWork() {
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpoints1);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, true,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -250,7 +250,7 @@ public void testDiscoveryGetInstanceOneNamespaceShouldWork() {
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpoints1);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -277,7 +277,7 @@ public void testDiscoveryGetInstanceWithNotReadyAddressesIncludedShouldWork() {
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpointWithoutReadyAddresses);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, true, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, true, null, Set.of(), null, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -318,7 +318,7 @@ public void instanceWithMultiplePortsAndPrimaryPortNameConfiguredWithLabelShould
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpointWithMultiplePorts);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -337,7 +337,7 @@ public void instanceWithMultiplePortsAndMisconfiguredPrimaryPortNameInLabelShoul
 				testEndpointWithMultiplePortsWithoutSupportedPortNames);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -353,7 +353,7 @@ public void instanceWithMultiplePortsAndGenericPrimaryPortNameConfiguredShouldWo
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpointWithMultiplePorts);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, false, null, Set.of(), null, "https", null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, "https", null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -369,7 +369,7 @@ public void instanceWithMultiplePortsAndMisconfiguredGenericPrimaryPortNameShoul
 				testEndpointWithMultiplePortsWithoutSupportedPortNames);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, false, null, Set.of(), null, "oops", null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, "oops", null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -384,7 +384,7 @@ public void instanceWithMultiplePortsAndWithoutPrimaryPortNameSpecifiedShouldFal
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpointWithMultiplePorts);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -399,7 +399,7 @@ public void instanceWithMultiplePortsAndWithoutPrimaryPortNameSpecifiedOrHttpsPo
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpointWithMultiplePortsWithoutHttps);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -415,7 +415,7 @@ public void instanceWithMultiplePortsAndWithoutAnyConfigurationShouldPickTheFirs
 				testEndpointWithMultiplePortsWithoutSupportedPortNames);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, true);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient("namespace1",
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);
@@ -430,7 +430,7 @@ public void getInstancesShouldReturnInstancesWithTheSameServiceIdFromNamespaces(
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpoints1, testEndpoints2);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, true,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, false);
 
 		KubernetesInformerDiscoveryClient discoveryClient = new KubernetesInformerDiscoveryClient(null,
 				sharedInformerFactory, serviceLister, endpointsLister, null, null, kubernetesDiscoveryProperties);

spring-cloud-kubernetes-client-discovery/src/test/java/org/springframework/cloud/kubernetes/client/discovery/reactive/KubernetesInformerReactiveDiscoveryClientTests.java

@@ -101,7 +101,7 @@ public void testDiscoveryGetInstanceAllNamespaceShouldWork() {
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpoints1);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, true,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, false);
 
 		KubernetesInformerReactiveDiscoveryClient discoveryClient = new KubernetesInformerReactiveDiscoveryClient(
 				new KubernetesNamespaceProvider(new MockEnvironment()), sharedInformerFactory, serviceLister,
@@ -120,7 +120,7 @@ public void testDiscoveryGetInstanceOneNamespaceShouldWork() {
 		Lister<V1Endpoints> endpointsLister = setupEndpointsLister(testEndpoints1);
 
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, false,
-				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0);
+				Set.of(), true, 60, false, null, Set.of(), null, null, null, 0, false);
 
 		KubernetesNamespaceProvider kubernetesNamespaceProvider = mock(KubernetesNamespaceProvider.class);
 		when(kubernetesNamespaceProvider.getNamespace()).thenReturn("namespace1");

spring-cloud-kubernetes-client-loadbalancer/src/test/java/org/springframework/cloud/kubernetes/client/loadbalancer/KubernetesClientServicesListSupplierTests.java

@@ -147,7 +147,7 @@ void getListAllNamespaces() {
 		when(kubernetesNamespaceProvider.getNamespace()).thenReturn("default");
 		KubernetesDiscoveryProperties kubernetesDiscoveryProperties = new KubernetesDiscoveryProperties(true, true,
 				Set.of(), true, 60, false, null, Set.of(), Map.of(), null,
-				KubernetesDiscoveryProperties.Metadata.DEFAULT, 0);
+				KubernetesDiscoveryProperties.Metadata.DEFAULT, 0, false);
 		CoreV1Api coreV1Api = new CoreV1Api();
 		KubernetesClientServiceInstanceMapper mapper = new KubernetesClientServiceInstanceMapper(
 				new KubernetesLoadBalancerProperties(), kubernetesDiscoveryProperties);

spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/discovery/KubernetesDiscoveryProperties.java

@@ -42,6 +42,7 @@
  * fetched from the Kubernetes API server.
  * @param primaryPortName If set then the port with a given name is used as primary when
  * multiple ports are defined for a service.
+ * @param useEndpointSlices use EndpointSlice instead of Endpoints
  */
 // @formatter:off
 @ConfigurationProperties("spring.cloud.kubernetes.discovery")
@@ -54,14 +55,15 @@ public record KubernetesDiscoveryProperties(
 		@DefaultValue({"443", "8443"}) Set<Integer> knownSecurePorts,
 		@DefaultValue Map<String, String> serviceLabels, String primaryPortName,
 		@DefaultValue Metadata metadata,
-		@DefaultValue("" + DEFAULT_ORDER) int order) {
+		@DefaultValue("" + DEFAULT_ORDER) int order,
+		boolean useEndpointSlices) {
 // @formatter:on
 
 	/**
 	 * Default instance.
 	 */
 	public static final KubernetesDiscoveryProperties DEFAULT = new KubernetesDiscoveryProperties(true, false, Set.of(),
-			true, 60, false, null, Set.of(), Map.of(), null, KubernetesDiscoveryProperties.Metadata.DEFAULT, 0);
+			true, 60, false, null, Set.of(), Map.of(), null, KubernetesDiscoveryProperties.Metadata.DEFAULT, 0, false);
 
 	/**
 	 * @param addLabels include labels as metadata

spring-cloud-kubernetes-commons/src/test/java/org/springframework/cloud/kubernetes/commons/discovery/KubernetesDiscoveryPropertiesTests.java

@@ -52,6 +52,7 @@ void testBindingWhenNoPropertiesProvided() {
 					assertThat(props.serviceLabels()).isEmpty();
 					assertThat(props.primaryPortName()).isNull();
 					assertThat(props.order()).isZero();
+					assertThat(props.useEndpointSlices()).isFalse();
 				});
 	}
 
@@ -61,6 +62,7 @@ void testBindingWhenSomePropertiesProvided() {
 				.withPropertyValues("spring.cloud.kubernetes.discovery.filter=some-filter",
 						"spring.cloud.kubernetes.discovery.knownSecurePorts[0]=222",
 						"spring.cloud.kubernetes.discovery.metadata.labelsPrefix=labelsPrefix",
+						"spring.cloud.kubernetes.discovery.use-endpoint-slices=true",
 						"spring.cloud.kubernetes.discovery.namespaces[0]=ns1",
 						"spring.cloud.kubernetes.discovery.namespaces[1]=ns2")
 				.run(context -> {
@@ -81,6 +83,7 @@ void testBindingWhenSomePropertiesProvided() {
 					assertThat(props.serviceLabels()).isEmpty();
 					assertThat(props.primaryPortName()).isNull();
 					assertThat(props.order()).isZero();
+					assertThat(props.useEndpointSlices()).isTrue();
 				});
 	}
 

spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/Fabric8CatalogWatchContext.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2012-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.kubernetes.fabric8.discovery;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Stream;
+
+import io.fabric8.kubernetes.api.model.ObjectReference;
+import io.fabric8.kubernetes.client.KubernetesClient;
+
+import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
+import org.springframework.cloud.kubernetes.commons.discovery.EndpointNameAndNamespace;
+import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
+
+/**
+ * A simple holder for some instances needed for either Endpoints or EndpointSlice catalog
+ * implementations.
+ *
+ * @author wind57
+ */
+record Fabric8CatalogWatchContext(KubernetesClient kubernetesClient, KubernetesDiscoveryProperties properties,
+		KubernetesNamespaceProvider namespaceProvider) {
+
+	static List<EndpointNameAndNamespace> state(Stream<ObjectReference> references) {
+		return references.filter(Objects::nonNull).map(x -> new EndpointNameAndNamespace(x.getName(), x.getNamespace()))
+				.sorted(Comparator.comparing(EndpointNameAndNamespace::endpointName, String::compareTo)).toList();
+	}
+
+}