Persistent state dependent resources

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

@@ -110,38 +110,38 @@ private void logForOperation(String operation, P primary, R desired) {
   }
 
   protected R handleCreate(R desired, P primary, Context<P> context) {
-    ResourceID resourceID = ResourceID.fromResource(primary);
     R created = creator.create(desired, primary, context);
     throwIfNull(created, primary, "Created resource");
-    onCreated(resourceID, created);
+    onCreated(primary, created, context);
     return created;
   }
 
   /**
    * Allows subclasses to perform additional processing (e.g. caching) on the created resource if
    * needed.
    *
-   * @param primaryResourceId the {@link ResourceID} of the primary resource associated with the
-   *        newly created resource
+   * @param primary the {@link ResourceID} of the primary resource associated with the newly created
+   *        resource
    * @param created the newly created resource
+   * @param context
    */
-  protected abstract void onCreated(ResourceID primaryResourceId, R created);
+  protected abstract void onCreated(P primary, R created, Context<P> context);
 
   /**
    * Allows subclasses to perform additional processing on the updated resource if needed.
    *
-   * @param primaryResourceId the {@link ResourceID} of the primary resource associated with the
-   *        newly updated resource
+   * @param primary the {@link ResourceID} of the primary resource associated with the newly updated
+   *        resource
    * @param updated the updated resource
    * @param actual the resource as it was before the update
+   * @param context
    */
-  protected abstract void onUpdated(ResourceID primaryResourceId, R updated, R actual);
+  protected abstract void onUpdated(P primary, R updated, R actual, Context<P> context);
 
   protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
-    ResourceID resourceID = ResourceID.fromResource(primary);
     R updated = updater.update(actual, desired, primary, context);
     throwIfNull(updated, primary, "Updated resource");
-    onUpdated(resourceID, updated, actual);
+    onUpdated(primary, updated, actual, context);
     return updated;
   }
 
@@ -154,8 +154,9 @@ public void delete(P primary, Context<P> context) {
     dependentResourceReconciler.delete(primary, context);
   }
 
-  protected void handleDelete(P primary, Context<P> context) {
-    throw new IllegalStateException("delete method be implemented if Deleter trait is supported");
+  protected void handleDelete(P primary, R secondary, Context<P> context) {
+    throw new IllegalStateException(
+        "handleDelete method must be implemented if Deleter trait is supported");
   }
 
   public void setResourceDiscriminator(

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

@@ -3,6 +3,7 @@
 import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceNotFoundException;
@@ -94,15 +95,17 @@ public Optional<ResourceEventSource<R, P>> eventSource() {
     return Optional.ofNullable(eventSource);
   }
 
-  protected void onCreated(ResourceID primaryResourceId, R created) {
+  protected void onCreated(P primary, R created, Context<P> context) {
     if (isCacheFillerEventSource) {
-      recentOperationCacheFiller().handleRecentResourceCreate(primaryResourceId, created);
+      recentOperationCacheFiller().handleRecentResourceCreate(ResourceID.fromResource(primary),
+          created);
     }
   }
 
-  protected void onUpdated(ResourceID primaryResourceId, R updated, R actual) {
+  protected void onUpdated(P primary, R updated, R actual, Context<P> context) {
     if (isCacheFillerEventSource) {
-      recentOperationCacheFiller().handleRecentResourceUpdate(primaryResourceId, updated, actual);
+      recentOperationCacheFiller().handleRecentResourceUpdate(ResourceID.fromResource(primary),
+          updated, actual);
     }
   }
 

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

@@ -0,0 +1,120 @@
+package io.javaoperatorsdk.operator.processing.dependent;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+
+public abstract class AbstractExternalDependentResource<R, P extends HasMetadata, T extends ResourceEventSource<R, P>>
+    extends AbstractEventSourceHolderDependentResource<R, P, T> {
+
+  private final boolean isDependentResourceWithExplicitState =
+      this instanceof DependentResourceWithExplicitState;
+  private final boolean isBulkDependentResource = this instanceof BulkDependentResource;
+  @SuppressWarnings("rawtypes")
+  private DependentResourceWithExplicitState dependentResourceWithExplicitState;
+  private InformerEventSource<?, P> externalStateEventSource;
+  private KubernetesClient kubernetesClient;
+
+  @SuppressWarnings("unchecked")
+  protected AbstractExternalDependentResource(Class<R> resourceType) {
+    super(resourceType);
+    if (isDependentResourceWithExplicitState) {
+      dependentResourceWithExplicitState = (DependentResourceWithExplicitState<R, P, ?>) this;
+    }
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public void resolveEventSource(EventSourceRetriever<P> eventSourceRetriever) {
+    super.resolveEventSource(eventSourceRetriever);
+    if (isDependentResourceWithExplicitState) {
+      externalStateEventSource = (InformerEventSource<?, P>) dependentResourceWithExplicitState
+          .eventSourceName()
+          .map(n -> eventSourceRetriever
+              .getResourceEventSourceFor(dependentResourceWithExplicitState.stateResourceClass(),
+                  (String) n))
+          .orElseGet(() -> eventSourceRetriever
+              .getResourceEventSourceFor(
+                  (Class<R>) dependentResourceWithExplicitState.stateResourceClass()));
+    }
+
+  }
+
+  @Override
+  protected void onCreated(P primary, R created, Context<P> context) {
+    super.onCreated(primary, created, context);
+    if (this instanceof DependentResourceWithExplicitState) {
+      handleExplicitStateCreation(primary, created, context);
+    }
+  }
+
+  @Override
+  public void delete(P primary, Context<P> context) {
+    if (isDependentResourceWithExplicitState && !isBulkDependentResource) {
+      var secondary = getSecondaryResource(primary, context);
+      super.delete(primary, context);
+      // deletes the state after the resource is deleted
+      handleExplicitStateDelete(primary, secondary.orElse(null), context);
+    } else {
+      super.delete(primary, context);
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private void handleExplicitStateDelete(P primary, R secondary, Context<P> context) {
+    var res = dependentResourceWithExplicitState.stateResource(primary, secondary);
+    dependentResourceWithExplicitState.getKubernetesClient().resource(res).delete();
+  }
+
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  protected void handleExplicitStateCreation(P primary, R created, Context<P> context) {
+    var resource = dependentResourceWithExplicitState.stateResource(primary, created);
+    var stateResource =
+        dependentResourceWithExplicitState.getKubernetesClient().resource(resource).create();
+    if (externalStateEventSource != null) {
+      ((RecentOperationCacheFiller) externalStateEventSource)
+          .handleRecentResourceCreate(ResourceID.fromResource(primary), stateResource);
+    }
+  }
+
+
+  @SuppressWarnings("unchecked")
+  public void deleteTargetResource(P primary, R resource, String key,
+      Context<P> context) {
+    if (isDependentResourceWithExplicitState) {
+      getKubernetesClient()
+          .resource(dependentResourceWithExplicitState.stateResource(primary, resource))
+          .delete();
+    }
+    handleDeleteTargetResource(primary, resource, key, context);
+  }
+
+  public void handleDeleteTargetResource(P primary, R resource, String key,
+      Context<P> context) {
+    throw new IllegalStateException("Override this method in case you manage an bulk resource");
+  }
+
+  @SuppressWarnings("rawtypes")
+  protected InformerEventSource getExternalStateEventSource() {
+    return externalStateEventSource;
+  }
+
+  /**
+   * It's here just to manage the explicit state resource in case the dependent resource implements
+   * {@link RecentOperationCacheFiller}.
+   *
+   * @return kubernetes client.
+   */
+  public KubernetesClient getKubernetesClient() {
+    return kubernetesClient;
+  }
+
+  public void setKubernetesClient(KubernetesClient kubernetesClient) {
+    this.kubernetesClient = kubernetesClient;
+  }
+}

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

@@ -11,7 +11,6 @@
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
 import io.javaoperatorsdk.operator.processing.dependent.Matcher.Result;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
 class BulkDependentResourceReconciler<R, P extends HasMetadata>
     implements DependentResourceReconciler<R, P> {
@@ -97,13 +96,13 @@ public Result<R> match(R resource, P primary, Context<P> context) {
     }
 
     @Override
-    protected void onCreated(ResourceID primaryResourceId, R created) {
-      asAbstractDependentResource().onCreated(primaryResourceId, created);
+    protected void onCreated(P primary, R created, Context<P> context) {
+      asAbstractDependentResource().onCreated(primary, created, context);
     }
 
     @Override
-    protected void onUpdated(ResourceID primaryResourceId, R updated, R actual) {
-      asAbstractDependentResource().onUpdated(primaryResourceId, updated, actual);
+    protected void onUpdated(P primary, R updated, R actual, Context<P> context) {
+      asAbstractDependentResource().onUpdated(primary, updated, actual, context);
     }
 
     @Override

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

@@ -0,0 +1,37 @@
+package io.javaoperatorsdk.operator.processing.dependent;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.KubernetesClientAware;
+
+/**
+ * Handles external resources where in order to address the resource additional information or
+ * persistent state (usually the ID of the resource) is needed to access the current state. These
+ * are non Kubernetes resources which when created their ID is generated, so cannot be determined
+ * based only on primary resources. In order to manage such dependent resource use this interface
+ * for a resource that extends {@link AbstractExternalDependentResource}.
+ */
+public interface DependentResourceWithExplicitState<R, P extends HasMetadata, S extends HasMetadata>
+    extends Creator<R, P>, Deleter<P>, KubernetesClientAware {
+
+  /**
+   * Only needs to be implemented if multiple event sources are present for the target resource
+   * class.
+   *
+   * @return name of the event source to access the state resources.
+   */
+  default Optional<String> eventSourceName() {
+    return Optional.empty();
+  }
+
+  /**
+   * Class of the state resource.
+   */
+  Class<S> stateResourceClass();
+
+  /** State resource which contains the target state. Usually an ID to address the resource */
+  S stateResource(P primary, R resource);
+
+}

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

@@ -21,6 +21,7 @@ public ReconcileResult<R> reconcile(P primary, Context<P> context) {
 
   @Override
   public void delete(P primary, Context<P> context) {
-    instance.handleDelete(primary, context);
+    var secondary = instance.getSecondaryResource(primary, context);
+    instance.handleDelete(primary, secondary.orElse(null), context);
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java

@@ -2,14 +2,13 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
-import io.javaoperatorsdk.operator.processing.dependent.AbstractEventSourceHolderDependentResource;
+import io.javaoperatorsdk.operator.processing.dependent.AbstractExternalDependentResource;
 import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
 import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 
 @Ignore
 public abstract class AbstractPollingDependentResource<R, P extends HasMetadata>
-    extends
-    AbstractEventSourceHolderDependentResource<R, P, ExternalResourceCachingEventSource<R, P>>
+    extends AbstractExternalDependentResource<R, P, ExternalResourceCachingEventSource<R, P>>
     implements CacheKeyMapper<R> {
 
   public static final int DEFAULT_POLLING_PERIOD = 5000;

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java

@@ -144,9 +144,10 @@ public Result<R> match(R actualResource, R desired, P primary, Context<P> contex
     return GenericKubernetesResourceMatcher.match(desired, actualResource, false);
   }
 
-  protected void handleDelete(P primary, Context<P> context) {
-    var resource = getSecondaryResource(primary, context);
-    resource.ifPresent(r -> client.resource(r).delete());
+  protected void handleDelete(P primary, R secondary, Context<P> context) {
+    if (secondary != null) {
+      client.resource(secondary).delete();
+    }
   }
 
   @SuppressWarnings("unused")

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java

@@ -7,7 +7,6 @@
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 
 import static org.junit.jupiter.api.Assertions.*;
@@ -85,10 +84,12 @@ public Optional<ConfigMap> getSecondaryResource(TestCustomResource primary,
     }
 
     @Override
-    protected void onCreated(ResourceID primaryResourceId, ConfigMap created) {}
+    protected void onCreated(TestCustomResource primary, ConfigMap created,
+        Context<TestCustomResource> context) {}
 
     @Override
-    protected void onUpdated(ResourceID primaryResourceId, ConfigMap updated, ConfigMap actual) {}
+    protected void onUpdated(TestCustomResource primary, ConfigMap updated, ConfigMap actual,
+        Context<TestCustomResource> context) {}
 
     @Override
     protected ConfigMap desired(TestCustomResource primary, Context<TestCustomResource> context) {

operator-framework/src/test/java/io/javaoperatorsdk/operator/EventSourceIT.java

@@ -16,6 +16,7 @@
 import static org.awaitility.Awaitility.await;
 
 class EventSourceIT {
+
   @RegisterExtension
   LocallyRunOperatorExtension operator =
       LocallyRunOperatorExtension.builder().withReconciler(EventSourceTestCustomReconciler.class)