#!/usr/bin/env python3 # pylint: disable=too-many-locals, line-too-long, too-many-statements """Mailcow check for domains""" from pprint import pprint from cmk.utils import debug # import necessary elements from API version 2 from cmk.agent_based.v2 import ( AgentSection, CheckPlugin, Service, Result, State, Metric, render, ) def get_state_upper( levels: tuple[int | float, int | float], value: int | float ) -> State: """returns OK/WARN/CRIT depending on the given parameters""" warn, crit = levels if value >= crit: return State.CRIT if value >= warn: return State.WARN return State.OK def parse_mailcow_domains(string_table): """the parse function""" parsed_data = {} # convert the raw output of the agent section to an meaningful structure # do type conversions and so on for line in string_table: domainname = line[0] value_active = int(line[1]) if value_active == 1: active = "yes" else: active = "no" # calculate creation and last modification date in human readable format create_time_value = line[2] if create_time_value == "None": create_time_data = "Not available" else: create_time_data = create_time_value modify_time_value = line[3] if modify_time_value == "None": modify_time_data = "Never" else: modify_time_data = modify_time_value # calculate percentage of used mailboxes max_number_of_mailboxes = int(line[4]) number_of_mailboxes = int(line[5]) percent_used_mailboxes = number_of_mailboxes * 100 / max_number_of_mailboxes # calculate percentage of used aliases max_number_of_aliases = int(line[6]) number_of_aliases = int(line[7]) percent_used_aliases = number_of_aliases * 100 / max_number_of_aliases # number of messages within domain total_number_of_messages = int(line[8]) # calculate storage used for all messages in domain total_number_of_bytes_used = int(line[9]) quota = int(line[10]) percent_storage_used_for_messages = total_number_of_bytes_used * 100 / quota # store all (calculated) data parsed_data[f"{domainname}"] = [ active, create_time_data, modify_time_data, max_number_of_mailboxes, number_of_mailboxes, percent_used_mailboxes, max_number_of_aliases, number_of_aliases, percent_used_aliases, total_number_of_messages, total_number_of_bytes_used, quota, percent_storage_used_for_messages, ] return parsed_data def discover_mailcow_domains(section): """the discover function""" # since we have a service with item here we must create one service per item for key in section: yield Service(item=key) def check_mailcow_domains(item, params, section): """the check function""" if debug.enabled(): pprint(section) domain = section.get(item) if not domain: # if a previously found domain does not exist anymore, create a meaningful result yield Result( state=State.UNKNOWN, summary="Domain not found anymore, maybe it was deleted?", ) return # get all infos regarding the domain active = domain[0] create_time = domain[1] modify_time = domain[2] max_number_of_mailboxes = domain[3] number_of_mailboxes = domain[4] percent_used_mailboxes = domain[5] max_number_of_aliases = domain[6] number_of_aliases = domain[7] percent_used_aliases = domain[8] total_number_of_messages = domain[9] total_number_of_bytes_used = domain[10] quota = domain[11] percent_storage_used_for_messages = domain[12] # create (main) service for used storage (domain quota) _type, levels = params["levels_mailcow_domains_quota_used"] state_quota = get_state_upper(levels, percent_storage_used_for_messages) # create graph for used quota yield Metric( "mailcow_domains_used_quota", percent_storage_used_for_messages, levels=levels ) summary_quota = ( f"Storage quota used is {render.percent(percent_storage_used_for_messages)}" ) details_quota = f"Storage quota: {render.bytes(total_number_of_bytes_used)} of {render.bytes(quota)} used" # create service yield Result(state=state_quota, summary=summary_quota, details=details_quota) # create some additional services and information only in service details notice = f"Active: {active}" yield Result(state=State.OK, notice=notice) notice = f"Creation date: {create_time}" yield Result(state=State.OK, notice=notice) notice = f"Last modified: {modify_time}" yield Result(state=State.OK, notice=notice) # create service for number of configured mailboxes (percent) _type, levels = params["levels_mailcow_domains_mailboxes_used"] state_mailboxes = get_state_upper(levels, percent_used_mailboxes) yield Metric("mailcow_domains_mailboxes", percent_used_mailboxes, levels=levels) notice = f"Used mailboxes: {render.percent(percent_used_mailboxes)}, {number_of_mailboxes} of {max_number_of_mailboxes} in use" yield Result(state=state_mailboxes, notice=notice) # create service for number of configured aliases (percent) _type, levels = params["levels_mailcow_domains_aliases_used"] state_aliases = get_state_upper(levels, percent_used_aliases) yield Metric("mailcow_domains_aliases", percent_used_aliases, levels=levels) notice = f"Used aliases: {render.percent(percent_used_aliases)}, {number_of_aliases} of {max_number_of_aliases} in use" yield Result(state=state_aliases, notice=notice) # create service for number of messages _type, levels = params["levels_mailcow_domains_num_messages"] state_messages = get_state_upper(levels, total_number_of_messages) yield Metric("mailcow_domains_messages", total_number_of_messages, levels=levels) notice = f"Number of messages: {total_number_of_messages}" yield Result(state=state_messages, notice=notice) # create service for number of configured aliases (absolute) _type, levels = params["levels_mailcow_domains_num_aliases"] state_aliases = get_state_upper(levels, number_of_aliases) yield Metric("mailcow_domains_configured_aliases", number_of_aliases, levels=levels) notice = ( f"Number of aliases: {number_of_aliases}, max {max_number_of_aliases} allowed" ) yield Result(state=state_aliases, notice=notice) # create service for number of configured mailboxes (absolute) _type, levels = params["levels_mailcow_domains_num_mailboxes"] state_mailboxes = get_state_upper(levels, number_of_mailboxes) yield Metric( "mailcow_domains_configured_mailboxes", number_of_mailboxes, levels=levels ) notice = f"Number of mailboxes: {number_of_mailboxes}, max {max_number_of_mailboxes} allowed" yield Result(state=state_mailboxes, notice=notice) # create the new agent section, must begin with "agent_section_" # and must be an instance of "AgentSection" agent_section_mailcow_domains = AgentSection( # "name" must exactly match the section name within the agent output name="mailcow_domains", # define the parse function, name is arbitrary, a good choice is to choose # "parse_" as prefix and append the section name parse_function=parse_mailcow_domains, ) # create the new check plugin, must begin with "check_plugin_" # and must be an instance of "CheckPlugin" check_plugin_mailcow_domains = CheckPlugin( # "name" should be the same as the corresponding section within the agent output name="mailcow_domains", # this is a service with item, so you have to include a place holder for the item id service_name="Mailcow domain %s", # define the discovery function, name is arbitrary, a good choice is to choose # "discover_" as prefix and append the section name discovery_function=discover_mailcow_domains, # define the check function, name is arbitrary, a good choice is to choose # "check_" as prefix and append the section name check_function=check_mailcow_domains, # define the default parameters check_default_parameters={ "levels_mailcow_domains_quota_used": ("fixed", (65.0, 85.0)), "levels_mailcow_domains_mailboxes_used": ("fixed", (65.0, 85.0)), "levels_mailcow_domains_aliases_used": ("fixed", (65.0, 85.0)), "levels_mailcow_domains_num_messages": ("fixed", (10_000, 25_000)), "levels_mailcow_domains_num_aliases": ("fixed", (100, 250)), "levels_mailcow_domains_num_mailboxes": ("fixed", (100, 250)), }, # connect to the ruleset where parameters can be defined # must match the name of the ruleset exactly check_ruleset_name="mailcow_domains", )