From 0bda86476dc07e5501ee9ed342cc240acb390f5d Mon Sep 17 00:00:00 2001 From: erabii Date: Tue, 30 Aug 2022 21:09:17 +0300 Subject: [PATCH] Add possibility to define the label for watcher 3 (#1063) * fix * simplify labels * fix * fix test * trigger * more simplifications * very dirty still * more changes * more changes * fix tests * fix test * trigger --- ...ientEventBasedConfigMapChangeDetector.java | 3 +- ...ClientEventBasedSecretsChangeDetector.java | 3 +- .../reload/ConfigurationUpdateStrategy.java | 2 +- ...ntBasedConfigMapWatcherChangeDetector.java | 27 ++-- ...ventBasedSecretsWatcherChangeDetector.java | 27 ++-- .../watcher/BusRefreshTrigger.java | 49 ++++++++ .../ConfigMapWatcherChangeDetector.java | 54 ++------ ...ConfigurationWatcherAutoConfiguration.java | 94 ++++++++------ ...urationWatcherConfigurationProperties.java | 5 + ...HttpBasedConfigMapWatchChangeDetector.java | 101 ++------------- .../HttpBasedSecretsWatchChangeDetector.java | 99 ++------------- .../watcher/HttpRefreshTrigger.java | 117 ++++++++++++++++++ .../configuration/watcher/RefreshTrigger.java | 34 +++++ .../watcher/SecretsWatcherChangeDetector.java | 53 ++------ .../configuration/watcher/WatcherUtil.java | 76 ++++++++++++ ...edConfigMapWatcherChangeDetectorTests.java | 4 +- ...asedSecretsWatcherChangeDetectorTests.java | 4 +- ...asedConfigMapWatchChangeDetectorTests.java | 7 +- ...pBasedSecretsWatchChangeDetectorTests.java | 6 +- 19 files changed, 410 insertions(+), 355 deletions(-) create mode 100644 spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusRefreshTrigger.java create mode 100644 spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpRefreshTrigger.java create mode 100644 spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/RefreshTrigger.java create mode 100644 spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/WatcherUtil.java diff --git a/spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedConfigMapChangeDetector.java b/spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedConfigMapChangeDetector.java index 5c3e2b3c..f25cf6a2 100644 --- a/spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedConfigMapChangeDetector.java +++ b/spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedConfigMapChangeDetector.java @@ -20,6 +20,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Set; +import io.kubernetes.client.common.KubernetesObject; import io.kubernetes.client.informer.ResourceEventHandler; import io.kubernetes.client.informer.SharedIndexInformer; import io.kubernetes.client.informer.SharedInformerFactory; @@ -145,7 +146,7 @@ public class KubernetesClientEventBasedConfigMapChangeDetector extends Configura factory.stopAllRegisteredInformers(); } - protected void onEvent(V1ConfigMap configMap) { + protected void onEvent(KubernetesObject configMap) { boolean reload = ConfigReloadUtil.reload("config-map", configMap.toString(), propertySourceLocator, environment, KubernetesClientConfigMapPropertySource.class); if (reload) { diff --git a/spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedSecretsChangeDetector.java b/spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedSecretsChangeDetector.java index 13af6d3a..bfa69d5b 100644 --- a/spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedSecretsChangeDetector.java +++ b/spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedSecretsChangeDetector.java @@ -20,6 +20,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Set; +import io.kubernetes.client.common.KubernetesObject; import io.kubernetes.client.informer.ResourceEventHandler; import io.kubernetes.client.informer.SharedIndexInformer; import io.kubernetes.client.informer.SharedInformerFactory; @@ -144,7 +145,7 @@ public class KubernetesClientEventBasedSecretsChangeDetector extends Configurati factory.stopAllRegisteredInformers(); } - protected void onEvent(V1Secret secret) { + protected void onEvent(KubernetesObject secret) { boolean reload = ConfigReloadUtil.reload("secrets", secret.toString(), propertySourceLocator, environment, KubernetesClientSecretsPropertySource.class); if (reload) { diff --git a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/config/reload/ConfigurationUpdateStrategy.java b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/config/reload/ConfigurationUpdateStrategy.java index bc7323de..459ea2e6 100644 --- a/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/config/reload/ConfigurationUpdateStrategy.java +++ b/spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/config/reload/ConfigurationUpdateStrategy.java @@ -24,7 +24,7 @@ import java.util.Objects; * * @author Nicola Ferraro */ -public final record ConfigurationUpdateStrategy(String name, Runnable reloadProcedure) { +public record ConfigurationUpdateStrategy(String name, Runnable reloadProcedure) { public ConfigurationUpdateStrategy(String name, Runnable reloadProcedure) { this.name = Objects.requireNonNull(name, "name cannot be null"); diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetector.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetector.java index b80f8e73..0b786baa 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetector.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetector.java @@ -16,18 +16,14 @@ package org.springframework.cloud.kubernetes.configuration.watcher; +import io.kubernetes.client.common.KubernetesObject; import io.kubernetes.client.openapi.apis.CoreV1Api; -import io.kubernetes.client.openapi.models.V1ConfigMap; import reactor.core.publisher.Mono; -import org.springframework.cloud.bus.BusProperties; -import org.springframework.cloud.bus.event.PathDestinationFactory; -import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent; import org.springframework.cloud.kubernetes.client.config.KubernetesClientConfigMapPropertySourceLocator; import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider; import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties; import org.springframework.cloud.kubernetes.commons.config.reload.ConfigurationUpdateStrategy; -import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @@ -35,29 +31,24 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; * @author Ryan Baxter * @author Kris Iyer */ -public class BusEventBasedConfigMapWatcherChangeDetector extends ConfigMapWatcherChangeDetector { +final class BusEventBasedConfigMapWatcherChangeDetector extends ConfigMapWatcherChangeDetector { - private final ApplicationEventPublisher applicationEventPublisher; + private final BusRefreshTrigger busRefreshTrigger; - private final BusProperties busProperties; - - public BusEventBasedConfigMapWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, + BusEventBasedConfigMapWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, KubernetesClientConfigMapPropertySourceLocator propertySourceLocator, - KubernetesNamespaceProvider kubernetesNamespaceProvider, BusProperties busProperties, + KubernetesNamespaceProvider kubernetesNamespaceProvider, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, - ThreadPoolTaskExecutor threadPoolTaskExecutor, ApplicationEventPublisher applicationEventPublisher) { + ThreadPoolTaskExecutor threadPoolTaskExecutor, BusRefreshTrigger busRefreshTrigger) { super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider, k8SConfigurationProperties, threadPoolTaskExecutor); - this.busProperties = busProperties; - this.applicationEventPublisher = applicationEventPublisher; + this.busRefreshTrigger = busRefreshTrigger; } @Override - protected Mono triggerRefresh(V1ConfigMap configMap) { - this.applicationEventPublisher.publishEvent(new RefreshRemoteApplicationEvent(configMap, busProperties.getId(), - new PathDestinationFactory().getDestination(configMap.getMetadata().getName()))); - return Mono.empty(); + public Mono triggerRefresh(KubernetesObject configMap) { + return busRefreshTrigger.triggerRefresh(configMap); } } diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetector.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetector.java index e4d9a3f8..b4d07097 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetector.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetector.java @@ -16,18 +16,14 @@ package org.springframework.cloud.kubernetes.configuration.watcher; +import io.kubernetes.client.common.KubernetesObject; import io.kubernetes.client.openapi.apis.CoreV1Api; -import io.kubernetes.client.openapi.models.V1Secret; import reactor.core.publisher.Mono; -import org.springframework.cloud.bus.BusProperties; -import org.springframework.cloud.bus.event.PathDestinationFactory; -import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent; import org.springframework.cloud.kubernetes.client.config.KubernetesClientSecretsPropertySourceLocator; import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider; import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties; import org.springframework.cloud.kubernetes.commons.config.reload.ConfigurationUpdateStrategy; -import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @@ -35,29 +31,24 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; * @author Ryan Baxter * @author Kris Iyer */ -public class BusEventBasedSecretsWatcherChangeDetector extends SecretsWatcherChangeDetector { +final class BusEventBasedSecretsWatcherChangeDetector extends SecretsWatcherChangeDetector { - private final ApplicationEventPublisher applicationEventPublisher; + private final BusRefreshTrigger busRefreshTrigger; - private final BusProperties busProperties; - - public BusEventBasedSecretsWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, + BusEventBasedSecretsWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, KubernetesClientSecretsPropertySourceLocator propertySourceLocator, - KubernetesNamespaceProvider kubernetesNamespaceProvider, BusProperties busProperties, + KubernetesNamespaceProvider kubernetesNamespaceProvider, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, - ThreadPoolTaskExecutor threadPoolTaskExecutor, ApplicationEventPublisher applicationEventPublisher) { + ThreadPoolTaskExecutor threadPoolTaskExecutor, BusRefreshTrigger busRefreshTrigger) { super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider, k8SConfigurationProperties, threadPoolTaskExecutor); - this.busProperties = busProperties; - this.applicationEventPublisher = applicationEventPublisher; + this.busRefreshTrigger = busRefreshTrigger; } @Override - protected Mono triggerRefresh(V1Secret secret) { - this.applicationEventPublisher.publishEvent(new RefreshRemoteApplicationEvent(secret, busProperties.getId(), - new PathDestinationFactory().getDestination(secret.getMetadata().getName()))); - return Mono.empty(); + public Mono triggerRefresh(KubernetesObject secret) { + return busRefreshTrigger.triggerRefresh(secret); } } diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusRefreshTrigger.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusRefreshTrigger.java new file mode 100644 index 00000000..da3e6d94 --- /dev/null +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusRefreshTrigger.java @@ -0,0 +1,49 @@ +/* + * 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.configuration.watcher; + +import io.kubernetes.client.common.KubernetesObject; +import reactor.core.publisher.Mono; + +import org.springframework.cloud.bus.event.PathDestinationFactory; +import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; + +/** + * An event publisher for an 'event bus' type of application. + * + * @author wind57 + */ +final class BusRefreshTrigger implements RefreshTrigger { + + private final ApplicationEventPublisher applicationEventPublisher; + + private final String busId; + + BusRefreshTrigger(ApplicationEventPublisher applicationEventPublisher, String busId) { + this.applicationEventPublisher = applicationEventPublisher; + this.busId = busId; + } + + @Override + public Mono triggerRefresh(KubernetesObject configMap) { + applicationEventPublisher.publishEvent(new RefreshRemoteApplicationEvent(configMap, busId, + new PathDestinationFactory().getDestination(configMap.getMetadata().getName()))); + return Mono.empty(); + } + +} diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigMapWatcherChangeDetector.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigMapWatcherChangeDetector.java index 2e089b4f..769ca0fb 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigMapWatcherChangeDetector.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigMapWatcherChangeDetector.java @@ -18,13 +18,9 @@ package org.springframework.cloud.kubernetes.configuration.watcher; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; +import io.kubernetes.client.common.KubernetesObject; import io.kubernetes.client.openapi.apis.CoreV1Api; -import io.kubernetes.client.openapi.models.V1ConfigMap; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import reactor.core.publisher.Mono; import org.springframework.cloud.kubernetes.client.config.KubernetesClientConfigMapPropertySourceLocator; import org.springframework.cloud.kubernetes.client.config.reload.KubernetesClientEventBasedConfigMapChangeDetector; @@ -38,60 +34,30 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; * @author Ryan Baxter * @author Kris Iyer */ -public abstract class ConfigMapWatcherChangeDetector extends KubernetesClientEventBasedConfigMapChangeDetector { - - protected Log log = LogFactory.getLog(getClass()); +abstract class ConfigMapWatcherChangeDetector extends KubernetesClientEventBasedConfigMapChangeDetector + implements RefreshTrigger { private final ScheduledExecutorService executorService; - protected ConfigurationWatcherConfigurationProperties k8SConfigurationProperties; + private final long refreshDelay; - public ConfigMapWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, + ConfigMapWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, KubernetesClientConfigMapPropertySourceLocator propertySourceLocator, KubernetesNamespaceProvider kubernetesNamespaceProvider, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, ThreadPoolTaskExecutor threadPoolTaskExecutor) { super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider); - this.executorService = Executors.newScheduledThreadPool(k8SConfigurationProperties.getThreadPoolSize(), threadPoolTaskExecutor); - this.k8SConfigurationProperties = k8SConfigurationProperties; + this.refreshDelay = k8SConfigurationProperties.getRefreshDelay().toMillis(); } @Override - protected void onEvent(V1ConfigMap configMap) { - if (isSpringCloudKubernetesConfig(configMap)) { - if (log.isDebugEnabled()) { - log.debug("Scheduling remote refresh event to be published for ConfigMap " - + configMap.getMetadata().getName() + " to be published in " - + k8SConfigurationProperties.getRefreshDelay().toMillis() + " milliseconds"); - } - executorService.schedule(() -> { - try { - triggerRefresh(configMap).subscribe(); - } - catch (Throwable t) { - log.warn("Error when refreshing ConfigMap " + configMap.getMetadata().getName(), t); - } - }, k8SConfigurationProperties.getRefreshDelay().toMillis(), TimeUnit.MILLISECONDS); - } - else { - if (log.isDebugEnabled()) { - log.debug("Not publishing event. ConfigMap " + configMap.getMetadata().getName() - + " does not contain the label " + ConfigurationWatcherConfigurationProperties.CONFIG_LABEL); - } - } - } - - protected abstract Mono triggerRefresh(V1ConfigMap configMap); - - private boolean isSpringCloudKubernetesConfig(V1ConfigMap configMap) { - if (configMap.getMetadata() == null || configMap.getMetadata().getLabels() == null) { - return false; - } - return Boolean.parseBoolean(configMap.getMetadata().getLabels() - .getOrDefault(ConfigurationWatcherConfigurationProperties.CONFIG_LABEL, "false")); + protected final void onEvent(KubernetesObject configMap) { + // this::refreshTrigger is coming from BusEventBasedConfigMapWatcherChangeDetector + WatcherUtil.onEvent(configMap, ConfigurationWatcherConfigurationProperties.CONFIG_LABEL, refreshDelay, + executorService, "config-map", this::triggerRefresh); } } diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherAutoConfiguration.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherAutoConfiguration.java index a1d73cdf..a3c6dd61 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherAutoConfiguration.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherAutoConfiguration.java @@ -19,6 +19,7 @@ package org.springframework.cloud.kubernetes.configuration.watcher; import io.kubernetes.client.openapi.apis.CoreV1Api; import org.springframework.boot.actuate.autoconfigure.amqp.RabbitHealthContributorAutoConfiguration; +import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.context.properties.EnableConfigurationProperties; @@ -45,8 +46,13 @@ import org.springframework.web.reactive.function.client.WebClient; */ @Configuration(proxyBeanMethods = false) @EnableConfigurationProperties({ ConfigurationWatcherConfigurationProperties.class }) +@Import({ ConfigurationWatcherAutoConfiguration.RefreshTriggerConfiguration.class }) public class ConfigurationWatcherAutoConfiguration { + private static final String AMQP = "bus-amqp"; + + private static final String KAFKA = "bus-kafka"; + @Bean @ConditionalOnMissingBean public WebClient webClient(WebClient.Builder webClientBuilder) { @@ -60,11 +66,11 @@ public class ConfigurationWatcherAutoConfiguration { CoreV1Api coreV1Api, KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, - KubernetesNamespaceProvider namespaceProvider, ThreadPoolTaskExecutor threadFactory, WebClient webClient, - KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient) { + KubernetesNamespaceProvider namespaceProvider, ThreadPoolTaskExecutor threadFactory, + HttpRefreshTrigger httpRefreshTrigger) { return new HttpBasedConfigMapWatchChangeDetector(coreV1Api, environment, properties, strategy, - configMapPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory, webClient, - kubernetesReactiveDiscoveryClient); + configMapPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory, + httpRefreshTrigger); } @Bean @@ -75,83 +81,99 @@ public class ConfigurationWatcherAutoConfiguration { KubernetesNamespaceProvider namespaceProvider, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, - ThreadPoolTaskExecutor threadFactory, WebClient webClient, - KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient) { + ThreadPoolTaskExecutor threadFactory, HttpRefreshTrigger httpRefreshTrigger) { return new HttpBasedSecretsWatchChangeDetector(coreV1Api, environment, properties, strategy, - secretsPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory, webClient, - kubernetesReactiveDiscoveryClient); + secretsPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory, + httpRefreshTrigger); } @Configuration - @Profile("bus-amqp") - @Import({ ContextFunctionCatalogAutoConfiguration.class, RabbitHealthContributorAutoConfiguration.class }) + @Profile(AMQP) + @Import({ ContextFunctionCatalogAutoConfiguration.class, RabbitHealthContributorAutoConfiguration.class, + RefreshTriggerConfiguration.class }) static class BusRabbitConfiguration { @Bean @ConditionalOnMissingBean(ConfigMapWatcherChangeDetector.class) @ConditionalOnBean(KubernetesClientConfigMapPropertySourceLocator.class) - public ConfigMapWatcherChangeDetector busConfigMapChangeWatcher(BusProperties busProperties, - AbstractEnvironment environment, CoreV1Api coreV1Api, - KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator, + public ConfigMapWatcherChangeDetector busConfigMapChangeWatcher(AbstractEnvironment environment, + CoreV1Api coreV1Api, KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator, KubernetesNamespaceProvider kubernetesNamespaceProvider, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, - ThreadPoolTaskExecutor threadFactory, ApplicationEventPublisher applicationEventPublisher) { + ThreadPoolTaskExecutor threadFactory, BusRefreshTrigger busRefreshTrigger) { return new BusEventBasedConfigMapWatcherChangeDetector(coreV1Api, environment, properties, strategy, - configMapPropertySourceLocator, kubernetesNamespaceProvider, busProperties, - k8SConfigurationProperties, threadFactory, applicationEventPublisher); + configMapPropertySourceLocator, kubernetesNamespaceProvider, k8SConfigurationProperties, + threadFactory, busRefreshTrigger); } @Bean @ConditionalOnMissingBean(SecretsWatcherChangeDetector.class) @ConditionalOnBean(KubernetesClientSecretsPropertySourceLocator.class) - public SecretsWatcherChangeDetector busSecretsChangeWatcher(BusProperties busProperties, - AbstractEnvironment environment, CoreV1Api coreV1Api, - KubernetesClientSecretsPropertySourceLocator secretsPropertySourceLocator, + public SecretsWatcherChangeDetector busSecretsChangeWatcher(AbstractEnvironment environment, + CoreV1Api coreV1Api, KubernetesClientSecretsPropertySourceLocator secretsPropertySourceLocator, ConfigReloadProperties properties, KubernetesNamespaceProvider kubernetesNamespaceProvider, ConfigurationUpdateStrategy strategy, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, - ThreadPoolTaskExecutor threadFactory, ApplicationEventPublisher applicationEventPublisher) { + ThreadPoolTaskExecutor threadFactory, BusRefreshTrigger busRefreshTrigger) { return new BusEventBasedSecretsWatcherChangeDetector(coreV1Api, environment, properties, strategy, - secretsPropertySourceLocator, kubernetesNamespaceProvider, busProperties, - k8SConfigurationProperties, threadFactory, applicationEventPublisher); + secretsPropertySourceLocator, kubernetesNamespaceProvider, k8SConfigurationProperties, + threadFactory, busRefreshTrigger); } } @Configuration - @Profile("bus-kafka") - @Import({ ContextFunctionCatalogAutoConfiguration.class }) + @Profile(KAFKA) + @Import({ ContextFunctionCatalogAutoConfiguration.class, RefreshTriggerConfiguration.class }) static class BusKafkaConfiguration { @Bean @ConditionalOnMissingBean(ConfigMapWatcherChangeDetector.class) @ConditionalOnBean(KubernetesClientConfigMapPropertySourceLocator.class) - public ConfigMapWatcherChangeDetector busConfigMapChangeWatcher(BusProperties busProperties, - AbstractEnvironment environment, CoreV1Api coreV1Api, - KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator, + public ConfigMapWatcherChangeDetector busConfigMapChangeWatcher(AbstractEnvironment environment, + CoreV1Api coreV1Api, KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator, ConfigReloadProperties properties, KubernetesNamespaceProvider namespaceProvider, ConfigurationUpdateStrategy strategy, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, - ThreadPoolTaskExecutor threadFactory, ApplicationEventPublisher applicationEventPublisher) { + ThreadPoolTaskExecutor threadFactory, BusRefreshTrigger busRefreshTrigger) { return new BusEventBasedConfigMapWatcherChangeDetector(coreV1Api, environment, properties, strategy, - configMapPropertySourceLocator, namespaceProvider, busProperties, k8SConfigurationProperties, - threadFactory, applicationEventPublisher); + configMapPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory, + busRefreshTrigger); } @Bean @ConditionalOnMissingBean(SecretsWatcherChangeDetector.class) @ConditionalOnBean(KubernetesClientSecretsPropertySourceLocator.class) - public SecretsWatcherChangeDetector busSecretsChangeWatcher(BusProperties busProperties, - AbstractEnvironment environment, CoreV1Api coreV1Api, - KubernetesClientSecretsPropertySourceLocator secretsPropertySourceLocator, + public SecretsWatcherChangeDetector busSecretsChangeWatcher(AbstractEnvironment environment, + CoreV1Api coreV1Api, KubernetesClientSecretsPropertySourceLocator secretsPropertySourceLocator, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, ThreadPoolTaskExecutor threadFactory, KubernetesNamespaceProvider namespaceProvider, - ApplicationEventPublisher applicationEventPublisher) { + BusRefreshTrigger busRefreshTrigger) { return new BusEventBasedSecretsWatcherChangeDetector(coreV1Api, environment, properties, strategy, - secretsPropertySourceLocator, namespaceProvider, busProperties, k8SConfigurationProperties, - threadFactory, applicationEventPublisher); + secretsPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory, + busRefreshTrigger); + } + + } + + @AutoConfiguration + static class RefreshTriggerConfiguration { + + @Bean + @ConditionalOnMissingBean + @Profile({ AMQP, KAFKA }) + public BusRefreshTrigger busRefreshTrigger(ApplicationEventPublisher applicationEventPublisher, + BusProperties busProperties) { + return new BusRefreshTrigger(applicationEventPublisher, busProperties.getId()); + } + + @Bean + @ConditionalOnMissingBean + public HttpRefreshTrigger httpRefreshTrigger(KubernetesInformerReactiveDiscoveryClient client, + ConfigurationWatcherConfigurationProperties properties, WebClient webClient) { + return new HttpRefreshTrigger(client, properties, webClient); } } diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherConfigurationProperties.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherConfigurationProperties.java index c8cdea8a..a25cee16 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherConfigurationProperties.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherConfigurationProperties.java @@ -38,6 +38,11 @@ public class ConfigurationWatcherConfigurationProperties { */ public static final String SECRET_LABEL = "spring.cloud.kubernetes.secret"; + /** + * Annotation key for actuator port and path. + */ + public static final String ANNOTATION_KEY = "boot.spring.io/actuator"; + /** * Amount of time to delay the posting of the event to allow the app volume to update * data. diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetector.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetector.java index 13962d99..331b7614 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetector.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetector.java @@ -16,125 +16,40 @@ package org.springframework.cloud.kubernetes.configuration.watcher; -import java.net.URI; - +import io.kubernetes.client.common.KubernetesObject; import io.kubernetes.client.openapi.apis.CoreV1Api; -import io.kubernetes.client.openapi.models.V1ConfigMap; -import io.kubernetes.client.openapi.models.V1ObjectMeta; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import org.springframework.cloud.client.ServiceInstance; import org.springframework.cloud.kubernetes.client.config.KubernetesClientConfigMapPropertySourceLocator; -import org.springframework.cloud.kubernetes.client.discovery.reactive.KubernetesInformerReactiveDiscoveryClient; import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider; import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties; import org.springframework.cloud.kubernetes.commons.config.reload.ConfigurationUpdateStrategy; import org.springframework.core.env.ConfigurableEnvironment; -import org.springframework.http.ResponseEntity; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; -import org.springframework.util.StringUtils; -import org.springframework.web.reactive.function.client.WebClient; -import org.springframework.web.util.UriComponentsBuilder; /** * @author Ryan Baxter * @author Kris Iyer */ -public class HttpBasedConfigMapWatchChangeDetector extends ConfigMapWatcherChangeDetector { +final class HttpBasedConfigMapWatchChangeDetector extends ConfigMapWatcherChangeDetector { - private Log log = LogFactory.getLog(getClass()); + private final HttpRefreshTrigger httpRefreshTrigger; - /** - * Annotation key for actuator port and path. - */ - public static String ANNOTATION_KEY = "boot.spring.io/actuator"; - - private WebClient webClient; - - private KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient; - - public HttpBasedConfigMapWatchChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, + HttpBasedConfigMapWatchChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, KubernetesClientConfigMapPropertySourceLocator propertySourceLocator, KubernetesNamespaceProvider kubernetesNamespaceProvider, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, - ThreadPoolTaskExecutor threadPoolTaskExecutor, WebClient webClient, - KubernetesInformerReactiveDiscoveryClient k8sReactiveDiscoveryClient) { + ThreadPoolTaskExecutor threadPoolTaskExecutor, HttpRefreshTrigger httpRefreshTrigger) { super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider, k8SConfigurationProperties, threadPoolTaskExecutor); - this.webClient = webClient; - this.kubernetesReactiveDiscoveryClient = k8sReactiveDiscoveryClient; - } - private void setActuatorUriFromAnnotation(UriComponentsBuilder actuatorUriBuilder, String metadataUri) { - URI annotationUri = URI.create(metadataUri); - actuatorUriBuilder.path(annotationUri.getPath() + "/refresh"); - - // The URI may not contain a host so if that is the case the port in the URI will - // be -1 - // The authority of the URI will be : for example :9090, we just need the - // 9090 in this case - if (annotationUri.getPort() < 0) { - if (annotationUri.getAuthority() != null) { - actuatorUriBuilder.port(annotationUri.getAuthority().replaceFirst(":", "")); - } - } - else { - actuatorUriBuilder.port(annotationUri.getPort()); - } - } - - private URI getActuatorUri(ServiceInstance si) { - String metadataUri = si.getMetadata().getOrDefault(ANNOTATION_KEY, ""); - if (log.isDebugEnabled()) { - log.debug("Metadata actuator uri is: " + metadataUri); - } - - UriComponentsBuilder actuatorUriBuilder = UriComponentsBuilder.newInstance().scheme(si.getScheme()) - .host(si.getHost()); - - if (!StringUtils.isEmpty(metadataUri)) { - if (log.isDebugEnabled()) { - log.debug("Found actuator URI in service instance metadata"); - } - setActuatorUriFromAnnotation(actuatorUriBuilder, metadataUri); - } - else { - Integer port = k8SConfigurationProperties.getActuatorPort() < 0 ? si.getPort() - : k8SConfigurationProperties.getActuatorPort(); - actuatorUriBuilder = actuatorUriBuilder.path(k8SConfigurationProperties.getActuatorPath() + "/refresh") - .port(port); - } - - return actuatorUriBuilder.build().toUri(); - } - - protected Flux> refresh(V1ObjectMeta objectMeta) { - - return kubernetesReactiveDiscoveryClient.getInstances(objectMeta.getName()).flatMap(si -> { - URI actuatorUri = getActuatorUri(si); - if (log.isDebugEnabled()) { - log.debug("Sending refresh request for " + objectMeta.getName() + " to URI " + actuatorUri.toString()); - } - Mono> response = webClient.post().uri(actuatorUri).retrieve().toBodilessEntity() - .doOnSuccess(re -> { - if (log.isDebugEnabled()) { - log.debug("Refresh sent to " + objectMeta.getName() + " at URI address " + actuatorUri - + " returned a " + re.getStatusCode().toString()); - } - }).doOnError(t -> { - log.warn("Refresh sent to " + objectMeta.getName() + " failed", t); - }); - return response; - }); + this.httpRefreshTrigger = httpRefreshTrigger; } @Override - protected Mono triggerRefresh(V1ConfigMap configMap) { - return refresh(configMap.getMetadata()).then(); + public Mono triggerRefresh(KubernetesObject configMap) { + return httpRefreshTrigger.triggerRefresh(configMap); } } diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetector.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetector.java index 9d10696a..e25f2767 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetector.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetector.java @@ -16,121 +16,40 @@ package org.springframework.cloud.kubernetes.configuration.watcher; -import java.net.URI; - +import io.kubernetes.client.common.KubernetesObject; import io.kubernetes.client.openapi.apis.CoreV1Api; -import io.kubernetes.client.openapi.models.V1ObjectMeta; -import io.kubernetes.client.openapi.models.V1Secret; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import org.springframework.cloud.client.ServiceInstance; import org.springframework.cloud.kubernetes.client.config.KubernetesClientSecretsPropertySourceLocator; -import org.springframework.cloud.kubernetes.client.discovery.reactive.KubernetesInformerReactiveDiscoveryClient; import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider; import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties; import org.springframework.cloud.kubernetes.commons.config.reload.ConfigurationUpdateStrategy; import org.springframework.core.env.ConfigurableEnvironment; -import org.springframework.http.ResponseEntity; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; -import org.springframework.util.StringUtils; -import org.springframework.web.reactive.function.client.WebClient; -import org.springframework.web.util.UriComponentsBuilder; /** * @author Ryan Baxter * @author Kris Iyer */ -public class HttpBasedSecretsWatchChangeDetector extends SecretsWatcherChangeDetector { +final class HttpBasedSecretsWatchChangeDetector extends SecretsWatcherChangeDetector { - /** - * Annotation key for actuator port and path. - */ - public static String ANNOTATION_KEY = "boot.spring.io/actuator"; + private final HttpRefreshTrigger httpRefreshTrigger; - private WebClient webClient; - - private KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient; - - public HttpBasedSecretsWatchChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, + HttpBasedSecretsWatchChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, KubernetesClientSecretsPropertySourceLocator propertySourceLocator, KubernetesNamespaceProvider kubernetesNamespaceProvider, ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, - ThreadPoolTaskExecutor threadPoolTaskExecutor, WebClient webClient, - KubernetesInformerReactiveDiscoveryClient k8sReactiveDiscoveryClient) { + ThreadPoolTaskExecutor threadPoolTaskExecutor, HttpRefreshTrigger httpRefreshTrigger) { super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider, k8SConfigurationProperties, threadPoolTaskExecutor); - this.webClient = webClient; - this.kubernetesReactiveDiscoveryClient = k8sReactiveDiscoveryClient; + + this.httpRefreshTrigger = httpRefreshTrigger; } @Override - protected Mono triggerRefresh(V1Secret secret) { - return refresh(secret.getMetadata()).then(); - } - - private void setActuatorUriFromAnnotation(UriComponentsBuilder actuatorUriBuilder, String metadataUri) { - URI annotationUri = URI.create(metadataUri); - actuatorUriBuilder.path(annotationUri.getPath() + "/refresh"); - - // The URI may not contain a host so if that is the case the port in the URI will - // be -1 - // The authority of the URI will be : for example :9090, we just need the - // 9090 in this case - if (annotationUri.getPort() < 0) { - if (annotationUri.getAuthority() != null) { - actuatorUriBuilder.port(annotationUri.getAuthority().replaceFirst(":", "")); - } - } - else { - actuatorUriBuilder.port(annotationUri.getPort()); - } - } - - private URI getActuatorUri(ServiceInstance si) { - String metadataUri = si.getMetadata().getOrDefault(ANNOTATION_KEY, ""); - if (log.isDebugEnabled()) { - log.debug("Metadata actuator uri is: " + metadataUri); - } - - UriComponentsBuilder actuatorUriBuilder = UriComponentsBuilder.newInstance().scheme(si.getScheme()) - .host(si.getHost()); - - if (!StringUtils.isEmpty(metadataUri)) { - if (log.isDebugEnabled()) { - log.debug("Found actuator URI in service instance metadata"); - } - setActuatorUriFromAnnotation(actuatorUriBuilder, metadataUri); - } - else { - Integer port = k8SConfigurationProperties.getActuatorPort() < 0 ? si.getPort() - : k8SConfigurationProperties.getActuatorPort(); - actuatorUriBuilder = actuatorUriBuilder.path(k8SConfigurationProperties.getActuatorPath() + "/refresh") - .port(port); - } - - return actuatorUriBuilder.build().toUri(); - } - - protected Flux> refresh(V1ObjectMeta objectMeta) { - - return kubernetesReactiveDiscoveryClient.getInstances(objectMeta.getName()).flatMap(si -> { - URI actuatorUri = getActuatorUri(si); - if (log.isDebugEnabled()) { - log.debug("Sending refresh request for " + objectMeta.getName() + " to URI " + actuatorUri.toString()); - } - Mono> response = webClient.post().uri(actuatorUri).retrieve().toBodilessEntity() - .doOnSuccess(re -> { - if (log.isDebugEnabled()) { - log.debug("Refresh sent to " + objectMeta.getName() + " at URI address " + actuatorUri - + " returned a " + re.getStatusCode().toString()); - } - }).doOnError(t -> { - log.warn("Refresh sent to " + objectMeta.getName() + " failed", t); - }); - return response; - }); + public Mono triggerRefresh(KubernetesObject secret) { + return httpRefreshTrigger.triggerRefresh(secret); } } diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpRefreshTrigger.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpRefreshTrigger.java new file mode 100644 index 00000000..dc032ec3 --- /dev/null +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpRefreshTrigger.java @@ -0,0 +1,117 @@ +/* + * 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.configuration.watcher; + +import java.net.URI; +import java.util.function.Consumer; + +import io.kubernetes.client.common.KubernetesObject; +import org.apache.commons.logging.LogFactory; +import reactor.core.publisher.Mono; + +import org.springframework.cloud.client.ServiceInstance; +import org.springframework.cloud.kubernetes.client.config.reload.KubernetesClientEventBasedSecretsChangeDetector; +import org.springframework.cloud.kubernetes.client.discovery.reactive.KubernetesInformerReactiveDiscoveryClient; +import org.springframework.core.log.LogAccessor; +import org.springframework.http.ResponseEntity; +import org.springframework.util.StringUtils; +import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.util.UriComponentsBuilder; + +/** + * @author wind57 + */ +final class HttpRefreshTrigger implements RefreshTrigger { + + private static final LogAccessor LOG = new LogAccessor( + LogFactory.getLog(KubernetesClientEventBasedSecretsChangeDetector.class)); + + private final KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient; + + private final ConfigurationWatcherConfigurationProperties k8SConfigurationProperties; + + private final WebClient webClient; + + HttpRefreshTrigger(KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient, + ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, WebClient webClient) { + this.kubernetesReactiveDiscoveryClient = kubernetesReactiveDiscoveryClient; + this.k8SConfigurationProperties = k8SConfigurationProperties; + this.webClient = webClient; + } + + @Override + public Mono triggerRefresh(KubernetesObject kubernetesObject) { + + String name = kubernetesObject.getMetadata().getName(); + + return kubernetesReactiveDiscoveryClient.getInstances(name).flatMap(si -> { + URI actuatorUri = getActuatorUri(si, k8SConfigurationProperties.getActuatorPath(), + k8SConfigurationProperties.getActuatorPort()); + LOG.debug(() -> "Sending refresh request for " + name + " to URI " + actuatorUri); + return webClient.post().uri(actuatorUri).retrieve().toBodilessEntity() + .doOnSuccess(onSuccess(name, actuatorUri)).doOnError(onError(name)); + }).then(); + } + + private Consumer> onSuccess(String name, URI actuatorUri) { + return re -> LOG.debug(() -> "Refresh sent to " + name + " at URI address " + actuatorUri + " returned a " + + re.getStatusCode()); + } + + private Consumer onError(String name) { + return t -> LOG.warn(t, () -> "Refresh sent to " + name + " failed"); + } + + private URI getActuatorUri(ServiceInstance si, String actuatorPath, int actuatorPort) { + String metadataUri = si.getMetadata().getOrDefault(ConfigurationWatcherConfigurationProperties.ANNOTATION_KEY, + ""); + LOG.debug(() -> "Metadata actuator uri is: " + metadataUri); + + UriComponentsBuilder actuatorUriBuilder = UriComponentsBuilder.newInstance().scheme(si.getScheme()) + .host(si.getHost()); + + if (StringUtils.hasText(metadataUri)) { + LOG.debug(() -> "Found actuator URI in service instance metadata"); + setActuatorUriFromAnnotation(actuatorUriBuilder, metadataUri); + } + else { + int port = actuatorPort < 0 ? si.getPort() : actuatorPort; + actuatorUriBuilder = actuatorUriBuilder.path(actuatorPath + "/refresh").port(port); + } + + return actuatorUriBuilder.build().toUri(); + } + + private void setActuatorUriFromAnnotation(UriComponentsBuilder actuatorUriBuilder, String metadataUri) { + URI annotationUri = URI.create(metadataUri); + actuatorUriBuilder.path(annotationUri.getPath() + "/refresh"); + + // The URI may not contain a host so if that is the case the port in the URI will + // be -1. The authority of the URI will be : for example :9090, we just need + // the + // 9090 in this case + if (annotationUri.getPort() < 0) { + if (annotationUri.getAuthority() != null) { + actuatorUriBuilder.port(annotationUri.getAuthority().replaceFirst(":", "")); + } + } + else { + actuatorUriBuilder.port(annotationUri.getPort()); + } + } + +} diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/RefreshTrigger.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/RefreshTrigger.java new file mode 100644 index 00000000..9844ab89 --- /dev/null +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/RefreshTrigger.java @@ -0,0 +1,34 @@ +/* + * 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.configuration.watcher; + +import io.kubernetes.client.common.KubernetesObject; +import reactor.core.publisher.Mono; + +/** + * Defines the refresh trigger contract. + * + * @author wind57 + */ +interface RefreshTrigger { + + /** + * @param kubernetesObject either a config-map or secret at the moment. + */ + Mono triggerRefresh(KubernetesObject kubernetesObject); + +} diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/SecretsWatcherChangeDetector.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/SecretsWatcherChangeDetector.java index 5b9c0ccb..2d8a942d 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/SecretsWatcherChangeDetector.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/SecretsWatcherChangeDetector.java @@ -18,13 +18,9 @@ package org.springframework.cloud.kubernetes.configuration.watcher; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; +import io.kubernetes.client.common.KubernetesObject; import io.kubernetes.client.openapi.apis.CoreV1Api; -import io.kubernetes.client.openapi.models.V1Secret; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import reactor.core.publisher.Mono; import org.springframework.cloud.kubernetes.client.config.KubernetesClientSecretsPropertySourceLocator; import org.springframework.cloud.kubernetes.client.config.reload.KubernetesClientEventBasedSecretsChangeDetector; @@ -38,15 +34,14 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; * @author Ryan Baxter * @author Kris Iyer */ -public abstract class SecretsWatcherChangeDetector extends KubernetesClientEventBasedSecretsChangeDetector { - - protected Log log = LogFactory.getLog(getClass()); +abstract class SecretsWatcherChangeDetector extends KubernetesClientEventBasedSecretsChangeDetector + implements RefreshTrigger { private final ScheduledExecutorService executorService; - protected ConfigurationWatcherConfigurationProperties k8SConfigurationProperties; + private final long refreshDelay; - public SecretsWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, + SecretsWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment, ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy, KubernetesClientSecretsPropertySourceLocator propertySourceLocator, KubernetesNamespaceProvider kubernetesNamespaceProvider, @@ -55,42 +50,14 @@ public abstract class SecretsWatcherChangeDetector extends KubernetesClientEvent super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider); this.executorService = Executors.newScheduledThreadPool(k8SConfigurationProperties.getThreadPoolSize(), threadPoolTaskExecutor); - this.k8SConfigurationProperties = k8SConfigurationProperties; + this.refreshDelay = k8SConfigurationProperties.getRefreshDelay().toMillis(); } @Override - protected void onEvent(V1Secret secret) { - if (isSpringCloudKubernetesSecret(secret)) { - if (log.isDebugEnabled()) { - log.debug("Scheduling remote refresh event to be published for Secret " + secret.getMetadata().getName() - + " to be published in " + k8SConfigurationProperties.getRefreshDelay().toMillis() - + " milliseconds"); - } - executorService.schedule(() -> { - try { - triggerRefresh(secret).subscribe(); - } - catch (Throwable t) { - log.warn("Error when refreshing ConfigMap " + secret.getMetadata().getName(), t); - } - }, k8SConfigurationProperties.getRefreshDelay().toMillis(), TimeUnit.MILLISECONDS); - } - else { - if (log.isDebugEnabled()) { - log.debug("Not publishing event. Secret " + secret.getMetadata().getName() - + " does not contain the label " + ConfigurationWatcherConfigurationProperties.SECRET_LABEL); - } - } - } - - protected abstract Mono triggerRefresh(V1Secret secret); - - private boolean isSpringCloudKubernetesSecret(V1Secret secret) { - if (secret.getMetadata() == null || secret.getMetadata().getLabels() == null) { - return false; - } - return Boolean.parseBoolean(secret.getMetadata().getLabels() - .getOrDefault(ConfigurationWatcherConfigurationProperties.SECRET_LABEL, "false")); + protected final void onEvent(KubernetesObject secret) { + // this::refreshTrigger is coming from BusEventBasedSecretsWatcherChangeDetector + WatcherUtil.onEvent(secret, ConfigurationWatcherConfigurationProperties.SECRET_LABEL, refreshDelay, + executorService, "secret", this::triggerRefresh); } } diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/WatcherUtil.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/WatcherUtil.java new file mode 100644 index 00000000..25fbe408 --- /dev/null +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/WatcherUtil.java @@ -0,0 +1,76 @@ +/* + * Copyright 2013-2020 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.configuration.watcher; + +import java.util.Collections; +import java.util.Optional; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; + +import io.kubernetes.client.common.KubernetesObject; +import org.apache.commons.logging.LogFactory; +import reactor.core.publisher.Mono; + +import org.springframework.core.log.LogAccessor; + +/** + * A common place where 'onEvent' code delegates to. + * + * @author wind57 + */ +final class WatcherUtil { + + private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(WatcherUtil.class)); + + private WatcherUtil() { + } + + static void onEvent(KubernetesObject kubernetesObject, String label, long refreshDelay, + ScheduledExecutorService executorService, String type, + Function> triggerRefresh) { + + String name = kubernetesObject.getMetadata().getName(); + boolean isSpringCloudKubernetes = isSpringCloudKubernetes(kubernetesObject, label); + + if (isSpringCloudKubernetes) { + + LOG.debug(() -> "Scheduling remote refresh event to be published for " + type + ": " + name + + " to be published in " + refreshDelay + " milliseconds"); + executorService.schedule(() -> { + try { + triggerRefresh.apply(kubernetesObject).subscribe(); + } + catch (Throwable t) { + LOG.warn(t, "Error when refreshing ConfigMap " + name); + } + }, refreshDelay, TimeUnit.MILLISECONDS); + } + else { + LOG.debug(() -> "Not publishing event." + type + ": + name + does not contain the label " + label); + } + } + + private static boolean isSpringCloudKubernetes(KubernetesObject kubernetesObject, String label) { + if (kubernetesObject.getMetadata() == null) { + return false; + } + return Boolean.parseBoolean(Optional.ofNullable(kubernetesObject.getMetadata().getLabels()) + .orElse(Collections.emptyMap()).getOrDefault(label, "false")); + } + +} diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetectorTests.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetectorTests.java index d1916ee7..2505a4a1 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetectorTests.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetectorTests.java @@ -80,8 +80,8 @@ class BusEventBasedConfigMapWatcherChangeDetectorTests { busProperties = new BusProperties(); changeDetector = new BusEventBasedConfigMapWatcherChangeDetector(coreV1Api, mockEnvironment, configReloadProperties, UPDATE_STRATEGY, configMapPropertySourceLocator, - new KubernetesNamespaceProvider(mockEnvironment), busProperties, - configurationWatcherConfigurationProperties, threadPoolTaskExecutor, applicationEventPublisher); + new KubernetesNamespaceProvider(mockEnvironment), configurationWatcherConfigurationProperties, + threadPoolTaskExecutor, new BusRefreshTrigger(applicationEventPublisher, busProperties.getId())); } @Test diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetectorTests.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetectorTests.java index 28fd3210..6b96219f 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetectorTests.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetectorTests.java @@ -80,8 +80,8 @@ class BusEventBasedSecretsWatcherChangeDetectorTests { busProperties = new BusProperties(); changeDetector = new BusEventBasedSecretsWatcherChangeDetector(coreV1Api, mockEnvironment, configReloadProperties, UPDATE_STRATEGY, secretsPropertySourceLocator, - new KubernetesNamespaceProvider(mockEnvironment), busProperties, - configurationWatcherConfigurationProperties, threadPoolTaskExecutor, applicationEventPublisher); + new KubernetesNamespaceProvider(mockEnvironment), configurationWatcherConfigurationProperties, + threadPoolTaskExecutor, new BusRefreshTrigger(applicationEventPublisher, busProperties.getId())); } @Test diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetectorTests.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetectorTests.java index 7f674124..dff7947e 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetectorTests.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetectorTests.java @@ -114,8 +114,8 @@ class HttpBasedConfigMapWatchChangeDetectorTests { changeDetector = new HttpBasedConfigMapWatchChangeDetector(coreV1Api, mockEnvironment, configReloadProperties, strategy, configMapPropertySourceLocator, new KubernetesNamespaceProvider(mockEnvironment), - configurationWatcherConfigurationProperties, threadPoolTaskExecutor, webClient, - reactiveDiscoveryClient); + configurationWatcherConfigurationProperties, threadPoolTaskExecutor, new HttpRefreshTrigger( + reactiveDiscoveryClient, configurationWatcherConfigurationProperties, webClient)); } @Test @@ -152,7 +152,8 @@ class HttpBasedConfigMapWatchChangeDetectorTests { int port = WIRE_MOCK_SERVER.port(); WireMock.configureFor("localhost", port); Map metadata = new HashMap<>(); - metadata.put(HttpBasedConfigMapWatchChangeDetector.ANNOTATION_KEY, "http://:" + port + "/my/custom/actuator"); + metadata.put(ConfigurationWatcherConfigurationProperties.ANNOTATION_KEY, + "http://:" + port + "/my/custom/actuator"); V1EndpointAddress fooEndpointAddress = new V1EndpointAddress(); fooEndpointAddress.setIp("127.0.0.1"); fooEndpointAddress.setHostname("localhost"); diff --git a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetectorTests.java b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetectorTests.java index 58c96595..25eb9b9a 100644 --- a/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetectorTests.java +++ b/spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetectorTests.java @@ -98,8 +98,8 @@ class HttpBasedSecretsWatchChangeDetectorTests { WebClient webClient = WebClient.builder().build(); changeDetector = new HttpBasedSecretsWatchChangeDetector(coreV1Api, mockEnvironment, configReloadProperties, updateStrategy, secretsPropertySourceLocator, new KubernetesNamespaceProvider(mockEnvironment), - configurationWatcherConfigurationProperties, threadPoolTaskExecutor, webClient, - reactiveDiscoveryClient); + configurationWatcherConfigurationProperties, threadPoolTaskExecutor, new HttpRefreshTrigger( + reactiveDiscoveryClient, configurationWatcherConfigurationProperties, webClient)); } @BeforeAll @@ -148,7 +148,7 @@ class HttpBasedSecretsWatchChangeDetectorTests { void triggerSecretRefreshWithAnnotationActuatorPath() { WireMock.configureFor("localhost", WIRE_MOCK_SERVER.port()); Map metadata = new HashMap<>(); - metadata.put(HttpBasedConfigMapWatchChangeDetector.ANNOTATION_KEY, + metadata.put(ConfigurationWatcherConfigurationProperties.ANNOTATION_KEY, "http://:" + WIRE_MOCK_SERVER.port() + "/my/custom/actuator"); V1EndpointAddress fooEndpointAddress = new V1EndpointAddress(); fooEndpointAddress.setIp("127.0.0.1");