AMBARI-18740. Perf: Automate deployment of PERF stack on 2500+ Ambari Agents.(vbrodetskyi)

contrib/utils/perf/deploy-gce-perf-cluster.py

@@ -0,0 +1,360 @@
+#!/usr/bin/env python
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import argparse
+import os
+import subprocess
+import sys
+import pprint
+import time
+import traceback
+import re
+import socket
+
+cluster_name="perf-cluster"
+ambari_repo_file_url="http://s3.amazonaws.com/dev.hortonworks.com/ambari/centos6/2.x/latest/trunk/ambaribn.repo"
+
+public_hostname_script="foo"
+hostname_script="foo"
+
+start_number=1
+number_of_agents_on_host=50
+
+
+class SSH:
+  """ Ssh implementation of this """
+  def __init__(self, user, sshkey_file, host, command, custom_option='', errorMessage = None):
+    self.user = user
+    self.sshkey_file = sshkey_file
+    self.host = host
+    self.command = command
+    self.errorMessage = errorMessage
+    self.custom_option = custom_option
+    pass
+
+
+  def run(self):
+    sshcommand = ["ssh",
+                  "-o", "ConnectTimeOut=180",
+                  "-o", "StrictHostKeyChecking=no",
+                  "-o", "BatchMode=yes",
+                  self.custom_option,
+                  "-i", self.sshkey_file,
+                  self.user + "@" + self.host, self.command]
+
+    if not self.custom_option:
+      del sshcommand[7]
+
+    i = 1
+    while True:
+      try:
+        sshstat = subprocess.Popen(sshcommand, stdout=subprocess.PIPE,
+                               stderr=subprocess.PIPE)
+        log = sshstat.communicate()
+        if sshstat.returncode != 0:
+          print "Executing SSH command on {0} failed: {1}".format(self.host, log)
+          print "\nRetrying SSH command one more time!"
+          if i >= 3:
+            break
+          i += 1
+          time.sleep(10)
+          continue
+        break
+      except:
+        print "Could not SSH to {0}, waiting for it to start".format(self.host)
+        i += 1
+        time.sleep(10)
+
+    if i >= 3:
+      print "Could not execute remote ssh command: " + ' '.join(sshcommand)
+      raise Exception("Could not connect to {0}. Giving up with erros: {1}".format(self.host, log))
+
+    errorMsg = log[1]
+    if self.errorMessage and sshstat.returncode != 0:
+      errorMsg = self.errorMessage + "\n" + errorMsg
+
+    print "SSH command execution finished"
+
+    return  {"exitstatus": sshstat.returncode, "log": log, "errormsg": errorMsg}
+
+
+class SCP:
+  """ SCP implementation that is thread based. The status can be returned using
+   status val """
+  def __init__(self, user, sshkey_file, host, inputFile, remote, errorMessage = None):
+    self.user = user
+    self.sshkey_file = sshkey_file
+    self.host = host
+    self.inputFile = inputFile
+    self.remote = remote
+    self.errorMessage = errorMessage
+    pass
+
+
+  def run(self):
+    scpcommand = ["scp",
+                  "-r",
+                  "-o", "ConnectTimeout=60",
+                  "-o", "BatchMode=yes",
+                  "-o", "StrictHostKeyChecking=no",
+                  "-i", self.sshkey_file, self.inputFile, self.user + "@" +
+                                                          self.host + ":" + self.remote]
+
+    i = 1
+    while True:
+      try:
+        scpstat = subprocess.Popen(scpcommand, stdout=subprocess.PIPE,
+                                   stderr=subprocess.PIPE)
+        log = scpstat.communicate()
+        if scpstat.returncode != 0:
+          print "Executing SCP command on {0} failed: {1}".format(self.host, log)
+          print "\nRetrying SCP command one more time!"
+          if i >= 3:
+            break
+          i += 1
+          time.sleep(10)
+          continue
+        break
+      except:
+        print "Could not SCP to {0}, waiting for it to start".format(self.host)
+        i += 1
+        time.sleep(10)
+
+      if i >= 3:
+        print "Could not execute remote scp command: " + ' '.join(scpcommand)
+        raise Exception("Could not connect to {0}. Giving up with erros: {1}".format(self.host, log))
+
+    errorMsg = log[1]
+    if self.errorMessage and scpstat.returncode != 0:
+      errorMsg = self.errorMessage + "\n" + errorMsg
+
+    print "SCP command execution finished"
+
+    return {"exitstatus": scpstat.returncode, "log": log, "errormsg": errorMsg}
+
+# main method to parse arguments from user and start work
+def main():
+  parser = argparse.ArgumentParser(
+    description='This script brings up a cluster with ambari installed, configured and started',
+    epilog='Only GCE is supported as of now!'
+  )
+
+  # options
+  parser.add_argument('--controller', type=str,
+                      action='store', help='GCE controller ip address.')
+
+  parser.add_argument('--key', type=str,
+                      action='store', help='Path to GCE ssh key.')
+
+  parser.add_argument('--cluster-prefix', type=str,
+                      action='store', help='Cluster name prefix.')
+
+  parser.add_argument('--agent-prefix', type=str,
+                      action='store', help='Agent name prefix.')
+
+  parser.add_argument('--agents-count', type=int,
+                      action='store', help='Agents count for whole cluster(multiples of 50).')
+
+  if len(sys.argv) <= 1:
+    parser.print_help()
+  else:
+    args = parser.parse_args()
+    do_work(args)
+
+# base method which process cluster deployment
+def deploy_cluster(args):
+  number_of_nodes=args.agents_count/number_of_agents_on_host
+  # trying to create cluster with needed params
+  print "Creating cluster {0}-{1} with {2} large nodes on centos6...".format(args.cluster_prefix, cluster_name, str(number_of_nodes))
+  execute_command(args, args.controller, "/usr/sbin/gce up {0}-{1} {2} --centos6 --large".format(args.cluster_prefix, cluster_name, str(number_of_nodes)),
+                  "Failed to create cluster, probably not enough resources!", "-tt")
+
+  # VMs are not accessible immediately
+  time.sleep(10)
+
+  # getting list of vms information like hostname and ip address
+  print "Getting list of virtual machines from cluster..."
+  vms = get_vms_list(args)
+
+  # check number of nodes in cluster to be the same as user asked
+  print "Checking count of created nodes in cluster..."
+  if not vms or len(vms) < number_of_nodes:
+    raise Exception("Can not bring up enough nodes. Requested {0}, but got: {1}. Probably not enough resources!".format(number_of_nodes, len(vms)))
+
+  print "GCE cluster was successfully created!"
+  pretty_print_vms(vms)
+
+  # installing/starting ambari-server and ambari-agents on each host
+  server_host_name = sorted(vms.items())[0][0]
+  server_installed=False
+
+  print "Creating server.sh script (which will be executed on server to install/configure/start ambari-server and ambari-agent)..."
+  create_server_script(args, server_host_name)
+
+  print "Creating agent.sh script (which will be executed on agent hosts to install/configure/start ambari-agent..."
+  create_agent_script(args, server_host_name)
+
+  time.sleep(10)
+  for (hostname, ip) in sorted(vms.items()):
+    print "=========================="
+    print "Working on {0}".format(hostname)
+    if not server_installed:
+      remote_path = "/server.sh"
+      local_path = "server.sh"
+      print "Copying server.sh to {0}...".format(hostname)
+      put_file(args, ip, local_path, remote_path, "Failed to copy file!")
+      print "Executing remote ssh command (set correct permissions and start executing server.sh in separate process) on {0}...".format(hostname)
+      execute_command(args, ip, "cd /; chmod 777 server.sh; nohup ./server.sh >/server.log 2>&1 &",
+                    "Install/configure/start server script failed!")
+      server_installed = True
+    else:
+      remote_path = "/agent.sh"
+      local_path = "agent.sh"
+      print "Copying agent.sh to {0}...".format(hostname)
+      put_file(args, ip, local_path, remote_path, "Failed to copy file!")
+      print "Executing remote ssh command (set correct permissions and start executing agent.sh in separate process) on {0}...".format(hostname)
+      execute_command(args, ip, "cd /; chmod 777 agent.sh; nohup ./agent.sh >/agent.log 2>&1 &",
+                    "Install/configure start agent script failed!")
+
+  print "All scripts where successfully copied and started on all hosts. " \
+        "\nPay attention that server.sh script need 5 minutes to finish and agent.sh need 3 minutes!"
+
+# check if all required params were passed by user
+# if all needed params available then start cluster deploy
+def do_work(args):
+  if not args.controller:
+    raise Failure("GCE controller ip address not defined!")
+
+  if not args.key:
+    raise Failure("Path to gce ssh key not defined!")
+
+  if not args.cluster_prefix:
+    raise Failure("Cluster name prefix not defined!")
+
+  if not args.agent_prefix:
+    raise Failure("Agent name prefix not defined!")
+
+  if not args.agents_count:
+    raise Failure("Agents count for whole cluster(multiples of 50) not defined!")
+
+  deploy_cluster(args)
+
+# creating server.sh script in the same dir where current script is located
+# server.sh script will install, configure and start ambari-server and ambari-agent on host
+def create_server_script(args, server_host_name):
+  file = open("server.sh", "w")
+
+  file.write("#!/bin/bash\n")
+
+  file.write("wget -O /etc/yum.repos.d/ambari.repo {0}\n".format(ambari_repo_file_url))
+  file.write("yum clean all; yum install git ambari-server ambari-agent -y\n")
+  file.write("cd /home; git clone https://github.com/apache/ambari.git\n")
+
+  file.write("cp -r /home/ambari/ambari-server/src/main/resources/stacks/PERF /var/lib/ambari-server/resources/stacks/PERF\n")
+  file.write("cp -r /home/ambari/ambari-server/src/main/resources/stacks/PERF /var/lib/ambari-agent/cache/stacks/PERF\n")
+
+  file.write("ambari-server setup -s\n")
+  file.write("sed -i -e 's/false/true/g' /var/lib/ambari-server/resources/stacks/PERF/1.0/metainfo.xml\n")
+  file.write("ambari-server start --skip-database-check\n")
+
+  file.write("sed -i -e 's/hostname=localhost/hostname={0}/g' /etc/ambari-agent/conf/ambari-agent.ini\n".format(server_host_name))
+  file.write("sed -i -e 's/agent]/agent]\\nhostname_script={0}\\npublic_hostname_script={1}\\n/1' /etc/ambari-agent/conf/ambari-agent.ini\n".format(hostname_script, public_hostname_script))
+  file.write("printf \"start={0}\\nnum={1}\\nprefix={2}\" > /etc/ambari-agent/conf/agent-multiplier.conf\n".format(start_number, number_of_agents_on_host, args.agent_prefix))
+  file.write("python /home/ambari/ambari-agent/conf/unix/agent-multiplier.py start\n")
+
+  file.write("exit 0")
+
+  file.close()
+
+# creating agent.sh script in the same dir where current script is located
+# agent.sh script will install, configure and start ambari-agent on host
+def create_agent_script(args, server_host_name):
+  file = open("agent.sh", "w")
+
+  file.write("#!/bin/bash\n")
+
+  file.write("wget -O /etc/yum.repos.d/ambari.repo {0}\n".format(ambari_repo_file_url))
+  file.write("yum clean all; yum install git ambari-agent -y\n")
+  file.write("cd /home; git clone https://github.com/apache/ambari.git\n")
+  file.write("cp -r /home/ambari/ambari-server/src/main/resources/stacks/PERF /var/lib/ambari-agent/cache/stacks/PERF\n")
+
+  file.write("sed -i -e 's/hostname=localhost/hostname={0}/g' /etc/ambari-agent/conf/ambari-agent.ini\n".format(server_host_name))
+  file.write("sed -i -e 's/agent]/agent]\\nhostname_script={0}\\npublic_hostname_script={1}\\n/1' /etc/ambari-agent/conf/ambari-agent.ini\n".format(hostname_script, public_hostname_script))
+  file.write("printf \"start={0}\\nnum={1}\\nprefix={2}\" > /etc/ambari-agent/conf/agent-multiplier.conf\n".format(start_number, number_of_agents_on_host, args.agent_prefix))
+  file.write("python /home/ambari/ambari-agent/conf/unix/agent-multiplier.py start\n")
+  file.write("exit 0")
+
+  file.close()
+
+# method to execute ssh commands via SSH class
+def execute_command(args, ip, cmd, fail_message, custom_option='', login='root'):
+  ssh = SSH(login, args.key, ip, cmd, custom_option, fail_message)
+  ssh_result = ssh.run()
+  status_code = ssh_result["exitstatus"]
+  if status_code != 0:
+    raise Exception(ssh_result["errormsg"])
+
+  return ssh_result["log"][0]
+
+# method to copy file from local to remote host via SCP class
+def put_file(args, ip, local_file, remote_file, fail_message, login='root'):
+  scp = SCP(login, args.key, ip, local_file,
+            remote_file, fail_message)
+  scp_result = scp.run()
+  status_code = scp_result["exitstatus"]
+  if status_code != 0:
+    raise Exception(scp_result["errormsg"])
+
+  return scp_result["log"][0]
+
+# method to parse "gce fqdn {cluster-name}" command output and get hosts and ips pairs for every host in cluster
+def get_vms_list(args):
+  gce_fqdb_cmd = '/usr/sbin/gce fqdn {0}-{1}'.format(
+    args.cluster_prefix, cluster_name)
+  out = execute_command(args, args.controller, gce_fqdb_cmd, "Failed to get VMs list!", "-tt")
+  lines = out.split('\n')
+  #print "LINES=" + str(lines)
+  if lines[0].startswith("Using profile") and not lines[1].strip():
+    result = {}
+    for s in lines[2:]:  # Ignore non-meaningful lines
+      if not s:
+        continue
+      match = re.match(r'^([\d\.]*)\s+([\w\.-]*)\s+([\w\.-]*)\s$', s, re.M)
+      if match:
+        result[match.group(2)] = match.group(1)
+      else:
+        raise Exception('Can not parse "{0}"'.format(s))
+    return result
+  else:
+    raise Exception('Can not parse "{0}"'.format(lines))
+
+
+def pretty_print_vms(vms):
+  print "----------------------------"
+  print "server ip: {0}".format(sorted(vms.items())[0][1])
+  print "Hostnames of nodes in cluster:"
+  for (hostname, ip) in sorted(vms.items()):
+    print hostname
+  print "----------------------------"
+
+
+if __name__ == "__main__":
+  main()
+
+