nngo/JQC_multi_tenant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
JQC_multi_tenant/scripts/db_health.py
T

1162 lines
48 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""
scripts/db_health.py -- MySQL health check + index optimizer
==============================================================
A standalone operations tool. It does NOT import the request layer, does not
touch the uploads tree, and by DEFAULT writes nothing at all -- the plain
invocation is strictly read-only (SELECTs against INFORMATION_SCHEMA / SHOW /
EXPLAIN).
Multi-tenant
------------
Every tenant has its OWN database built from the same migration chain, so a
schema check is per tenant, not per deployment. --tenant selects what to open:
--tenant default the SQLALCHEMY_DATABASE_URI bind (the default; this is
the whole story in single-tenant mode, and the fallback
bind in MT mode)
--tenant <slug> one tenant's database, via the control plane
--tenant all every non-deleted tenant in turn
Tenant modes need CONTROL_DATABASE_URL + CONTROL_FERNET_KEY, exactly like
control/backup.py:
set -a; . /etc/jqc/control.env; set +a
The SERVER-level checks (max_connections, wait_timeout, buffer pool, slow log,
sql_mode) describe the MySQL instance, not a schema, so they run ONCE against
the first database opened rather than once per tenant. Everything schema-shaped
-- hygiene, footprint, indexes, foreign keys, EXPLAIN -- runs per tenant, and
each finding is tagged with the tenant it came from.
What it checks
--------------
1. SQLAlchemy engine options pool_pre_ping / pool_recycle -- the two that
decide whether you get "MySQL server has
gone away" after an idle overnight.
2. Server + connection settings max_connections vs the Gunicorn pool,
wait_timeout vs pool_recycle,
innodb_buffer_pool_size vs the live data
size, slow-query-log state, sql_mode.
3. Schema hygiene non-InnoDB tables, non-utf8mb4 charsets,
mixed collations (a collation mismatch on a
join column silently disables the index).
4. Table + index footprint row estimates, data/index bytes.
5. Missing indexes a curated, individually justified list
checked against what is really there.
6. Redundant indexes leftmost-prefix duplicates. FK-backed
indexes are excluded from drop candidates --
dropping one fails with errno 150.
7. Unindexed foreign keys should be empty on InnoDB; a gap here means
every parent DELETE scans the child table.
8. EXPLAIN on hot queries the dashboard, both list pages, the SLA
cron, the mobile notification poll. Flags
full table scans, filesorts, temp tables.
What it can change (only when explicitly asked)
-----------------------------------------------
--apply-indexes CREATE the missing recommended indexes (idempotent --
re-checks INFORMATION_SCHEMA first, per CLAUDE.md rule 16;
CREATE INDEX IF NOT EXISTS needs MySQL >= 8.0.12).
--analyze ANALYZE TABLE on every table. Cheap, non-locking on
InnoDB, refreshes optimizer cardinality statistics.
Safe to run in production.
--optimize OPTIMIZE TABLE -- rebuilds each table to reclaim space
after large deletes. Requires --yes. This LOCKS each
table for the rebuild: maintenance window only.
It never DROPs anything. Redundant indexes are reported with the exact SQL to
run by hand, because "unused" is a judgement call the tool should not make for
you.
Usage
-----
cd /home/jqc/janitorial_qc
source venv/bin/activate
python scripts/db_health.py # read-only report
python scripts/db_health.py --json /tmp/db.json # + machine-readable
python scripts/db_health.py --no-explain # skip the EXPLAIN pass
python scripts/db_health.py --emit-migration migrations/versions/phase54_perf_indexes.py \
--revision phase54_perf_indexes
python scripts/db_health.py --apply-indexes # create missing indexes
python scripts/db_health.py --analyze # refresh statistics
python scripts/db_health.py --optimize --yes # rebuild (locks!)
Exit codes
----------
0 no findings at or above --fail-on
1 findings at or above --fail-on (default 'none' -- never fails)
2 could not connect / not a MySQL database / refused a destructive flag /
could not reach the control plane for a tenant selection
"""
import os
import re
import sys
import json
import argparse
from datetime import datetime
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from sqlalchemy import text # noqa: E402
from app import create_app, db # noqa: E402
# -- Output helpers ----------------------------------------------------------
class C:
OK = '\033[92m'
WARN = '\033[93m'
ERR = '\033[91m'
DIM = '\033[90m'
BOLD = '\033[1m'
END = '\033[0m'
_NO_COLOR = (not sys.stdout.isatty()) or bool(os.environ.get('NO_COLOR'))
def _c(code, s):
return s if _NO_COLOR else f'{code}{s}{C.END}'
def header(title):
print()
print(_c(C.BOLD, f'-- {title} ' + '-' * max(0, 72 - len(title))))
SEVERITY_ORDER = {'info': 0, 'low': 1, 'medium': 2, 'high': 3}
FINDINGS = []
# Which tenant the checks are currently running against. Findings from a
# --tenant all run are otherwise indistinguishable in the JSON report, and
# "issues.ix_issues_status_reported is missing" means something quite different
# depending on WHOSE database it is missing from.
CURRENT_TENANT = 'default'
def finding(severity, area, message, fix=None):
"""Record a finding and print it. severity: info|low|medium|high."""
FINDINGS.append({'severity': severity, 'area': area, 'tenant': CURRENT_TENANT,
'message': message, 'fix': fix})
tag = {
'info': _c(C.DIM, 'INFO '),
'low': _c(C.DIM, 'LOW '),
'medium': _c(C.WARN, 'MED '),
'high': _c(C.ERR, 'HIGH '),
}[severity]
print(f' {tag} {message}')
if fix:
for line in fix.strip().splitlines():
print(_c(C.DIM, f' {line}'))
def ok(message):
print(f' {_c(C.OK, "OK ")} {message}')
def human_bytes(n):
n = float(n or 0)
for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
if n < 1024:
return f'{n:.1f} {unit}'
n /= 1024
return f'{n:.1f} PB'
# -- The recommended index set -----------------------------------------------
#
# Each entry is justified by a specific query in the application. An index that
# nothing runs is pure write-amplification, so nothing speculative belongs here.
#
# (table, index_name, 'col, col', priority, why)
#
# priority: 'high' -- a hot path scans without it (every page load / every cron)
# 'medium' -- a heavy but less frequent report or filter
# 'low' -- worth having as the table grows
RECOMMENDED_INDEXES = [
# --- notifications: polled by the bell on every page AND by every iPad ---
('notifications', 'ix_notif_user_read_created', 'user_id, is_read, created_at',
'high',
"GET /api/v1/notifications and the in-app bell both run "
"WHERE user_id=? AND is_read=0 [AND created_at > ?] ORDER BY created_at. "
"Only user_id is indexed today, so MySQL reads every notification ever "
"sent to that user and then sorts them."),
('notifications', 'ix_notif_digest_user', 'digest_pending, user_id',
'medium',
"send_pending_digests() runs SELECT DISTINCT user_id WHERE "
"digest_pending=1. The existing single-column digest_pending index cannot "
"satisfy the DISTINCT from the index alone."),
# --- issues: the SLA cron scans these every 30 minutes, forever ----------
('issues', 'ix_issues_status_reported', 'status, reported_at',
'high',
"send_sla_alerts() selects every open/in_progress/pending_verification "
"issue every 30 min, and the issues list orders by reported_at DESC after "
"filtering on status. One composite serves both; ix_issues_status then "
"becomes a redundant prefix of it."),
('issues', 'ix_issues_resolved_at', 'resolved_at',
'medium',
"The dashboard 'Resolved Today' card and the SLA Compliance report both "
"filter on resolved_at ranges; the column is unindexed."),
('issues', 'ix_issues_area_status', 'area_id, status',
'medium',
"Every facility-scoped issue query outer-joins areas and ORs on "
"Area.facility_id. area_id carries only the plain FK index; pairing it "
"with status lets the legacy area-routed half of the OR resolve from the "
"index instead of reading rows."),
('issues', 'ix_issues_handler_status', 'handler_type, status',
'low',
"Dashboard handler-split cards and the ?handler_type= list filter."),
# --- inspections --------------------------------------------------------
('inspections', 'ix_insp_followup_status', 'follow_up_required, status',
'medium',
"The Pending Follow-ups dashboard card and the ?status=follow_up filter "
"run WHERE follow_up_required=1 AND status='completed' AND NOT EXISTS("
"child re-inspection)."),
('inspections', 'ix_insp_followup_assignee', 'follow_up_assigned_to',
'low',
"phase56 follow-up ownership (Inspection.follow_up_owned_by) filters on "
"this column. It is a late-added nullable FK -- confirm it is indexed."),
# --- audit_logs: append-only, and the fastest-growing table here ---------
('audit_logs', 'ix_audit_entity', 'entity_type, entity_id',
'medium',
"The audit trail is filtered by entity. entity_type alone is very low "
"cardinality (about a dozen distinct values), so it is close to useless "
"on its own."),
('audit_logs', 'ix_audit_user_created', 'user_id, created_at',
'low',
"Per-user audit filtering with the default created_at DESC ordering."),
# --- issue_comments -----------------------------------------------------
('issue_comments', 'ix_comments_issue_created', 'issue_id, created_at',
'medium',
"Both the web issue detail and GET /api/v1/issues/<id>/comments read the "
"thread ordered by created_at. The plain FK index on issue_id leaves the "
"sort to a filesort."),
# --- inspection_schedules: cron materialises due schedules --------------
# MT names this table inspection_schedules (phase34) and the due column
# next_run_at; phase34 already ships ix_ischd_active_next over exactly these
# two, so this entry normally reports as present. It stays listed because
# the checker is the record of WHICH index a hot query depends on.
('inspection_schedules', 'ix_ischd_active_next', 'active, next_run_at',
'medium',
"POST /inspection-schedules/run and the schedules page both read "
"WHERE active=1 AND next_run_at <= now ORDER BY next_run_at."),
# --- issue_links --------------------------------------------------------
# phase57 creates ix_issue_links_issue_id / ix_issue_links_linked_issue_id
# itself. Listed so the checker keeps agreeing with the schema: both are
# read on every issue detail page (Issue.all_links reads BOTH directions).
('issue_links', 'ix_issue_links_issue_id', 'issue_id', 'low',
"Issue.all_links() reads links_from on every issue detail page."),
('issue_links', 'ix_issue_links_linked_issue_id', 'linked_issue_id', 'low',
"Issue.all_links() reads links_to on the same page -- a link is stored "
"once and shown on both issues, so both directions are hot."),
# --- support ------------------------------------------------------------
('support_chat_messages', 'ix_scm_session_created', 'session_id, created_at',
'low',
"Transcript rendering reads a session's messages in created_at order."),
]
# -- Representative hot queries for the EXPLAIN pass -------------------------
#
# Literal-free: every value is a bound parameter, so this pass reflects what the
# app actually sends and cannot itself be a vector for anything.
EXPLAIN_QUERIES = [
('dashboard: open issues (scoped)',
"SELECT i.id FROM issues i LEFT JOIN areas a ON i.area_id = a.id "
"WHERE i.status IN ('open','in_progress')",
{}),
('dashboard: resolved today',
"SELECT COUNT(*) FROM issues WHERE status='resolved' "
"AND resolved_at >= :d0 AND resolved_at < :d1",
{'d0': '2026-01-01 00:00:00', 'd1': '2026-01-02 00:00:00'}),
('dashboard: pending follow-ups',
"SELECT COUNT(*) FROM inspections i WHERE i.follow_up_required = 1 "
"AND i.status = 'completed' AND NOT EXISTS "
"(SELECT 1 FROM inspections c WHERE c.parent_inspection_id = i.id)",
{}),
('issues list: page 1, newest first',
"SELECT i.id FROM issues i LEFT JOIN areas a ON i.area_id = a.id "
"ORDER BY i.reported_at DESC LIMIT 25",
{}),
('inspections list: page 1, newest first',
"SELECT i.id FROM inspections i ORDER BY i.inspection_date DESC LIMIT 25",
{}),
('SLA cron: candidate scan (every 30 min)',
"SELECT id FROM issues WHERE status IN "
"('open','in_progress','pending_verification')",
{}),
('mobile: unread notification poll',
"SELECT id FROM notifications WHERE user_id = :uid AND is_read = 0 "
"AND created_at > :since ORDER BY created_at ASC LIMIT 50",
{'uid': 1, 'since': '2026-01-01 00:00:00'}),
('digest cron: pending recipients',
"SELECT DISTINCT user_id FROM notifications WHERE digest_pending = 1",
{}),
('reports: avg score by facility (30d)',
"SELECT f.id, AVG(i.overall_score) FROM facilities f "
"JOIN inspections i ON f.id = i.facility_id "
"WHERE i.inspection_date >= :d0 AND i.status='completed' "
"AND i.overall_score IS NOT NULL GROUP BY f.id",
{'d0': '2026-01-01 00:00:00'}),
('audit trail: newest first',
"SELECT id FROM audit_logs ORDER BY created_at DESC LIMIT 50",
{}),
]
# -- Introspection -----------------------------------------------------------
def db_name(conn):
return conn.execute(text('SELECT DATABASE()')).scalar()
def is_mysql(conn):
return conn.dialect.name in ('mysql', 'mariadb')
def server_version(conn):
return conn.execute(text('SELECT VERSION()')).scalar()
def global_vars(conn, names):
out = {}
for n in names:
try:
row = conn.execute(
text('SHOW GLOBAL VARIABLES LIKE :n'), {'n': n}).fetchone()
if row:
out[row[0]] = row[1]
except Exception:
pass
return out
def fetch_tables(conn, schema):
rows = conn.execute(text("""
SELECT TABLE_NAME, ENGINE, TABLE_ROWS, DATA_LENGTH, INDEX_LENGTH,
TABLE_COLLATION
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = :s AND TABLE_TYPE = 'BASE TABLE'
ORDER BY (DATA_LENGTH + INDEX_LENGTH) DESC
"""), {'s': schema}).fetchall()
return [{'name': r[0], 'engine': r[1], 'rows': r[2] or 0,
'data': r[3] or 0, 'index': r[4] or 0, 'collation': r[5]}
for r in rows]
def fetch_indexes(conn, schema):
"""Return {table: {index_name: {'cols': [...], 'unique': bool}}}."""
rows = conn.execute(text("""
SELECT TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX, COLUMN_NAME, NON_UNIQUE
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = :s
ORDER BY TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX
"""), {'s': schema}).fetchall()
out = {}
for tbl, idx, seq, col, non_unique in rows:
entry = out.setdefault(tbl, {}).setdefault(
idx, {'cols': [], 'unique': not non_unique})
entry['cols'].append(col)
return out
def fetch_fk_protection(conn, schema):
"""Index names and columns MySQL relies on for FOREIGN KEY constraints.
Dropping such an index fails with errno 150, so these must never be
offered as drop candidates no matter how redundant they look.
"""
rows = conn.execute(text("""
SELECT TABLE_NAME, CONSTRAINT_NAME
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS
WHERE TABLE_SCHEMA = :s AND CONSTRAINT_TYPE = 'FOREIGN KEY'
"""), {'s': schema}).fetchall()
fk_names = {}
for tbl, name in rows:
fk_names.setdefault(tbl, set()).add(name)
rows = conn.execute(text("""
SELECT TABLE_NAME, COLUMN_NAME
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
WHERE TABLE_SCHEMA = :s AND REFERENCED_TABLE_NAME IS NOT NULL
"""), {'s': schema}).fetchall()
fk_cols = {}
for tbl, col in rows:
fk_cols.setdefault(tbl, set()).add(col)
return fk_names, fk_cols
def index_exists(conn, schema, table, index_name):
return conn.execute(text("""
SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = :s AND TABLE_NAME = :t AND INDEX_NAME = :i
"""), {'s': schema, 't': table, 'i': index_name}).scalar() > 0
def table_exists(conn, schema, table):
return conn.execute(text("""
SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = :s AND TABLE_NAME = :t
"""), {'s': schema, 't': table}).scalar() > 0
def column_exists(conn, schema, table, column):
return conn.execute(text("""
SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = :s AND TABLE_NAME = :t AND COLUMN_NAME = :c
"""), {'s': schema, 't': table, 'c': column}).scalar() > 0
# -- Checks ------------------------------------------------------------------
def check_engine_options(app):
header('SQLAlchemy engine options')
opts = app.config.get('SQLALCHEMY_ENGINE_OPTIONS') or {}
uri = app.config.get('SQLALCHEMY_DATABASE_URI', '')
if not opts:
finding('high', 'engine',
'SQLALCHEMY_ENGINE_OPTIONS is not set -- the pool runs on '
'library defaults with no liveness check and no recycle.',
"Add to config.Config:\n"
" SQLALCHEMY_ENGINE_OPTIONS = {\n"
" 'pool_pre_ping': True, # kills 'MySQL server has gone away'\n"
" 'pool_recycle': 1800, # must be < MySQL wait_timeout\n"
" 'pool_size': 5, 'max_overflow': 5,\n"
" 'pool_timeout': 30,\n"
" }")
else:
if not opts.get('pool_pre_ping'):
finding('high', 'engine',
'pool_pre_ping is off -- an idle connection killed by MySQL '
'surfaces as OperationalError 2006 on the next request.',
"SQLALCHEMY_ENGINE_OPTIONS['pool_pre_ping'] = True")
else:
ok('pool_pre_ping enabled')
recycle = opts.get('pool_recycle')
if not recycle or recycle < 0:
finding('medium', 'engine',
'pool_recycle is unset -- connections are held indefinitely '
'and can outlive the server-side wait_timeout.',
"SQLALCHEMY_ENGINE_OPTIONS['pool_recycle'] = 1800")
else:
ok(f'pool_recycle = {recycle}s')
if 'mysql' in uri and 'charset=' not in uri:
finding('low', 'engine',
'DATABASE_URL has no ?charset= -- PyMySQL negotiates the server '
'default, which on older installs is latin1.',
'Append ?charset=utf8mb4 to DATABASE_URL.')
# The options above govern the DEFAULT bind only. In MT mode nearly every
# request runs on a per-tenant engine built by app/tenancy/engine_cache.py,
# which has its own pool settings -- so reporting only the default pool
# would understate the connection ceiling by the factor that matters.
if app.config.get('MULTI_TENANT_ENABLED'):
print(f' {_c(C.DIM, "per-tenant pool")} '
f'size={app.config.get("TENANT_ENGINE_POOL_SIZE")} '
f'overflow={app.config.get("TENANT_ENGINE_MAX_OVERFLOW")} '
f'recycle={app.config.get("TENANT_ENGINE_POOL_RECYCLE")}s '
f'{_c(C.DIM, "engine cache cap")} '
f'{app.config.get("TENANT_ENGINE_CACHE_MAX")}')
ok('per-tenant engines set pool_pre_ping (engine_cache.get_tenant_engine)')
return opts
def check_server(conn, opts, tables, mt_config=None):
header('MySQL server settings')
print(f' {_c(C.DIM, "version")} {server_version(conn)}')
g = global_vars(conn, [
'max_connections', 'wait_timeout', 'interactive_timeout',
'innodb_buffer_pool_size', 'slow_query_log', 'long_query_time',
'sql_mode', 'character_set_server', 'collation_server',
'innodb_flush_log_at_trx_commit', 'table_open_cache',
'performance_schema',
])
# --- pool sizing vs max_connections -------------------------------------
try:
maxc = int(g.get('max_connections', 0))
except (TypeError, ValueError):
maxc = 0
pool_size = (opts or {}).get('pool_size', 5)
overflow = (opts or {}).get('max_overflow', 10)
try:
import multiprocessing
workers = multiprocessing.cpu_count() * 2 + 1 # gunicorn_config.py
except Exception:
workers = 9
# In MT the default bind is the SMALL half of the total. Each worker also
# caches up to TENANT_ENGINE_CACHE_MAX per-tenant engines, each with its own
# pool, so the real ceiling is
# workers x [ (default pool) + cache_cap x (tenant pool) ]
# -- the number that actually meets max_connections. Reporting only the
# default pool is how a box ends up at "Too many connections" while the
# health check says there is headroom.
per_worker = pool_size + overflow
detail = (f'pool_size {pool_size} + overflow {overflow}')
if mt_config:
t_size = int(mt_config.get('TENANT_ENGINE_POOL_SIZE') or 2)
t_over = int(mt_config.get('TENANT_ENGINE_MAX_OVERFLOW') or 3)
t_cap = int(mt_config.get('TENANT_ENGINE_CACHE_MAX') or 32)
per_worker += t_cap * (t_size + t_over)
detail += (f', plus {t_cap} cached tenant engines x '
f'(pool {t_size} + overflow {t_over})')
worst_case = workers * per_worker
print(f' {_c(C.DIM, "max_connections")} {maxc} '
f'{_c(C.DIM, "gunicorn workers")} ~{workers} '
f'{_c(C.DIM, "worst-case pool")} {worst_case}')
if maxc and worst_case > maxc * 0.8:
finding('high', 'server',
f'{workers} sync workers x ({detail}) = {worst_case} possible '
f'connections against max_connections={maxc}. Under load that '
f'returns "Too many connections" instead of queueing.',
'Cap the pools (DB_POOL_SIZE / DB_MAX_OVERFLOW for the default '
'bind, TENANT_ENGINE_POOL_SIZE / TENANT_ENGINE_MAX_OVERFLOW / '
'TENANT_ENGINE_CACHE_MAX for the tenant engines) or raise '
'max_connections. A sync worker only ever needs one connection '
'per engine in steady state, and the cache cap is the multiplier '
'worth lowering first.')
elif maxc:
ok(f'connection ceiling has headroom ({worst_case} worst case / {maxc})')
# --- wait_timeout vs pool_recycle ---------------------------------------
try:
wait = int(g.get('wait_timeout', 0))
except (TypeError, ValueError):
wait = 0
recycle = (opts or {}).get('pool_recycle')
if wait and recycle and recycle >= wait:
finding('high', 'server',
f'pool_recycle ({recycle}s) is not below wait_timeout ({wait}s) '
f'-- the server closes the connection before SQLAlchemy retires '
f'it, which is exactly the case pool_recycle exists to prevent.',
f'Set pool_recycle to roughly {int(wait * 0.6)}.')
elif wait:
ok(f'wait_timeout = {wait}s')
# --- buffer pool vs live data -------------------------------------------
try:
pool_bytes = int(g.get('innodb_buffer_pool_size', 0))
except (TypeError, ValueError):
pool_bytes = 0
total = sum(t['data'] + t['index'] for t in tables)
print(f' {_c(C.DIM, "innodb_buffer_pool_size")} {human_bytes(pool_bytes)} '
f'{_c(C.DIM, "data+indexes")} {human_bytes(total)}')
if pool_bytes and total > pool_bytes:
finding('medium', 'server',
f'The working set ({human_bytes(total)}) exceeds the InnoDB '
f'buffer pool ({human_bytes(pool_bytes)}) -- hot pages are being '
f'evicted and re-read from disk.',
'On a dedicated box set innodb_buffer_pool_size to about '
'60-70% of RAM.')
elif pool_bytes:
ok('the entire dataset fits in the InnoDB buffer pool')
# --- slow query log ------------------------------------------------------
if str(g.get('slow_query_log', 'OFF')).upper() in ('OFF', '0'):
finding('medium', 'observability',
'slow_query_log is OFF -- there is no record of which queries '
'are actually slow in production, so tuning stays guesswork.',
"SET GLOBAL slow_query_log = 'ON';\n"
"SET GLOBAL long_query_time = 0.5;\n"
"(and persist both in my.cnf so they survive a restart)")
else:
ok(f"slow_query_log ON (long_query_time={g.get('long_query_time')})")
# --- sql_mode ------------------------------------------------------------
mode = g.get('sql_mode', '') or ''
if 'STRICT_TRANS_TABLES' not in mode and 'STRICT_ALL_TABLES' not in mode:
finding('medium', 'integrity',
'sql_mode has no STRICT mode -- an over-long string or an '
'invalid ENUM value is silently coerced instead of raising.',
'Add STRICT_TRANS_TABLES to sql_mode in my.cnf.')
else:
ok('STRICT mode enabled')
return g
def check_schema_hygiene(conn, schema, tables):
header('Schema hygiene')
bad_engine = [t for t in tables if (t['engine'] or '').upper() != 'INNODB']
if bad_engine:
finding('high', 'schema',
'Non-InnoDB tables found: ' +
', '.join(f"{t['name']} ({t['engine']})" for t in bad_engine),
'MyISAM has no transactions and locks the whole table on write.\n'
+ '\n'.join(f'ALTER TABLE {t["name"]} ENGINE=InnoDB;'
for t in bad_engine))
else:
ok('all tables are InnoDB')
collations = {}
for t in tables:
collations.setdefault(t['collation'] or 'unknown', []).append(t['name'])
non_utf8mb4 = {c: n for c, n in collations.items()
if not str(c).startswith('utf8mb4')}
if non_utf8mb4:
for coll, names in non_utf8mb4.items():
finding('medium', 'schema',
f'{len(names)} table(s) are {coll}, not utf8mb4: '
+ ', '.join(names[:8]) + ('...' if len(names) > 8 else ''),
'utf8mb3/latin1 cannot store emoji or many non-Latin names, '
'and a join between two different collations cannot use an '
'index on that column.\n'
+ '\n'.join(
f'ALTER TABLE {n} CONVERT TO CHARACTER SET utf8mb4 '
f'COLLATE utf8mb4_unicode_ci;' for n in names[:8]))
else:
ok('all tables are utf8mb4')
if len(collations) > 1:
finding('low', 'schema',
'Mixed collations across tables: '
+ ', '.join(sorted(str(c) for c in collations))
+ '. Joining two columns with different collations silently '
'disables index use on the join.')
def check_table_footprint(tables):
header('Table footprint (largest first)')
print(' ' + _c(C.DIM, 'table'.ljust(32) + 'rows~'.rjust(12)
+ 'data'.rjust(12) + 'index'.rjust(12)))
for t in tables[:15]:
print(f' {t["name"][:32].ljust(32)}'
f'{str(t["rows"]).rjust(12)}'
f'{human_bytes(t["data"]).rjust(12)}'
f'{human_bytes(t["index"]).rjust(12)}')
for t in tables:
if t['index'] > t['data'] * 2 and t['data'] > 10 * 1024 * 1024:
finding('low', 'indexes',
f'{t["name"]}: index bytes ({human_bytes(t["index"])}) are '
f'more than twice the data ({human_bytes(t["data"])}) -- '
f'likely over-indexed, which slows every write.')
def check_missing_indexes(conn, schema, existing):
header('Missing indexes (recommended set)')
missing = []
for table, name, cols, prio, why in RECOMMENDED_INDEXES:
if not table_exists(conn, schema, table):
continue
col_list = [c.strip() for c in cols.split(',')]
if any(not column_exists(conn, schema, table, c) for c in col_list):
continue # column absent in this schema version -- skip quietly
tbl_idx = existing.get(table, {})
# Covered when some existing index STARTS with exactly these columns in
# this order: a longer index with the right prefix serves the same
# queries, so adding a duplicate would only cost write time.
covered = any(idx['cols'][:len(col_list)] == col_list
for idx in tbl_idx.values())
if covered:
continue
missing.append((table, name, cols, prio, why))
finding(prio, 'indexes',
f'{table}.{name} ({cols}) is missing',
f'{why}\nCREATE INDEX {name} ON {table} ({cols});')
if not missing:
ok('every recommended index is present')
return missing
def check_redundant_indexes(existing, fk_names, fk_cols):
header('Redundant indexes')
found = False
for table, idxs in sorted(existing.items()):
names = sorted(idxs.keys())
for a in names:
if a == 'PRIMARY':
continue
for b in names:
if a == b or b == 'PRIMARY':
continue
ca, cb = idxs[a]['cols'], idxs[b]['cols']
# a is redundant when b starts with all of a's columns and b is
# strictly longer.
if len(ca) >= len(cb) or cb[:len(ca)] != ca:
continue
if idxs[a]['unique']:
continue # a unique index is a constraint, not a hint
protected = (a in fk_names.get(table, set())
or (len(ca) == 1
and ca[0] in fk_cols.get(table, set())))
if protected:
finding('info', 'indexes',
f'{table}.{a} ({", ".join(ca)}) is a prefix of '
f'{b} ({", ".join(cb)}) but backs a FOREIGN KEY -- '
f'leave it alone (dropping it fails, errno 150).')
else:
found = True
finding('low', 'indexes',
f'{table}.{a} ({", ".join(ca)}) is fully covered by '
f'{b} ({", ".join(cb)}) -- it costs write time and '
f'buffer-pool space for nothing.',
'-- confirm against your slow log first, then:\n'
f'DROP INDEX {a} ON {table};')
break
if not found:
ok('no droppable duplicate indexes')
def check_unindexed_fks(conn, schema, existing):
header('Foreign keys without a usable index')
rows = conn.execute(text("""
SELECT TABLE_NAME, COLUMN_NAME
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
WHERE TABLE_SCHEMA = :s AND REFERENCED_TABLE_NAME IS NOT NULL
"""), {'s': schema}).fetchall()
bad = []
for tbl, col in rows:
idxs = existing.get(tbl, {})
if not any(i['cols'] and i['cols'][0] == col for i in idxs.values()):
bad.append((tbl, col))
if bad:
for tbl, col in bad:
finding('high', 'indexes',
f'{tbl}.{col} is a foreign key with no leading index -- '
f'every DELETE on the parent table scans this one.',
f'CREATE INDEX ix_{tbl}_{col} ON {tbl} ({col});')
else:
ok('every foreign key column has a leading index')
def check_explain(conn):
header('EXPLAIN on representative hot queries')
for label, sql, params in EXPLAIN_QUERIES:
try:
rows = conn.execute(text('EXPLAIN ' + sql), params).mappings().all()
except Exception as exc:
print(f' {_c(C.DIM, "skip ")} {label}: '
f'{str(exc).splitlines()[0][:90]}')
continue
problems = []
for r in rows:
typ = (r.get('type') or '').upper()
extra = r.get('Extra') or ''
tbl = r.get('table') or '?'
nrows = r.get('rows') or 0
if typ == 'ALL':
problems.append(f'full scan of {tbl} (~{nrows} rows)')
if 'Using filesort' in extra:
problems.append(f'filesort on {tbl}')
if 'Using temporary' in extra:
problems.append(f'temp table on {tbl}')
if problems:
finding('medium', 'query-plan',
f'{label}: ' + '; '.join(sorted(set(problems))), sql)
else:
ok(label)
# -- Mutating actions (opt-in only) ------------------------------------------
def apply_indexes(conn, schema, missing):
header('Applying missing indexes')
if not missing:
ok('nothing to apply')
return
for table, name, cols, prio, why in missing:
# Re-check rather than trusting the earlier pass: CREATE INDEX IF NOT
# EXISTS is unavailable before MySQL 8.0.12 (CLAUDE.md rule 16).
if index_exists(conn, schema, table, name):
print(f' {_c(C.DIM, "skip ")} {name} already exists')
continue
sql = f'CREATE INDEX {name} ON {table} ({cols})'
print(f' {_c(C.BOLD, "run ")} {sql}')
conn.execute(text(sql))
print(f' {_c(C.OK, "done ")} {name}')
def run_analyze(conn, tables):
header('ANALYZE TABLE (refresh optimizer statistics)')
for t in tables:
conn.execute(text(f'ANALYZE TABLE `{t["name"]}`'))
print(f' {_c(C.OK, "done ")} {t["name"]}')
def run_optimize(conn, tables):
header('OPTIMIZE TABLE (rebuild -- locks each table)')
for t in tables:
before = t['data'] + t['index']
conn.execute(text(f'OPTIMIZE TABLE `{t["name"]}`'))
print(f' {_c(C.OK, "done ")} {t["name"]} (was {human_bytes(before)})')
# -- Alembic migration emitter -----------------------------------------------
MIGRATION_TEMPLATE = '''"""Performance indexes -- generated by scripts/db_health.py
Adds the indexes the health check found missing. Every statement is guarded by
an INFORMATION_SCHEMA lookup, so this migration is safe to re-run (CLAUDE.md
rule 16: CREATE INDEX IF NOT EXISTS needs MySQL >= 8.0.12).
Generated: {generated}
Revision ID: {revision}
Revises: {down_revision}
"""
from alembic import op
import sqlalchemy as sa
# Revision ids must be <= 32 chars -- alembic_version.version_num is VARCHAR(32).
revision = '{revision}'
down_revision = '{down_revision}'
branch_labels = None
depends_on = None
INDEXES = [
{index_rows}
]
def _index_exists(conn, table, name):
return conn.execute(sa.text("""
SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t AND INDEX_NAME = :i
"""), {{'t': table, 'i': name}}).scalar() > 0
def upgrade():
conn = op.get_bind()
for table, name, cols in INDEXES:
if not _index_exists(conn, table, name):
op.execute(sa.text(f'CREATE INDEX {{name}} ON {{table}} ({{cols}})'))
def downgrade():
conn = op.get_bind()
for table, name, cols in INDEXES:
if _index_exists(conn, table, name):
op.execute(sa.text(f'DROP INDEX {{name}} ON {{table}}'))
'''
def emit_migration(path, missing, revision, down_revision):
rows = '\n'.join(
f' ({table!r}, {name!r}, {cols!r}),'
for table, name, cols, prio, why in missing
) or ' # nothing was missing at generation time'
body = MIGRATION_TEMPLATE.format(
generated=datetime.now().strftime('%Y-%m-%d %H:%M'),
revision=revision,
down_revision=down_revision,
index_rows=rows,
)
with open(path, 'w', encoding='utf-8') as fh:
fh.write(body)
print(f'\n {_c(C.OK, "wrote")} {path}')
print(_c(C.DIM, ' Review it, confirm down_revision is the real head '
'(flask db heads), then run flask db upgrade.'))
def current_head():
"""Best-effort read of the current Alembic head from the versions dir."""
vdir = os.path.join(os.path.dirname(__file__), '..', 'migrations', 'versions')
revs, downs = set(), set()
try:
for fn in os.listdir(vdir):
if not fn.endswith('.py'):
continue
with open(os.path.join(vdir, fn), encoding='utf-8') as fh:
src = fh.read()
m = re.search(r"^revision\s*=\s*['\"]([^'\"]+)", src, re.M)
d = re.search(r"^down_revision\s*=\s*['\"]([^'\"]+)", src, re.M)
if m:
revs.add(m.group(1))
if d:
downs.add(d.group(1))
except OSError:
return '<REPLACE_WITH_CURRENT_HEAD>'
heads = revs - downs
return sorted(heads)[0] if len(heads) == 1 else '<REPLACE_WITH_CURRENT_HEAD>'
# -- Tenant selection --------------------------------------------------------
def select_targets(selector):
"""Resolve --tenant into an ordered list of (label, db_uri | None).
A db_uri of None means "use the app's default bind" -- the only target in
single-tenant mode, and still a real database in MT (it is the fallback
bind and what the exempt paths run on).
Mirrors control/backup.py::_select_tenants so the two agree on what "all"
means: every tenant whose status is not 'deleted'. Reads the control plane
and nothing else -- this function never writes anywhere.
"""
if selector in (None, '', 'default'):
return [('default', None)]
from control.base import control_session
from control.models import Tenant
with control_session() as sess:
if selector == 'all':
tenants = (sess.query(Tenant)
.filter(Tenant.status != 'deleted')
.order_by(Tenant.id).all())
if not tenants:
raise ValueError('No tenants are registered in the control DB.')
else:
tenants = sess.query(Tenant).filter_by(slug=selector).all()
if not tenants:
raise ValueError(f'Tenant "{selector}" not found in the control DB.')
# db_uri decrypts the stored password, so read it inside the session.
return [(t.slug, t.db_uri) for t in tenants]
# -- Main --------------------------------------------------------------------
def main():
p = argparse.ArgumentParser(
description='MySQL health check and index optimizer for JQC. '
'Read-only unless an --apply/--analyze/--optimize flag is '
'given.')
p.add_argument('--tenant', default='default', metavar='SLUG|all|default',
help='which database to inspect: "default" (the '
'SQLALCHEMY_DATABASE_URI bind, the default), a tenant '
'slug, or "all" for every registered tenant. Tenant '
'modes read the control plane, so source '
'/etc/jqc/control.env first.')
p.add_argument('--json', metavar='PATH', help='write the full report as JSON')
p.add_argument('--no-explain', action='store_true',
help='skip the EXPLAIN pass')
p.add_argument('--apply-indexes', action='store_true',
help='CREATE the missing recommended indexes (idempotent)')
p.add_argument('--analyze', action='store_true',
help='ANALYZE TABLE on every table (safe, non-locking)')
p.add_argument('--optimize', action='store_true',
help='OPTIMIZE TABLE on every table -- LOCKS each table; '
'needs --yes')
p.add_argument('--emit-migration', metavar='PATH',
help='write an Alembic migration for the missing indexes')
p.add_argument('--revision', default='phaseNN_perf_indexes',
help='revision id for --emit-migration (max 32 chars)')
p.add_argument('--yes', action='store_true',
help='confirm locking operations')
p.add_argument('--fail-on', choices=['none', 'low', 'medium', 'high'],
default='none',
help='exit 1 when a finding of this severity or above exists')
args = p.parse_args()
if args.optimize and not args.yes:
print('--optimize rebuilds every table and holds a lock for the '
'duration. Re-run with --yes inside a maintenance window.')
return 2
if len(args.revision) > 32:
print('--revision must be <= 32 characters (alembic_version.version_num '
'is VARCHAR(32)).')
return 2
app = create_app(os.getenv('FLASK_ENV') or 'production')
with app.app_context():
engine_opts = check_engine_options(app)
mt_config = None
if app.config.get('MULTI_TENANT_ENABLED'):
mt_config = {k: app.config.get(k) for k in (
'TENANT_ENGINE_POOL_SIZE', 'TENANT_ENGINE_MAX_OVERFLOW',
'TENANT_ENGINE_CACHE_MAX')}
try:
targets = select_targets(args.tenant)
except Exception as exc:
print(_c(C.ERR, f'\nCould not resolve --tenant {args.tenant}: {exc}'))
print(_c(C.DIM, ' Tenant selection needs CONTROL_DATABASE_URL '
'and CONTROL_FERNET_KEY:\n'
' set -a; . /etc/jqc/control.env; set +a'))
return 2
# The server-level settings describe the MySQL instance, not a schema,
# so they are checked once against the first database opened rather than
# re-reported identically for every tenant.
server = None
per_tenant = {}
connected = 0
for label, db_uri in targets:
globals()['CURRENT_TENANT'] = label
# A tenant engine here is short-lived and deliberately tiny: this is
# a one-shot CLI, not a worker, and --tenant all would otherwise open
# a full pool per tenant against the very max_connections it is
# checking. NullPool keeps it to one connection at a time.
own_engine = None
try:
if db_uri is None:
conn = db.engine.connect()
else:
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool
own_engine = create_engine(db_uri, poolclass=NullPool,
future=True)
conn = own_engine.connect()
except Exception as exc:
finding('high', 'connection',
f'could not connect to "{label}": {exc}')
if own_engine is not None:
own_engine.dispose()
continue
try:
with conn:
if not is_mysql(conn):
print(_c(C.ERR,
f'\nThis script targets MySQL/MariaDB; "{label}" '
f'is "{conn.dialect.name}". Skipping.'))
continue
connected += 1
schema = db_name(conn)
print(_c(C.BOLD,
f'\nJQC database health check -- tenant "{label}", '
f'schema "{schema}" @ {datetime.now():%Y-%m-%d %H:%M}'))
tables = fetch_tables(conn, schema)
existing = fetch_indexes(conn, schema)
fk_names, fk_cols = fetch_fk_protection(conn, schema)
if server is None:
server = check_server(conn, engine_opts, tables,
mt_config=mt_config)
check_schema_hygiene(conn, schema, tables)
check_table_footprint(tables)
missing = check_missing_indexes(conn, schema, existing)
check_redundant_indexes(existing, fk_names, fk_cols)
check_unindexed_fks(conn, schema, existing)
if not args.no_explain:
check_explain(conn)
per_tenant[label] = {'schema': schema, 'tables': tables,
'missing_indexes': missing}
# -- opt-in mutations, per tenant ------------------------
if args.apply_indexes:
with conn.begin():
apply_indexes(conn, schema, missing)
if args.analyze:
run_analyze(conn, tables)
if args.optimize:
run_optimize(conn, tables)
finally:
if own_engine is not None:
own_engine.dispose()
globals()['CURRENT_TENANT'] = 'default'
if not connected:
print(_c(C.ERR, '\nNo database could be inspected.'))
return 2
# The migration is emitted ONCE for the union of what every inspected
# tenant is missing: the chain is shared, so an index needed by one
# tenant is created in all of them. Emitting one per tenant would
# produce conflicting revisions for a single chain.
if args.emit_migration:
seen, union = set(), []
for data in per_tenant.values():
for entry in data['missing_indexes']:
key = (entry[0], entry[1])
if key not in seen:
seen.add(key)
union.append(entry)
emit_migration(args.emit_migration, union,
args.revision, current_head())
# -- summary ---------------------------------------------------------
header('Summary')
counts = {'high': 0, 'medium': 0, 'low': 0, 'info': 0}
for f in FINDINGS:
counts[f['severity']] += 1
print(f' {_c(C.DIM, "databases inspected")} {connected} '
f'{_c(C.ERR, "high")} {counts["high"]} '
f'{_c(C.WARN, "medium")} {counts["medium"]} '
f'{_c(C.DIM, "low")} {counts["low"]} '
f'{_c(C.DIM, "info")} {counts["info"]}')
if len(per_tenant) > 1:
for label, data in per_tenant.items():
n = len(data['missing_indexes'])
tag = _c(C.OK, 'OK ') if not n else _c(C.WARN, f'{n:<4}')
note = ('missing index(es)' if n
else '-- all recommended indexes present')
print(f' {tag} {label} ({data["schema"]}) {note}')
if not (args.apply_indexes or args.analyze or args.optimize):
print(_c(C.DIM, ' (read-only run -- nothing was modified)'))
if args.json:
report = {
'generated_at': datetime.now().isoformat(timespec='seconds'),
'tenant_selector': args.tenant,
'server': server,
'engine_options': {k: str(v) for k, v in (engine_opts or {}).items()},
'tenant_engine_options': {k: str(v) for k, v
in (mt_config or {}).items()},
'tenants': {
label: {
'schema': data['schema'],
'tables': data['tables'],
'missing_indexes': [
{'table': t, 'name': n, 'columns': c,
'priority': pr, 'why': w}
for t, n, c, pr, w in data['missing_indexes']
],
}
for label, data in per_tenant.items()
},
'findings': FINDINGS,
'counts': counts,
}
with open(args.json, 'w', encoding='utf-8') as fh:
json.dump(report, fh, indent=2, default=str)
print(f' report written to {args.json}')
if args.fail_on != 'none':
threshold = SEVERITY_ORDER[args.fail_on]
if any(SEVERITY_ORDER[f['severity']] >= threshold for f in FINDINGS):
return 1
return 0
if __name__ == '__main__':
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink