Broker Module

The Statusengine Broker Module is a small C++ library that gets loaded into your Naemon or Nagios Core. It grabs status information as it happens, encodes it as JSON and publishes it to a Gearman job server or to RabbitMQ.

Because a queue sits in between, the monitoring core never waits on a database or on disk I/O. Publishing is a local hand-off; everything after that is somebody else’s problem. The worker’s , specifically.

How it hooks into the core

The module is a NEB (Naemon Event Broker ↗ ) module, which gets loaded by the monitoring core. As a user, you only need to add the broker_module=/path/to/module.so line into your config.

extern "C" { NEB_API_VERSION(CURRENT_NEB_API_VERSION) }

extern "C" int nebmodule_init(int, char *args, nebmodule *handle) {
    return statusengine::Nebmodule::Instance().Init(handle, std::string(args));
}
extern "C" int nebmodule_deinit(int, int reason) {
    return statusengine::Nebmodule::Instance().Deinit(reason);
}
int nebmodule_callback(int event_type, void *data) {
    return statusengine::Nebmodule::Instance().Callback(event_type, data);
}

Supported monitoring cores

Either way you need the headers of a compiled core, so build and install Naemon or Nagios before building the broker.

Building from source

Dependencies

apt install git python3-pip gcc g++ cmake build-essential libglib2.0-dev \
  libgearman-dev uuid-dev libuchardet-dev pkg-config \
  libssl-dev librabbitmq-dev
pip3 install meson ninja

dnf install git python3-pip gcc gcc-c++ cmake3 pkgconfig librabbitmq-devel \
  libgearman-devel libuchardet-devel openssl-devel glib2-devel
pip3 install meson ninja

Get the sources

cd /tmp
git clone https://github.com/statusengine/broker
cd broker

Configure and build

export PKG_CONFIG_PATH=/opt/naemon/lib/pkgconfig/
meson setup --buildtype=release build
ninja -C build

meson setup -Dnagios=true -Dnagios_include_dir=/opt/nagios/include \
  --buildtype=release build
ninja -C build

Install

ninja -C build install

The default prefix is /usr/local, so the library lands at /usr/local/lib/libstatusengine.so. Pass --prefix=/opt/naemon to meson setup to install it next to your core instead.

Build options

Disabling a transport you do not use drops its client library from the dependency list, which is worth doing if you are packaging the module.

Loading the module

Add one line to naemon.cfg or nagios.cfg. On Naemon, you can also drop a file into module-conf.d/:

broker_module=/opt/naemon/lib/libstatusengine.so /path/to/statusengine.toml

Then restart the core. On success the module logs its startup with the prefix Statusengine:. If it cannot reach the configured queue, nebmodule_init() returns non-zero and the core refuses to start.

A configuration matched to the Go worker

This is a minimal Gearman configuration that publishes exactly the twelve queues the worker reads and nothing else:

[[Gearman]]
URL = "127.0.0.1:4730"

HostStatus                    = "statusngin_hoststatus"
ServiceStatus                 = "statusngin_servicestatus"
HostCheck                     = "statusngin_hostchecks"
ServiceCheck                  = "statusngin_servicechecks"
ServicePerfData               = "statusngin_service_perfdata"
StateChange                   = "statusngin_statechanges"
LogData                       = "statusngin_logentries"
NotificationData              = "statusngin_notifications"
ContactNotificationMethodData = "statusngin_contactnotificationmethod"
AcknowledgementData           = "statusngin_acknowledgements"
DowntimeData                  = "statusngin_downtimes"
RestartData                   = "statusngin_core_restart"

# Inbound: external commands the worker publishes are handed to the core here.
WorkerCommand = "statusngin_cmd"

[Bulk]
Queues = ["HostStatus", "HostCheck", "ServiceStatus", "ServiceCheck",
          "ServicePerfData", "StateChange", "LogData", "NotificationData"]

[Log]
Level = "Warning"

Publishing a queue the worker does not consume is not an error. The messages simply accumulate on the queue server until something drains them, which on a busy installation is a slow way to run out of memory.

A full configuration example for the broker module can be found here: statusengine.toml ↗

Queue identifier reference

An identifier is the left-hand side of a Identifier = "queue name" line inside a [[Gearman]] or [[Rabbitmq]] section. The name on the right is yours to choose; the names below are the ones the shipped statusengine.toml suggests, and the Go worker expects those exact names. An identifier you do not write down publishes nothing - see the callout above .

Outbound: core to queue

23 identifiers, in the order src/Queue.h declares them. The Go worker has a consumer for 12 of them. The twelve in the configuration above . The other 11 are not consumed by Statusengine itself.

statusngin_programmstatus is spelled with two m in the shipped configuration. It is only a suggestion like every other name here, but changing it means changing it on both sides.

Inbound: queue to core

Three identifiers work the other way round. The module consumes them inside the core’s event loop and applies what it finds, which is what makes the queue two-way — and what the [Worker] section’s time budget exists to bound.

Bulk messages

Any outbound identifier named in [Bulk] Queues is batched: instead of one message per event, the module collects events and publishes {"messages": [ … ], "format": "none"}. The setting is global. It applies to every [[Gearman]] and [[Rabbitmq]] connection and a batch is flushed when it reaches Maximum messages (200) or after FlushInterval seconds (10), whichever comes first.

Bulking is what keeps a busy core from producing one queue job per check. It costs a little latency, bounded by FlushInterval, and the shipped default covers the queues where the volume actually is:

[Bulk]
Queues = ["HostStatus", "HostCheck", "ServiceStatus", "ServiceCheck",
          "ServicePerfData", "StateChange", "LogData", "NotificationData",
          "OCHP", "OCSP"]

OCSP and OCHP

Naemon and Nagios can run a command after every check, called ocsp_command and ochp_command, the obsessive compulsive service and host processors. They work, and they fork a process for every single check result, which is the one thing a busy core cannot afford.

The OCSP and OCHP queues replace that with a copy on a queue. Name them in any [[Gearman]] or [[Rabbitmq]] section and every processed check is published a second time:

OCSP = "statusngin_ocsp"
OCHP = "statusngin_ochp"

Nothing is forked, nothing runs inside the core’s event loop, and whatever reads the queue can be on a different machine. The payload is the same object ServiceCheck and HostCheck publish. The separate queue exists so a second consumer can have its own copy, at its own pace, without competing with the worker for the check queues.

Two things to point at it:

• Another Statusengine node. Its WorkerOCSP and WorkerOCHP queues take exactly this payload and submit it to that core as a passive check result — see Inbound: queue to core . This is how check execution is distributed across nodes.
• Anything you write yourself, for everything else that wants a copy of every check result as it happens.

Reading the queue

Useful far beyond writing a consumer: this is how you look at what the module is actually publishing, and how you empty a queue that has run away.

apt-get install gearman-tools jq

One job, pretty-printed, then exit:

gearman -w -c 1 -f statusngin_ocsp | jq .

gearman -w -f statusngin_ocsp > /dev/null

Add -h and -p for a job server that is not on localhost:4730. If the tool cannot reach one at all it prints nothing and waits rather than reporting an error, so silence here means the connection, not an empty queue. gearadmin --status says which.

{
  "messages": [
    {
      "type": 701,
      "flags": 0,
      "attr": 0,
      "timestamp": 1614507120,
      "timestamp_usec": 41434,
      "servicecheck": {
        "host_name": "linksys-srw224p",
        "service_description": "PING",
        "command_line": "$USER1$/check_ping -H $HOSTADDRESS$ -w $ARG1$ -c $ARG2$ -p 5",
        "command_name": "check_ping!200.0,20%!600.0,60%",
        "output": "PING CRITICAL - Packet loss = 100%",
        "long_output": "",
        "perf_data": "rta=600.000000ms;200.000000;600.000000;0.000000 pl=100%;20;60;0",
        "check_type": 0,
        "current_attempt": 3,
        "max_attempts": 3,
        "state_type": 1,
        "state": 2,
        "timeout": 60,
        "start_time": 1614507110,
        "end_time": 1614507120,
        "early_timeout": 0,
        "execution_time": 10.004374,
        "latency": 0.03640900179743767,
        "return_code": 2
      }
    }
  ],
  "format": "none"
}

<?php

class StatusengineOcspProcessor {
    private \GearmanWorker $GearmanWorker;

    public function __construct() {
        $this->GearmanWorker = new \GearmanWorker();
        $this->GearmanWorker->addServer('127.0.0.1', 4730);

        // Consume statusngin_ocsp and hand each job to handleOcsp() below.
        $this->GearmanWorker->addFunction('statusngin_ocsp', [$this, 'handleOcsp']);
    }

    public function loop(): void {
        while (true) {
            $this->GearmanWorker->work();
        }
    }

    public function handleOcsp(\GearmanJob $job): void {
        print_r(json_decode($job->workload()));
    }
}

(new StatusengineOcspProcessor())->loop();

Submitting data to the core

Everything so far has been one way. WorkerCommand is the way back: the module consumes it and applies what it finds to the running core, which is what lets the worker’s /commands endpoint reach Naemon at all. WorkerOCSP and WorkerOCHP — described above — take one fixed payload each; this queue takes four different commands.

Every message is an object with a Command and a Data:

{"Command": "check_result", "Data": { … }}

An unknown Command, or a message missing either key, is logged and dropped.

check_result

Submits a passive check result. Data is a check result object; a service_description is what makes it a service check rather than a host check.

{
  "Command": "check_result",
  "Data": {
    "host_name": "localhost",
    "service_description": "PING",
    "output": "PING OK - Packet loss = 0%, RTA = 0.05 ms",
    "long_output": "",
    "perf_data": "rta=0.055000ms;100.000000;500.000000;0.000000 pl=0%;20;60;0",
    "return_code": 0,
    "check_type": 1,
    "start_time": 1614507125,
    "end_time": 1614507129,
    "early_timeout": 0,
    "latency": 0.037,
    "exited_ok": 1
  }
}

The three output fields are joined back into the single string the core wants before the result is submitted: output|perf_data on the first line, then long_output on the next. You do not build that string yourself. Use the individual fields instead.

schedule_check

Moves the next check of a host or service to a given time. schedule_time is a Unix timestamp and the check is scheduled at it, not merely no later than it.

{
  "Command": "schedule_check",
  "Data": {
    "host_name": "localhost",
    "service_description": "PING",
    "schedule_time": 1614507200
  }
}

host_name and a non-zero schedule_time are both required. Leaving service_description out schedules the host check instead. An object the core does not know is logged by name and ignored, so a typo in a host name fails quietly rather than loudly. Look for Received schedule_check command for unknown host in the core’s log.

delete_downtime

Deletes downtimes matching what you give it. Only host_name is required; service_description, start_time, end_time and comment narrow it further, and leaving them all out deletes every downtime on that host.

{
  "Command": "delete_downtime",
  "Data": {
    "host_name": "localhost",
    "service_description": "PING",
    "start_time": 1614507000,
    "end_time": 1614510600,
    "comment": "Ansible run"
  }
}

raw

Data is not an object here but a string, handed to the core’s external command processor unchanged — the same line you would otherwise echo into naemon.cmd. That makes every external command ↗ available without the module needing to know about it.

{"Command": "raw", "Data": "[1614507120] SCHEDULE_FORCED_HOST_CHECK;localhost;1614507120"}

Bulk messages here too

The inbound side accepts the same messages envelope as the outbound queues, so a client can submit many commands in one job:

{"messages": [
  {"Command": "raw", "Data": "[1614507120] SCHEDULE_FORCED_HOST_CHECK;localhost;1614507120"},
  {"Command": "raw", "Data": "[1614507120] SCHEDULE_FORCED_HOST_CHECK;router;1614507120"}
]}