ClickHouse/utils/clickhouse-diagnostics/clickhouse-diagnostics
2021-12-25 15:45:08 +03:00

961 lines
26 KiB
Python

#!/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
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, port=8123, user=None):
self._session = requests.Session()
if user:
self._session.headers['X-ClickHouse-User'] = user
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, host):
self.args = args
self.host = 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 Yandex 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()
host = socket.getfqdn()
timestamp = datetime.strftime(datetime.now(), '%Y-%m-%d %H:%M:%S')
client = ClickhouseClient(host=host)
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, host)
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)
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 __name__ == '__main__':
main()