refactor: rename internal package to source, moving LifecycleAware (#716)

Co-authored-by: Chris Laprun <[email protected]>

Author: csviri

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

@@ -13,12 +13,12 @@
 import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.Version;
-import io.javaoperatorsdk.operator.api.LifecycleAware;
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.processing.Controller;
+import io.javaoperatorsdk.operator.processing.LifecycleAware;
 
 @SuppressWarnings("rawtypes")
 public class Operator implements AutoCloseable, LifecycleAware {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfiguration.java

@@ -7,8 +7,8 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.ControllerUtils;
-import io.javaoperatorsdk.operator.processing.event.internal.ResourceEventFilter;
-import io.javaoperatorsdk.operator.processing.event.internal.ResourceEventFilters;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilter;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilters;
 
 public interface ControllerConfiguration<R extends HasMetadata> {
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java

@@ -5,7 +5,7 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.processing.event.internal.ResourceEventFilter;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilter;
 
 public class ControllerConfigurationOverrider<R extends HasMetadata> {
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java

@@ -4,7 +4,7 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.processing.event.internal.ResourceEventFilter;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilter;
 
 public class DefaultControllerConfiguration<R extends HasMetadata>
     implements ControllerConfiguration<R> {

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

@@ -5,7 +5,7 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
-import io.javaoperatorsdk.operator.processing.event.internal.ResourceEventFilter;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilter;
 
 @Retention(RetentionPolicy.RUNTIME)
 @Target({ElementType.TYPE})

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

@@ -1,7 +1,7 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.processing.event.EventSourceRegistry;
+import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry;
 
 public interface EventSourceInitializer<T extends HasMetadata> {
 

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

@@ -11,7 +11,6 @@
 import io.javaoperatorsdk.operator.CustomResourceUtils;
 import io.javaoperatorsdk.operator.MissingCRDException;
 import io.javaoperatorsdk.operator.OperatorException;
-import io.javaoperatorsdk.operator.api.LifecycleAware;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics.ControllerExecution;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
@@ -20,15 +19,14 @@
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
-import io.javaoperatorsdk.operator.processing.event.EventSourceRegistry;
+import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry;
 
 public class Controller<R extends HasMetadata> implements Reconciler<R>,
     LifecycleAware, EventSourceInitializer<R> {
   private final Reconciler<R> reconciler;
   private final ControllerConfiguration<R> configuration;
   private final KubernetesClient kubernetesClient;
   private EventSourceManager<R> eventSourceManager;
-  private EventProcessor<R> eventProcessor;
 
   public Controller(Reconciler<R> reconciler,
       ControllerConfiguration<R> configuration,
@@ -170,10 +168,6 @@ public void start() throws OperatorException {
       }
 
       eventSourceManager = new EventSourceManager<>(this);
-      eventProcessor =
-          new EventProcessor<>(this, eventSourceManager.getControllerResourceEventSource());
-      eventProcessor.setEventSourceManager(eventSourceManager);
-      eventSourceManager.setEventProcessor(eventProcessor);
       if (reconciler instanceof EventSourceInitializer) {
         ((EventSourceInitializer<R>) reconciler).prepareEventSources(eventSourceManager);
       }
@@ -183,7 +177,6 @@ public void start() throws OperatorException {
                 + controllerName
                 + "' is configured to watch the current namespace but it couldn't be inferred from the current configuration.");
       }
-      eventProcessor.start();
       eventSourceManager.start();
     } catch (MissingCRDException e) {
       throwMissingCRDException(crdName, specVersion, controllerName);
@@ -223,8 +216,5 @@ public void stop() {
     if (eventSourceManager != null) {
       eventSourceManager.stop();
     }
-    if (eventProcessor != null) {
-      eventProcessor.stop();
-    }
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/LifecycleAware.java renamed to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/LifecycleAware.java

@@ -1,4 +1,4 @@
-package io.javaoperatorsdk.operator.api;
+package io.javaoperatorsdk.operator.processing;
 
 import io.javaoperatorsdk.operator.OperatorException;
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/EventMarker.java renamed to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventMarker.java

@@ -1,10 +1,7 @@
-package io.javaoperatorsdk.operator.processing;
+package io.javaoperatorsdk.operator.processing.event;
 
 import java.util.HashMap;
 
-import io.javaoperatorsdk.operator.processing.event.Event;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
-
 /**
  * Manages the state of received events. Basically there can be only three distinct states relevant
  * for event processing. Either an event is received, so we eventually process or no event for
@@ -13,7 +10,7 @@
  * events are irrelevant for us from this point. Note that the dependant resources are either
  * cleaned up by K8S garbage collection or by the controller implementation for cleanup.
  */
-public class EventMarker {
+class EventMarker {
 
   public enum EventingState {
     /** Event but NOT Delete event present */

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/EventProcessor.java renamed to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java

@@ -1,4 +1,4 @@
-package io.javaoperatorsdk.operator.processing;
+package io.javaoperatorsdk.operator.processing.event;
 
 import java.util.HashMap;
 import java.util.HashSet;
@@ -14,17 +14,16 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.OperatorException;
-import io.javaoperatorsdk.operator.api.LifecycleAware;
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics;
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
-import io.javaoperatorsdk.operator.processing.event.Event;
-import io.javaoperatorsdk.operator.processing.event.EventHandler;
-import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
-import io.javaoperatorsdk.operator.processing.event.internal.ResourceAction;
-import io.javaoperatorsdk.operator.processing.event.internal.ResourceEvent;
+import io.javaoperatorsdk.operator.processing.LifecycleAware;
+import io.javaoperatorsdk.operator.processing.MDCUtils;
+import io.javaoperatorsdk.operator.processing.ResourceCache;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceAction;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEvent;
+import io.javaoperatorsdk.operator.processing.event.source.TimerEventSource;
 import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
 import io.javaoperatorsdk.operator.processing.retry.Retry;
 import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
@@ -36,8 +35,7 @@
  * Event handler that makes sure that events are processed in a "single threaded" way per resource
  * UID, while buffering events which are received during an execution.
  */
-public class EventProcessor<R extends HasMetadata>
-    implements EventHandler, LifecycleAware {
+class EventProcessor<R extends HasMetadata> implements EventHandler, LifecycleAware {
 
   private static final Logger log = LoggerFactory.getLogger(EventProcessor.class);
 
@@ -51,32 +49,34 @@ public class EventProcessor<R extends HasMetadata>
   private final Metrics metrics;
   private volatile boolean running;
   private final ResourceCache<R> resourceCache;
-  private EventSourceManager<R> eventSourceManager;
+  private final EventSourceManager<R> eventSourceManager;
   private final EventMarker eventMarker;
 
-  public EventProcessor(Controller<R> controller, ResourceCache<R> resourceCache) {
+  EventProcessor(EventSourceManager<R> eventSourceManager) {
     this(
-        resourceCache,
+        eventSourceManager.getControllerResourceEventSource(),
         ExecutorServiceManager.instance().executorService(),
-        controller.getConfiguration().getName(),
-        new ReconciliationDispatcher<>(controller),
-        GenericRetry.fromConfiguration(controller.getConfiguration().getRetryConfiguration()),
-        controller.getConfiguration().getConfigurationService().getMetrics(),
-        new EventMarker());
+        eventSourceManager.getController().getConfiguration().getName(),
+        new ReconciliationDispatcher<>(eventSourceManager.getController()),
+        GenericRetry.fromConfiguration(
+            eventSourceManager.getController().getConfiguration().getRetryConfiguration()),
+        eventSourceManager.getController().getConfiguration().getConfigurationService()
+            .getMetrics(),
+        eventSourceManager);
   }
 
   EventProcessor(ReconciliationDispatcher<R> reconciliationDispatcher,
-      ResourceCache<R> resourceCache,
+      EventSourceManager<R> eventSourceManager,
       String relatedControllerName,
-      Retry retry, EventMarker eventMarker) {
-    this(resourceCache, null, relatedControllerName, reconciliationDispatcher, retry, null,
-        eventMarker);
+      Retry retry) {
+    this(eventSourceManager.getControllerResourceEventSource(), null, relatedControllerName,
+        reconciliationDispatcher, retry, null, eventSourceManager);
   }
 
   private EventProcessor(ResourceCache<R> resourceCache, ExecutorService executor,
       String relatedControllerName,
       ReconciliationDispatcher<R> reconciliationDispatcher, Retry retry, Metrics metrics,
-      EventMarker eventMarker) {
+      EventSourceManager<R> eventSourceManager) {
     this.running = true;
     this.executor =
         executor == null
@@ -88,11 +88,12 @@ private EventProcessor(ResourceCache<R> resourceCache, ExecutorService executor,
     this.retry = retry;
     this.resourceCache = resourceCache;
     this.metrics = metrics != null ? metrics : Metrics.NOOP;
-    this.eventMarker = eventMarker;
+    this.eventMarker = new EventMarker();
+    this.eventSourceManager = eventSourceManager;
   }
 
-  public void setEventSourceManager(EventSourceManager<R> eventSourceManager) {
-    this.eventSourceManager = eventSourceManager;
+  EventMarker getEventMarker() {
+    return eventMarker;
   }
 
   @Override
@@ -243,9 +244,12 @@ private boolean isCacheReadyForInstantReconciliation(ExecutionScope<R> execution
 
   private void reScheduleExecutionIfInstructed(PostExecutionControl<R> postExecutionControl,
       R customResource) {
-    postExecutionControl.getReScheduleDelay().ifPresent(delay -> eventSourceManager
-        .getRetryAndRescheduleTimerEventSource()
-        .scheduleOnce(customResource, delay));
+    postExecutionControl.getReScheduleDelay()
+        .ifPresent(delay -> retryEventSource().scheduleOnce(customResource, delay));
+  }
+
+  TimerEventSource<R> retryEventSource() {
+    return eventSourceManager.retryEventSource();
   }
 
   /**
@@ -275,9 +279,7 @@ private void handleRetryOnException(ExecutionScope<R> executionScope,
               delay,
               customResourceID);
           metrics.failedReconciliation(customResourceID, exception);
-          eventSourceManager
-              .getRetryAndRescheduleTimerEventSource()
-              .scheduleOnce(executionScope.getResource(), delay);
+          retryEventSource().scheduleOnce(executionScope.getResource(), delay);
         },
         () -> log.error("Exhausted retries for {}", executionScope));
   }
@@ -289,9 +291,7 @@ private void cleanupOnSuccessfulExecution(ExecutionScope<R> executionScope) {
     if (isRetryConfigured()) {
       retryState.remove(executionScope.getCustomResourceID());
     }
-    eventSourceManager
-        .getRetryAndRescheduleTimerEventSource()
-        .cancelOnceSchedule(executionScope.getCustomResourceID());
+    retryEventSource().cancelOnceSchedule(executionScope.getCustomResourceID());
   }
 
   private RetryExecution getOrInitRetryExecution(ExecutionScope<R> executionScope) {