fix: init sources before starting, wait for controller to be started

Author: metacosm

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

@@ -34,6 +34,7 @@ public Operator(ConfigurationService configurationService) {
   public Operator(KubernetesClient kubernetesClient, ConfigurationService configurationService) {
     this.kubernetesClient = kubernetesClient;
     this.configurationService = configurationService;
+    ExecutorServiceManager.init(configurationService);
   }
 
   /** Adds a shutdown hook that automatically calls {@link #close()} when the app shuts down. */
@@ -85,7 +86,6 @@ public void start() {
       throw new OperatorException(error, e);
     }
 
-    ExecutorServiceManager.init(configurationService);
     controllers.start();
   }
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java

@@ -4,7 +4,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.Controller;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.DependentResourceEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry;
 
 public class DefaultContext<P extends HasMetadata> implements Context<P> {
@@ -32,9 +32,8 @@ public EventSourceRegistry<P> getEventSourceRegistry() {
   @Override
   public <T extends HasMetadata> T getSecondaryResource(Class<T> expectedType,
       String... qualifier) {
-    return getEventSourceRegistry()
-        .getResourceEventSourceFor(expectedType, qualifier)
-        .getResourceCache()
-        .get(ResourceID.fromResource(primaryResource)).orElse(null);
+    final var eventSource = (DependentResourceEventSource<T, P>) getEventSourceRegistry()
+        .getResourceEventSourceFor(expectedType, qualifier);
+    return eventSource == null ? null : eventSource.getAssociated(primaryResource);
   }
 }

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

@@ -1,6 +1,11 @@
 package io.javaoperatorsdk.operator.processing;
 
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.KubernetesResourceList;
@@ -31,21 +36,46 @@
 
 public class Controller<R extends HasMetadata> implements Reconciler<R>,
     LifecycleAware, EventSourceInitializer<R> {
+
+  private static final Logger log = LoggerFactory.getLogger(Controller.class);
   private final Reconciler<R> reconciler;
   private final ControllerConfiguration<R> configuration;
   private final KubernetesClient kubernetesClient;
-  private EventSourceManager<R> eventSourceManager;
+  private final EventSourceManager<R> eventSourceManager;
+  private final AtomicBoolean started = new AtomicBoolean(false);
 
   public Controller(Reconciler<R> reconciler,
       ControllerConfiguration<R> configuration,
       KubernetesClient kubernetesClient) {
     this.reconciler = reconciler;
     this.configuration = configuration;
     this.kubernetesClient = kubernetesClient;
+
+    eventSourceManager = new EventSourceManager<>(this);
+    prepareEventSources(eventSourceManager,
+        configuration.getConfigurationService().getResourceCloner());
+  }
+
+  private void waitUntilStarted() {
+    if (!started.get()) {
+      AtomicInteger count = new AtomicInteger(0);
+      final var waitTime = 50;
+      while (!started.get()) {
+        try {
+          count.getAndIncrement();
+          Thread.sleep(waitTime);
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+        }
+      }
+      log.info("Waited {}ms for controller '{}' to finish initializing", count.get() * waitTime,
+          configuration.getName());
+    }
   }
 
   @Override
   public DeleteControl cleanup(R resource, Context<R> context) {
+    waitUntilStarted();
     return configuration.getConfigurationService().getMetrics().timeControllerExecution(
         new ControllerExecution<>() {
           @Override
@@ -72,6 +102,7 @@ public DeleteControl execute() {
 
   @Override
   public UpdateControl<R> reconcile(R resource, Context<R> context) {
+    waitUntilStarted();
     final var metrics = configuration.getConfigurationService().getMetrics();
 
     configuration.getDependentResources().forEach(dependent -> {
@@ -154,10 +185,11 @@ protected ResourceEventFilter initFilter(ResourceConfiguration configuration) {
         @Override
         protected EventSourceWrapper wrapEventSource(
             FilterWatchListDeletable filteredBySelectorClient, Cloner cloner) {
-          final var source = new DependentResourceEventSource(filteredBySelectorClient, cloner,
+          final var dependent = new DependentResourceEventSource(filteredBySelectorClient, cloner,
               dependentConfiguration);
-          eventSourceRegistry.registerEventSource(source);
-          return source;
+          // make sure we're set to receive events
+          eventSourceRegistry.registerEventSource(dependent);
+          return dependent;
         }
       };
     });
@@ -235,15 +267,14 @@ public void start() throws OperatorException {
         CustomResourceUtils.assertCustomResource(resClass, crd);
       }
 
-      eventSourceManager = new EventSourceManager<>(this);
-      prepareEventSources(eventSourceManager, configurationService.getResourceCloner());
       if (failOnMissingCurrentNS()) {
         throw new OperatorException(
             "Controller '"
                 + controllerName
                 + "' is configured to watch the current namespace but it couldn't be inferred from the current configuration.");
       }
       eventSourceManager.start();
+      started.set(true);
     } catch (MissingCRDException e) {
       throwMissingCRDException(crdName, specVersion, controllerName);
     }
@@ -282,6 +313,7 @@ public void stop() {
     if (eventSourceManager != null) {
       eventSourceManager.stop();
     }
+    started.set(false);
   }
 
   public EventSourceRegistry<R> getEventSourceRegistry() {

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

@@ -41,10 +41,11 @@ public class EventSourceManager<R extends HasMetadata>
 
   public EventSourceManager(Controller<R> controller) {
     this.controller = controller;
+    initRetryEventSource();
+
     controllerResourceEventSource = new ControllerResourceEventSource<>(controller, this);
     this.eventProcessor = new EventProcessor<>(this);
     registerEventSource(controllerResourceEventSource);
-    initRetryEventSource();
   }
 
   private void initRetryEventSource() {
@@ -62,9 +63,10 @@ public void start() throws OperatorException {
     eventProcessor.start();
     lock.lock();
     try {
-      log.debug("Starting event sources.");
       for (var eventSource : eventSources.values()) {
         try {
+          log.debug("Starting source {} for {}", eventSource.getClass(),
+              eventSource.getResourceClass());
           eventSource.start();
         } catch (Exception e) {
           log.warn("Error starting {} -> {}", eventSource, e);
@@ -134,15 +136,15 @@ private String keyFor(Class<?> dependentType, String... qualifier) {
       key += "-" + qualifier[0];
     }
 
-    // make sure we process controller and timer event sources first
+    // make sure timer event source is started first, then controller event source
     // this is needed so that these sources are set when informer sources start so that events can
     // properly be processed
     if (controllerResourceEventSource != null
         && className.equals(controllerResourceEventSource.getResourceClass().getCanonicalName())) {
-      key = 0 + key;
+      key = 1 + key;
     } else if (retryAndRescheduleTimerEventSource != null && className
         .equals(retryAndRescheduleTimerEventSource.getResourceClass().getCanonicalName())) {
-      key = 1 + key;
+      key = 0 + key;
     }
     return key;
   }

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

@@ -23,7 +23,8 @@
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getVersion;
 
 public abstract class AbstractResourceEventSource<T extends HasMetadata, U extends ResourceConfiguration<T, U>, V extends EventSourceWrapper<T>, P extends HasMetadata>
-    extends AbstractEventSource<P> implements ResourceEventHandler<T>, ResourceEventSource<T, P> {
+    extends AbstractEventSource<P>
+    implements ResourceEventHandler<T> {
 
   private static final String ANY_NAMESPACE_MAP_KEY = "anyNamespace";
   private static final Logger log = LoggerFactory.getLogger(AbstractResourceEventSource.class);
@@ -33,7 +34,7 @@ public abstract class AbstractResourceEventSource<T extends HasMetadata, U exten
   private final U configuration;
   private final MixedOperation<T, KubernetesResourceList<T>, Resource<T>> client;
   private final Cloner cloner;
-  private ResourceCache<T> cache;
+  private final ResourceCache<T> cache;
 
   public AbstractResourceEventSource(U configuration,
       MixedOperation<T, KubernetesResourceList<T>, Resource<T>> client, Cloner cloner,
@@ -44,9 +45,12 @@ public AbstractResourceEventSource(U configuration,
     this.filter = initFilter(configuration);
     this.cloner = cloner;
     setEventRegistry(registry);
+
+    initSources();
+
+    this.cache = new AggregateResourceCache<>(sources);
   }
 
-  @Override
   public U getConfiguration() {
     return configuration;
   }
@@ -86,6 +90,10 @@ public void onDelete(T resource, boolean b) {
 
   @Override
   public void start() throws OperatorException {
+    sources.values().parallelStream().forEach(LifecycleAware::start);
+  }
+
+  private void initSources() {
     final var targetNamespaces = configuration.getEffectiveNamespaces();
     final var labelSelector = configuration.getLabelSelector();
 
@@ -103,11 +111,6 @@ public void start() throws OperatorException {
                 ns);
           });
     }
-    this.cache = new AggregateResourceCache<>(sources);
-
-    // start sources only after the cache is populated so that events don't start coming in before
-    // it's ready
-    sources.values().parallelStream().forEach(LifecycleAware::start);
   }
 
 

sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/WebappReconciler.java

@@ -25,16 +25,19 @@
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceConfiguration;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.AssociatedSecondaryRetriever;
 import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry;
 import io.javaoperatorsdk.operator.processing.event.source.PrimaryResourcesRetriever;
+import io.javaoperatorsdk.operator.sample.WebappReconciler.TomcatRetriever;
 import io.javaoperatorsdk.operator.sample.WebappReconciler.WebappRetriever;
 
 import okhttp3.Response;
 
 @ControllerConfiguration(
     dependents = @DependentResourceConfiguration(
         creatable = false, resourceType = Tomcat.class,
-        associatedPrimariesRetriever = WebappRetriever.class))
+        associatedPrimariesRetriever = WebappRetriever.class,
+        associatedSecondaryRetriever = TomcatRetriever.class))
 public class WebappReconciler implements Reconciler<Webapp> {
 
   private KubernetesClient kubernetesClient;
@@ -60,6 +63,16 @@ public Set<ResourceID> associatedPrimaryResources(Tomcat t,
     }
   }
 
+  public static class TomcatRetriever implements AssociatedSecondaryRetriever<Tomcat, Webapp> {
+
+    @Override
+    public Tomcat associatedSecondary(Webapp primary, EventSourceRegistry<Webapp> registry) {
+      return registry.getResourceEventSourceFor(Tomcat.class).getResourceCache()
+          .get(new ResourceID(primary.getSpec().getTomcat(), primary.getMetadata().getNamespace()))
+          .orElse(null);
+    }
+  }
+
   /**
    * This method will be called not only on changes to Webapp objects but also when Tomcat objects
    * change.