Merge branch '3.1.x' into 3.2.x
This commit is contained in:
Ryan Baxter
@@ -34,7 +34,7 @@ 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.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 @@ import static org.springframework.cloud.kubernetes.fabric8.catalog.watch.TestAss
|
||||
*/
|
||||
@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;
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,241 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-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.k8s.client.catalog.watcher;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
|
||||
import io.kubernetes.client.openapi.models.V1Deployment;
|
||||
import io.kubernetes.client.openapi.models.V1Ingress;
|
||||
import io.kubernetes.client.openapi.models.V1Service;
|
||||
import org.junit.jupiter.api.AfterAll;
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.MethodOrderer;
|
||||
import org.junit.jupiter.api.Order;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.TestMethodOrder;
|
||||
import org.testcontainers.k3s.K3sContainer;
|
||||
import reactor.netty.http.client.HttpClient;
|
||||
import reactor.util.retry.Retry;
|
||||
import reactor.util.retry.RetryBackoffSpec;
|
||||
|
||||
import org.springframework.cloud.kubernetes.commons.discovery.EndpointNameAndNamespace;
|
||||
import org.springframework.cloud.kubernetes.integration.tests.commons.Commons;
|
||||
import org.springframework.cloud.kubernetes.integration.tests.commons.Images;
|
||||
import org.springframework.cloud.kubernetes.integration.tests.commons.Phase;
|
||||
import org.springframework.cloud.kubernetes.integration.tests.commons.native_client.Util;
|
||||
import org.springframework.core.ParameterizedTypeReference;
|
||||
import org.springframework.core.ResolvableType;
|
||||
import org.springframework.http.HttpMethod;
|
||||
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
|
||||
import org.springframework.web.reactive.function.client.WebClient;
|
||||
|
||||
import static org.awaitility.Awaitility.await;
|
||||
import static org.springframework.cloud.kubernetes.integration.tests.commons.Commons.waitForLogStatement;
|
||||
|
||||
/**
|
||||
* @author wind57
|
||||
*/
|
||||
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
|
||||
class KubernetesClientCatalogWatchIT {
|
||||
|
||||
private static final String APP_NAME = "spring-cloud-kubernetes-k8s-client-catalog-watcher";
|
||||
|
||||
private static final String NAMESPACE = "default";
|
||||
|
||||
private static final String NAMESPACE_A = "namespacea";
|
||||
|
||||
private static final String NAMESPACE_B = "namespaceb";
|
||||
|
||||
private static final K3sContainer K3S = Commons.container();
|
||||
|
||||
private static final String DOCKER_IMAGE = "docker.io/springcloud/" + APP_NAME + ":" + Commons.pomVersion();
|
||||
|
||||
private static Util util;
|
||||
|
||||
@BeforeAll
|
||||
static void beforeAll() throws Exception {
|
||||
K3S.start();
|
||||
Commons.validateImage(APP_NAME, K3S);
|
||||
Commons.loadSpringCloudKubernetesImage(APP_NAME, K3S);
|
||||
|
||||
Images.loadBusybox(K3S);
|
||||
|
||||
util = new Util(K3S);
|
||||
util.setUp(NAMESPACE);
|
||||
app(Phase.CREATE);
|
||||
}
|
||||
|
||||
@AfterAll
|
||||
static void afterAll() {
|
||||
util.deleteClusterWide(NAMESPACE, Set.of(NAMESPACE_A, NAMESPACE_B));
|
||||
util.deleteNamespace(NAMESPACE_A);
|
||||
util.deleteNamespace(NAMESPACE_B);
|
||||
app(Phase.DELETE);
|
||||
}
|
||||
|
||||
@BeforeEach
|
||||
void beforeEach() {
|
||||
util.busybox(NAMESPACE, Phase.CREATE);
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* - we deploy a busybox service with 2 replica pods
|
||||
* - we receive an event from KubernetesCatalogWatcher, assert what is inside it
|
||||
* - delete the busybox service
|
||||
* - assert that we receive only spring-cloud-kubernetes-client-catalog-watcher pod
|
||||
* </pre>
|
||||
*/
|
||||
@Test
|
||||
@Order(1)
|
||||
void testCatalogWatchWithEndpoints() {
|
||||
waitForLogStatement("stateGenerator is of type: KubernetesEndpointsCatalogWatch", K3S, APP_NAME);
|
||||
test();
|
||||
}
|
||||
|
||||
@Test
|
||||
@Order(2)
|
||||
void testCatalogWatchWithEndpointSlices() {
|
||||
KubernetesClientCatalogWatchUtils.patchForEndpointSlices(APP_NAME, NAMESPACE, DOCKER_IMAGE);
|
||||
waitForLogStatement("stateGenerator is of type: KubernetesEndpointSlicesCatalogWatch", K3S, APP_NAME);
|
||||
test();
|
||||
|
||||
testCatalogWatchWithEndpointsNamespaces();
|
||||
}
|
||||
|
||||
void testCatalogWatchWithEndpointsNamespaces() {
|
||||
util.createNamespace(NAMESPACE_A);
|
||||
util.createNamespace(NAMESPACE_B);
|
||||
util.setUpClusterWide(NAMESPACE, Set.of(NAMESPACE_A, NAMESPACE_B));
|
||||
util.busybox(NAMESPACE_A, Phase.CREATE);
|
||||
util.busybox(NAMESPACE_B, Phase.CREATE);
|
||||
|
||||
KubernetesClientCatalogWatchUtils.patchForEndpointsNamespaces(APP_NAME, NAMESPACE, DOCKER_IMAGE);
|
||||
KubernetesClientCatalogWatchNamespacesDelegate.testCatalogWatchWithEndpointsNamespaces(APP_NAME);
|
||||
|
||||
util.busybox(NAMESPACE_A, Phase.CREATE);
|
||||
util.busybox(NAMESPACE_B, Phase.CREATE);
|
||||
KubernetesClientCatalogWatchUtils.patchForEndpointSlicesNamespaces(APP_NAME, NAMESPACE, DOCKER_IMAGE);
|
||||
KubernetesClientCatalogWatchNamespacesDelegate.testCatalogWatchWithEndpointSlicesNamespaces(APP_NAME);
|
||||
}
|
||||
|
||||
/**
|
||||
* the test is the same for both endpoints and endpoint slices, the set-up for them is
|
||||
* different.
|
||||
*/
|
||||
private void test() {
|
||||
|
||||
WebClient client = builder().baseUrl("http://localhost/result").build();
|
||||
EndpointNameAndNamespace[] holder = new EndpointNameAndNamespace[2];
|
||||
ResolvableType resolvableType = ResolvableType.forClassWithGenerics(List.class, EndpointNameAndNamespace.class);
|
||||
|
||||
await().pollInterval(Duration.ofSeconds(1)).atMost(Duration.ofSeconds(240)).until(() -> {
|
||||
List<EndpointNameAndNamespace> result = (List<EndpointNameAndNamespace>) client.method(HttpMethod.GET)
|
||||
.retrieve()
|
||||
.bodyToMono(ParameterizedTypeReference.forType(resolvableType.getType()))
|
||||
.retryWhen(retrySpec())
|
||||
.block();
|
||||
|
||||
// we get 3 pods as input, but because they are sorted by name in the catalog
|
||||
// watcher implementation
|
||||
// we will get the first busybox instances here.
|
||||
|
||||
if (result != null) {
|
||||
if (result.size() != 3) {
|
||||
return false;
|
||||
}
|
||||
holder[0] = result.get(0);
|
||||
holder[1] = result.get(1);
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
});
|
||||
|
||||
EndpointNameAndNamespace resultOne = holder[0];
|
||||
EndpointNameAndNamespace resultTwo = holder[1];
|
||||
|
||||
Assertions.assertNotNull(resultOne);
|
||||
Assertions.assertNotNull(resultTwo);
|
||||
|
||||
Assertions.assertTrue(resultOne.endpointName().contains("busybox"));
|
||||
Assertions.assertTrue(resultTwo.endpointName().contains("busybox"));
|
||||
Assertions.assertEquals("default", resultOne.namespace());
|
||||
Assertions.assertEquals("default", resultTwo.namespace());
|
||||
|
||||
util.busybox(NAMESPACE, Phase.DELETE);
|
||||
|
||||
// what we get after delete
|
||||
EndpointNameAndNamespace[] afterDelete = new EndpointNameAndNamespace[1];
|
||||
|
||||
await().pollInterval(Duration.ofSeconds(1)).atMost(Duration.ofSeconds(240)).until(() -> {
|
||||
List<EndpointNameAndNamespace> result = (List<EndpointNameAndNamespace>) client.method(HttpMethod.GET)
|
||||
.retrieve()
|
||||
.bodyToMono(ParameterizedTypeReference.forType(resolvableType.getType()))
|
||||
.retryWhen(retrySpec())
|
||||
.block();
|
||||
|
||||
// we need to get the event from KubernetesCatalogWatch, but that happens
|
||||
// on periodic bases. So in order to be sure we got the event we care about
|
||||
// we wait until the result has a single entry, which means busybox was
|
||||
// deleted
|
||||
// + KubernetesCatalogWatch received the new update.
|
||||
if (result != null && result.size() != 1) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// we will only receive one pod here, our own
|
||||
if (result != null) {
|
||||
afterDelete[0] = result.get(0);
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
});
|
||||
|
||||
Assertions.assertTrue(afterDelete[0].endpointName().contains(APP_NAME));
|
||||
Assertions.assertEquals("default", afterDelete[0].namespace());
|
||||
|
||||
}
|
||||
|
||||
private static void app(Phase phase) {
|
||||
V1Deployment deployment = (V1Deployment) util.yaml("app/watcher-deployment.yaml");
|
||||
V1Service service = (V1Service) util.yaml("app/watcher-service.yaml");
|
||||
V1Ingress ingress = (V1Ingress) util.yaml("app/watcher-ingress.yaml");
|
||||
|
||||
if (phase.equals(Phase.CREATE)) {
|
||||
util.createAndWait(NAMESPACE, null, deployment, service, ingress, true);
|
||||
}
|
||||
else if (phase.equals(Phase.DELETE)) {
|
||||
util.deleteAndWait(NAMESPACE, deployment, service, ingress);
|
||||
}
|
||||
}
|
||||
|
||||
private WebClient.Builder builder() {
|
||||
return WebClient.builder().clientConnector(new ReactorClientHttpConnector(HttpClient.create()));
|
||||
}
|
||||
|
||||
private RetryBackoffSpec retrySpec() {
|
||||
return Retry.fixedDelay(15, Duration.ofSeconds(1)).filter(Objects::nonNull);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,153 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-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.k8s.client.catalog.watcher;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
import java.util.Comparator;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.testcontainers.k3s.K3sContainer;
|
||||
import reactor.netty.http.client.HttpClient;
|
||||
import reactor.util.retry.Retry;
|
||||
import reactor.util.retry.RetryBackoffSpec;
|
||||
|
||||
import org.springframework.cloud.kubernetes.commons.discovery.EndpointNameAndNamespace;
|
||||
import org.springframework.cloud.kubernetes.integration.tests.commons.Commons;
|
||||
import org.springframework.cloud.kubernetes.integration.tests.commons.Phase;
|
||||
import org.springframework.cloud.kubernetes.integration.tests.commons.native_client.Util;
|
||||
import org.springframework.core.ParameterizedTypeReference;
|
||||
import org.springframework.core.ResolvableType;
|
||||
import org.springframework.http.HttpMethod;
|
||||
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
|
||||
import org.springframework.web.reactive.function.client.WebClient;
|
||||
|
||||
import static org.awaitility.Awaitility.await;
|
||||
import static org.springframework.cloud.kubernetes.integration.tests.commons.Commons.waitForLogStatement;
|
||||
|
||||
final class KubernetesClientCatalogWatchNamespacesDelegate {
|
||||
|
||||
private KubernetesClientCatalogWatchNamespacesDelegate() {
|
||||
|
||||
}
|
||||
|
||||
private static final String NAMESPACE_A = "namespacea";
|
||||
|
||||
private static final String NAMESPACE_B = "namespaceb";
|
||||
|
||||
private static final K3sContainer K3S = Commons.container();
|
||||
|
||||
private static Util util;
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* - we deploy one busybox service with 2 replica pods in namespace namespacea
|
||||
* - we deploy one busybox service with 2 replica pods in namespace namespaceb
|
||||
* - we enable the search to be made in namespacea and default ones
|
||||
* - we receive an event from KubernetesCatalogWatcher, assert what is inside it
|
||||
* - delete both busybox services in namespacea and namespaceb
|
||||
* - assert that we receive only spring-cloud-kubernetes-client-catalog-watcher pod
|
||||
* </pre>
|
||||
*/
|
||||
static void testCatalogWatchWithEndpointsNamespaces(String deploymentName) {
|
||||
waitForLogStatement("stateGenerator is of type: KubernetesEndpointsCatalogWatch", K3S, deploymentName);
|
||||
testForNamespacesFilter();
|
||||
}
|
||||
|
||||
static void testCatalogWatchWithEndpointSlicesNamespaces(String deploymentName) {
|
||||
waitForLogStatement("stateGenerator is of type: KubernetesEndpointSlicesCatalogWatch", K3S, deploymentName);
|
||||
testForNamespacesFilter();
|
||||
}
|
||||
|
||||
/**
|
||||
* the test is the same for both endpoints and endpoint slices, the set-up for them is
|
||||
* different.
|
||||
*/
|
||||
private static void testForNamespacesFilter() {
|
||||
|
||||
WebClient client = builder().baseUrl("http://localhost/result").build();
|
||||
EndpointNameAndNamespace[] holder = new EndpointNameAndNamespace[4];
|
||||
ResolvableType resolvableType = ResolvableType.forClassWithGenerics(List.class, EndpointNameAndNamespace.class);
|
||||
|
||||
await().pollInterval(Duration.ofSeconds(1)).atMost(Duration.ofSeconds(240)).until(() -> {
|
||||
List<EndpointNameAndNamespace> result = (List<EndpointNameAndNamespace>) client.method(HttpMethod.GET)
|
||||
.retrieve()
|
||||
.bodyToMono(ParameterizedTypeReference.forType(resolvableType.getType()))
|
||||
.retryWhen(retrySpec())
|
||||
.block();
|
||||
|
||||
if (result != null) {
|
||||
// 2 from namespace-a, 2 from namespace-b
|
||||
Assertions.assertEquals(result.size(), 4);
|
||||
holder[0] = result.get(0);
|
||||
holder[1] = result.get(1);
|
||||
holder[2] = result.get(2);
|
||||
holder[3] = result.get(3);
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
});
|
||||
|
||||
EndpointNameAndNamespace resultOne = holder[0];
|
||||
EndpointNameAndNamespace resultTwo = holder[1];
|
||||
EndpointNameAndNamespace resultThree = holder[2];
|
||||
EndpointNameAndNamespace resultFour = holder[3];
|
||||
|
||||
Assertions.assertTrue(resultOne.endpointName().contains("busybox"));
|
||||
Assertions.assertTrue(resultTwo.endpointName().contains("busybox"));
|
||||
Assertions.assertTrue(resultThree.endpointName().contains("busybox"));
|
||||
Assertions.assertTrue(resultFour.endpointName().contains("busybox"));
|
||||
|
||||
List<EndpointNameAndNamespace> sorted = Arrays.stream(holder)
|
||||
.sorted(Comparator.comparing(EndpointNameAndNamespace::namespace))
|
||||
.toList();
|
||||
|
||||
Assertions.assertEquals(NAMESPACE_A, sorted.get(0).namespace());
|
||||
Assertions.assertEquals(NAMESPACE_A, sorted.get(1).namespace());
|
||||
Assertions.assertEquals(NAMESPACE_B, sorted.get(2).namespace());
|
||||
Assertions.assertEquals(NAMESPACE_B, sorted.get(3).namespace());
|
||||
|
||||
util = new Util(K3S);
|
||||
util.busybox(NAMESPACE_A, Phase.DELETE);
|
||||
util.busybox(NAMESPACE_B, Phase.DELETE);
|
||||
|
||||
await().pollInterval(Duration.ofSeconds(1)).atMost(Duration.ofSeconds(240)).until(() -> {
|
||||
List<EndpointNameAndNamespace> result = (List<EndpointNameAndNamespace>) client.method(HttpMethod.GET)
|
||||
.retrieve()
|
||||
.bodyToMono(ParameterizedTypeReference.forType(resolvableType.getType()))
|
||||
.retryWhen(retrySpec())
|
||||
.block();
|
||||
|
||||
// there is no update to receive anymore, as there is nothing in namespacea
|
||||
// and namespaceb
|
||||
return result.size() == 0;
|
||||
});
|
||||
|
||||
}
|
||||
|
||||
private static WebClient.Builder builder() {
|
||||
return WebClient.builder().clientConnector(new ReactorClientHttpConnector(HttpClient.create()));
|
||||
}
|
||||
|
||||
private static RetryBackoffSpec retrySpec() {
|
||||
return Retry.fixedDelay(15, Duration.ofSeconds(1)).filter(Objects::nonNull);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,138 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-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.k8s.client.catalog.watcher;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
import static org.springframework.cloud.kubernetes.integration.tests.commons.native_client.Util.patchWithReplace;
|
||||
|
||||
/**
|
||||
* @author wind57
|
||||
*/
|
||||
final class KubernetesClientCatalogWatchUtils {
|
||||
|
||||
private static final Map<String, String> POD_LABELS = Map.of("app",
|
||||
"spring-cloud-kubernetes-k8s-client-catalog-watcher");
|
||||
|
||||
private KubernetesClientCatalogWatchUtils() {
|
||||
|
||||
}
|
||||
|
||||
private static final String BODY_ONE = """
|
||||
{
|
||||
"spec": {
|
||||
"template": {
|
||||
"spec": {
|
||||
"containers": [{
|
||||
"name": "spring-cloud-kubernetes-k8s-client-catalog-watcher",
|
||||
"image": "image_name_here",
|
||||
"env": [
|
||||
{
|
||||
"name": "SPRING_CLOUD_KUBERNETES_DISCOVERY_USE_ENDPOINT_SLICES",
|
||||
"value": "TRUE"
|
||||
},
|
||||
{
|
||||
"name": "LOGGING_LEVEL_ORG_SPRINGFRAMEWORK_CLOUD_KUBERNETES_CLIENT_DISCOVERY_CATALOG",
|
||||
"value": "DEBUG"
|
||||
}
|
||||
]
|
||||
}]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
""";
|
||||
|
||||
private static final String BODY_TWO = """
|
||||
{
|
||||
"spec": {
|
||||
"template": {
|
||||
"spec": {
|
||||
"containers": [{
|
||||
"name": "spring-cloud-kubernetes-k8s-client-catalog-watcher",
|
||||
"image": "image_name_here",
|
||||
"env": [
|
||||
{
|
||||
"name": "SPRING_CLOUD_KUBERNETES_DISCOVERY_USE_ENDPOINT_SLICES",
|
||||
"value": "FALSE"
|
||||
},
|
||||
{
|
||||
"name": "LOGGING_LEVEL_ORG_SPRINGFRAMEWORK_CLOUD_KUBERNETES_CLIENT_DISCOVERY_CATALOG",
|
||||
"value": "DEBUG"
|
||||
},
|
||||
{
|
||||
"name": "SPRING_CLOUD_KUBERNETES_DISCOVERY_NAMESPACES_0",
|
||||
"value": "namespacea"
|
||||
},
|
||||
{
|
||||
"name": "SPRING_CLOUD_KUBERNETES_DISCOVERY_NAMESPACES_1",
|
||||
"value": "namespaceb"
|
||||
}
|
||||
]
|
||||
}]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
""";
|
||||
|
||||
private static final String BODY_THREE = """
|
||||
{
|
||||
"spec": {
|
||||
"template": {
|
||||
"spec": {
|
||||
"containers": [{
|
||||
"name": "spring-cloud-kubernetes-k8s-client-catalog-watcher",
|
||||
"image": "image_name_here",
|
||||
"env": [
|
||||
{
|
||||
"name": "SPRING_CLOUD_KUBERNETES_DISCOVERY_USE_ENDPOINT_SLICES",
|
||||
"value": "TRUE"
|
||||
},
|
||||
{
|
||||
"name": "LOGGING_LEVEL_ORG_SPRINGFRAMEWORK_CLOUD_KUBERNETES_CLIENT_DISCOVERY_CATALOG",
|
||||
"value": "DEBUG"
|
||||
},
|
||||
{
|
||||
"name": "SPRING_CLOUD_KUBERNETES_DISCOVERY_NAMESPACES_0",
|
||||
"value": "namespacea"
|
||||
},
|
||||
{
|
||||
"name": "SPRING_CLOUD_KUBERNETES_DISCOVERY_NAMESPACES_1",
|
||||
"value": "namespaceb"
|
||||
}
|
||||
]
|
||||
}]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
""";
|
||||
|
||||
static void patchForEndpointSlices(String deploymentName, String namespace, String imageName) {
|
||||
patchWithReplace(imageName, deploymentName, namespace, BODY_ONE, POD_LABELS);
|
||||
}
|
||||
|
||||
static void patchForEndpointsNamespaces(String deploymentName, String namespace, String imageName) {
|
||||
patchWithReplace(imageName, deploymentName, namespace, BODY_TWO, POD_LABELS);
|
||||
}
|
||||
|
||||
static void patchForEndpointSlicesNamespaces(String deploymentName, String namespace, String imageName) {
|
||||
patchWithReplace(imageName, deploymentName, namespace, BODY_THREE, POD_LABELS);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
/*
|
||||
* 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.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
|
||||
import org.assertj.core.api.Assertions;
|
||||
import org.awaitility.Awaitility;
|
||||
|
||||
import org.springframework.boot.test.system.CapturedOutput;
|
||||
import org.springframework.cloud.kubernetes.commons.discovery.EndpointNameAndNamespace;
|
||||
import org.springframework.cloud.kubernetes.integration.tests.commons.Phase;
|
||||
import org.springframework.cloud.kubernetes.integration.tests.commons.fabric8_client.Util;
|
||||
import org.springframework.core.ParameterizedTypeReference;
|
||||
import org.springframework.core.ResolvableType;
|
||||
import org.springframework.http.HttpMethod;
|
||||
import org.springframework.web.reactive.function.client.WebClient;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
import static org.springframework.cloud.kubernetes.integration.tests.commons.Commons.builder;
|
||||
import static org.springframework.cloud.kubernetes.integration.tests.commons.Commons.retrySpec;
|
||||
|
||||
/**
|
||||
* @author wind57
|
||||
*/
|
||||
final class TestAssertions {
|
||||
|
||||
private TestAssertions() {
|
||||
|
||||
}
|
||||
|
||||
static void assertLogStatement(CapturedOutput output, String textToAssert) {
|
||||
Awaitility.await()
|
||||
.during(Duration.ofSeconds(5))
|
||||
.pollInterval(Duration.ofMillis(200))
|
||||
.untilAsserted(() -> Assertions.assertThat(output.getOut()).contains(textToAssert));
|
||||
}
|
||||
|
||||
/**
|
||||
* the checks are the same for both endpoints and endpoint slices, while the set-up
|
||||
* for them is different.
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
static void invokeAndAssert(Util util, Set<String> namespaces, int port, String assertionNamespace) {
|
||||
|
||||
WebClient client = builder().baseUrl("http://localhost:" + port + "/result").build();
|
||||
EndpointNameAndNamespace[] holder = new EndpointNameAndNamespace[2];
|
||||
ResolvableType resolvableType = ResolvableType.forClassWithGenerics(List.class, EndpointNameAndNamespace.class);
|
||||
|
||||
await().pollInterval(Duration.ofMillis(200)).atMost(Duration.ofSeconds(30)).until(() -> {
|
||||
List<EndpointNameAndNamespace> result = (List<EndpointNameAndNamespace>) client.method(HttpMethod.GET)
|
||||
.retrieve()
|
||||
.bodyToMono(ParameterizedTypeReference.forType(resolvableType.getType()))
|
||||
.retryWhen(retrySpec())
|
||||
.block();
|
||||
|
||||
if (result != null) {
|
||||
if (result.size() != 2) {
|
||||
return false;
|
||||
}
|
||||
holder[0] = result.get(0);
|
||||
holder[1] = result.get(1);
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
});
|
||||
|
||||
EndpointNameAndNamespace resultOne = holder[0];
|
||||
EndpointNameAndNamespace resultTwo = holder[1];
|
||||
|
||||
assertThat(resultOne).isNotNull();
|
||||
assertThat(resultTwo).isNotNull();
|
||||
|
||||
assertThat(resultOne.endpointName()).contains("busybox");
|
||||
assertThat(resultTwo.endpointName()).contains("busybox");
|
||||
|
||||
assertThat(resultOne.namespace()).isEqualTo(assertionNamespace);
|
||||
assertThat(resultTwo.namespace()).isEqualTo(assertionNamespace);
|
||||
|
||||
namespaces.forEach(namespace -> util.busybox(namespace, Phase.DELETE));
|
||||
|
||||
await().pollInterval(Duration.ofSeconds(1)).atMost(Duration.ofSeconds(240)).until(() -> {
|
||||
List<EndpointNameAndNamespace> result = (List<EndpointNameAndNamespace>) client.method(HttpMethod.GET)
|
||||
.retrieve()
|
||||
.bodyToMono(ParameterizedTypeReference.forType(resolvableType.getType()))
|
||||
.retryWhen(retrySpec())
|
||||
.block();
|
||||
|
||||
// we need to get the event from KubernetesCatalogWatch, but that happens
|
||||
// on periodic bases. So in order to be sure we got the event we care about
|
||||
// we wait until there is no entry, which means busybox was deleted
|
||||
// and KubernetesCatalogWatch received that update.
|
||||
return Objects.requireNonNull(result).isEmpty();
|
||||
});
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user