AMBARI-5722. All Services Fail To Deploy Due To Agent Parsing Exception (dlysnichenko)

ambari-agent/src/main/python/ambari_agent/manifestGenerator.py

@@ -29,6 +29,7 @@ from ambari_agent import AgentException
 
 HOSTS_LIST_KEY = "all_hosts"
 PING_PORTS_KEY = "all_ping_ports"
+AMBARI_SERVER_HOST = "ambari_server_host"
 
 logger = logging.getLogger()
 
@@ -108,6 +109,7 @@ def decompressClusterHostInfo(clusterHostInfo):
   #Pop info not related to host roles  
   hostsList = info.pop(HOSTS_LIST_KEY)
   pingPorts = info.pop(PING_PORTS_KEY)
+  ambariServerHost = info.pop(AMBARI_SERVER_HOST)
 
   decompressedMap = {}
 
@@ -127,6 +129,8 @@ def decompressClusterHostInfo(clusterHostInfo):
   decompressedMap[PING_PORTS_KEY] = pingPorts
   #Add hosts list to result
   decompressedMap[HOSTS_LIST_KEY] = hostsList
+  #Add ambari-server host to result
+  decompressedMap[AMBARI_SERVER_HOST] = ambariServerHost
   
   return decompressedMap
 

ambari-agent/src/test/python/ambari_agent/TestManifestGenerator.py

@@ -192,31 +192,39 @@ class TestManifestGenerator(TestCase):
       exceptionWasTrown = True
     self.assertTrue(exceptionWasTrown)
     
-    def testDecompressClusterHostInfo(self):
-        
-      info = { "jtnode_host"        : ["5"],
-               "hbase_master_hosts" : ["5"],
-               "all_hosts"          : ["h8", "h9", "h5", "h4", "h7", "h6", "h1", "h3", "h2", "h10"],
-               "namenode_host"      : ["6"],
-               "mapred_tt_hosts"    : ["0", "7-9", "2","3", "5"],
-               "slave_hosts"        : ["3", "0", "1", "5-9"],
-               "snamenode_host"     : ["8"],
-               "ping_ports"         : ["8670:1,5-8", "8673:9", "8672:0,4", "8671:2,3"],
-               "hbase_rs_hosts"     : ["3", "1", "5", "8", "9"]
-      }
-        
-      decompressedInfo = manifestGenerator.decompressClusterHostInfo(clusterHostInfo)
-      
-      self.assertTrue(decompressedInfo.has_key("all_hosts"))
-      
-      allHosts = decompressedInfo.pop("all_hosts")
-      
-      self.assertEquals(info.get("all_hosts"), decompressedInfo.get("all_hosts"))
-      
-      pingPorts = decompressedInfo.pop("all_ping_ports")
-      
-      self.assertEquals(pingPorts, manifestGenerator.convertMappedRangeToList(info.get("all_ping_ports")))
-      
-      for k,v in decompressedInfo.items():
-        self.assertEquals(v, manifestGenerator.convertRangeToList(info.get(k)))
+  def testDecompressClusterHostInfo(self):
+
+    all_hosts_key = "all_hosts"
+    all_ping_ports_key = "all_ping_ports"
+    ambari_server_host_key = "ambari_server_host"
+    info = { "jtnode_host"        : ["5"],
+             "hbase_master_hosts" : ["5"],
+             all_hosts_key: ["h8", "h9", "h5", "h4", "h7", "h6", "h1", "h3", "h2", "h10"],
+             "namenode_host"      : ["6"],
+             "mapred_tt_hosts"    : ["0", "7-9", "2","3", "5"],
+             "slave_hosts"        : ["3", "0", "1", "5-9"],
+             "snamenode_host"     : ["8"],
+             all_ping_ports_key: ["8670:1,5-8", "8673:9", "8672:0,4", "8671:2,3"],
+             "hbase_rs_hosts"     : ["3", "1", "5", "8", "9"],
+             ambari_server_host_key: ["h0"]
+    }
+
+    decompressedInfo = manifestGenerator.decompressClusterHostInfo(info)
+
+    self.assertTrue(decompressedInfo.has_key(all_hosts_key))
+
+    self.assertEquals(info.pop(all_hosts_key), decompressedInfo.get(all_hosts_key))
+
+    self.assertEquals(['8672', '8670', '8671', '8671', '8672',
+                       '8670', '8670', '8670', '8670', '8673'],
+                      decompressedInfo.get(all_ping_ports_key))
+
+    self.assertEquals(["h0"], decompressedInfo.get(ambari_server_host_key))
+
+    for k,v in decompressedInfo.items():
+      if k is all_ping_ports_key:
+        continue # Skip checking this list
+      # Check that list contains only host names
+      non_converted = [x for x in v if not x.startswith("h")]
+      self.assertEquals(0, len(non_converted))
 

ambari-server/src/main/java/org/apache/ambari/server/utils/StageUtils.java

@@ -48,6 +48,7 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
@@ -236,12 +237,6 @@ public class StageUtils {
     List<String> hostsList = new ArrayList<String>(hostsSet);
 
     //     Fill host roles
-    // Fill server host
-    TreeSet<Integer> serverHost = new TreeSet<Integer>();
-    int hostIndex = hostsList.indexOf(getHostName());
-    serverHost.add(hostIndex);
-    hostRolesInfo.put(AMBARI_SERVER_HOST, serverHost);
-
     // Fill hosts for services
     for (Entry<String, Service> serviceEntry : cluster.getServices().entrySet()) {
 
@@ -273,7 +268,7 @@ public class StageUtils {
               hostRolesInfo.put(roleName, hostsForComponentsHost);
             }
 
-            hostIndex = hostsList.indexOf(hostName);
+            int hostIndex = hostsList.indexOf(hostName);
             //Add index of host to current host role
             hostsForComponentsHost.add(hostIndex);
           }
@@ -288,7 +283,7 @@ public class StageUtils {
                 hostRolesInfo.put(decomRoleName, hostsForComponentsHost);
               }
 
-              hostIndex = hostsList.indexOf(hostName);
+              int hostIndex = hostsList.indexOf(hostName);
               //Add index of host to current host role
               hostsForComponentsHost.add(hostIndex);
             }
@@ -308,6 +303,17 @@ public class StageUtils {
     clusterHostInfo.put(HOSTS_LIST, hostsSet);
     clusterHostInfo.put(PORTS, replaceMappedRanges(portsList));
 
+    // Fill server host
+    /*
+     * Note: We don't replace server host name by an index (like we do
+     * with component hostnames), because if ambari-agent is not installed
+     * at ambari-server host, then allHosts map will not contain
+     * ambari-server hostname.
+     */
+    TreeSet<String> serverHost = new TreeSet<String>();
+    serverHost.add(getHostName());
+    clusterHostInfo.put(AMBARI_SERVER_HOST, serverHost);
+
     return clusterHostInfo;
   }
 

ambari-server/src/test/java/org/apache/ambari/server/utils/TestStageUtils.java

@@ -164,13 +164,12 @@ public class TestStageUtils {
   }
 
   @Test
-  @Ignore
   public void testGetClusterHostInfo() throws AmbariException, UnknownHostException {
     Clusters fsm = injector.getInstance(Clusters.class);
-    String h1 = "h1";
+    String h0 = "h0";
 
     List<String> hostList = new ArrayList<String>();
-    hostList.add(h1);
+    hostList.add("h1");
     hostList.add("h2");
     hostList.add("h3");
     hostList.add("h4");
@@ -182,7 +181,7 @@ public class TestStageUtils {
     hostList.add("h10");
 
     mockStaticPartial(StageUtils.class, "getHostName");
-    expect(StageUtils.getHostName()).andReturn(h1).anyTimes();
+    expect(StageUtils.getHostName()).andReturn(h0).anyTimes();
     replayAll();
 
     List<Integer> pingPorts = Arrays.asList(StageUtils.DEFAULT_PING_PORT,
@@ -226,14 +225,14 @@ public class TestStageUtils {
     //Add HBASE service
     Map<String, List<Integer>> hbaseTopology = new HashMap<String, List<Integer>>(); 
     hbaseTopology.put("HBASE_MASTER", Collections.singletonList(5));
-    List<Integer> regionServiceIndexes = Arrays.asList(1,3,5,8,9);;
+    List<Integer> regionServiceIndexes = Arrays.asList(1,3,5,8,9);
     hbaseTopology.put("HBASE_REGIONSERVER", regionServiceIndexes);
     addService(fsm.getCluster("c1"), hostList, hbaseTopology , "HBASE", injector);
     
     //Add MAPREDUCE service
     Map<String, List<Integer>> mrTopology = new HashMap<String, List<Integer>>(); 
     mrTopology.put("JOBTRACKER", Collections.singletonList(5));
-    List<Integer> taskTrackerIndexes = Arrays.asList(1,2,3,4,5,7,9);;
+    List<Integer> taskTrackerIndexes = Arrays.asList(1,2,3,4,5,7,9);
     mrTopology.put("TASKTRACKER", taskTrackerIndexes);
     addService(fsm.getCluster("c1"), hostList, mrTopology , "MAPREDUCE", injector);
     
@@ -309,9 +308,8 @@ public class TestStageUtils {
     // check server hostname field
     assertTrue(info.containsKey(StageUtils.AMBARI_SERVER_HOST));
     Set<String> serverHost = info.get(StageUtils.AMBARI_SERVER_HOST);
-    assertEquals(1, serverHost.toArray().length);
-    int serverHostIndex = Integer.valueOf(serverHost.iterator().next());
-    assertEquals(h1, allHostsList.get(serverHostIndex));
+    assertEquals(1, serverHost.size());
+    assertEquals(h0, serverHost.iterator().next());
   }
 
   private void checkServiceCompression(Map<String, Set<String>> info,