improve: mamed event sources and equality of event sources and their implications

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

Author: csviri

caffeine-bounded-cache-support/src/test/java/io/javaoperatorsdk/operator/processing/event/source/cache/sample/AbstractTestReconciler.java

@@ -1,6 +1,7 @@
 package io.javaoperatorsdk.operator.processing.event.source.cache.sample;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 
 import org.slf4j.Logger;
@@ -69,7 +70,7 @@ protected void createConfigMap(P resource, Context<P> context) {
   }
 
   @Override
-  public Map<String, EventSource> prepareEventSources(
+  public List<EventSource> prepareEventSources(
       EventSourceContext<P> context) {
 
     var boundedItemStore =
@@ -82,7 +83,7 @@ public Map<String, EventSource> prepareEventSources(
             Mappers.fromOwnerReference(this instanceof BoundedCacheClusterScopeTestReconciler))
         .build(), context);
 
-    return EventSourceUtils.nameEventSources(es);
+    return List.of(es);
   }
 
   private void ensureStatus(P resource) {

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

@@ -5,10 +5,20 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.Workflow;
+import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
 
+// todo cleanup / delete
 public class EventSourceUtils {
+
+  @SuppressWarnings("unchecked")
+  public static  <R extends HasMetadata> List<EventSource> dependentEventSources(EventSourceContext<R> eventSourceContext,DependentResource... dependentResources) {
+      return Arrays.stream(dependentResources)
+              .flatMap(dr-> dr.eventSource(eventSourceContext).stream()).toList();
+  }
+
   /**
    * Utility method to easily create map with generated name for event sources. This is for the use
    * case when the event sources are not access explicitly by name in the reconciler.
@@ -25,15 +35,11 @@ public static Map<String, EventSource> nameEventSources(EventSource... eventSour
   }
 
   @SuppressWarnings("unchecked")
-  public static <K extends HasMetadata> Map<String, EventSource> eventSourcesFromWorkflow(
+  public static <K extends HasMetadata> List<EventSource> eventSourcesFromWorkflow(
       EventSourceContext<K> context,
       Workflow<K> workflow) {
-    Map<String, EventSource> result = new HashMap<>();
-    for (var e : workflow.getDependentResourcesByNameWithoutActivationCondition().entrySet()) {
-      var eventSource = e.getValue().eventSource(context);
-      eventSource.ifPresent(es -> result.put(e.getKey(), (EventSource) es));
-    }
-    return result;
+    return workflow.getDependentResourcesByNameWithoutActivationCondition().stream()
+            .flatMap(dr->dr.eventSource(context).stream()).toList();
   }
 
   @SuppressWarnings("rawtypes")

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

@@ -19,17 +19,17 @@ public interface Reconciler<P extends HasMetadata> {
    */
   UpdateControl<P> reconcile(P resource, Context<P> context) throws Exception;
 
-
   /**
    * Prepares a map of {@link EventSource} implementations keyed by the name with which they need to
    * be registered by the SDK.
    *
    * @param context a {@link EventSourceContext} providing access to information useful to event
    *        sources
-   * @return a map of event sources to register
+   * @return a list of event sources
    */
-  default Map<String, EventSource> prepareEventSources(EventSourceContext<P> context) {
-    return Map.of();
+  // todo should be ? extends EventSource
+  default List<EventSource> prepareEventSources(EventSourceContext<P> context) {
+    return Collections.emptyList();
   }
 
 }

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

@@ -49,6 +49,8 @@ default Optional<? extends ResourceEventSource<R, P>> eventSource(
     return Optional.empty();
   }
 
+
+
   /**
    * Retrieves the secondary resource (if it exists) associated with the specified primary resource
    * for this DependentResource.

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

@@ -238,20 +238,20 @@ public void initAndRegisterEventSources(EventSourceContext<P> context) {
         managedWorkflow.getDependentResourcesByNameWithoutActivationCondition();
     final var size = dependentResourcesByName.size();
     if (size > 0) {
-      dependentResourcesByName.forEach((key, dependentResource) -> {
+      dependentResourcesByName.forEach(dependentResource -> {
         if (dependentResource instanceof EventSourceProvider) {
           final var provider = (EventSourceProvider) dependentResource;
           final var source = provider.initEventSource(context);
-          eventSourceManager.registerEventSource(key, source);
+          eventSourceManager.registerEventSource(source);
         } else {
           Optional<ResourceEventSource> eventSource = dependentResource.eventSource(context);
-          eventSource.ifPresent(es -> eventSourceManager.registerEventSource(key, es));
+          eventSource.ifPresent(es -> eventSourceManager.registerEventSource(es));
         }
       });
 
       // resolve event sources referenced by name for dependents that reuse an existing event source
       final Map<String, List<EventSourceReferencer>> unresolvable = new HashMap<>(size);
-      dependentResourcesByName.values().stream()
+      dependentResourcesByName.stream()
           .filter(EventSourceReferencer.class::isInstance)
           .map(EventSourceReferencer.class::cast)
           .forEach(dr -> {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java

@@ -139,8 +139,7 @@ protected <R> void registerOrDeregisterEventSourceBasedOnActivation(
         var eventSource =
             dr.eventSource(eventSourceRetriever.eventSourceContextForDynamicRegistration());
         var es = eventSource.orElseThrow();
-        eventSourceRetriever.dynamicallyRegisterEventSource(dr.name(), es);
-
+        eventSourceRetriever.dynamicallyRegisterEventSource(es);
       } else {
         eventSourceRetriever.dynamicallyDeRegisterEventSource(dr.name());
       }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultWorkflow.java

@@ -148,14 +148,8 @@ public Map<String, DependentResource> getDependentResourcesByName() {
     return resources;
   }
 
-  public Map<String, DependentResource> getDependentResourcesByNameWithoutActivationCondition() {
-    final var resources = new HashMap<String, DependentResource>(dependentResourceNodes.size());
-    dependentResourceNodes
-        .forEach((name, node) -> {
-          if (node.getActivationCondition().isEmpty()) {
-            resources.put(name, node.getDependentResource());
-          }
-        });
-    return resources;
+  public List<DependentResource> getDependentResourcesByNameWithoutActivationCondition() {
+    return dependentResourceNodes.values().stream().filter(n->n.getActivationCondition().isEmpty())
+            .map(DependentResourceNode::getDependentResource).toList();
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/Workflow.java

@@ -1,6 +1,7 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -44,7 +45,7 @@ default Map<String, DependentResource> getDependentResourcesByName() {
   }
 
   @SuppressWarnings("rawtypes")
-  default Map<String, DependentResource> getDependentResourcesByNameWithoutActivationCondition() {
-    return Collections.emptyMap();
+  default List<DependentResource> getDependentResourcesByNameWithoutActivationCondition() {
+    return Collections.emptyList();
   }
 }

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

@@ -140,30 +140,23 @@ private Void stopEventSource(NamedEventSource eventSource) {
     return null;
   }
 
-  public final void registerEventSource(EventSource eventSource) throws OperatorException {
-    registerEventSource(null, eventSource);
-  }
-
   @SuppressWarnings("rawtypes")
-  public final synchronized void registerEventSource(String name, EventSource eventSource)
+  public final synchronized void registerEventSource(EventSource eventSource)
       throws OperatorException {
     Objects.requireNonNull(eventSource, "EventSource must not be null");
     try {
-      if (name == null || name.isBlank()) {
-        name = EventSourceUtils.generateNameFor(eventSource);
-      }
       if (eventSource instanceof ManagedInformerEventSource) {
         var managedInformerEventSource = ((ManagedInformerEventSource) eventSource);
         managedInformerEventSource.setConfigurationService(
             controller.getConfiguration().getConfigurationService());
       }
-      final var named = new NamedEventSource(eventSource, name);
+      final var named = new NamedEventSource(eventSource, eventSource.name());
       eventSources.add(named);
       named.setEventHandler(controller.getEventProcessor());
     } catch (IllegalStateException | MissingCRDException e) {
       throw e; // leave untouched
     } catch (Exception e) {
-      throw new OperatorException("Couldn't register event source: " + name + " for "
+      throw new OperatorException("Couldn't register event source: " + eventSource.name() + " for "
           + controller.getConfiguration().getName() + " controller", e);
     }
   }
@@ -229,14 +222,13 @@ public <R> List<ResourceEventSource<R, P>> getResourceEventSourcesFor(Class<R> d
   }
 
   @Override
-  public EventSource dynamicallyRegisterEventSource(String name,
-      EventSource eventSource) {
+  public EventSource dynamicallyRegisterEventSource(EventSource eventSource) {
     synchronized (this) {
-      var actual = eventSources.existing(name, eventSource);
+      var actual = eventSources.existing(eventSource.name(), eventSource);
       if (actual != null) {
         eventSource = actual.eventSource();
       } else {
-        registerEventSource(name, eventSource);
+        registerEventSource(eventSource);
       }
     }
     // The start itself is blocking thus blocking only the threads which are attempt to start the

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

@@ -39,11 +39,10 @@ default <R> ResourceEventSource<R, P> getResourceEventSourceFor(Class<R> depende
    * name will ever be registered.
    * </p>
    *
-   * @param name of the event source
    * @param eventSource to register
    * @return the actual event source registered. Might not be the same as the parameter.
    */
-  EventSource dynamicallyRegisterEventSource(String name, EventSource eventSource);
+  EventSource dynamicallyRegisterEventSource(EventSource eventSource);
 
 
   /**

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

@@ -87,7 +87,7 @@ 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)) {
+    if (existing != null && !eventSource.scopeEquals(existing)) {
       throw new IllegalArgumentException("Event source " + existing.original()
           + " is already registered for the "
           + keyAsString(getResourceType(original), name)

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

@@ -16,8 +16,15 @@ public abstract class AbstractResourceEventSource<R, P extends HasMetadata>
   protected OnDeleteFilter<? super R> onDeleteFilter;
   protected GenericFilter<? super R> genericFilter;
 
+  private final String name;
+
   protected AbstractResourceEventSource(Class<R> resourceClass) {
+    this(resourceClass,resourceClass.getName());
+  }
+
+  protected AbstractResourceEventSource(Class<R> resourceClass, String name) {
     this.resourceClass = resourceClass;
+    this.name = name == null ? resourceClass.getName() : name;
   }
 
   @Override
@@ -42,4 +49,9 @@ public void setOnDeleteFilter(
   public void setGenericFilter(GenericFilter<? super R> genericFilter) {
     this.genericFilter = genericFilter;
   }
+
+  @Override
+  public String name() {
+    return name;
+  }
 }

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

@@ -28,4 +28,14 @@ default EventSourceStartPriority priority() {
   default Status getStatus() {
     return Status.UNKNOWN;
   }
+
+  default String name() {
+    return getClass().getSimpleName();
+  }
+
+  // todo maybe different special method name?
+  default boolean scopeEquals(EventSource es) {
+    return this == es;
+  }
+
 }

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

@@ -43,8 +43,13 @@ public abstract class ExternalResourceCachingEventSource<R, P extends HasMetadat
   protected Map<ResourceID, Map<String, R>> cache = new ConcurrentHashMap<>();
 
   protected ExternalResourceCachingEventSource(Class<R> resourceClass,
+                                               CacheKeyMapper<R> cacheKeyMapper) {
+    this(null,resourceClass,cacheKeyMapper);
+  }
+
+  protected ExternalResourceCachingEventSource(String name,Class<R> resourceClass,
       CacheKeyMapper<R> cacheKeyMapper) {
-    super(resourceClass);
+    super(resourceClass,name);
     this.cacheKeyMapper = cacheKeyMapper;
   }
 

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

@@ -39,4 +39,10 @@ default Optional<R> getSecondaryResource(P primary) {
   void setOnDeleteFilter(OnDeleteFilter<? super R> onDeleteFilter);
 
   void setGenericFilter(GenericFilter<? super R> genericFilter);
+
+  @Override
+  default String name() {
+    return resourceType().getSimpleName();
+  }
+
 }

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

@@ -26,12 +26,13 @@ public class ControllerResourceEventSource<T extends HasMetadata>
     implements ResourceEventHandler<T> {
 
   private static final Logger log = LoggerFactory.getLogger(ControllerResourceEventSource.class);
+  public static final String NAME = "ControllerResourceEventSource";
 
   private final Controller<T> controller;
 
   @SuppressWarnings({"unchecked", "rawtypes"})
   public ControllerResourceEventSource(Controller<T> controller) {
-    super(controller.getCRClient(), controller.getConfiguration(), false);
+    super(NAME,controller.getCRClient(), controller.getConfiguration(), false);
     this.controller = controller;
 
     final var config = controller.getConfiguration();

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

@@ -76,20 +76,27 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
   private final PrimaryToSecondaryMapper<P> primaryToSecondaryMapper;
   private final String id = UUID.randomUUID().toString();
 
+  public InformerEventSource(String name,
+          InformerConfiguration<R> configuration, EventSourceContext<P> context) {
+    this(name,configuration, context.getClient(),
+            context.getControllerConfiguration().getConfigurationService()
+                    .parseResourceVersionsForEventFilteringAndCaching());
+  }
+
   public InformerEventSource(
       InformerConfiguration<R> configuration, EventSourceContext<P> context) {
-    this(configuration, context.getClient(),
+    this(null,configuration, context.getClient(),
         context.getControllerConfiguration().getConfigurationService()
             .parseResourceVersionsForEventFilteringAndCaching());
   }
 
   public InformerEventSource(InformerConfiguration<R> configuration, KubernetesClient client) {
-    this(configuration, client, false);
+    this(null,configuration, client, false);
   }
 
-  public InformerEventSource(InformerConfiguration<R> configuration, KubernetesClient client,
+  public InformerEventSource(String name,InformerConfiguration<R> configuration, KubernetesClient client,
       boolean parseResourceVersions) {
-    super(
+    super(name,
         configuration.getGroupVersionKind()
             .map(gvk -> client.genericKubernetesResources(gvk.apiVersion(), gvk.getKind()))
             .orElseGet(() -> (MixedOperation) client.resources(configuration.getResourceClass())),