improve: thread-safe idempotent dynamic informer event source start (#2174)

Signed-off-by: Attila Mészáros <[email protected]>

Author: csviri

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

@@ -229,13 +229,18 @@ public <R> List<ResourceEventSource<R, P>> getResourceEventSourcesFor(Class<R> d
   }
 
   @Override
-  public synchronized EventSource dynamicallyRegisterEventSource(String name,
+  public EventSource dynamicallyRegisterEventSource(String name,
       EventSource eventSource) {
-    var es = eventSources.existing(name, eventSource);
-    if (es != null) {
-      return es;
+    synchronized (this) {
+      var actual = eventSources.existing(name, eventSource);
+      if (actual != null) {
+        eventSource = actual;
+      } else {
+        registerEventSource(name, eventSource);
+      }
     }
-    registerEventSource(name, eventSource);
+    // The start itself is blocking thus blocking only the threads which are attempt to start the
+    // actual event source. Think of this as a form of lock striping.
     eventSource.start();
     return eventSource;
   }

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

@@ -45,6 +45,7 @@ default <R> ResourceEventSource<R, P> getResourceEventSourceFor(Class<R> depende
    */
   EventSource dynamicallyRegisterEventSource(String name, EventSource eventSource);
 
+
   /**
    * De-registers (and stops) the {@link EventSource} associated with the specified name. If no such
    * source exists, this method will do nothing.

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSource.java

@@ -55,7 +55,7 @@ public ControllerResourceEventSource(Controller<T> controller) {
   }
 
   @Override
-  public void start() {
+  public synchronized void start() {
     try {
       super.start();
     } catch (KubernetesClientException e) {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java

@@ -76,7 +76,10 @@ public void changeNamespaces(Set<String> namespaces) {
   }
 
   @Override
-  public void start() {
+  public synchronized void start() {
+    if (isRunning()) {
+      return;
+    }
     temporaryResourceCache = new TemporaryResourceCache<>(this, parseResourceVersions);
     this.cache = new InformerManager<>(client, configuration, this);
     cache.setConfigurationService(configurationService);
@@ -86,7 +89,10 @@ public void start() {
   }
 
   @Override
-  public void stop() {
+  public synchronized void stop() {
+    if (!isRunning()) {
+      return;
+    }
     super.stop();
     manager().stop();
   }

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

@@ -0,0 +1,62 @@
+package io.javaoperatorsdk.operator;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.Secret;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.dynamicgenericeventsourceregistration.DynamicGenericEventSourceRegistrationCustomResource;
+import io.javaoperatorsdk.operator.sample.dynamicgenericeventsourceregistration.DynamicGenericEventSourceRegistrationReconciler;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class DynamicGenericEventSourceRegistrationIT {
+
+  public static final String TEST_RESOURCE_NAME = "test1";
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder()
+          .withReconciler(DynamicGenericEventSourceRegistrationReconciler.class)
+          .build();
+
+  @Test
+  void registersEventSourcesDynamically() {
+    var reconciler =
+        extension.getReconcilerOfType(DynamicGenericEventSourceRegistrationReconciler.class);
+    extension.create(testResource());
+
+    await().pollDelay(Duration.ofMillis(150)).untilAsserted(() -> {
+      var cm = extension.get(ConfigMap.class, TEST_RESOURCE_NAME);
+      var secret = extension.get(Secret.class, TEST_RESOURCE_NAME);
+      assertThat(cm).isNotNull();
+      assertThat(secret).isNotNull();
+    });
+    var executions = reconciler.getNumberOfExecutions();
+    assertThat(reconciler.getNumberOfEventSources()).isEqualTo(2);
+    assertThat(executions).isLessThanOrEqualTo(3);
+
+    var cm = extension.get(ConfigMap.class, TEST_RESOURCE_NAME);
+    cm.getData().put("key2", "val2");
+
+    extension.replace(cm); // triggers the reconciliation
+
+    await().untilAsserted(() -> {
+      assertThat(reconciler.getNumberOfExecutions() - executions).isEqualTo(2);
+    });
+    assertThat(reconciler.getNumberOfEventSources()).isEqualTo(2);
+  }
+
+
+  DynamicGenericEventSourceRegistrationCustomResource testResource() {
+    var res = new DynamicGenericEventSourceRegistrationCustomResource();
+    res.setMetadata(new ObjectMetaBuilder()
+        .withName(TEST_RESOURCE_NAME)
+        .build());
+    return res;
+  }
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/dynamicgenericeventsourceregistration/DynamicGenericEventSourceRegistrationCustomResource.java

@@ -0,0 +1,16 @@
+package io.javaoperatorsdk.operator.sample.dynamicgenericeventsourceregistration;
+
+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("dger")
+public class DynamicGenericEventSourceRegistrationCustomResource
+    extends CustomResource<Void, Void>
+    implements Namespaced {
+
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/dynamicgenericeventsourceregistration/DynamicGenericEventSourceRegistrationReconciler.java

@@ -0,0 +1,90 @@
+package io.javaoperatorsdk.operator.sample.dynamicgenericeventsourceregistration;
+
+import java.util.Base64;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.api.model.*;
+import io.fabric8.kubernetes.client.dsl.NonDeletingOperation;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
+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.processing.GroupVersionKind;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+
+@ControllerConfiguration
+public class DynamicGenericEventSourceRegistrationReconciler
+    implements Reconciler<DynamicGenericEventSourceRegistrationCustomResource> {
+
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+  private final AtomicInteger numberOfEventSources = new AtomicInteger();
+
+  @Override
+  public UpdateControl<DynamicGenericEventSourceRegistrationCustomResource> reconcile(
+      DynamicGenericEventSourceRegistrationCustomResource primary,
+      Context<DynamicGenericEventSourceRegistrationCustomResource> context) {
+
+    numberOfExecutions.addAndGet(1);
+
+    context.eventSourceRetriever().dynamicallyRegisterEventSource(ConfigMap.class.getSimpleName(),
+        genericInformerFor(ConfigMap.class, context));
+    context.eventSourceRetriever().dynamicallyRegisterEventSource(Secret.class.getSimpleName(),
+        genericInformerFor(Secret.class, context));
+
+    context.getClient().resource(secret(primary)).createOr(NonDeletingOperation::update);
+    context.getClient().resource(configMap(primary)).createOr(NonDeletingOperation::update);
+
+    numberOfEventSources.set(context.eventSourceRetriever()
+        .getResourceEventSourcesFor(GenericKubernetesResource.class).size());
+
+    return UpdateControl.noUpdate();
+  }
+
+  private Secret secret(DynamicGenericEventSourceRegistrationCustomResource primary) {
+    var secret = new SecretBuilder()
+        .withMetadata(new ObjectMetaBuilder()
+            .withName(primary.getMetadata().getName())
+            .withNamespace(primary.getMetadata().getNamespace())
+            .build())
+        .withData(Map.of("key", Base64.getEncoder().encodeToString("val".getBytes())))
+        .build();
+    secret.addOwnerReference(primary);
+    return secret;
+  }
+
+  private ConfigMap configMap(DynamicGenericEventSourceRegistrationCustomResource primary) {
+    var cm = new ConfigMapBuilder()
+        .withMetadata(new ObjectMetaBuilder()
+            .withName(primary.getMetadata().getName())
+            .withNamespace(primary.getMetadata().getNamespace())
+            .build())
+        .withData(Map.of("key", "val"))
+        .build();
+    cm.addOwnerReference(primary);
+    return cm;
+  }
+
+  private GroupVersionKind gvkFor(Class<? extends HasMetadata> clazz) {
+    return new GroupVersionKind(HasMetadata.getApiVersion(clazz), HasMetadata.getKind(clazz));
+  }
+
+  private InformerEventSource<GenericKubernetesResource, DynamicGenericEventSourceRegistrationCustomResource> genericInformerFor(
+      Class<? extends HasMetadata> clazz,
+      Context<DynamicGenericEventSourceRegistrationCustomResource> context) {
+
+    return new InformerEventSource<>(
+        InformerConfiguration.from(gvkFor(clazz),
+            context.eventSourceRetriever().eventSourceContextForDynamicRegistration()).build(),
+        context.eventSourceRetriever().eventSourceContextForDynamicRegistration());
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+  public int getNumberOfEventSources() {
+    return numberOfEventSources.get();
+  }
+}