Refactor k8s client catalog watcher it (#1836)

* placeholder commit Signed-off-by: wind57 <eugen.rabii@gmail.com> * started work Signed-off-by: wind57 <eugen.rabii@gmail.com> * dirty Signed-off-by: wind57 <eugen.rabii@gmail.com> * refactored Signed-off-by: wind57 <eugen.rabii@gmail.com> --------- Signed-off-by: wind57 <eugen.rabii@gmail.com>
spring-cloud · Jan 23, 2025 · fd21434 · fd21434
1 parent 1eec4a0
commit fd21434
Show file tree

Hide file tree

Showing 8 changed files with 401 additions and 534 deletions.
diff --git a/...c8CatalogWatchEndpointSlicesFilterIT.java → .../Fabric8CatalogWatchEndpointSlicesIT.java b/...c8CatalogWatchEndpointSlicesFilterIT.java → .../Fabric8CatalogWatchEndpointSlicesIT.java
@@ -34,7 +34,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Primary;
 
-import static org.springframework.cloud.kubernetes.fabric8.catalog.watch.Fabric8CatalogWatchEndpointSlicesFilterIT.TestConfig;
+import static org.springframework.cloud.kubernetes.fabric8.catalog.watch.Fabric8CatalogWatchEndpointSlicesIT.TestConfig;
 import static org.springframework.cloud.kubernetes.fabric8.catalog.watch.TestAssertions.assertLogStatement;
 import static org.springframework.cloud.kubernetes.fabric8.catalog.watch.TestAssertions.invokeAndAssert;
 
@@ -43,7 +43,7 @@
  */
 @SpringBootTest(classes = { KubernetesCatalogWatchAutoConfiguration.class, TestConfig.class, Application.class },
 		webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
-class Fabric8CatalogWatchEndpointSlicesFilterIT extends Fabric8CatalogWatchBase {
+class Fabric8CatalogWatchEndpointSlicesIT extends Fabric8CatalogWatchBase {
 
 	@LocalServerPort
 	private int port;

diff --git a/...amework/cloud/kubernetes/k8s/client/catalog/watcher/KubernetesClientCatalogWatchBase.java b/...amework/cloud/kubernetes/k8s/client/catalog/watcher/KubernetesClientCatalogWatchBase.java
@@ -0,0 +1,84 @@
+/*
+ * Copyright 2012-2024 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.k8s.client.catalog.watcher;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.Map;
+import java.util.Set;
+
+import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.util.Config;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.k3s.K3sContainer;
+
+import org.springframework.boot.test.system.OutputCaptureExtension;
+import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
+import org.springframework.cloud.kubernetes.integration.tests.commons.Commons;
+import org.springframework.cloud.kubernetes.integration.tests.commons.fabric8_client.Util;
+import org.springframework.test.context.TestPropertySource;
+
+/**
+ * @author wind57
+ */
+
+@TestPropertySource(
+		properties = { "spring.main.cloud-platform=kubernetes", "spring.cloud.config.import-check.enabled=false",
+				"spring.cloud.kubernetes.discovery.catalogServicesWatchDelay=2000",
+				"spring.cloud.kubernetes.client.namespace=default",
+				"logging.level.org.springframework.cloud.kubernetes.client.discovery.catalog=DEBUG" })
+@ExtendWith(OutputCaptureExtension.class)
+abstract class KubernetesClientCatalogWatchBase {
+
+	protected static final String NAMESPACE = "default";
+
+	protected static final String NAMESPACE_A = "a";
+
+	protected static final String NAMESPACE_B = "b";
+
+	protected static final K3sContainer K3S = Commons.container();
+
+	protected static Util util;
+
+	@BeforeAll
+	protected static void beforeAll() {
+		K3S.start();
+		util = new Util(K3S);
+	}
+
+	protected static KubernetesDiscoveryProperties discoveryProperties(boolean useEndpointSlices,
+			Set<String> namespaces) {
+		return new KubernetesDiscoveryProperties(true, false, namespaces, true, 60, false, null, Set.of(443, 8443),
+				Map.of(), null, KubernetesDiscoveryProperties.Metadata.DEFAULT, 0, useEndpointSlices, false, null);
+	}
+
+	protected static ApiClient apiClient() {
+		String kubeConfigYaml = K3S.getKubeConfigYaml();
+
+		ApiClient client;
+		try {
+			client = Config.fromConfig(new StringReader(kubeConfigYaml));
+		}
+		catch (IOException e) {
+			throw new RuntimeException(e);
+		}
+		return new CoreV1Api(client).getApiClient();
+	}
+
+}
diff --git a/...d/kubernetes/k8s/client/catalog/watcher/KubernetesClientCatalogWatchEndpointSlicesIT.java b/...d/kubernetes/k8s/client/catalog/watcher/KubernetesClientCatalogWatchEndpointSlicesIT.java
@@ -0,0 +1,99 @@
+/*
+ * Copyright 2013-2025 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.k8s.client.catalog.watcher;
+
+import java.util.Set;
+
+import io.kubernetes.client.openapi.ApiClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
+import org.springframework.cloud.kubernetes.integration.tests.commons.Images;
+import org.springframework.cloud.kubernetes.integration.tests.commons.Phase;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Primary;
+
+import static org.springframework.cloud.kubernetes.k8s.client.catalog.watcher.TestAssertions.assertLogStatement;
+import static org.springframework.cloud.kubernetes.k8s.client.catalog.watcher.TestAssertions.invokeAndAssert;
+
+@SpringBootTest(classes = { KubernetesClientCatalogWatchEndpointSlicesIT.TestConfig.class, Application.class },
+		webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+class KubernetesClientCatalogWatchEndpointSlicesIT extends KubernetesClientCatalogWatchBase {
+
+	@LocalServerPort
+	private int port;
+
+	@BeforeEach
+	void beforeEach() {
+
+		util.createNamespace(NAMESPACE_A);
+		util.createNamespace(NAMESPACE_B);
+
+		Images.loadBusybox(K3S);
+
+		util.busybox(NAMESPACE_A, Phase.CREATE);
+		util.busybox(NAMESPACE_B, Phase.CREATE);
+
+	}
+
+	@AfterEach
+	void afterEach() {
+		// busybox is deleted as part of the assertions, thus not seen here
+		util.deleteNamespace(NAMESPACE_A);
+		util.deleteNamespace(NAMESPACE_B);
+	}
+
+	/**
+	 * <pre>
+	 *     - we deploy a busybox service with 2 replica pods in two namespaces : a, b
+	 *     - we use endpoints
+	 *     - we enable namespace filtering for 'default' and 'a'
+	 *     - we receive an event from KubernetesCatalogWatcher, assert what is inside it
+	 *     - delete the busybox service in 'a' and 'b'
+	 *     - assert that we receive an empty response
+	 * </pre>
+	 */
+	@Test
+	void testCatalogWatchWithEndpoints(CapturedOutput output) {
+		assertLogStatement(output, "stateGenerator is of type: KubernetesEndpointSlicesCatalogWatch");
+		invokeAndAssert(util, Set.of(NAMESPACE_A, NAMESPACE_B), port, NAMESPACE_A);
+	}
+
+	@TestConfiguration
+	static class TestConfig {
+
+		@Bean
+		@Primary
+		ApiClient client() {
+			return apiClient();
+		}
+
+		@Bean
+		@Primary
+		KubernetesDiscoveryProperties kubernetesDiscoveryProperties() {
+			return discoveryProperties(true, Set.of(NAMESPACE, NAMESPACE_A));
+		}
+
+	}
+
+}
diff --git a/.../cloud/kubernetes/k8s/client/catalog/watcher/KubernetesClientCatalogWatchEndpointsIT.java b/.../cloud/kubernetes/k8s/client/catalog/watcher/KubernetesClientCatalogWatchEndpointsIT.java
@@ -0,0 +1,99 @@
+/*
+ * Copyright 2013-2025 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.k8s.client.catalog.watcher;
+
+import java.util.Set;
+
+import io.kubernetes.client.openapi.ApiClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
+import org.springframework.cloud.kubernetes.integration.tests.commons.Images;
+import org.springframework.cloud.kubernetes.integration.tests.commons.Phase;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Primary;
+
+import static org.springframework.cloud.kubernetes.k8s.client.catalog.watcher.TestAssertions.assertLogStatement;
+import static org.springframework.cloud.kubernetes.k8s.client.catalog.watcher.TestAssertions.invokeAndAssert;
+
+@SpringBootTest(classes = { KubernetesClientCatalogWatchEndpointsIT.TestConfig.class, Application.class },
+		webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+class KubernetesClientCatalogWatchEndpointsIT extends KubernetesClientCatalogWatchBase {
+
+	@LocalServerPort
+	private int port;
+
+	@BeforeEach
+	void beforeEach() {
+
+		util.createNamespace(NAMESPACE_A);
+		util.createNamespace(NAMESPACE_B);
+
+		Images.loadBusybox(K3S);
+
+		util.busybox(NAMESPACE_A, Phase.CREATE);
+		util.busybox(NAMESPACE_B, Phase.CREATE);
+
+	}
+
+	@AfterEach
+	void afterEach() {
+		// busybox is deleted as part of the assertions, thus not seen here
+		util.deleteNamespace(NAMESPACE_A);
+		util.deleteNamespace(NAMESPACE_B);
+	}
+
+	/**
+	 * <pre>
+	 *     - we deploy a busybox service with 2 replica pods in two namespaces : a, b
+	 *     - we use endpoints
+	 *     - we enable namespace filtering for 'default' and 'a'
+	 *     - we receive an event from KubernetesCatalogWatcher, assert what is inside it
+	 *     - delete the busybox service in 'a' and 'b'
+	 *     - assert that we receive an empty response
+	 * </pre>
+	 */
+	@Test
+	void testCatalogWatchWithEndpoints(CapturedOutput output) {
+		assertLogStatement(output, "stateGenerator is of type: KubernetesEndpointsCatalogWatch");
+		invokeAndAssert(util, Set.of(NAMESPACE_A, NAMESPACE_B), port, NAMESPACE_A);
+	}
+
+	@TestConfiguration
+	static class TestConfig {
+
+		@Bean
+		@Primary
+		ApiClient client() {
+			return apiClient();
+		}
+
+		@Bean
+		@Primary
+		KubernetesDiscoveryProperties kubernetesDiscoveryProperties() {
+			return discoveryProperties(false, Set.of(NAMESPACE, NAMESPACE_A));
+		}
+
+	}
+
+}