leader election impl

Author: csviri

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

@@ -9,16 +9,16 @@
 import org.slf4j.LoggerFactory;
 
 import io.javaoperatorsdk.operator.processing.Controller;
-import io.javaoperatorsdk.operator.processing.LifecycleAware;
 
 /**
  * Not confuse with controller manager form go operators. The highest level aggregate is
  * {@link Operator} in JOSDK.
  */
-class ControllerManager implements LifecycleAware {
+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;
 
@@ -31,8 +31,8 @@ public synchronized void shouldStart() {
     }
   }
 
-  public synchronized void start() {
-    controllers().parallelStream().forEach(Controller::start);
+  public synchronized void start(boolean startEventProcessor) {
+    controllers().parallelStream().forEach(c -> c.start(startEventProcessor));
     started = true;
   }
 
@@ -41,10 +41,13 @@ public synchronized void stop() {
       log.debug("closing {}", closeable);
       closeable.stop();
     });
-
     started = false;
   }
 
+  public synchronized void startEventProcessing() {
+    controllers().parallelStream().forEach(Controller::startEventProcessing);
+  }
+
   @SuppressWarnings("unchecked")
   synchronized void add(Controller controller) {
     final var configuration = controller.getConfiguration();
@@ -57,17 +60,16 @@ synchronized void add(Controller controller) {
           + "' is already registered for resource '" + resourceTypeName + "'");
     }
     controllers.put(resourceTypeName, controller);
-    if (started) {
-      controller.start();
-    }
   }
 
+  @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();
   }

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

@@ -1,6 +1,7 @@
 package io.javaoperatorsdk.operator;
 
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -20,16 +21,18 @@ public class LeaderElectionManager {
   private static final Logger log = LoggerFactory.getLogger(LeaderElectionManager.class);
 
   private LeaderElector leaderElector = null;
-  private ControllerManager controllerManager;
+  private final ControllerManager controllerManager;
+  private String identity;
+  private CompletableFuture<?> leaderElectionFuture;
 
   public LeaderElectionManager(ControllerManager controllerManager) {
     this.controllerManager = controllerManager;
   }
 
   public void init(LeaderElectionConfiguration config, KubernetesClient client) {
-    Lock lock = new LeaseLock(config.getLeaseNamespace(), config.getLeaseName(), identity(config));
-    // todo releaseOnCancel
-    // todo use this executor service?
+    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(
@@ -38,7 +41,6 @@ public void init(LeaderElectionConfiguration config, KubernetesClient client) {
         .build();
   }
 
-
   public boolean isLeaderElectionOn() {
     return leaderElector != null;
   }
@@ -50,11 +52,15 @@ private LeaderCallbacks leaderCallbacks() {
   }
 
   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 a very long time.
+    System.exit(1);
   }
 
   private String identity(LeaderElectionConfiguration config) {
@@ -65,4 +71,15 @@ private String identity(LeaderElectionConfiguration config) {
     return identity;
   }
 
+  public void start() {
+    if (isLeaderElectionOn()) {
+      leaderElectionFuture = leaderElector.start();
+    }
+  }
+
+  public void stop() {
+    if (leaderElectionFuture != null) {
+      leaderElectionFuture.cancel(false);
+    }
+  }
 }

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

@@ -11,6 +11,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
 import io.fabric8.kubernetes.client.Version;
 import io.javaoperatorsdk.operator.api.config.*;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
@@ -24,9 +25,10 @@ public class Operator implements LifecycleAware {
   private final ControllerManager controllerManager = new ControllerManager();
   private final LeaderElectionManager leaderElectionManager =
       new LeaderElectionManager(controllerManager);
+  private volatile boolean started = false;
 
   public Operator() {
-    this(new DefaultKubernetesClient(), ConfigurationServiceProvider.instance());
+    this(new KubernetesClientBuilder().build(), ConfigurationServiceProvider.instance());
   }
 
   public Operator(KubernetesClient kubernetesClient) {
@@ -42,7 +44,7 @@ public Operator(ConfigurationService configurationService) {
   }
 
   public Operator(Consumer<ConfigurationServiceOverrider> overrider) {
-    this(new DefaultKubernetesClient(), overrider);
+    this(new KubernetesClientBuilder().build(), overrider);
   }
 
   public Operator(KubernetesClient client, Consumer<ConfigurationServiceOverrider> overrider) {
@@ -82,9 +84,11 @@ public KubernetesClient getKubernetesClient() {
    */
   public void start() {
     try {
-
+      if (started) {
+        return;
+      }
+      started = true;
       controllerManager.shouldStart();
-
       final var version = ConfigurationServiceProvider.instance().getVersion();
       log.info(
           "Operator SDK {} (commit: {}) built on {} starting...",
@@ -94,9 +98,9 @@ public void start() {
 
       final var clientVersion = Version.clientVersion();
       log.info("Client version: {}", clientVersion);
-
       ExecutorServiceManager.init();
-      controllerManager.start();
+      controllerManager.start(!leaderElectionManager.isLeaderElectionOn());
+      leaderElectionManager.start();
     } catch (Exception e) {
       log.error("Error starting operator", e);
       stop();
@@ -111,8 +115,8 @@ public void stop() throws OperatorException {
         "Operator SDK {} is shutting down...", configurationService.getVersion().getSdkVersion());
 
     controllerManager.stop();
-
     ExecutorServiceManager.stop();
+    leaderElectionManager.stop();
     if (configurationService.closeClientOnStop()) {
       kubernetesClient.close();
     }
@@ -148,6 +152,9 @@ public <P extends HasMetadata> RegisteredController<P> register(Reconciler<P> re
   public <P extends HasMetadata> RegisteredController<P> register(Reconciler<P> reconciler,
       ControllerConfiguration<P> configuration)
       throws OperatorException {
+    if (started) {
+      throw new OperatorException("Operator already started. Register all the controllers before.");
+    }
 
     if (configuration == null) {
       throw new OperatorException(

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java

@@ -45,7 +45,7 @@
 @SuppressWarnings({"unchecked", "rawtypes"})
 @Ignore
 public class Controller<P extends HasMetadata>
-    implements Reconciler<P>, Cleaner<P>, LifecycleAware,
+    implements Reconciler<P>, Cleaner<P>,
     RegisteredController<P> {
 
   private static final Logger log = LoggerFactory.getLogger(Controller.class);
@@ -62,7 +62,6 @@ public class Controller<P extends HasMetadata>
   private final GroupVersionKind associatedGVK;
   private final EventProcessor<P> eventProcessor;
 
-
   public Controller(Reconciler<P> reconciler,
       ControllerConfiguration<P> configuration,
       KubernetesClient kubernetesClient) {
@@ -272,7 +271,7 @@ public MixedOperation<P, KubernetesResourceList<P>, Resource<P>> getCRClient() {
    *
    * @throws OperatorException if a problem occurred during the registration process
    */
-  public void start() throws OperatorException {
+  public synchronized void start(boolean startEventProcessor) throws OperatorException {
     final Class<P> resClass = configuration.getResourceClass();
     final String controllerName = configuration.getName();
     final var crdName = configuration.getResourceTypeName();
@@ -290,14 +289,17 @@ public void start() throws OperatorException {
 
       initAndRegisterEventSources(context);
       eventSourceManager.start();
-      eventProcessor.start();
+      if (startEventProcessor) {
+        eventProcessor.start();
+      }
       log.info("'{}' controller started, pending event sources initialization", controllerName);
     } catch (MissingCRDException e) {
       stop();
       throwMissingCRDException(crdName, specVersion, controllerName);
     }
   }
 
+
   private void validateCRDWithLocalModelIfRequired(Class<P> resClass, String controllerName,
       String crdName, String specVersion) {
     final CustomResourceDefinition crd;
@@ -323,14 +325,11 @@ public void changeNamespaces(Set<String> namespaces) {
     eventProcessor.start();
   }
 
-  public void startEventProcessing() {
+  public synchronized void startEventProcessing() {
+    log.info("Started event processing for controller: {}", configuration.getName());
     eventProcessor.start();
   }
 
-  public void stopEventProcessing() {
-    eventProcessor.stop();
-  }
-
   private void throwMissingCRDException(String crdName, String specVersion, String controllerName) {
     throw new MissingCRDException(
         crdName,
@@ -363,13 +362,13 @@ public EventSourceManager<P> getEventSourceManager() {
     return eventSourceManager;
   }
 
-  public void stop() {
-    if (eventSourceManager != null) {
-      eventSourceManager.stop();
-    }
+  public synchronized void stop() {
     if (eventProcessor != null) {
       eventProcessor.stop();
     }
+    if (eventSourceManager != null) {
+      eventSourceManager.stop();
+    }
   }
 
   public boolean useFinalizer() {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java

@@ -24,7 +24,7 @@ void crdShouldNotBeCheckedForNativeResources() {
     final var configuration = MockControllerConfiguration.forResource(Secret.class);
 
     final var controller = new Controller<Secret>(reconciler, configuration, client);
-    controller.start();
+    controller.start(true);
     verify(client, never()).apiextensions();
   }
 
@@ -36,7 +36,7 @@ void crdShouldNotBeCheckedForCustomResourcesIfDisabled() {
     try {
       ConfigurationServiceProvider.overrideCurrent(o -> o.checkingCRDAndValidateLocalModel(false));
       final var controller = new Controller<TestCustomResource>(reconciler, configuration, client);
-      controller.start();
+      controller.start(true);
       verify(client, never()).apiextensions();
     } finally {
       ConfigurationServiceProvider.reset();

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java

@@ -121,7 +121,7 @@ public boolean useFinalizer() {
         return useFinalizer;
       }
     };
-    controller.start();
+    controller.start(true);
 
     return new ReconciliationDispatcher<>(controller, customResourceFacade);
   }