fix: change namespace starts processor on namespace change even if no…

…t leader (#2344)



Signed-off-by: Attila Mészáros <csviri@gmail.com>

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

@@ -368,17 +368,25 @@ private void validateCRDWithLocalModelIfRequired(Class<P> resClass, String contr
     }
   }
 
-  public void changeNamespaces(Set<String> namespaces) {
+  public synchronized void changeNamespaces(Set<String> namespaces) {
     if (namespaces.contains(WATCH_CURRENT_NAMESPACE)) {
       throw new OperatorException("Unexpected value in target namespaces: " + namespaces);
     }
     if (namespaces.contains(Constants.WATCH_ALL_NAMESPACES) && namespaces.size() > 1) {
       throw new OperatorException(
           "Watching all namespaces, but additional specific namespace is present");
     }
-    eventProcessor.stop();
+    // if the processor was not running, for example because the controller
+    // was not leading in a HA setup, we don't want to stop and
+    // mainly start the processor on namespace change.
+    boolean eventProcessorWasRunning = eventProcessor.isRunning();
+    if (eventProcessorWasRunning) {
+      eventProcessor.stop();
+    }
     eventSourceManager.changeNamespaces(namespaces);
-    eventProcessor.start();
+    if (eventProcessorWasRunning) {
+      eventProcessor.start();
+    }
   }
 
   public synchronized void startEventProcessing() {

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

@@ -473,4 +473,8 @@ private String controllerName() {
   public synchronized boolean isUnderProcessing(ResourceID resourceID) {
     return isControllerUnderExecution(resourceStateManager.getOrCreate(resourceID));
   }
+
+  public synchronized boolean isRunning() {
+    return running;
+  }
 }

operator-framework/src/test/java/io/javaoperatorsdk/operator/LeaderElectionChangeNamespaceIT.java

@@ -0,0 +1,98 @@
+package io.javaoperatorsdk.operator;
+
+import java.time.Duration;
+import java.time.ZonedDateTime;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.coordination.v1.Lease;
+import io.fabric8.kubernetes.api.model.coordination.v1.LeaseSpecBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import io.javaoperatorsdk.operator.api.config.LeaderElectionConfiguration;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.leaderelectionchangenamespace.LeaderElectionChangeNamespaceCustomResource;
+import io.javaoperatorsdk.operator.sample.leaderelectionchangenamespace.LeaderElectionChangeNamespaceReconciler;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+public class LeaderElectionChangeNamespaceIT {
+
+  public static final String LEASE_NAME = "nschangelease";
+
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder()
+          .withConfigurationService(o -> o.withLeaderElectionConfiguration(
+              new LeaderElectionConfiguration(LEASE_NAME)))
+          .withReconciler(new LeaderElectionChangeNamespaceReconciler())
+          .build();
+
+  private static KubernetesClient client = new KubernetesClientBuilder().build();
+
+  @BeforeAll
+  static void createLeaseManually() {
+    client.resource(lease()).create();
+  }
+
+  @AfterAll
+  static void deleteLeaseManually() {
+    client.resource(lease()).delete();
+  }
+
+  @Test
+  @DisplayName("If operator is not a leader, namespace change should not start processor")
+  void noReconcileOnChangeNamespace() {
+    extension.create(testResource());
+
+    var reconciler = extension.getReconcilerOfType(LeaderElectionChangeNamespaceReconciler.class);
+    await().pollDelay(Duration.ofSeconds(1))
+        .timeout(Duration.ofSeconds(3))
+        .untilAsserted(() -> {
+          assertThat(reconciler.getNumberOfExecutions()).isEqualTo(0);
+        });
+
+    extension.getRegisteredControllerForReconcile(LeaderElectionChangeNamespaceReconciler.class)
+        .changeNamespaces("default", extension.getNamespace());
+
+    await().pollDelay(Duration.ofSeconds(1))
+        .timeout(Duration.ofSeconds(3))
+        .untilAsserted(() -> {
+          assertThat(reconciler.getNumberOfExecutions()).isEqualTo(0);
+        });
+  }
+
+
+  LeaderElectionChangeNamespaceCustomResource testResource() {
+    var resource = new LeaderElectionChangeNamespaceCustomResource();
+    resource.setMetadata(new ObjectMetaBuilder()
+        .withName("test1")
+        .build());
+    return resource;
+  }
+
+  static Lease lease() {
+    var lease = new Lease();
+    lease.setMetadata(new ObjectMetaBuilder()
+        .withName(LEASE_NAME)
+        .withNamespace("default")
+        .build());
+    var time = ZonedDateTime.now();
+    lease.setSpec(new LeaseSpecBuilder()
+        .withAcquireTime(ZonedDateTime.now())
+        .withRenewTime(time)
+        .withAcquireTime(time)
+        .withHolderIdentity("non-operator-identity")
+        .withLeaseTransitions(0)
+        .withLeaseDurationSeconds(30)
+        .build());
+
+    return lease;
+  }
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/leaderelectionchangenamespace/LeaderElectionChangeNamespaceCustomResource.java

@@ -0,0 +1,15 @@
+package io.javaoperatorsdk.operator.sample.leaderelectionchangenamespace;
+
+import io.fabric8.kubernetes.api.model.Namespaced;
+import io.fabric8.kubernetes.client.CustomResource;
+import io.fabric8.kubernetes.model.annotation.Group;
+import io.fabric8.kubernetes.model.annotation.ShortNames;
+import io.fabric8.kubernetes.model.annotation.Version;
+
+@Group("sample.javaoperatorsdk")
+@Version("v1")
+@ShortNames("lcn")
+public class LeaderElectionChangeNamespaceCustomResource
+    extends CustomResource<Void, Void>
+    implements Namespaced {
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/leaderelectionchangenamespace/LeaderElectionChangeNamespaceReconciler.java

@@ -0,0 +1,29 @@
+package io.javaoperatorsdk.operator.sample.leaderelectionchangenamespace;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider;
+
+@ControllerConfiguration()
+public class LeaderElectionChangeNamespaceReconciler
+    implements Reconciler<LeaderElectionChangeNamespaceCustomResource>, TestExecutionInfoProvider {
+
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  @Override
+  public UpdateControl<LeaderElectionChangeNamespaceCustomResource> reconcile(
+      LeaderElectionChangeNamespaceCustomResource resource,
+      Context<LeaderElectionChangeNamespaceCustomResource> context) {
+    numberOfExecutions.addAndGet(1);
+    return UpdateControl.noUpdate();
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+}