improve: checks on inputs and lock method version of PrimaryUpdateAndCacheUtils

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

Author: csviri

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

@@ -7,8 +7,10 @@
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.dsl.base.PatchContext;
 import io.fabric8.kubernetes.client.dsl.base.PatchType;
+import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.api.reconciler.support.PrimaryResourceCache;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
@@ -20,6 +22,8 @@
  */
 public class PrimaryUpdateAndCacheUtils {
 
+  public static final int DEFAULT_MAX_RETRY = 3;
+
   private PrimaryUpdateAndCacheUtils() {}
 
   private static final Logger log = LoggerFactory.getLogger(PrimaryUpdateAndCacheUtils.class);
@@ -40,6 +44,15 @@ public static <P extends HasMetadata> P updateStatusAndCacheResource(
         primary, context, () -> context.getClient().resource(primary).updateStatus());
   }
 
+  public static <P extends HasMetadata> P updateStatusAndCacheResourceWithLock(
+      P primary, Context<P> context, UnaryOperator<P> modificationFunction) {
+    return updateAndCacheResourceWithLock(
+        primary,
+        context,
+        modificationFunction,
+        r -> context.getClient().resource(r).updateStatus());
+  }
+
   /**
    * Patches status with and makes sure that the up-to-date primary resource will be present during
    * the next reconciliation. Using JSON Merge patch.
@@ -56,6 +69,12 @@ public static <P extends HasMetadata> P patchStatusAndCacheResource(
         primary, context, () -> context.getClient().resource(primary).patchStatus());
   }
 
+  public static <P extends HasMetadata> P patchStatusAndCacheResourceWithLock(
+      P primary, Context<P> context, UnaryOperator<P> modificationFunction) {
+    return updateAndCacheResourceWithLock(
+        primary, context, modificationFunction, r -> context.getClient().resource(r).patchStatus());
+  }
+
   /**
    * Patches status and makes sure that the up-to-date primary resource will be present during the
    * next reconciliation. Using JSON Patch.
@@ -72,6 +91,15 @@ public static <P extends HasMetadata> P editStatusAndCacheResource(
         primary, context, () -> context.getClient().resource(primary).editStatus(operation));
   }
 
+  public static <P extends HasMetadata> P editStatusAndCacheResourceWithLock(
+      P primary, Context<P> context, UnaryOperator<P> modificationFunction) {
+    return updateAndCacheResourceWithLock(
+        primary,
+        context,
+        UnaryOperator.identity(),
+        r -> context.getClient().resource(r).editStatus(modificationFunction));
+  }
+
   /**
    * Patches the resource with supplied method and makes sure that the up-to-date primary resource
    * will be present during the next reconciliation.
@@ -124,6 +152,25 @@ public static <P extends HasMetadata> P ssaPatchStatusAndCacheResource(
     return res;
   }
 
+  public static <P extends HasMetadata> P ssaPatchStatusAndCacheResourceWithLock(
+      P primary, P freshResourceWithStatus, Context<P> context) {
+    return updateAndCacheResourceWithLock(
+        primary,
+        context,
+        r -> freshResourceWithStatus,
+        r ->
+            context
+                .getClient()
+                .resource(r)
+                .subresource("status")
+                .patch(
+                    new PatchContext.Builder()
+                        .withForce(true)
+                        .withFieldManager(context.getControllerConfiguration().fieldManager())
+                        .withPatchType(PatchType.SERVER_SIDE_APPLY)
+                        .build()));
+  }
+
   /**
    * Patches the resource status and caches the response in provided {@link PrimaryResourceCache}.
    * Uses Server Side Apply.
@@ -228,4 +275,71 @@ private static <P extends HasMetadata> void logWarnIfResourceVersionPresent(P pr
               + "using optimistic locking is discouraged for this purpose. ");
     }
   }
+
+  public static <P extends HasMetadata> P updateAndCacheResourceWithLock(
+      P primary,
+      Context<P> context,
+      UnaryOperator<P> modificationFunction,
+      UnaryOperator<P> updateMethod) {
+    return updateAndCacheResourceWithLock(
+        primary, context, modificationFunction, updateMethod, DEFAULT_MAX_RETRY);
+  }
+
+  @SuppressWarnings("unchecked")
+  public static <P extends HasMetadata> P updateAndCacheResourceWithLock(
+      P primary,
+      Context<P> context,
+      UnaryOperator<P> modificationFunction,
+      UnaryOperator<P> updateMethod,
+      int maxRetry) {
+
+    if (log.isDebugEnabled()) {
+      log.debug("Conflict retrying update for: {}", ResourceID.fromResource(primary));
+    }
+    int retryIndex = 0;
+    while (true) {
+      try {
+        var modified = modificationFunction.apply(primary);
+        modified.getMetadata().setResourceVersion(primary.getMetadata().getResourceVersion());
+        var updated = updateMethod.apply(modified);
+        context
+            .eventSourceRetriever()
+            .getControllerEventSource()
+            .handleRecentResourceUpdate(ResourceID.fromResource(primary), updated, primary);
+        return updated;
+      } catch (KubernetesClientException e) {
+        log.trace("Exception during patch for resource: {}", primary);
+        retryIndex++;
+        // only retry on conflict (409) and unprocessable content (422) which
+        // can happen if JSON Patch is not a valid request since there was
+        // a concurrent request which already removed another finalizer:
+        // List element removal from a list is by index in JSON Patch
+        // so if addressing a second finalizer but first is meanwhile removed
+        // it is a wrong request.
+        if (e.getCode() != 409 && e.getCode() != 422) {
+          throw e;
+        }
+        if (retryIndex >= maxRetry) {
+          throw new OperatorException(
+              "Exceeded maximum ("
+                  + maxRetry
+                  + ") retry attempts to patch resource: "
+                  + ResourceID.fromResource(primary));
+        }
+        log.debug(
+            "Retrying patch for resource name: {}, namespace: {}; HTTP code: {}",
+            primary.getMetadata().getName(),
+            primary.getMetadata().getNamespace(),
+            e.getCode());
+        primary =
+            (P)
+                context
+                    .getClient()
+                    .resources(primary.getClass())
+                    .inNamespace(primary.getMetadata().getNamespace())
+                    .withName(primary.getMetadata().getName())
+                    .get();
+      }
+    }
+  }
 }