cmk-bonobo/Mailcow-CheckMK
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

initial upload of version 2.1.0 which uses the new plugin API

Browse Source
This commit is contained in:
Ralf Mellis 2025-04-05 11:47:59 +02:00
parent 358f84307c
commit cc2aaee908
34 changed files with 1204 additions and 1238 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
README.md
View File

@ -12,22 +12,21 @@ Mailboxes with quota usage, number of messages, last logins via IMAP/POP3/SMTP
## Caveats:
Tested only with dockerized versions of Mailcow
Tested with Mailcow version 2022-07a and higher
Tested with Mailcow version 2023-01 and higher
Please feel free to add other working environments via review/email.
## Upgrade from older MKPs (before 1.2.0):
If you upgrade from a already installed version before 1.2.0, you have to re-create your rules for the "Mailcow Server Information" (reason: massive parameter changes).
Upgrade from older MKPs (before 2.1.0): If you upgrade from a already installed version before 2.1.0, you have to re-create your rules for the "Mailcow Server Information" (reason: parameter handling changes and migration to the new plugin API).
HINT: It is a good idea to take a screenshot from all other Mailcow related rules as well.
## Steps to accomplish this without problems:
1. Take a screenshot of your settings in the above mentioned ruleset
2. Assure that you have access to the API keys you used within the current rules
3. Delete all rules for "Mailcow Server Information"
4. Install and enable the new MKP
5. Re-create your rules with the previously saved information from steps 1 and 2
6. Apply your changes
Steps to accomplish this without problems:
Take a screenshot of your settings in the above mentioned ruleset
Assure that you have access to the API keys you used within the current rules
Delete all rules for "Mailcow Server Information"
Install and enable the new MKP
Re-create your rules with the previously saved information from steps 1 and 2
Apply your changes
## General installation instructions:
@ -57,6 +56,8 @@ If you upgrade from a already installed version before 1.2.0, you have to re-cre
## Version history:
2025/03/30 2.1.0 Migration to new plugin API done
2024/01/26 1.2.0 Added password store option for the API key, added version check
2023/09/22: 1.1.4 Bugfix for the last bugfix :-)

162
local/lib/python3/cmk/base/plugins/agent_based/mailcow_domains.py
View File

@ -1,162 +0,0 @@
#!/usr/bin/env python3
import time
import random
from pprint import pprint
from .agent_based_api.v1 import get_value_store, get_rate, register, render, Service, Result, State, Metric
def getStateUpper(levels, value):
warn, crit = levels
if value >= crit:
return State.CRIT
if value >= warn:
return State.WARN
return State.OK
# services with item --> storageid and usage of value_store (to be able to calculate rates)
def discover_mailcow_domains(section):
for key in section:
yield(Service(item = key))
def check_mailcow_domains(item, params, section):
domain = item
# get all values from section
active = section[domain][0]
create_time = section[domain][1]
modify_time = section[domain][2]
max_number_of_mailboxes = section[domain][3]
number_of_mailboxes = section[domain][4]
percent_used_mailboxes = section[domain][5]
max_number_of_aliases = section[domain][6]
number_of_aliases = section[domain][7]
percent_used_aliases = section[domain][8]
total_number_of_messages = section[domain][9]
total_number_of_bytes_used = section[domain][10]
quota = section[domain][11]
percent_storage_used_for_messages = section[domain][12]
# create (main) service for used storage (domain quota)
warn, crit = params["levels_mailcow_domains_quota_used"]
levels = (warn, crit)
state_quota = getStateUpper(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 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)
warn, crit = params["levels_mailcow_domains_mailboxes_used"]
levels = (warn, crit)
state_mailboxes = getStateUpper(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)
warn, crit = params["levels_mailcow_domains_aliases_used"]
levels = (warn, crit)
state_aliases = getStateUpper(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
warn, crit = params["levels_mailcow_domains_num_messages"]
levels = (warn, crit)
state_messages = getStateUpper(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)
warn, crit = params["levels_mailcow_domains_num_aliases"]
levels = (warn, crit)
state_aliases = getStateUpper(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)
warn, crit = params["levels_mailcow_domains_num_mailboxes"]
levels = (warn, crit)
state_mailboxes = getStateUpper(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))
def parse_mailcow_domains_section(string_table):
# convert the raw output of the agent section into a meaningful structure, do type conversions and so on
parsed_data = {}
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
register.agent_section(
name = "mailcow_domains",
parse_function = parse_mailcow_domains_section,
)
register.check_plugin(
name = "mailcow_domains",
service_name = "Mailcow domain %s",
discovery_function = discover_mailcow_domains,
check_function = check_mailcow_domains,
check_default_parameters = {
"levels_mailcow_domains_quota_used": (65.0, 85.0),
"levels_mailcow_domains_mailboxes_used": (65.0, 85.0),
"levels_mailcow_domains_aliases_used": (65.0, 85.0),
"levels_mailcow_domains_num_messages": (10000, 25000),
"levels_mailcow_domains_num_aliases": (100, 250),
"levels_mailcow_domains_num_mailboxes": (100, 250),
},
check_ruleset_name="mailcow_domains",
)

142
local/lib/python3/cmk/base/plugins/agent_based/mailcow_info.py
View File

@ -1,142 +0,0 @@
#!/usr/bin/env python3
from pprint import pprint
from .agent_based_api.v1 import register, render, Service, Result, State, Metric
def getStateUpper(levels, value):
warn, crit = levels
if value >= crit:
return State.CRIT
if value >= warn:
return State.WARN
return State.OK
def getStateLower(levels, value):
warn, crit = levels
if value < crit:
return State.CRIT
if value < warn:
return State.WARN
return State.OK
def discover_mailcow_info(section):
yield (Service())
def check_mailcow_info(params, section):
for key in section:
if key == "mailcow":
# get thresholds
levels_num_domains = params["levels_num_domains"]
levels_num_mailboxes = params["levels_num_mailboxes"]
levels_num_global_messages = params["levels_num_global_messages"]
version = section[key]["version"]
git_version = section[key]["git_version"]
check_version_enabled = section[key]["check_version_enabled"]
update_available = section[key]["update_available"]
num_domains = section[key]["num_domains"]
num_mailboxes = section[key]["num_mailboxes"]
num_global_messages = section[key]["num_global_messages"]
# create graphs for number of domains, mailboxes and messages
yield (Metric("mc_num_domains", num_domains, levels=levels_num_domains))
yield (
Metric("mc_num_mailboxes", num_mailboxes, levels=levels_num_mailboxes)
)
yield (
Metric(
"mc_num_global_messages",
num_global_messages,
levels=levels_num_global_messages,
)
)
# create overall result
if check_version_enabled:
if update_available:
summary = f"Update available: Running version is {version}, Github version is: {git_version}"
state = State.WARN
else:
summary = f"No update available: Running version is {version}, Github version is: {git_version}"
state = State.OK
else:
summary = f"Version is {version}, Update check is disabled"
state = State.OK
details = f"Mailcow version: {version}\nNumber of domains: {num_domains}\nNumber of mailboxes: {num_mailboxes}\nNumber of messages: {num_global_messages}"
yield Result(state=state, summary=summary, details=details)
# Create result for number of domains
warn, crit = levels_num_domains
state = getStateUpper((warn, crit), num_domains)
notice = f"Number of domains: {num_domains}"
if state != State.OK:
yield (Result(state=state, notice=notice))
# Create result for number of mailboxes
warn, crit = levels_num_mailboxes
state = getStateUpper((warn, crit), num_mailboxes)
notice = f"Number of mailboxes: {num_mailboxes}"
if state != State.OK:
yield (Result(state=state, notice=notice))
# Create result for number of global messages
warn, crit = levels_num_global_messages
state = getStateUpper((warn, crit), num_global_messages)
notice = f"Number of messages: {num_global_messages}"
if state != State.OK:
yield (Result(state=state, notice=notice))
def parse_mailcow_info_section(string_table):
# pprint(string_table)
parsed_data = {
"mailcow": {},
}
# we expect only one line
line = string_table[0]
version = line[0]
num_domains = int(line[1])
num_mailboxes = int(line[2])
num_global_messages = int(line[3])
git_version = line[4]
update_available = line[5]
if update_available == "True":
update_available = True
else:
update_available = False
check_version_enabled = line[6]
if check_version_enabled == "True":
check_version_enabled = True
else:
check_version_enabled = False
parsed_data["mailcow"]["version"] = version
parsed_data["mailcow"]["num_domains"] = num_domains
parsed_data["mailcow"]["num_mailboxes"] = num_mailboxes
parsed_data["mailcow"]["num_global_messages"] = num_global_messages
parsed_data["mailcow"]["git_version"] = git_version
parsed_data["mailcow"]["update_available"] = update_available
parsed_data["mailcow"]["check_version_enabled"] = check_version_enabled
# pprint(parsed_data)
return parsed_data
register.agent_section(
name="mailcow_info",
parse_function=parse_mailcow_info_section,
)
register.check_plugin(
name="mailcow_info",
service_name="Mailcow Info",
discovery_function=discover_mailcow_info,
check_function=check_mailcow_info,
check_default_parameters={
"levels_num_domains": (100, 200),
"levels_num_mailboxes": (500, 1000),
"levels_num_global_messages": (100000, 250000),
},
check_ruleset_name="mailcow_info",
)

155
local/lib/python3/cmk/base/plugins/agent_based/mailcow_mailboxes.py
View File

@ -1,155 +0,0 @@
#!/usr/bin/env python3
import time
import random
from pprint import pprint
from .agent_based_api.v1 import get_value_store, get_rate, register, render, Service, Result, State, Metric
def getStateUpper(levels, value):
warn, crit = levels
if value >= crit:
return State.CRIT
if value >= warn:
return State.WARN
return State.OK
def discover_mailcow_mailboxes(section):
for key in section:
yield(Service(item = key))
def check_mailcow_mailboxes(item, params, section):
mailbox = item
# get all values from section
active = section[mailbox][0]
create_time = section[mailbox][1]
modify_time = section[mailbox][2]
display_name = section[mailbox][3]
number_of_messages = section[mailbox][4]
_percent_in_use = section[mailbox][5]
quota = section[mailbox][6]
total_number_of_bytes_used = section[mailbox][7]
percent_storage_used_for_messages = section[mailbox][8]
last_imap_login_data = section[mailbox][9]
last_pop3_login_data = section[mailbox][10]
last_smtp_login_data = section[mailbox][11]
# create (main) service for used storage (mailbox quota)
warn, crit = params["levels_mailcow_mailboxes_quota_used"]
levels = (warn, crit)
state_quota = getStateUpper(levels, percent_storage_used_for_messages)
# create graph for used quota
yield Metric("mailcow_mailboxes_used_quota", percent_storage_used_for_messages, levels=levels)
summary_quota = f"Storage quota for mailbox of '{display_name}' is {render.percent(percent_storage_used_for_messages)}"
details_quota = f"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 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))
notice = f"Last IMAP login: {last_imap_login_data} ago"
yield(Result(state=State.OK, notice=notice))
notice = f"Last POP3 login: {last_pop3_login_data} ago"
yield(Result(state=State.OK, notice=notice))
notice = f"Last SMTP login: {last_smtp_login_data} ago"
yield(Result(state=State.OK, notice=notice))
# create service for number of messages
warn, crit = params["levels_mailcow_mailboxes_num_messages"]
levels = (warn, crit)
state_messages = getStateUpper(levels, number_of_messages)
yield Metric("mailcow_mailboxes_messages", number_of_messages, levels=levels)
notice = f"Number of messages: {number_of_messages}"
yield(Result(state=state_messages, notice=notice))
def parse_mailcow_mailboxes_section(string_table):
# convert the raw output of the agent section into a meaningful structure, do type conversions and so on
parsed_data = {}
for line in string_table:
mailboxname = 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
# get display name
display_name = line[4]
# number of messages within mailbox
number_of_messages = int(line[5])
# calculate storage used for all messages in mailbox
quota = int(line[7])
total_number_of_bytes_used = int(line[8])
if quota == 0:
# quota is not set, if this is the case, line[6] contains no numeric value, but the char "-"
# so set all usage counters to zero
percent_in_use = 0
percent_storage_used_for_messages = 0
else:
# percent in use, rounded to full percent (calculated by Mailcow)
percent_in_use = int(line[6])
# let's calculate our own value
percent_storage_used_for_messages = total_number_of_bytes_used * 100 / quota
# get time of last login for IMAP/POP3/SMTP (seconds since epoch)
last_imap_login = int(line[9])
last_pop3_login = int(line[10])
last_smtp_login = int(line[11])
# transfer these times into a human friendly format
if last_imap_login == 0:
last_imap_login_data = "Never"
else:
curr_time = int(time.time())
diff_time = curr_time - last_imap_login
last_imap_login_data = render.timespan(diff_time)
if last_pop3_login == 0:
last_pop3_login_data = "Never"
else:
curr_time = int(time.time())
diff_time = curr_time - last_pop3_login
last_pop3_login_data = render.timespan(diff_time)
if last_smtp_login == 0:
last_smtp_login_data = "Never"
else:
curr_time = int(time.time())
diff_time = curr_time - last_smtp_login
last_smtp_login_data = render.timespan(diff_time)
# store all (calculated) data
parsed_data[f"{mailboxname}"] = [active, create_time_data, modify_time_data,
display_name, number_of_messages, percent_in_use,
quota, total_number_of_bytes_used, percent_storage_used_for_messages,
last_imap_login_data, last_pop3_login_data, last_smtp_login_data
]
return parsed_data
register.agent_section(
name = "mailcow_mailboxes",
parse_function = parse_mailcow_mailboxes_section,
)
register.check_plugin(
name = "mailcow_mailboxes",
service_name = "Mailcow mailbox %s",
discovery_function = discover_mailcow_mailboxes,
check_function = check_mailcow_mailboxes,
check_default_parameters = {
"levels_mailcow_mailboxes_quota_used": (65.0, 85.0),
"levels_mailcow_mailboxes_num_messages": (1000, 2500),
},
check_ruleset_name="mailcow_mailboxes",
)

19
local/share/check_mk/checks/agent_mailcow
View File

@ -1,19 +0,0 @@
def agent_mailcow_arguments(params, hostname, ipaddress):
return [
"--hostname",
params["hostname"],
"--apikey",
passwordstore_get_cmdline("%s", params["apikey"]),
"--port",
params["port"],
"--check-version",
params["check_version"],
"--no-https",
params["no_https"],
"--no-cert-check",
params["no_cert_check"],
ipaddress,
]
special_agent_info["mailcow"] = agent_mailcow_arguments

72
local/share/check_mk/web/plugins/metrics/mailcow_metrics.py
View File

@ -1,72 +0,0 @@
#!/usr/bin/env python3
from cmk.gui.i18n import _
from cmk.gui.plugins.metrics import (
metric_info,
graph_info,
)
metric_info["mc_num_domains"] = {
"title": _("Number of Domains"),
"unit": "count",
"color": "44/a",
}
metric_info["mc_num_mailboxes"] = {
"title": _("Number of Mailboxes"),
"unit": "count",
"color": "44/b",
}
metric_info["mc_num_global_messages"] = {
"title": _("Number of Messages"),
"unit": "count",
"color": "42/a",
}
metric_info["mailcow_domains_used_quota"] = {
"title": _("Domain Quota Used"),
"unit": "%",
"color": "24/a",
}
metric_info["mailcow_domains_mailboxes"] = {
"title": _("Domain Mailboxes Quota Used"),
"unit": "%",
"color": "44/b",
}
metric_info["mailcow_domains_aliases"] = {
"title": _("Domain Aliases Quota Used"),
"unit": "%",
"color": "44/a",
}
metric_info["mailcow_domains_messages"] = {
"title": _("Global Number of Messages"),
"unit": "count",
"color": "24/a",
}
metric_info["mailcow_domains_configured_aliases"] = {
"title": _("Number of Configured Aliases"),
"unit": "count",
"color": "24/a",
}
metric_info["mailcow_domains_configured_mailboxes"] = {
"title": _("Number of Configured Mailboxes"),
"unit": "count",
"color": "24/b",
}
metric_info["mailcow_mailboxes_used_quota"] = {
"title": _("Mailbox Quota Used"),
"unit": "%",
"color": "24/a",
}
metric_info["mailcow_mailboxes_messages"] = {
"title": _("Number of Messages"),
"unit": "count",
"color": "24/b",
}

44
local/share/check_mk/web/plugins/perfometer/mailcow_perfometers.py
View File

@ -1,44 +0,0 @@
#!/usr/bin/env python3
from cmk.gui.plugins.metrics import perfometer_info
perfometer_info.append(
{
"type": "stacked",
"perfometers": [
{
"type": "linear",
"segments": [
"mc_num_domains",
"mc_num_mailboxes",
"mc_num_global_messages",
],
},
],
}
)
perfometer_info.append(
{
"type": "stacked",
"perfometers": [
{
"type": "linear",
"segments": ["mailcow_domains_used_quota"],
"total": 100.0,
},
],
}
)
perfometer_info.append(
{
"type": "stacked",
"perfometers": [
{
"type": "linear",
"segments": ["mailcow_mailboxes_used_quota"],
"total": 100.0,
},
],
}
)

115
local/share/check_mk/web/plugins/wato/mailcow_domains_rules.py
View File

@ -1,115 +0,0 @@
from cmk.gui.i18n import _
from cmk.gui.plugins.wato import (
CheckParameterRulespecWithItem,
rulespec_registry,
RulespecGroupCheckParametersApplications
)
from cmk.gui.valuespec import (
Dictionary,
ListChoice,
TextAscii,
Percentage,
Tuple,
Float,
Integer
)
def _item_spec_mailcow_domains():
return TextAscii(
title=_("Domain name")
)
def _parameter_spec_mailcow_domains():
return Dictionary(
elements=[
("levels_mailcow_domains_quota_used", Tuple(
title=_("Mailcow domains quota usage for storage"),
elements=[
Percentage(
title=_("Warning at"),
default_value=65.0,
),
Percentage(
title=_("Critical at"),
default_value=85.0,
)
],
)),
("levels_mailcow_domains_mailboxes_used", Tuple(
title=_("Mailcow domains mailboxes usage"),
elements=[
Percentage(
title=_("Warning at"),
default_value=65.0,
),
Percentage(
title=_("Critical at"),
default_value=85.0,
)
],
)),
("levels_mailcow_domains_aliases_used", Tuple(
title=_("Mailcow domains aliases usage"),
elements=[
Percentage(
title=_("Warning at"),
default_value=65.0,
),
Percentage(
title=_("Critical at"),
default_value=85.0,
)
],
)),
("levels_mailcow_domains_num_messages", Tuple(
title=_("Number of messages"),
elements=[
Integer(
title=_("Warning at"),
default_value=10000,
),
Integer(
title=_("Critical at"),
default_value=25000,
)
],
)),
("levels_mailcow_domains_num_aliases", Tuple(
title=_("Number of configured aliases"),
elements=[
Integer(
title=_("Warning at"),
default_value=100,
),
Integer(
title=_("Critical at"),
default_value=250,
)
],
)),
("levels_mailcow_domains_num_mailboxes", Tuple(
title=_("Number of configured mailboxes"),
elements=[
Integer(
title=_("Warning at"),
default_value=100,
),
Integer(
title=_("Critical at"),
default_value=250,
)
],
)),
],
)
rulespec_registry.register(
CheckParameterRulespecWithItem(
check_group_name="mailcow_domains",
group=RulespecGroupCheckParametersApplications,
match_type="dict",
item_spec=_item_spec_mailcow_domains,
parameter_valuespec=_parameter_spec_mailcow_domains,
title=lambda: _("Levels for Mailcow domains"),
)
)

87
local/share/check_mk/web/plugins/wato/mailcow_info_rules.py
View File

@ -1,87 +0,0 @@
from cmk.gui.i18n import _
from cmk.gui.plugins.wato import (
CheckParameterRulespecWithoutItem,
rulespec_registry,
RulespecGroupCheckParametersApplications,
)
from cmk.gui.valuespec import (
Dictionary,
ListChoice,
Tuple,
Percentage,
Integer,
Float,
)
def _parameter_spec_mailcow_info():
return Dictionary(
elements=[
(
"levels_num_domains",
Tuple(
title=_("Number of email domains"),
elements=[
Integer(
title=_("Warning at"),
size=32,
default_value=100,
),
Integer(
title=_("Critical at"),
size=32,
default_value=200,
),
],
),
),
(
"levels_num_mailboxes",
Tuple(
title=_("Number of mailboxes"),
elements=[
Integer(
title=_("Warning at"),
size=32,
default_value=500,
),
Integer(
title=_("Critical at"),
size=32,
default_value=1000,
),
],
),
),
(
"levels_num_global_messages",
Tuple(
title=_("Number of messages"),
elements=[
Integer(
title=_("Warning at"),
size=32,
default_value=100000,
),
Integer(
title=_("Critical at"),
size=32,
default_value=250000,
),
],
),
),
],
)
rulespec_registry.register(
CheckParameterRulespecWithoutItem(
check_group_name="mailcow_info",
group=RulespecGroupCheckParametersApplications,
match_type="dict",
parameter_valuespec=_parameter_spec_mailcow_info,
title=lambda: _("Levels for Mailcow info"),
)
)

65
local/share/check_mk/web/plugins/wato/mailcow_mailboxes_rules.py
View File

@ -1,65 +0,0 @@
from cmk.gui.i18n import _
from cmk.gui.plugins.wato import (
CheckParameterRulespecWithItem,
rulespec_registry,
RulespecGroupCheckParametersApplications
)
from cmk.gui.valuespec import (
Dictionary,
ListChoice,
TextAscii,
Percentage,
Tuple,
Float,
Integer
)
def _item_spec_mailcow_mailboxes():
return TextAscii(
title=_("Mailbox name")
)
def _parameter_spec_mailcow_mailboxes():
return Dictionary(
elements=[
("levels_mailcow_mailboxes_quota_used", Tuple(
title=_("Mailcow mailbox quota usage for storage"),
elements=[
Percentage(
title=_("Warning at"),
default_value=65.0,
),
Percentage(
title=_("Critical at"),
default_value=85.0,
)
],
)),
("levels_mailcow_mailboxes_num_messages", Tuple(
title=_("Number of messages in mailbox"),
elements=[
Integer(
title=_("Warning at"),
size=32,
default_value=1000,
),
Integer(
title=_("Critical at"),
size=32,
default_value=2500,
)
],
)),
],
)
rulespec_registry.register(
CheckParameterRulespecWithItem(
check_group_name="mailcow_mailboxes",
group=RulespecGroupCheckParametersApplications,
match_type="dict",
item_spec=_item_spec_mailcow_mailboxes,
parameter_valuespec=_parameter_spec_mailcow_mailboxes,
title=lambda: _("Levels for Mailcow mailboxes"),
)
)

98
local/share/check_mk/web/plugins/wato/mailcow_params.py
View File

@ -1,98 +0,0 @@
from cmk.gui.i18n import _
from cmk.gui.plugins.wato.special_agents.common import (
RulespecGroupDatasourceProgramsApps,
RulespecGroupDatasourceProgramsCustom,
RulespecGroupDatasourceProgramsHardware,
RulespecGroupDatasourceProgramsOS,
RulespecGroupDatasourceProgramsTesting,
)
from cmk.gui.plugins.wato.utils import (
IndividualOrStoredPassword,
rulespec_registry,
CheckParameterRulespecWithItem,
CheckParameterRulespecWithoutItem,
HostRulespec,
Rulespec,
)
from cmk.gui.valuespec import (
Dictionary,
ListChoice,
Checkbox,
TextAscii,
Password,
NetworkPort,
Integer,
)
def _valuespec_special_agent_mailcow():
return Dictionary(
title=_("Mailcow Server Information"),
help=_("Checking Mailcow instances via API"),
elements=[
(
"hostname",
TextAscii(
title=_("Hostname"),
size=32,
allow_empty=False,
help=_("Hostname of Mailcow server (bare FQDN or IP), mandatory"),
),
),
(
"apikey",
IndividualOrStoredPassword(
title=_("API Key"),
size=32,
allow_empty=False,
help=_("API Key, mandatory"),
),
),
(
"port",
TextAscii(
title=_("Port"),
allow_empty=True,
help=_("Specify port if not listening to HTTPS/HTTP, optional"),
),
),
(
"check_version",
Checkbox(
title=_("Check version"),
help=_("Checks the running version against the public Github repo"),
),
),
(
"no_https",
Checkbox(
title=_("Disable HTTPS"),
help=_(
"Activate to disable TLS encryption (not recommended), optional"
),
),
),
(
"no_cert_check",
Checkbox(
title=_("Disable certificate validation"),
help=_(
"Activate to disable certificate validation (not recommended), optional"
),
),
),
],
optional_keys=[],
)
rulespec_registry.register(
HostRulespec(
group=RulespecGroupDatasourceProgramsApps,
name="special_agents:mailcow",
valuespec=_valuespec_special_agent_mailcow,
)
)

BIN
mailcow/agent_based/__pycache__/mailcow_domains.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
mailcow/agent_based/__pycache__/mailcow_info.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
mailcow/agent_based/__pycache__/mailcow_mailboxes.cpython-312.pyc Normal file
View File

Binary file not shown.

219
mailcow/agent_based/mailcow_domains.py Normal file
View File

@ -0,0 +1,219 @@
#!/usr/bin/env python3
# pylint: disable=too-many-locals, line-too-long, too-many-statements
"""Mailcow check for domains"""
# 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"""
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",
)

155
mailcow/agent_based/mailcow_info.py Normal file
View File

@ -0,0 +1,155 @@
#!/usr/bin/env python3
# pylint: disable=line-too-long, simplifiable-if-statement, missing-module-docstring, too-many-locals
# import necessary elements from API version 2
from cmk.agent_based.v2 import (
AgentSection,
CheckPlugin,
Service,
Result,
State,
Metric,
)
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_info(string_table) -> dict:
"""the parse function"""
parsed_data = {}
# we only expect one line with 7 entries
line = string_table[0]
version = line[0]
num_domains = int(line[1])
num_mailboxes = int(line[2])
num_global_messages = int(line[3])
git_version = line[4]
update_available = line[5]
if update_available == "True":
update_available = True
else:
update_available = False
check_version_enabled = line[6]
if check_version_enabled == "True":
check_version_enabled = True
else:
check_version_enabled = False
parsed_data["version"] = version
parsed_data["num_domains"] = num_domains
parsed_data["num_mailboxes"] = num_mailboxes
parsed_data["num_global_messages"] = num_global_messages
parsed_data["git_version"] = git_version
parsed_data["update_available"] = update_available
parsed_data["check_version_enabled"] = check_version_enabled
return parsed_data
def discover_mailcow_info(section):
"""the discover function"""
yield Service()
def check_mailcow_info(params, section):
"""the check function"""
# get thresholds
_type, levels_num_domains = params["levels_num_domains"]
_type, levels_num_mailboxes = params["levels_num_mailboxes"]
_type, levels_num_global_messages = params["levels_num_global_messages"]
# get all section data
version: str = section["version"]
git_version: str = section["git_version"]
check_version_enabled: bool = section["check_version_enabled"]
update_available: bool = section["update_available"]
num_domains: int = section["num_domains"]
num_mailboxes: int = section["num_mailboxes"]
num_global_messages: int = section["num_global_messages"]
# create graphs for number of domains, mailboxes and messages
yield Metric(name="mc_num_domains", value=num_domains, levels=levels_num_domains)
yield Metric(
name="mc_num_mailboxes", value=num_mailboxes, levels=levels_num_mailboxes
)
yield Metric(
name="mc_num_global_messages",
value=num_global_messages,
levels=levels_num_global_messages,
)
# create overall result
if check_version_enabled:
if update_available:
summary = f"Update available: Running version is {version}, Github version is: {git_version}"
state = State.WARN
else:
summary = f"No update available: Running version is {version}, Github version is: {git_version}"
state = State.OK
else:
summary = f"Version is {version}, Update check is disabled"
state = State.OK
details = f"Mailcow version: {version}\nNumber of domains: {num_domains}\nNumber of mailboxes: {num_mailboxes}\nNumber of messages: {num_global_messages}"
yield Result(state=state, summary=summary, details=details)
# Create result for number of domains
warn, crit = levels_num_domains
state = get_state_upper((warn, crit), num_domains)
notice = f"Number of domains: {num_domains}"
if state != State.OK:
yield Result(state=state, notice=notice)
# Create result for number of mailboxes
warn, crit = levels_num_mailboxes
state = get_state_upper((warn, crit), num_mailboxes)
notice = f"Number of mailboxes: {num_mailboxes}"
if state != State.OK:
yield Result(state=state, notice=notice)
# Create result for number of global messages
warn, crit = levels_num_global_messages
state = get_state_upper((warn, crit), num_global_messages)
notice = f"Number of messages: {num_global_messages}"
if state != State.OK:
yield Result(state=state, notice=notice)
# create the new agent section, must begin with "agent_section_"
# and must be an instance of "AgentSection"
agent_section_mailcow_info = AgentSection(
# "name" must exactly match the section name within the agent output
name="mailcow_info",
# 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_info,
)
# create the new check plugin, must begin with "check_plugin_"
# and must be an instance of "CheckPlugin"
check_plugin_mailcow_info = CheckPlugin(
# "name" should be the same as the corresponding section within the agent output
name="mailcow_info",
service_name="Mailcow info",
# 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_info,
# 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_info,
# define the default parameters
check_default_parameters={
"levels_num_domains": ("fixed", (100, 200)),
"levels_num_mailboxes": ("fixed", (500, 1000)),
"levels_num_global_messages": ("fixed", (100_000, 250_000)),
},
# connect to the ruleset where parameters can be defined
# must match the name of the ruleset exactly
check_ruleset_name="mailcow_info",
)

211
mailcow/agent_based/mailcow_mailboxes.py Normal file
View File

@ -0,0 +1,211 @@
#!/usr/bin/env python3
# pylint: disable=too-many-locals, line-too-long, too-many-statements, too-many-branches
"""Mailcow check for mailboxes"""
import time
# 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_mailboxes(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:
mailboxname = 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
# get display name
display_name = line[4]
# number of messages within mailbox
number_of_messages = int(line[5])
# calculate storage used for all messages in mailbox
quota = int(line[7])
total_number_of_bytes_used = int(line[8])
if quota == 0:
# quota is not set, if this is the case, line[6] contains no numeric value, but the char "-"
# so set all usage counters to zero
percent_in_use = 0
percent_storage_used_for_messages = 0
else:
# percent in use, rounded to full percent (calculated by Mailcow)
percent_in_use = int(line[6])
# let's calculate our own value
percent_storage_used_for_messages = total_number_of_bytes_used * 100 / quota
# get time of last login for IMAP/POP3/SMTP (seconds since epoch)
last_imap_login = int(line[9])
last_pop3_login = int(line[10])
last_smtp_login = int(line[11])
# transfer these times into a human friendly format
if last_imap_login == 0:
last_imap_login_data = "Never"
else:
curr_time = int(time.time())
diff_time = curr_time - last_imap_login
last_imap_login_data = render.timespan(diff_time)
if last_pop3_login == 0:
last_pop3_login_data = "Never"
else:
curr_time = int(time.time())
diff_time = curr_time - last_pop3_login
last_pop3_login_data = render.timespan(diff_time)
if last_smtp_login == 0:
last_smtp_login_data = "Never"
else:
curr_time = int(time.time())
diff_time = curr_time - last_smtp_login
last_smtp_login_data = render.timespan(diff_time)
# store all (calculated) data
parsed_data[f"{mailboxname}"] = [
active,
create_time_data,
modify_time_data,
display_name,
number_of_messages,
percent_in_use,
quota,
total_number_of_bytes_used,
percent_storage_used_for_messages,
last_imap_login_data,
last_pop3_login_data,
last_smtp_login_data,
]
return parsed_data
def discover_mailcow_mailboxes(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_mailboxes(item, params, section):
"""the check function"""
mailbox = section.get(item)
if not mailbox:
# if a previously found domain does not exist anymore, create a meaningful result
yield Result(
state=State.UNKNOWN,
summary="Mailbox not found anymore, maybe it was deleted?",
)
return
# get all infos regarding the mailbox
active = mailbox[0]
create_time = mailbox[1]
modify_time = mailbox[2]
display_name = mailbox[3]
number_of_messages = mailbox[4]
_percent_in_use = mailbox[5]
quota = mailbox[6]
total_number_of_bytes_used = mailbox[7]
percent_storage_used_for_messages = mailbox[8]
last_imap_login_data = mailbox[9]
last_pop3_login_data = mailbox[10]
last_smtp_login_data = mailbox[11]
# create (main) service for used storage (mailbox quota)
_type, levels = params["levels_mailcow_mailboxes_quota_used"]
state_quota = get_state_upper(levels, percent_storage_used_for_messages)
# create graph for used quota
yield Metric(
"mailcow_mailboxes_used_quota", percent_storage_used_for_messages, levels=levels
)
summary_quota = f"Storage quota for mailbox of '{display_name}' is {render.percent(percent_storage_used_for_messages)}"
details_quota = f"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 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)
notice = f"Last IMAP login: {last_imap_login_data} ago"
yield Result(state=State.OK, notice=notice)
notice = f"Last POP3 login: {last_pop3_login_data} ago"
yield Result(state=State.OK, notice=notice)
notice = f"Last SMTP login: {last_smtp_login_data} ago"
yield Result(state=State.OK, notice=notice)
# create service for number of messages
_type, levels = params["levels_mailcow_mailboxes_num_messages"]
state_messages = get_state_upper(levels, number_of_messages)
yield Metric("mailcow_mailboxes_messages", number_of_messages, levels=levels)
notice = f"Number of messages: {number_of_messages}"
yield Result(state=state_messages, notice=notice)
# create the new agent section, must begin with "agent_section_"
# and must be an instance of "AgentSection"
agent_section_mailcow_mailboxes = AgentSection(
# "name" must exactly match the section name within the agent output
name="mailcow_mailboxes",
# 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_mailboxes,
)
# create the new check plugin, must begin with "check_plugin_"
# and must be an instance of "CheckPlugin"
check_plugin_mailcow_mailboxes = CheckPlugin(
# "name" should be the same as the corresponding section within the agent output
name="mailcow_mailboxes",
# this is a service with item, so you have to include a place holder for the item id
service_name="Mailcow mailbox %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_mailboxes,
# 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_mailboxes,
# define the default parameters
check_default_parameters={
"levels_mailcow_mailboxes_quota_used": ("fixed", (65.0, 85.0)),
"levels_mailcow_mailboxes_num_messages": ("fixed", (1000, 2500)),
},
# connect to the ruleset where parameters can be defined
# must match the name of the ruleset exactly
check_ruleset_name="mailcow_mailboxes",
)

2
local/share/check_mk/checkman/mailcow_domains → mailcow/checkman/mailcow_domains
View File

@ -17,4 +17,4 @@ description:
item:
domainname
inventory:
one service is created for each domain
one service is created for each email domain

0
local/share/check_mk/checkman/mailcow_info → mailcow/checkman/mailcow_info
View File

0
local/share/check_mk/checkman/mailcow_mailboxes → mailcow/checkman/mailcow_mailboxes
View File

BIN
mailcow/graphing/__pycache__/graph_mailcow.cpython-312.pyc Normal file
View File

Binary file not shown.

112
mailcow/graphing/graph_mailcow.py Normal file
View File

@ -0,0 +1,112 @@
#!/usr/bin/env python3
# pylint: disable=unused-import
"""graphing and metrics definitions"""
from cmk.graphing.v1 import Title
from cmk.graphing.v1.graphs import Graph, MinimalRange
from cmk.graphing.v1.metrics import Color, DecimalNotation, Metric, Unit
from cmk.graphing.v1.perfometers import Closed, FocusRange, Perfometer
## Metrics
# metrics must begin with "metric_" and be an instance of "Metric"
# Global metrics
metric_mc_num_domains = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mc_num_domains",
title=Title("Number of email domains"),
unit=Unit(DecimalNotation("")),
color=Color.DARK_ORANGE,
)
metric_mc_num_global_messages = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mc_num_global_messages",
title=Title("Global number of messages"),
unit=Unit(DecimalNotation("")),
color=Color.DARK_CYAN,
)
metric_mc_num_mailboxes = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mc_num_mailboxes",
title=Title("Number of mailboxes"),
unit=Unit(DecimalNotation("")),
color=Color.DARK_GREEN,
)
# Mailbox Metrics
metric_mc_mailboxes_used_quota = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mailcow_mailboxes_used_quota",
title=Title("Percentage of mailbox quota used"),
unit=Unit(DecimalNotation("%")),
color=Color.DARK_ORANGE,
)
metric_mc_mailboxes_messages = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mailcow_mailboxes_messages",
title=Title("Number of messages in mailbox"),
unit=Unit(DecimalNotation("")),
color=Color.DARK_RED,
)
# Domain metrics
metric_mc_domains_used_quota = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mailcow_domains_used_quota",
title=Title("Percentage of domain quota used"),
unit=Unit(DecimalNotation("%")),
color=Color.DARK_ORANGE,
)
metric_mc_domains_mailboxes = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mailcow_domains_mailboxes",
title=Title("Percentage of maximum mailboxes used"),
unit=Unit(DecimalNotation("%")),
color=Color.DARK_GREEN,
)
metric_mc_domains_aliases = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mailcow_domains_aliases",
title=Title("Percentage of maximum aliases used"),
unit=Unit(DecimalNotation("%")),
color=Color.DARK_BLUE,
)
metric_mc_domains_messages = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mailcow_domains_messages",
title=Title("Number of messages"),
unit=Unit(DecimalNotation("")),
color=Color.DARK_RED,
)
metric_mc_domains_configured_aliases = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mailcow_domains_configured_aliases",
title=Title("Number of configured aliases"),
unit=Unit(DecimalNotation("")),
color=Color.DARK_CYAN,
)
metric_mc_domains_configured_mailboxes = Metric(
# "name" must be exactly the "metric_name" within the check function
name="mailcow_domains_configured_mailboxes",
title=Title("Number of configured mailboxes"),
unit=Unit(DecimalNotation("")),
color=Color.DARK_PINK,
)
## Perfometers
# perfometers must begin with "perfometer_" and be an instance of "Perfometer"
perfometer_mc_domains_used_quota = Perfometer(
name="mailcow_domains_used_quota",
focus_range=FocusRange(Closed(0), Closed(100)),
# "segments" must be exactly the name of the metric
segments=["mailcow_domains_used_quota"],
)
perfometer_mc_mailboxes_used_quota = Perfometer(
name="mailcow_mailboxes_used_quota",
focus_range=FocusRange(Closed(0), Closed(100)),
# "segments" must be exactly the name of the metric
segments=["mailcow_mailboxes_used_quota"],
)

400
local/share/check_mk/agents/special/agent_mailcow → mailcow/libexec/agent_mailcow
View File

@ -1,206 +1,70 @@
#!/usr/bin/env python3
# pylint: disable=line-too-long, too-many-branches, too-many-locals
# pylint: disable=bare-except, pointless-string-statement, consider-using-dict-items
# pylint: disable=too-many-arguments, too-many-statements, simplifiable-if-statement, too-many-positional-arguments
# pylint: disable=possibly-used-before-assignment, used-before-assignment
# pylint: disable=missing-class-docstring, missing-module-docstring, missing-function-docstring
"""CheckMK special agent for Mailcow systems"""
import getopt
import json
import argparse
import sys
import requests
import urllib3
import json
import os
import cmk.utils.password_store
from pprint import pprint
from requests.structures import CaseInsensitiveDict
from requests.auth import HTTPBasicAuth
from time import (
time,
localtime,
strftime,
)
def showUsage() -> None:
sys.stderr.write(
"""CheckMK Mailcow Special Agent
USAGE: agent_nextcloud -H [hostname] -k [apikey]
agent_nextcloud -h
OPTIONS:
-H, --hostname Hostname (FQDN or IP) of Nextcloud server
-k, --apikey API Key
-P, --port Port
--check-version True|False If "True": Running version will be checked against official Github repo
--no-https True|False If "True": Disable HTTPS, use HTTP (not recommended!)
--no-cert-check True|False If "True": Disable TLS certificate check (not recommended!)
-h, --help Show this help message and exit
"""
def get_args() -> argparse.Namespace:
parser: argparse.ArgumentParser = argparse.ArgumentParser(
description="Mailcow server parameters"
)
parser.add_argument(
"--hostname", required=True, type=str, help="Hostname or IP of mailcow server"
)
parser.add_argument("--apikey", required=True, type=str, help="API key")
parser.add_argument(
"--port",
required=False,
type=int,
help="Port where the server is listening on",
)
parser.add_argument(
"--no-https",
required=False,
default=False,
help="Disable HTTPS",
action="store_true",
)
parser.add_argument(
"--no-cert-check",
required=False,
default=False,
help="Disable certificate checks",
action="store_true",
)
parser.add_argument(
"--check-version",
required=False,
default=False,
help="Enable version check",
action="store_true",
)
args = parser.parse_args()
return args
# set this to True to produce debug output (this clutters the agent output)
# be aware: activating this logs very sensitive information to debug files in ~/tmp
# !!DO NOT FORGET to delete these files after debugging is done!!
MC_API_BASE: str = "api/v1/get"
TIMEOUT: int = 30
DEBUG = False
mc_api_base = "api/v1/get"
opt_hostname = ""
opt_apikey = ""
opt_port = ""
opt_check_version = False
opt_no_https = False
opt_no_cert_check = False
short_options = "hH:k:P:"
long_options = [
"hostname=",
"apikey=",
"port=",
"check-version=",
"no-https=",
"no-cert-check=",
"help",
]
domain_data = {}
mailbox_data = {}
debugLogFilename = ""
SEP = "|"
# SEP = "\t"
TIMEFMT = "%Y-%m-%d %H:%M:%S"
FLOATFMT = "{:.4f}"
"""
Decorator function for debugging purposes
creates a file with many information regarding the function call, like:
timestamp
name of function
runtime
number of arguments
number of keyword arguments
return value of function call
type of return value
all parameters given to function
"""
domain_data: dict = {}
mailbox_data: dict = {}
def debugLog(function):
def wrapper(*args, **kwargs):
# execute function and measure runtime
start = time()
value = function(*args, **kwargs)
end = time()
# get number of args and kwargs
len_args = len(args)
len_kwargs = len(kwargs)
# format the output
seconds = FLOATFMT.format(end - start)
local_time = strftime(TIMEFMT, localtime(start))
# get function name
fname = function.__name__
# create output
# out1: Timestamp|Name of Function|Runtime|Num Args|Num Kwargs|Return Value|Return Value Type
out1 = f"{local_time}{SEP}{fname}{SEP}{seconds}{SEP}{len_args}{SEP}{len_kwargs}{SEP}{value}{SEP}{type(value)}"
# out2: all arguments
out2 = ""
for arg in args:
out2 = out2 + SEP + str(arg)
# out3: all keyword arguments
out3 = ""
if len_kwargs > 0:
for key, val in kwargs.items():
out3 = out3 + SEP + key + ":" + str(val)
# write output to file
if debugLogFilename != "":
try:
with open(debugLogFilename, "a+") as f:
f.write(f"{out1}{out2}{out3}\n")
except:
sys.stderr.write(
f"Something went wrong when writing to file {debugLogFilename}\n"
)
sys.exit(1)
else:
sys.stderr.write(f"Debug activated, but no debug filename given.\n")
sys.exit(1)
return value
return wrapper
def getDebugFilename(hostname: str) -> str:
home_path = os.getenv("HOME")
tmp_path = f"{home_path}/tmp"
file_name = f"{tmp_path}/mailcow_{hostname}_debug.log"
return file_name
def getOptions() -> None:
global opt_hostname
global opt_apikey
global opt_port
global opt_check_version
global opt_no_https
global opt_no_cert_check
opts, args = getopt.getopt(sys.argv[1:], short_options, long_options)
for opt, arg in opts:
if opt in ["-H", "--hostname"]:
opt_hostname = arg
elif opt in ["-k", "--apikey"]:
opt_apikey = arg
elif opt in ["-P", "--port"]:
opt_port = arg
elif opt in ["--check-version"]:
if arg == "True":
opt_check_version = True
else:
opt_check_version = False
elif opt in ["--no-https"]:
if arg == "True":
opt_no_https = True
else:
opt_no_https = False
elif opt in ["--no-cert-check"]:
if arg == "True":
opt_no_cert_check = True
else:
opt_no_cert_check = False
elif opt in ["-h", "--help"]:
showUsage()
sys.exit(0)
if DEBUG:
home_path = os.getenv("HOME")
tmp_path = f"{home_path}/tmp"
help_file = f"{tmp_path}/mailcow_{opt_hostname}_debug.txt"
with open(help_file, "a") as file:
file.write(
f"Number of Arguments: {len(sys.argv)}, Argument List: {str(sys.argv)}\n"
)
# @debugLog
def showOptions() -> None:
print(f"Hostname: {opt_hostname}")
print(f"Username: {opt_apikey}")
print(f"Port: {opt_port}")
print(f"Check Version: {opt_check_version}")
print(f"No HTTPS: {opt_no_https}")
print(f"No TLS Check: {opt_no_cert_check}")
home_path = os.getenv("HOME")
tmp_path = f"{home_path}/tmp"
help_file = f"{tmp_path}/mailcow_{opt_hostname}_debug.txt"
with open(help_file, "a") as file:
file.write(
f"Hostname: {opt_hostname}, Port: {opt_port}, No HTTPS: {opt_no_https}, No Cert Check: {opt_no_cert_check}\n"
)
# @debugLog
def getDomainInfo(headers: str, verify: bool, base_url: str) -> int:
url = f"{base_url}/domain/all"
response = requests.get(url, headers=headers, verify=verify)
def get_domain_info(headers: str, verify: bool, base_url: str) -> int:
"""retrieves info about all configured domains"""
url: str = f"{base_url}/domain/all"
response = requests.get(url=url, headers=headers, verify=verify, timeout=TIMEOUT)
if response.status_code == 200:
jsdata = response.text
data = json.loads(jsdata) # returns a list of dictionaries
@ -243,22 +107,21 @@ def getDomainInfo(headers: str, verify: bool, base_url: str) -> int:
i += 1
# return number of email domains
return i
else:
sys.stderr.write(
f"Request response code is {response.status_code} with URL {url}\n"
)
sys.exit(1)
sys.stderr.write(
f"Request response code is {response.status_code} with URL {url}\n"
)
sys.exit(1)
# @debugLog
def getMailboxInfo(headers: str, verify: bool, base_url: str) -> tuple:
url = f"{base_url}/mailbox/all"
response = requests.get(url, headers=headers, verify=verify)
def get_mailbox_info(headers: str, verify: bool, base_url: str) -> tuple:
"""retrieves info about all configured mailboxes"""
url: str = f"{base_url}/mailbox/all"
response = requests.get(url=url, headers=headers, verify=verify, timeout=TIMEOUT)
if response.status_code == 200:
jsdata = response.text
data = json.loads(jsdata) # returns a list of dictionaries
i = 0
global_num_messages = 0
i: int = 0
global_num_messages: int = 0
while i < len(data):
# get status of mailbox (0-->inactive or 1-->active)
active = data[i].get("active")
@ -282,7 +145,7 @@ def getMailboxInfo(headers: str, verify: bool, base_url: str) -> tuple:
last_imap_login = data[i].get("last_imap_login")
last_pop3_login = data[i].get("last_pop3_login")
last_smtp_login = data[i].get("last_smtp_login")
mb = {
mailbox = {
"active": active,
"created": created,
"modified": modified,
@ -297,27 +160,26 @@ def getMailboxInfo(headers: str, verify: bool, base_url: str) -> tuple:
"last_smtp_login": last_smtp_login,
}
mailbox_data[username] = {}
mailbox_data[username] = mb
mailbox_data[username] = mailbox
i += 1
global_num_messages += num_messages
# return number of mailboxes and global number of messages
return i, global_num_messages
else:
sys.stderr.write(
f"Request response code is {response.status_code} with URL {url}\n"
)
sys.exit(1)
sys.stderr.write(
f"Request response code is {response.status_code} with URL {url}\n"
)
sys.exit(1)
def getGitVersion() -> str:
url = "https://api.github.com/repos/mailcow/mailcow-dockerized/releases/latest"
git_version = ""
response = requests.get(url)
def get_git_version() -> str:
"""gets the actual version of Mailcow from Github"""
url: str = "https://api.github.com/repos/mailcow/mailcow-dockerized/releases/latest"
git_version: str = ""
response = requests.get(url=url, timeout=TIMEOUT)
if response.status_code == 200:
jsdata = response.text
data = json.loads(jsdata) # returns a dictionary
git_version = data["tag_name"]
# print(git_version)
else:
sys.stderr.write(
f"Request response code is {response.status_code} with URL {url}\n"
@ -326,8 +188,9 @@ def getGitVersion() -> str:
return git_version
def compareVersions(mc_ver: str, git_ver: str) -> bool:
update_available = False
def compare_versions(mc_ver: str, git_ver: str) -> bool:
"""compares the actual Github version with the version of the running Mailcow system"""
update_available: bool = False
try:
mc_year, mc_month = mc_ver.split("-")
git_year, git_month = git_ver.split("-")
@ -360,15 +223,18 @@ def compareVersions(mc_ver: str, git_ver: str) -> bool:
elif git_month_ver > mc_month_ver:
update_available = True
except:
# TBD
pass
return update_available
# @debugLog
def getMailcowInfo(headers: str, verify: bool, base_url: str) -> dict:
def get_mailcow_info(
headers: str, verify: bool, base_url: str, check_version: bool
) -> dict:
"""retrieves several global information about the mailcow system"""
info_data = {}
url = f"{base_url}/status/version"
response = requests.get(url, headers=headers, verify=verify)
response = requests.get(url=url, headers=headers, verify=verify, timeout=TIMEOUT)
if response.status_code == 200:
jsdata = response.text
data = json.loads(jsdata) # returns a dictionary
@ -376,24 +242,21 @@ def getMailcowInfo(headers: str, verify: bool, base_url: str) -> dict:
# get Mailcow version
mc_version = data["version"]
# if enabled, check this version against the official Mailcow repo on Github
if opt_check_version:
git_version = getGitVersion()
update_available = compareVersions(mc_version, git_version)
# update_available = compareVersions("2023-01a", "2023-02")
# print(update_available)
if check_version:
git_version = get_git_version()
update_available = compare_versions(mc_version, git_version)
info_data["git_version"] = git_version
info_data["update_available"] = update_available
else:
info_data["git_version"] = "Version check disabled"
info_data["update_available"] = False
info_data["mc_version"] = mc_version
info_data["check_version_enabled"] = opt_check_version
info_data["check_version_enabled"] = check_version
return info_data
else:
sys.stderr.write(
f"Request response code is {response.status_code} with URL {url}\n"
)
sys.exit(1)
sys.stderr.write(
f"Request response code is {response.status_code} with URL {url}\n"
)
sys.exit(1)
"""
@ -418,8 +281,8 @@ user1@dom2.de;1;2022-04-30 09:55:37;2022-04-30 09:55:37;Melissa;53460;33;1996488
"""
# @debugLog
def doCmkOutputMailboxes() -> None:
def do_cmk_output_mailboxes() -> None:
"""prints out agent section for mailboxes"""
print("<<<mailcow_mailboxes:sep(59)>>>")
for mb in mailbox_data:
active = mailbox_data[mb]["active"]
@ -440,8 +303,7 @@ def doCmkOutputMailboxes() -> None:
)
# @debugLog
def doCmkOutputMailcow(
def do_cmk_output_mailcow(
version: str,
num_domains: int,
num_mailboxes: int,
@ -450,6 +312,7 @@ def doCmkOutputMailcow(
update_available: bool,
check_version_enabled: bool,
) -> None:
"""prints out agent section for global mailcow info"""
print("<<<mailcow_info:sep(59)>>>")
# strip semicolons, if present, since we use it as delimiter
version = version.replace(";", "")
@ -460,11 +323,11 @@ def doCmkOutputMailcow(
"""
Output is as follows:
0 domain_name
0 domain name
1 active 1 --> active, 0 --> not active
2 creation date "None" if ???
3 last modified date "None" if never modified
4 max number mailboxes
4 max number of mailboxes
5 number of mailboxes
6 max number of aliases
7 number of aliases
@ -479,8 +342,8 @@ dom3.de;1;2022-04-29 13:36:08;2022-04-29 21:26:19;10;1;100;3;28132;12852485367;2
"""
# @debugLog
def doCmkOutputDomains() -> None:
def do_cmk_output_domains() -> None:
"""print out agent section for domains"""
print("<<<mailcow_domains:sep(59)>>>")
for dom in domain_data:
active = domain_data[dom]["active"]
@ -499,74 +362,77 @@ def doCmkOutputDomains() -> None:
def main():
global debugLogFilename
cmk.utils.password_store.replace_passwords()
getOptions()
"""main function"""
# get and check parameters
params: argparse.Namespace = get_args()
# do some parameter checks
if opt_hostname == "":
sys.stderr.write(f"No hostname given.\n")
showUsage()
if params.hostname is None:
sys.stderr.write("No hostname given.\n")
sys.exit(1)
else:
hostname = opt_hostname
if opt_apikey == "":
sys.stderr.write(f"No API key given.\n")
showUsage()
hostname = params.hostname
if params.apikey is None:
sys.stderr.write("No API key given.\n")
sys.exit(1)
if opt_no_cert_check:
else:
apikey = params.apikey
if params.no_cert_check:
# disable certificate warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
verify = False
else:
verify = True
if opt_port == "":
if opt_no_https:
if params.check_version:
check_version = True
else:
check_version = False
if params.port is None:
if params.no_https:
protocol = "http"
port = "80"
else:
protocol = "https"
port = "443"
else:
if opt_no_https:
if params.no_https:
protocol = "http"
else:
protocol = "https"
port = opt_port
port = params.port
if protocol == "http" and port == "443":
sys.stderr.write(f"Combining HTTP with port 443 is not supported.\n")
sys.stderr.write("Combining HTTP with port 443 is not supported.\n")
sys.exit(1)
if protocol == "https" and port == "80":
sys.stderr.write(f"Combining HTTPS with port 80 is not supported.\n")
sys.stderr.write("Combining HTTPS with port 80 is not supported.\n")
sys.exit(1)
headers = CaseInsensitiveDict()
headers["Accept"] = "application/json"
headers["X-API-Key"] = opt_apikey
headers["X-API-Key"] = apikey
# now "hostname" contains the FQDN of the host running Mailcow
# now "protocol" is http or https
# now "port" contains the port number to use
# now "verify" signals whether certificate checking will be done (True) or not (False)
# now "check_version" signals whether the Mailcow version will be checked against the Github version
# now "headers" contains the accepted format (JSON) and the API key to use
debugLogFilename = getDebugFilename(hostname)
if DEBUG:
showOptions()
print(
f"hostname: {hostname}, protocol: {protocol}, port: {port}, verify: {verify}"
)
base_url = f"{protocol}://{hostname}:{port}/{mc_api_base}"
base_url = f"{protocol}://{hostname}:{port}/{MC_API_BASE}"
# get domain data
num_domains = getDomainInfo(headers, verify, base_url)
num_domains = get_domain_info(headers=headers, verify=verify, base_url=base_url)
# get mailbox data
num_mailboxes, num_global_messages = getMailboxInfo(headers, verify, base_url)
num_mailboxes, num_global_messages = get_mailbox_info(
headers=headers, verify=verify, base_url=base_url
)
# get global Mailcow info
mailcow_info = getMailcowInfo(headers, verify, base_url)
mailcow_info = get_mailcow_info(
headers=headers, verify=verify, base_url=base_url, check_version=check_version
)
mailcow_version = mailcow_info["mc_version"]
github_version = mailcow_info["git_version"]
update_available = mailcow_info["update_available"]
check_version_enabled = mailcow_info["check_version_enabled"]
# create agent output
doCmkOutputDomains()
doCmkOutputMailboxes()
doCmkOutputMailcow(
do_cmk_output_domains()
do_cmk_output_mailboxes()
do_cmk_output_mailcow(
mailcow_version,
num_domains,
num_mailboxes,

BIN
mailcow/rulesets/__pycache__/rs_mailcow_domains.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
mailcow/rulesets/__pycache__/rs_mailcow_info.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
mailcow/rulesets/__pycache__/rs_mailcow_mailboxes.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
mailcow/rulesets/__pycache__/rs_mailcow_params.cpython-312.pyc Normal file
View File

Binary file not shown.

93
mailcow/rulesets/rs_mailcow_domains.py Normal file
View File

@ -0,0 +1,93 @@
#!/user/bin/env python3
"""parameter form ruleset for mailcow domains"""
from cmk.rulesets.v1 import Title
from cmk.rulesets.v1.form_specs import (
DefaultValue,
DictElement,
Dictionary,
LevelDirection,
SimpleLevels,
Percentage,
Integer,
)
from cmk.rulesets.v1.rule_specs import CheckParameters, HostAndItemCondition, Topic
# function name should begin with an underscore to limit it's visibility
def _parameter_form():
return Dictionary(
elements={
"levels_mailcow_domains_quota_used": DictElement(
parameter_form=SimpleLevels(
title=Title("Mailcow domains quota usage for storage"),
form_spec_template=Percentage(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(65.0, 85.0)),
),
required=True,
),
"levels_mailcow_domains_mailboxes_used": DictElement(
parameter_form=SimpleLevels(
title=Title("Mailcow domains mailboxes usage"),
form_spec_template=Percentage(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(65.0, 85.0)),
),
required=True,
),
"levels_mailcow_domains_aliases_used": DictElement(
parameter_form=SimpleLevels(
title=Title("Mailcow domains aliases usage"),
form_spec_template=Percentage(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(65.0, 85.0)),
),
required=True,
),
"levels_mailcow_domains_num_messages": DictElement(
parameter_form=SimpleLevels(
title=Title("Number of messages"),
form_spec_template=Integer(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(10_000, 25_000)),
),
required=True,
),
"levels_mailcow_domains_num_aliases": DictElement(
parameter_form=SimpleLevels(
title=Title("Number of configured aliases"),
form_spec_template=Integer(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(100, 250)),
),
required=True,
),
"levels_mailcow_domains_num_mailboxes": DictElement(
parameter_form=SimpleLevels(
title=Title("Number of configured mailboxes"),
form_spec_template=Integer(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(100, 250)),
),
required=True,
),
}
)
# name must begin with "rule_spec_", should refer to the used check plugin
# must be an instance of "CheckParameters"
rule_spec_mailcow_domains = CheckParameters(
# "name" should be the same as the check plugin
name="mailcow_domains",
# the title is shown in the GUI
title=Title("Mailcow domain levels"),
# this ruleset can be found under Setup|Service monitoring rules|Various
topic=Topic.APPLICATIONS,
# define the name of the function which creates the GUI elements
parameter_form=_parameter_form,
# define the label in the GUI where you can restrict the
# settings to one or mor specific users (item)
condition=HostAndItemCondition(item_title=Title("Domain")),
)

65
mailcow/rulesets/rs_mailcow_info.py Normal file
View File

@ -0,0 +1,65 @@
#!/usr/bin/env python3
"""
defines the form for typing in all needed levels
regarding number of domains, mailboxes and global message count
"""
from cmk.rulesets.v1 import Help, Title
from cmk.rulesets.v1.form_specs import (
DictElement,
Dictionary,
SimpleLevels,
LevelDirection,
DefaultValue,
Integer,
)
from cmk.rulesets.v1.rule_specs import (
CheckParameters,
Topic,
HostCondition,
)
def _parameter_form() -> Dictionary:
return Dictionary(
title=Title("Levels for global Mailcow information"),
help_text=Help("Checking Mailcow systems via API"),
elements={
"levels_num_domains": DictElement(
parameter_form=SimpleLevels(
title=Title("Levels for number of email domains"),
form_spec_template=Integer(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(100, 200)),
),
required=True,
),
"levels_num_mailboxes": DictElement(
parameter_form=SimpleLevels(
title=Title("Levels for number of mailboxes"),
form_spec_template=Integer(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(500, 1000)),
),
required=True,
),
"levels_num_global_messages": DictElement(
parameter_form=SimpleLevels(
title=Title("Levels for number of messages"),
form_spec_template=Integer(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(100_000, 250_000)),
),
required=True,
),
},
)
rule_spec_mailcow_info = CheckParameters(
name="mailcow_info",
title=Title("Mailcow global levels"),
topic=Topic.APPLICATIONS,
parameter_form=_parameter_form,
condition=HostCondition(),
)

57
mailcow/rulesets/rs_mailcow_mailboxes.py Normal file
View File

@ -0,0 +1,57 @@
#!/user/bin/env python3
"""parameter form ruleset for mailcow domains"""
from cmk.rulesets.v1 import Title
from cmk.rulesets.v1.form_specs import (
DefaultValue,
DictElement,
Dictionary,
LevelDirection,
SimpleLevels,
Percentage,
Integer,
)
from cmk.rulesets.v1.rule_specs import CheckParameters, HostAndItemCondition, Topic
# function name should begin with an underscore to limit it's visibility
def _parameter_form():
return Dictionary(
elements={
"levels_mailcow_mailboxes_quota_used": DictElement(
parameter_form=SimpleLevels(
title=Title("Mailcow mailbox quota usage for storage"),
form_spec_template=Percentage(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(65.0, 85.0)),
),
required=True,
),
"levels_mailcow_mailboxes_num_messages": DictElement(
parameter_form=SimpleLevels(
title=Title("Number of messages in mailbox"),
form_spec_template=Integer(),
level_direction=LevelDirection.UPPER,
prefill_fixed_levels=DefaultValue(value=(1000, 2500)),
),
required=True,
),
}
)
# name must begin with "rule_spec_", should refer to the used check plugin
# must be an instance of "CheckParameters"
rule_spec_mailcow_mailboxes = CheckParameters(
# "name" should be the same as the check plugin
name="mailcow_mailboxes",
# the title is shown in the GUI
title=Title("Mailcow mailbox levels"),
# this ruleset can be found under Setup|Service monitoring rules|Various
topic=Topic.APPLICATIONS,
# define the name of the function which creates the GUI elements
parameter_form=_parameter_form,
# define the label in the GUI where you can restrict the
# settings to one or mor specific users (item)
condition=HostAndItemCondition(item_title=Title("Mailbox")),
)

98
mailcow/rulesets/rs_mailcow_params.py Normal file
View File

@ -0,0 +1,98 @@
#!/usr/bin/env python3
# pylint: disable=line-too-long
"""defines the form for typing in all needed HAL9002 parameters"""
from cmk.rulesets.v1 import Help, Title, Label
from cmk.rulesets.v1.form_specs import (
DictElement,
Dictionary,
String,
validators,
BooleanChoice,
Integer,
Password,
migrate_to_password,
InputHint,
DefaultValue,
)
from cmk.rulesets.v1.rule_specs import SpecialAgent, Topic
def _form_spec_special_agent_mailcow() -> Dictionary:
return Dictionary(
title=Title("Mailcow Server Information"),
help_text=Help("Checking Mailcow systems via API"),
elements={
"hostname": DictElement(
required=True,
parameter_form=String(
title=Title("Hostname"),
help_text=Help(
"Hostname of Mailcow server (bare FQDN or IP), mandatory, eg. mailcow.yourdomain.tld"
),
custom_validate=(validators.LengthInRange(min_value=1),),
prefill=InputHint("mailcow.yourdomain.tld"),
),
),
"apikey": DictElement(
required=True,
parameter_form=Password(
title=Title("API key"),
help_text=Help("Specify API key, mandatory"),
custom_validate=(validators.LengthInRange(min_value=1),),
migrate=migrate_to_password,
),
),
"port": DictElement(
required=False,
parameter_form=Integer(
title=Title("Port"),
help_text=Help(
"Specify port the Mailcow system ist listening on, mandatory"
),
prefill=DefaultValue(443),
custom_validate=(validators.NetworkPort(),),
),
),
"check_version": DictElement(
required=False,
parameter_form=BooleanChoice(
title=Title("Check Mailcow version"),
help_text=Help(
"Activate to check version of running Mailcow against actual Github version, optional"
),
label=Label(
"Enable version check (requires access to Github internet site)"
),
),
),
"no_https": DictElement(
required=False,
parameter_form=BooleanChoice(
title=Title("Disable HTTPS"),
help_text=Help(
"Activate to disable encryption (not recommended), optional"
),
label=Label("Disable HTTPS"),
),
),
"no_cert_check": DictElement(
required=False,
parameter_form=BooleanChoice(
title=Title("Disable certificate validation"),
help_text=Help(
"Activate to disable certificate validation (not recommended), optional"
),
label=Label("Disable certificate validation"),
),
),
},
)
rule_spec_mailcow = SpecialAgent(
name="mailcow",
title=Title("Mailcow connection parameters"),
topic=Topic.APPLICATIONS,
parameter_form=_form_spec_special_agent_mailcow,
)

BIN
mailcow/server_side_calls/__pycache__/agent_mailcow.cpython-312.pyc Normal file
View File

Binary file not shown.

48
mailcow/server_side_calls/agent_mailcow.py Normal file
View File

@ -0,0 +1,48 @@
#!/usr/bin/env python3
# pylint: disable=missing-class-docstring, missing-module-docstring, missing-function-docstring
from collections.abc import Iterable
from pydantic import BaseModel
from cmk.server_side_calls.v1 import (
HostConfig,
Secret,
SpecialAgentCommand,
SpecialAgentConfig,
)
class MailcowParams(BaseModel):
hostname: str | None = None
apikey: Secret | None = None
port: int | None = None
no_https: bool = False
no_cert_check: bool = False
check_version: bool = False
def agent_mailcow_arguments(
params: MailcowParams, _host_config: HostConfig
) -> Iterable[SpecialAgentCommand]:
command_arguments: list[str | Secret] = []
if params.hostname is not None:
command_arguments += ["--hostname", params.hostname]
if params.apikey is not None:
command_arguments += ["--apikey", params.apikey.unsafe()]
if params.port is not None:
command_arguments += ["--port", str(params.port)]
if params.no_https:
command_arguments.append("--no-https")
if params.no_cert_check:
command_arguments.append("--no-cert-check")
if params.check_version:
command_arguments.append("--check-version")
yield SpecialAgentCommand(command_arguments=command_arguments)
special_agent_mailcow = SpecialAgentConfig(
name="mailcow",
parameter_parser=MailcowParams.model_validate,
commands_function=agent_mailcow_arguments,
)

BIN
mkp/Mailcow-2.1.0.mkp Normal file
View File

Binary file not shown.