| #!/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 operates Palo node. | |
| """ | |
| import json | |
| import os | |
| import pexpect | |
| import sys | |
| import threading | |
| import time | |
| import random | |
| import socket | |
| sys.path.append('../deploy') | |
| import env_config | |
| import palo_logger | |
| import palo_client | |
| import util | |
| import palo_job | |
| # 日志 异常 对象 | |
| LOG = palo_logger.Logger.getLogger() | |
| L = palo_logger.StructedLogMessage | |
| class Node(object): | |
| """ | |
| palo node operation | |
| """ | |
| 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.__java_home = env_config.JAVA_HOME | |
| self.__client = None | |
| self.__query_port = env_config.fe_query_port | |
| self.__be_ip_list = [socket.gethostbyname(be) for be in self.__be_list] | |
| self.init_client() | |
| def init_client(self, host=None, user=None, password=None): | |
| """get palo client, for get fe master""" | |
| user = 'root' if user is None else user | |
| password = '' if password is None else password | |
| host = random.choice(self.__follower_list + self.__observer_list + [self.__master]) if host is None else host | |
| try: | |
| self.__client = palo_client.get_client(host, self.__query_port, | |
| user=user, password=password) | |
| except Exception as e: | |
| LOG.info(L("INIT CLIENT FAILED", host=host, port=self.__query_port)) | |
| def stop_fe(self, host_name): | |
| """stop fe node""" | |
| LOG.info(L('STOP FE.')) | |
| cmd = 'cd %s/fe;sh bin/stop_fe.sh' % self.__fe_path | |
| status, output = self.__exec_cmd(cmd, host_name=host_name) | |
| if status == 0: | |
| LOG.info(L('STOP FE SUCC.', ret=True)) | |
| return True | |
| else: | |
| LOG.info(L('STOP FE FAILED.', ret=False)) | |
| return False | |
| def start_fe(self, host_name): | |
| """start fe node""" | |
| LOG.info(L('START FE.')) | |
| cmd = 'export JAVA_HOME=%s;cd %s/fe;sh bin/start_fe.sh --daemon' % (self.__java_home, self.__fe_path) | |
| status, output = self.__exec_cmd(cmd, host_name) | |
| if status == 0: | |
| LOG.info(L('START FE SUCC.', ret=True)) | |
| return True | |
| else: | |
| LOG.info(L('START FE FAILED.', ret=False)) | |
| return False | |
| def is_fe_alive(self, host_name): | |
| """check is fe alive | |
| """ | |
| LOG.info(L('CHECK FE STATUE.')) | |
| cmd = "cd %s/fe;kill -0 `cat bin/fe.pid` >/dev/null 2>&1" % self.__fe_path | |
| status, output = self.__exec_cmd(cmd, host_name) | |
| if status == 0: | |
| LOG.info(L('CHECK FE STATUE. FE is alive')) | |
| return True | |
| else: | |
| LOG.info(L('CHECK FE STATUE. FE is dead')) | |
| return False | |
| def is_be_alive(self, host_name): | |
| """check is be alive | |
| """ | |
| LOG.info(L('CHECK BE STATUS.')) | |
| cmd = "cd %s/be;kill -0 `cat bin/be.pid` >/dev/null 2>&1" % self.__be_path | |
| status, output = self.__exec_cmd(cmd, host_name) | |
| if status == 0: | |
| LOG.info(L('CHECK BE STATUS. BE is alive')) | |
| return True | |
| else: | |
| LOG.info(L('CHECK BE STATUS. BE is dead')) | |
| return False | |
| def __exec_cmd(self, cmd, host_name, timeout=120): | |
| 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}) | |
| LOG.info(L("execute CMD", exe_cmd=exe_cmd, status=status, output=output)) | |
| return status, output | |
| def stop_be(self, host_name): | |
| """Stop BE | |
| """ | |
| LOG.info(L('STOP BE.')) | |
| cmd_a = 'cd %s/be;sh bin/stop_be.sh' % self.__be_path | |
| status, output = self.__exec_cmd(cmd_a, host_name) | |
| if status == 0: | |
| LOG.info(L('STOP BE SUCC.')) | |
| return True | |
| else: | |
| LOG.info(L('STOP BE FAILED.')) | |
| return False | |
| def start_be(self, host_name): | |
| """Start BE | |
| """ | |
| LOG.info(L('START BE.')) | |
| cmd = 'export PATH=$PATH:/sbin; export JAVA_HOME=%s; cd %s/be;sh bin/start_be.sh --daemon' \ | |
| % (self.__java_home, self.__be_path) | |
| status, output = self.__exec_cmd(cmd, host_name) | |
| if status == 0: | |
| LOG.info(L('START BE SUCC.')) | |
| return True | |
| else: | |
| LOG.info(L('START BE FAILED.')) | |
| return False | |
| def get_observer(self): | |
| """get a observer""" | |
| return random.choice(self.__observer_list) | |
| def get_master(self): | |
| """get fe master""" | |
| retry_times = 10 | |
| if self.__client is not None and self.__client.connection: | |
| while retry_times > 0: | |
| retry_times -= 1 | |
| try: | |
| self.init_client() | |
| self.__master = self.__client.get_master_host() | |
| break | |
| except Exception as e: | |
| LOG.info(L("reconnect to fe")) | |
| time.sleep(3) | |
| else: | |
| raise Exception('can not connect to palo and get master') | |
| return self.__master | |
| def get_follower(self): | |
| """get fe follower not master""" | |
| retry_times = 10 | |
| if self.__client is not None and self.__client.connection: | |
| while retry_times > 0: | |
| retry_times -= 1 | |
| try: | |
| self.init_client() | |
| ret = self.__client.get_fe_list() | |
| break | |
| except Exception as e: | |
| LOG.info(L("reconnect to fe")) | |
| time.sleep(3) | |
| else: | |
| raise Exception('can not connect to palo and get master') | |
| self.__follower_list = util.get_attr_condition_list(ret, palo_job.FrontendInfo.Role, | |
| 'FOLLOWER', palo_job.FrontendInfo.Host) | |
| return random.choice(self.__follower_list) | |
| def get_fe_list(self): | |
| """get fe list""" | |
| return self.__observer_list + [self.__master] + self.__follower_list | |
| def get_be_list(self): | |
| """get be list""" | |
| return self.__be_list | |
| def get_be_ip_list(self): | |
| """get be ip list""" | |
| return self.__be_ip_list | |
| def restart_fe(self, host_name, wait_time=10): | |
| """restart fe""" | |
| self.stop_fe(host_name) | |
| time.sleep(wait_time) | |
| self.start_fe(host_name) | |
| def restart_be(self, host_name, wait_time=10): | |
| """retart be""" | |
| self.stop_be(host_name) | |
| time.sleep(wait_time) | |
| self.start_be(host_name) | |
| def is_be_core(self, host_name): | |
| """check if be has core file""" | |
| cmd = "cat %s/be/log/be.out" % self.__be_path | |
| status, output = self.__exec_cmd(cmd, host_name) | |
| print(output.replace("'s password", "'s be.out")) | |
| LOG.info(L('CHECK BE CORE.', be=host_name)) | |
| cmd = "ls -lh %s/be/core.*" % (self.__be_path) | |
| status, output = self.__exec_cmd(cmd, host_name) | |
| if status == 0: | |
| LOG.info(L('BE HAS COREFILE.', be=host_name)) | |
| return True | |
| else: | |
| LOG.info(L('BE HAS NO COREFILE.', be=host_name)) | |
| return False | |
| def get_image_version(self, host_name): | |
| """ | |
| get fe image version | |
| may have image.[log-id] & image.ckpt temp file | |
| """ | |
| cmd = 'ls %s/fe/palo-meta/image/image.[0-9]*' % self.__fe_path | |
| status, output = self.__exec_cmd(cmd, host_name) | |
| image_version_list = list() | |
| if status == 0: | |
| output_list = output.split('\r\n') | |
| for output in output_list: | |
| output = output.strip() | |
| images = output.split('image.') | |
| if len(images) != 2: | |
| continue | |
| version = images[-1] | |
| image_version_list.append(version) | |
| LOG.info(L('get fe image file version', fe=host_name, version=image_version_list)) | |
| return image_version_list | |
| else: | |
| LOG.warning(L('get fe image file version failed', fe=host_name, msg=output, status=status)) | |
| return None | |
| def check_cluster(self, start_if_dead=True, fe_check=True, be_check=True): | |
| """check fe status and be status, start if dead""" | |
| if fe_check: | |
| for fe in self.get_fe_list(): | |
| if not self.is_fe_alive(fe): | |
| self.start_fe(fe) | |
| if be_check: | |
| for be in self.get_be_list(): | |
| if not self.is_be_alive(be): | |
| self.start_be(be) | |
| def modify_be_conf(self, hostname, option, value): | |
| """modify be conf, and restart""" | |
| if isinstance(value, str): | |
| value_s = value.replace('/', '\/') | |
| else: | |
| value_s = value | |
| # 获取文件中是否有option，如果有则sed修改配置，如果没有则echo追加到文件末尾 | |
| cmd = "grep -q '^{option}' {filepath} && " \ | |
| "sed -i 's/^{option}.*/{option} = {value_a}/g' {filepath} || " \ | |
| "echo '\n{option} = {value}' >> {filepath}".format(option=option, value=value, | |
| filepath=self.__be_path + '/be/conf/be.conf', | |
| value_a=value_s) | |
| status, output = self.__exec_cmd(cmd, hostname) | |
| if status == 0: | |
| LOG.info(L("modify be conf succeed, will restart be", be=hostname, config="%s=%s" % (option, value))) | |
| self.restart_be(hostname) | |
| return True | |
| else: | |
| LOG.warning(L("modify be conf failed", be=hostname, config="%s=%s" % (option, value))) | |
| return False | |
| def modify_fe_conf(self, hostname, option, value): | |
| """modify fe conf, and restart""" | |
| if isinstance(value, str): | |
| value_s = value.replace('/', '\/') | |
| else: | |
| value_s = value | |
| # 获取文件中是否有option，如果有则sed修改配置，如果没有则echo追加到文件末尾 | |
| cmd = "grep -q '^{option}' {filepath} && " \ | |
| "sed -i 's/^{option}.*/{option} = {value_a}/g' {filepath} || " \ | |
| "echo '\n{option} = {value}' >> {filepath}".format(option=option, value=value, | |
| filepath=self.__fe_path + '/fe/conf/be.conf', | |
| value_a=value_s) | |
| status, output = self.__exec_cmd(cmd, hostname) | |
| if status == 0: | |
| LOG.info(L("modify be conf succeed, will restart be", be=hostname, config="%s=%s" % (option, value))) | |
| self.restart_be(hostname) | |
| return True | |
| else: | |
| LOG.warning(L("modify be conf failed", be=hostname, config="%s=%s" % (option, value), msg=output)) | |
| return False | |
| if __name__ == '__main__': | |
| env = Node() | |
| env.check_cluster(fe_check=False) | |