palo_env.py

#!/bin/env python
# -*- coding: utf-8 -*-
# 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.

"""
This module describe environment info about Palo.

Date:    2015/10/07 17:23:06
"""

import os
import pexpect
import threading
import time
import env_config
from lib import palo_client


class PaloEnv(object):
    """
    Palo部署及运行环境信息
    """
    def __init__(self):
        self.__master = env_config.master
        self.__follower_list = env_config.follower_list
        self.__observer_list = env_config.observer_list
        self.__be_list = env_config.be_list
        self.__fe_path = env_config.fe_path
        self.__be_path = env_config.be_path
        self.__host_username = env_config.host_username
        self.__host_password = env_config.host_password
        self.__fe_query_port = env_config.fe_query_port
        self.__be_heartbeat_port = self.__fe_query_port + 20
        self.__be_data_path_list = env_config.be_data_path_list
        self.__master_lock = threading.Lock()
        self.__follower_lock = threading.Lock()
        self.__observer_lock = threading.Lock()
        self.__be_lock = threading.Lock()
        self.__dynamic_add_fe_list = env_config.dynamic_add_fe_list
        self.__dynamic_add_be_list = env_config.dynamic_add_be_list
        self.__dynamic_add_fe_lock = threading.Lock()
        self.__dynamic_add_be_lock = threading.Lock()

    def init(self):
        """init
        """
        if not self.reset_master():
            return False
        return True

    def get_master(self):
        """返回mster
        """
        with self.__master_lock:
            return self.__master

    def set_master(self, host_name):
        """设置mster
        """
        assert host_name
        with self.__master_lock:
            self.__master = host_name

    def remove_master(self):
        """remove mster
        """
        with self.__master_lock:
            self.__master = None

    def __get_client(self, host_name, port):
        """
        """
        assert host_name
        client = palo_client.PaloClient(host_name, port)
        if client.init():
            time.sleep(10)
            return client
        else:
            return None

    def reset_master(self):
        """reset master
        """
        # 如果没有get到master, 重试，等待重新选了一个新的master
        ix = 0
        while ix < 120:
            if self.reset_master_once():
                return True
            time.sleep(1)
            ix += 1
        else:
            return False

    def reset_master_once(self):
        """reset master once
        """
        cur_master = self.get_master()
        electable_tuple = (self.get_master(),) + self.get_follower_tuple() if \
                self.get_master() else self.get_follower_tuple()
        for host_name in electable_tuple:
            client = self.__get_client(host_name, self.get_fe_query_port())
            if not client:
                return False
            if client.is_master():
                if host_name != cur_master:
                    if cur_master:
                        self.add_follower(cur_master)
                    self.remove_follower(host_name)
                    self.set_master(host_name)
                return True
        return False

    def get_follower_tuple(self):
        """返回follower
        """
        with self.__follower_lock:
            return tuple(self.__follower_list)

    def add_follower(self, host_name):
        """增加follower
        """
        assert host_name
        if host_name not in self.__follower_list:
            with self.__follower_lock:
                self.__follower_list.append(host_name)

    def remove_follower(self, host_name):
        """删除follower
        """
        assert host_name
        if host_name in self.__follower_list:
            with self.__follower_lock:
                self.__follower_list.remove(host_name)

    def get_observer_tuple(self):
        """返回observer
        """
        with self.__follower_lock:
            return tuple(self.__observer_list)

    def add_observer(self, host_name):
        """增加observer
        """
        assert host_name
        if host_name not in self.__observer_list:
            with self.__observer_lock:
                self.__observer_list.append(host_name)

    def remove_observer(self, host_name):
        """删除observer
        """
        assert host_name
        if host_name in self.__observer_list:
            with self.__observer_lock:
                self.__observer_list.remove(host_name)

    def get_fe_tuple(self):
        """返回FE
        """
        return (self.get_master(),) + self.get_follower_tuple() + self.get_observer_tuple()

    def get_dynamic_add_fe_tuple(self):
        """返回dynamic added fe
        """
        with self.__dynamic_add_fe_lock:
            return tuple(self.__dynamic_add_fe_list)

    def add_dynamic_add_fe(self, host_name):
        """增加dynamic added fe
        """
        assert host_name
        if host_name not in self.__dynamic_add_fe_list:
            with self.__dynamic_add_fe_lock:
                self.__dynamic_add_fe_list.append(host_name)

    def remove_dynamic_add_fe(self, host_name):
        """删除dynamic added fe
        """
        assert host_name
        if host_name in self.__dynamic_add_fe_list:
            with self.__dynamic_add_fe_lock:
                self.__dynamic_add_fe_list.remove(host_name)

    def get_be_tuple(self):
        """返回BE
        """
        with self.__be_lock:
            return tuple(self.__be_list)

    def add_be(self, host_name):
        """增加BE
        """
        assert host_name
        if host_name not in self.__be_list:
            with self.__be_lock:
                self.__be_list.append(host_name)

    def remove_be(self, host_name):
        """删除BE
        """
        assert host_name
        if host_name in self.__be_list:
            with self.__be_lock:
                self.__be_list.remove(host_name)

    def get_dynamic_add_be_tuple(self):
        """返回dynamic added be
        """
        with self.__dynamic_add_be_lock:
            return tuple(self.__dynamic_add_be_list)

    def add_dynamic_add_be(self, host_name):
        """增加dynamic added be
        """
        assert host_name
        if host_name not in self.__dynamic_add_be_list:
            with self.__dynamic_add_be_lock:
                self.__dynamic_add_be_list.append(host_name)

    def remove_dynamic_add_be(self, host_name):
        """删除dynamic added be
        """
        assert host_name
        if host_name in self.__dynamic_add_be_list:
            with self.__dynamic_add_be_lock:
                self.__dynamic_add_be_list.remove(host_name)

    def get_fe_query_port(self):
        """fe query port
        """
        return self.__fe_query_port

    def get_be_heartbeat_port(self):
        """be port
        """
        return self.__be_heartbeat_port

    def __exec_cmd(self, cmd, host_name, timeout=120):
        """exec cmd"""
        exe_cmd = 'ssh %s@%s "%s"' % (self.__host_username, host_name, cmd)
        output, status = pexpect.run(exe_cmd, timeout=timeout, withexitstatus=True, \
                events = {"continue connecting":"yes\n", "password:":"%s\n" % self.__host_password})
        if 'PALO_CLIENT_LOG_SQL' in os.environ.keys():
            palo_client.LOG.info(palo_client.L("execute CMD", exe_cmd=exe_cmd, \
                    status=status, output=output))
        return status, output

    def stop_master(self):
        """Stop master
        """
        # 结束当前master进程
        cmd = 'cd %s/fe;sh bin/stop_fe.sh' % (self.__fe_path)
        ix = 0
        while ix < 120:
            self.__exec_cmd(cmd, host_name=self.__master)
            if not self.is_fe_alive(self.get_master()):
                self.remove_master()
                return True
            time.sleep(1)
            ix += 1
        else:
            return False

    def start_electable(self, host_name):
        """Start electable 
        """
        return self.start_follower(host_name)

    def start_follower(self, host_name):
        """Start follower
        """
        if host_name in self.get_follower_tuple():
            return False
        cmd = 'cd %s/fe;sh bin/start_fe.sh' % self.__fe_path
        self.__exec_cmd(cmd, host_name)
        time.sleep(3)
        if self.__get_client(host_name, self.get_fe_query_port()):
            self.add_follower(host_name)
            time.sleep(10)
            return True
        else:
            return False

    def stop_follower(self, host_name):
        """stop follower
        """
        if host_name not in self.get_follower_tuple():
            return False
        cmd = 'cd %s/fe;sh bin/stop_fe.sh' % (self.__fe_path)
        ix = 0
        while ix < 120:
            self.__exec_cmd(cmd, host_name)
            if not self.is_fe_alive(host_name):
                self.remove_follower(host_name)
                return True
            time.sleep(1)
            ix += 1
        else:
            return False

    def start_observer(self, host_name):
        """Start master
        """
        if host_name in self.get_observer_tuple():
            return False
        cmd = 'cd %s/fe;sh bin/start_fe.sh' % (self.__fe_path)
        self.__exec_cmd(cmd, host_name)
        time.sleep(3)
        if self.__get_client(host_name, self.get_fe_query_port()):
            self.add_observer(host_name)
            time.sleep(10)
            return True
        else:
            return False

    def stop_observer(self, host_name):
        """stop observer
        """
        if host_name not in self.get_observer_tuple():
            return False
        cmd = 'cd %s/fe;sh bin/stop_fe.sh' % (self.__fe_path)
        ix = 0
        while ix < 120:
            self.__exec_cmd(cmd, host_name)
            if not self.is_fe_alive(host_name):
                self.remove_observer(host_name)
                return True
            time.sleep(1)
            ix += 1
        else:
            return False

    def is_fe_alive(self, host_name):
        """check is fe alive
        """
        cmd = "cd %s/fe;if [ -f bin/fe.pid ]; then if kill -0 `cat bin/fe.pid` " \
                ">/dev/null 2>&1; then echo 'running'; fi; fi" % (self.__fe_path)
        if self.__exec_cmd(cmd, host_name)[1].rstrip('\r\n') == 'running':
            return True
        else:
            return False

    def is_be_alive(self, host_name):
        """check is be alive
        """
        cmd = "cd %s/be;if [ -f bin/be.pid ]; then if kill -0 `cat bin/be.pid` " \
                ">/dev/null 2>&1; then echo 'running'; fi; fi" % (self.__be_path)
        if self.__exec_cmd(cmd, host_name)[1].rstrip('\r\n') == 'running':
            return True
        else:
            return False

    def restart_master(self):
        """restart master
        """
#       follower_tuple = self.get_follower_tuple()
#       for follower in follower_tuple:
#           if not self.stop_follower(follower):
#               return False
        old_master = self.get_master()
        if not self.stop_master():
            return False
        if not self.start_electable(old_master):
            return False
#       for follower in follow_tuple:
#           if not self.start_follower(follower):
#               return False
#       return True
        if not self.reset_master():
            return False
        return True

    def switch_master(self):
        """主备切换
        """
        cur_master = self.get_master()
        if not self.stop_master():
            return False
        if not self.reset_master():
            return False
        if not self.start_electable(cur_master):
            return False
        return True

    def restart_follower(self, host_name=None):
        """restart follower
        """
        if not self.stop_follower(host_name):
            return False
        if not self.start_follower(host_name):
            return False
        return True

    def restart_observer(self, host_name=None):
        """restart observer
        """
        if not self.stop_observer(host_name):
            return False
        if not self.start_observer(host_name):
            return False
        return True

    def stop_be(self, host_name):
        """Stop BE
        """
        cmd_a = 'cd %s/be;sh bin/stop_be.sh' % (self.__be_path)
        cmd_b = 'pkill -f %s/be/lib/palo_be' % (self.__be_path)
        ix = 0
        while ix < 120:
            self.__exec_cmd(cmd_a, host_name)
            self.__exec_cmd(cmd_b, host_name)
            if not self.is_be_alive(host_name):
                self.remove_be(host_name)
                return True
            time.sleep(1)
            ix += 1
        return False

    def start_be(self, host_name):
        """Start BE
        """
        if host_name in self.get_be_tuple():
            return False
        cmd= 'cd %s/be;sh bin/start_be.sh' % (self.__be_path)
        ix = 0
        while ix < 120:
            self.__exec_cmd(cmd, host_name)
            if self.is_be_alive(host_name):
                self.add_be(host_name)
                time.sleep(10)
                return True
            time.sleep(1)
            ix += 1
        return False

    def restart_be(self, host_name):
        """Restart be
        """
        if not self.stop_be(host_name):
            return False
        if not self.start_be(host_name):
            return False
        return True

    def stop_all_fe(self):
        """stop all fe
        """
        for host_name in self.get_follower_tuple():
            if not self.stop_follower(host_name):
                return False
            
        for host_name in self.get_observer_tuple():
            if not self.stop_observer(host_name):
                return False

        if not self.stop_master():
            return False

        return True

    def stop_all_be(self):
        """stop all be
        """
        for host_name in self.get_be_tuple():
            if not self.stop_be(host_name):
                return False
        return True

    def update_fe_config(self, option_value_dict):
        """Update FE config
        """
        master = self.get_master()
        follower_tuple = self.get_follower_tuple()
        observer_tuple = self.get_observer_tuple()
        fe_tuple = self.get_fe_tuple()

        if not self.stop_all_fe():
            return False

        configfile_path = '%s/fe/conf/fe.conf' % (self.__fe_path)
        for host_name in fe_tuple:
            for option, value in option_value_dict.iteritems():
                if not self.modify_config(host_name, configfile_path, option, value):
                    return False

        self.start_electable(master)

        follower_threads = []
        observer_threads = []
        for host_name in follower_tuple:
            t = threading.Thread(target=self.start_follower, args=(host_name,))
            t.start()
            follower_threads.append(t)

        for host_name in observer_tuple:
            t = threading.Thread(target=self.start_observer, args=(host_name,))
            t.start()
            observer_threads.append(t)

        for t in follower_threads:
            t.join()
        for t in observer_threads:
            t.join()

        for host_name in follower_tuple + observer_tuple:
            if not self.is_fe_alive(host_name):
                return False

        if not self.reset_master():
            return False

        return True

    def update_be_config(self, option_value_dict):
        """Update BE config
        """
        be_tuple = self.get_be_tuple()
        if not self.stop_all_be():
            return False

        configfile_path = '%s/be/conf/be.conf' % (self.__be_path)
        for host_name in be_tuple:
            for option, value in option_value_dict.iteritems():
                if not self.modify_config(host_name, configfile_path, option, value):
                    return False

        be_threads = []
        for host_name in self.be_tuple:
            t = threading.Thread(target=self.start_be, args=(host_name,))
            t.start()
            be_threads.append(t)

        for t in be_threads:
            t.join()

        for host_name in be_tuple:
            if not self.is_be_alive(host_name):
                return False

        return True

    def modify_config(self, host_name, filepath, option, value):
        """修改配置
        """
        cmd = "grep -q '^{option}' {filepath} && "\
                "sed -i 's/^{option}.*/{option} = {value}/g' {filepath} || "\
                "echo '{option} = {value}' >> {filepath}".format( \
                option=option, value=value, filepath=filepath)
        status, output = self.__exec_cmd(cmd, host_name)
        if status != 0 or output:
            return False
        cmd = "grep -q '^{} = {}' {}".format(option, value, filepath)
        status, output = self.__exec_cmd(cmd, host_name)
        if status != 0:
            return False
        return True

    def remove_config(self, host_name, filepath, option):
        """删除配置
        """
        cmd = 'sed -i "s/^%s.*/ /" %s' % (option, filepath)
        self.__exec_cmd(cmd, host_name)
        cmd = "grep -q '^{} {}'".format(option, filepath)
        status, output = self.__exec_cmd(cmd, host_name)
        if status == 0:
            return False
        return True

    def clean_fe(self, host_name):
        """clean fe
        """
        cmd= 'cd %s/fe;rm -rf log palo-meta/image palo-meta/bdb temp' % (self.__fe_path)
        return self.__exec_cmd(cmd, host_name)

    def clean_be(self, host_name):
        """clean be
        """
        data_path = '/* '.join(self.__be_data_path_list) + '/*'
        cmd= 'cd %s/be;rm -rf log unused bin/be.pid %s' % (self.__be_path, data_path)
        return self.__exec_cmd(cmd, host_name)

    def clean_start(self):
        """clean start
        """

        self.stop_all_fe()
        self.stop_all_be()

        master = env_config.master
        follower_tuple = tuple(env_config.follower_list)
        observer_tuple = tuple(env_config.observer_list)
        be_tuple = tuple(env_config.be_list)

        for host_name in (master,) + follower_tuple + observer_tuple:
            self.clean_fe(host_name)
        for host_name in be_tuple:
            self.clean_be(host_name)

        self.start_electable(master,)

        follower_threads = []
        observer_threads = []
        be_threads = []

        for host_name in follower_tuple:
            t = threading.Thread(target=self.start_follower, args=(host_name,))
            t.start()
            follower_threads.append(t)
        for host_name in observer_tuple:
            t = threading.Thread(target=self.start_observer, args=(host_name,))
            t.start()
            observer_threads.append(t)
        for t in follower_threads:
            t.join()
        for t in observer_threads:
            t.join()

        for host_name in be_tuple:
            t = threading.Thread(target=self.start_be, args=(host_name,))
            t.start()
            be_threads.append(t)
        for t in be_threads:
            t.join()
        
        for host_name in follower_tuple:
            if not self.is_fe_alive(host_name):
                return False
        for host_name in observer_tuple:
            if not self.is_fe_alive(host_name):
                return False
        for host_name in be_tuple:
            if not self.is_be_alive(host_name):
                return False

        if not self.reset_master():
            return False
        client = self.__get_client(self.get_master(), self.get_fe_query_port())
        if not client:
            return False
        be_heartbeat_port = self.get_be_heartbeat_port()
        backend_list = ['%s:%s' % (be, be_heartbeat_port) for be in self.get_be_tuple()]
        return client.add_backend_list(backend_list)

    def check_be(self, host_name, tablet_id):
        """check whether base expansion done
        """
        assert host_name
        data_path = '/data '.join(self.__be_data_path_list) + '/data'
        cmd = 'find %s -name %s_0_*.dat' % (data_path, tablet_id)
        status, output = self.__exec_cmd(cmd, host_name)
        assert not status
        file_name = output.split('/')[-1]
        versions = file_name.split('_')
        if versions[2] > 1:
            return True
        else:
            return False

    def check_ce(self, host_name, tablet_id):
        """check whether cumulative done
        """
        assert host_name
        data_path = '/data '.join(self.__be_data_path_list) + '/data'
        cmd = 'find %s -name %s_*.dat' % (data_path, tablet_id)
        status, output = self.__exec_cmd(cmd, host_name)
        assert not status
        file_path_list = output.split('\r\n')
        for file_path in file_path_list:
            file_name = file_path.split('/')[-1]
            versions = file_name.split('_')
            if versions[1] > 0 and versions[1] != versions[2]:
                return True
        return False