404Forbiden

✏️ Editing: nagios.py

import agent_util
import os
import sys

DEFAULT_NAGIOS_FOLDER = "/usr/share/fm-agent/nagios"

def parse_performance_data(output, log):
    """
    Parse Nagios performance data, as defined in their API docs at
    https://assets.nagios.com/downloads/nagioscore/docs/nagioscore/3/en/pluginapi.html

Returns a list of dictionaries, each with the following keys (if found in output):
     - label
     - value
     - unit
     - min_value
     - max_value
    """
    lines = output.strip().split("\n")

# Grab the first line, after the | which is performance data
    parts = []
    if "|" in lines[0]:
        parts.extend(lines[0].split("|", 1)[1].split(" "))

# Look in rest of output for a | indicating that all the rest is performance data
    found_perf = False
    for line in lines[1:]:
        if not found_perf and "|" in line:
            found_perf = True
            line = line.split("|", 1)[1]
        if found_perf:
            parts.extend(line.split(" "))

metrics = []

# Parse each part into component pieces
    for part in parts:
        metric = {}
        try:
            pieces = part.strip().strip(";").split(";")
            label, value = pieces[0].split("=")
            metric["label"] = label.strip("'").strip()

if value == "U":
                value = None
            else:
                # Split the value from the unit, if it's there
                unit = ""
                for i, char in enumerate(value):
                    if char not in "-0123456789.":
                        unit = value[i:]
                        value = value[:i]
                        break
                value = float(value)

metric["value"] = value
            metric["unit"] = unit

# Extract min and max, if present
            if len(pieces) >= 4:
                metric["min_value"] = float(pieces[3])
            if len(pieces) >= 5:
                metric["max_value"] = float(pieces[4])

metrics.append(metric)
        except:
            log.exception("Error parsing Nagios output: %s" % part)

return metrics

class NagiosPlugin(agent_util.Plugin):
    textkey = "nagios"
    label = "Nagios"

@classmethod
    def get_metadata(self, config):
        status = agent_util.SUPPORTED
        msg = None

custom_folder = config.get("plugins_location")

plugins = self._get_all_plugins(DEFAULT_NAGIOS_FOLDER, custom_folder)

if not plugins:
            status = agent_util.UNSUPPORTED
            msg = "No nagios plugin found on %s %s" % (
                DEFAULT_NAGIOS_FOLDER,
                custom_folder or "",
            )
            return {}

metric_options = []
        for dir, plugin in plugins:
            self.log.info("Found Nagios plugin: %s" % plugin)

# Add an option for the return status
            metric_options.append(
                {
                    "nagios_script": plugin,
                    "metric": "status",
                    "resource": "%s: status" % (plugin),
                }
            )

try:
                ret_code, output = agent_util.execute_command(os.path.join(dir, plugin))
                self.log.info("Nagios: %s %s" % (ret_code, output))
                for metric in parse_performance_data(output, self.log):
                    metric_options.append(
                        {
                            "nagios_script": plugin,
                            "metric": metric["label"],
                            "resource": "%s: %s" % (plugin, metric["label"]),
                        }
                    )
            except:
                self.log.exception(
                    "Error gathering metadata for Nagios plugin %s" % plugin
                )

self.log.info(
            "%s Nagios options found: %s" % (len(metric_options), metric_options)
        )

options_schema = {
            "nagios_script": "string",
            "metric": "string",
            "resource": "string",
        }
        metadata = {
            "nagios_metric": {
                "label": "Nagios Metric",
                "options": metric_options,
                "options_schema": options_schema,
                "status": status,
                "error_message": msg,
            }
        }

return metadata

@staticmethod
    def _get_all_plugins(*args):
        plugins = []
        for arg in args:
            if arg and os.path.isdir(arg):
                for file in os.listdir(arg):
                    if os.access(os.path.join(arg, file), os.X_OK):
                        plugins.append((arg, file))
        return plugins

def check(self, textkey, plugin_metric, config):
        split = plugin_metric.split(":")
        plugin_name = split[0].strip()
        metric_name = split[1].strip()

custom_folder = config.get("plugins_location")
        plugins = self._get_all_plugins(DEFAULT_NAGIOS_FOLDER, custom_folder)

for dir, plugin in plugins:
            if plugin == plugin_name:
                self.log.debug(
                    "Executing %s to get %s" % (os.path.join(dir, plugin), textkey)
                )
                ret_code, output = agent_util.execute_command(os.path.join(dir, plugin))
                self.log.debug("Nagios: %s %s" % (ret_code, output))

if metric_name == "status":
                    return ret_code

for metric in parse_performance_data(output, self.log):
                    if metric["label"] == metric_name:
                        return metric["value"]

self.log.info("No matching Nagios plugin found for %s" % textkey)
        return None

Sid Gifari From Gifari Industries - BD Cyber Security Team