Use ConfigMapList's resourceVersion for watching (#300)

Rationale: avoid `ERROR` notifications during rolling upgrades.

Using the greatest `resourceVersion` of all the `ConfigMap`s returned
within the `ConfigMapList` works as expected for *fresh* deployments.
But, when performing a *rolling upgrade* (and depending on the upgrade
strategy), the watcher happens to frequently stop after having received
an `ERROR` notification:

> [ERROR] [KubernetesConfigMapWatcher] [] Kubernetes API returned an
error for a ConfigMap watch event: ConfigMapWatchEvent{type=ERROR,
object=ConfigMap{metadata=Metadata{name='null', namespace='null',
uid='null', labels={}, resourceVersion=null}, data={}}}

What's actually streamed in that case is a `Status` object such as:
```json
{
  "type": "ERROR",
  "object": {
    "kind": "Status",
    "apiVersion": "v1",
    "metadata": {},
    "status": "Expired",
    "message": "too old resource version: 123 (456)",
    "reason": "Gone",
    "code": 410
  }
}
```

A few references:
* ManageIQ/kubeclient#452
* https://www.baeldung.com/java-kubernetes-watch#1-resource-versions

It's possible to recover by adding some logic to reinstall the watcher
starting with the newly advertised `resourceVersion`, but this may be
avoided at all by starting the initial watch at the `resourceVersion`
of the `ConfigMapList` itself: this one won't expire.

The proposed implementation consists in storing last received
`resourceVersion` as an additional `PropertySource` (through a dedicated
name and version key) and later using it when installing the watcher.

Co-authored-by: Regis Desgroppes <rdesgroppes@users.noreply.github.com>
Co-authored-by: Pavol Gressa <pavol.gressa@oracle.com>

Co-authored-by: Georg Rollinger <georg.rollinger@posteo.net>
Co-authored-by: Pavol Gressa <pavol.gressa@oracle.com>

kubernetes-discovery-client/src/main/java/io/micronaut/kubernetes/client/v1/configmaps/ConfigMapList.java

@@ -16,6 +16,7 @@
 package io.micronaut.kubernetes.client.v1.configmaps;
 
 import io.micronaut.core.annotation.Introspected;
+import io.micronaut.kubernetes.client.v1.KubernetesObject;
 
 import java.util.Collections;
 import java.util.List;
@@ -28,7 +29,7 @@
  * @since 1.0.0
  */
 @Introspected
-public class ConfigMapList {
+public class ConfigMapList extends KubernetesObject {
 
     private List<ConfigMap> items;
 

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

@@ -39,7 +39,7 @@
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 
-import static io.micronaut.kubernetes.configuration.KubernetesConfigurationClient.KUBERNETES_CONFIG_MAP_NAME_SUFFIX;
+import static io.micronaut.kubernetes.configuration.KubernetesConfigurationClient.KUBERNETES_CONFIG_MAP_LIST_NAME;
 import static io.micronaut.kubernetes.util.KubernetesUtils.computePodLabelSelector;
 
 /**
@@ -109,8 +109,8 @@ private long computeLastResourceVersion() {
         long lastResourceVersion = environment
                 .getPropertySources()
                 .stream()
-                .filter(propertySource -> propertySource.getName().endsWith(KUBERNETES_CONFIG_MAP_NAME_SUFFIX))
-                .map(propertySource -> propertySource.get(KubernetesConfigurationClient.CONFIG_MAP_RESOURCE_VERSION))
+                .filter(propertySource -> propertySource.getName().equals(KUBERNETES_CONFIG_MAP_LIST_NAME))
+                .map(propertySource -> propertySource.get(KubernetesConfigurationClient.CONFIG_MAP_LIST_RESOURCE_VERSION))
                 .map(o -> Long.parseLong(o.toString()))
                 .max(Long::compareTo)
                 .orElse(0L);

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

@@ -45,6 +45,7 @@
 
 import static io.micronaut.kubernetes.client.v1.secrets.Secret.OPAQUE_SECRET_TYPE;
 import static io.micronaut.kubernetes.util.KubernetesUtils.computePodLabelSelector;
+import static java.util.Collections.singletonMap;
 
 /**
  * A {@link ConfigurationClient} implementation that provides {@link PropertySource}s read from Kubernetes ConfigMap's.
@@ -59,7 +60,9 @@
 @BootstrapContextCompatible
 public class KubernetesConfigurationClient implements ConfigurationClient {
 
+    public static final String CONFIG_MAP_LIST_RESOURCE_VERSION = "configMapListResourceVersion";
     public static final String CONFIG_MAP_RESOURCE_VERSION = "configMapResourceVersion";
+    public static final String KUBERNETES_CONFIG_MAP_LIST_NAME = "Kubernetes ConfigMapList";
     public static final String KUBERNETES_CONFIG_MAP_NAME_SUFFIX = " (Kubernetes ConfigMap)";
     public static final String KUBERNETES_SECRET_NAME_SUFFIX = " (Kubernetes Secret)";
 
@@ -154,15 +157,30 @@ private Flowable<PropertySource> getPropertySourcesFromConfigMaps() {
                         LOG.debug("Found {} config maps. Applying includes/excludes filters (if any)", configMapList.getItems().size());
                     }
                 })
-                .flatMapIterable(ConfigMapList::getItems)
-                .filter(includesFilter)
-                .filter(excludesFilter)
-                .doOnNext(configMap -> {
-                    if (LOG.isDebugEnabled()) {
-                        LOG.debug("Adding config map with name {}", configMap.getMetadata().getName());
-                    }
-                })
-                .map(KubernetesUtils::configMapAsPropertySource);
+                .flatMap(configMapList -> Flowable.just(configMapListAsPropertySource(configMapList))
+                        .mergeWith(Flowable.fromIterable(configMapList.getItems())
+                                .filter(includesFilter)
+                                .filter(excludesFilter)
+                                .doOnNext(configMap -> {
+                                    if (LOG.isDebugEnabled()) {
+                                        LOG.debug("Adding config map with name {}", configMap.getMetadata().getName());
+                                    }
+                                })
+                                .map(KubernetesUtils::configMapAsPropertySource)));
+    }
+
+    /**
+     * Converts a {@link ConfigMapList} into a {@link PropertySource}.
+     *
+     * @param configMapList the ConfigMapList
+     * @return A PropertySource
+     */
+    private static PropertySource configMapListAsPropertySource(ConfigMapList configMapList) {
+        String resourceVersion = configMapList.getMetadata().getResourceVersion();
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Adding config map list with version {}", resourceVersion);
+        }
+        return PropertySource.of(KUBERNETES_CONFIG_MAP_LIST_NAME, singletonMap(CONFIG_MAP_LIST_RESOURCE_VERSION, resourceVersion), EnvironmentPropertySource.POSITION + 100);
     }
 
     private Flowable<PropertySource> getPropertySourcesFromSecrets() {

kubernetes-discovery-client/src/test/groovy/io/micronaut/kubernetes/configuration/KubernetesConfigurationClientLabelsSpec.groovy

@@ -49,7 +49,8 @@ class KubernetesConfigurationClientLabelsSpec extends KubernetesSpecification {
         def propertySources = Flowable.fromPublisher(configurationClient.getPropertySources(applicationContext.environment)).blockingIterable()
 
         then:
-        propertySources.size() == 0
+        propertySources.size() == 1
+        propertySources.first().name == KubernetesConfigurationClient.KUBERNETES_CONFIG_MAP_LIST_NAME
     }
 
     void "it can filter secrets by labels"() {