feat: expose event source metadata (#1617)

* feat: early initialization of event sources

* feat: expose event source metadata

* fix: properly return underlying EventSource when needed, add tests

* fix: remove use of deprecated method, add javadoc

* fix: remove dependent metadata on EventSourceMetadata

* fix: throw instead of return null (which shouldn't actually occur here)

* fix: do not expose EventSource on EventSourceMetadata

Author: metacosm

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

@@ -34,7 +34,7 @@ public Optional<RetryInfo> getRetryInfo() {
 
   @Override
   public <T> Set<T> getSecondaryResources(Class<T> expectedType) {
-    return controller.getEventSourceManager().getEventSourcesFor(expectedType).stream()
+    return controller.getEventSourceManager().getResourceEventSourcesFor(expectedType).stream()
         .map(es -> es.getSecondaryResources(primaryResource))
         .flatMap(Set::stream)
         .collect(Collectors.toSet());

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

@@ -83,9 +83,14 @@ public Controller(Reconciler<P> reconciler,
     isCleaner = reconciler instanceof Cleaner;
     managedWorkflow =
         ManagedWorkflow.workflowFor(kubernetesClient, configuration.getDependentResources());
+
     eventSourceManager = new EventSourceManager<>(this);
     eventProcessor = new EventProcessor<>(eventSourceManager);
     eventSourceManager.postProcessDefaultEventSourcesAfterProcessorInitializer();
+
+    final var context = new EventSourceContext<>(
+        eventSourceManager.getControllerResourceEventSource(), configuration, kubernetesClient);
+    initAndRegisterEventSources(context);
   }
 
   @Override
@@ -241,8 +246,7 @@ public void initAndRegisterEventSources(EventSourceContext<P> context) {
           .map(EventSourceReferencer.class::cast)
           .forEach(dr -> {
             try {
-              ((EventSourceReferencer<P>) dr)
-                  .resolveEventSource(eventSourceManager);
+              ((EventSourceReferencer<P>) dr).resolveEventSource(eventSourceManager);
             } catch (EventSourceNotFoundException e) {
               unresolvable.computeIfAbsent(e.getEventSourceName(), s -> new ArrayList<>()).add(dr);
             }
@@ -318,10 +322,6 @@ public synchronized void start(boolean startEventProcessor) throws OperatorExcep
     try {
       // check that the custom resource is known by the cluster if configured that way
       validateCRDWithLocalModelIfRequired(resClass, controllerName, crdName, specVersion);
-      final var context = new EventSourceContext<>(
-          eventSourceManager.getControllerResourceEventSource(), configuration, kubernetesClient);
-
-      initAndRegisterEventSources(context);
       eventSourceManager.start();
       if (startEventProcessor) {
         eventProcessor.start();

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

@@ -5,6 +5,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -126,8 +127,9 @@ public final synchronized void registerEventSource(String name, EventSource even
       if (name == null || name.isBlank()) {
         name = EventSourceInitializer.generateNameFor(eventSource);
       }
-      eventSources.add(name, eventSource);
-      eventSource.setEventHandler(controller.getEventProcessor());
+      final var named = new NamedEventSource(eventSource, name);
+      eventSources.add(named);
+      named.setEventHandler(controller.getEventProcessor());
     } catch (IllegalStateException | MissingCRDException e) {
       throw e; // leave untouched
     } catch (Exception e) {
@@ -138,22 +140,24 @@ public final synchronized void registerEventSource(String name, EventSource even
 
   @SuppressWarnings("unchecked")
   public void broadcastOnResourceEvent(ResourceAction action, P resource, P oldResource) {
-    eventSources.additionalNamedEventSources().forEach(eventSource -> {
-      if (eventSource.original() instanceof ResourceEventAware) {
-        var lifecycleAwareES = ((ResourceEventAware<P>) eventSource.original());
-        switch (action) {
-          case ADDED:
-            lifecycleAwareES.onResourceCreated(resource);
-            break;
-          case UPDATED:
-            lifecycleAwareES.onResourceUpdated(resource, oldResource);
-            break;
-          case DELETED:
-            lifecycleAwareES.onResourceDeleted(resource);
-            break;
-        }
-      }
-    });
+    eventSources.additionalNamedEventSources()
+        .map(NamedEventSource::original)
+        .forEach(source -> {
+          if (source instanceof ResourceEventAware) {
+            var lifecycleAwareES = ((ResourceEventAware<P>) source);
+            switch (action) {
+              case ADDED:
+                lifecycleAwareES.onResourceCreated(resource);
+                break;
+              case UPDATED:
+                lifecycleAwareES.onResourceUpdated(resource, oldResource);
+                break;
+              case DELETED:
+                lifecycleAwareES.onResourceDeleted(resource);
+                break;
+            }
+          }
+        });
   }
 
   public void changeNamespaces(Set<String> namespaces) {
@@ -174,6 +178,11 @@ public Set<EventSource> getRegisteredEventSources() {
         .collect(Collectors.toCollection(LinkedHashSet::new));
   }
 
+  @SuppressWarnings("unused")
+  public Stream<? extends EventSourceMetadata> getNamedEventSourcesStream() {
+    return eventSources.flatMappedSources();
+  }
+
   public ControllerResourceEventSource<P> getControllerResourceEventSource() {
     return eventSources.controllerResourceEventSource();
   }
@@ -187,9 +196,12 @@ public <R> List<ResourceEventSource<R, P>> getResourceEventSourcesFor(Class<R> d
     return eventSources.getEventSources(dependentType);
   }
 
+  /**
+   * @deprecated Use {@link #getResourceEventSourceFor(Class)} instead
+   */
   @Deprecated
   public <R> List<ResourceEventSource<R, P>> getEventSourcesFor(Class<R> dependentType) {
-    return eventSources.getEventSources(dependentType);
+    return getResourceEventSourcesFor(dependentType);
   }
 
   @Override

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

@@ -0,0 +1,13 @@
+package io.javaoperatorsdk.operator.processing.event;
+
+import java.util.Optional;
+
+public interface EventSourceMetadata {
+  String name();
+
+  Class<?> type();
+
+  Optional<Class<?>> resourceType();
+
+  Optional<?> configuration();
+}

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

@@ -1,6 +1,10 @@
 package io.javaoperatorsdk.operator.processing.event;
 
-import java.util.*;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.stream.Collectors;
@@ -20,15 +24,14 @@ class EventSources<R extends HasMetadata> {
   public static final String RETRY_RESCHEDULE_TIMER_EVENT_SOURCE_NAME =
       "RetryAndRescheduleTimerEventSource";
 
-  private final ConcurrentNavigableMap<String, Map<String, EventSource>> sources =
+  private final ConcurrentNavigableMap<String, Map<String, NamedEventSource>> sources =
       new ConcurrentSkipListMap<>();
   private final TimerEventSource<R> retryAndRescheduleTimerEventSource = new TimerEventSource<>();
   private ControllerResourceEventSource<R> controllerResourceEventSource;
 
 
-  ControllerResourceEventSource<R> initControllerEventSource(Controller<R> controller) {
+  void initControllerEventSource(Controller<R> controller) {
     controllerResourceEventSource = new ControllerResourceEventSource<>(controller);
-    return controllerResourceEventSource;
   }
 
   ControllerResourceEventSource<R> controllerResourceEventSource() {
@@ -49,7 +52,7 @@ public Stream<NamedEventSource> additionalNamedEventSources() {
   Stream<EventSource> additionalEventSources() {
     return Stream.concat(
         Stream.of(retryEventSource()).filter(Objects::nonNull),
-        sources.values().stream().flatMap(c -> c.values().stream()));
+        flatMappedSources().map(NamedEventSource::original));
   }
 
   NamedEventSource namedControllerResourceEventSource() {
@@ -58,29 +61,32 @@ NamedEventSource namedControllerResourceEventSource() {
   }
 
   Stream<NamedEventSource> flatMappedSources() {
-    return sources.values().stream().flatMap(c -> c.entrySet().stream()
-        .map(esEntry -> new NamedEventSource(esEntry.getValue(), esEntry.getKey())));
+    return sources.values().stream().flatMap(c -> c.values().stream());
   }
 
   public void clear() {
     sources.clear();
   }
 
-  public boolean contains(String name, EventSource source) {
+  private NamedEventSource existing(String name, EventSource source) {
     final var eventSources = sources.get(keyFor(source));
     if (eventSources == null || eventSources.isEmpty()) {
-      return false;
+      return null;
     }
-    return eventSources.containsKey(name);
+    return eventSources.get(name);
   }
 
-  public void add(String name, EventSource eventSource) {
-    if (contains(name, eventSource)) {
-      throw new IllegalArgumentException("An event source is already registered for the "
-          + keyAsString(getResourceType(eventSource), name)
+  public void add(NamedEventSource eventSource) {
+    final var name = eventSource.name();
+    final var original = eventSource.original();
+    final var existing = existing(name, original);
+    if (existing != null && !eventSource.equals(existing)) {
+      throw new IllegalArgumentException("Event source " + existing.original()
+          + " is already registered for the "
+          + keyAsString(getResourceType(original), name)
           + " class/name combination");
     }
-    sources.computeIfAbsent(keyFor(eventSource), k -> new HashMap<>()).put(name, eventSource);
+    sources.computeIfAbsent(keyFor(original), k -> new HashMap<>()).put(name, eventSource);
   }
 
   @SuppressWarnings("rawtypes")
@@ -91,6 +97,10 @@ private Class<?> getResourceType(EventSource source) {
   }
 
   private String keyFor(EventSource source) {
+    if (source instanceof NamedEventSource) {
+      source = ((NamedEventSource) source).original();
+    }
+
     return keyFor(getResourceType(source));
   }
 
@@ -100,16 +110,20 @@ private String keyFor(Class<?> dependentType) {
 
   @SuppressWarnings("unchecked")
   public <S> ResourceEventSource<S, R> get(Class<S> dependentType, String name) {
+    if (dependentType == null) {
+      throw new IllegalArgumentException("Must pass a dependent type to retrieve event sources");
+    }
+
     final var sourcesForType = sources.get(keyFor(dependentType));
     if (sourcesForType == null || sourcesForType.isEmpty()) {
       throw new IllegalArgumentException(
           "There is no event source found for class:" + dependentType.getName());
     }
 
     final var size = sourcesForType.size();
-    final EventSource source;
+    NamedEventSource source;
     if (size == 1 && name == null) {
-      source = sourcesForType.values().stream().findFirst().orElse(null);
+      source = sourcesForType.values().stream().findFirst().orElseThrow();
     } else {
       if (name == null || name.isBlank()) {
         throw new IllegalArgumentException("There are multiple EventSources registered for type "
@@ -125,15 +139,16 @@ public <S> ResourceEventSource<S, R> get(Class<S> dependentType, String name) {
       }
     }
 
-    if (!(source instanceof ResourceEventSource)) {
+    EventSource original = source.original();
+    if (!(original instanceof ResourceEventSource)) {
       throw new IllegalArgumentException(source + " associated with "
           + keyAsString(dependentType, name) + " is not a "
           + ResourceEventSource.class.getSimpleName());
     }
-    final var res = (ResourceEventSource<S, R>) source;
+    final var res = (ResourceEventSource<S, R>) original;
     final var resourceClass = res.resourceType();
     if (!resourceClass.isAssignableFrom(dependentType)) {
-      throw new IllegalArgumentException(source + " associated with "
+      throw new IllegalArgumentException(original + " associated with "
           + keyAsString(dependentType, name)
           + " is handling " + resourceClass.getName() + " resources but asked for "
           + dependentType.getName());
@@ -151,7 +166,12 @@ private String keyAsString(Class dependentType, String name) {
   @SuppressWarnings("unchecked")
   public <S> List<ResourceEventSource<S, R>> getEventSources(Class<S> dependentType) {
     final var sourcesForType = sources.get(keyFor(dependentType));
+    if (sourcesForType == null) {
+      return Collections.emptyList();
+    }
+
     return sourcesForType.values().stream()
+        .map(NamedEventSource::original)
         .filter(ResourceEventSource.class::isInstance)
         .map(es -> (ResourceEventSource<S, R>) es)
         .collect(Collectors.toList());

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

@@ -1,12 +1,15 @@
 package io.javaoperatorsdk.operator.processing.event;
 
 import java.util.Objects;
+import java.util.Optional;
 
 import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.processing.event.source.Configurable;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSourceStartPriority;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 
-class NamedEventSource implements EventSource {
+class NamedEventSource implements EventSource, EventSourceMetadata {
 
   private final EventSource original;
   private final String name;
@@ -35,6 +38,35 @@ public String name() {
     return name;
   }
 
+  @Override
+  public Class<?> type() {
+    return original.getClass();
+  }
+
+  @Override
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  public Optional<Class<?>> resourceType() {
+    if (original instanceof ResourceEventSource) {
+      ResourceEventSource resourceEventSource = (ResourceEventSource) original;
+      return Optional.of(resourceEventSource.resourceType());
+    }
+    return Optional.empty();
+  }
+
+  @Override
+  @SuppressWarnings("rawtypes")
+  public Optional<?> configuration() {
+    if (original instanceof Configurable) {
+      Configurable configurable = (Configurable) original;
+      return Optional.ofNullable(configurable.configuration());
+    }
+    return Optional.empty();
+  }
+
+  public EventSource eventSource() {
+    return original;
+  }
+
   @Override
   public String toString() {
     return original + " named: '" + name + "'}";

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

@@ -0,0 +1,5 @@
+package io.javaoperatorsdk.operator.processing.event.source;
+
+public interface Configurable<C> {
+  C configuration();
+}

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

@@ -1,6 +1,10 @@
 package io.javaoperatorsdk.operator.processing.event.source.informer;
 
-import java.util.*;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -72,6 +76,10 @@ void initSources(MixedOperation<T, KubernetesResourceList<T>, Resource<T>> clien
     }
   }
 
+  C configuration() {
+    return configuration;
+  }
+
   public void changeNamespaces(Set<String> namespaces) {
     var sourcesToRemove = sources.keySet().stream()
         .filter(k -> !namespaces.contains(k)).collect(Collectors.toSet());