zabbix-deploy/collector/l2db.py

# -*- coding: utf-8 -*-
"""Small DB bridge for the out-of-process L2 collectors.

The collectors run on the Docker host, outside the Zabbix PHP container.  We
deliberately do not keep a second set of MySQL credentials: the credentials
are read from the already-running Zabbix MySQL container's environment (the
same MYSQL_* values that the official Zabbix compose file passed to MySQL and
Zabbix).  Queries are executed inside that container, so the DB host never
needs to be resolvable from the host shell.
"""
import json
import os
import subprocess


def _container_name():
    return os.environ.get('L2_DB_CONTAINER') or os.environ.get('L2_MYSQL_CONTAINER') or 'zabbix-mysql'


def _container_env():
    """Return the real environment of the Zabbix MySQL container as a dict."""
    name = _container_name()
    out = subprocess.check_output(
        ['docker', 'inspect', name, '--format', '{{json .Config.Env}}'],
        text=True, timeout=30
    )
    env = {}
    for item in json.loads(out or '[]'):
        key, sep, value = str(item).partition('=')
        if sep:
            env[key] = value
    return env


def credentials():
    """Return (container, user, password, database) without hard-coded secrets."""
    env = _container_env()
    user = env.get('MYSQL_USER') or env.get('DB_SERVER_USER') or ''
    password = env.get('MYSQL_PASSWORD') or env.get('DB_SERVER_PASS') or ''
    database = env.get('MYSQL_DATABASE') or env.get('DB_SERVER_DBNAME') or 'zabbix'
    if not user or not password:
        raise RuntimeError('Zabbix MySQL container has no MYSQL_USER/MYSQL_PASSWORD')
    return _container_name(), user, password, database


def run_sql(sql, timeout=120):
    """Execute SQL inside the Zabbix MySQL container.

    Returns the completed process.  stdin is used instead of argv so long
    compressed history blobs never hit the command-line length limit.
    """
    container, user, password, database = credentials()
    cmd = [
        'docker', 'exec', '-i', container, 'mysql',
        '-N', '-B', '--default-character-set=utf8mb4',
        '-u' + user, '-p' + password, database
    ]
    return subprocess.run(cmd, input=sql.encode('utf-8'), capture_output=True, timeout=timeout)


def query(sql, timeout=120):
    """Run a SELECT and return rows as lists of strings (TSV, no header)."""
    r = run_sql(sql, timeout=timeout)
    if r.returncode != 0:
        raise RuntimeError((r.stderr or b'').decode('utf-8', 'ignore')[:500])
    text = (r.stdout or b'').decode('utf-8', 'ignore')
    return [line.split('\t') for line in text.splitlines() if line != '']


def scalar(sql, timeout=120):
    rows = query(sql, timeout=timeout)
    return rows[0][0] if rows and rows[0] else None