fix-1828: started work

docs/modules/ROOT/pages/spring-cloud-kubernetes-configuration-watcher.adoc

@@ -98,6 +98,74 @@ This tells how many milliseconds should we wait before firing the event from con
 
 You need to "match" this _eventually_ part to that value in milliseconds on your cluster.
 
+## High Availability
+
+The Configuration Watcher can run with multiple replicas, with only one replica actively
+watching ConfigMaps and Secrets at a time. High availability is disabled by default. Enable
+both the native Kubernetes leader election and Configuration Watcher HA:
+
+[source,yaml]
+----
+spring:
+  cloud:
+    kubernetes:
+      leader:
+        election:
+          enabled: true
+      configuration:
+        watcher:
+          ha:
+            enabled: true
+----
+
+The two settings have different responsibilities:
+
+* `spring.cloud.kubernetes.leader.election.enabled` enables the native Kubernetes leader
+  election used to choose the active watcher replica.
+* `spring.cloud.kubernetes.configuration.watcher.ha.enabled` makes the Configuration Watcher
+  start its ConfigMap and Secret informers only after it becomes the leader, and stop them when
+  it loses leadership.
+
+The leader-election lock and the Configuration Watcher HA state are stored separately. The
+Configuration Watcher state is persisted in a Kubernetes `Lease`. Its defaults are:
+
+* `spring.cloud.kubernetes.configuration.watcher.ha.lease-name`: `configuration-watcher-ha`
+* `spring.cloud.kubernetes.configuration.watcher.ha.lease-namespace`: `default`
+
+The state lease stores the last processed informer resource version separately for each watched
+namespace and resource type. When a new replica becomes the leader, it reads those checkpoints
+and starts each informer from its stored resource version. This allows changes observed while no
+watcher was leader to be delivered to the new leader. The watcher does not refresh every resource
+after a leadership change; it only replays events for the configured ConfigMap and Secret
+informers.
+
+Processing is at least once. If a watcher loses leadership after an event was delivered but before
+its checkpoint was persisted, the replacement leader may process that event again. Refresh targets
+must therefore tolerate repeated refresh notifications.
+
+The service account used by the Configuration Watcher needs permission to read and update the HA
+state lease in the configured lease namespace. Add the following rule to the Role in that
+namespace:
+
+[source,yaml]
+----
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["get", "create", "update"]
+----
+
+The HA state lease defaults to the `default` namespace even when the Configuration Watcher runs
+in another namespace. If you configure a different `lease-namespace`, create the corresponding
+Role and RoleBinding there, or grant equivalent cluster-scoped permissions.
+
+Configure at least two replicas in the Configuration Watcher Deployment to enable failover:
+
+[source,yaml]
+----
+spec:
+  replicas: 2
+----
+
 Spring Cloud Kubernetes Configuration Watcher can send refresh notifications to applications in two ways.
 
 1.  Over HTTP, in which case the application being notified, must have the `/refresh` actuator endpoint exposed and accessible from within the cluster
@@ -155,6 +223,9 @@ items:
       - apiGroups: ["", "extensions", "apps"]
         resources: ["configmaps", "pods", "services", "endpoints", "secrets"]
         verbs: ["get", "list", "watch"]
+      - apiGroups: ["coordination.k8s.io"]
+        resources: ["leases"]
+        verbs: ["get", "create", "update"]
   - apiVersion: apps/v1
     kind: Deployment
     metadata:

spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/ConfigMapResourceEventHandler.java

@@ -21,7 +21,8 @@
 
 import io.kubernetes.client.informer.ResourceEventHandler;
 import io.kubernetes.client.openapi.models.V1ConfigMap;
-import org.apache.commons.logging.LogFactory;
+import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import jakarta.annotation.Nullable;
 
 import org.springframework.core.log.LogAccessor;
 
@@ -30,19 +31,25 @@
  */
 final class ConfigMapResourceEventHandler implements ResourceEventHandler<V1ConfigMap> {
 
-	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(ConfigMapResourceEventHandler.class));
+	private static final LogAccessor LOG = new LogAccessor(ConfigMapResourceEventHandler.class);
 
 	private final Consumer<V1ConfigMap> onEvent;
 
-	ConfigMapResourceEventHandler(Consumer<V1ConfigMap> onEvent) {
+	@Nullable
+	private final Consumer<NamespaceAndResourceVersion> resourceVersionWriter;
+
+	ConfigMapResourceEventHandler(Consumer<V1ConfigMap> onEvent,
+			@Nullable Consumer<NamespaceAndResourceVersion> resourceVersionWriter) {
 		this.onEvent = onEvent;
+		this.resourceVersionWriter = resourceVersionWriter;
 	}
 
 	@Override
 	public void onAdd(V1ConfigMap configMap) {
 		LOG.debug(() -> "ConfigMap " + configMap.getMetadata().getName() + " was added in namespace "
 				+ configMap.getMetadata().getNamespace());
 		onEvent.accept(configMap);
+		writeResourceVersion(configMap);
 	}
 
 	@Override
@@ -55,13 +62,23 @@ public void onUpdate(V1ConfigMap oldConfigMap, V1ConfigMap newConfigMap) {
 		else {
 			onEvent.accept(newConfigMap);
 		}
+		writeResourceVersion(newConfigMap);
 	}
 
 	@Override
 	public void onDelete(V1ConfigMap configMap, boolean deletedFinalStateUnknown) {
 		LOG.debug(() -> "ConfigMap " + configMap.getMetadata().getName() + " was deleted in namespace "
 				+ configMap.getMetadata().getNamespace());
 		onEvent.accept(configMap);
+		writeResourceVersion(configMap);
+	}
+
+	private void writeResourceVersion(V1ConfigMap configMap) {
+		if (resourceVersionWriter != null) {
+			V1ObjectMeta metadata = configMap.getMetadata();
+			resourceVersionWriter
+				.accept(new NamespaceAndResourceVersion(metadata.getNamespace(), metadata.getResourceVersion()));
+		}
 	}
 
 }

spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/InformerResourceVersionResolver.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2013-present 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.client.config.reload;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * Resolves the resource version for each informer request.
+ *
+ * <p>
+ * When HA is disabled, the resolver always returns the resource version supplied by the
+ * informer. When HA is enabled, it returns the persisted resource version exactly once,
+ * for the first request, so that the informer can replay events from the stored
+ * checkpoint. Every subsequent request must use the resource version supplied by the
+ * informer because that value advances as the informer processes list and watch calls.
+ *
+ * @author wind57
+ */
+final class InformerResourceVersionResolver {
+
+	private static final LogAccessor LOG = new LogAccessor(InformerResourceVersionResolver.class);
+
+	// resource versions that we have in our custom lease, these are per namespace
+	private final Map<String, String> checkpointResourceVersions;
+
+	private final boolean haEnabled;
+
+	// key is the namespace, value is whether its stored resource
+	// version was already consumed or not
+	private final Map<String, AtomicBoolean> checkpointResourceVersionConsumed = new ConcurrentHashMap<>();
+
+	InformerResourceVersionResolver(Map<String, String> checkpointResourceVersions, boolean haEnabled) {
+		this.checkpointResourceVersions = checkpointResourceVersions;
+		this.haEnabled = haEnabled;
+	}
+
+	String resolve(String namespace, String informerResourceVersion) {
+
+		// if HA is not enabled, we do not restore from a checkpoint
+		if (!haEnabled) {
+			return informerResourceVersion;
+		}
+
+		String checkpointResourceVersion = checkpointResourceVersions.get(namespace);
+		// there is no previous checkpoint ( maybe it's the first time app is started in
+		// HA mode)
+		if (checkpointResourceVersion == null) {
+			return informerResourceVersion;
+		}
+
+		// Consume the checkpoint only once.
+		// Subsequent versions come from the informer, so it can progress.
+		boolean checkpointNotConsumed = checkpointResourceVersionConsumed
+			.computeIfAbsent(namespace, ignored -> new AtomicBoolean())
+			.compareAndSet(false, true);
+
+		if (checkpointNotConsumed) {
+			LOG.info(() -> "replaying events in namespace " + namespace + " from resource version "
+					+ checkpointResourceVersion);
+			return checkpointResourceVersion;
+		}
+		return informerResourceVersion;
+	}
+
+}