AMBARI-8724. Rolling Upgrade - Upgrade Pack to restart standby ResourceManager first, failover, and then upgrade the remaining RM (ncole)

ambari-server/src/main/java/org/apache/ambari/server/stack/MasterHostResolver.java

@@ -18,21 +18,21 @@
 
 package org.apache.ambari.server.stack;
 
-import com.google.common.reflect.TypeToken;
-import org.apache.ambari.server.state.Cluster;
-import org.apache.ambari.server.utils.HTTPUtils;
-
 import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.ambari.server.state.Cluster;
+import org.apache.ambari.server.utils.HTTPUtils;
 import org.apache.ambari.server.utils.StageUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.reflect.TypeToken;
+
 
 public class MasterHostResolver {
 
@@ -42,7 +42,8 @@ public class MasterHostResolver {
 
   enum Service {
     HDFS,
-    HBASE
+    HBASE,
+    YARN
   }
 
   /**
@@ -53,10 +54,6 @@ public class MasterHostResolver {
     STANDBY
   }
 
-  public MasterHostResolver() {
-    ;
-  }
-
   public MasterHostResolver(Cluster cluster) {
     this.cluster = cluster;
   }
@@ -74,15 +71,21 @@ public class MasterHostResolver {
       return null;
     }
 
-    Service s = Service.valueOf(serviceName.toUpperCase());
-
     Set<String> componentHosts = cluster.getHosts(serviceName, componentName);
     if (0 == componentHosts.size()) {
       return null;
     }
-
+    
     hostsType.hosts = componentHosts;
 
+    Service s = null;
+    try {
+      s = Service.valueOf(serviceName.toUpperCase());
+    } catch (Exception e) {
+      // !!! nothing to do
+      return hostsType;
+    }
+
     switch (s) {
       case HDFS:
         if (componentName.equalsIgnoreCase("NAMENODE")) {
@@ -90,13 +93,19 @@ public class MasterHostResolver {
           if (pair != null) {
             hostsType.master = pair.containsKey(Status.ACTIVE) ? pair.get(Status.ACTIVE) :  null;
             hostsType.secondary = pair.containsKey(Status.STANDBY) ? pair.get(Status.STANDBY) :  null;
+          } else {
+            hostsType.master = componentHosts.iterator().next();
           }
         }
         break;
+      case YARN:
+        if (componentName.equalsIgnoreCase("RESOURCEMANAGER")) {
+          resolveResourceManagers(hostsType);
+        }
+        break;
       case HBASE:
-        if (componentName.equalsIgnoreCase("HBASE_REGIONSERVER")) {
-          // TODO Rolling Upgrade, fill for this Component.
-          ;
+        if (componentName.equalsIgnoreCase("HBASE_MASTER")) {
+          resolveHBaseMasters(hostsType);
         }
         break;
     }
@@ -110,42 +119,20 @@ public class MasterHostResolver {
    */
   private Map<Status, String> getNameNodePair(Set<String> hosts) {
     Map<Status, String> stateToHost = new HashMap<Status, String>();
+    
     if (hosts != null && hosts.size() == 2) {
-      Iterator iter = hosts.iterator();
-
-      while(iter.hasNext()) {
-        String hostname = (String) iter.next();
-        try {
-          // TODO Rolling Upgrade, don't hardcode jmx port number
-          // E.g.,
-          // dfs.namenode.http-address.dev.nn1 : c6401.ambari.apache.org:50070
-          // dfs.namenode.http-address.dev.nn2 : c6402.ambari.apache.org:50070
-          String endpoint = "http://" + hostname + ":50070/jmx?qry=Hadoop:service=NameNode,name=NameNodeStatus";
-          String response = HTTPUtils.requestURL(endpoint);
-
-          if (response != null && !response.isEmpty()) {
-            Map<String, ArrayList<HashMap<String, String>>> nameNodeInfo = new HashMap<String, ArrayList<HashMap<String, String>>>();
-            Type type = new TypeToken<Map<String, ArrayList<HashMap<String, String>>>>() {}.getType();
-            nameNodeInfo = StageUtils.getGson().fromJson(response, type);
-
-            try {
-              String state = nameNodeInfo.get("beans").get(0).get("State");
-
-              if (state.equalsIgnoreCase(Status.ACTIVE.toString()) || state.equalsIgnoreCase(Status.STANDBY.toString())) {
-                Status status = Status.valueOf(state.toUpperCase());
-                stateToHost.put(status, hostname);
-              }
-            } catch (Exception e) {
-              throw new Exception("Response from endpoint " + endpoint + " was not formatted correctly. Value: " + response);
-            }
-          } else {
-            throw new Exception("Response from endpoint " + endpoint + " was empty.");
+      for (String hostname : hosts) {
+        String state = queryJmxBeanValue(hostname, 50070,
+            "Hadoop:service=NameNode,name=NameNodeStatus", "State", true);
+        
+        if (null != state &&
+            (state.equalsIgnoreCase(Status.ACTIVE.toString()) ||
+                state.equalsIgnoreCase(Status.STANDBY.toString()))) {
+            Status status = Status.valueOf(state.toUpperCase());
+            stateToHost.put(status, hostname);
           }
-        } catch (Exception e) {
-          LOG.warn("Failed to parse namenode jmx endpoint to get state for host " + hostname + ". Error: " + e.getMessage());
         }
-      }
-
+        
       if (stateToHost.containsKey(Status.ACTIVE) && stateToHost.containsKey(Status.STANDBY) && !stateToHost.get(Status.ACTIVE).equalsIgnoreCase(stateToHost.get(Status.STANDBY))) {
         return stateToHost;
       }
@@ -153,4 +140,75 @@ public class MasterHostResolver {
 
     return null;
   }
+  
+  private void resolveResourceManagers(HostsType hostType) {
+    // !!! for RM, only the master returns jmx
+    Set<String> orderedHosts = new LinkedHashSet<String>(hostType.hosts);
+    
+    for (String hostname : hostType.hosts) {
+      
+      String value = queryJmxBeanValue(hostname, 8088,
+          "Hadoop:service=ResourceManager,name=RMNMInfo", "modelerType", true);
+      
+      if (null != value) {
+        if (null == hostType.master) {
+          hostType.master = hostname;
+        }
+        
+        // !!! quick and dirty to make sure the master is last in the list
+        orderedHosts.remove(hostname);
+        orderedHosts.add(hostname);
+      }
+      
+    }
+    hostType.hosts = orderedHosts;
+  }
+  
+  private void resolveHBaseMasters(HostsType hostsType) {
+    
+    for (String hostname : hostsType.hosts) {
+      
+      String value = queryJmxBeanValue(hostname, 60010,
+          "Hadoop:service=HBase,name=Master,sub=Server", "tag.isActiveMaster", false);
+      
+      if (null != value) {
+        Boolean bool = Boolean.valueOf(value);
+        if (bool.booleanValue()) {
+          hostsType.master = hostname;
+        } else {
+          hostsType.secondary = hostname;
+        }
+      }
+      
+    }
+  }
+  
+  private String queryJmxBeanValue(String hostname, int port, String beanName, String attributeName,
+      boolean asQuery) {
+    
+    String endPoint = asQuery ?
+        String.format("http://%s:%s/jmx?qry=%s", hostname, port, beanName) :
+          String.format("http://%s:%s/jmx?get=%s::%s", hostname, port, beanName, attributeName);
+    
+    String response = HTTPUtils.requestURL(endPoint);
+    
+    if (null == response || response.isEmpty()) {
+      return null;
+    }
+    
+    Type type = new TypeToken<Map<String, ArrayList<HashMap<String, String>>>>() {}.getType();
+
+    try {
+      Map<String, ArrayList<HashMap<String, String>>> jmxBeans =
+          StageUtils.getGson().fromJson(response, type);
+      
+      return jmxBeans.get("beans").get(0).get(attributeName);
+    } catch (Exception e) {
+      LOG.info("Could not load JMX from {}/{} from {}", beanName, attributeName, hostname, e);
+    }
+    
+    return null;
+    
+  }
+  
 }

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

@@ -100,19 +100,17 @@ public class UpgradeHelper {
             continue;
           }
 
-          Set<String> componentHosts = cluster.getHosts(service.serviceName, component);
-          if (0 == componentHosts.size()) {
+          HostsType hostsType = mhr.getMasterAndHosts(service.serviceName, component);
+          if (null == hostsType) {
             continue;
           }
-          HostsType hostsType = new HostsType();
-          hostsType.hosts = componentHosts;
-
+          
           ProcessingComponent pc = allTasks.get(service.serviceName).get(component);
-
+          
           // Special case for NAMENODE
           if (service.serviceName.equalsIgnoreCase("HDFS") && component.equalsIgnoreCase("NAMENODE")) {
-            hostsType = mhr.getMasterAndHosts(service.serviceName, component);
-            if (hostsType != null && hostsType.master != null && componentHosts.contains(hostsType.master) && hostsType.secondary != null && componentHosts.contains(hostsType.secondary)) {
+            // !!! revisit if needed
+            if (hostsType.master != null && hostsType.secondary != null) {
               // The order is important, first do the standby, then the active namenode.
               Set<String> order = new LinkedHashSet<String>();
 
@@ -122,15 +120,14 @@ public class UpgradeHelper {
 
               // Override the hosts with the ordered collection
               hostsType.hosts = order;
-              builder.add(hostsType, service.serviceName, pc);
+              
             } else {
-              throw new AmbariException(MessageFormat.format("Could not find active and standby namenodes using hosts: {0}", StringUtils.join(componentHosts, ", ").toString()));
+//              throw new AmbariException(MessageFormat.format("Could not find active and standby namenodes using hosts: {0}", StringUtils.join(hostsType.hosts, ", ").toString()));
             }
-          }
-          /*
-          TODO Rolling Upgrade, write logic for HBASE
-          */
-          else {
+            
+            builder.add(hostsType, service.serviceName, pc);
+            
+          } else {
             builder.add(hostsType, service.serviceName, pc);
           }
         }

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

@@ -19,6 +19,7 @@ package org.apache.ambari.server.state.stack.upgrade;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 
@@ -31,10 +32,8 @@ import javax.xml.bind.annotation.XmlTransient;
 import javax.xml.bind.annotation.XmlType;
 
 import org.apache.ambari.server.stack.HostsType;
+import org.apache.ambari.server.stack.MasterHostResolver;
 import org.apache.ambari.server.state.Cluster;
-import org.apache.ambari.server.state.UpgradeHelper;
-import org.apache.ambari.server.state.cluster.ClusterImpl;
-import org.apache.ambari.server.state.stack.UpgradePack;
 import org.apache.ambari.server.state.stack.UpgradePack.ProcessingComponent;
 
 /**
@@ -73,13 +72,13 @@ public class ClusterGrouping extends Grouping {
   }
 
   public class ClusterBuilder extends StageWrapperBuilder {
-    private Cluster m_cluster = null;
+    private MasterHostResolver m_resolver = null;
 
     /**
      * @param cluster the cluster to use with this builder
      */
     public void setHelpers(Cluster cluster) {
-      m_cluster = cluster;
+      m_resolver = new MasterHostResolver(cluster);
     }
 
     @Override
@@ -101,15 +100,28 @@ public class ClusterGrouping extends Grouping {
         StageWrapper wrapper = null;
 
         if (null != execution.service && null != execution.component) {
-          Set<String> hosts = m_cluster.getHosts(execution.service, execution.component);
+
+          HostsType hosts = m_resolver.getMasterAndHosts(execution.service, execution.component);
+
+          if (null == hosts) {
+            continue;
+          }
+
+          Set<String> realHosts = new LinkedHashSet<String>(hosts.hosts);
+
           // !!! FIXME other types
-          if (hosts.size() > 0 && task.getType() == Task.Type.EXECUTE) {
+          if (task.getType() == Task.Type.EXECUTE) {
+            ExecuteTask et = (ExecuteTask) task;
+
+            if (null != et.hosts && "master".equals(et.hosts) && null != hosts.master) {
+              realHosts = Collections.singleton(hosts.master);
+            }
+
             wrapper = new StageWrapper(
                 StageWrapper.Type.RU_TASKS,
                 execution.title,
-                new TaskWrapper(execution.service, execution.component, hosts, task));
+                new TaskWrapper(execution.service, execution.component, realHosts, task));
           }
-
         } else {
           switch (task.getType()) {
             case MANUAL:

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

@@ -130,7 +130,7 @@ public class ColocatedGrouping extends Grouping {
         LOG.debug("RU initial: {}", initialBatch);
         LOG.debug("RU final: {}", finalBatches);
       }
-
+      
       results.addAll(fromProxies(initialBatch));
 
       // !!! TODO when manual tasks are ready
@@ -157,7 +157,7 @@ public class ColocatedGrouping extends Grouping {
 
         // !!! stage per host, per type
         StageWrapper wrapper = null;
-        StageWrapper execwrapper = null;
+        List<StageWrapper> execwrappers = new ArrayList<StageWrapper>();
 
         for (TaskProxy t : entry.getValue()) {
           serviceChecks.add(t.service);
@@ -167,19 +167,18 @@ public class ColocatedGrouping extends Grouping {
               wrapper = new StageWrapper(StageWrapper.Type.RU_TASKS, t.message, t.getTasksArray());
             }
           } else {
-            if (null == execwrapper) {
-              execwrapper = new StageWrapper(StageWrapper.Type.RESTART, t.message, t.getTasksArray());
-            }
+            execwrappers.add(new StageWrapper(StageWrapper.Type.RESTART, t.message, t.getTasksArray()));
           }
         }
 
         if (null != wrapper) {
           results.add(wrapper);
         }
-
-        if (null != execwrapper) {
-          results.add(execwrapper);
+        
+        if (execwrappers.size() > 0) {
+          results.addAll(execwrappers);
         }
+
       }
 
       if (serviceChecks.size() > 0) {

ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/upgrade-2.2.xml

@@ -31,8 +31,8 @@
 
 <upgrade xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
   <target>2.2.*.*</target>
-  <order>
 
+  <order>
     <group name="ZOOKEEPER" title="Zookeeper">
       <service name="ZOOKEEPER">
         <component>ZOOKEEPER_SERVER</component>
@@ -54,6 +54,7 @@
         <component>APP_TIMELINE_SERVER</component>
         <component>RESOURCEMANAGER</component>
       </service>
+
       <service name="HBASE">
         <component>HBASE_MASTER</component>
       </service>
@@ -63,7 +64,7 @@
       <service name="HDFS">
         <component>DATANODE</component>
       </service>
-
+      
       <service name="HBASE">
         <component>HBASE_REGIONSERVER</component>
       </service>
@@ -78,20 +79,22 @@
       </batch>
     </group>
     
-    <!--
     <group xsi:type="cluster" name="POST_CLUSTER" title="Finalize Upgrade">
+      <!-- 
       <execute-stage title="Confirm Finalize">
         <task xsi:type="manual">
           <message>Please confirm you are ready to finalize</message>
         </task>
       </execute-stage>
+      -->      
       <execute-stage service="HDFS" component="NAMENODE" title="Execute HDFS Finalize">
-        <task xsi:type="execute">
-          <command>ls</command>
+        <task xsi:type="execute" hosts="master">
+          <first>su - {{hadoop-env/hdfs_user}} -c 'hdfs dfsadmin -rollingUpgrade finalize'</first>
+          <command>su - {{hadoop-env/hdfs_user}} -c 'hdfs dfsadmin -safemode leave'</command>
         </task>
       </execute-stage>
     </group>
-    -->    
+   
   </order>
   
   
@@ -242,7 +245,7 @@
     <service name="HBASE">
       <component name="HBASE_MASTER">
         <pre-upgrade>
-          <task xsi:type="execute">
+          <task xsi:type="execute" hosts="master">
             <command>su - {{hbase-env/hbase_user}} -c 'echo "snapshot_all" | hbase shell'</command>
           </task>
         </pre-upgrade>

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

@@ -17,11 +17,15 @@
  */
 package org.apache.ambari.server.state;
 
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 
@@ -34,17 +38,16 @@ import org.apache.ambari.server.stack.HostsType;
 import org.apache.ambari.server.stack.MasterHostResolver;
 import org.apache.ambari.server.state.UpgradeHelper.UpgradeGroupHolder;
 import org.apache.ambari.server.state.stack.UpgradePack;
+import org.easymock.EasyMock;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.Mockito;
 
+import com.google.inject.AbstractModule;
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.google.inject.persist.PersistService;
-import org.mockito.Mockito;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 /**
  * Tests the {@link UpgradeHelper} class
@@ -65,8 +68,8 @@ public class UpgradeHelperTest {
     helper = injector.getInstance(OrmTestHelper.class);
     ambariMetaInfo = injector.getInstance(AmbariMetaInfo.class);
     ambariMetaInfo.init();
-
-    m_masterHostResolver = mock(MasterHostResolver.class);
+    
+    m_masterHostResolver = EasyMock.createMock(MasterHostResolver.class);
   }
 
   @After
@@ -86,15 +89,12 @@ public class UpgradeHelperTest {
 
     Cluster cluster = makeCluster();
 
-    HostsType hostsType = new HostsType();
-    hostsType.hosts = cluster.getHosts("HDFS", "NAMENODE");
-    hostsType.master = "h1";
-    hostsType.secondary = "h2";
-    when(m_masterHostResolver.getMasterAndHosts(Mockito.matches("HDFS"), Mockito.matches("NAMENODE"))).thenReturn(hostsType);
-
+    
     UpgradeHelper helper = new UpgradeHelper();
     List<UpgradeGroupHolder> groups = helper.createUpgrade(cluster, m_masterHostResolver, upgrade);
 
+    
+    
     assertEquals(5, groups.size());
 
     assertEquals("PRE_CLUSTER", groups.get(0).name);
@@ -167,8 +167,39 @@ public class UpgradeHelperTest {
     sc.addServiceComponentHost("h1");
     sc.addServiceComponentHost("h3");
 
-    return c;
+    HostsType type = new HostsType();
+    type.hosts = new HashSet<String>(Arrays.asList("h1", "h2", "h3"));
+    expect(m_masterHostResolver.getMasterAndHosts("ZOOKEEPER", "ZOOKEEPER_SERVER")).andReturn(type).anyTimes();
+    
+    type = new HostsType();
+    type.hosts = new HashSet<String>(Arrays.asList("h1", "h2"));
+    type.master = "h1";
+    type.secondary = "h2";
+    expect(m_masterHostResolver.getMasterAndHosts("HDFS", "NAMENODE")).andReturn(type).anyTimes();
+    
+    type = new HostsType();
+    type.hosts = new HashSet<String>(Arrays.asList("h2", "h3"));
+    expect(m_masterHostResolver.getMasterAndHosts("HDFS", "DATANODE")).andReturn(type).anyTimes();
+    
+    type = new HostsType();
+    type.hosts = new HashSet<String>(Arrays.asList("h2"));
+    expect(m_masterHostResolver.getMasterAndHosts("YARN", "RESOURCEMANAGER")).andReturn(type).anyTimes();
+
+    type = new HostsType();
+    type.hosts = new HashSet<String>(Arrays.asList("h1", "h3"));
+    expect(m_masterHostResolver.getMasterAndHosts("YARN", "NODEMANAGER")).andReturn(type).anyTimes();
+    
+    
+    replay(m_masterHostResolver);
 
+    return c;
+  }
+  
+  
+  private static class MockModule extends AbstractModule {
+    protected void configure() {
+      
+    }
   }
 
 }