ClickHouse/tests/integration/test_cluster_copier/test_three_nodes.py

import os
import sys
import time
import logging
import pytest

from helpers.cluster import ClickHouseCluster
from helpers.test_tools import TSV

import docker

CURRENT_TEST_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.dirname(CURRENT_TEST_DIR))

cluster = ClickHouseCluster(__file__, name="copier_test_three_nodes")


@pytest.fixture(scope="module")
def started_cluster():
    global cluster
    try:

        for name in ["first", "second", "third"]:
            cluster.add_instance(
                name,
                main_configs=[
                    "configs_three_nodes/conf.d/clusters.xml",
                    "configs_three_nodes/conf.d/ddl.xml",
                ],
                user_configs=["configs_three_nodes/users.xml"],
                with_zookeeper=True,
            )

        cluster.start()
        yield cluster

    finally:
        cluster.shutdown()


class Task:
    def __init__(self, cluster):
        self.cluster = cluster
        self.zk_task_path = "/clickhouse-copier/task"
        self.container_task_file = "/task_taxi_data.xml"

        for instance_name, _ in cluster.instances.items():
            instance = cluster.instances[instance_name]
            instance.copy_file_to_container(
                os.path.join(CURRENT_TEST_DIR, "./task_taxi_data.xml"),
                self.container_task_file,
            )
            logging.debug(
                f"Copied task file to container of '{instance_name}' instance. Path {self.container_task_file}"
            )

    def start(self):
        for name in ["first", "second", "third"]:
            node = cluster.instances[name]
            node.query("DROP DATABASE IF EXISTS dailyhistory SYNC;")
            node.query("DROP DATABASE IF EXISTS monthlyhistory SYNC;")

        first = cluster.instances["first"]

        # daily partition database
        first.query("CREATE DATABASE IF NOT EXISTS dailyhistory on cluster events;")
        first.query(
            """CREATE TABLE dailyhistory.yellow_tripdata_staging ON CLUSTER events
        (
            id UUID DEFAULT generateUUIDv4(),
            vendor_id String,
            tpep_pickup_datetime DateTime('UTC'),
            tpep_dropoff_datetime DateTime('UTC'),
            passenger_count Nullable(Float64),
            trip_distance String,
            pickup_longitude Float64,
            pickup_latitude Float64,
            rate_code_id String,
            store_and_fwd_flag String,
            dropoff_longitude Float64,
            dropoff_latitude Float64,
            payment_type String,
            fare_amount String,
            extra String,
            mta_tax String,
            tip_amount String,
            tolls_amount String,
            improvement_surcharge String,
            total_amount String,
            pickup_location_id String,
            dropoff_location_id String,
            congestion_surcharge String,
            junk1 String,  junk2 String
        )
        Engine = ReplacingMergeTree()
        PRIMARY KEY (tpep_pickup_datetime, id)
        ORDER BY (tpep_pickup_datetime, id)
        PARTITION BY (toYYYYMMDD(tpep_pickup_datetime))"""
        )

        first.query(
            """CREATE TABLE dailyhistory.yellow_tripdata
            ON CLUSTER events
            AS dailyhistory.yellow_tripdata_staging
            ENGINE = Distributed('events', 'dailyhistory', yellow_tripdata_staging, sipHash64(id) % 3);"""
        )

        first.query(
            """INSERT INTO dailyhistory.yellow_tripdata
            SELECT * FROM generateRandom(
                'id UUID DEFAULT generateUUIDv4(),
                vendor_id String,
                tpep_pickup_datetime DateTime(\\'UTC\\'),
                tpep_dropoff_datetime DateTime(\\'UTC\\'),
                passenger_count Nullable(Float64),
                trip_distance String,
                pickup_longitude Float64,
                pickup_latitude Float64,
                rate_code_id String,
                store_and_fwd_flag String,
                dropoff_longitude Float64,
                dropoff_latitude Float64,
                payment_type String,
                fare_amount String,
                extra String,
                mta_tax String,
                tip_amount String,
                tolls_amount String,
                improvement_surcharge String,
                total_amount String,
                pickup_location_id String,
                dropoff_location_id String,
                congestion_surcharge String,
                junk1 String,
                junk2 String',
            1, 10, 2) LIMIT 50;"""
        )

        # monthly partition database
        first.query("create database IF NOT EXISTS monthlyhistory on cluster events;")
        first.query(
            """CREATE TABLE monthlyhistory.yellow_tripdata_staging ON CLUSTER events
        (
            id UUID DEFAULT generateUUIDv4(),
            vendor_id String,
            tpep_pickup_datetime DateTime('UTC'),
            tpep_dropoff_datetime DateTime('UTC'),
            passenger_count Nullable(Float64),
            trip_distance String,
            pickup_longitude Float64,
            pickup_latitude Float64,
            rate_code_id String,
            store_and_fwd_flag String,
            dropoff_longitude Float64,
            dropoff_latitude Float64,
            payment_type String,
            fare_amount String,
            extra String,
            mta_tax String,
            tip_amount String,
            tolls_amount String,
            improvement_surcharge String,
            total_amount String,
            pickup_location_id String,
            dropoff_location_id String,
            congestion_surcharge String,
            junk1 String,
            junk2 String
        )
        Engine = ReplacingMergeTree()
        PRIMARY KEY (tpep_pickup_datetime, id)
        ORDER BY (tpep_pickup_datetime, id)
        PARTITION BY (pickup_location_id, toYYYYMM(tpep_pickup_datetime))"""
        )

        first.query(
            """CREATE TABLE monthlyhistory.yellow_tripdata
            ON CLUSTER events
            AS monthlyhistory.yellow_tripdata_staging
            ENGINE = Distributed('events', 'monthlyhistory', yellow_tripdata_staging, sipHash64(id) % 3);"""
        )

    def check(self):
        first = cluster.instances["first"]
        a = TSV(first.query("SELECT count() from dailyhistory.yellow_tripdata"))
        b = TSV(first.query("SELECT count() from monthlyhistory.yellow_tripdata"))
        assert a == b, "Distributed tables"

        for instance_name, instance in cluster.instances.items():
            instance = cluster.instances[instance_name]
            a = instance.query(
                "SELECT count() from dailyhistory.yellow_tripdata_staging"
            )
            b = instance.query(
                "SELECT count() from monthlyhistory.yellow_tripdata_staging"
            )
            assert a == b, "MergeTree tables on each shard"

            a = TSV(
                instance.query(
                    "SELECT sipHash64(*) from dailyhistory.yellow_tripdata_staging ORDER BY id"
                )
            )
            b = TSV(
                instance.query(
                    "SELECT sipHash64(*) from monthlyhistory.yellow_tripdata_staging ORDER BY id"
                )
            )

            assert a == b, "Data on each shard"

        for name in ["first", "second", "third"]:
            node = cluster.instances[name]
            node.query("DROP DATABASE IF EXISTS dailyhistory SYNC;")
            node.query("DROP DATABASE IF EXISTS monthlyhistory SYNC;")


def execute_task(started_cluster, task, cmd_options):
    task.start()

    zk = started_cluster.get_kazoo_client("zoo1")
    logging.debug("Use ZooKeeper server: {}:{}".format(zk.hosts[0][0], zk.hosts[0][1]))

    # Run cluster-copier processes on each node
    docker_api = started_cluster.docker_client.api
    copiers_exec_ids = []

    cmd = [
        "/usr/bin/clickhouse",
        "copier",
        "--config",
        "/etc/clickhouse-server/config-copier.xml",
        "--task-path",
        task.zk_task_path,
        "--task-file",
        task.container_task_file,
        "--task-upload-force",
        "true",
        "--base-dir",
        "/var/log/clickhouse-server/copier",
    ]
    cmd += cmd_options

    logging.debug(f"execute_task cmd: {cmd}")

    for instance_name in started_cluster.instances.keys():
        instance = started_cluster.instances[instance_name]
        container = instance.get_docker_handle()
        instance.copy_file_to_container(
            os.path.join(CURRENT_TEST_DIR, "configs_three_nodes/config-copier.xml"),
            "/etc/clickhouse-server/config-copier.xml",
        )
        logging.info("Copied copier config to {}".format(instance.name))
        exec_id = docker_api.exec_create(container.id, cmd, stderr=True)
        output = docker_api.exec_start(exec_id).decode("utf8")
        logging.info(output)
        copiers_exec_ids.append(exec_id)
        logging.info(
            "Copier for {} ({}) has started".format(instance.name, instance.ip_address)
        )

    # time.sleep(1000)

    # Wait for copiers stopping and check their return codes
    for exec_id, instance in zip(
        copiers_exec_ids, iter(started_cluster.instances.values())
    ):
        while True:
            res = docker_api.exec_inspect(exec_id)
            if not res["Running"]:
                break
            time.sleep(1)

        assert res["ExitCode"] == 0, "Instance: {} ({}). Info: {}".format(
            instance.name, instance.ip_address, repr(res)
        )

    try:
        task.check()
    finally:
        zk.delete(task.zk_task_path, recursive=True)


# Tests
@pytest.mark.timeout(600)
def test(started_cluster):
    execute_task(started_cluster, Task(started_cluster), [])