Owls 106174 - improve handling of cluster resource Added/Modified/Deleted watch notifications (#3915)

* Add shouldContinue for cluster makeright and add domainUid label

Author: doxiao

integration-tests/src/test/java/oracle/weblogic/kubernetes/ItKubernetesDomainEvents.java

@@ -35,6 +35,7 @@
 import oracle.weblogic.kubernetes.annotations.Namespaces;
 import oracle.weblogic.kubernetes.logging.LoggingFacade;
 import oracle.weblogic.kubernetes.utils.DomainUtils;
+import oracle.weblogic.kubernetes.utils.K8sEvents;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.DisplayName;
@@ -399,6 +400,12 @@ void testK8SEventsMultiClusterEvents() {
     checkEvent(opNamespace, domainNamespace3, domainUid,
             DOMAIN_AVAILABLE, "Normal", timestamp);
     logger.info("verify the Cluster_Available event is generated");
+    checkEvent(opNamespace, domainNamespace3, domainUid,
+        CLUSTER_CHANGED, "Normal", timestamp);
+    assertEquals(1, getOpGeneratedEventCount(domainNamespace3, domainUid,
+        CLUSTER_CHANGED, timestamp2));
+    assertEquals(1, K8sEvents.getOpGeneratedEventCountForResource(domainNamespace3, domainUid, cluster2Name,
+        CLUSTER_CHANGED, timestamp));
     checkEvent(opNamespace, domainNamespace3, domainUid,
         CLUSTER_AVAILABLE, "Normal", timestamp);
     logger.info("verify the Completed event is generated");

integration-tests/src/test/java/oracle/weblogic/kubernetes/utils/K8sEvents.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020, 2022, Oracle and/or its affiliates.
+// Copyright (c) 2020, 2023, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.weblogic.kubernetes.utils;
@@ -371,6 +371,38 @@ public static int getOpGeneratedEventCount(
     return count;
   }
 
+  /**
+   * Get the event count between a specific timestamp for the given resource (domain or cluster).
+   *
+   * @param domainNamespace namespace in which the domain exists
+   * @param domainUid UID of the domain
+   * @param resourceName the Uid/name of the resource
+   * @param reason event to check for Created, Changed, deleted, processing etc
+   * @param timestamp the timestamp after which to see events
+   * @return count number of events count
+   */
+  public static int getOpGeneratedEventCountForResource(
+      String domainNamespace, String domainUid, String resourceName, String reason, OffsetDateTime timestamp) {
+    int count = 0;
+    try {
+      List<CoreV1Event> events = Kubernetes.listOpGeneratedNamespacedEvents(domainNamespace);
+      for (CoreV1Event event : events) {
+        Map<String, String> labels = event.getMetadata().getLabels();
+        if (event.getReason().equals(reason)
+            && labels.get("weblogic.domainUID").equals(domainUid)
+            && event.getInvolvedObject().getName().equals(resourceName)
+            && (isEqualOrAfter(timestamp, event))) {
+          logger.info(Yaml.dump(event));
+          count++;
+        }
+      }
+    } catch (ApiException ex) {
+      Logger.getLogger(K8sEvents.class.getName()).log(Level.SEVERE, null, ex);
+      return -1;
+    }
+    return count;
+  }
+
   /**
    * Check if a given event is logged only once for the given pod.
    *

operator/src/main/java/oracle/kubernetes/operator/ClusterResourceStatusUpdater.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2022, Oracle and/or its affiliates.
+// Copyright (c) 2022, 2023, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
@@ -36,6 +36,7 @@
 
 import static oracle.kubernetes.operator.KubernetesConstants.CLUSTER;
 import static oracle.kubernetes.operator.KubernetesConstants.HTTP_NOT_FOUND;
+import static oracle.kubernetes.operator.helpers.EventHelper.createClusterResourceEventData;
 
 /**
  * Updates for status of Cluster resources.
@@ -255,14 +256,14 @@ private List<EventData> getClusterStatusConditionTrueEvents(List<ClusterConditio
 
     private EventData toTrueClusterResourceEvent(ClusterCondition condition) {
       return Optional.ofNullable(condition.getType().getAddedEvent())
-          .map(eventItem -> new ClusterResourceEventData(eventItem, resource))
+          .map(eventItem -> createClusterResourceEventData(eventItem, resource, domain.getDomainUid()))
           .map(this::initializeClusterResourceEventData)
           .orElse(null);
     }
 
     private EventData toFalseClusterResourceEvent(ClusterCondition removedCondition) {
       return Optional.ofNullable(removedCondition.getType().getRemovedEvent())
-          .map(eventItem -> new ClusterResourceEventData(eventItem, resource))
+          .map(eventItem -> createClusterResourceEventData(eventItem, resource, domain.getDomainUid()))
           .map(this::initializeClusterResourceEventData)
           .orElse(null);
     }

operator/src/main/java/oracle/kubernetes/operator/DomainProcessor.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2018, 2022, Oracle and/or its affiliates.
+// Copyright (c) 2018, 2023, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
@@ -34,6 +34,18 @@ public interface DomainProcessor {
    */
   MakeRightDomainOperation createMakeRightOperation(DomainPresenceInfo liveInfo);
 
+  /**
+   * Ensures that a cluster event is generated for a cluster resource no matter whether it is referenced by a domain
+   * or not.
+   *
+   * @param clusterEvent the event that needs to be generated
+   * @param cluster the cluster resource that the event is associated with
+   * @param domainUid the UID of the domain that the cluster is referenced by
+   * @return Make-right operation
+   */
+  MakeRightClusterOperation createMakeRightOperationForClusterEvent(
+      EventItem clusterEvent, ClusterResource cluster, String domainUid);
+
   /**
    * Ensures that a cluster event is generated for a cluster resource no matter whether it is referenced by a domain
    * or not.

operator/src/main/java/oracle/kubernetes/operator/DomainProcessorImpl.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2018, 2022, Oracle and/or its affiliates.
+// Copyright (c) 2018, 2023, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
@@ -36,7 +36,6 @@
 import oracle.kubernetes.operator.helpers.ConfigMapHelper;
 import oracle.kubernetes.operator.helpers.DomainPresenceInfo;
 import oracle.kubernetes.operator.helpers.EventHelper;
-import oracle.kubernetes.operator.helpers.EventHelper.ClusterResourceEventData;
 import oracle.kubernetes.operator.helpers.EventHelper.EventData;
 import oracle.kubernetes.operator.helpers.EventHelper.EventItem;
 import oracle.kubernetes.operator.helpers.KubernetesEventObjects;
@@ -141,6 +140,14 @@ private static DomainPresenceInfo getExistingDomainPresenceInfo(DomainPresenceIn
     return getExistingDomainPresenceInfo(newPresence.getNamespace(), newPresence.getDomainUid());
   }
 
+  private static ClusterPresenceInfo getExistingClusterPresenceInfo(String ns, String clusterName) {
+    return clusters.computeIfAbsent(ns, k -> new ConcurrentHashMap<>()).get(clusterName);
+  }
+
+  private static ClusterPresenceInfo getExistingClusterPresenceInfo(ClusterPresenceInfo newPresence) {
+    return getExistingClusterPresenceInfo(newPresence.getNamespace(), newPresence.getResourceName());
+  }
+
   @Override
   public Map<String, Map<String,DomainPresenceInfo>>  getDomainPresenceInfoMap() {
     return domains;
@@ -374,7 +381,9 @@ public void runMakeRight(MakeRightClusterOperation operation) {
     final ClusterPresenceInfo liveInfo = operation.getPresenceInfo();
     if (delegate.isNamespaceRunning(liveInfo.getNamespace())) {
       try (ThreadLoggingContext ignored = setThreadContext().presenceInfo(liveInfo)) {
-        new ClusterPlan(operation, delegate).execute();
+        if (shouldContinue(operation, liveInfo)) {
+          new ClusterPlan(operation, delegate).execute();
+        }
       }
     }
   }
@@ -394,10 +403,38 @@ private boolean shouldContinue(MakeRightDomainOperation operation, DomainPresenc
     }
   }
 
+  private boolean shouldContinue(MakeRightClusterOperation operation, ClusterPresenceInfo liveInfo) {
+    final ClusterPresenceInfo cachedInfo = getExistingClusterPresenceInfo(liveInfo);
+    if (hasDeletedClusterEventData(operation)) {
+      return findClusterPresenceInfo(liveInfo.getNamespace(), liveInfo.getResourceName());
+    } else if (isNewCluster(cachedInfo)) {
+      return true;
+    } else if (liveInfo.isFromOutOfDateEvent(operation, cachedInfo)) {
+      return false;
+    } else if (liveInfo.isClusterGenerationChanged(cachedInfo)) {
+      return true;
+    } else {
+      cachedInfo.setCluster(liveInfo.getCluster());
+      return false;
+    }
+  }
+
   private boolean isNewDomain(DomainPresenceInfo cachedInfo) {
     return Optional.ofNullable(cachedInfo).map(DomainPresenceInfo::getDomain).orElse(null) == null;
   }
 
+  private boolean isNewCluster(ClusterPresenceInfo cachedInfo) {
+    return Optional.ofNullable(cachedInfo).map(ClusterPresenceInfo::getCluster).orElse(null) == null;
+  }
+
+  private boolean findClusterPresenceInfo(String namespace, String clusterName) {
+    return Optional.ofNullable(clusters.get(namespace)).orElse(Collections.emptyMap()).get(clusterName) != null;
+  }
+
+  private boolean hasDeletedClusterEventData(MakeRightClusterOperation operation) {
+    return operation.getEventData() != null && operation.getEventData().getItem().name().equals("CLUSTER_DELETED");
+  }
+
   private void logStartingDomain(DomainPresenceInfo presenceInfo) {
     LOGGER.fine(MessageKeys.PROCESSING_DOMAIN, presenceInfo.getDomainUid());
   }
@@ -430,12 +467,24 @@ public void registerDomainPresenceInfo(DomainPresenceInfo info) {
           .put(info.getDomainUid(), info);
   }
 
+  @Override
+  public void registerClusterPresenceInfo(ClusterPresenceInfo info) {
+    clusters
+        .computeIfAbsent(info.getNamespace(), k -> new ConcurrentHashMap<>())
+        .put(info.getResourceName(), info);
+  }
+
   @Override
   public void unregisterDomainPresenceInfo(DomainPresenceInfo info) {
     unregisterPresenceInfo(info.getNamespace(), info.getDomainUid());
     unregisterEventK8SObject(info.getNamespace(), info.getDomainUid());
   }
 
+  @Override
+  public void unregisterClusterPresenceInfo(ClusterPresenceInfo info) {
+    unregisterPresenceInfoForCluster(info.getNamespace(), info.getResourceName());
+  }
+
   private static void unregisterEventK8SObject(String ns, String domainUid) {
     Optional.ofNullable(domainEventK8SObjects.get(ns)).ifPresent(m -> m.remove(domainUid));
   }
@@ -444,6 +493,10 @@ private static void unregisterPresenceInfo(String ns, String domainUid) {
     Optional.ofNullable(domains.get(ns)).ifPresent(m -> m.remove(domainUid));
   }
 
+  private static void unregisterPresenceInfoForCluster(String ns, String clusterName) {
+    Optional.ofNullable(clusters.get(ns)).ifPresent(m -> m.remove(clusterName));
+  }
+
   @Override
   public void endScheduledDomainStatusUpdates(DomainPresenceInfo info) {
     Map<String, ScheduledFuture<?>> map = statusUpdaters.get(info.getNamespace());
@@ -727,14 +780,15 @@ public void dispatchClusterWatch(Watch.Response<ClusterResource> item) {
   }
 
   private void handleAddedCluster(ClusterResource cluster) {
-    createMakeRightOperationForClusterEvent(EventItem.CLUSTER_CREATED, cluster).execute();
     List<DomainPresenceInfo> hostingDomains =
         getExistingDomainPresenceInfoForCluster(cluster.getNamespace(), cluster.getMetadata().getName());
     if (hostingDomains.isEmpty()) {
       LOGGER.info(MessageKeys.WATCH_CLUSTER_WITHOUT_DOMAIN, cluster.getMetadata().getName());
+      createMakeRightOperationForClusterEvent(EventItem.CLUSTER_CREATED, cluster, null).execute();
     } else {
       hostingDomains.forEach(info -> {
         LOGGER.info(MessageKeys.WATCH_CLUSTER, cluster.getMetadata().getName(), info.getDomainUid());
+        createMakeRightOperationForClusterEvent(EventItem.CLUSTER_CREATED, cluster, info.getDomainUid()).execute();
         createMakeRightOperation(info)
             .interrupt()
             .withExplicitRecheck()
@@ -744,11 +798,11 @@ private void handleAddedCluster(ClusterResource cluster) {
   }
 
   private void handleModifiedCluster(ClusterResource cluster) {
-    createMakeRightOperationForClusterEvent(EventItem.CLUSTER_CHANGED, cluster).execute();
     List<DomainPresenceInfo> hostingDomains =
         getExistingDomainPresenceInfoForCluster(cluster.getNamespace(), cluster.getMetadata().getName());
     if (hostingDomains.isEmpty()) {
       LOGGER.info(MessageKeys.WATCH_CLUSTER_WITHOUT_DOMAIN, cluster.getMetadata().getName());
+      createMakeRightOperationForClusterEvent(EventItem.CLUSTER_CHANGED, cluster, null).execute();
     } else {
       hostingDomains.forEach(info -> {
         ClusterResource cachedResource = info.getClusterResource(cluster.getClusterName());
@@ -757,6 +811,7 @@ private void handleModifiedCluster(ClusterResource cluster) {
         }
 
         LOGGER.fine(MessageKeys.WATCH_CLUSTER, cluster.getMetadata().getName(), info.getDomainUid());
+        createMakeRightOperationForClusterEvent(EventItem.CLUSTER_CHANGED, cluster, info.getDomainUid()).execute();
         createMakeRightOperation(info)
             .interrupt()
             .withExplicitRecheck()
@@ -766,14 +821,15 @@ private void handleModifiedCluster(ClusterResource cluster) {
   }
 
   private void handleDeletedCluster(ClusterResource cluster) {
-    createMakeRightOperationForClusterEvent(EventItem.CLUSTER_DELETED, cluster).execute();
     List<DomainPresenceInfo> hostingDomains =
         getExistingDomainPresenceInfoForCluster(cluster.getNamespace(), cluster.getMetadata().getName());
     if (hostingDomains.isEmpty()) {
       LOGGER.info(MessageKeys.WATCH_CLUSTER_WITHOUT_DOMAIN, cluster.getMetadata().getName());
+      createMakeRightOperationForClusterEvent(EventItem.CLUSTER_DELETED, cluster, null).execute();
     } else {
       hostingDomains.forEach(info -> {
         LOGGER.info(MessageKeys.WATCH_CLUSTER_DELETED, cluster.getMetadata().getName(), info.getDomainUid());
+        createMakeRightOperationForClusterEvent(EventItem.CLUSTER_DELETED, cluster, info.getDomainUid()).execute();
         info.removeClusterResource(cluster.getClusterName());
         createMakeRightOperation(info)
             .interrupt()
@@ -785,14 +841,16 @@ private void handleDeletedCluster(ClusterResource cluster) {
 
   @Override
   public MakeRightClusterOperation createMakeRightOperationForClusterEvent(
-      EventItem clusterEvent, ClusterResource cluster) {
+      EventItem clusterEvent, ClusterResource cluster, String domainUid) {
     return delegate.createMakeRightOperation(this, createInfoForClusterEventOnly(cluster))
         .interrupt()
-        .withEventData(createClusterResourceEventData(clusterEvent, cluster));
+        .withEventData(EventHelper.createClusterResourceEventData(clusterEvent, cluster, domainUid));
   }
 
-  private ClusterResourceEventData createClusterResourceEventData(EventItem clusterEvent, ClusterResource cluster) {
-    return new ClusterResourceEventData(clusterEvent, cluster);
+  @Override
+  public MakeRightClusterOperation createMakeRightOperationForClusterEvent(
+      EventItem clusterEvent, ClusterResource cluster) {
+    return createMakeRightOperationForClusterEvent(clusterEvent, cluster, null);
   }
 
   @NotNull
@@ -913,17 +971,12 @@ private void addServerToMaps(Map<String, ServerHealth> serverHealthMap,
 
   }
 
-  private class DomainPlan extends Plan<MakeRightDomainOperation> {
+  private static class DomainPlan extends Plan<MakeRightDomainOperation> {
 
     public DomainPlan(MakeRightDomainOperation operation, DomainProcessorDelegate delegate) {
       super(operation, delegate);
     }
 
-    @Override
-    protected void cacheResourcePresenceInfo(ResourcePresenceInfo presenceInfo) {
-      //No-op.
-    }
-
     @Override
     public CompletionCallback createCompletionCallback() {
       return new DomainPlanCompletionCallback();
@@ -1001,32 +1054,20 @@ private long delayUntilNextRetry(@Nonnull DomainPresenceInfo domainPresenceInfo)
       return interval.getSeconds();
 
     }
-
   }
 
-  private class ClusterPlan extends Plan<MakeRightClusterOperation> {
+  private static class ClusterPlan extends Plan<MakeRightClusterOperation> {
 
     public ClusterPlan(MakeRightClusterOperation operation, DomainProcessorDelegate delegate) {
       super(operation, delegate);
     }
 
-    @Override
-    protected void cacheResourcePresenceInfo(ResourcePresenceInfo presenceInfo) {
-      if (operation.getEventData().getItem() == EventHelper.EventItem.CLUSTER_DELETED) {
-        Optional.ofNullable(clusters.get(presenceInfo.getNamespace()))
-            .ifPresent(m -> m.remove(presenceInfo.getResourceName()));
-      } else {
-        clusters.computeIfAbsent(presenceInfo.getNamespace(), c -> new ConcurrentHashMap<>())
-            .computeIfAbsent(presenceInfo.getResourceName(), k -> (ClusterPresenceInfo) presenceInfo);
-      }
-    }
-
     @Override
     public CompletionCallback createCompletionCallback() {
       return new ClusterPlanCompletionCallback();
     }
 
-    class ClusterPlanCompletionCallback implements CompletionCallback {
+    static class ClusterPlanCompletionCallback implements CompletionCallback {
 
       @Override
       public void onCompletion(Packet packet) {
@@ -1044,7 +1085,7 @@ private void reportFailure(Throwable throwable) {
     }
   }
 
-  private abstract class Plan<T extends MakeRightOperation> {
+  private abstract static class Plan<T extends MakeRightOperation> {
 
     final T operation;
     protected final ResourcePresenceInfo presenceInfo;
@@ -1059,11 +1100,8 @@ public Plan(T operation, DomainProcessorDelegate delegate) {
       this.firstStep = operation.createSteps();
       this.packet = operation.createPacket();
       this.gate = getMakeRightFiberGate(delegate, this.presenceInfo.getNamespace());
-      cacheResourcePresenceInfo(presenceInfo);
     }
 
-    protected abstract void cacheResourcePresenceInfo(ResourcePresenceInfo presenceInfo);
-
     private FiberGate getMakeRightFiberGate(DomainProcessorDelegate delegate, String ns) {
       return makeRightFiberGates.computeIfAbsent(ns, k -> delegate.createFiberGate());
     }