test_sys_update_restart.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.

"""
/***************************************************************************
  *
  * @file test_sys_partition_multi_col.py
  * @brief Test for partition by multi column
  *
  **************************************************************************/
"""
import time

from data import schema as DATA
from data import load_file as FILE
from lib import palo_client
from lib import util
from lib import palo_config
from lib import common
from lib import palo_job
from lib import palo_task
from lib import node_op

LOG = palo_client.LOG
L = palo_client.L
config = palo_config.config
broker_info = palo_config.broker_info


def setup_module():
    """
    setUp
    """
    global node_operator
    node_operator = node_op.Node()
    node_operator.check_cluster()


def teardown_module():
    """
    tearDown
    """
    node_operator.check_cluster()


def test_update_on_observer():
    """
    {
    "title": "test_update_on_observer",
    "describe": "在非fe master上执行update操作，预期执行成功",
    "tag": "function,p0,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client = common.create_workspace(database_name)
    partition_info = palo_client.PartitionInfo("k2",
                                               ["p1", "p2", "p3", "p4", "p5"],
                                               ["0", "100", "200", "300", "MAXVALUE"])
    distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
    keys_desc = "UNIQUE KEY(k1, k2, k3)"
    ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
                              partition_info=partition_info, distribution_info=distribution_info)
    assert ret, 'create table failed'
    ret = client.stream_load(table_name, FILE.partition_local_file)
    assert ret, 'stream load failed'
    ret1 = client.execute('select k1, k2, v6+1 from %s order by k1' % table_name)
    observer = node_operator.get_observer()
    observer_client = common.get_client(observer)
    observer_client.use(database_name)
    retry_times = 10
    while retry_times > 0:
        try:
            ret = observer_client.update(table_name, 'v6=v6+1', 'k1 > 0')
            break
        except Exception as e:
            LOG.warning(L('retry', msg=str(e)))
        retry_times -= 1
        time.sleep(3)
    assert ret, 'update failed'
    ret2 = observer_client.execute('select k1, k2, v6 from %s order by k1' % table_name)
    util.check(ret1, ret2, True)
    follower = node_operator.get_follower()
    follower_client = common.get_client(follower)
    follower_client.use(database_name)
    ret = follower_client.update(table_name, 'v6=1.0', 'k1 > 0')
    assert ret, 'update failed'
    sql1 = 'select 1.0 from %s' % table_name
    sql2 = 'select v6 from %s order by k1' % table_name
    common.check2(follower_client, sql1, sql2=sql2)
    client.clean(database_name)


def test_update_when_switch_fe_master():
    """
    {
    "title": "test_update_when_switch_fe_master",
    "describe": "在执行update操作的时候，fe进行切主",
    "tag": "function,p0,fuzz"
    }
    """
    # init
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client = common.create_workspace(database_name)
    partition_info = palo_client.PartitionInfo("k2",
                                               ["p1", "p2", "p3", "p4", "p5"],
                                               ["0", "100", "200", "300", "MAXVALUE"])
    distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
    keys_desc = "UNIQUE KEY(k1, k2, k3)"
    ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
                              partition_info=partition_info, distribution_info=distribution_info)
    assert ret, 'create table failed'
    ret = client.stream_load(table_name, FILE.partition_local_file)
    assert ret, 'stream load failed'
    # 持续执行update
    observer = node_operator.get_observer()
    client1 = common.get_client(observer)
    update_task = palo_task.SyncTask(client1.update, table_name, ['v6=v6+1'], 'k1 > 0', database_name)
    update_thread = palo_task.TaskThread(update_task)
    update_thread.start()
    time.sleep(5)
    # upadte的过程中切主
    master = node_operator.get_master()
    node_operator.restart_fe(master)
    time.sleep(30)
    # 停止update
    time.sleep(5)
    update_thread.stop()
    update_thread.join()
    # 校验
    assert node_operator.is_fe_alive(master), 'fe restart failed'
    assert client1.select_all(table_name, database_name)
    client1.connect()
    assert client1.get_master()
    client1.clean(database_name)
    retry_times = 10
    while retry_times > 0:
        try:
            client.connect()
            client.get_alive_backend_list()[0][palo_job.BackendProcInfo.Host]
            break
        except Exception as e:
            retry_times -= 1
            time.sleep(3)
            LOG.info(L("show proc failed"))


def test_update_load_fe_image():
    """
    {
    "title": "test_update_when_switch_fe_master",
    "describe": "在执行update操作的时候生成image，fe宕机，持续执行update操作到有新的image生成，fe切主加载image，数据正确",
    "tag": "function,p0,fuzz"
    }
    """
    # check cluster status
    # init
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client = common.create_workspace(database_name)
    partition_info = palo_client.PartitionInfo("k2",
                                               ["p1", "p2", "p3", "p4", "p5"],
                                               ["0", "100", "200", "300", "MAXVALUE"])
    distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
    keys_desc = "UNIQUE KEY(k1, k2, k3)"
    ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
                              partition_info=partition_info, distribution_info=distribution_info)
    assert ret, 'create table failed'
    ret = client.stream_load(table_name, FILE.partition_local_file)
    assert ret, 'stream load failed'
    # 获取fe image
    master = node_operator.get_master()
    version = node_operator.get_image_version(master)
    assert version, 'get fe version failed.version is %s' % version
    print(version)
    node_operator.restart_fe(master)
    time.sleep(30)
    new_master = node_operator.get_master()
    # 持续执行update
    observer = node_operator.get_observer()
    client1 = common.get_client(observer)
    update_task = palo_task.SyncTask(client1.update, table_name, ['v6=v6+1'], 'k1 > 0', database_name)
    update_thread = palo_task.TaskThread(update_task)
    update_thread.start()
    # 检测新master上是否有新image生成，如果有则停止update
    timeout = 300
    while timeout > 0:
        new_version = node_operator.get_image_version(new_master)
        print(new_version)
        if new_version and max(new_version) > max(version):
            break
        time.sleep(3)
        timeout -= 3
    # 停止update
    update_thread.stop()
    update_thread.join()
    time.sleep(5)
    # 再次切主
    node_operator.restart_fe(new_master)
    new_master = node_operator.get_master()
    time.sleep(30)
    # 查询表
    client.connect()
    assert client.select_all(table_name)
    client.clean(database_name)
    retry_times = 10
    while retry_times > 0:
        try:
            client.get_alive_backend_list()[0][palo_job.BackendProcInfo.Host]
            break
        except Exception as e:
            retry_times -= 1
            time.sleep(3)
            LOG.info(L("show proc failed"))


def test_update_when_one_be_down():
    """
    {
    "title": "test_update_when_one_be_down",
    "describe": "在执行update操作的时候，有一个be宕机",
    "tag": "function,p0,fuzz"
    }
    """
    # init
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client = common.create_workspace(database_name)
    partition_info = palo_client.PartitionInfo("k2",
                                               ["p1", "p2", "p3", "p4", "p5"],
                                               ["0", "100", "200", "300", "MAXVALUE"])
    distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
    keys_desc = "UNIQUE KEY(k1, k2, k3)"
    ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
                              partition_info=partition_info, distribution_info=distribution_info)
    assert ret, 'create table failed'
    ret = client.stream_load(table_name, FILE.partition_local_file)
    assert ret, 'stream load failed'
    # 持续执行update
    update_task = palo_task.SyncTask(client.update, table_name, ['v6=v6+1'], 'k1 > 0', database_name)
    update_thread = palo_task.TaskThread(update_task)
    update_thread.start()
    time.sleep(5)
    # update的过程中be宕机
    be_host = client.get_alive_backend_list()[0][palo_job.BackendProcInfo.Host]
    print('stop be ', be_host)
    assert node_operator.stop_be(be_host)
    update_succ_count = int(update_task.succ_count)
    print(update_succ_count)
    # 停止update
    time.sleep(30)
    update_thread.stop()
    update_thread.join()
    print(update_task.succ_count)
    assert update_task.succ_count > update_succ_count
    assert node_operator.start_be(be_host)
    time.sleep(5)
    client.clean(database_name)


def test_update_when_be_restart():
    """
    {
    "title": "test_update_when_multi_be_down",
    "describe": "在执行update操作的时候，有be重启",
    "tag": "function,p0,fuzz"
    }
    """
    # init
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client = common.create_workspace(database_name)
    partition_info = palo_client.PartitionInfo("k2",
                                               ["p1", "p2", "p3", "p4", "p5"],
                                               ["0", "100", "200", "300", "MAXVALUE"])
    distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
    keys_desc = "UNIQUE KEY(k1, k2, k3)"
    ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
                              partition_info=partition_info, distribution_info=distribution_info)
    assert ret, 'create table failed'
    ret = client.stream_load(table_name, FILE.partition_local_file)
    assert ret, 'stream load failed'
    # 持续执行update
    update_task = palo_task.SyncTask(client.update, table_name, ['v6=v6+1'], 'k1 > 0', database_name)
    update_thread = palo_task.TaskThread(update_task)
    update_thread.start()
    time.sleep(5)
    # upadte的过程中be宕机
    be_host = node_operator.get_be_list()[-1]
    print('stop be ', be_host)
    node_operator.restart_be(be_host)
    update_succ_count = int(update_task.succ_count)
    print(update_succ_count)
    time.sleep(10)
    assert node_operator.is_be_alive(be_host)
    print(update_succ_count)
    # 停止update
    time.sleep(30)
    update_thread.stop()
    update_thread.join()
    print(update_task.succ_count)
    print(update_succ_count)
    assert update_task.succ_count > update_succ_count
    client.clean(database_name)