datastore/ci/jobs/scripts/functional_tests/setup_log_cluster.sh

#!/bin/bash

set -e
# This script sets up export of system log tables to a remote server.
# Remote tables are created if not exist, and augmented with extra columns,
# and their names will contain a hash of the table structure,
# which allows exporting tables from servers of different versions.

# Config file contains KEY=VALUE pairs with any necessary parameters like:
# CLICKHOUSE_CI_LOGS_HOST - remote host
# CLICKHOUSE_CI_LOGS_USER - password for user
# CLICKHOUSE_CI_LOGS_PASSWORD - password for user

# Pre-configured destination cluster, where to export the data
CLICKHOUSE_CI_LOGS_CLUSTER=${CLICKHOUSE_CI_LOGS_CLUSTER:-system_logs_export}

EXTRA_COLUMNS=${EXTRA_COLUMNS:-"repo LowCardinality(String), pull_request_number UInt32, commit_sha String, check_start_time DateTime('UTC'), check_name LowCardinality(String), instance_type LowCardinality(String), instance_id String, INDEX ix_repo (repo) TYPE set(100), INDEX ix_pr (pull_request_number) TYPE set(100), INDEX ix_commit (commit_sha) TYPE set(100), INDEX ix_check_time (check_start_time) TYPE minmax, "}
echo "EXTRA_COLUMNS_EXPRESSION=${EXTRA_COLUMNS_EXPRESSION:?}"
EXTRA_ORDER_BY_COLUMNS=${EXTRA_ORDER_BY_COLUMNS:-"check_name"}

function __set_connection_args()
{
    # It's impossible to use a generic $CONNECTION_ARGS string, it's unsafe from word splitting perspective.
    # That's why we must stick to the generated option
    CONNECTION_ARGS=(
        --receive_timeout=45 --send_timeout=45 --secure
        --user "${CLICKHOUSE_CI_LOGS_USER:?}" --host "${CLICKHOUSE_CI_LOGS_HOST:?}"
        --password "${CLICKHOUSE_CI_LOGS_PASSWORD:?}"
    )
}

function __shadow_credentials()
{
    # The function completely screws the output, it shouldn't be used in normal functions, only in ()
    # The only way to substitute the env as a plain text is using perl 's/\Qsomething\E/another/
    exec &> >(perl -pe '
        s(\Q$ENV{CLICKHOUSE_CI_LOGS_HOST}\E)[CLICKHOUSE_CI_LOGS_HOST]g;
        s(\Q$ENV{CLICKHOUSE_CI_LOGS_USER}\E)[CLICKHOUSE_CI_LOGS_USER]g;
        s(\Q$ENV{CLICKHOUSE_CI_LOGS_PASSWORD}\E)[CLICKHOUSE_CI_LOGS_PASSWORD]g;
    ')
}

function check_logs_credentials()
{
    # The function connects with given credentials, and if it's unable to execute the simplest query, returns exit code

    set +x
    echo "Check CI Log cluster..."
    __set_connection_args
    __shadow_credentials
    local code
    # The remote CI logs cluster occasionally fails a probe for reasons that are
    # transient and unrelated to our credentials: it resets the connection, or
    # it is momentarily over its memory limit and rejects the query with
    # MEMORY_LIMIT_EXCEEDED (its RSS crosses the cap for a few seconds under
    # load from other CI jobs). Retry only those transient cases. We match on
    # the stderr message rather than the exit status: the shell keeps only the
    # low 8 bits of the client's exit code, so unrelated errors (e.g. 466/722)
    # alias to 210, and the client also raises NETWORK_ERROR for permanent
    # cases (protocol mismatch, generic NetException). A short --connect_timeout
    # (instead of the 10s default) is applied to the probe only, so a real
    # outage or a non-recoverable error (auth, config, DNS) fails fast: this
    # probe is best-effort and must not slow every CI job. Export queries keep
    # the default timeouts.
    local attempt err
    # BEGIN: logs-cluster connectivity probe loop
    for attempt in 1 2 3; do
        # Catch both success and error to not fail on `set -e`. Capture stderr
        # so we can distinguish transient failures from permanent ones.
        err=$(clickhouse-client "${CONNECTION_ARGS[@]:?}" --connect_timeout 3 -q 'SELECT 1 FORMAT Null' 2>&1) && return 0 || code=$?
        # Only transient failures are worth retrying; anything else will not
        # recover on retry, so stop immediately.
        if [[ "$err" != *"Connection reset by peer"* && "$err" != *"MEMORY_LIMIT_EXCEEDED"* ]]; then
            break
        fi
        echo "Attempt ${attempt}/3 to connect to CI Logs cluster failed (transient error), retrying"
        [ "$attempt" -lt 3 ] && sleep "$((attempt + 1))"
    done
    # END: logs-cluster connectivity probe loop
    echo 'Failed to connect to CI Logs cluster'
    echo "$err"
    return "$code"
}

function setup_logs_replication()
{
    # The function is launched in a separate shell instance to not expose the
    # exported values
    set +x

    if [[ -n "$CLICKHOUSE_CI_LOGS_HOST" ]]; then
        check_logs_credentials
    else
        echo 'No CI logs creds found, tables check will be skipped'
    fi

    echo "My hostname is ${HOSTNAME}"

    echo 'Create all configured system logs'
    clickhouse-client --query "SYSTEM FLUSH LOGS"

    debug_or_sanitizer_build=$(clickhouse-client -q "WITH ((SELECT value FROM system.build_options WHERE name='BUILD_TYPE') AS build, (SELECT value FROM system.build_options WHERE name='CXX_FLAGS') as flags) SELECT build='Debug' OR flags LIKE '%fsanitize%'")
    echo "Build is debug or sanitizer: $debug_or_sanitizer_build"

    # For each system log table:
    echo 'Create %_log tables'
    clickhouse-client --query "SHOW TABLES FROM system LIKE '%\\_log'" | while read -r table
    do
        EXTRA_COLUMNS_FOR_TABLE="${EXTRA_COLUMNS}"
        EXTRA_COLUMNS_EXPRESSION_FOR_TABLE="${EXTRA_COLUMNS_EXPRESSION}"

        # Calculate hash of its structure according to the columns and their types, including extra columns
        hash=$(clickhouse-client --query "
            SELECT sipHash64(groupArray((SELECT columns FROM external)), groupArray((name, type)))
            FROM (SELECT name, type FROM system.columns
                WHERE database = 'system' AND table = '$table'
                ORDER BY position)
            " --external --name=external --file=- --structure='columns String' <<< "$EXTRA_COLUMNS_FOR_TABLE"
        )

        # Create the destination table with adapted name and structure:
        statement=$(clickhouse-client --format TSVRaw --query "SHOW CREATE TABLE system.${table}" | sed -r -e '
            s/^\($/('"$EXTRA_COLUMNS_FOR_TABLE"'/;
            s/^ORDER BY (([^\(].+?)|\((.+?)\))$/ORDER BY ('"$EXTRA_ORDER_BY_COLUMNS"', \2\3)/;
            s/^CREATE TABLE system\.\w+_log$/CREATE TABLE IF NOT EXISTS '"$table"'_'"$hash"'/;
            /^TTL /d
            /^SETTINGS /d
            /^COMMENT /d
            s/ COMMENT \x27([^\x27\\]|\\.)*\x27//g
            ')
        statement+=" SETTINGS use_const_adaptive_granularity = 1"

        echo -e "Creating remote destination table ${table}_${hash} with statement:" >&2

        echo "::group::${table}"
        # there's the only way big "$statement" can be printed without causing EAGAIN error
        # cat: write error: Resource temporarily unavailable
        statement_print="${statement}"
        if [ "${#statement_print}" -gt 4000 ]; then
          statement_print="${statement::1999}\n…\n${statement:${#statement}-1999}"
        fi
        echo -e "$statement_print"
        echo "::endgroup::"

        # The remote aggregation tables do not need per-column comments, so the
        # `sed` above strips them. Without that, wide tables like `metric_log`
        # (~1500 columns, each with a long description) expand into a `CREATE`
        # larger than the remote server's default `max_query_size` (256 KiB)
        # and the parse fails with `Code: 62. Max query size exceeded`.
        echo "$statement" | clickhouse-client --database_replicated_initial_query_timeout_sec=10 \
            --distributed_ddl_task_timeout=30 --distributed_ddl_output_mode=throw_only_active \
            "${CONNECTION_ARGS[@]:?}" || continue

        echo "Creating table system.${table}_sender" >&2

        # Create Distributed table and materialized view to watch on the original table:
        clickhouse-client --query "
            CREATE TABLE system.${table}_sender
            ENGINE = Distributed(${CLICKHOUSE_CI_LOGS_CLUSTER}, default, ${table}_${hash})
            SETTINGS flush_on_detach=0
            EMPTY AS
            SELECT ${EXTRA_COLUMNS_EXPRESSION_FOR_TABLE}, *
            FROM system.${table}
        " || continue

        echo "Creating materialized view system.${table}_watcher" >&2

        clickhouse-client --query "
            CREATE MATERIALIZED VIEW system.${table}_watcher
            TO system.${table}_sender
            DEFINER = ci_logs_sender
            AS
            SELECT ${EXTRA_COLUMNS_EXPRESSION_FOR_TABLE}, * FROM system.${table}
        " || continue
    done
}

function stop_logs_replication()
{
    # The `_sender` tables are `Distributed` engine with `flush_on_detach=0`
    # (set so a slow/unreachable remote cluster can't hang server shutdown),
    # and inserts into them via the `_watcher` materialized views are batched
    # and sent to the remote cluster in the background. Without an explicit
    # `SYSTEM FLUSH DISTRIBUTED` here, whatever is still queued locally - e.g.
    # the just-flushed rows from the caller's preceding `SYSTEM FLUSH LOGS` -
    # is silently discarded when the table below is dropped. Bound it with the
    # same `timeout` pattern as the drop step so an unreachable cluster cannot
    # block teardown indefinitely.
    echo "Flush pending log records to the remote cluster"
    ( clickhouse-client --query "select database||'.'||table from system.tables where database = 'system' and table like '%_sender'" | {
        tee /dev/stderr
    } | {
        timeout --verbose --preserve-status --signal TERM --kill-after 1m 5m xargs -n1 -P10 -r -i clickhouse-client --query "SYSTEM FLUSH DISTRIBUTED {}"
    } ) || echo "WARNING: failed to flush pending log records, some rows may be lost"

    echo "Detach all logs replication"
    clickhouse-client --query "select database||'.'||table from system.tables where database = 'system' and (table like '%_sender' or table like '%_watcher')" | {
        tee /dev/stderr
    } | {
        timeout --verbose --preserve-status --signal TERM --kill-after 5m 15m xargs -n1 -P10 -r -i clickhouse-client --query "drop table {}"
    }
}


while [[ "$#" -gt 0 ]]; do
    case $1 in
        --stop-log-replication)
            echo "Stopping log replication..."
            stop_logs_replication
            ;;
        --setup-logs-replication)
            echo "Setting up log replication..."
            setup_logs_replication
            ;;
        *)
            echo "Unknown option: $1"
            echo "Usage: $0 [--stop-log-replication | --setup-logs-replication ]"
            exit 1
            ;;
    esac
    shift
done