icinga2-checks/check_mysql_slave.py

#!/usr/bin/env python
import argparse
import sys
import traceback

import mysql.connector

from checker import nagios, print_icinga2_check_status
from checker.types import try_int

"""
## Usage example ##

Target delay = 300 seconds
Warning deviation = 10%
Critical deviation = 15%

This will set the warning levels to 330 and 270 seconds, if the delay is greater or less than these values it will return WARNING.
The critical levels will be 345 and 255 seconds. 
"""


def main(args):
    cursor = cnx = None
    try:
        cnx = mysql.connector.connect(user=args.username, password=args.password, host=args.host)
        cursor = cnx.cursor(dictionary=True)
        cursor.execute("SHOW SLAVE STATUS")
        slave_status = cursor.fetchone()

        if slave_status is None:
            print("UNKNOWN - Could not retrieve slave status")
            sys.exit(3)

        slave_io_running = slave_status['Slave_IO_Running']
        slave_sql_running = slave_status['Slave_SQL_Running']
        replication_delay = try_int(slave_status['Seconds_Behind_Master'])
        last_io_error = slave_status['Last_IO_Error']
        last_sql_error = slave_status['Last_SQL_Error']

        exit_code = nagios.STATE_OK
        exit_msg = []

        if slave_sql_running != 'Yes':
            exit_code = nagios.STATE_CRIT
            exit_msg.append('Slave SQL is not running!')
        if slave_io_running != 'Yes':
            exit_code = nagios.STATE_CRIT
            exit_msg.append('Slave IO is not running!')
        if last_sql_error:
            exit_code = nagios.STATE_CRIT
            exit_msg.append(f'Last SQL Error: {last_sql_error.strip(".")}.')
        if last_io_error:
            exit_code = nagios.STATE_CRIT
            exit_msg.append(f'Last IO Error: {last_io_error.strip(".")}.')

        if args.target_delay:
            warn_deviation_max = args.target_delay * (1 + (args.warning_deviation / 100))
            warn_deviation_min = args.target_delay * (1 - (args.warning_deviation / 100))
            crit_deviation_max = args.target_delay * (1 + (args.critical_deviation / 100))
            crit_deviation_min = args.target_delay * (1 - (args.critical_deviation / 100))
            if replication_delay <= crit_deviation_min:
                exit_code = nagios.STATE_CRIT
                exit_msg.append('Replication is delayed!')
            if replication_delay >= crit_deviation_max:
                exit_code = nagios.STATE_CRIT
                exit_msg.append('Replication is ahead???')
            if replication_delay <= warn_deviation_min:
                exit_code = nagios.STATE_WARN
                exit_msg.append('Replication is delayed!')
            if replication_delay >= warn_deviation_max:
                exit_code = nagios.STATE_WARN
                exit_msg.append('Replication is ahead???')

        if exit_code == nagios.STATE_OK:
            exit_msg.append('Slave is healthy! Slave SQL: running. Slave IO: running.')

        exit_msg.append(f'Slave is {replication_delay} seconds behind master.')

        perfdata = {
            "replication_delay": {
                "value": replication_delay,
                "min": 0,
                "unit": "s",
            },
        }

        text_result = ' '.join(exit_msg)
        print_icinga2_check_status(text_result, exit_code, perfdata)
        sys.exit(exit_code)

    except mysql.connector.Error as e:
        print("UNKNOWN - Could not connect to database!")
        print(e)
        sys.exit(nagios.STATE_UNKNOWN)
    except Exception as e:
        print(f"UNKNOWN - {e}")
        traceback.print_exc()
        sys.exit(nagios.STATE_UNKNOWN)

    finally:
        if cursor:
            cursor.close()
        if cnx:
            cnx.close()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Check a MySQL slave.')
    parser.add_argument('--host', required=True, help='The IP of the slave to connect to.')
    parser.add_argument('--username', required=True, help='Username.')
    parser.add_argument('--password', required=True, help='Password.')
    parser.add_argument('--target-delay', default=None, type=int, help='The target delay in seconds.')
    parser.add_argument('--warning-deviation', default=10, type=int, help='If the delay deviates more than this percentage from the target delay, return warning.')
    parser.add_argument('--critical-deviation', default=15, type=int, help='If the delay deviates more than this percentage from the target delay, return critical.')
    args = parser.parse_args()
    main(args)