Nagios RabbitMQ consumption check

check_rabbitmq_consumption.php

#!/bin/env php
<?php

array_shift($argv);

function printUsage() {
    echo "\n";
    echo 'Nagios consumer check for RabbitMQ on localhost'."\n";
    echo '-----------------------------------------------'."\n";
    echo "\n";
    echo 'Note that message age must be higher than check interval on Nagios!'."\n";
    echo "\n";
    echo "Options:\n";
    echo "  -h This help\n";
    echo "  -u Username to use for connecting to RabbitMQ (default:guest)\n";
    echo "  -p Password to use for connecting to RabbitMQ (default:guest)\n";
    echo "  -n Port number to use for connectin to RabbtiMQ (default:15672)\n";
    echo "  -q Queue name to monitor (REQUIRED)\n";
    echo "  -s Max age of unconsumed message in seconds before warning is raised (default:1800)\n";
    echo "  -S Max age of unconsumed message in seconds before error is raised (default:3600)\n";
// @todo maybe add this later on
//    echo "  -c Max number of messages in queue. A warning is raised if this value is exceeded (default:20)\n";
//    echo "  -C Max number of messages in queue. An error is raised if this value is exceeded (default:100)\n";
    echo "\n";
}


// Default values
$port = 15672;
$durationWarning = 1800;
$durationError = 3600;
$countWarning = 20;
$countError = 100;
$user = 'guest';
$pass = 'guest';
$queueName = '';
$dbFile = '/tmp/nrpe/nagios_check_rabbitmq';

if (!file_exists('/tmp/nrpe')) {
    mkdir('/tmp/nrpe');
}


$errors = array();

foreach ($argv as $key => $arg) {

    switch ($arg) {
        case '-h':
            printUsage();
            exit;
        case '-n':
            $port = $argv[$key+1];
            if (!preg_match('/^\d+$/', $port)) {
                $errors[] = 'Invalid portnumber given for -n';
            }
            break;
        case '-s':
            $durationWarning = $argv[$key+1];
            if (!preg_match('/^\d+$/', $durationWarning)) {
                $errors[] = 'Invalid number given for -s';
            }
            break;
        case '-S':
            $durationError = $argv[$key+1];
            if (!preg_match('/^\d+$/', $durationError)) {
                $errors[] = 'Invalid number given for -S';
            }
            break;
        case '-c':
            $countWarning = $argv[$key+1];
            if (!preg_match('/^\d+$/', $countWarning)) {
                $errors[] = 'Invalid number given for -c';
            }
            break;
        case '-C':
            $countError = $argv[$key+1];
            if (!preg_match('/^\d+$/', $countError)) {
                $errors[] = 'Invalid number given for -C';
            }
            break;
        case '-q':
            $queueName = $argv[$key+1];
            if (empty($queueName)) {
                $errors[] = 'No queuename given';
            }
            break;
        case '-u':
            $user = $argv[$key+1];
            if (empty($user)) {
                $errors[] = 'No user given';
            }
            break;
        case '-p':
            $pass = $argv[$key+1];
            if (empty($pass)) {
                $errors[] = 'No pass given';
            }
            break;
        default:
            // probably a value
    }
}


if (empty($queueName) && !in_array('No queuename given', $errors)) {
    $errors[] = 'No queuename given';
}

if (count($errors)) {
    echo "Some errors have been found \n\n";

    foreach ($errors as $error) {
        echo "  - " . $error . "\n";
    }

    printUsage();
    exit;
}


$ch = curl_init();
curl_setopt($ch, CURLOPT_URL, sprintf("http://localhost:%s/api/queues", $port));
curl_setopt($ch, CURLOPT_USERPWD, $user . ":" . $pass);
curl_setopt($ch, CURLOPT_VERBOSE, 0);
curl_setopt($ch, CURLOPT_HEADER, 0);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);

$data = curl_exec($ch);
$queues = json_decode($data, true);
$found = false;

foreach ($queues as $queue) {

    if ($queue['name'] != $queueName) {
        continue;
    }

    $found = true;

    if (intval($queue['messages']) === 0) {
        echo sprintf("OK - 0 messages in queue\n");
        exit(0);
        break;
    }

    if (!array_key_exists('message_stats', $queue) || !array_key_exists('ack', $queue['message_stats'])) {
        // warning no consumer has ever connected to this queue
        if ($queue['messages'] > 0) {
            echo sprintf('CRITICAL - No consumer has ever connected to queue %s with %s messages' . "\n", $queueName, $queue['messages']);
            exit(2);
        }
        echo sprintf('WARNING - No consumer has ever connected to queue %s' . "\n", $queueName);
        exit(1);
    }

    $queueCount = intval($queue['messages']);
    $ackCount = intval($queue['message_stats']['ack']);

    $db = new PDO(sprintf('sqlite:%s', $dbFile));
    $result = $db->query('CREATE TABLE IF NOT EXISTS monitor (time INTEGER, ackcount INTEGER)');

    if (!file_exists($dbFile)) {
        echo "ERROR - Couldnt create RabbitMq tracker (sqlite db)";
        exit(3);
    }

    $result = $db->query('SELECT * FROM monitor')->fetchAll();

    if (count($result) == 0) {
        // insert
        $db->exec(sprintf('INSERT INTO monitor VALUES (%s, %s)', time(), $ackCount));
        $result = $db->query('SELECT * FROM monitor')->fetchAll();
    }

    $monitorData = $result[0];
    $timeDiff = time() - $monitorData['time'];
    $oldAckCount = $monitorData['ackcount'];

    // There are some messages in queue, so check message age
    if ($oldAckCount == $ackCount) {

        // No messages have been processed during last check, but message still exists in queue, so check age compared to last ack increment time
        if ($timeDiff > $durationError) {
            echo sprintf("CRITICAL - No messages have been consumed during last %s seconds. %s message(s) in queue\n", $durationError, $queueCount);
            exit(2);
        }
        if ($timeDiff > $durationWarning) {
            echo sprintf("WARNING - No messages have been consumed during last %s seconds. %s message(s) in queue\n", $durationWarning, $queueCount);
            exit(1);
        }

        echo sprintf("OK - %s messages in queue\n", $queueCount);
        exit(0);

    } else {
        // update with current time
        $result = $db->exec(sprintf('UPDATE monitor SET time = %s AND ackcount = %s', time(), $ackCount));
        // Some messages have been processed, so consumer has consumed something meanwhile
        echo sprintf("OK - %s messages have been processed since last check. Current ackcount %s, %s message(s) in queue\n", $ackCount - $oldAckCount, $ackCount, $queueCount);
        exit(0);
    }

}

if (!$found) {
    echo sprintf("ERROR - queue named '%s' is not found on this instance\n", $queueName);
    exit(3);
}