#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) Forvia <https://www.forvia.com/>
# Developer : Jan KUCERA <jan.kucera@external.forvia.com>
from pathlib import Path
from typing import Any
from .bakery_api.v1 import FileGenerator, OS, Plugin, register
def get_vss_shadow_storage_files(conf: Any) -> FileGenerator:
yield Plugin(base_os=OS.WINDOWS,
source=Path("vss_shadow_storage.cmd"),
interval=conf.get("interval"))
register.bakery_plugin(
name="vss_shadow_storage",
files_function=get_vss_shadow_storage_files,
)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) Forvia <https://www.forvia.com/>
# Developer : Jan KUCERA <jan.kucera@external.forvia.com>
import re
from dataclasses import dataclass
from typing import Any, List, Mapping
from cmk.base.plugins.agent_based.agent_based_api.v1 import (
register,
Result,
State,
Service,
)
from .agent_based_api.v1.type_defs import CheckResult, DiscoveryResult, StringTable
@dataclass(frozen=True)
class VSSVolume:
volume: str
used_space: int
allocated_space: int
maximum_space: int
used_percentage: int
allocated_percentage: int
def convert_to_bytes(size: str) -> int:
if size == "UNBOUNDED":
return -1
number, unit = size.split()
number = float(number)
units = {"KB": 1, "MB": 2, "GB": 3, "TB": 4}
return int(number * 1024 ** units.get(unit, 0))
def parse_vss_data(string_table: StringTable) -> List[VSSVolume]:
volumes = []
for i in range(2, len(string_table), 6):
if i + 5 >= len(string_table):
break
volume_line = " ".join(string_table[i + 1])
used_line = " ".join(string_table[i + 3])
allocated_line = " ".join(string_table[i + 4])
maximum_line = " ".join(string_table[i + 5])
volume = volume_line.split("(")[1].split(")")[0]
used_space, used_percentage = used_line.split(": ")[1].split(" (")
allocated_space, allocated_percentage = allocated_line.split(": ")[1].split(" (")
maximum_space = maximum_line.split(": ")[1].split(" (")[0]
volumes.append(VSSVolume(
volume=volume,
used_space=convert_to_bytes(used_space),
allocated_space=convert_to_bytes(allocated_space),
maximum_space=convert_to_bytes(maximum_space),
used_percentage=int(used_percentage[:-2]),
allocated_percentage=int(allocated_percentage[:-2]),
))
return volumes
def discover_vss_volumes(params: Mapping[str, Any], section: List[VSSVolume]) -> DiscoveryResult:
included_regex = re.compile(params.get('vss_volume_regex', '.*'))
for volume in section:
if included_regex.match(volume.volume):
yield Service(item=volume.volume)
def check_vss_volumes(item: str, params: Mapping[str, Any], section: List[VSSVolume]) -> CheckResult:
threshold_allocated_lower = params.get('threshold_allocated_lower', (0,))[0]
threshold_allocated_upper = params.get('threshold_allocated_upper', (100,))[0]
threshold_maximum_lower = params.get('threshold_maximum_lower', (0,))[0]
threshold_maximum_upper = params.get('threshold_maximum_upper', (100,))[0]
unbounded_handling = params.get('unbounded_handling', 'auto-detect')
for volume in section:
if volume.volume == item:
summary = f"Used: {volume.used_space} bytes ({volume.used_percentage}%), Allocated: {volume.allocated_space} bytes ({volume.allocated_percentage}%), Maximum: {'UNBOUNDED' if volume.maximum_space == -1 else f'{volume.maximum_space} bytes'}"
if volume.maximum_space == -1:
yield Result(
state=State.UNKNOWN if unbounded_handling == 'unset' else State.CRIT,
summary=summary,
details=f"Maximum storage is UNBOUNDED and {unbounded_handling}"
)
else:
yield Result(state=State.OK, summary=summary, details="Maximum storage is BOUNDED")
if volume.allocated_percentage < threshold_allocated_lower:
yield Result(state=State.CRIT,
summary=f"Allocated space is below lower threshold: {volume.allocated_percentage}%")
elif volume.allocated_percentage > threshold_allocated_upper:
yield Result(state=State.CRIT,
summary=f"Allocated space is above upper threshold: {volume.allocated_percentage}%")
if volume.maximum_space != -1:
if volume.maximum_space < threshold_maximum_lower:
yield Result(state=State.CRIT,
summary=f"Maximum space is below lower threshold: {volume.maximum_space} bytes")
elif volume.maximum_space > threshold_maximum_upper:
yield Result(state=State.CRIT,
summary=f"Maximum space is above upper threshold: {volume.maximum_space} bytes")
yield Result(state=State.OK, summary=summary)
return
yield Result(state=State.UNKNOWN, summary=f"Volume {item} not found")
register.agent_section(
name="vss_shadow_storage",
parse_function=parse_vss_data,
)
register.check_plugin(
name="vss_shadow_storage",
sections=["vss_shadow_storage"],
service_name="VSS Volume %s",
discovery_function=discover_vss_volumes,
check_function=check_vss_volumes,
discovery_default_parameters={"vss_volume_regex": ".*"},
check_default_parameters={
"threshold_allocated_upper": (100,),
"threshold_allocated_lower": (0,),
"threshold_maximum_upper": (100,),
"threshold_maximum_lower": (0,),
"unbounded_handling": "set"
},
discovery_ruleset_name="vss_shadow_storage",
)
#!/usr/bin/env python3
# -*- encoding: utf-8; py-indent-offset: 4 -*-
# Copyright (C) Forvia <https://www.forvia.com/>
# Developer : Jan KUCERA <jan.kucera@external.forvia.com>
from cmk.gui.i18n import _
from cmk.gui.plugins.metrics import metric_info
metric_info["vss_used_space"] = {
"title": _("Used Shadow Copy Storage space"),
"unit": "bytes",
"color": "11/a",
}
metric_info["vss_allocated_space"] = {
"title": _("Allocated Shadow Copy Storage space"),
"unit": "bytes",
"color": "22/a",
}
metric_info["vss_maximum_space"] = {
"title": _("Maximum Shadow Copy Storage space"),
"unit": "bytes",
"color": "33/a",
}
metric_info["vss_used_percentage"] = {
"title": _("Used Shadow Copy Storage space percentage"),
"unit": "%",
"color": "44/a",
}
metric_info["vss_allocated_percentage"] = {
"title": _("Allocated Shadow Copy Storage space percentage"),
"unit": "%",
"color": "55/a",
}
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) Forvia <https://www.forvia.com/>
# Developer : Jan KUCERA <jan.kucera@external.forvia.com>
from cmk.gui.cee.plugins.wato.agent_bakery.rulespecs.utils import RulespecGroupMonitoringAgentsAgentPlugins
from cmk.gui.i18n import _
from cmk.gui.plugins.wato import (
rulespec_registry,
HostRulespec,
)
from cmk.gui.valuespec import (
Dictionary,
DropdownChoice,
Age
)
def _valuespec_agent_config_vss_volumes():
return Dictionary(
title=_("Shadow Copy Storage Monitoring (Windows)"),
help=_("This will deploy the agent plugin <tt>vssadmin_volumes.cmd</tt> for monitoring."),
elements=[("activated",
DropdownChoice(title=_("Windows VSS"),
choices=[
(True, _("Deploy plugin")),
(None, _("Do not deploy plugin")),
])),
(
"interval",
Age(
title=_("Run asynchronously"),
label=_("Interval for collecting data"),
default_value=3600,
),
),
]
)
rulespec_registry.register(
HostRulespec(
group=RulespecGroupMonitoringAgentsAgentPlugins,
name="agent_config:vss_shadow_storage",
valuespec=_valuespec_agent_config_vss_volumes,
))
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) Forvia <https://www.forvia.com/>
# Developer : Jan KUCERA <jan.kucera@external.forvia.com>
from cmk.gui.i18n import _
from cmk.gui.plugins.wato import (
CheckParameterRulespecWithItem,
rulespec_registry,
RulespecGroupCheckParametersApplications,
HostRulespec,
RulespecGroupCheckParametersDiscovery
)
from cmk.gui.valuespec import (
Dictionary,
Integer,
TextAscii,
Tuple,
TextInput
)
def _parameter_valuespec_vss_volumes():
return Dictionary(elements=[
("threshold_allocated_upper",
Tuple(
title=_("Upper threshold for allocated space percentage"),
elements=[
Integer(title=_("Warning at"), unit="%"),
Integer(title=_("Critical at"), unit="%"),
],
)),
("threshold_allocated_lower",
Tuple(
title=_("Lower threshold for allocated space percentage"),
elements=[
Integer(title=_("Warning below"), unit="%"),
Integer(title=_("Critical below"), unit="%"),
],
)),
("threshold_maximum_upper",
Tuple(
title=_("Upper threshold for maximum space"),
elements=[
Integer(title=_("Warning at"), unit="bytes"),
Integer(title=_("Critical at"), unit="bytes"),
],
)),
("threshold_maximum_lower",
Tuple(
title=_("Lower threshold for maximum space"),
elements=[
Integer(title=_("Warning below"), unit="bytes"),
Integer(title=_("Critical below"), unit="bytes"),
],
)),
("unbounded_handling",
TextAscii(
title=_("Handling of unbounded storage"),
help=_("Specify how to handle unbounded storage: 'unset', 'set', or 'auto-detect'.")
)),
], )
rulespec_registry.register(
CheckParameterRulespecWithItem(
check_group_name="vss_shadow_storage",
group=RulespecGroupCheckParametersApplications,
item_spec=lambda: TextAscii(title=_("Volume name")),
match_type="dict",
parameter_valuespec=_parameter_valuespec_vss_volumes,
title=lambda: _("VSS Shadow Storage"),
))
def _valuespec_vss_shadow_storage_inventory():
return Dictionary(
title=_("VSS shadow storage discovery"),
help=_("This selects which VSS volumes are discovered."),
elements=[
(
"vss_volume_regex",
TextInput(
title=_("Filter relevant VSS volumes by regex"),
help=_(
" You can provide filtering against a specific regex you are"
" interested in. This is to only catch volumes of interest."
),
),
),
],
)
rulespec_registry.register(
HostRulespec(
group=RulespecGroupCheckParametersDiscovery,
name="vss_shadow_storage_inventory",
valuespec=_valuespec_vss_shadow_storage_inventory,
)
)