feat: support leader Election (#1358)

Fixes #411

Co-authored-by: Chris Laprun <[email protected]>

Author: csviri

.github/workflows/e2e-test.yml

@@ -21,6 +21,7 @@ jobs:
           - "sample-operators/mysql-schema"
           - "sample-operators/tomcat-operator"
           - "sample-operators/webpage"
+          - "sample-operators/leader-election"
     runs-on: ubuntu-latest
     steps:
       - name: Checkout

docs/documentation/features.md

@@ -683,6 +683,19 @@ See also
 the [integration test](https://github.com/java-operator-sdk/java-operator-sdk/blob/ec37025a15046d8f409c77616110024bf32c3416/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestReconciler.java)
 for this feature.
 
+## Leader Election
+
+Operators are generally deployed with a single running or active instance. However, it is 
+possible to deploy multiple instances in such a way that only one, called the "leader", processes the 
+events. This is achieved via a mechanism called "leader election". While all the instances are 
+running, and even start their event sources to populate the caches, only the leader will process 
+the events. This means that should the leader change for any reason, for example because it 
+crashed, the other instances are already warmed up and ready to pick up where the previous 
+leader left off should one of them become elected leader.
+
+See sample configuration in the [E2E test](https://github.com/java-operator-sdk/java-operator-sdk/blob/144947d89323f1c65de6e86bd8b9a6a8ffe714ff/sample-operators/leader-election/src/main/java/io/javaoperatorsdk/operator/sample/LeaderElectionTestOperator.java#L26-L30)
+.
+
 ## Monitoring with Micrometer
 
 ## Automatic Generation of CRDs

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ControllerManager.java

@@ -0,0 +1,81 @@
+package io.javaoperatorsdk.operator;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.javaoperatorsdk.operator.processing.Controller;
+
+/**
+ * Not to be confused with the controller manager concept from Go's controller-runtime project. In
+ * JOSDK, the equivalent concept is {@link Operator}.
+ */
+class ControllerManager {
+
+  private static final Logger log = LoggerFactory.getLogger(ControllerManager.class);
+
+  @SuppressWarnings("rawtypes")
+  private final Map<String, Controller> controllers = new HashMap<>();
+  private boolean started = false;
+
+  public synchronized void shouldStart() {
+    if (started) {
+      return;
+    }
+    if (controllers.isEmpty()) {
+      throw new OperatorException("No Controller exists. Exiting!");
+    }
+  }
+
+  public synchronized void start(boolean startEventProcessor) {
+    controllers().parallelStream().forEach(c -> c.start(startEventProcessor));
+    started = true;
+  }
+
+  public synchronized void stop() {
+    controllers().parallelStream().forEach(closeable -> {
+      log.debug("closing {}", closeable);
+      closeable.stop();
+    });
+    started = false;
+  }
+
+  public synchronized void startEventProcessing() {
+    controllers().parallelStream().forEach(Controller::startEventProcessing);
+  }
+
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  synchronized void add(Controller controller) {
+    final var configuration = controller.getConfiguration();
+    final var resourceTypeName = ReconcilerUtils
+        .getResourceTypeNameWithVersion(configuration.getResourceClass());
+    final var existing = controllers.get(resourceTypeName);
+    if (existing != null) {
+      throw new OperatorException("Cannot register controller '" + configuration.getName()
+          + "': another controller named '" + existing.getConfiguration().getName()
+          + "' is already registered for resource '" + resourceTypeName + "'");
+    }
+    controllers.put(resourceTypeName, controller);
+  }
+
+  @SuppressWarnings("rawtypes")
+  synchronized Optional<Controller> get(String name) {
+    return controllers().stream()
+        .filter(c -> name.equals(c.getConfiguration().getName()))
+        .findFirst();
+  }
+
+  @SuppressWarnings("rawtypes")
+  synchronized Collection<Controller> controllers() {
+    return controllers.values();
+  }
+
+  synchronized int size() {
+    return controllers.size();
+  }
+}
+

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/LeaderElectionManager.java

@@ -0,0 +1,85 @@
+package io.javaoperatorsdk.operator;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfig;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectorBuilder;
+import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.LeaseLock;
+import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.Lock;
+import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
+import io.javaoperatorsdk.operator.api.config.LeaderElectionConfiguration;
+
+public class LeaderElectionManager {
+
+  private static final Logger log = LoggerFactory.getLogger(LeaderElectionManager.class);
+
+  private LeaderElector leaderElector = null;
+  private final ControllerManager controllerManager;
+  private String identity;
+  private CompletableFuture<?> leaderElectionFuture;
+
+  public LeaderElectionManager(ControllerManager controllerManager) {
+    this.controllerManager = controllerManager;
+  }
+
+  public void init(LeaderElectionConfiguration config, KubernetesClient client) {
+    this.identity = identity(config);
+    Lock lock = new LeaseLock(config.getLeaseNamespace(), config.getLeaseName(), identity);
+    // releaseOnCancel is not used in the underlying implementation
+    leaderElector = new LeaderElectorBuilder(client,
+        ConfigurationServiceProvider.instance().getExecutorService())
+        .withConfig(
+            new LeaderElectionConfig(lock, config.getLeaseDuration(), config.getRenewDeadline(),
+                config.getRetryPeriod(), leaderCallbacks(), true, config.getLeaseName()))
+        .build();
+  }
+
+  public boolean isLeaderElectionEnabled() {
+    return leaderElector != null;
+  }
+
+  private LeaderCallbacks leaderCallbacks() {
+    return new LeaderCallbacks(this::startLeading, this::stopLeading, leader -> {
+      log.info("New leader with identity: {}", leader);
+    });
+  }
+
+  private void startLeading() {
+    controllerManager.startEventProcessing();
+  }
+
+  private void stopLeading() {
+    log.info("Stopped leading for identity: {}. Exiting.", identity);
+    // When leader stops leading the process ends immediately to prevent multiple reconciliations
+    // running parallel.
+    // Note that some reconciliations might run for a very long time.
+    System.exit(1);
+  }
+
+  private String identity(LeaderElectionConfiguration config) {
+    String id = config.getIdentity().orElse(System.getenv("HOSTNAME"));
+    if (id == null || id.isBlank()) {
+      id = UUID.randomUUID().toString();
+    }
+    return id;
+  }
+
+  public void start() {
+    if (isLeaderElectionEnabled()) {
+      leaderElectionFuture = leaderElector.start();
+    }
+  }
+
+  public void stop() {
+    if (leaderElectionFuture != null) {
+      leaderElectionFuture.cancel(false);
+    }
+  }
+}