CheckMK custom monitoring script

readme.md

example monitoring script. cmk.py is a libray file used by check_dir_size.py. Place both files in /usr/lib/check_mk_agent/local.

Stolen from CheckMK forum: https://forum.checkmk.com/t/folder-size-age-of-s3-buckets/34329/5

check_dir_size.py

#!/usr/bin/env python3

""" This module is used for monitoring directory sizes and reporting these to CheckMK.
    The status is automatically calculated based on WARN and CRIT levels of directory sizes.
    If a max_size (see: dictionary host_paths) has been configured, WARN is 80% and CRIT is 90% of max_size.
"""

###############
### imports ###
###############

import glob
import socket
from pathlib import Path
from cmk import CMKService

#################
### functions ###
#################

def get_foldersize(fullpath_str):
    """ Function expects a full path and will add the size of every file (except symlinks) of every subdirectory and return the total size in Bytes """
    fullpath_pathlib = Path(fullpath_str)
    total_size = 0
    for file in fullpath_pathlib.iterdir():
        if file.is_dir():
            total_size += get_foldersize(file)
        elif file.is_symlink():
            continue
        else:
            total_size += file.stat().st_size
    return total_size

def sizeof_format(num, suffix='B'):
    """ Function takes a size in Bytes and converts it to the matching unit like MiB / GiB / ... """
    for unit in [' ',' Ki',' Mi',' Gi',' Ti',' Pi',' Ei',' Zi']:
        if abs(num) < 1024.0:
            return "%3.1f%s%s" % (num, unit, suffix)
        num /= 1024.0
    return "%.1f%s%s" % (num, 'Yi', suffix)

#################
### variables ###
#################

# constants
STATE_OK = 0
STATE_WARN = 1
STATE_CRIT = 2
STATE_UNKNOWN = 3
NLDELIMIT = "\\\\n"

# variables
host_paths = [  # a dictionary with hostnames and folder paths, so you can monitor different folders on different hosts
    { 'hostname': 'SERVER_HOSTNAME', 'path': 'PATH_TO_FOLDER', 'max_size': 20*1024*1024*1024 } # size in Bytes
]

#################
### main code ###
#################

cmk_list = []

hostname = socket.gethostname()
if '.' in hostname:
    hostname = hostname.split('.')[0]

# search for the right hostname
for host in host_paths:
    if host['hostname'] == hostname:
        folder = host['path']
        size = get_foldersize(host['path'])
        max_size = host['max_size']

        cmk = CMKService(f"\"DirSize {folder}\"", uses_metrics=True, state_calculated=True)

        # create short output string
        if max_size != 0:
            cmk.add_short(f"{sizeof_format(size)} / {sizeof_format(max_size)}")
            cmk.add_metric('fs_size', size, int(max_size*0.8), max_size)
        else:
            cmk.add_short(f"{sizeof_format(size)}")

        cmk_list.append(cmk)

for cmkservice in cmk_list:
    print(cmkservice.get_serviceString())

cmk.py

#/usr/bin/env python3
"""
    This Module is used to aggregate all important things
    needed to create a CheckMK Service in a unified way.
"""

STATE_OK = 0
STATE_WARN = 1
STATE_CRIT = 2
STATE_UNKN = 3
NLDELIMIT = "\\n"

class CMKService():
    """
        This Class holds all information needed for CheckMK.
        - Service Name
        - Service Metrics
        - Service State
        - Service Output (short and long)
    """

    def __init__(self, name, uses_metrics=False, state_calculated=False):
        self._name = name
        self._uses_metrics = uses_metrics
        self._state_calculated = state_calculated

        self._state = STATE_UNKN
        self._metrics = []  # List of dictionaries
        self._short = []    # short service text messages, which you see in the summary
        self._long = []     # long service text messages, accessible via click on service

        if state_calculated:
            self._state_calculated = True
            self._state = 'P'

    def add_short(self, text):
        """ Adds a short (single line) service text """
        self._short.append(text)
    def add_long(self, text):
        """ Adds a long (multi-line) service text """
        self._long.append(text)

    def add_metric(self, name, value, warn=None, crit=None):
        """
            This method adds a Metric to the CheckMK Service.
            WARN and CRIT values are optional.
        """
        self._metrics.append({
            'name': name,
            'value': value,
            'warn': warn,
            'crit': crit,
        })

    def short_isEmpty(self):
        if len(self._short) == 0:
            return True
        return False

    def get_serviceString(self):
        """
            This method returns a complete Service String.
            The output string is already formatted the way CheckMK needs it.
        """
        service_str = ''
        service_str += str(self._state)+' '+self._name

        if self._uses_metrics:
            tmp = ''
            for metric in self._metrics:
                # if the string has a metric already (not 'falsy' in a boolean context),
                # add the checkmk metric seperator '|'
                if tmp: tmp += '|'
                tmp += metric['name']+'='+str(metric['value'])
                if metric['warn']: tmp += ";"+str(metric['warn'])
                if metric['crit']: tmp += ";"+str(metric['crit'])
            service_str += " "+tmp
        else: service_str += ' -'

        tmp = ''
        for short in self._short:
            if tmp: tmp += ' // '
            tmp += short
        service_str += " "+tmp

        tmp = ''
        for long in self._long:
            tmp += NLDELIMIT+''+long
        service_str += tmp

        return service_str

    def uses_metrics(self):
        """ Returns BOOL TRUE/FALSE if Service uses Metrics """
        return self._uses_metrics
    def state_calculated(self):
        """ Returns BOOL TRUE/FALSE if state is Self-calculated ('P') """
        return self._state_calculated

    # Getters / Setters
    @property
    def state(self):
        return self._state
    @state.setter
    def state(self, new_state):
        self._state = new_state