icinga2-checks/check_nginx.py

#!/usr/bin/env python

import argparse
import sys
import traceback

from checker import nagios, print_icinga2_check_status
from checker.http import fetch_with_retry
from checker.linuxfabric.base import get_state


def parse_nginx_status(status):
    try:
        lines = status.strip().split("\n")
        data = {}
        for line in lines:
            parts = line.split()
            if "Active" in line:
                data["active_connections"] = int(parts[2])
            elif "server accepts handled requests" in line:
                continue
            elif len(parts) == 3 and parts[1].isdigit():
                data["accepted"] = int(parts[0])
                data["handled"] = int(parts[1])
                data["requests"] = int(parts[2])
            elif "Reading" in line:
                data["reading"] = int(parts[1])
                data["writing"] = int(parts[3])
                data["waiting"] = int(parts[5])
        return data
    except Exception as e:
        print_icinga2_check_status(f'failed to parse status page: {e}', nagios.UNKNOWN)
        sys.exit(nagios.UNKNOWN)


def main():
    parser = argparse.ArgumentParser(description="Check Nginx status using stub_status.")
    parser.add_argument("--url", required=True, help="URL to Nginx stub_status.")
    parser.add_argument("--critical-active", type=int, default=None, help="Critical threshold for active connections. Default: 0 (disabled)")
    parser.add_argument("--warning-active", type=int, default=None, help="Warning threshold for active connections. Default: 0 (disabled)")
    parser.add_argument("--critical-waiting", type=int, default=None, help="Critical threshold for waiting connections. Default: 0 (disabled)")
    parser.add_argument("--warning-waiting", type=int, default=None, help="Warning threshold for waiting connections. Default: 0 (disabled)")
    args = parser.parse_args()

    status = fetch_with_retry(args.url).text
    data = parse_nginx_status(status)

    perfdata_dict = {
        "active_connections": {
            "value": data["active_connections"],
            "warn": args.warning_active,
            "crit": args.critical_active,
        },
        "waiting": {
            "value": data["waiting"],
            "warn": args.warning_waiting,
            "crit": args.critical_waiting,
        },
        "accepted": {"value": data["accepted"]},
        "handled": {"value": data["handled"]},
        "requests": {"value": data["requests"]},
        "reading": {"value": data["reading"]},
        "writing": {"value": data["writing"]},
    }

    return_code = nagios.STATE_OK
    if args.warning_active or args.critical_active:
        active_connections_state = get_state(
            data["active_connections"], args.warning_active, args.critical_active, "ge"
        )
    else:
        active_connections_state = nagios.STATE_OK
    return_code = max(active_connections_state, return_code)

    if args.warning_waiting or args.critical_waiting:
        waiting_state = get_state(data["waiting"], args.warning_waiting, args.critical_waiting, "ge")
    else:
        waiting_state = nagios.STATE_OK
    return_code = max(waiting_state, return_code)

    # if active_connections_state == nagios.STATE_CRIT or waiting_state == nagios.STATE_CRIT:
    #     return_code = nagios.CRITICAL
    # elif active_connections_state == nagios.STATE_WARN or waiting_state == nagios.STATE_WARN:
    #     return_code = nagios.WARNING
    # else:
    #     return_code = nagios.OK

    status_str = "Active connections: {active_connections}, Waiting: {waiting}, Accepted: {accepted}, Handled: {handled}, Requests: {requests}, Reading: {reading}, Writing: {writing}".format(**data)
    print_icinga2_check_status(status_str, return_code, perfdata_dict)
    sys.exit(return_code)


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print_icinga2_check_status(f'exception "{e}" \n {traceback.format_exc()}', nagios.UNKNOWN)
        sys.exit(nagios.UNKNOWN)