AMBARI-21679. Service Checks Will Run Multiple Times In Patch/Service Upgrades (ncole)

ambari-server/src/main/java/org/apache/ambari/server/state/UpgradeHelper.java

@@ -63,6 +63,8 @@ import org.apache.ambari.server.state.stack.upgrade.Direction;
 import org.apache.ambari.server.state.stack.upgrade.Grouping;
 import org.apache.ambari.server.state.stack.upgrade.ManualTask;
 import org.apache.ambari.server.state.stack.upgrade.RestartTask;
+import org.apache.ambari.server.state.stack.upgrade.ServiceCheckGrouping;
+import org.apache.ambari.server.state.stack.upgrade.ServiceCheckGrouping.ServiceCheckStageWrapper;
 import org.apache.ambari.server.state.stack.upgrade.StageWrapper;
 import org.apache.ambari.server.state.stack.upgrade.StageWrapperBuilder;
 import org.apache.ambari.server.state.stack.upgrade.StartTask;
@@ -77,6 +79,7 @@ import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.collect.Lists;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
 import com.google.inject.Singleton;
@@ -299,6 +302,7 @@ public class UpgradeHelper {
     Map<String, Map<String, ProcessingComponent>> allTasks = upgradePack.getTasks();
     List<UpgradeGroupHolder> groups = new ArrayList<>();
 
+    UpgradeGroupHolder previousGroupHolder = null;
     for (Grouping group : upgradePack.getGroups(context.getDirection())) {
 
       // !!! grouping is not scoped to context
@@ -320,6 +324,7 @@ public class UpgradeHelper {
       groupHolder.skippable = group.skippable;
       groupHolder.supportsAutoSkipOnFailure = group.supportsAutoSkipOnFailure;
       groupHolder.allowRetry = group.allowRetry;
+      groupHolder.processingGroup = group.isProcessingGroup();
 
       // !!! all downgrades are skippable
       if (context.getDirection().isDowngrade()) {
@@ -496,9 +501,22 @@ public class UpgradeHelper {
       List<StageWrapper> proxies = builder.build(context);
 
       if (CollectionUtils.isNotEmpty(proxies)) {
+
         groupHolder.items = proxies;
         postProcess(context, groupHolder);
-        groups.add(groupHolder);
+
+        // !!! prevent service checks from running twice.  merge the stage wrappers
+        if (ServiceCheckGrouping.class.isInstance(group)) {
+          if (null != previousGroupHolder && ServiceCheckGrouping.class.equals(previousGroupHolder.groupClass)) {
+            mergeServiceChecks(groupHolder, previousGroupHolder);
+          } else {
+            groups.add(groupHolder);
+          }
+        } else {
+          groups.add(groupHolder);
+        }
+
+        previousGroupHolder = groupHolder;
       }
     }
 
@@ -518,9 +536,52 @@ public class UpgradeHelper {
       }
     }
 
+    // !!! strip off the first service check if nothing has been processed
+    Iterator<UpgradeGroupHolder> iterator = groups.iterator();
+    boolean canServiceCheck = false;
+    while (iterator.hasNext()) {
+      UpgradeGroupHolder holder = iterator.next();
+
+      if (ServiceCheckGrouping.class.equals(holder.groupClass) && !canServiceCheck) {
+        iterator.remove();
+      }
+
+      canServiceCheck |= holder.processingGroup;
+    }
+
     return groups;
   }
 
+  /**
+   * Merges two service check groups when they have been orchestrated back-to-back.
+   * @param newHolder   the "new" group holder, which was orchestrated after the "old" one
+   * @param oldHolder   the "old" group holder, which is one that was already orchestrated
+   */
+  @SuppressWarnings("unchecked")
+  private void mergeServiceChecks(UpgradeGroupHolder newHolder, UpgradeGroupHolder oldHolder) {
+
+    LinkedHashSet<StageWrapper> priority = new LinkedHashSet<>();
+    LinkedHashSet<StageWrapper> others = new LinkedHashSet<>();
+
+    for (List<StageWrapper> holderItems : new List[] { oldHolder.items, newHolder.items }) {
+      for (StageWrapper stageWrapper : holderItems) {
+        ServiceCheckStageWrapper wrapper = (ServiceCheckStageWrapper) stageWrapper;
+
+        if (wrapper.priority) {
+          priority.add(stageWrapper);
+        } else {
+          others.add(stageWrapper);
+        }
+      }
+    }
+
+    // !!! remove duplicate wrappers that are now in the priority list
+    others = new LinkedHashSet<>(CollectionUtils.subtract(others, priority));
+
+    oldHolder.items = Lists.newLinkedList(priority);
+    oldHolder.items.addAll(others);
+  }
+
   /**
    * Walks through the UpgradeGroupHolder and updates titles and manual tasks,
    * replacing keyword tokens needed for display purposes
@@ -642,6 +703,11 @@ public class UpgradeHelper {
    * Short-lived objects that hold information about upgrade groups
    */
   public static class UpgradeGroupHolder {
+    /**
+     *
+     */
+    private boolean processingGroup;
+
     /**
      * The name
      */

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ClusterGrouping.java

@@ -70,6 +70,10 @@ public class ClusterGrouping extends Grouping {
     return new ClusterBuilder(this);
   }
 
+  @Override
+  protected boolean serviceCheckAfterProcessing() {
+    return false;
+  }
 
   /**
    * Represents a single-stage execution that happens as part of a cluster-wide

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/Grouping.java

@@ -92,6 +92,21 @@ public class Grouping {
   @XmlElement(name = "condition")
   public Condition condition;
 
+  /**
+   * @return {@code true} when the grouping is used to upgrade services and that it is
+   * appropriate to run service checks after orchestration.
+   */
+  public final boolean isProcessingGroup() {
+    return serviceCheckAfterProcessing();
+  }
+
+  /**
+   * Overridable function to indicate if full service checks can be run
+   */
+  protected boolean serviceCheckAfterProcessing() {
+    return true;
+  }
+
   /**
    * Gets the default builder.
    */

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ServiceCheckGrouping.java

@@ -67,6 +67,11 @@ public class ServiceCheckGrouping extends Grouping {
   @XmlElement(name="service")
   private Set<String> excludeServices = new HashSet<>();
 
+  @Override
+  protected boolean serviceCheckAfterProcessing() {
+    return false;
+  }
+
   /**
    * {@inheritDoc}
    */
@@ -137,11 +142,8 @@ public class ServiceCheckGrouping extends Grouping {
       // create stages for the priorities
       for (String service : priorityServices) {
         if (checkServiceValidity(upgradeContext, service, serviceMap)) {
-          StageWrapper wrapper = new StageWrapper(
-            StageWrapper.Type.SERVICE_CHECK,
-            "Service Check " + upgradeContext.getServiceDisplay(service),
-            new TaskWrapper(service, "", Collections.<String>emptySet(),
-              new ServiceCheckTask()));
+          StageWrapper wrapper = new ServiceCheckStageWrapper(service,
+              upgradeContext.getServiceDisplay(service), true);
 
           result.add(wrapper);
           clusterServices.remove(service);
@@ -156,11 +158,9 @@ public class ServiceCheckGrouping extends Grouping {
           }
 
           if (checkServiceValidity(upgradeContext, service, serviceMap)) {
-            StageWrapper wrapper = new StageWrapper(
-              StageWrapper.Type.SERVICE_CHECK,
-              "Service Check " + upgradeContext.getServiceDisplay(service),
-              new TaskWrapper(service, "", Collections.<String>emptySet(),
-                new ServiceCheckTask()));
+            StageWrapper wrapper = new ServiceCheckStageWrapper(service,
+                upgradeContext.getServiceDisplay(service), false);
+
             result.add(wrapper);
           }
         }
@@ -266,4 +266,44 @@ public class ServiceCheckGrouping extends Grouping {
       }
     }
   }
+
+  /**
+   * Special type of stage wrapper that allows inspection of the service check for
+   * merging if required.  This prevents consecutive service checks from running, particularly
+   * for Patch or Maintenance types of upgrades.
+   */
+  public static class ServiceCheckStageWrapper extends StageWrapper {
+    public String service;
+    public boolean priority;
+
+    ServiceCheckStageWrapper(String service, String serviceDisplay, boolean priority) {
+      super(StageWrapper.Type.SERVICE_CHECK,
+          String.format("Service Check %s", serviceDisplay),
+          new TaskWrapper(service, "", Collections.<String>emptySet(), new ServiceCheckTask()));
+
+      this.service = service;
+      this.priority = priority;
+    }
+
+    @Override
+    public int hashCode() {
+      return service.hashCode();
+    }
+
+    @Override
+    public boolean equals(Object other) {
+      if (!other.getClass().equals(getClass())) {
+        return false;
+      }
+
+      if (other == this) {
+        return true;
+      }
+
+      return ((ServiceCheckStageWrapper) other).service.equals(service);
+
+    }
+
+  }
+
 }

ambari-server/src/test/java/org/apache/ambari/server/state/UpgradeHelperTest.java

@@ -77,6 +77,7 @@ import org.apache.ambari.server.state.stack.upgrade.HostOrderItem;
 import org.apache.ambari.server.state.stack.upgrade.HostOrderItem.HostOrderActionType;
 import org.apache.ambari.server.state.stack.upgrade.ManualTask;
 import org.apache.ambari.server.state.stack.upgrade.SecurityCondition;
+import org.apache.ambari.server.state.stack.upgrade.ServiceCheckGrouping;
 import org.apache.ambari.server.state.stack.upgrade.StageWrapper;
 import org.apache.ambari.server.state.stack.upgrade.StopGrouping;
 import org.apache.ambari.server.state.stack.upgrade.Task;
@@ -1109,11 +1110,11 @@ public class UpgradeHelperTest extends EasyMockSupport {
 
     List<UpgradeGroupHolder> groups = m_upgradeHelper.createSequence(upgrade, context);
 
-    assertEquals(5, groups.size());
+    assertEquals(6, groups.size());
 
     // grab the manual task out of ZK which has placeholder text
 
-    UpgradeGroupHolder zookeeperGroup = groups.get(3);
+    UpgradeGroupHolder zookeeperGroup = groups.get(4);
     assertEquals("ZOOKEEPER", zookeeperGroup.name);
     ManualTask manualTask = (ManualTask) zookeeperGroup.items.get(0).getTasks().get(
         0).getTasks().get(0);
@@ -2498,8 +2499,71 @@ public class UpgradeHelperTest extends EasyMockSupport {
 
     assertNotNull(clusterEnvMap);
     assertTrue(clusterEnvMap.containsKey("a"));
+
+    // Do stacks cleanup
+    stackManagerMock.invalidateCurrentPaths();
+    ambariMetaInfo.init();
+  }
+
+  @Test
+  public void testSequentialServiceChecks() throws Exception {
+    Map<String, UpgradePack> upgrades = ambariMetaInfo.getUpgradePacks("HDP", "2.1.1");
+    assertTrue(upgrades.containsKey("upgrade_test_checks"));
+    UpgradePack upgrade = upgrades.get("upgrade_test_checks");
+    assertNotNull(upgrade);
+
+    Cluster cluster = makeCluster();
+    cluster.deleteService("HDFS");
+    cluster.deleteService("YARN");
+
+    UpgradeContext context = getMockUpgradeContext(cluster, Direction.UPGRADE,
+        UpgradeType.ROLLING, repositoryVersion2110);
+
+    List<UpgradeGroupHolder> groups = m_upgradeHelper.createSequence(upgrade, context);
+    assertEquals(5, groups.size());
+
+    UpgradeGroupHolder serviceCheckGroup = groups.get(2);
+    assertEquals(ServiceCheckGrouping.class, serviceCheckGroup.groupClass);
+    assertEquals(3, serviceCheckGroup.items.size());
+
+    StageWrapper wrapper = serviceCheckGroup.items.get(0);
+    assertEquals(ServiceCheckGrouping.ServiceCheckStageWrapper.class, wrapper.getClass());
+    assertTrue(wrapper.getText().contains("ZooKeeper"));
+
+    // Do stacks cleanup
+    stackManagerMock.invalidateCurrentPaths();
+    ambariMetaInfo.init();
   }
 
+  @Test
+  public void testPrematureServiceChecks() throws Exception {
+    Map<String, UpgradePack> upgrades = ambariMetaInfo.getUpgradePacks("HDP", "2.1.1");
+    assertTrue(upgrades.containsKey("upgrade_test_checks"));
+    UpgradePack upgrade = upgrades.get("upgrade_test_checks");
+    assertNotNull(upgrade);
+
+    Cluster cluster = makeCluster();
+    cluster.deleteService("HDFS");
+    cluster.deleteService("YARN");
+    cluster.deleteService("ZOOKEEPER");
+
+    UpgradeContext context = getMockUpgradeContext(cluster, Direction.UPGRADE,
+        UpgradeType.ROLLING, repositoryVersion2110);
+
+    List<UpgradeGroupHolder> groups = m_upgradeHelper.createSequence(upgrade, context);
+
+    assertEquals(3, groups.size());
+
+    for (UpgradeGroupHolder holder : groups) {
+      assertFalse(ServiceCheckGrouping.class.equals(holder.groupClass));
+    }
+
+    // Do stacks cleanup
+    stackManagerMock.invalidateCurrentPaths();
+    ambariMetaInfo.init();
+  }
+
+
   /**
    * @param cluster
    * @param direction

ambari-server/src/test/resources/stacks/HDP/2.1.1/upgrades/upgrade_test_checks.xml

@@ -68,6 +68,7 @@
         <component>ZOOKEEPER_CLIENT</component>
       </service>
     </group>
+    
     <group name="CORE_MASTER" title="Core Masters">
       <service name="HDFS">
         <component>JOURNALNODE</component>
@@ -108,11 +109,18 @@
     
     <group name="SERVICE_CHECK_2" title="Post-Slave Service Checks" xsi:type="service-check">
       <priority>
+        <service>ZOOKEEPER</service>
         <service>HDFS</service>
         <service>YARN</service>
       </priority>
     </group>
     
+    <group name="OOZIE" title="Oozie">
+      <service name="OOZIE">
+        <component>OOZIE_SERVER</component>
+      </service>
+    </group>
+    
     
     <group xsi:type="cluster" name="POST_CLUSTER" title="Finalize Upgrade">
       <execute-stage title="Confirm Finalize">
@@ -213,5 +221,12 @@
         <upgrade />
       </component>
     </service>
+    <service name="OOZIE">
+      <component name="OOZIE_SERVER">
+        <upgrade>
+          <task xsi:type="restart-task" />
+        </upgrade>
+      </component>
+	</service>
   </processing>
 </upgrade>