feat: retrieve secondary resources from Context

Author: metacosm

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

@@ -1,13 +1,16 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
+import java.util.Optional;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry;
-import java.util.Optional;
 
 public interface Context<P extends HasMetadata> {
 
   Optional<RetryInfo> getRetryInfo();
 
   EventSourceRegistry<P> getEventSourceRegistry();
 
+  <T extends HasMetadata> T getSecondaryResourceAssociatedWith(Class<T> expectedType,
+      String... qualifier);
 }

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

@@ -1,18 +1,22 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
+import java.util.Optional;
+
 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.EventSourceRegistry;
-import java.util.Optional;
 
 public class DefaultContext<P extends HasMetadata> implements Context<P> {
 
   private final RetryInfo retryInfo;
-  private final EventSourceRegistry<P> registry;
+  private final Controller<P> controller;
+  private final P primaryResource;
 
-  public DefaultContext(RetryInfo retryInfo,
-      EventSourceRegistry<P> registry) {
+  public DefaultContext(RetryInfo retryInfo, Controller<P> controller, P primaryResource) {
     this.retryInfo = retryInfo;
-    this.registry = registry;
+    this.controller = controller;
+    this.primaryResource = primaryResource;
   }
 
   @Override
@@ -22,6 +26,15 @@ public Optional<RetryInfo> getRetryInfo() {
 
   @Override
   public EventSourceRegistry<P> getEventSourceRegistry() {
-    return registry;
+    return controller.getEventSourceRegistry();
+  }
+
+  @Override
+  public <T extends HasMetadata> T getSecondaryResourceAssociatedWith(Class<T> expectedType,
+      String... qualifier) {
+    return getEventSourceRegistry()
+        .getResourceEventSourceFor(expectedType, qualifier)
+        .getResourceCache()
+        .get(ResourceID.fromResource(primaryResource)).orElse(null);
   }
 }

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

@@ -45,7 +45,7 @@ public Controller(Reconciler<R> reconciler,
   }
 
   @Override
-  public DeleteControl cleanup(R resource, Context context) {
+  public DeleteControl cleanup(R resource, Context<R> context) {
     return configuration.getConfigurationService().getMetrics().timeControllerExecution(
         new ControllerExecution<>() {
           @Override
@@ -71,7 +71,7 @@ public DeleteControl execute() {
   }
 
   @Override
-  public UpdateControl<R> reconcile(R resource, Context context) {
+  public UpdateControl<R> reconcile(R resource, Context<R> context) {
     final var metrics = configuration.getConfigurationService().getMetrics();
 
     configuration.getDependentResources().forEach(dependent -> {

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

@@ -17,6 +17,7 @@
 import io.javaoperatorsdk.operator.processing.event.source.ControllerResourceEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.TimerEventSource;
 
 public class EventSourceManager<R extends HasMetadata>
@@ -25,7 +26,7 @@ public class EventSourceManager<R extends HasMetadata>
   private static final Logger log = LoggerFactory.getLogger(EventSourceManager.class);
 
   private final ReentrantLock lock = new ReentrantLock();
-  private final ConcurrentNavigableMap<String, EventSource> eventSources =
+  private final ConcurrentNavigableMap<String, EventSource<R>> eventSources =
       new ConcurrentSkipListMap<>();
   private final EventProcessor<R> eventProcessor;
   private TimerEventSource<R> retryAndRescheduleTimerEventSource;
@@ -110,16 +111,36 @@ public final void registerEventSource(EventSource eventSource)
   }
 
   private String keyFor(EventSource source) {
-    var name = source.getClass().getName();
+    String name;
+    if (source instanceof ResourceEventSource) {
+      ResourceEventSource resourceEventSource = (ResourceEventSource) source;
+      final var configuration = resourceEventSource.getConfiguration();
+      // todo: extract qualifier from configuration
+      name = keyFor(configuration.getResourceClass());
+    } else {
+      name = keyFor(source.getResourceClass());
+    }
+    return name;
+  }
+
+  private String keyFor(Class<?> dependentType, String... qualifier) {
+    final var className = dependentType.getCanonicalName();
+    var key = className;
+    if (qualifier != null && qualifier.length > 0) {
+      key += "-" + qualifier[0];
+    }
+
     // make sure we process controller and timer event sources first
     // this is needed so that these sources are set when informer sources start so that events can
     // properly be processed
-    if (source == controllerResourceEventSource) {
-      name = 0 + name;
-    } else if (source == retryAndRescheduleTimerEventSource) {
-      name = 1 + name;
+    if (controllerResourceEventSource != null
+        && className.equals(controllerResourceEventSource.getResourceClass().getCanonicalName())) {
+      key = 0 + key;
+    } else if (retryAndRescheduleTimerEventSource != null && className
+        .equals(retryAndRescheduleTimerEventSource.getResourceClass().getCanonicalName())) {
+      key = 1 + key;
     }
-    return name;
+    return key;
   }
 
   public void cleanupForCustomResource(ResourceID customResourceUid) {
@@ -134,7 +155,7 @@ public void cleanupForCustomResource(ResourceID customResourceUid) {
   }
 
   @Override
-  public Set<EventSource> getRegisteredEventSources() {
+  public Set<EventSource<R>> getRegisteredEventSources() {
     return Set.copyOf(eventSources.values());
   }
 
@@ -143,6 +164,38 @@ public ControllerResourceEventSource<R> getControllerResourceEventSource() {
     return controllerResourceEventSource;
   }
 
+  @Override
+  public <S extends HasMetadata> ResourceEventSource<S, R> getResourceEventSourceFor(
+      Class<S> dependentType,
+      String... qualifier) {
+    final var eventSource = eventSources.get(keyFor(dependentType, qualifier));
+    if (eventSource == null) {
+      return null;
+    }
+    if (!(eventSource instanceof ResourceEventSource)) {
+      throw new IllegalArgumentException(eventSource + " associated with "
+          + keyAsString(dependentType, qualifier) + " is not a "
+          + ResourceEventSource.class.getSimpleName());
+    }
+    final var source = (ResourceEventSource<S, R>) eventSource;
+    final var configuration = source.getConfiguration();
+    final var resourceClass = configuration.getResourceClass();
+    if (!resourceClass.isAssignableFrom(dependentType)) {
+      throw new IllegalArgumentException(eventSource + " associated with "
+          + keyAsString(dependentType, qualifier)
+          + " is handling " + resourceClass.getName() + " resources but asked for "
+          + dependentType.getName());
+    }
+    return (ResourceEventSource<S, R>) eventSource;
+  }
+
+  @SuppressWarnings("rawtypes")
+  private String keyAsString(Class dependentType, String... qualifier) {
+    return qualifier != null && qualifier.length > 0
+        ? "(" + dependentType.getName() + ", " + qualifier[0] + ")"
+        : dependentType.getName();
+  }
+
   TimerEventSource<R> retryEventSource() {
     return retryAndRescheduleTimerEventSource;
   }

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

@@ -67,8 +67,7 @@ private PostExecutionControl<R> handleDispatch(ExecutionScope<R> executionScope)
       return PostExecutionControl.defaultDispatch();
     }
 
-    Context<R> context =
-        new DefaultContext<>(executionScope.getRetryInfo(), controller.getEventSourceRegistry());
+    Context<R> context = new DefaultContext<>(executionScope.getRetryInfo(), controller, resource);
     if (markedForDeletion) {
       return handleCleanup(resource, context);
     } else {

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

@@ -7,6 +7,16 @@ public abstract class AbstractEventSource<P extends HasMetadata> implements Even
 
   private volatile EventHandler eventHandler;
   private volatile EventSourceRegistry<P> eventSourceRegistry;
+  private final Class<?> resourceClass;
+
+  protected AbstractEventSource(Class<?> resourceClass) {
+    this.resourceClass = resourceClass;
+  }
+
+  @Override
+  public Class<?> getResourceClass() {
+    return resourceClass;
+  }
 
   @Override
   public void setEventHandler(EventHandler eventHandler) {

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

@@ -38,13 +38,19 @@ public abstract class AbstractResourceEventSource<T extends HasMetadata, U exten
   public AbstractResourceEventSource(U configuration,
       MixedOperation<T, KubernetesResourceList<T>, Resource<T>> client, Cloner cloner,
       EventSourceRegistry<P> registry) {
+    super(configuration.getResourceClass());
     this.configuration = configuration;
     this.client = client;
     this.filter = initFilter(configuration);
     this.cloner = cloner;
     setEventRegistry(registry);
   }
 
+  @Override
+  public U getConfiguration() {
+    return configuration;
+  }
+
   protected abstract ResourceEventFilter<T, U> initFilter(U configuration);
 
   protected abstract V wrapEventSource(

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

@@ -19,4 +19,6 @@ public interface EventSource<P extends HasMetadata> extends LifecycleAware {
    * @param customResourceUid - id of custom resource
    */
   default void cleanupForResource(ResourceID customResourceUid) {}
+
+  Class<?> getResourceClass();
 }

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

@@ -18,8 +18,11 @@ public interface EventSourceRegistry<T extends HasMetadata> {
   void registerEventSource(EventSource<? extends HasMetadata> eventSource)
       throws IllegalStateException, OperatorException;
 
-  Set<EventSource> getRegisteredEventSources();
+  Set<EventSource<T>> getRegisteredEventSources();
 
   ControllerResourceEventSource<T> getControllerResourceEventSource();
 
+  <R extends HasMetadata> ResourceEventSource<R, T> getResourceEventSourceFor(
+      Class<R> dependentType,
+      String... qualifier);
 }

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

@@ -46,6 +46,7 @@ public InformerEventSource(SharedInformer<T> sharedInformer,
       AssociatedSecondaryRetriever<T, P> associatedWith,
       boolean skipUpdateEventPropagationIfNoChange,
       Cloner cloner) {
+    super(sharedInformer.getApiTypeClass());
     this.sharedInformer = sharedInformer;
     this.associatedPrimaries = associatedPrimaries;
     this.skipUpdateEventPropagationIfNoChange = skipUpdateEventPropagationIfNoChange;

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

@@ -1,8 +1,11 @@
 package io.javaoperatorsdk.operator.processing.event.source;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.ResourceConfiguration;
 
 public interface ResourceEventSource<T extends HasMetadata, P extends HasMetadata>
     extends EventSource<P> {
   ResourceCache<T> getResourceCache();
+
+  ResourceConfiguration<T, ? extends ResourceConfiguration> getConfiguration();
 }

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

@@ -20,6 +20,11 @@ public class TimerEventSource<R extends HasMetadata> extends AbstractEventSource
   private final AtomicBoolean running = new AtomicBoolean();
   private final Map<ResourceID, EventProducerTimeTask> onceTasks = new ConcurrentHashMap<>();
 
+  public TimerEventSource() {
+    // we need to return the source class so that it can be used to register the event source
+    super(TimerEventSource.class);
+  }
+
 
   public void scheduleOnce(R resource, long delay) {
     if (!running.get()) {

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

@@ -14,6 +14,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 class EventSourceManagerTest {
 
@@ -23,6 +24,7 @@ class EventSourceManagerTest {
   @Test
   public void registersEventSource() {
     EventSource eventSource = mock(EventSource.class);
+    when(eventSource.getResourceClass()).thenReturn(String.class);
 
     eventSourceManager.registerEventSource(eventSource);
 
@@ -35,7 +37,9 @@ public void registersEventSource() {
   @Test
   public void closeShouldCascadeToEventSources() throws IOException {
     EventSource eventSource = mock(EventSource.class);
+    when(eventSource.getResourceClass()).thenReturn(String.class);
     EventSource eventSource2 = mock(EventSource.class);
+    when(eventSource2.getResourceClass()).thenReturn(Object.class);
     eventSourceManager.registerEventSource(eventSource);
     eventSourceManager.registerEventSource(eventSource2);
 
@@ -48,7 +52,9 @@ public void closeShouldCascadeToEventSources() throws IOException {
   @Test
   public void startCascadesToEventSources() {
     EventSource eventSource = mock(EventSource.class);
+    when(eventSource.getResourceClass()).thenReturn(String.class);
     EventSource eventSource2 = mock(EventSource.class);
+    when(eventSource2.getResourceClass()).thenReturn(Object.class);
     eventSourceManager.registerEventSource(eventSource);
     eventSourceManager.registerEventSource(eventSource2);
 
@@ -62,6 +68,7 @@ public void startCascadesToEventSources() {
   public void deRegistersEventSources() {
     CustomResource customResource = TestUtils.testCustomResource();
     EventSource eventSource = mock(EventSource.class);
+    when(eventSource.getResourceClass()).thenReturn(String.class);
     eventSourceManager.registerEventSource(eventSource);
 
     eventSourceManager

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

@@ -274,7 +274,7 @@ public boolean isLastAttempt() {
         ArgumentCaptor.forClass(Context.class);
     verify(reconciler, times(1))
         .reconcile(any(), contextArgumentCaptor.capture());
-    Context context = contextArgumentCaptor.getValue();
+    Context<?> context = contextArgumentCaptor.getValue();
     final var retryInfo = context.getRetryInfo().get();
     assertThat(retryInfo.getAttemptCount()).isEqualTo(2);
     assertThat(retryInfo.isLastAttempt()).isEqualTo(true);

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/errorstatushandler/ErrorStatusHandlerTestReconciler.java

@@ -22,7 +22,8 @@ public class ErrorStatusHandlerTestReconciler
 
   @Override
   public UpdateControl<ErrorStatusHandlerTestCustomResource> reconcile(
-      ErrorStatusHandlerTestCustomResource resource, Context context) {
+      ErrorStatusHandlerTestCustomResource resource,
+      Context<ErrorStatusHandlerTestCustomResource> context) {
     var number = numberOfExecutions.addAndGet(1);
     var retryAttempt = -1;
     if (context.getRetryInfo().isPresent()) {

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomReconciler.java

@@ -32,7 +32,7 @@ public RetryTestCustomReconciler(int numberOfExecutionFails) {
 
   @Override
   public UpdateControl<RetryTestCustomResource> reconcile(
-      RetryTestCustomResource resource, Context context) {
+      RetryTestCustomResource resource, Context<RetryTestCustomResource> context) {
     numberOfExecutions.addAndGet(1);
 
     if (!resource.getMetadata().getFinalizers().contains(FINALIZER_NAME)) {

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

@@ -32,7 +32,8 @@
 import okhttp3.Response;
 
 @ControllerConfiguration(
-    dependents = @DependentResourceConfiguration(creatable = false, resourceType = Tomcat.class,
+    dependents = @DependentResourceConfiguration(
+        creatable = false, resourceType = Tomcat.class,
         associatedPrimariesRetriever = WebappRetriever.class))
 public class WebappReconciler implements Reconciler<Webapp> {
 
@@ -64,15 +65,13 @@ public Set<ResourceID> associatedPrimaryResources(Tomcat t,
    * change.
    */
   @Override
-  public UpdateControl<Webapp> reconcile(Webapp webapp, Context context) {
+  public UpdateControl<Webapp> reconcile(Webapp webapp, Context<Webapp> context) {
     if (webapp.getStatus() != null
         && Objects.equals(webapp.getSpec().getUrl(), webapp.getStatus().getDeployedArtifact())) {
       return UpdateControl.noUpdate();
     }
 
-    var tomcatClient = kubernetesClient.customResources(Tomcat.class);
-    Tomcat tomcat = tomcatClient.inNamespace(webapp.getMetadata().getNamespace())
-        .withName(webapp.getSpec().getTomcat()).get();
+    Tomcat tomcat = context.getSecondaryResourceAssociatedWith(Tomcat.class);
     if (tomcat == null) {
       throw new IllegalStateException("Cannot find Tomcat " + webapp.getSpec().getTomcat()
           + " for Webapp " + webapp.getMetadata().getName() + " in namespace "