Send a RefreshEvent when the environment is refreshed because of a Co…

…nfigMap change (micronaut-projects#301)

* Make `environment` final

The injected `Environment` instance remains the same over
`KubernetesConfigMapWatcher`s lifetime.

A subsequent commit will anyway replace occurrences of
`environment.refresh()` with `environment.refreshAndDiff()`, which does
not return a (potentially new) `Environment`.

* Remove direct mutations of `environment`

Commit 8c29208 introduced a workaround
on micronaut-projects/micronaut-core#1903 so that updated
`PropertySource`s are reflected in the environment after a refresh.

Commit 6d50e2b introduced a local cache
for `PropertySource`s so that the client doesn't trigger each time a
`RefreshEvent` gets published.

Following that, `PropertySource`s were directly added to/removed from
two distinct storages:
1. the above mentioned local cache,
2. the current `Environment`.

Removing direct mutations of the current `Environment` allows:
1. to get rid of the workaround on
   micronaut-projects/micronaut-core#1903 which has since been addressed
   by micronaut-projects/micronaut-core#1965,
2. `environment.refreshAndDiff()` to find and return changes that
   happened between the former state (`copiedCatalog`: the current
   `Environment`) and the new one (`catalog`: the local cache).

Changes in `PropertySource`s will anyway get automatically propagated to
the environment.

* Send a RefreshEvent when a ConfigMap change affects the Environment

Co-authored-by: Regis Desgroppes <rdesgroppes@users.noreply.github.com>

Co-authored-by: Regis Desgroppes <rdesgroppes@users.noreply.github.com>

kubernetes-discovery-client/src/main/java/io/micronaut/kubernetes/configuration/KubernetesConfigMapWatcher.java

@@ -19,13 +19,15 @@
 import io.micronaut.context.env.Environment;
 import io.micronaut.context.env.PropertySource;
 import io.micronaut.context.event.ApplicationEventListener;
+import io.micronaut.context.event.ApplicationEventPublisher;
 import io.micronaut.core.util.StringUtils;
 import io.micronaut.discovery.event.ServiceReadyEvent;
 import io.micronaut.kubernetes.client.v1.KubernetesClient;
 import io.micronaut.kubernetes.client.v1.KubernetesConfiguration;
 import io.micronaut.kubernetes.client.v1.configmaps.ConfigMap;
 import io.micronaut.kubernetes.client.v1.configmaps.ConfigMapWatchEvent;
 import io.micronaut.kubernetes.util.KubernetesUtils;
+import io.micronaut.runtime.context.scope.refresh.RefreshEvent;
 import io.reactivex.Flowable;
 import io.reactivex.schedulers.Schedulers;
 import org.slf4j.Logger;
@@ -55,18 +57,20 @@ public class KubernetesConfigMapWatcher implements ApplicationEventListener<Serv
 
     private static final Logger LOG = LoggerFactory.getLogger(KubernetesConfigMapWatcher.class);
 
-    private Environment environment;
+    private final Environment environment;
     private final KubernetesClient client;
     private final KubernetesConfiguration configuration;
     private final ExecutorService executorService;
+    private final ApplicationEventPublisher eventPublisher;
 
     /**
      * @param environment the {@link Environment}
      * @param client the {{@link KubernetesClient}}
      * @param configuration the {@link KubernetesConfiguration}
      * @param executorService the IO {@link ExecutorService} where the watch publisher will be scheduled on
+     * @param eventPublisher the {@link ApplicationEventPublisher}
      */
-    public KubernetesConfigMapWatcher(Environment environment, KubernetesClient client, KubernetesConfiguration configuration, @Named("io") ExecutorService executorService) {
+    public KubernetesConfigMapWatcher(Environment environment, KubernetesClient client, KubernetesConfiguration configuration, @Named("io") ExecutorService executorService, ApplicationEventPublisher eventPublisher) {
         if (LOG.isDebugEnabled()) {
             LOG.debug("Initializing {}", getClass().getName());
         }
@@ -75,6 +79,7 @@ public KubernetesConfigMapWatcher(Environment environment, KubernetesClient clie
         this.client = client;
         this.configuration = configuration;
         this.executorService = executorService;
+        this.eventPublisher = eventPublisher;
     }
 
     @SuppressWarnings("ResultOfMethodCallIgnored")
@@ -101,7 +106,7 @@ public void onApplicationEvent(ServiceReadyEvent event) {
     }
 
     private long computeLastResourceVersion() {
-        long lastResourceVersion = this.environment
+        long lastResourceVersion = environment
                 .getPropertySources()
                 .stream()
                 .filter(propertySource -> propertySource.getName().endsWith(KUBERNETES_CONFIG_MAP_NAME_SUFFIX))
@@ -144,9 +149,8 @@ private void processConfigMapAdded(ConfigMap configMap) {
             propertySource = KubernetesUtils.configMapAsPropertySource(configMap);
         }
         if (passesIncludesExcludesLabelsFilters(configMap)) {
-            this.environment.addPropertySource(propertySource);
             KubernetesConfigurationClient.addPropertySourceToCache(propertySource);
-            this.environment = environment.refresh();
+            refreshEnvironment();
         }
     }
 
@@ -156,11 +160,9 @@ private void processConfigMapModified(ConfigMap configMap) {
             propertySource = KubernetesUtils.configMapAsPropertySource(configMap);
         }
         if (passesIncludesExcludesLabelsFilters(configMap)) {
-            this.environment.removePropertySource(propertySource);
-            this.environment.addPropertySource(propertySource);
             KubernetesConfigurationClient.removePropertySourceFromCache(propertySource.getName());
             KubernetesConfigurationClient.addPropertySourceToCache(propertySource);
-            this.environment = environment.refresh();
+            refreshEnvironment();
         }
     }
 
@@ -170,9 +172,23 @@ private void processConfigMapDeleted(ConfigMap configMap) {
             propertySource = KubernetesUtils.configMapAsPropertySource(configMap);
         }
         if (passesIncludesExcludesLabelsFilters(configMap)) {
-            this.environment.removePropertySource(propertySource);
             KubernetesConfigurationClient.removePropertySourceFromCache(propertySource.getName());
-            this.environment = environment.refresh();
+            refreshEnvironment();
+        }
+    }
+
+    /**
+     * Send a {@link RefreshEvent} when a {@link ConfigMap} change affects the {@link Environment}
+     *
+     * @see io.micronaut.management.endpoint.refresh.RefreshEndpoint#refresh(Boolean)
+     */
+    private void refreshEnvironment() {
+        final Map<String, Object> changes = environment.refreshAndDiff();
+        if (LOG.isTraceEnabled()) {
+            LOG.trace("Changes in ConfigMap property sources: [{}]", String.join(", ", changes.keySet()));
+        }
+        if (!changes.isEmpty()) {
+            eventPublisher.publishEvent(new RefreshEvent(changes));
         }
     }
 

kubernetes-discovery-client/src/main/java/io/micronaut/kubernetes/configuration/KubernetesConfigurationClient.java

@@ -128,7 +128,7 @@ public static void addPropertySourceToCache(PropertySource propertySource) {
      */
     static void removePropertySourceFromCache(String name) {
         if (LOG.isTraceEnabled()) {
-            LOG.trace("Removing property source {} to cache", name);
+            LOG.trace("Removing property source {} from cache", name);
         }
         propertySources.remove(name);
     }