#!/usr/bin/env python3
# -*- coding: utf-8; py-indent-offset: 4 -*-
#
# Author:  Linuxfabrik GmbH, Zurich, Switzerland
# Contact: info (at) linuxfabrik (dot) ch
#          https://www.linuxfabrik.ch/
# License: The Unlicense, see LICENSE file.

# https://github.com/Linuxfabrik/monitoring-plugins/blob/main/CONTRIBUTING.md

"""See the check's README for more details."""

import argparse
import datetime
import os
import re
import stat
import sys

import lib.args
import lib.base
import lib.cache
import lib.db_mysql
import lib.db_sqlite
import lib.disk
import lib.human
import lib.logmatch
import lib.logsource
import lib.net
import lib.time
import lib.txt
from lib.globals import STATE_CRIT, STATE_OK, STATE_UNKNOWN, STATE_WARN

__author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
__version__ = '2026083001'

DESCRIPTION = """Scans the MySQL/MariaDB error log for errors, warnings, startups and
shutdowns. On MySQL 8.0.22+ the plugin prefers the `performance_schema.error_log` table
(reachable over the network, no shell access to the log file needed). Otherwise it
reads the on-disk log file, or fetches recent log lines from a container
(`docker:`/`podman:`/`kubectl:`) or systemd unit (`systemd:`).
The on-disk file path is taken from MySQL/MariaDB's `log_error` variable, with
common fallback locations probed when that variable is empty, and the journal of the
database unit is read along with it, because a server that fails to start writes why to
its standard error and never reaches the error log. What both hold is counted once. The
discovered path is cached so the check still works briefly when the database is down. The
most recent rotated file is read along with the live one, so the window does not end
where logrotate last ran.
Severity is detected from the bracketed log tags (`[ERROR]`, `[Warning]`), which
matches MySQL/MariaDB output and avoids false positives on lines that merely
mention "error" or "warning". Two of those lines are counted by how often they
arrive within `--lookback` instead, per client host, because they are written in
ones and twos on a host where nothing is wrong and only say something in bulk: a
login the server turned away, and a connection a client dropped without saying
goodbye. Counting either by its level would leave the check permanently yellow on
an ordinary application server. Recommendations are grouped under a single block at
the end of the output.
Reading the on-disk log file usually requires root/sudo (typical mysql logs are
owned by `mysql:mysql` mode `0640`). The `performance_schema.error_log` path needs
SELECT on that table but no filesystem access.
Alerts on every error and warning the log carries, when the logins a single source had
turned away or the connections a single client dropped cross their rates within
`--lookback`, when a configured log is missing or unreadable, and when an on-disk log
has grown past 32 MiB."""

DEFAULT_CACHE_EXPIRE = 5 * 24 * 60  # in minutes (= 5 days)
DEFAULT_DEFAULTS_FILE = '/var/spool/icinga2/.my.cnf'
DEFAULT_DEFAULTS_GROUP = 'client'
DEFAULT_ABORTED_CONNECTIONS_CRITICAL = 200
DEFAULT_ABORTED_CONNECTIONS_WARNING = 20
DEFAULT_ACCESS_DENIED_CRITICAL = 60
DEFAULT_ACCESS_DENIED_WARNING = 6
DEFAULT_ICINGA_CALLBACK = False
DEFAULT_INSECURE = True
DEFAULT_NO_MATCH_SEVERITY = 'ok'
DEFAULT_LOOKBACK = 600  # seconds
DEFAULT_NO_PROXY = False
DEFAULT_PER_SOURCE = True
DEFAULT_HOSTNAME = '127.0.0.1'
DEFAULT_PORT = '3306'
DEFAULT_TIMEOUT = 3

LOGFILE_BIG_THRESHOLD = 32 * 1024 * 1024  # 32 MiB - matches mysqltuner's cutoff

# A login the server turned away: `Access denied for user 'root'@'198.51.100.7'
# (using password: YES)`. It is the one line in this log that says somebody is
# working on the server rather than that the server is unwell, so it is counted
# against a rate rather than reported one by one, and it is kept out of the
# error and warning counts below - MariaDB writes it as `[Warning]`, and without
# this a single mistyped password would leave the check yellow.
# The host it names is what the rate is counted per. It is whatever the server
# resolved the peer to, an address or a name, and it is taken as it stands.
# Verified against MariaDB 11.8 and MySQL 8.4.
ACCESS_DENIED_REGEX = re.compile(r"Access denied for user '[^']*'@'(?P<source>[^']*)'")

# A client that went away without saying goodbye. The server writes one of these
# for every connection that ends without `COM_QUIT`, which every application that
# lets its connections fall out of scope produces all day - a batch job, a PHP
# request, a monitoring check. It is a `[Warning]` on MariaDB, so counting it by
# its level leaves the check permanently yellow on a host where nothing is wrong;
# a burst of them, on the other hand, is a network or an application falling over.
ABORTED_CONNECTION_REGEX = re.compile(
    r"Aborted connection \d+ to db: '[^']*' user: '[^']*' host: '(?P<source>[^']*)'"
)

# What the two servers have to be told before they write that line at all.
# MariaDB writes it at `log_warnings` 2, which is its default. MySQL keeps it at
# the third verbosity level and its default is the second, so on MySQL the line
# is missing until somebody raises it. Measured on MariaDB 11.8 and MySQL 8.4.
MINIMUM_LOG_ERROR_VERBOSITY = 3
MINIMUM_LOG_WARNINGS = 2
ACK_RETENTION_DAYS = 30
MAXLINES = 30000  # maximum log lines to consider from the end of the source
# Rotated predecessors of the log file to read along with it. Without one the
# window would end at the last rotation, and an event from before it would stop
# being reported the moment logrotate runs.
ROTATED_FILES = 1

# The unit the distributions run the database server as. Only the fixed locations systemd
# reads unit files from are looked at, and only for the names the distributions
# use, so nothing a log or a configuration file holds can point the check at
# another unit.
UNIT_CANDIDATES = ('mariadb.service', 'mysqld.service', 'mysql.service')
UNIT_DIRECTORIES = (
    '/etc/systemd/system',
    '/usr/lib/systemd/system',
    '/lib/systemd/system',
)

# How long a path may be in the summary before it is abbreviated for display.
# The paths the distributions use fit within it untouched; a host serving a
# dozen sites names a dozen logs in one directory, and there the directory is
# the part that says nothing.
SOURCE_PATH_MAX_LEN = 32


def parse_args():
    """Parse command line arguments using argparse."""
    parser = argparse.ArgumentParser(
        description=DESCRIPTION,
        epilog=lib.args.epilog(__file__),
        formatter_class=lib.args.HelpFormatter,
    )

    parser.add_argument(
        '-V',
        '--version',
        action='version',
        version=f'%(prog)s: v{__version__} by {__author__}',
    )

    parser.add_argument(
        '--aborted-connections-critical',
        help='Number of connections the clients dropped within `--lookback` '
        'that returns CRITICAL. '
        'Counted per client host, so one application falling over reaches it '
        'while the same number spread over a fleet does not. '
        '0 turns the threshold off. '
        'Example: `--aborted-connections-critical=500`. '
        'Default: %(default)s',
        dest='ABORTED_CONNECTIONS_CRITICAL',
        type=int,
        default=DEFAULT_ABORTED_CONNECTIONS_CRITICAL,
    )

    parser.add_argument(
        '--aborted-connections-warning',
        help='Number of connections the clients dropped within `--lookback` '
        'that returns WARNING. '
        'Counted per client host, so one application falling over reaches it '
        'while the same number spread over a fleet does not. '
        '0 turns the threshold off. '
        'Example: `--aborted-connections-warning=50`. '
        'Default: %(default)s',
        dest='ABORTED_CONNECTIONS_WARNING',
        type=int,
        default=DEFAULT_ABORTED_CONNECTIONS_WARNING,
    )

    parser.add_argument(
        '--access-denied-critical',
        help='Number of logins the server turned away within `--lookback` that '
        'returns CRITICAL. '
        '0 turns the threshold off. '
        'Example: `--access-denied-critical=200`. '
        'Default: %(default)s',
        dest='ACCESS_DENIED_CRITICAL',
        type=int,
        default=DEFAULT_ACCESS_DENIED_CRITICAL,
    )

    parser.add_argument(
        '--access-denied-warning',
        help='Number of logins the server turned away within `--lookback` that '
        'returns WARNING. '
        'Counted per source address, so a run against one account from one host '
        'reaches it while the same number of typos across a fleet does not. '
        '0 turns the threshold off. '
        'Example: `--access-denied-warning=1`. '
        'Default: %(default)s',
        dest='ACCESS_DENIED_WARNING',
        type=int,
        default=DEFAULT_ACCESS_DENIED_WARNING,
    )

    parser.add_argument(
        '--always-ok',
        help=lib.args.help('--always-ok'),
        dest='ALWAYS_OK',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '--cache-expire',
        help=lib.args.help('--cache-expire') + ' Default: %(default)s',
        dest='CACHE_EXPIRE',
        type=int,
        default=DEFAULT_CACHE_EXPIRE,
    )

    parser.add_argument(
        '--defaults-file',
        help='MySQL/MariaDB cnf file to read user, host and password from. '
        'Example: `--defaults-file=/var/spool/icinga2/.my.cnf`. '
        'Default: %(default)s',
        dest='DEFAULTS_FILE',
        default=DEFAULT_DEFAULTS_FILE,
    )

    parser.add_argument(
        '--defaults-group',
        help=lib.args.help('--defaults-group') + ' Default: %(default)s',
        dest='DEFAULTS_GROUP',
        default=DEFAULT_DEFAULTS_GROUP,
    )

    parser.add_argument(
        '-H',
        '--hostname',
        help='MySQL/MariaDB hostname or IP address. Default: %(default)s',
        dest='HOSTNAME',
        default=DEFAULT_HOSTNAME,
    )

    parser.add_argument(
        '--icinga-callback',
        help=lib.args.help('--icinga-callback'),
        dest='ICINGA_CALLBACK',
        action='store_true',
        default=DEFAULT_ICINGA_CALLBACK,
    )

    parser.add_argument(
        '--icinga-password',
        help=lib.args.help('--icinga-password'),
        dest='ICINGA_PASSWORD',
    )

    parser.add_argument(
        '--icinga-service-name',
        help=lib.args.help('--icinga-service-name'),
        dest='ICINGA_SERVICE_NAME',
    )

    parser.add_argument(
        '--icinga-url',
        help=lib.args.help('--icinga-url'),
        dest='ICINGA_URL',
    )

    parser.add_argument(
        '--icinga-username',
        help=lib.args.help('--icinga-username'),
        dest='ICINGA_USERNAME',
    )

    parser.add_argument(
        '--ignore',
        help='Ignore a log line matching this Python regular expression. '
        'The log line is lowercased before matching, so write the pattern in '
        'lowercase (or use the `(?i)` flag). '
        'Can be specified multiple times. '
        "Example: `--ignore='(?i)linuxfabrik'`.",
        action='append',
        default=None,
        dest='IGNORE',
    )

    parser.add_argument(
        '--ignore-pattern',
        help=argparse.SUPPRESS,
        action='append',
        default=None,
        dest='IGNORE_PATTERN',
    )

    parser.add_argument(
        '--ignore-regex',
        help=argparse.SUPPRESS,
        action='append',
        default=None,
        dest='IGNORE_REGEX',
    )

    parser.add_argument(
        '--insecure',
        help='Applies to the connection to the monitoring server that `--icinga-callback` makes. '
        + lib.args.help('--insecure'),
        dest='INSECURE',
        action='store_true',
        default=DEFAULT_INSECURE,
    )

    parser.add_argument(
        '--lookback',
        help='Logins the server turned away are counted within this window '
        'rather than reported one by one. '
        + lib.args.help('--lookback')
        + ' Example: `--lookback=3600`. '
        'Default: %(default)s (seconds)',
        dest='LOOKBACK',
        type=int,
        default=DEFAULT_LOOKBACK,
    )

    parser.add_argument(
        '--match',
        help='Only consider a log line matching this Python regular expression. '
        'The log line is lowercased before matching, so write the pattern in '
        'lowercase (or use the `(?i)` flag). '
        'Can be specified multiple times. '
        + lib.args.MATCH_IGNORE_PRECEDENCE
        + " Example: `--match='innodb'`.",
        action='append',
        default=None,
        dest='MATCH',
    )

    parser.add_argument(
        '--no-insecure',
        help='Applies to the connection to the monitoring server that `--icinga-callback` makes. '
        + lib.args.help('--no-insecure'),
        dest='INSECURE',
        action='store_false',
        default=DEFAULT_INSECURE,
    )

    parser.add_argument(
        '--no-match-severity',
        help=lib.args.help('--no-match-severity') + ' Default: %(default)s',
        dest='NO_MATCH_SEVERITY',
        choices=['ok', 'warn', 'crit', 'unknown'],
        default=DEFAULT_NO_MATCH_SEVERITY,
    )

    parser.add_argument(
        '--no-per-source',
        help=lib.args.help('--no-per-source'),
        dest='PER_SOURCE',
        action='store_false',
        default=DEFAULT_PER_SOURCE,
    )

    parser.add_argument(
        '--no-perfdata',
        help=lib.args.help('--no-perfdata'),
        dest='NO_PERFDATA',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '--no-proxy',
        help='Applies to the connection to the monitoring server that `--icinga-callback` makes. '
        + lib.args.help('--no-proxy'),
        dest='NO_PROXY',
        action='store_true',
        default=DEFAULT_NO_PROXY,
    )

    parser.add_argument(
        '--per-source',
        help=lib.args.help('--per-source') + ' Default: %(default)s',
        dest='PER_SOURCE',
        action='store_true',
        default=DEFAULT_PER_SOURCE,
    )

    parser.add_argument(
        '--port',
        help='MySQL/MariaDB port number. Default: %(default)s',
        dest='PORT',
        type=int,
        default=DEFAULT_PORT,
    )

    parser.add_argument(
        '--proxy',
        help='Applies to the connection to the monitoring server that `--icinga-callback` makes. '
        + lib.args.help('--proxy'),
        dest='PROXY',
        default=None,
    )

    parser.add_argument(
        '--server-log',
        help='Log source to read from. '
        'Accepts a file path, `docker:CONTAINER`, `podman:CONTAINER`, '
        '`kubectl:CONTAINER` or `systemd:UNITNAME`. '
        'Can be specified multiple times, and everything named is then read as '
        'one window; a source named twice is read once. '
        'If omitted, the check first probes `performance_schema.error_log` '
        '(MySQL 8.0.22+) and then falls back to the file from `log_error`, read '
        'along with the journal of the database unit; what the two share is '
        'counted once.',
        action='append',
        default=None,
        dest='SERVER_LOG',
    )

    parser.add_argument(
        '--timeout',
        help=lib.args.help('--timeout') + ' Default: %(default)s (seconds)',
        dest='TIMEOUT',
        type=int,
        default=DEFAULT_TIMEOUT,
    )

    args, _ = parser.parse_known_args()
    return args


def get_vars(conn):
    # Do not implement `get_all_vars()`, just fetch the ones we need for this check.
    # Without the GLOBAL modifier, SHOW VARIABLES displays the values that are used for
    # the current connection to MariaDB.
    sql = """
        show global variables
        where variable_name like 'datadir'
            or variable_name like 'hostname'
            or variable_name like 'log_error'
            or variable_name like 'log_error_verbosity'
            or variable_name like 'log_warnings';
          """
    return lib.base.coe(lib.db_mysql.select(conn, sql))


def get_log_file_real_path(file, hostname, datadir):
    if file and lib.disk.file_exists(file, allow_empty=True):
        return file
    if lib.disk.file_exists(f'{hostname}.log', allow_empty=True):
        return f'{hostname}.log'
    if lib.disk.file_exists(f'{hostname}.err', allow_empty=True):
        return f'{hostname}.err'
    if lib.disk.file_exists(os.path.join(datadir, f'{hostname}.err'), allow_empty=True):
        return os.path.join(datadir, f'{hostname}.err')
    if lib.disk.file_exists(os.path.join(datadir, f'{hostname}.log'), allow_empty=True):
        return os.path.join(datadir, f'{hostname}.log')
    if lib.disk.file_exists(os.path.join(datadir, 'mysql_error.log'), allow_empty=True):
        return os.path.join(datadir, 'mysql_error.log')
    if lib.disk.file_exists('/var/log/mysql.log', allow_empty=True):
        return '/var/log/mysql.log'
    if lib.disk.file_exists('/var/log/mysqld.log', allow_empty=True):
        return '/var/log/mysqld.log'
    if lib.disk.file_exists(f'/var/log/mysql/{hostname}.err', allow_empty=True):
        return f'/var/log/mysql/{hostname}.err'
    if lib.disk.file_exists(f'/var/log/mysql/{hostname}.log', allow_empty=True):
        return f'/var/log/mysql/{hostname}.log'
    if lib.disk.file_exists('/var/log/mysql/mysql_error.log', allow_empty=True):
        return '/var/log/mysql/mysql_error.log'
    return file


def has_pfs_error_log(conn):
    """True if `performance_schema.error_log` exists and is visible to this user.
    information_schema.tables only lists tables the current user can access, so a
    non-empty result implies the user has at least some privilege on it.
    """
    sql = (
        'SELECT 1 FROM information_schema.tables'
        " WHERE table_schema = 'performance_schema'"
        " AND table_name = 'error_log' LIMIT 1;"
    )
    success, rows = lib.db_mysql.select(conn, sql)
    return success and bool(rows)


def read_pfs_error_log(conn):
    """Pull the most recent rows from `performance_schema.error_log` and rebuild
    text lines that look like ordinary MySQL/MariaDB log entries (so the
    downstream bracket-tag parser does not need its own PFS branch). Returns the
    joined log text on success or None on SQL failure (caller falls back to
    file mode).
    """
    sql = (
        'SELECT LOGGED, PRIO, ERROR_CODE, SUBSYSTEM, DATA'
        ' FROM performance_schema.error_log'
        f' ORDER BY LOGGED DESC LIMIT {MAXLINES};'
    )
    success, rows = lib.db_mysql.select(conn, sql)
    if not success:
        return None
    lines = []
    # Reverse to put oldest first, so "last: ..." in the summary points to the
    # most recent row.
    for row in reversed(rows or []):
        ts = row.get('LOGGED', '')
        prio = row.get('PRIO', '')
        err_code = row.get('ERROR_CODE', '')
        subsystem = row.get('SUBSYSTEM', '')
        data = row.get('DATA', '')
        lines.append(f'{ts} 0 [{prio}] [{err_code}] [{subsystem}] {data}')
    return '\n'.join(lines)


def get_source(log_line):
    """Return the host a line names as the peer, or None where it names none.

    The server writes whatever it resolved the peer to, so this is an address on
    one host and a name (`localhost`) on the next. An address is brought into one
    form, because the same client otherwise counts twice when the server writes
    it as `::ffff:198.51.100.7` here and as `198.51.100.7` there; a name is kept
    as it stands, which is all that can be done with it and enough to group by.
    """
    match = ACCESS_DENIED_REGEX.search(log_line) or ABORTED_CONNECTION_REGEX.search(
        log_line
    )
    if not match:
        return None
    source = match.group('source')
    return lib.net.normalize_address(source) or source or None


def get_units(config_root=''):
    """Return the the database server units this host has a unit file for.

    A unit file two names point at is one unit: the MariaDB package ships
    `mysqld.service` and `mysql.service` as symlinks to `mariadb.service`, and
    reading that journal three times would count every line three times.
    """
    found = []
    seen = set()
    for candidate in UNIT_CANDIDATES:
        for directory in UNIT_DIRECTORIES:
            for filename in lib.disk.glob(f'{config_root}{directory}/{candidate}'):
                if not lib.disk.file_exists(filename, allow_empty=True):
                    continue
                identity = os.path.realpath(filename)
                if identity in seen:
                    continue
                seen.add(identity)
                found.append(os.path.basename(filename))
    return found


def dedup_key(line):
    """Identify the event in a line, whichever transport delivered it.

    The error log and the journal of the unit are both read, and the server
    writes its own timestamp into either, so what follows the syslog prefix the
    journal adds is the line the file holds.
    """
    written_at = lib.logsource.timestamp(line)
    return (
        written_at.replace(microsecond=0) if written_at else None,
        lib.logsource.strip_syslog_prefix(line),
    )


def main():
    """The main function. This is where the magic happens."""

    # logic taken from mysqltuner.pl:log_file_recommendations(),
    # verified in sync with MySQLTuner

    # parse the command line
    try:
        args = parse_args()
    except SystemExit:
        sys.exit(STATE_UNKNOWN)

    if args.IGNORE is None:
        args.IGNORE = []
    if args.IGNORE_PATTERN is None:
        args.IGNORE_PATTERN = []
    if args.MATCH is None:
        args.MATCH = []
    # `--ignore-regex` is the former name of `--ignore` and means the same
    # thing, so its values simply join the list. `--ignore-pattern` stays a
    # plain substring filter of its own, because folding a substring into a
    # regex list would change what it matches as soon as it carries a
    # metacharacter.
    if args.IGNORE_REGEX:
        args.IGNORE += args.IGNORE_REGEX

    if args.ICINGA_CALLBACK and not all(
        (
            args.ICINGA_URL,
            args.ICINGA_PASSWORD,
            args.ICINGA_USERNAME,
            args.ICINGA_SERVICE_NAME,
        )
    ):
        lib.base.cu(
            '`--icinga-callback` requires `--icinga-url`, `--icinga-password`, '
            '`--icinga-username` and `--icinga-service-name`.'
        )

    # fetch data
    # A size is known for plain on-disk files only (so we can emit the 32 MiB
    # cutoff check and a perfdata series), keyed by the source it belongs to.
    # Where several files are read, the series is what they add up to.
    sizes = {}
    # Every source that was read, each with the rotated predecessors read along
    # with it, in the shape the summary spells them out in. The structured table
    # has neither a size nor a predecessor and stands here on its own.
    read_sources = []
    # What the read had to say about itself, and what it could not read at all.
    # The structured table has neither, which is why both start out empty rather
    # than being taken off the result of a read that did not happen.
    read_notice = ''
    read_failed = []
    read_duplicates = 0
    source_truncated = False
    logfile = None
    log_error = None
    server_logs = args.SERVER_LOG or []
    myvar = {}

    if not server_logs:
        mysql_connection = {
            'defaults_file': args.DEFAULTS_FILE,
            'defaults_group': args.DEFAULTS_GROUP,
            'timeout': args.TIMEOUT,
        }
        success, conn = lib.db_mysql.connect(mysql_connection)
        if success:
            lib.base.coe(lib.db_mysql.check_privileges(conn))
            myvar = lib.db_mysql.lod2dict(get_vars(conn))
            # Prefer the structured PFS table on MySQL 8.0.22+ where it exists
            # and is visible to this user. Avoids needing shell access to the
            # error log file (works against remote DBs too).
            if has_pfs_error_log(conn):
                pfs_text = read_pfs_error_log(conn)
                if pfs_text is not None:
                    logfile = pfs_text
                    log_error = 'performance_schema.error_log'
                    read_sources = [
                        {'label': 'performance_schema.error_log', 'rotated': []}
                    ]
            lib.db_mysql.close(conn)
        if log_error is None:
            # DB unreachable or PFS not available: use the file path from the
            # config, fall back to the cached value from a previous run.
            log_error = myvar.get('log_error')
            if not log_error:
                log_error = lib.cache.get(
                    f'{args.HOSTNAME}-{args.PORT}',
                    filename='linuxfabrik-monitoring-plugins-mysql-logfile.db',
                )
            if log_error:
                log_error = get_log_file_real_path(
                    log_error,
                    myvar.get('hostname', ''),
                    myvar.get('datadir', ''),
                )
        if logfile is None:
            # The unit alongside the file, not instead of it: a server that
            # failed to start wrote why to its standard error and never reached
            # the error log. What both carry is counted once. The structured
            # table above answers for itself and needs neither.
            #
            # And where `log_error` names no file at all, the unit is the whole
            # answer rather than a reason to give up: a server left at its
            # default writes to its standard error, which under systemd is the
            # journal, and in a container it is all there is.
            server_logs = [log_error] if log_error else []
            server_logs += [f'systemd:{unit}' for unit in get_units()]
        if not server_logs and logfile is None:
            lib.base.cu(
                'No log file set and no database unit on this host (set '
                '`log_error` in the MySQL/MariaDB configuration, or use the '
                "check's `--server-log` parameter)."
            )

    # Cache the discovered on-disk path so the next run still has it when the
    # DB is unreachable. PFS does not need caching (only works with a live conn).
    if log_error and log_error != 'performance_schema.error_log':
        lib.cache.set(
            f'{args.HOSTNAME}-{args.PORT}',
            log_error,
            lib.time.now() + args.CACHE_EXPIRE * 60,
            filename='linuxfabrik-monitoring-plugins-mysql-logfile.db',
        )

    # read log content (unless PFS already provided it)
    if logfile is None:
        if 'stderr' in server_logs:
            lib.base.cu("log_error is set to STDERR, so this check can't read stderr.")
        resolved = []
        for server_log in server_logs:
            kind, _, target = lib.base.coe(lib.logsource.parse(server_log))
            if kind != lib.logsource.KIND_FILE:
                resolved.append(server_log)
                continue
            if not os.path.isabs(target):
                server_log = os.path.join(myvar.get('datadir', ''), target)
            resolved.append(server_log)
            # Report a path that is not a regular file before reading it, so the
            # admin hears about a directory or a device rather than about an I/O
            # error further down. With one source that is the whole answer; with
            # several it is one source of many and the run goes on.
            log_stat = lib.disk.stat(server_log)
            if log_stat is None or not stat.S_ISREG(log_stat.st_mode):
                if len(server_logs) == 1:
                    lib.base.oao(
                        f'Logging seems to be configured, but `{server_log}`'
                        f' does not seem to be an existing regular file. Check'
                        f' the path and file permissions, or provide the'
                        f' `--server-log` parameter.',
                        STATE_WARN,
                    )
                continue
            sizes[server_log] = log_stat.st_size
        server_logs = resolved
        if len(server_logs) == 1 and sizes.get(server_logs[0]) == 0:
            # An empty log file is a deterministic "no errors / warnings
            # observed" state, not an unknown one - typical right after
            # logrotate fires. Both the auto-detected and the explicit
            # --server-log code paths land here.
            lib.base.oao(
                f'Log file `{server_logs[0]}` is empty. Assuming log-rotation.',
                STATE_OK,
            )
        # The plugin runs as root via sudo, so restrict an on-disk log to the
        # system log directory and the default MySQL data directory. The datadir
        # reported by the server is deliberately NOT trusted here, because an
        # attacker controls which server the check connects to
        # (`--defaults-file`/`--hostname`) and could report `datadir=/etc`. The
        # plugin's own unit-test/ directory is allowed so the fixture tests
        # work; it does not exist next to the flat, root-owned plugin in
        # production. Bind-mount a custom data/log directory under /var/log to
        # include it (see the README).
        allowed_roots = [
            '/var/log',
            '/var/lib/mysql',
            os.path.join(os.path.dirname(os.path.realpath(sys.argv[0])), 'unit-test'),
        ]
        # No position is kept on purpose: this check reports on a window of the
        # log, counting startups and shutdowns next to errors and warnings, so
        # every run has to see the whole window rather than only what is new.
        result = lib.base.coe(
            lib.logsource.read_many(
                server_logs,
                allowed_roots=allowed_roots,
                dedup_key=dedup_key,
                max_lines=MAXLINES,
                rotated=ROTATED_FILES,
                timeout=args.TIMEOUT,
            )
        )
        logfile = '\n'.join(result['lines'])
        read_sources = result['sources']
        read_notice = result['notice']
        read_failed = result['failed']
        read_duplicates = result['duplicates']
        source_truncated = result['truncated']
    # What to call the whole of what was read, for the messages that have to name
    # it before the summary below spells every source out.

    # init some vars
    state = STATE_OK
    sections = []
    facts = []
    # All recommendations from all WARN/CRIT paths land here and render once at
    # the end as a `Recommendations:\n* ...` bulleted block, regardless of
    # which combinations of paths fire.
    recommendations = []
    last_errs, last_warns, last_shutdowns, last_starts = [], [], [], []
    last_denied = []
    last_aborted = []
    aborted_state = STATE_OK
    rate_since = datetime.datetime.now() - datetime.timedelta(seconds=args.LOOKBACK)
    considered_cnt = 0
    suppressed_cnt = 0
    compiled_ignore = [
        lib.base.coe(item) for item in lib.txt.compile_regex(args.IGNORE, '--ignore')
    ]
    compiled_match = [
        lib.base.coe(item) for item in lib.txt.compile_regex(args.MATCH, '--match')
    ]

    # Persisted acknowledgements are only needed when the callback is in use.
    # Each unique combination of filters gets its own state database, so two
    # services watching the same error log for different things do not share
    # what has been acknowledged. Only the filters actually set take part in the
    # identifier, so adding a filter later does not orphan the state of a
    # service that does not use it.
    acked_fingerprints = set()
    ack_conn = None
    if args.ICINGA_CALLBACK:
        instance_payload = {'hostname': args.HOSTNAME, 'port': args.PORT}
        for name, value in (
            ('ignore', args.IGNORE),
            ('ignore_pattern', args.IGNORE_PATTERN),
            ('match', args.MATCH),
        ):
            if value:
                instance_payload[name] = value
        ack_conn = lib.base.coe(
            lib.logmatch.connect(
                'mysql-logfile', lib.logmatch.instance_id(instance_payload)
            )
        )
        # Drop acknowledgements older than the retention: by that age the line
        # has left the window this check reads and can no longer re-appear.
        lib.base.coe(lib.logmatch.prune(ack_conn, retention=ACK_RETENTION_DAYS))
        acked_fingerprints = lib.base.coe(lib.logmatch.suppressed(ack_conn))

    # analyze data
    for log_line in logfile.splitlines():
        haystack = log_line.lower()
        if any(pattern.lower() in haystack for pattern in args.IGNORE_PATTERN) or any(
            item.search(haystack) for item in compiled_ignore
        ):
            continue
        # `--match` (include) is applied first, then `--ignore` (exclude), so a
        # line hit by `--ignore` is dropped even if it also matches `--match`.
        if compiled_match and not any(item.search(haystack) for item in compiled_match):
            continue
        considered_cnt += 1
        # Drop a line an operator has already taken on. The whole log line is
        # keyed, timestamp included, so the very same message logged again later
        # is a new event and alerts again.
        if acked_fingerprints and lib.logmatch.key(log_line) in acked_fingerprints:
            suppressed_cnt += 1
            continue
        # A login the server turned away is judged by how often it arrives and
        # is therefore counted here and nowhere else. Counting it in the level
        # counts as well would leave the check yellow for a single mistyped
        # password, because MariaDB writes the line as `[Warning]`.
        if ACCESS_DENIED_REGEX.search(log_line):
            last_denied.append(log_line)
            continue
        # The same for a connection the client dropped: on MariaDB it is a
        # `[Warning]` too, and an application that never closes cleanly writes
        # hundreds of them a day without anything being wrong.
        if ABORTED_CONNECTION_REGEX.search(log_line):
            last_aborted.append(log_line)
            continue
        # MySQL/MariaDB tag the severity in brackets like `[ERROR]` / `[Warning]`.
        # Match the bracketed form so words like "errors" inside table names or
        # paths do not falsely trip the count (mysqltuner uses the same regex).
        if '[error]' in haystack:
            last_errs.append(log_line)
        if '[warning]' in haystack:
            last_warns.append(log_line)
        if 'shutdown complete' in haystack and 'innodb' not in haystack:
            last_shutdowns.append(log_line)
        if 'ready for connections' in log_line:
            last_starts.append(log_line)

    # Everything collected is put back into the order it was written in. The
    # sources are read one after the other, so a line from the file sits before a
    # newer one from the journal however late it was written, and "the last one"
    # would name something days old.
    last_errs = lib.logsource.sort_by_time(last_errs)
    last_warns = lib.logsource.sort_by_time(last_warns)
    last_denied = lib.logsource.sort_by_time(last_denied)
    last_aborted = lib.logsource.sort_by_time(last_aborted)
    last_starts = lib.logsource.sort_by_time(last_starts)
    last_shutdowns = lib.logsource.sort_by_time(last_shutdowns)

    log_lines = logfile.splitlines()

    # `--match` narrowed the run down to nothing, so this run looked at no line
    # at all rather than at a quiet error log. Only reachable when the operator
    # set a filter, so a log that simply has nothing to report stays OK.
    if compiled_match and log_lines and not considered_cnt:
        if ack_conn is not None:
            lib.db_sqlite.close(ack_conn)
        lib.base.oao(
            f'Nothing checked: `--match` dropped all {len(log_lines)} '
            f'{lib.txt.pluralize("line", len(log_lines))} of {log_error}.',
            lib.base.str2state(args.NO_MATCH_SEVERITY),
            always_ok=args.ALWAYS_OK,
        )

    # build the message

    # What was read, and where from. The size stands next to the live file
    # rather than next to the line count, because it describes that one file and
    # nothing else - on a log rotated an hour ago a small file said to hold tens
    # of thousands of lines reads like a miscount, when almost all of them came
    # from the rotated predecessor. Naming the predecessor says where they did
    # come from. The count matters because everything below is counted within
    # it: a run that reports no startup at all is telling the truth about a
    # handful of lines rather than about the day. And where the window stopped
    # at the cap this check reads rather than at the start of the log, it says
    # so.
    # Which stretch of time those lines cover, because the count alone does not
    # say whether the window is an hour or a week: on a busy host the cap is
    # reached within hours, on a quiet one the same 30000 lines reach back
    # months, and every count below has to be read against that. Taken from the
    # ends of each source rather than from the ends of everything read: the
    # sources arrive one after the other, so the newest line of the first one
    # sits in the middle. A source that stamps no line simply leaves it out.
    covered = ''
    window_from, window_to = lib.logsource.covered_window(
        [item['lines'] for item in read_sources]
    )
    if window_from is not None and window_to is not None:
        span = int((window_to - window_from).total_seconds())
        covered = (
            f'{window_from:%Y-%m-%d %H:%M} .. {window_to:%Y-%m-%d %H:%M}'
            f' ({lib.human.seconds2human(span)}): '
        )
    line_cnt = len(log_lines)
    lines_read = (
        f'{lib.human.number2human(line_cnt)} {lib.txt.pluralize("line", line_cnt)}'
    )
    if source_truncated:
        lines_read = f'the most recent {lines_read}'
    threshold_human = lib.human.bytes2human(LOGFILE_BIG_THRESHOLD)
    oversized = any(size >= LOGFILE_BIG_THRESHOLD for size in sizes.values())
    # Every source by name, with its own size and its own rotated predecessor,
    # because a run that read three files and found nothing in two of them has
    # said something different from a run that read one.
    # The sources get a section of their own, the way the lines behind every
    # count do: naming them in the summary would bury the verdict under a
    # paragraph of paths, and the summary line is what a monitoring server shows
    # in a list. The paths are not abbreviated there: a bullet has the room, and
    # that is where an administrator copies them from.
    described = [
        lib.logsource.describe(
            item, sizes.get(item['label']), size_threshold=LOGFILE_BIG_THRESHOLD
        )
        for item in read_sources
    ]
    sources = f'{len(described)} {lib.txt.pluralize("source", len(described))}'

    if read_duplicates:
        # Named rather than silently dropped: an administrator who sees two
        # sources read and one count has to be able to tell that the check knows
        # they hold the same events.
        shared = read_duplicates
        source_fact_tail = (
            f', {lib.human.number2human(shared)} '
            f'{lib.txt.pluralize("line", shared)} they share counted once'
        )
    else:
        source_fact_tail = ''
    source_heading = f'Read {lines_read} from {sources}{source_fact_tail}:'
    if oversized:
        # The size is a fact about the host, not about this run, so it belongs in
        # the summary even though the sizes themselves are listed further down.
        state = lib.base.get_worst(state, STATE_WARN)
        facts.append(
            f'A log file is larger than {threshold_human}'
            f'{lib.base.state2str(STATE_WARN, prefix=" ")}'
        )
        recommendations.append(
            f'Log file is > {threshold_human}; analyse why or set up log'
            f' rotation (e.g. logrotate)'
        )

    # The lines behind these two counts are listed further down, so there is no
    # recommendation to make here: "check the errors" is what the listing is for.
    # Only the named situations below say what to do about them.
    n_err = len(last_errs)
    if n_err:
        state = lib.base.get_worst(state, STATE_CRIT)
        facts.append(
            f'{n_err} error {lib.txt.pluralize("line", n_err)} found'
            f'{lib.base.state2str(STATE_CRIT, prefix=" ")}'
            f' (last: {last_errs[-1]})'
        )

    n_warn = len(last_warns)
    if n_warn:
        state = lib.base.get_worst(state, STATE_WARN)
        facts.append(
            f'{n_warn} warning {lib.txt.pluralize("line", n_warn)} found'
            f'{lib.base.state2str(STATE_WARN, prefix=" ")}'
            f' (last: {last_warns[-1]})'
        )

    if not n_err and not n_warn:
        # Naming both levels one by one would bury the one sentence a healthy
        # host is supposed to be.
        facts.append('No errors or warnings found')

    # What the server turned away, counted per source: a run against one account
    # from one host reaches the threshold while the same number of typos across a
    # fleet does not. This is the one line in the log that is about somebody
    # working on the server rather than about the server itself, which is why it
    # is judged by a rate and not by its level.
    denied_window = lib.logsource.count_within(
        last_denied,
        rate_since,
        key=get_source if args.PER_SOURCE else None,
    )
    if last_denied:
        denied_state = lib.base.get_state(
            denied_window['count'],
            args.ACCESS_DENIED_WARNING or None,
            args.ACCESS_DENIED_CRITICAL or None,
        )
        state = lib.base.get_worst(state, denied_state)
        fact = (
            f'{denied_window["count"]} denied '
            f'{lib.txt.pluralize("login", denied_window["count"])}'
        )
        if denied_window['busiest']:
            fact += f' from {denied_window["busiest"]}'
        fact += (
            f' in the last {lib.human.seconds2human(args.LOOKBACK)}'
            f'{lib.base.state2str(denied_state, prefix=" ")}'
        )
        extras = []
        if denied_window['total'] != denied_window['count']:
            extras.append(
                f'{denied_window["total"]} in total from {denied_window["sources"]}'
                f' {lib.txt.pluralize("address", denied_window["sources"], "es")}'
            )
        if len(last_denied) != denied_window['total']:
            extras.append(f'{len(last_denied)} in the window read')
        if extras:
            fact += ' (' + ', '.join(extras) + ')'
        facts.append(fact)
        if denied_state != STATE_OK:
            recommendations.append(
                'Logins are being turned away in bulk, which is what a guessing'
                ' run against a database account looks like; the accounts and'
                ' hosts in the log say whether that is it or an application'
                ' still using credentials that were changed'
            )

    # What the clients dropped, counted the same way and for the same reason: a
    # connection that ends without `COM_QUIT` is normal in ones and twos and says
    # an application or a network is falling over when it arrives in bulk.
    aborted_window = lib.logsource.count_within(
        last_aborted,
        rate_since,
        key=get_source if args.PER_SOURCE else None,
    )
    if last_aborted:
        aborted_state = lib.base.get_state(
            aborted_window['count'],
            args.ABORTED_CONNECTIONS_WARNING or None,
            args.ABORTED_CONNECTIONS_CRITICAL or None,
        )
        state = lib.base.get_worst(state, aborted_state)
        fact = (
            f'{aborted_window["count"]} aborted '
            f'{lib.txt.pluralize("connection", aborted_window["count"])}'
        )
        if aborted_window['busiest']:
            fact += f' from {aborted_window["busiest"]}'
        fact += (
            f' in the last {lib.human.seconds2human(args.LOOKBACK)}'
            f'{lib.base.state2str(aborted_state, prefix=" ")}'
        )
        extras = []
        if aborted_window['total'] != aborted_window['count']:
            extras.append(
                f'{aborted_window["total"]} in total from '
                f'{aborted_window["sources"]}'
                f' {lib.txt.pluralize("host", aborted_window["sources"])}'
            )
        if len(last_aborted) != aborted_window['total']:
            extras.append(f'{len(last_aborted)} in the window read')
        if extras:
            fact += ' (' + ', '.join(extras) + ')'
        facts.append(fact)
        if aborted_state != STATE_OK:
            recommendations.append(
                'Connections are being dropped in bulk; the client hosts in the'
                ' log say whether one application is falling over, a network is'
                ' losing packets, or a connector needs its timeouts and'
                ' `max_allowed_packet` looked at'
            )

    # A counter at zero says nothing the performance data does not, and a
    # sentence about it would only push the verdict further right.
    n_start = len(last_starts)
    if n_start:
        facts.append(
            f'{n_start} {lib.txt.pluralize("startup", n_start)} detected'
            f' (last: {last_starts[-1]})'
        )

    n_shut = len(last_shutdowns)
    if n_shut:
        facts.append(
            f'{n_shut} {lib.txt.pluralize("shutdown", n_shut)} detected'
            f' (last: {last_shutdowns[-1]})'
        )

    # A server told to keep quiet writes no denied login at all, so a report
    # without any says nothing about the host. It goes first, ahead of the
    # verdict, because it decides what the verdict is worth, and it raises no
    # state of its own: the setting is a decision somebody took, and the check
    # only says what follows from it. Only reachable where the check reached the
    # server and could ask it.
    verbosity = myvar.get('log_error_verbosity')
    warnings_level = myvar.get('log_warnings')
    if verbosity is not None and int(verbosity) < MINIMUM_LOG_ERROR_VERBOSITY:
        facts.insert(
            0,
            f'The server runs at `log_error_verbosity {verbosity}` and writes a'
            f' denied login only from {MINIMUM_LOG_ERROR_VERBOSITY} on, so it'
            f' logs none of them',
        )
        recommendations.append(
            f'Set `log_error_verbosity = {MINIMUM_LOG_ERROR_VERBOSITY}` to have'
            f' the server log the logins it turns away'
        )
    elif warnings_level is not None and int(warnings_level) < MINIMUM_LOG_WARNINGS:
        facts.insert(
            0,
            f'The server runs at `log_warnings {warnings_level}` and writes a'
            f' denied login only from {MINIMUM_LOG_WARNINGS} on, so it logs none'
            f' of them',
        )
        recommendations.append(
            f'Set `log_warnings = {MINIMUM_LOG_WARNINGS}` to have the server log'
            f' the logins it turns away'
        )

    if read_notice:
        facts.append(read_notice.rstrip('.'))

    # A source that could not be read at all leaves a hole in the window: with a
    # second source still delivering, the run goes on, but reporting the state of
    # what happened to work would hide that nothing is watching the rest.
    if read_failed:
        state = lib.base.get_worst(state, STATE_WARN)
        facts.append(
            '; '.join(item.rstrip('.') for item in read_failed)
            + lib.base.state2str(STATE_WARN, prefix=' ')
        )
        recommendations.append(
            'Check the log sources this check was told to read; '
            'one of them could not be read at all'
        )

    # The window the summary reports on comes first, because every count in it
    # is a count within that window and reads differently against an hour than
    # against a week.
    sections.append(covered + '. '.join(facts) + '.')

    # The lines behind the counts, under the headings the rest of the family
    # uses. A startup and a shutdown get none: the fact above names the last of
    # each, and listing them again is a list of things that went right.
    if last_errs:
        sections.append(
            'Error lines:\n'
            + '\n'.join(f'* {line}' for line in lib.txt.shorten_list(last_errs))
        )
    if last_warns:
        sections.append(
            'Warning lines:\n'
            + '\n'.join(f'* {line}' for line in lib.txt.shorten_list(last_warns))
        )
    if last_denied:
        sections.append(
            'Denied logins:\n'
            + '\n'.join(f'* {line}' for line in lib.txt.shorten_list(last_denied))
        )
    if last_aborted and aborted_state != STATE_OK:
        # Only once the rate says something: a handful of them is what every
        # application that lets a connection fall out of scope produces, and
        # listing ten of those under a green check is noise.
        sections.append(
            'Aborted connections:\n'
            + '\n'.join(f'* {line}' for line in lib.txt.shorten_list(last_aborted))
        )

    # What was read, last of the sections: the verdict and the lines behind it
    # come first, the provenance after them.
    sections.append(
        source_heading + '\n' + '\n'.join(f'* {item}' for item in described)
    )

    if recommendations:
        sections.append(
            'Recommendations:\n' + '\n'.join(f'* {r}' for r in recommendations)
        )

    msg = '\n\n'.join(sections)

    # perfdata
    perfdata = ''
    if sizes:
        perfdata += lib.base.get_perfdata(
            'mysql_logfile_size',
            sum(sizes.values()),
            uom='B',
            warn=LOGFILE_BIG_THRESHOLD,
            _min=0,
        )
    # The counters of offending lines alert from their first hit on, hence
    # `warn='0'` ("outside 0..0") rather than `warn=1`. The startup and shutdown
    # counters carry none, because neither is a problem in itself.
    perfdata += lib.base.get_perfdata(
        'mysql_error_lines', n_err, uom=None, warn='0', _min=0
    )
    perfdata += lib.base.get_perfdata(
        'mysql_warning_lines', n_warn, uom=None, warn='0', _min=0
    )
    # The rate counter trends what the state follows: how many the busiest single
    # source produced within the window, not how many the log holds.
    perfdata += lib.base.get_perfdata(
        'mysql_access_denied',
        denied_window['count'],
        uom=None,
        warn=args.ACCESS_DENIED_WARNING or None,
        crit=args.ACCESS_DENIED_CRITICAL or None,
        _min=0,
    )
    perfdata += lib.base.get_perfdata(
        'mysql_aborted_connections',
        aborted_window['count'],
        uom=None,
        warn=args.ABORTED_CONNECTIONS_WARNING or None,
        crit=args.ABORTED_CONNECTIONS_CRITICAL or None,
        _min=0,
    )
    perfdata += lib.base.get_perfdata('mysql_startups', n_start, uom=None, _min=0)
    perfdata += lib.base.get_perfdata('mysql_shutdowns', n_shut, uom=None, _min=0)

    # Ask the monitoring server about the acknowledgement. Where the service is
    # acknowledged, persist the lines currently being reported so an unchanged
    # error log does not raise them again on the following runs.
    if args.ICINGA_CALLBACK and state != STATE_OK:
        acknowledged, note = lib.base.coe(
            lib.logmatch.service_acknowledged(
                args.ICINGA_URL,
                args.ICINGA_USERNAME,
                args.ICINGA_PASSWORD,
                args.ICINGA_SERVICE_NAME,
                insecure=args.INSECURE,
                no_proxy=args.NO_PROXY,
                proxy=args.PROXY,
                timeout=args.TIMEOUT,
            )
        )
        if acknowledged:
            lib.base.coe(
                lib.logmatch.acknowledge(
                    ack_conn,
                    [
                        {'key': lib.logmatch.key(line), 'line': line}
                        for line in last_errs + last_warns
                    ],
                )
            )
            state = STATE_OK
        if note:
            msg += f'\n\n{note}'
    if suppressed_cnt:
        msg += (
            f'\n\n{suppressed_cnt} acknowledged'
            f' {lib.txt.pluralize("line", suppressed_cnt)} suppressed.'
        )
    if ack_conn is not None:
        lib.db_sqlite.close(ack_conn)

    # over and out
    lib.base.oao(
        msg, state, perfdata, always_ok=args.ALWAYS_OK, no_perfdata=args.NO_PERFDATA
    )


if __name__ == '__main__':
    try:
        main()
    except Exception:
        lib.base.cu()
