improve: KubernetesDependentResource uses resource operations directly (#3146)

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

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

@@ -28,6 +28,7 @@
 import io.fabric8.kubernetes.client.dsl.base.PatchType;
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.informer.ManagedInformerEventSource;
 
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID;
@@ -80,6 +81,40 @@ public <R extends HasMetadata> R serverSideApply(R resource) {
                         .build()));
   }
 
+  /**
+   * Updates the resource and caches the response if needed, thus making sure that next
+   * reconciliation will see to updated resource - or more recent one if additional update happened
+   * after this update; In addition to that it filters out the event from the update, so
+   * reconciliation is not triggered by own update.
+   *
+   * <p>You are free to control the optimistic locking by setting the resource version in resource
+   * metadata. In case of SSA we advise not to do updates with optimistic locking.
+   *
+   * @param resource fresh resource for server side apply
+   * @return updated resource
+   * @param informerEventSource InformerEventSource to use for resource caching and filtering
+   * @param <R> resource type
+   */
+  public <R extends HasMetadata> R serverSideApply(
+      R resource, InformerEventSource<R, P> informerEventSource) {
+    if (informerEventSource == null) {
+      return serverSideApply(resource);
+    }
+    return resourcePatch(
+        resource,
+        r ->
+            context
+                .getClient()
+                .resource(r)
+                .patch(
+                    new PatchContext.Builder()
+                        .withForce(true)
+                        .withFieldManager(context.getControllerConfiguration().fieldManager())
+                        .withPatchType(PatchType.SERVER_SIDE_APPLY)
+                        .build()),
+        informerEventSource);
+  }
+
   /**
    * Server-Side Apply the resource status subresource.
    *
@@ -189,6 +224,69 @@ public <R extends HasMetadata> R update(R resource) {
     return resourcePatch(resource, r -> context.getClient().resource(r).update());
   }
 
+  /**
+   * Updates the resource and caches the response if needed, thus making sure that next
+   * reconciliation will see to updated resource - or more recent one if additional update happened
+   * after this update; In addition to that it filters out the event from this update, so
+   * reconciliation is not triggered by own update.
+   *
+   * <p>You are free to control the optimistic locking by setting the resource version in resource
+   * metadata.
+   *
+   * @param resource resource to update
+   * @return updated resource
+   * @param informerEventSource InformerEventSource to use for resource caching and filtering
+   * @param <R> resource type
+   */
+  public <R extends HasMetadata> R update(
+      R resource, InformerEventSource<R, P> informerEventSource) {
+    if (informerEventSource == null) {
+      return update(resource);
+    }
+    return resourcePatch(
+        resource, r -> context.getClient().resource(r).update(), informerEventSource);
+  }
+
+  /**
+   * Creates the resource and caches the response if needed, thus making sure that next
+   * reconciliation will see to updated resource - or more recent one if additional update happened
+   * after this update; In addition to that it filters out the event from this update, so
+   * reconciliation is not triggered by own update.
+   *
+   * <p>You are free to control the optimistic locking by setting the resource version in resource
+   * metadata.
+   *
+   * @param resource resource to update
+   * @return updated resource
+   * @param <R> resource type
+   */
+  public <R extends HasMetadata> R create(R resource) {
+    return resourcePatch(resource, r -> context.getClient().resource(r).create());
+  }
+
+  /**
+   * Creates the resource and caches the response if needed, thus making sure that next
+   * reconciliation will see to updated resource - or more recent one if additional update happened
+   * after this update; In addition to that it filters out the event from this update, so
+   * reconciliation is not triggered by own update.
+   *
+   * <p>You are free to control the optimistic locking by setting the resource version in resource
+   * metadata.
+   *
+   * @param resource resource to update
+   * @return updated resource
+   * @param informerEventSource InformerEventSource to use for resource caching and filtering
+   * @param <R> resource type
+   */
+  public <R extends HasMetadata> R create(
+      R resource, InformerEventSource<R, P> informerEventSource) {
+    if (informerEventSource == null) {
+      return create(resource);
+    }
+    return resourcePatch(
+        resource, r -> context.getClient().resource(r).create(), informerEventSource);
+  }
+
   /**
    * Updates the resource status subresource.
    *

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

@@ -25,7 +25,6 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.Namespaced;
-import io.fabric8.kubernetes.client.dsl.Resource;
 import io.javaoperatorsdk.operator.api.config.dependent.Configured;
 import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
@@ -71,28 +70,10 @@ public void configureWith(KubernetesDependentResourceConfig<R> config) {
     this.kubernetesDependentResourceConfig = config;
   }
 
-  @Override
-  protected R handleCreate(R desired, P primary, Context<P> context) {
-    return eventSource()
-        .orElseThrow()
-        .eventFilteringUpdateAndCacheResource(
-            desired,
-            toCreate -> KubernetesDependentResource.super.handleCreate(toCreate, primary, context));
-  }
-
-  @Override
-  protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
-    return eventSource()
-        .orElseThrow()
-        .eventFilteringUpdateAndCacheResource(
-            desired,
-            toUpdate ->
-                KubernetesDependentResource.super.handleUpdate(actual, toUpdate, primary, context));
-  }
-
   @SuppressWarnings("unused")
   public R create(R desired, P primary, Context<P> context) {
-    if (useSSA(context)) {
+    var ssa = useSSA(context);
+    if (ssa) {
       // setting resource version for SSA so only created if it doesn't exist already
       var createIfNotExisting =
           kubernetesDependentResourceConfig == null
@@ -104,35 +85,40 @@ public R create(R desired, P primary, Context<P> context) {
       }
     }
     addMetadata(false, null, desired, primary, context);
-    final var resource = prepare(context, desired, primary, "Creating");
-    return useSSA(context)
-        ? resource
-            .fieldManager(context.getControllerConfiguration().fieldManager())
-            .forceConflicts()
-            .serverSideApply()
-        : resource.create();
+    log.debug(
+        "Creating target resource with type: {}, with id: {} use ssa: {}",
+        desired.getClass(),
+        ResourceID.fromResource(desired),
+        ssa);
+
+    return ssa
+        ? context.resourceOperations().serverSideApply(desired, eventSource().orElse(null))
+        : context.resourceOperations().create(desired, eventSource().orElse(null));
   }
 
   public R update(R actual, R desired, P primary, Context<P> context) {
-    boolean useSSA = useSSA(context);
+    boolean ssa = useSSA(context);
     if (log.isDebugEnabled()) {
       log.debug(
           "Updating actual resource: {} version: {}; SSA: {}",
           ResourceID.fromResource(actual),
           actual.getMetadata().getResourceVersion(),
-          useSSA);
+          ssa);
     }
     R updatedResource;
     addMetadata(false, actual, desired, primary, context);
-    if (useSSA) {
+    log.debug(
+        "Updating target resource with type: {}, with id: {} use ssa: {}",
+        desired.getClass(),
+        ResourceID.fromResource(desired),
+        ssa);
+    if (ssa) {
       updatedResource =
-          prepare(context, desired, primary, "Updating")
-              .fieldManager(context.getControllerConfiguration().fieldManager())
-              .forceConflicts()
-              .serverSideApply();
+          context.resourceOperations().serverSideApply(desired, eventSource().orElse(null));
     } else {
       var updatedActual = GenericResourceUpdater.updateResource(actual, desired, context);
-      updatedResource = prepare(context, updatedActual, primary, "Updating").update();
+      updatedResource =
+          context.resourceOperations().update(updatedActual, eventSource().orElse(null));
     }
     log.debug(
         "Resource version after update: {}", updatedResource.getMetadata().getResourceVersion());
@@ -203,17 +189,6 @@ public void deleteTargetResource(P primary, R resource, ResourceID key, Context<
     context.getClient().resource(resource).delete();
   }
 
-  @SuppressWarnings("unused")
-  protected Resource<R> prepare(Context<P> context, R desired, P primary, String actionName) {
-    log.debug(
-        "{} target resource with type: {}, with id: {}",
-        actionName,
-        desired.getClass(),
-        ResourceID.fromResource(desired));
-
-    return context.getClient().resource(desired);
-  }
-
   protected void addReferenceHandlingMetadata(R desired, P primary) {
     if (addOwnerReference()) {
       desired.addOwnerReference(primary);