ambari/contrib/agent-simulator/cluster.py

"""
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 subprocess
import datetime
from config import Config
from docker import Docker
from vm import VM
import os
import time
from log import Log
from data import Data


class Cluster:
    """
    The Cluster instance holds a list of VMs,
    it has the methods to request cluster, generate information and run all Ambari-agent and Ambari-server
    """

    # The constants represent the state of the cluster
    # A newly requested cluster is in FREE state
    STATE_FREE = "FREE"
    # A cluster with running Ambari-server and Ambari-agents is in RUNNING state
    STATE_RUNNING = "RUNNING"
    # A cluster is merged into another cluster and running, is in MERGE state
    # the name of the extended cluster is directly following the state String in JSON
    STATE_MERGE = "MERGE"

    def __init__(self):
        self.cluster_name = ""
        self.state = ""
        self.create_time = ""
        # The list should only has one or zero VM, which holds the Ambari-server
        self.ambari_server_vm = []
        # The list of VMs, with Ambari-agents directly inside (not in Docker)
        self.service_server_vm_list = []
        # The list of VMs, each will hold multiple Docker containers with Ambari-agent inside
        self.ambari_agent_vm_list = []

    def _get_int_interval(self, int_list):
        """
        get the interval of the integer list
        example: input[4,5,6,1,2,3], output [(1,3),(4,6)]
        example: input[4,5,6,100,2,3], output [(2,3),(4,6),(100,100)]
        :param int_list: the list of integer
        :return: a tuple, each tuple has 2 integer, representing one interval
        """
        interval_list = []
        int_list.sort()

        begin = None
        end = None
        for integer in int_list:
            if begin is None:
                begin = integer
                end = integer
            else:
                if integer == end + 1:
                    end = integer
                else:
                    interval_list.append((begin, end))
                    begin = integer
                    end = integer

        if begin is not None:
            interval_list.append((begin, end))

        return interval_list

    def print_description(self):
        print "cluster name: ", self.cluster_name
        print "create time: ", self.create_time
        print "state: ", self.state
        print

        print "Ambari Server: "
        ambari_server_vm = self.get_ambari_server_vm()
        if ambari_server_vm is None:
            print "None"
        else:
            print ambari_server_vm.domain_name, " ", ambari_server_vm.external_ip, " ",\
                ambari_server_vm.weave_internal_ip
        print

        print "Service Server with Ambari Agent directly installed: "
        if len(self.service_server_vm_list) == 0:
            print "None"
        for vm in self.service_server_vm_list:
            print vm.weave_domain_name, " ", vm.external_ip, " ", vm.weave_internal_ip
        print

        print "Ambari Agent in Docker Container: "
        int_list = []
        for vm in self.ambari_agent_vm_list:
            for docker in vm.docker_list:
                int_list.append(int(docker.get_index()))
        interval_list = self._get_int_interval(int_list)
        for interval in interval_list:
            interval_str = ""
            if interval[0] == interval[1]:
                interval_str = str(interval(0))
            else:
                interval_str = "[{0}-{1}]".format(interval[0], interval[1])
            print Docker.get_pattern_presentation(self.cluster_name, interval_str)
        print

    def get_agent_vm(self, vm_ip):
        """
        get the VM instance which holds Docker containers from the cluster instance
        :param vm_ip: the external IP of the target VM
        :return: the VM instance with the specified iP
        """
        for vm in self.ambari_agent_vm_list:
            if vm.external_ip == vm_ip:
                return vm

    def get_ambari_server_vm(self):
        """
        get the VM instance which hold the Ambari-server
        :return: the VM instance hold the Ambari-server, or None if no Ambari-server in this cluster
        """
        for vm in self.ambari_server_vm:
            return vm
        return None

    def get_service_server_vm(self, vm_ip):
        """
        get the VM instance which directly hold the Ambari-agent
        :param vm_ip: the external IP of the target VM
        :return:
        """
        for vm in self.service_server_vm_list:
            if vm.external_ip == vm_ip:
                return vm

    def to_json(self):
        """
        create a map to hold the information of the Cluster instance
        :return: A map, which is JSON format object.
        """
        cluster = {}
        cluster["cluster_name"] = self.cluster_name
        cluster["create_time"] = self.create_time
        cluster["state"] = self.state

        cluster["ambari_server_vm"] = []
        for vm in self.ambari_server_vm:
            cluster["ambari_server_vm"].append(vm.to_json())

        cluster["service_server_vm_list"] = []
        for vm in self.service_server_vm_list:
            cluster["service_server_vm_list"].append(vm.to_json())

        cluster["ambari_agent_vm_list"] = []
        for vm in self.ambari_agent_vm_list:
            cluster["ambari_agent_vm_list"].append(vm.to_json())

        return cluster

    @staticmethod
    def load_from_json(cluster_name):
        """
        load the cluster information from json file
        :param cluster_name: the name of the cluster
        :return: a Cluster instance or None if no such cluster
        """
        data = Data()
        json_data = data.read_cluster_json(cluster_name)
        if json_data is None:
            return None

        ambari_server_vm = []
        service_server_vm_list = []
        ambari_agent_vm_list = []

        for vm_json in json_data["ambari_server_vm"]:
            ambari_server_vm.append(VM.load_from_json(vm_json))

        for vm_json in json_data["service_server_vm_list"]:
            service_server_vm_list.append(VM.load_from_json(vm_json))

        for vm_json in json_data["ambari_agent_vm_list"]:
            ambari_agent_vm_list.append(VM.load_from_json(vm_json))

        cluster = Cluster()
        cluster.cluster_name = cluster_name
        cluster.state = json_data["state"]
        cluster.create_time = json_data["create_time"]
        cluster.ambari_server_vm = ambari_server_vm
        cluster.service_server_vm_list = service_server_vm_list
        cluster.ambari_agent_vm_list = ambari_agent_vm_list
        return cluster

    def _extract_vm_fqdn_ip(self, gce_info_file_name):
        """
        exatract domain name and IP address of VMs from the output file of GCE
        :param gce_info_file_name: output file of "GCE info" command
        :return: A list of tuple, each tuple has domain name and IP of a VM
        """
        lines = []
        with open(gce_info_file_name) as f:
            lines = f.readlines()

        vm_list = []
        # the first line in the output file is title
        for line in lines[1:]:
            tokens = line.split()
            fqdn_ip = (tokens[0], tokens[1])
            vm_list.append(fqdn_ip)
        return vm_list

    def request_vm(self, name, vm_num, gce_vm_type, gce_vm_os, gce_extra_cmd):
        """
        Request VMs from GCE
        :param name: the name prefix of all requesting VMs
        :param vm_num: the number of VM
        :param gce_vm_type: the type of VM
        :param gce_vm_os: the OS of VM
        :param gce_extra_cmd: extra command for requesting the VMs
        :return: A list of tuple, each tuple has domain name and IP of a VM
        """
        gce_key = Config.ATTRIBUTES["gce_controller_key_file"]
        gce_login = "{0}@{1}".format(Config.ATTRIBUTES["gce_controller_user"], Config.ATTRIBUTES["gce_controller_ip"])
        gce_up_cmd = "gce up {0} {1} {2} {3} {4}".format(name, vm_num, gce_vm_type, gce_vm_os, gce_extra_cmd)
        subprocess.call(["ssh", "-o", "StrictHostKeyChecking=no", "-i", gce_key, gce_login, gce_up_cmd])

        Log.write("cluster launched, wait for cluster info ... ...")

        fqdn_ip_pairs = []
        # wait for long enough. the more VM, more time it takes.
        for retry in range(max(6, vm_num)):
            time.sleep(10)

            # request cluster info
            with open(Config.ATTRIBUTES["gce_info_output"], "w") as gce_info_output_file:
                gce_info_cmd = "gce info {0}".format(name)
                subprocess.call(["ssh", "-o", "StrictHostKeyChecking=no", "-i", gce_key, gce_login, gce_info_cmd],
                                stdout=gce_info_output_file)

            fqdn_ip_pairs = self._extract_vm_fqdn_ip(Config.ATTRIBUTES["gce_info_output"])

            if len(fqdn_ip_pairs) == vm_num:
                Log.write("Get info for all ", str(len(fqdn_ip_pairs)), " VMs successfully")
                break
            Log.write("Only get info for ", str(len(fqdn_ip_pairs)), " VMs, retry ... ...")
        return fqdn_ip_pairs

    def request_ambari_server_vm(self, name):
        """
        request a VM for holding Ambari-server
        :param name: the name prefix of all requesting VMs
        :return: A list of tuple, each tuple has domain name and IP of a VM
        """
        # only 1 ambari server
        vm_num = 1
        gce_vm_type = Config.ATTRIBUTES["ambari_server_vm_type"]
        gce_vm_os = Config.ATTRIBUTES["ambari_server_vm_os"]

        gce_extra_cmd = ""
        if "ambari_server_vm_extra" in Config.ATTRIBUTES:
            gce_extra_cmd = Config.ATTRIBUTES["ambari_server_vm_extra"]

        fqdn_ip_pairs = self.request_vm(name, vm_num, gce_vm_type, gce_vm_os, gce_extra_cmd)
        return fqdn_ip_pairs

    def reqeust_service_server_vm(self, vm_num, name):
        """
        Request VMs to directly hold Ambari-agent (not inside Docker)
        :param vm_num: the number of VM to request
        :param name: the name prefix of all requesting VMs
        :return: A list of tuple, each tuple has domain name and IP of a VM
        """
        gce_vm_type = Config.ATTRIBUTES["service_server_vm_type"]
        gce_vm_os = Config.ATTRIBUTES["service_server_vm_os"]

        gce_extra_cmd = ""
        if "service_server_vm_extra" in Config.ATTRIBUTES:
            gce_extra_cmd = Config.ATTRIBUTES["service_server_vm_extra"]

        fqdn_ip_pairs = self.request_vm(name, vm_num, gce_vm_type, gce_vm_os, gce_extra_cmd)
        return fqdn_ip_pairs

    def request_agent_vm(self, vm_num, name):
        """
        Request VMs to hold Docker containers, each with Ambari-agent inside
        :param vm_num: the number of VM to request
        :param name: the name prefix of all requesting VMs
        :return: A list of tuple, each tuple has domain name and IP of a VM
        """
        gce_vm_type = Config.ATTRIBUTES["ambari_agent_vm_type"]
        gce_vm_os = Config.ATTRIBUTES["ambari_agent_vm_os"]
        gce_extra_disk = ""
        if "ambari_agent_vm_extra_disk" in Config.ATTRIBUTES:
            gce_extra_disk = Config.ATTRIBUTES["ambari_agent_vm_extra_disk"]

        fqdn_ip_pairs = self.request_vm(name, vm_num, gce_vm_type, gce_vm_os, gce_extra_disk)
        return fqdn_ip_pairs

    def request_gce_cluster(self, ambari_agent_vm_num, docker_num,
                            service_server_num, with_ambari_server, cluster_name):
        """
        Request a cluster from GCE
        :param ambari_agent_vm_num: number of VMs to hold Docker containers
        :param docker_num: number of Docker containers inside each VM
        :param service_server_num: number of VMs which has Ambari-agent directly installed (not in Docker)
        :param with_ambari_server: True or False, whether to request a VM to hold Ambari-server
        :param cluster_name: the name of the cluster
        :return: None
        """
        ambari_server_fqdn_ip_pairs = []
        if with_ambari_server is True:
            ambari_server_fqdn_ip_pairs = self.request_ambari_server_vm(VM.get_ambari_server_vm_name(cluster_name))
        service_server_fqdn_ip_pairs = self.reqeust_service_server_vm(service_server_num,
                                                                      VM.get_service_server_vm_name(cluster_name))
        ambari_agent_fqdn_ip_pairs = self.request_agent_vm(ambari_agent_vm_num,
                                                           VM.get_ambari_agent_vm_name(cluster_name))

        # prepare all attributes of the cluster, write to a file
        self.generate_cluster_info(cluster_name, ambari_server_fqdn_ip_pairs, service_server_fqdn_ip_pairs,
                                   ambari_agent_fqdn_ip_pairs, docker_num)

    def generate_cluster_info(self, cluster_name, ambari_server_fqdn_ip_pairs, service_server_fqdn_ip_pairs,
                              ambari_agent_fqdn_ip_pairs, docker_num):
        """
        generate VM and docker info for this cluster
        set up parameter of the class instance as this info
        :param cluster_name: the name of the cluster
        :param ambari_server_fqdn_ip_pairs: the domain name and IP pairs for Ambari-server
        :param service_server_fqdn_ip_pairs: the domain name and IP pairs for VMs with Ambari-agent installed
        :param ambari_agent_fqdn_ip_pairs: the domain name and IP pairs for VM with Docker containers
        :param docker_num: the number of Dockers inside each VMs
        :return: None
        """
        weave_ip_base = Config.ATTRIBUTES["weave_ip_base"]
        weave_ip_mask = Config.ATTRIBUTES["weave_ip_mask"]
        current_ip = weave_ip_base

        for vm_domain_name, vm_ip in ambari_server_fqdn_ip_pairs:
            current_ip = self._increase_ip(current_ip, 1)
            weave_dns_ip = current_ip
            vm = VM(vm_ip, vm_domain_name, weave_dns_ip, weave_ip_mask)
            current_ip = self._increase_ip(current_ip, 1)
            vm.weave_internal_ip = current_ip
            self.ambari_server_vm.append(vm)

        for vm_domain_name, vm_ip in service_server_fqdn_ip_pairs:
            current_ip = self._increase_ip(current_ip, 1)
            weave_dns_ip = current_ip
            vm = VM(vm_ip, vm_domain_name, weave_dns_ip, weave_ip_mask)
            current_ip = self._increase_ip(current_ip, 1)
            vm.weave_internal_ip = current_ip
            self.service_server_vm_list.append(vm)

        vm_index = 0
        for vm_domain_name, vm_ip in ambari_agent_fqdn_ip_pairs:
            current_ip = self._increase_ip(current_ip, 1)
            weave_dns_ip = current_ip
            vm = VM(vm_ip, vm_domain_name, weave_dns_ip, weave_ip_mask)

            for docker_index in range(0, docker_num):
                current_ip = self._increase_ip(current_ip, 1)
                docker_ip_str = current_ip

                total_docker_index = vm_index * docker_num + docker_index
                docker_domain_name = Docker.get_weave_domain_name(cluster_name, total_docker_index)

                docker = Docker(docker_ip_str, str(weave_ip_mask), docker_domain_name)
                vm.add_docker(docker)

            vm_index += 1
            self.ambari_agent_vm_list.append(vm)

        self.cluster_name = cluster_name
        self.create_time = str(datetime.datetime.now())
        self.state = Cluster.STATE_FREE

        # update config file.
        # This step makes the user avoid reconfiguring the IP for next cluster creation
        Config.update("weave", "weave_ip_base", current_ip)

    def _increase_ip(self, base_ip_str, increase):
        """
        increase the IP address.
        example: 192.168.1.1, increased by 1: 192.168.1.2
        example: 192.168.1.254, increased by 2: 192.168.2.1
        :param base_ip_str: the IP to be increased
        :param increase: the amount of increase
        :return: the new IP address, in String
        """
        base_ip = base_ip_str.split(".")
        new_ip = [int(base_ip[0]), int(base_ip[1]), int(base_ip[2]), int(base_ip[3])]
        new_ip[3] = new_ip[3] + increase
        for index in reversed(range(0, 4)):
            if new_ip[index] > 255:
                new_ip[index - 1] += (new_ip[index] / 256)
                new_ip[index] %= 256
        return "{0}.{1}.{2}.{3}".format(new_ip[0], new_ip[1], new_ip[2], new_ip[3])

    def _scp_upload(self, vm_external_ip):
        """
        upload all the code in a VM
        :param vm_external_ip: the external IP of the VM
        :return: None
        """
        # upload necessary file to VM
        vm_directory = "{0}@{1}:{2}".format(Config.ATTRIBUTES["vm_user"], vm_external_ip,
                                            Config.ATTRIBUTES["vm_code_directory"])
        vm_key = Config.ATTRIBUTES["vm_key_file"]

        upload_return_code = 0
        with open(os.devnull, 'w') as shutup:
            upload_return_code = subprocess.call(["scp", "-o", "StrictHostKeyChecking=no", "-i",
                                                  vm_key, "-r", ".", vm_directory],
                                                 stdout=shutup, stderr=shutup)
        if upload_return_code == 0:
            Log.write("VM ", vm_external_ip, " file upload succeed")
        else:
            Log.write("VM ", vm_external_ip, " file upload fail")

    def _set_executable_permission(self, vm_external_ip):
        """
        Set all shell file to be executable
        :param vm_external_ip: the external IP of the VM
        :return: None
        """
        vm_ssh_login = "{0}@{1}".format(Config.ATTRIBUTES["vm_user"], vm_external_ip)
        vm_ssh_cd_cmd = "cd {0}".format(Config.ATTRIBUTES["vm_code_directory"])
        vm_ssh_chmod_cmd = "chmod a+x **/*.sh"
        vm_ssh_cmd = "{0};{1}".format(vm_ssh_cd_cmd, vm_ssh_chmod_cmd)
        vm_key = Config.ATTRIBUTES["vm_key_file"]

        with open(os.devnull, 'w') as shutup:
            subprocess.Popen(["ssh", "-o", "StrictHostKeyChecking=no", "-t", "-i", vm_key,
                              vm_ssh_login, vm_ssh_cmd],
                             stdout=shutup, stderr=shutup)

    def run_cluster(self, server_weave_ip, server_external_ip):
        """
        Run all Ambari-agents and Ambari-server in the cluster in parallel
        Wait until all processes finish
        :param server_weave_ip: the Weave IP of Ambari-server
        :param server_external_ip: the external IP of Ambari-server
        :return: None
        """
        process_list = {}
        process_list.update(self.run_ambari_server_asyn())
        process_list.update(self.run_service_server_asyn(server_weave_ip, server_external_ip))
        process_list.update(self.run_docker_on_cluster_asyn(server_weave_ip, server_external_ip))

        terminate_state_list = {}
        for hostname in process_list:
            terminate_state_list[hostname] = False

        Log.write("Wait for all VMs to finish configuration ... ...")

        # Wait for all configuration subprocesses
        while True:
            all_finished = True
            for hostname in process_list:
                output_file, output_file_path, process = process_list[hostname]
                if terminate_state_list[hostname] is False:
                    all_finished = False
                    returncode = process.poll()
                    if returncode is None:
                        continue
                    else:
                        Log.write("VM ", hostname, " configuration completed, return code: ", str(returncode),
                                  ", output file path: ", output_file_path)
                        terminate_state_list[hostname] = True
                        output_file.close()
                else:
                    pass
            if all_finished:
                break
            time.sleep(5)

        Log.write("All VM configuration completed.")

    def run_ambari_server_asyn(self):
        """
        Run configuration for Ambari-server in this cluster
        Set up Ambari-server and Weave network
        The method is NON-BLOCK
        :return: a map of tuple, the key of the map is the host name of the VM,
                the tuple has 3 elements: the file handler of the output of the VM,
                the file path of the output of the VM,
                and the process object of configuration for the VM
        """
        process_list = {}

        for vm in self.ambari_server_vm:
            vm_external_ip = vm.external_ip
            self._scp_upload(vm_external_ip)
            self._set_executable_permission(vm_external_ip)

            vm_output_file_path = vm.get_ssh_output_file_path()
            vm_output_file = open(vm_output_file_path, "w")

            # ssh install server
            vm_ssh_login = "{0}@{1}".format(Config.ATTRIBUTES["vm_user"], vm_external_ip)
            vm_ssh_cd_cmd = "cd {0}".format(Config.ATTRIBUTES["vm_code_directory"])
            vm_ssh_python_cmd = "python launcher_ambari_server.py {0}".format(self.cluster_name)
            vm_ssh_cmd = "{0};{1}".format(vm_ssh_cd_cmd, vm_ssh_python_cmd)
            vm_key = Config.ATTRIBUTES["vm_key_file"]
            Log.write(vm_ssh_python_cmd)

            process = subprocess.Popen(["ssh", "-o", "StrictHostKeyChecking=no", "-t", "-i", vm_key,
                                       vm_ssh_login, vm_ssh_cmd],
                                       stdout=vm_output_file, stderr=vm_output_file)
            process_list[vm.hostname] = (vm_output_file, vm_output_file_path, process)
            Log.write("Configuring VM ", vm.hostname, " ... ...")
        return process_list

    def run_service_server_asyn(self, server_weave_ip, server_external_ip):
        """
        Run configuration, set up Ambari-agent in this VM, and the Weave network
        :param server_weave_ip: the Weave IP of the Ambari-server
        :param server_external_ip: the external IP of the Ambari-server
        The method is NON-BLOCK
        :return: a map of tuple, the key of the map is the host name of the VM,
                the tuple has 3 elements: the file handler of the output of the VM,
                the file path of the output of the VM,
                and the process object of configuration for the VM
        """
        process_list = {}

        for vm in self.service_server_vm_list:
            vm_external_ip = vm.external_ip
            self._scp_upload(vm_external_ip)
            self._set_executable_permission(vm_external_ip)

            vm_output_file_path = vm.get_ssh_output_file_path()
            vm_output_file = open(vm_output_file_path, "w")

            # ssh install server
            vm_ssh_login = "{0}@{1}".format(Config.ATTRIBUTES["vm_user"], vm_external_ip)
            vm_ssh_cd_cmd = "cd {0}".format(Config.ATTRIBUTES["vm_code_directory"])
            vm_ssh_python_cmd = "python launcher_service_server.py {0} {1} {2} {3}".format(
                vm_external_ip, server_weave_ip, server_external_ip, self.cluster_name)
            vm_ssh_cmd = "{0};{1}".format(vm_ssh_cd_cmd, vm_ssh_python_cmd)
            vm_key = Config.ATTRIBUTES["vm_key_file"]
            Log.write(vm_ssh_python_cmd)

            process = subprocess.Popen(["ssh", "-o", "StrictHostKeyChecking=no", "-t", "-i", vm_key,
                                        vm_ssh_login, vm_ssh_cmd],
                                       stdout=vm_output_file, stderr=vm_output_file)

            process_list[vm.hostname] = (vm_output_file, vm_output_file_path, process)
            Log.write("Configuring VM ", vm.hostname, " ... ...")
        return process_list

    def run_docker_on_cluster_asyn(self, server_weave_ip, server_external_ip):
        """
        Run configuration, set up Docker and Weave network
        run all containers, each with Ambari-agent inside
        :param server_weave_ip: the Weave IP of the Ambari-server
        :param server_external_ip: the external IP of the Ambari-server
        The method is NON-BLOCK
        :return: a map of tuple, the key of the map is the host name of the VM,
                the tuple has 3 elements: the file handler of the output of the VM,
                the file path of the output of the VM,
                and the process object of configuration for the VM
        """
        process_list = {}

        for vm in self.ambari_agent_vm_list:
            vm_external_ip = vm.external_ip
            self._scp_upload(vm_external_ip)
            self._set_executable_permission(vm_external_ip)

            vm_output_file_path = vm.get_ssh_output_file_path()
            vm_output_file = open(vm_output_file_path, "w")

            vm_ssh_login = "{0}@{1}".format(Config.ATTRIBUTES["vm_user"], vm_external_ip)
            vm_ssh_cd_cmd = "cd {0}".format(Config.ATTRIBUTES["vm_code_directory"])
            vm_ssh_python_cmd = "python launcher_docker.py {0} {1} {2} {3}".format(
                vm_external_ip, server_weave_ip, server_external_ip, self.cluster_name)
            vm_ssh_cmd = "{0};{1}".format(vm_ssh_cd_cmd, vm_ssh_python_cmd)
            vm_key = Config.ATTRIBUTES["vm_key_file"]
            Log.write(vm_ssh_python_cmd)

            process = subprocess.Popen(["ssh", "-o", "StrictHostKeyChecking=no", "-t", "-i", vm_key,
                                        vm_ssh_login, vm_ssh_cmd],
                                       stdout=vm_output_file, stderr=vm_output_file)

            process_list[vm.hostname] = (vm_output_file, vm_output_file_path, process)
            Log.write("Configuring VM ", vm.hostname, " ... ...")

        return process_list