fix: make sure event sources are properly registered

Author: metacosm

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

@@ -144,7 +144,8 @@ public void prepareEventSources(EventSourceRegistry<R> eventSourceRegistry, Clon
     configuration.getDependentResources().forEach(dependent -> {
       final var dependentConfiguration = dependent.getConfiguration();
       final var source = new AbstractResourceEventSource<>(dependentConfiguration,
-          kubernetesClient.resources(dependentConfiguration.getResourceClass()), cloner) {
+          kubernetesClient.resources(dependentConfiguration.getResourceClass()), cloner,
+          eventSourceRegistry) {
         @Override
         protected ResourceEventFilter initFilter(ResourceConfiguration configuration) {
           return configuration.getEventFilter();
@@ -156,7 +157,8 @@ protected EventSourceWrapper wrapEventSource(
           return new InformerEventSourceEventSourceWrapper(filteredBySelectorClient, cloner,
               dependentConfiguration.getPrimaryResourcesRetriever(),
               dependentConfiguration.getAssociatedResourceRetriever(),
-              dependentConfiguration.skipUpdateIfUnchanged());
+              dependentConfiguration.skipUpdateIfUnchanged(),
+              eventSourceRegistry);
         }
       };
       eventSourceRegistry.registerEventSource(source);

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

@@ -39,7 +39,7 @@ public class EventSourceManager<R extends HasMetadata>
 
   public EventSourceManager(Controller<R> controller) {
     this.controller = controller;
-    controllerResourceEventSource = new ControllerResourceEventSource<>(controller);
+    controllerResourceEventSource = new ControllerResourceEventSource<>(controller, this);
     this.eventProcessor = new EventProcessor<>(this);
     registerEventSource(controllerResourceEventSource);
     initRetryEventSource();

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

@@ -36,11 +36,13 @@ public abstract class AbstractResourceEventSource<T extends HasMetadata, U exten
   private ResourceCache<T> cache;
 
   public AbstractResourceEventSource(U configuration,
-      MixedOperation<T, KubernetesResourceList<T>, Resource<T>> client, Cloner cloner) {
+      MixedOperation<T, KubernetesResourceList<T>, Resource<T>> client, Cloner cloner,
+      EventSourceRegistry<P> registry) {
     this.configuration = configuration;
     this.client = client;
     this.filter = initFilter(configuration);
     this.cloner = cloner;
+    setEventRegistry(registry);
   }
 
   protected abstract ResourceEventFilter<T, U> initFilter(U configuration);

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

@@ -23,10 +23,11 @@ public class ControllerResourceEventSource<T extends HasMetadata>
   private final Controller<T> controller;
   private OnceWhitelistEventFilterEventFilter<T> onceWhitelistEventFilterEventFilter;
 
-  public ControllerResourceEventSource(Controller<T> controller) {
+  public ControllerResourceEventSource(Controller<T> controller, EventSourceRegistry<T> registry) {
     super(controller.getConfiguration(),
         controller.getCRClient(),
-        controller.getConfiguration().getConfigurationService().getResourceCloner());
+        controller.getConfiguration().getConfigurationService().getResourceCloner(),
+        registry);
     this.controller = controller;
   }
 

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

@@ -54,7 +54,7 @@ public InformerEventSource(SharedInformer<T> sharedInformer,
           "Informer is already running on event source creation, this is not desirable and may " +
               "lead to non deterministic behavior.");
     }
-    this.cache = new InformerResourceCache<T>(sharedInformer, cloner);
+    this.cache = new InformerResourceCache<>(sharedInformer, cloner);
 
     this.associatedWith = Objects.requireNonNullElseGet(associatedWith,
         () -> (cr, registry) -> cache.get(ResourceID.fromResource(cr)).orElse(null));

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

@@ -22,11 +22,12 @@ public InformerEventSourceEventSourceWrapper(
       FilterWatchListDeletable<T, KubernetesResourceList<T>> client, Cloner cloner,
       PrimaryResourcesRetriever<T, P> secondaryToPrimaryResourcesIdSet,
       AssociatedSecondaryRetriever<T, P> associatedWith,
-      boolean skipUpdateEventPropagationIfNoChange) {
+      boolean skipUpdateEventPropagationIfNoChange, EventSourceRegistry<P> registry) {
     final var informer = client.runnableInformer(0);
     source = new InformerEventSource<>(informer,
         secondaryToPrimaryResourcesIdSet,
         associatedWith, skipUpdateEventPropagationIfNoChange, cloner);
+    registry.registerEventSource(source);
     this.cache = new InformerResourceCache<>(informer, cloner);
   }
 

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

@@ -30,9 +30,10 @@ class ControllerResourceEventSourceTest {
   private static final MixedOperation<TestCustomResource, KubernetesResourceList<TestCustomResource>, Resource<TestCustomResource>> client =
       mock(MixedOperation.class);
   EventHandler eventHandler = mock(EventHandler.class);
+  EventSourceRegistry registry = mock(EventSourceRegistry.class);
 
   private ControllerResourceEventSource<TestCustomResource> controllerResourceEventSource =
-      new ControllerResourceEventSource<>(new TestController(true));
+      new ControllerResourceEventSource<>(new TestController(true), registry);
 
   @BeforeEach
   public void setup() {
@@ -89,7 +90,7 @@ public void normalExecutionIfGenerationChanges() {
   @Test
   public void handlesAllEventIfNotGenerationAware() {
     controllerResourceEventSource =
-        new ControllerResourceEventSource<>(new TestController(false));
+        new ControllerResourceEventSource<>(new TestController(false), registry);
     setup();
 
     TestCustomResource customResource1 = TestUtils.testCustomResource();

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

@@ -30,6 +30,7 @@ class ResourceEventFilterTest {
   public static final String FINALIZER = "finalizer";
 
   private EventHandler eventHandler;
+  private EventSourceRegistry registry = mock(EventSourceRegistry.class);
 
   @BeforeEach
   public void before() {
@@ -46,7 +47,7 @@ public void eventFilteredByCustomPredicate() {
             newResource.getStatus().getConfigMapStatus()));
 
     var controller = new TestController(config);
-    var eventSource = new ControllerResourceEventSource<>(controller);
+    var eventSource = new ControllerResourceEventSource<>(controller, registry);
     eventSource.setEventHandler(eventHandler);
 
     TestCustomResource cr = TestUtils.testCustomResource();
@@ -74,7 +75,7 @@ public void eventFilteredByCustomPredicateAndGenerationAware() {
             newResource.getStatus().getConfigMapStatus()));
 
     var controller = new TestController(config);
-    var eventSource = new ControllerResourceEventSource<>(controller);
+    var eventSource = new ControllerResourceEventSource<>(controller, registry);
     eventSource.setEventHandler(eventHandler);
 
     TestCustomResource cr = TestUtils.testCustomResource();
@@ -104,7 +105,7 @@ public void observedGenerationFiltering() {
         .thenReturn(ConfigurationService.DEFAULT_CLONER);
 
     var controller = new ObservedGenController(config);
-    var eventSource = new ControllerResourceEventSource<>(controller);
+    var eventSource = new ControllerResourceEventSource<>(controller, registry);
     eventSource.setEventHandler(eventHandler);
 
     ObservedGenCustomResource cr = new ObservedGenCustomResource();
@@ -135,7 +136,7 @@ public void eventNotFilteredByCustomPredicateIfFinalizerIsRequired() {
         .thenReturn(ConfigurationService.DEFAULT_CLONER);
 
     var controller = new TestController(config);
-    var eventSource = new ControllerResourceEventSource<>(controller);
+    var eventSource = new ControllerResourceEventSource<>(controller, registry);
     eventSource.setEventHandler(eventHandler);
 
     TestCustomResource cr = TestUtils.testCustomResource();

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

@@ -3,6 +3,7 @@
 import java.io.ByteArrayOutputStream;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -17,22 +18,23 @@
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.dsl.ExecListener;
 import io.fabric8.kubernetes.client.dsl.ExecWatch;
-import io.javaoperatorsdk.operator.api.config.Cloner;
-import io.fabric8.kubernetes.client.informers.cache.Cache;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
-import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 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.EventSourceRegistry;
-import io.javaoperatorsdk.operator.processing.event.source.InformerEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.PrimaryResourcesRetriever;
+import io.javaoperatorsdk.operator.sample.WebappReconciler.WebappRetriever;
 
 import okhttp3.Response;
 
-@ControllerConfiguration
-public class WebappReconciler implements Reconciler<Webapp>, EventSourceInitializer<Webapp> {
+@ControllerConfiguration(
+    dependents = @DependentResourceConfiguration(creatable = false, resourceType = Tomcat.class,
+        associatedPrimariesRetriever = WebappRetriever.class))
+public class WebappReconciler implements Reconciler<Webapp> {
 
   private KubernetesClient kubernetesClient;
 
@@ -42,22 +44,19 @@ public WebappReconciler(KubernetesClient kubernetesClient) {
     this.kubernetesClient = kubernetesClient;
   }
 
-  private InformerEventSource<Tomcat> tomcatEventSource;
-
-  @Override
-  public void prepareEventSources(EventSourceRegistry<Webapp> eventSourceRegistry, Cloner cloner) {
-    tomcatEventSource =
-        new InformerEventSource<>(kubernetesClient, Tomcat.class, (t, registry) -> {
-          // To create an event to a related WebApp resource and trigger the reconciliation
-          // we need to find which WebApp this Tomcat custom resource is related to.
-          // To find the related customResourceId of the WebApp resource we traverse the cache to
-          // and identify it based on naming convention.
-          return eventSourceRegistry.getControllerResourceEventSource().getResourceCache()
-              .list(webApp -> webApp.getSpec().getTomcat().equals(t.getMetadata().getName()))
-              .map(ResourceID::fromResource)
-              .collect(Collectors.toSet());
-        }, cloner);
-    eventSourceRegistry.registerEventSource(tomcatEventSource);
+  public static class WebappRetriever implements PrimaryResourcesRetriever<Tomcat, Webapp> {
+    @Override
+    public Set<ResourceID> associatedPrimaryResources(Tomcat t,
+        EventSourceRegistry<Webapp> registry) {
+      // To create an event to a related WebApp resource and trigger the reconciliation
+      // we need to find which WebApp this Tomcat custom resource is related to.
+      // To find the related customResourceId of the WebApp resource we traverse the cache to
+      // and identify it based on naming convention.
+      return registry.getControllerResourceEventSource().getResourceCache()
+          .list(webApp -> webApp.getSpec().getTomcat().equals(t.getMetadata().getName()))
+          .map(ResourceID::fromResource)
+          .collect(Collectors.toSet());
+    }
   }
 
   /**
@@ -71,9 +70,9 @@ public UpdateControl<Webapp> reconcile(Webapp webapp, Context context) {
       return UpdateControl.noUpdate();
     }
 
-    Tomcat tomcat = tomcatEventSource.getStore()
-        .getByKey(Cache.namespaceKeyFunc(webapp.getMetadata().getNamespace(),
-            webapp.getSpec().getTomcat()));
+    var tomcatClient = kubernetesClient.customResources(Tomcat.class);
+    Tomcat tomcat = tomcatClient.inNamespace(webapp.getMetadata().getNamespace())
+        .withName(webapp.getSpec().getTomcat()).get();
     if (tomcat == null) {
       throw new IllegalStateException("Cannot find Tomcat " + webapp.getSpec().getTomcat()
           + " for Webapp " + webapp.getMetadata().getName() + " in namespace "