mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-12-17 20:02:05 +00:00
965 lines
27 KiB
Python
Executable File
965 lines
27 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import argparse
|
|
import gzip
|
|
import io
|
|
import json
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
from copy import deepcopy
|
|
from datetime import datetime
|
|
from typing import MutableMapping
|
|
|
|
import jinja2
|
|
from pandas import describe_option
|
|
import requests
|
|
import sqlparse
|
|
import tenacity
|
|
import xmltodict
|
|
import yaml
|
|
|
|
SELECT_VERSION = r'SELECT version()'
|
|
|
|
SELECT_UPTIME = r'''
|
|
{% if version_ge('21.3') -%}
|
|
SELECT formatReadableTimeDelta(uptime())
|
|
{% else -%}
|
|
SELECT
|
|
toString(floor(uptime() / 3600 / 24)) || ' days ' ||
|
|
toString(floor(uptime() % (24 * 3600) / 3600, 1)) || ' hours'
|
|
{% endif -%}
|
|
'''
|
|
|
|
SELECT_SYSTEM_TABLES = "SELECT name FROM system.tables WHERE database = 'system'"
|
|
|
|
SELECT_DATABASE_ENGINES = r'''SELECT
|
|
engine,
|
|
count() "count"
|
|
FROM system.databases
|
|
GROUP BY engine
|
|
'''
|
|
|
|
SELECT_DATABASES = r'''SELECT
|
|
name,
|
|
engine,
|
|
tables,
|
|
partitions,
|
|
parts,
|
|
formatReadableSize(bytes_on_disk) "disk_size"
|
|
FROM system.databases db
|
|
LEFT JOIN
|
|
(
|
|
SELECT
|
|
database,
|
|
uniq(table) "tables",
|
|
uniq(table, partition) "partitions",
|
|
count() AS parts,
|
|
sum(bytes_on_disk) "bytes_on_disk"
|
|
FROM system.parts
|
|
WHERE active
|
|
GROUP BY database
|
|
) AS db_stats ON db.name = db_stats.database
|
|
ORDER BY bytes_on_disk DESC
|
|
LIMIT 10
|
|
'''
|
|
|
|
SELECT_TABLE_ENGINES = r'''SELECT
|
|
engine,
|
|
count() "count"
|
|
FROM system.tables
|
|
WHERE database != 'system'
|
|
GROUP BY engine
|
|
'''
|
|
|
|
SELECT_DICTIONARIES = r'''SELECT
|
|
source,
|
|
type,
|
|
status,
|
|
count() "count"
|
|
FROM system.dictionaries
|
|
GROUP BY source, type, status
|
|
ORDER BY status DESC, source
|
|
'''
|
|
|
|
SELECT_ACCESS = "SHOW ACCESS"
|
|
|
|
SELECT_QUOTA_USAGE = "SHOW QUOTA"
|
|
|
|
SELECT_REPLICAS = r'''SELECT
|
|
database,
|
|
table,
|
|
is_leader,
|
|
is_readonly,
|
|
absolute_delay,
|
|
queue_size,
|
|
inserts_in_queue,
|
|
merges_in_queue
|
|
FROM system.replicas
|
|
ORDER BY absolute_delay DESC
|
|
LIMIT 10
|
|
'''
|
|
|
|
SELECT_REPLICATION_QUEUE = r'''SELECT
|
|
database,
|
|
table,
|
|
replica_name,
|
|
position,
|
|
node_name,
|
|
type,
|
|
source_replica,
|
|
parts_to_merge,
|
|
new_part_name,
|
|
create_time,
|
|
required_quorum,
|
|
is_detach,
|
|
is_currently_executing,
|
|
num_tries,
|
|
last_attempt_time,
|
|
last_exception,
|
|
concat('time: ', toString(last_postpone_time), ', number: ', toString(num_postponed), ', reason: ', postpone_reason) postpone
|
|
FROM system.replication_queue
|
|
ORDER BY create_time ASC
|
|
LIMIT 20
|
|
'''
|
|
|
|
SELECT_REPLICATED_FETCHES = r'''SELECT
|
|
database,
|
|
table,
|
|
round(elapsed, 1) "elapsed",
|
|
round(100 * progress, 1) "progress",
|
|
partition_id,
|
|
result_part_name,
|
|
result_part_path,
|
|
total_size_bytes_compressed,
|
|
bytes_read_compressed,
|
|
source_replica_path,
|
|
source_replica_hostname,
|
|
source_replica_port,
|
|
interserver_scheme,
|
|
to_detached,
|
|
thread_id
|
|
FROM system.replicated_fetches
|
|
'''
|
|
|
|
SELECT_PARTS_PER_TABLE = r'''SELECT
|
|
database,
|
|
table,
|
|
count() "partitions",
|
|
sum(part_count) "parts",
|
|
max(part_count) "max_parts_per_partition"
|
|
FROM
|
|
(
|
|
SELECT
|
|
database,
|
|
table,
|
|
partition,
|
|
count() "part_count"
|
|
FROM system.parts
|
|
WHERE active
|
|
GROUP BY database, table, partition
|
|
) partitions
|
|
GROUP BY database, table
|
|
ORDER BY max_parts_per_partition DESC
|
|
LIMIT 10
|
|
'''
|
|
|
|
SELECT_MERGES = r'''SELECT
|
|
database,
|
|
table,
|
|
round(elapsed, 1) "elapsed",
|
|
round(100 * progress, 1) "progress",
|
|
is_mutation,
|
|
partition_id,
|
|
{% if version_ge('20.3') -%}
|
|
result_part_path,
|
|
source_part_paths,
|
|
{% endif -%}
|
|
num_parts,
|
|
formatReadableSize(total_size_bytes_compressed) "total_size_compressed",
|
|
formatReadableSize(bytes_read_uncompressed) "read_uncompressed",
|
|
formatReadableSize(bytes_written_uncompressed) "written_uncompressed",
|
|
columns_written,
|
|
{% if version_ge('20.3') -%}
|
|
formatReadableSize(memory_usage) "memory_usage",
|
|
thread_id
|
|
{% else -%}
|
|
formatReadableSize(memory_usage) "memory_usage"
|
|
{% endif -%}
|
|
FROM system.merges
|
|
'''
|
|
|
|
SELECT_MUTATIONS = r'''SELECT
|
|
database,
|
|
table,
|
|
mutation_id,
|
|
command,
|
|
create_time,
|
|
{% if version_ge('20.3') -%}
|
|
parts_to_do_names,
|
|
{% endif -%}
|
|
parts_to_do,
|
|
is_done,
|
|
latest_failed_part,
|
|
latest_fail_time,
|
|
latest_fail_reason
|
|
FROM system.mutations
|
|
WHERE NOT is_done
|
|
ORDER BY create_time DESC
|
|
'''
|
|
|
|
SELECT_RECENT_DATA_PARTS = r'''SELECT
|
|
database,
|
|
table,
|
|
engine,
|
|
partition_id,
|
|
name,
|
|
{% if version_ge('20.3') -%}
|
|
part_type,
|
|
{% endif -%}
|
|
active,
|
|
level,
|
|
{% if version_ge('20.3') -%}
|
|
disk_name,
|
|
{% endif -%}
|
|
path,
|
|
marks,
|
|
rows,
|
|
bytes_on_disk,
|
|
data_compressed_bytes,
|
|
data_uncompressed_bytes,
|
|
marks_bytes,
|
|
modification_time,
|
|
remove_time,
|
|
refcount,
|
|
is_frozen,
|
|
min_date,
|
|
max_date,
|
|
min_time,
|
|
max_time,
|
|
min_block_number,
|
|
max_block_number
|
|
FROM system.parts
|
|
WHERE modification_time > now() - INTERVAL 3 MINUTE
|
|
ORDER BY modification_time DESC
|
|
'''
|
|
|
|
SELECT_DETACHED_DATA_PARTS = r'''SELECT
|
|
database,
|
|
table,
|
|
partition_id,
|
|
name,
|
|
disk,
|
|
reason,
|
|
min_block_number,
|
|
max_block_number,
|
|
level
|
|
FROM system.detached_parts
|
|
'''
|
|
|
|
SELECT_PROCESSES = r'''SELECT
|
|
elapsed,
|
|
query_id,
|
|
{% if normalize_queries -%}
|
|
normalizeQuery(query) AS normalized_query,
|
|
{% else -%}
|
|
query,
|
|
{% endif -%}
|
|
is_cancelled,
|
|
concat(toString(read_rows), ' rows / ', formatReadableSize(read_bytes)) AS read,
|
|
concat(toString(written_rows), ' rows / ', formatReadableSize(written_bytes)) AS written,
|
|
formatReadableSize(memory_usage) AS "memory usage",
|
|
user,
|
|
multiIf(empty(client_name), http_user_agent, concat(client_name, ' ', toString(client_version_major), '.', toString(client_version_minor), '.', toString(client_version_patch))) AS client,
|
|
{% if version_ge('21.3') -%}
|
|
thread_ids,
|
|
{% endif -%}
|
|
{% if version_ge('21.8') -%}
|
|
ProfileEvents,
|
|
Settings
|
|
{% else -%}
|
|
ProfileEvents.Names,
|
|
ProfileEvents.Values,
|
|
Settings.Names,
|
|
Settings.Values
|
|
{% endif -%}
|
|
FROM system.processes
|
|
ORDER BY elapsed DESC
|
|
'''
|
|
|
|
SELECT_TOP_QUERIES_BY_DURATION = r'''SELECT
|
|
type,
|
|
query_start_time,
|
|
query_duration_ms,
|
|
query_id,
|
|
query_kind,
|
|
is_initial_query,
|
|
{% if normalize_queries -%}
|
|
normalizeQuery(query) AS normalized_query,
|
|
{% else -%}
|
|
query,
|
|
{% endif -%}
|
|
concat(toString(read_rows), ' rows / ', formatReadableSize(read_bytes)) AS read,
|
|
concat(toString(written_rows), ' rows / ', formatReadableSize(written_bytes)) AS written,
|
|
concat(toString(result_rows), ' rows / ', formatReadableSize(result_bytes)) AS result,
|
|
formatReadableSize(memory_usage) AS "memory usage",
|
|
exception,
|
|
'\n' || stack_trace AS stack_trace,
|
|
user,
|
|
initial_user,
|
|
multiIf(empty(client_name), http_user_agent, concat(client_name, ' ', toString(client_version_major), '.', toString(client_version_minor), '.', toString(client_version_patch))) AS client,
|
|
client_hostname,
|
|
{% if version_ge('21.3') -%}
|
|
databases,
|
|
tables,
|
|
columns,
|
|
used_aggregate_functions,
|
|
used_aggregate_function_combinators,
|
|
used_database_engines,
|
|
used_data_type_families,
|
|
used_dictionaries,
|
|
used_formats,
|
|
used_functions,
|
|
used_storages,
|
|
used_table_functions,
|
|
thread_ids,
|
|
{% endif -%}
|
|
{% if version_ge('21.8') -%}
|
|
ProfileEvents,
|
|
Settings
|
|
{% else -%}
|
|
ProfileEvents.Names,
|
|
ProfileEvents.Values,
|
|
Settings.Names,
|
|
Settings.Values
|
|
{% endif -%}
|
|
FROM system.query_log
|
|
WHERE type != 'QueryStart'
|
|
AND event_date >= today() - 1
|
|
AND event_time >= now() - INTERVAL 1 DAY
|
|
ORDER BY query_duration_ms DESC
|
|
LIMIT 10
|
|
'''
|
|
|
|
SELECT_TOP_QUERIES_BY_MEMORY_USAGE = r'''SELECT
|
|
type,
|
|
query_start_time,
|
|
query_duration_ms,
|
|
query_id,
|
|
query_kind,
|
|
is_initial_query,
|
|
{% if normalize_queries -%}
|
|
normalizeQuery(query) AS normalized_query,
|
|
{% else -%}
|
|
query,
|
|
{% endif -%}
|
|
concat(toString(read_rows), ' rows / ', formatReadableSize(read_bytes)) AS read,
|
|
concat(toString(written_rows), ' rows / ', formatReadableSize(written_bytes)) AS written,
|
|
concat(toString(result_rows), ' rows / ', formatReadableSize(result_bytes)) AS result,
|
|
formatReadableSize(memory_usage) AS "memory usage",
|
|
exception,
|
|
'\n' || stack_trace AS stack_trace,
|
|
user,
|
|
initial_user,
|
|
multiIf(empty(client_name), http_user_agent, concat(client_name, ' ', toString(client_version_major), '.', toString(client_version_minor), '.', toString(client_version_patch))) AS client,
|
|
client_hostname,
|
|
{% if version_ge('21.3') -%}
|
|
databases,
|
|
tables,
|
|
columns,
|
|
used_aggregate_functions,
|
|
used_aggregate_function_combinators,
|
|
used_database_engines,
|
|
used_data_type_families,
|
|
used_dictionaries,
|
|
used_formats,
|
|
used_functions,
|
|
used_storages,
|
|
used_table_functions,
|
|
thread_ids,
|
|
{% endif -%}
|
|
{% if version_ge('21.8') -%}
|
|
ProfileEvents,
|
|
Settings
|
|
{% else -%}
|
|
ProfileEvents.Names,
|
|
ProfileEvents.Values,
|
|
Settings.Names,
|
|
Settings.Values
|
|
{% endif -%}
|
|
FROM system.query_log
|
|
WHERE type != 'QueryStart'
|
|
AND event_date >= today() - 1
|
|
AND event_time >= now() - INTERVAL 1 DAY
|
|
ORDER BY memory_usage DESC
|
|
LIMIT 10
|
|
'''
|
|
|
|
SELECT_FAILED_QUERIES = r'''SELECT
|
|
type,
|
|
query_start_time,
|
|
query_duration_ms,
|
|
query_id,
|
|
query_kind,
|
|
is_initial_query,
|
|
{% if normalize_queries -%}
|
|
normalizeQuery(query) AS normalized_query,
|
|
{% else -%}
|
|
query,
|
|
{% endif -%}
|
|
concat(toString(read_rows), ' rows / ', formatReadableSize(read_bytes)) AS read,
|
|
concat(toString(written_rows), ' rows / ', formatReadableSize(written_bytes)) AS written,
|
|
concat(toString(result_rows), ' rows / ', formatReadableSize(result_bytes)) AS result,
|
|
formatReadableSize(memory_usage) AS "memory usage",
|
|
exception,
|
|
'\n' || stack_trace AS stack_trace,
|
|
user,
|
|
initial_user,
|
|
multiIf(empty(client_name), http_user_agent, concat(client_name, ' ', toString(client_version_major), '.', toString(client_version_minor), '.', toString(client_version_patch))) AS client,
|
|
client_hostname,
|
|
{% if version_ge('21.3') -%}
|
|
databases,
|
|
tables,
|
|
columns,
|
|
used_aggregate_functions,
|
|
used_aggregate_function_combinators,
|
|
used_database_engines,
|
|
used_data_type_families,
|
|
used_dictionaries,
|
|
used_formats,
|
|
used_functions,
|
|
used_storages,
|
|
used_table_functions,
|
|
thread_ids,
|
|
{% endif -%}
|
|
{% if version_ge('21.8') -%}
|
|
ProfileEvents,
|
|
Settings
|
|
{% else -%}
|
|
ProfileEvents.Names,
|
|
ProfileEvents.Values,
|
|
Settings.Names,
|
|
Settings.Values
|
|
{% endif -%}
|
|
FROM system.query_log
|
|
WHERE type != 'QueryStart'
|
|
AND event_date >= today() - 1
|
|
AND event_time >= now() - INTERVAL 1 DAY
|
|
AND exception != ''
|
|
ORDER BY query_start_time DESC
|
|
LIMIT 10
|
|
'''
|
|
|
|
SELECT_STACK_TRACES = r'''SELECT
|
|
'\n' || arrayStringConcat(
|
|
arrayMap(
|
|
x,
|
|
y -> concat(x, ': ', y),
|
|
arrayMap(x -> addressToLine(x), trace),
|
|
arrayMap(x -> demangle(addressToSymbol(x)), trace)),
|
|
'\n') AS trace
|
|
FROM system.stack_trace
|
|
'''
|
|
|
|
SELECT_CRASH_LOG = r'''SELECT
|
|
event_time,
|
|
signal,
|
|
thread_id,
|
|
query_id,
|
|
'\n' || arrayStringConcat(trace_full, '\n') AS trace,
|
|
version
|
|
FROM system.crash_log
|
|
ORDER BY event_time DESC
|
|
'''
|
|
|
|
|
|
def retry(exception_types, max_attempts=5, max_interval=5):
|
|
"""
|
|
Function decorator that retries wrapped function on failures.
|
|
"""
|
|
return tenacity.retry(
|
|
retry=tenacity.retry_if_exception_type(exception_types),
|
|
wait=tenacity.wait_random_exponential(multiplier=0.5, max=max_interval),
|
|
stop=tenacity.stop_after_attempt(max_attempts),
|
|
reraise=True)
|
|
|
|
|
|
class ClickhouseError(Exception):
|
|
"""
|
|
ClickHouse interaction error.
|
|
"""
|
|
|
|
def __init__(self, response):
|
|
self.response = response
|
|
super().__init__(self.response.text.strip())
|
|
|
|
|
|
class ClickhouseClient:
|
|
"""
|
|
ClickHouse client.
|
|
"""
|
|
|
|
def __init__(self, *, host="localhost", port=8123, user="default", password):
|
|
self._session = requests.Session()
|
|
if user:
|
|
self._session.headers['X-ClickHouse-User'] = user
|
|
self._session.headers['X-ClickHouse-Key'] = password
|
|
self._url = f'http://{host}:{port}'
|
|
self._timeout = 60
|
|
self._ch_version = None
|
|
|
|
@property
|
|
def clickhouse_version(self):
|
|
if self._ch_version is None:
|
|
self._ch_version = self.query(SELECT_VERSION)
|
|
|
|
return self._ch_version
|
|
|
|
@retry(requests.exceptions.ConnectionError)
|
|
def query(self, query, query_args=None, format=None, post_data=None, timeout=None, echo=False, dry_run=False):
|
|
"""
|
|
Execute query.
|
|
"""
|
|
if query_args:
|
|
query = self.render_query(query, **query_args)
|
|
|
|
if format:
|
|
query += f' FORMAT {format}'
|
|
|
|
if timeout is None:
|
|
timeout = self._timeout
|
|
|
|
if echo:
|
|
print(sqlparse.format(query, reindent=True), '\n')
|
|
|
|
if dry_run:
|
|
return None
|
|
|
|
try:
|
|
response = self._session.post(self._url,
|
|
params={
|
|
'query': query,
|
|
},
|
|
json=post_data,
|
|
timeout=timeout)
|
|
|
|
response.raise_for_status()
|
|
|
|
if format in ('JSON', 'JSONCompact'):
|
|
return response.json()
|
|
|
|
return response.text.strip()
|
|
except requests.exceptions.HTTPError as e:
|
|
raise ClickhouseError(e.response) from None
|
|
|
|
def render_query(self, query, **kwargs):
|
|
env = jinja2.Environment()
|
|
|
|
env.globals['version_ge'] = lambda version: version_ge(self.clickhouse_version, version)
|
|
|
|
template = env.from_string(query)
|
|
return template.render(kwargs)
|
|
|
|
|
|
class ClickhouseConfig:
|
|
"""
|
|
ClickHouse server configuration.
|
|
"""
|
|
|
|
def __init__(self, config):
|
|
self._config = config
|
|
|
|
def dump(self, mask_secrets=True):
|
|
config = deepcopy(self._config)
|
|
if mask_secrets:
|
|
self._mask_secrets(config)
|
|
|
|
return xmltodict.unparse(config, pretty=True)
|
|
|
|
@classmethod
|
|
def load(cls):
|
|
return ClickhouseConfig(cls._load_config('/var/lib/clickhouse/preprocessed_configs/config.xml'))
|
|
|
|
@staticmethod
|
|
def _load_config(config_path):
|
|
with open(config_path, 'r') as file:
|
|
return xmltodict.parse(file.read())
|
|
|
|
@classmethod
|
|
def _mask_secrets(cls, config):
|
|
if isinstance(config, MutableMapping):
|
|
for key, value in list(config.items()):
|
|
if isinstance(value, MutableMapping):
|
|
cls._mask_secrets(config[key])
|
|
elif key in ('password', 'secret_access_key', 'header', 'identity'):
|
|
config[key] = '*****'
|
|
|
|
|
|
class DiagnosticsData:
|
|
"""
|
|
Diagnostics data.
|
|
"""
|
|
|
|
def __init__(self, args):
|
|
self.args = args
|
|
self.host = args.host
|
|
self._sections = [{'section': None, 'data': {}}]
|
|
|
|
def add_string(self, name, value, section=None):
|
|
self._section(section)[name] = {
|
|
'type': 'string',
|
|
'value': value,
|
|
}
|
|
|
|
def add_xml_document(self, name, document, section=None):
|
|
self._section(section)[name] = {
|
|
'type': 'xml',
|
|
'value': document,
|
|
}
|
|
|
|
def add_query(self, name, query, result, section=None):
|
|
self._section(section)[name] = {
|
|
'type': 'query',
|
|
'query': query,
|
|
'result': result,
|
|
}
|
|
|
|
def add_command(self, name, command, result, section=None):
|
|
self._section(section)[name] = {
|
|
'type': 'command',
|
|
'command': command,
|
|
'result': result,
|
|
}
|
|
|
|
def dump(self, format):
|
|
if format.startswith('json'):
|
|
result = self._dump_json()
|
|
elif format.startswith('yaml'):
|
|
result = self._dump_yaml()
|
|
else:
|
|
result = self._dump_wiki()
|
|
|
|
if format.endswith('.gz'):
|
|
compressor = gzip.GzipFile(mode='wb', fileobj=sys.stdout.buffer)
|
|
compressor.write(result.encode())
|
|
else:
|
|
print(result)
|
|
|
|
def _section(self, name=None):
|
|
if self._sections[-1]['section'] != name:
|
|
self._sections.append({'section': name, 'data': {}})
|
|
|
|
return self._sections[-1]['data']
|
|
|
|
def _dump_json(self):
|
|
"""
|
|
Dump diagnostic data in JSON format.
|
|
"""
|
|
return json.dumps(self._sections, indent=2, ensure_ascii=False)
|
|
|
|
def _dump_yaml(self):
|
|
"""
|
|
Dump diagnostic data in YAML format.
|
|
"""
|
|
return yaml.dump(self._sections, default_flow_style=False, allow_unicode=True)
|
|
|
|
def _dump_wiki(self):
|
|
"""
|
|
Dump diagnostic data in Wiki format.
|
|
"""
|
|
|
|
def _write_title(buffer, value):
|
|
buffer.write(f'### {value}\n')
|
|
|
|
def _write_subtitle(buffer, value):
|
|
buffer.write(f'#### {value}\n')
|
|
|
|
def _write_string_item(buffer, name, item):
|
|
value = item['value']
|
|
if value != '':
|
|
value = f'**{value}**'
|
|
buffer.write(f'{name}: {value}\n')
|
|
|
|
def _write_xml_item(buffer, section_name, name, item):
|
|
if section_name:
|
|
buffer.write(f'##### {name}\n')
|
|
else:
|
|
_write_subtitle(buffer, name)
|
|
|
|
_write_result(buffer, item['value'], format='XML')
|
|
|
|
def _write_query_item(buffer, section_name, name, item):
|
|
if section_name:
|
|
buffer.write(f'##### {name}\n')
|
|
else:
|
|
_write_subtitle(buffer, name)
|
|
|
|
_write_query(buffer, item['query'])
|
|
_write_result(buffer, item['result'])
|
|
|
|
def _write_command_item(buffer, section_name, name, item):
|
|
if section_name:
|
|
buffer.write(f'##### {name}\n')
|
|
else:
|
|
_write_subtitle(buffer, name)
|
|
|
|
_write_command(buffer, item['command'])
|
|
_write_result(buffer, item['result'])
|
|
|
|
def _write_unknown_item(buffer, section_name, name, item):
|
|
if section_name:
|
|
buffer.write(f'**{name}**\n')
|
|
else:
|
|
_write_subtitle(buffer, name)
|
|
|
|
json.dump(item, buffer, indent=2)
|
|
|
|
def _write_query(buffer, query):
|
|
buffer.write('**query**\n')
|
|
buffer.write('```sql\n')
|
|
buffer.write(query)
|
|
buffer.write('\n```\n')
|
|
|
|
def _write_command(buffer, command):
|
|
buffer.write('**command**\n')
|
|
buffer.write('```\n')
|
|
buffer.write(command)
|
|
buffer.write('\n```\n')
|
|
|
|
def _write_result(buffer, result, format=None):
|
|
buffer.write('**result**\n')
|
|
buffer.write(f'```{format}\n' if format else '```\n')
|
|
buffer.write(result)
|
|
buffer.write('\n```\n')
|
|
|
|
buffer = io.StringIO()
|
|
|
|
_write_title(buffer, f'Diagnostics data for host {self.host}')
|
|
for section in self._sections:
|
|
section_name = section['section']
|
|
if section_name:
|
|
_write_subtitle(buffer, section_name)
|
|
|
|
for name, item in section['data'].items():
|
|
if item['type'] == 'string':
|
|
_write_string_item(buffer, name, item)
|
|
elif item['type'] == 'query':
|
|
_write_query_item(buffer, section_name, name, item)
|
|
elif item['type'] == 'command':
|
|
_write_command_item(buffer, section_name, name, item)
|
|
elif item['type'] == 'xml':
|
|
_write_xml_item(buffer, section_name, name, item)
|
|
else:
|
|
_write_unknown_item(buffer, section_name, name, item)
|
|
|
|
return buffer.getvalue()
|
|
|
|
|
|
def main():
|
|
"""
|
|
Program entry point.
|
|
"""
|
|
args = parse_args()
|
|
timestamp = datetime.strftime(datetime.now(), '%Y-%m-%d %H:%M:%S')
|
|
client = ClickhouseClient(host=args.host, port=args.port, user=args.user, password=args.password)
|
|
ch_config = ClickhouseConfig.load()
|
|
version = client.clickhouse_version
|
|
system_tables = [row[0] for row in execute_query(client, SELECT_SYSTEM_TABLES, format='JSONCompact')['data']]
|
|
|
|
diagnostics = DiagnosticsData(args)
|
|
diagnostics.add_string('Version', version)
|
|
diagnostics.add_string('Timestamp', timestamp)
|
|
diagnostics.add_string('Uptime', execute_query(client, SELECT_UPTIME))
|
|
|
|
diagnostics.add_xml_document('ClickHouse configuration', ch_config.dump())
|
|
|
|
if version_ge(version, '20.8'):
|
|
add_query(diagnostics, 'Access configuration',
|
|
client=client,
|
|
query=SELECT_ACCESS,
|
|
format='TSVRaw')
|
|
add_query(diagnostics, 'Quotas',
|
|
client=client,
|
|
query=SELECT_QUOTA_USAGE,
|
|
format='Vertical')
|
|
|
|
add_query(diagnostics, 'Database engines',
|
|
client=client,
|
|
query=SELECT_DATABASE_ENGINES,
|
|
format='PrettyCompactNoEscapes',
|
|
section='Schema')
|
|
add_query(diagnostics, 'Databases (top 10 by size)',
|
|
client=client,
|
|
query=SELECT_DATABASES,
|
|
format='PrettyCompactNoEscapes',
|
|
section='Schema')
|
|
add_query(diagnostics, 'Table engines',
|
|
client=client,
|
|
query=SELECT_TABLE_ENGINES,
|
|
format='PrettyCompactNoEscapes',
|
|
section='Schema')
|
|
add_query(diagnostics, 'Dictionaries',
|
|
client=client,
|
|
query=SELECT_DICTIONARIES,
|
|
format='PrettyCompactNoEscapes',
|
|
section='Schema')
|
|
|
|
add_query(diagnostics, 'Replicated tables (top 10 by absolute delay)',
|
|
client=client,
|
|
query=SELECT_REPLICAS,
|
|
format='PrettyCompactNoEscapes',
|
|
section='Replication')
|
|
add_query(diagnostics, 'Replication queue (top 20 oldest tasks)',
|
|
client=client,
|
|
query=SELECT_REPLICATION_QUEUE,
|
|
format='Vertical',
|
|
section='Replication')
|
|
if version_ge(version, '21.3'):
|
|
add_query(diagnostics, 'Replicated fetches',
|
|
client=client,
|
|
query=SELECT_REPLICATED_FETCHES,
|
|
format='Vertical',
|
|
section='Replication')
|
|
|
|
add_query(diagnostics, 'Top 10 tables by max parts per partition',
|
|
client=client,
|
|
query=SELECT_PARTS_PER_TABLE,
|
|
format='PrettyCompactNoEscapes')
|
|
add_query(diagnostics, 'Merges in progress',
|
|
client=client,
|
|
query=SELECT_MERGES,
|
|
format='Vertical')
|
|
add_query(diagnostics, 'Mutations in progress',
|
|
client=client,
|
|
query=SELECT_MUTATIONS,
|
|
format='Vertical')
|
|
add_query(diagnostics, 'Recent data parts (modification time within last 3 minutes)',
|
|
client=client,
|
|
query=SELECT_RECENT_DATA_PARTS,
|
|
format='Vertical')
|
|
|
|
add_query(diagnostics, 'system.detached_parts',
|
|
client=client,
|
|
query=SELECT_DETACHED_DATA_PARTS,
|
|
format='PrettyCompactNoEscapes',
|
|
section='Detached data')
|
|
add_command(diagnostics, 'Disk space usage',
|
|
command='du -sh -L -c /var/lib/clickhouse/data/*/*/detached/* | sort -rsh',
|
|
section='Detached data')
|
|
|
|
add_query(diagnostics, 'Queries in progress (process list)',
|
|
client=client,
|
|
query=SELECT_PROCESSES,
|
|
format='Vertical',
|
|
section='Queries')
|
|
add_query(diagnostics, 'Top 10 queries by duration',
|
|
client=client,
|
|
query=SELECT_TOP_QUERIES_BY_DURATION,
|
|
format='Vertical',
|
|
section='Queries')
|
|
add_query(diagnostics, 'Top 10 queries by memory usage',
|
|
client=client,
|
|
query=SELECT_TOP_QUERIES_BY_MEMORY_USAGE,
|
|
format='Vertical',
|
|
section='Queries')
|
|
add_query(diagnostics, 'Last 10 failed queries',
|
|
client=client,
|
|
query=SELECT_FAILED_QUERIES,
|
|
format='Vertical',
|
|
section='Queries')
|
|
|
|
add_query(diagnostics, 'Stack traces',
|
|
client=client,
|
|
query=SELECT_STACK_TRACES,
|
|
format='Vertical')
|
|
|
|
if 'crash_log' in system_tables:
|
|
add_query(diagnostics, 'Crash log',
|
|
client=client,
|
|
query=SELECT_CRASH_LOG,
|
|
format='Vertical')
|
|
|
|
add_command(diagnostics, 'uname', 'uname -a')
|
|
|
|
diagnostics.dump(args.format)
|
|
|
|
|
|
def parse_args():
|
|
"""
|
|
Parse command-line arguments.
|
|
"""
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--format',
|
|
choices=['json', 'yaml', 'json.gz', 'yaml.gz', 'wiki', 'wiki.gz'],
|
|
default='wiki')
|
|
parser.add_argument('--normalize-queries',
|
|
action='store_true',
|
|
default=False)
|
|
parser.add_argument('--host', dest="host", help="clickhouse host")
|
|
parser.add_argument('--port', dest="port", default=8123, help="clickhouse http port")
|
|
parser.add_argument('--user', dest="user", default="default", help="clickhouse user")
|
|
parser.add_argument('--password', dest="password", help="clickhouse password")
|
|
return parser.parse_args()
|
|
|
|
|
|
def add_query(diagnostics, name, client, query, format, section=None):
|
|
query_args = {
|
|
'normalize_queries': diagnostics.args.normalize_queries,
|
|
}
|
|
query = client.render_query(query, **query_args)
|
|
diagnostics.add_query(
|
|
name=name,
|
|
query=query,
|
|
result=execute_query(client, query, render_query=False, format=format),
|
|
section=section)
|
|
|
|
|
|
def execute_query(client, query, render_query=True, format=None):
|
|
if render_query:
|
|
query = client.render_query(query)
|
|
|
|
try:
|
|
return client.query(query, format=format)
|
|
except Exception as e:
|
|
return repr(e)
|
|
|
|
|
|
def add_command(diagnostics, name, command, section=None):
|
|
diagnostics.add_command(
|
|
name=name,
|
|
command=command,
|
|
result=execute_command(command),
|
|
section=section)
|
|
|
|
|
|
def execute_command(command, input=None):
|
|
proc = subprocess.Popen(command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
|
|
if isinstance(input, str):
|
|
input = input.encode()
|
|
|
|
stdout, stderr = proc.communicate(input=input)
|
|
|
|
if proc.returncode:
|
|
return f'failed with exit code {proc.returncode}\n{stderr.decode()}'
|
|
|
|
return stdout.decode()
|
|
|
|
|
|
def version_ge(version1, version2):
|
|
"""
|
|
Return True if version1 is greater or equal than version2.
|
|
"""
|
|
return parse_version(version1) >= parse_version(version2)
|
|
|
|
|
|
def parse_version(version):
|
|
"""
|
|
Parse version string.
|
|
"""
|
|
return [int(x) for x in version.strip().split('.') if x.isnumeric()]
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|