cmk.agent_based.v2

New in this version

This section lists the most important changes you have to be aware of when migrating your plug-in to this API version.

Note that changes are expressed in relation to the API version 1.

You can find a script in doc/treasures/migration_helpers/ that will do most of the migration for you. This script is not officially supported, so use at your own risk.

type_defs module is dissolved

The type_defs module has been dissolved. All types are now directly imported from the v2 module.

check_levels signature changed

The new check_levels() function is designed to work well with the levels elements from the new rulesets API v1. These can be found in cmk.rulesets.v1.form_specs.

The types of the arguments have been added to the API and can be found in cmk.agent_based.v2.

Registration is replaced by a discovery approach

This is the main reason for the introduction of this new API version. Plugins are no longer registered during import, but only created and picked up later by the backend. To realize this, we introduced four new classes:

AgentSection replacing register.agent_section()
SimpleSNMPSection and SNMPSection replacing register.snmp_section()
CheckPlugin replacing register.check_plugin()
InventoryPlugin replacing register.inventory_plugin()

The arguments of these have barely changed (see next paragraph), resulting in easy to automate changes (see this commit for instance).

Changed arguments and validation for Agent and SNMP sections

We slightly adopted the arguments to the above-mentioned Section classes. We now favor type annotations over runtime validation. To get slightly easier type annotations, parse_function is no longer optional.

Removed wrapper for regex creation `regex()`

This has been removed. See its documentation for the reasoning. You can use pythons re.compile() as drop-in replacement.

Added rendering function `time_offset()`

On popular demand we add a function to render a number of seconds that might be negative.

entry_point_prefixes()

Return the types of plug-ins and their respective prefixes that can be discovered by Checkmk.

These types can be used to create plug-ins that can be discovered by Checkmk. To be discovered, the plug-in must be of one of the types returned by this function and its name must start with the corresponding prefix.

Return type:: Mapping[type[Union[AgentSection[Any], CheckPlugin, InventoryPlugin, SimpleSNMPSection[Any, Any], SNMPSection[Any, Any]]], str]

Example:

>>> for plugin_type, prefix in entry_point_prefixes().items():
...     print(f'{prefix}... = {plugin_type.__name__}(...)')
snmp_section_... = SimpleSNMPSection(...)
snmp_section_... = SNMPSection(...)
agent_section_... = AgentSection(...)
check_plugin_... = CheckPlugin(...)
inventory_plugin_... = InventoryPlugin(...)

class AgentSection(*, name: str, parse_function: Callable[[List[List[str]]], _Section | None], host_label_function: Callable[[_Section], Generator[HostLabel, None, None]] | None = None, host_label_default_parameters: None = None, host_label_ruleset_name: None = None, host_label_ruleset_type: RuleSetType = RuleSetType.MERGED, parsed_section_name: str | None = None, supersedes: list[str] | None = None)

class AgentSection(*, name: str, parse_function: Callable[[List[List[str]]], _Section | None], host_label_function: Callable[[Mapping[str, object], _Section], Generator[HostLabel, None, None]], host_label_default_parameters: Mapping[str, object], host_label_ruleset_name: str, host_label_ruleset_type: Literal[RuleSetType.MERGED] = RuleSetType.MERGED, parsed_section_name: str | None = None, supersedes: list[str] | None = None)

class AgentSection(*, name: str, parse_function: Callable[[List[List[str]]], _Section | None], host_label_function: Callable[[Sequence[Mapping[str, object]], _Section], Generator[HostLabel, None, None]], host_label_default_parameters: Mapping[str, object], host_label_ruleset_name: str, host_label_ruleset_type: Literal[RuleSetType.ALL], parsed_section_name: str | None = None, supersedes: list[str] | None = None)

An AgentSection to plug into Checkmk

Instances of this class will only be picked up by Checkmk if their names start with agent_section_.

The section marked by ‘<<<name>>>’ in the raw agent output will be processed according to the functions and options given to this function:

Parameters:

name (str) – The unique name of the section to be registered. It must match the section header of the agent output (‘<<<name>>>’).
parse_function (Callable[[List[List[str]]], Optional[TypeVar(_Section, bound= object)]]) – The function responsible for parsing the raw agent data. It must accept exactly one argument by the name ‘string_table’. It may return an arbitrary object. Note that if the return value is None, no further processing will take place (just as if the agent had not sent any data). This function may raise arbitrary exceptions, which will be dealt with by the checking engine. You should expect well formatted data.
parsed_section_name (str | None) – The name under which the parsed section will be available to the plug-ins. Defaults to the original name.
host_label_function (Callable[[TypeVar(_Section, bound= object)], Generator[HostLabel, None, None]] | Callable[[Mapping[str, object], TypeVar(_Section, bound= object)], Generator[HostLabel, None, None]] | Callable[[Sequence[Mapping[str, object]], TypeVar(_Section, bound= object)], Generator[HostLabel, None, None]] | None) – The function responsible for extracting host labels from the parsed data. For unparameterized host label functions, it must accept exactly one argument by the name ‘section’. When used in conjunction with a ruleset, it must accept two arguments: ‘params’ and ‘section’. The type of ‘params’ depends on the ruleset type. It will be a single mapping for MERGED rulesets and a sequence of mappings for ALL rulesets. ‘section will be the parsed data as returned by the parse function. It is expected to yield objects of type HostLabel.
host_label_default_parameters (Mapping[str, object] | None) – Default parameters for the host label function. Must match the ValueSpec of the corresponding WATO ruleset, if it exists.
host_label_ruleset_name (str | None) – The name of the host label ruleset.
host_label_ruleset_type (RuleSetType) – The ruleset type is either RuleSetType.ALL or RuleSetType.MERGED. It describes whether this plug-in needs the merged result of the effective rules, or every individual rule matching for the current host.
supersedes (list[str] | None) – A list of section names which are superseded by this section. If this section will be parsed to something that is not None (see above) all superseded section will not be considered at all.

class CheckPlugin(*, name, sections=None, service_name, discovery_function, discovery_default_parameters=None, discovery_ruleset_name=None, discovery_ruleset_type=RuleSetType.MERGED, check_function, check_default_parameters=None, check_ruleset_name=None, cluster_check_function=None)

A CheckPlugin to plug into Checkmk.

Instances of this class will only be picked up by Checkmk if their names start with check_plugin_.

Parameters:

name (str) – The unique name of the check plug-in. It must only contain the characters ‘A-Z’, ‘a-z’, ‘0-9’ and the underscore.
sections (list[str] | None) – An optional list of section names that this plug-in subscribes to. They correspond to the ‘parsed_section_name’ specified in agent_section() and snmp_section(). The corresponding sections are passed to the discovery and check function. The functions arguments must be called ‘section_<name1>, section_<name2>’ ect. Defaults to a list containing as only element a name equal to the name of the check plug-in.
service_name (str) – The template for the service name. The check function must accept ‘item’ as first argument if and only if “%s” is present in the value of “service_name”.
discovery_function (Callable[..., Iterable[Service]]) – The discovery_function. Arguments must be ‘params’ (if discovery parameters are defined) and ‘section’ (if the plug-in subscribes to a single section), or ‘section_<name1>, section_<name2>’ ect. corresponding to the sections. It is expected to be a generator of Service instances.
discovery_default_parameters (Mapping[str, object] | None) – Default parameters for the discovery function. Must match the ValueSpec of the corresponding WATO ruleset, if it exists.
discovery_ruleset_name (str | None) – The name of the discovery ruleset.
discovery_ruleset_type (RuleSetType) – The ruleset type is either RuleSetType.ALL or RuleSetType.MERGED. It describes whether this plug-in needs the merged result of the effective rules, or every individual rule matching for the current host.
check_function (Callable[..., Iterable[IgnoreResults | Metric | Result]]) – The check_function. Arguments must be ‘item’ (if the service has an item), ‘params’ (if check default parameters are defined) and ‘section’ (if the plug-in subscribes to a single section), or ‘section_<name1>, section_<name2>’ ect. corresponding to the sections.
check_default_parameters (Mapping[str, object] | None) – Default parameters for the check function. Must match the ValueSpec of the corresponding WATO ruleset, if it exists.
check_ruleset_name (str | None) – The name of the check ruleset.
cluster_check_function (Callable[..., Iterable[IgnoreResults | Metric | Result]] | None) – The cluster check function. If this function is not specified, the corresponding services will not be available for clusters. The arguments are the same as the ones for the check function, except that the sections are dicts (node name -> node section).

class SNMPSection(*, name: str, detect: SNMPDetectSpecification, fetch: Sequence[SNMPTree], parse_function: Callable[[Sequence[_TableTypeT]], _Section | None], host_label_function: Callable[[_Section], Generator[HostLabel, None, None]] | None = None, host_label_default_parameters: None = None, host_label_ruleset_name: None = None, host_label_ruleset_type: RuleSetType = RuleSetType.MERGED, parsed_section_name: str | None = None, supersedes: list[str] | None = None)

class SNMPSection(*, name: str, detect: SNMPDetectSpecification, fetch: Sequence[SNMPTree], parse_function: Callable[[Sequence[_TableTypeT]], _Section | None], host_label_function: Callable[[Sequence[Mapping[str, object]], _Section], Generator[HostLabel, None, None]], host_label_default_parameters: Mapping[str, object], host_label_ruleset_name: str, host_label_ruleset_type: Literal[RuleSetType.MERGED] = RuleSetType.MERGED, parsed_section_name: str | None = None, supersedes: list[str] | None = None)

class SNMPSection(*, name: str, detect: SNMPDetectSpecification, fetch: Sequence[SNMPTree], parse_function: Callable[[Sequence[_TableTypeT]], _Section | None], host_label_function: Callable[[Sequence[Mapping[str, object]], _Section], Generator[HostLabel, None, None]], host_label_default_parameters: Mapping[str, object], host_label_ruleset_name: str, host_label_ruleset_type: Literal[RuleSetType.ALL], parsed_section_name: str | None = None, supersedes: list[str] | None = None)

An SNMPSection to plug into Checkmk

Instances of this class will only be picked up by Checkmk if their names start with snmp_section_.

The snmp information will be gathered and parsed according to the functions and options given to this function:

Parameters:

name (str) – The unique name of the section to be registered.
detect (SNMPDetectSpecification) – The conditions on single OIDs that will result in the attempt to fetch snmp data and discover services. This should only match devices to which the section is applicable. It is highly recommended to check the system description OID at the very first, as this will make the discovery much more responsive and consume less resources.
fetch (Sequence[SNMPTree]) – The specification of snmp data that should be fetched from the device. It must be a non-empty list of SNMPTree objects. The parse function will be passed a non-empty list of StringTable instances accordingly.
parse_function (Callable[[Sequence[TypeVar(_TableTypeT, List[List[Union[str, List[int]]]], List[List[str]])]], Optional[TypeVar(_Section, bound= object)]]) – The function responsible for parsing the raw snmp data. It must accept exactly one argument by the name ‘string_table’. It will be passed either a list of StringTable`s, matching the length of the value of the `fetch argument. It may return an arbitrary object. Note that if the return value is None, no further processing will take place (just as if the agent had not sent any data). This function may raise arbitrary exceptions, which will be dealt with by the checking engine. You should expect well formatted data.
parsed_section_name (str | None) – The name under which the parsed section will be available to the plug-ins. Defaults to the original name.
host_label_function (Callable[[TypeVar(_Section, bound= object)], Generator[HostLabel, None, None]] | Callable[[Mapping[str, object], TypeVar(_Section, bound= object)], Generator[HostLabel, None, None]] | Callable[[Sequence[Mapping[str, object]], TypeVar(_Section, bound= object)], Generator[HostLabel, None, None]] | None) – The function responsible for extracting host labels from the parsed data. It must accept exactly one argument by the name ‘section’. When the function is called, it will be passed the parsed data as returned by the parse function. It is expected to yield objects of type HostLabel.
host_label_default_parameters (Mapping[str, object] | None) – Default parameters for the host label function. Must match the ValueSpec of the corresponding WATO ruleset, if it exists.
host_label_ruleset_name (str | None) – The name of the host label ruleset.
host_label_ruleset_type (RuleSetType) – The ruleset type is either RuleSetType.ALL or RuleSetType.MERGED. It describes whether this plug-in needs the merged result of the effective rules, or every individual rule matching for the current host.
supersedes (list[str] | None) – A list of section names which are superseded by this section. If this section will be parsed to something that is not None (see above) all superseded section will not be considered at all.

class SimpleSNMPSection(*, name: str, detect: SNMPDetectSpecification, fetch: SNMPTree, parse_function: Callable[[_TableTypeT], _Section | None], host_label_function: Callable[[_Section], Generator[HostLabel, None, None]] | None = None, host_label_default_parameters: None = None, host_label_ruleset_name: None = None, host_label_ruleset_type: RuleSetType = RuleSetType.MERGED, parsed_section_name: str | None = None, supersedes: list[str] | None = None)

class SimpleSNMPSection(*, name: str, detect: SNMPDetectSpecification, fetch: SNMPTree, parse_function: Callable[[_TableTypeT], _Section | None], host_label_function: Callable[[Mapping[str, object], _Section], Generator[HostLabel, None, None]], host_label_default_parameters: Mapping[str, object], host_label_ruleset_name: str, host_label_ruleset_type: Literal[RuleSetType.MERGED] = RuleSetType.MERGED, parsed_section_name: str | None = None, supersedes: list[str] | None = None)

class SimpleSNMPSection(*, name: str, detect: SNMPDetectSpecification, fetch: SNMPTree, parse_function: Callable[[_TableTypeT], _Section | None], host_label_function: Callable[[Sequence[Mapping[str, object]], _Section], Generator[HostLabel, None, None]], host_label_default_parameters: Mapping[str, object], host_label_ruleset_name: str, host_label_ruleset_type: Literal[RuleSetType.ALL], parsed_section_name: str | None = None, supersedes: list[str] | None = None)

A SimpleSNMPSection to plug into Checkmk

Instances of this class will only be picked up by Checkmk if their names start with snmp_section_.

The snmp information will be gathered and parsed according to the functions and options given to this function:

Parameters:

name (str) – The unique name of the section to be registered.
detect (SNMPDetectSpecification) – The conditions on single OIDs that will result in the attempt to fetch snmp data and discover services. This should only match devices to which the section is applicable. It is highly recommended to check the system description OID at the very first, as this will make the discovery much more responsive and consume less resources.
fetch (SNMPTree) – The specification of snmp data that should be fetched from the device. It must be an SNMPTree object. The parse function will be passed a single StringTable.
parse_function (Callable[[TypeVar(_TableTypeT, List[List[Union[str, List[int]]]], List[List[str]])], Optional[TypeVar(_Section, bound= object)]]) – The function responsible for parsing the raw snmp data. It must accept exactly one argument by the name ‘string_table’. It will be passed a StringTable. It may return an arbitrary object. Note that if the return value is None, no further processing will take place (just as if the agent had not sent any data). This function may raise arbitrary exceptions, which will be dealt with by the checking engine. You should expect well formatted data.
parsed_section_name (str | None) – The name under which the parsed section will be available to the plug-ins. Defaults to the original name.
host_label_function (Callable[[TypeVar(_Section, bound= object)], Generator[HostLabel, None, None]] | Callable[[Mapping[str, object], TypeVar(_Section, bound= object)], Generator[HostLabel, None, None]] | Callable[[Sequence[Mapping[str, object]], TypeVar(_Section, bound= object)], Generator[HostLabel, None, None]] | None) – The function responsible for extracting host labels from the parsed data. For unparameterized host label functions, it must accept exactly one argument by the name ‘section’. When used in conjunction with a ruleset, it must accept two arguments: ‘params’ and ‘section’. The type of ‘params’ depends on the ruleset type. It will be a single mapping for MERGED rulesets and a sequence of mappings for ALL rulesets. ‘section will be the parsed data as returned by the parse function. It is expected to yield objects of type HostLabel.
host_label_default_parameters (Mapping[str, object] | None) – Default parameters for the host label function. Must match the ValueSpec of the corresponding WATO ruleset, if it exists.
host_label_ruleset_name (str | None) – The name of the host label ruleset.
host_label_ruleset_type (RuleSetType) – The ruleset type is either RuleSetType.ALL or RuleSetType.MERGED. It describes whether this plug-in needs the merged result of the effective rules, or every individual rule matching for the current host.
supersedes (list[str] | None) – A list of section names which are superseded by this section. If this section will be parsed to something that is not None (see above) all superseded section will not be considered at all.

class SNMPDetectSpecification(iterable=(), /)

A specification for SNMP device detection

Note that the structure of this object is not part of the API, and may change at any time.

class InventoryPlugin(*, name, sections=None, inventory_function, inventory_default_parameters=None, inventory_ruleset_name=None)

An InventoryPlugin to plug into Checkmk.

Instances of this class will only be picked up by Checkmk if their names start with inventory_plugin_.

Parameters:

name (str) – The unique name of the plug-in. It must only contain the characters ‘A-Z’, ‘a-z’, ‘0-9’ and the underscore.
sections (list[str] | None) – An optional list of section names that this plug-in subscribes to. They correspond to the ‘parsed_section_name’ specified in agent_section() and snmp_section(). The corresponding sections are passed to the discovery and check function. The functions arguments must be called ‘section_<name1>, section_<name2>’ ect. Defaults to a list containing as only element a name equal to the name of the inventory plug-in.
inventory_function (Callable[..., Iterable[Attributes | TableRow]]) – The inventory_function. Arguments must be ‘params’ (if inventory parameters are defined) and ‘section’ (if the plug-in subscribes to a single section), or ‘section_<name1>, section_<name2>’ ect. corresponding to the sections. It is expected to be a generator of Attributes or TableRow instances.
inventory_default_parameters (Mapping[str, object] | None) – Default parameters for the inventory function. Must match the ValueSpec of the corresponding WATO ruleset, if it exists.
inventory_ruleset_name (str | None) – The name of the inventory ruleset.

all_of(spec_0, spec_1, *specs)

Detect the device if all passed specifications are met

Parameters:

spec_0 (SNMPDetectSpecification) – A valid specification for SNMP device detection
spec_1 (SNMPDetectSpecification) – A valid specification for SNMP device detection

Return type:

SNMPDetectSpecification

Returns:

A valid specification for SNMP device detection

Example

>>> DETECT = all_of(exists("1.2.3.4"), contains("1.2.3.5", "foo"))

any_of(*specs)

Detect the device if any of the passed specifications are met

Parameters:: spec – A valid specification for SNMP device detection
Return type:: SNMPDetectSpecification
Returns:: A valid specification for SNMP device detection

Example

>>> DETECT = any_of(exists("1.2.3.4"), exists("1.2.3.5"))

exists(oidstr)

Detect the device if the OID exists at all

Parameters:: oidstr (str) – The OID that is required to exist
Return type:: SNMPDetectSpecification
Returns:: A valid specification for SNMP device detection

Example

>>> DETECT = exists("1.2.3")

equals(oidstr, value)

Detect the device if the value of the OID equals the given string

Parameters:

oidstr (str) – The OID to match the value against
value (str) – The expected value of the OID

Return type:

SNMPDetectSpecification

Returns:

A valid specification for SNMP device detection

Example

>>> DETECT = equals("1.2.3", "MySwitch")

startswith(oidstr, value)

Detect the device if the value of the OID starts with the given string

Parameters:

oidstr (str) – The OID to match the value against
value (str) – The expected start of the OIDs value

Return type:

SNMPDetectSpecification

Returns:

A valid specification for SNMP device detection

Example

>>> DETECT = startswith("1.2.3", "Sol")

endswith(oidstr, value)

Detect the device if the value of the OID ends with the given string

Parameters:

oidstr (str) – The OID to match the value against
value (str) – The expected end of the OIDs value

Return type:

SNMPDetectSpecification

Returns:

A valid specification for SNMP device detection

Example

>>> DETECT = endswith("1.2.3", "nix")

contains(oidstr, value)

Detect the device if the value of the OID contains the given string

Parameters:

oidstr (str) – The OID to match the value against
value (str) – The substring expected to be in the OIDs value

Return type:

SNMPDetectSpecification

Returns:

A valid specification for SNMP device detection

Example

>>> DETECT = contains("1.2.3", "isco")

matches(oidstr, value)

Detect the device if the value of the OID matches the expression

Parameters:

oidstr (str) – The OID to match the value against
value (str) – The regular expression that the value of the OID should match

Return type:

SNMPDetectSpecification

Returns:

A valid specification for SNMP device detection

Example

>>> DETECT = matches("1.2.3.4", ".* Server")

not_exists(oidstr)

The negation of exists()

Return type:: SNMPDetectSpecification

not_equals(oidstr, value)

The negation of equals()

Return type:: SNMPDetectSpecification

not_contains(oidstr, value)

The negation of contains()

Return type:: SNMPDetectSpecification

not_endswith(oidstr, value)

The negation of endswith()

Return type:: SNMPDetectSpecification

not_matches(oidstr, value)

The negation of matches()

Return type:: SNMPDetectSpecification

not_startswith(oidstr, value)

The negation of startswith()

Return type:: SNMPDetectSpecification

class Attributes(*, path: list[str], inventory_attributes: Mapping[str, int | float | str | bool | None] | None = None, status_attributes: Mapping[str, int | float | str | bool | None] | None = None): Attributes to be written at a node in the HW/SW Inventory

check_levels(value, *, levels_upper=None, levels_lower=None, metric_name=None, render_func=None, label=None, boundaries=None, notice_only=False)

Generic function for checking a value against levels.

Parameters:

value (float) – The currently measured value
levels_upper (Union[tuple[Literal['no_levels'], None], tuple[Literal['fixed'], tuple[TypeVar(_NumberT, int, float), TypeVar(_NumberT, int, float)]], tuple[Literal['predictive'], tuple[str, float | None, tuple[TypeVar(_NumberT, int, float), TypeVar(_NumberT, int, float)] | None]], None]) – Upper level parameters created by the :class:Levels form spec
levels_lower (Union[tuple[Literal['no_levels'], None], tuple[Literal['fixed'], tuple[TypeVar(_NumberT, int, float), TypeVar(_NumberT, int, float)]], tuple[Literal['predictive'], tuple[str, float | None, tuple[TypeVar(_NumberT, int, float), TypeVar(_NumberT, int, float)] | None]], None]) – Lower level parameters created by the :class:Levels form spec
metric_name (str | None) – The name of the datasource in the RRD that corresponds to this value or None in order not to generate a metric.
render_func (Callable[[float], str] | None) – A single argument function to convert the value from float into a human-readable string.
label (str | None) – The label to prepend to the output.
boundaries (tuple[float | None, float | None] | None) – Minimum and maximum to add to the metric.
notice_only (bool) – Only show up in service output if not OK (otherwise in details). See notice keyword of Result class.

Return type:

Iterable[Result | Metric]

Example

>>> result, metric = check_levels(
...     23.0,
...     levels_upper=("fixed", (12., 42.)),
...     metric_name="temperature",
...     label="Fridge",
...     render_func=lambda v: "%.1f°" % v,
... )
>>> print(result.summary)
Fridge: 23.0° (warn/crit at 12.0°/42.0°)
>>> print(metric)
Metric('temperature', 23.0, levels=(12.0, 42.0))

get_average(value_store, key, time, value, backlog_minutes)

Return new average based on current value and last average

Parameters:

value_store (MutableMapping[str, object]) – The Mapping that holds the last value. Usually this will be the value store provided by the API.
key (str) – Unique ID for storing this average until the next check
time (float) – Timestamp of new value
value (float) – The new value
backlog_minutes (float) – Averaging horizon in minutes

This function returns the new average value aₙ as the weighted sum of the current value xₙ and the last average:

aₙ = (1 - w)xₙ + waₙ₋₁

= (1-w) ∑ᵢ₌₀ⁿ wⁱxₙ₋ᵢ

This results in a so-called “exponential moving average”.

The weight is chosen such that for long-running timeseries the “backlog” (all recorded values in the last n minutes) will make up 50% of the weighted average.

Assuming k values in the backlog, compute their combined weight such that they sum up to the backlog weight b (0.5 in our case):

b = (1-w) ∑ᵢ₌₀ᵏ⁻¹ wⁱ => w = (1 - b) ** (1/k) (“geometric sum”)

Until the averaging horizon has been reached, we set the horizon to the time passed since starting to average. This:

Avoids giving undue weight to the first value

Helps to arrive at a meaningful average more quickly

Return type:: float
Returns:: The computed average

get_rate(value_store, key, time, value, *, raise_overflow=False)

Update value store.
Calculate rate based on current value and time and last value and time

Parameters:

value_store (MutableMapping[str, Any]) – The mapping that holds the last value. Usually this will be the value store provided by the APIs get_value_store().
key (str) – Unique ID for storing the time/value pair until the next check
time (float) – Timestamp of new value
value (float) – The new value
raise_overflow (bool) – Raise a GetRateError if the rate is negative

This function returns the rate of a measurement rₙ as the quotient of the value and time provided to the current function call (xₙ, tₙ) and the value and time provided to the previous function call (xₙ₋₁, tₙ₋₁):

rₙ = (xₙ - xₙ₋₁) / (tₙ - tₙ₋₁)

Note that the function simply computes the quotient of the values and times given, regardless of any unit. You might as well pass something different than the time. However, this function is written with the use case of passing timestamps in mind.

A GetRateError will be raised if one of the following happens:

the function is called for the first time

the time has not changed

the rate is negative and raise_overflow is set to True (useful for instance when dealing with counters)

In general there is no need to catch a GetRateError, as it inherits IgnoreResultsError.

Example

>>> # in practice: my_store = get_value_store()
>>> my_store = {}
>>> try:
...     rate = get_rate(my_store, 'my_rate', 10, 23)
... except GetRateError:
...     pass  # this fails the first time, because my_store is empty.
>>> my_store  # now remembers the last time/value
{'my_rate': (10, 23)}
>>> # Assume in the next check cycle (60 seconds later) the value has increased to 56.
>>> # get_rate uses the new and old values to compute (56 - 23) / (70 - 10)
>>> get_rate(my_store, 'my_rate', 70, 56)
0.55

Return type:: float
Returns:: The computed rate

get_value_store()

Get the value store for the current service from Checkmk

The returned value store object can be used to persist values between different check executions. It is a MutableMapping, so it can be used just like a dictionary.

Return type:: MutableMapping[str, Any]

class HostLabel(name: str, value: str)

Representing a host label in Checkmk

This class creates a host label that can be yielded by a host_label_function as regisitered with the section.

>>> my_label = HostLabel("my_key", "my_value")

class IgnoreResults(value='currently no results')

A result to make the service go stale, but carry on with the check function

Yielding a result of type IgnoreResults will have a similar effect as raising an IgnoreResultsError, with the difference that the execution of the check funtion will not be interrupted.

yield IgnoreResults("Good luck next time!")
return

is equivalent to

raise IgnoreResultsError("Good luck next time!")

This is useful for instance if you want to initialize all counters, before returning.

exception IgnoreResultsError

Raising an IgnoreResultsError from within a check function makes the service go stale.

Example

>>> def check_db_table(item, section):
...     if item not in section:
...         # avoid a lot of UNKNOWN services:
...         raise IgnoreResultsError("Login to database failed")
...     # do your work here
>>>

Create a metric for a service

Parameters:

name – The name of the metric.
value – The measured value.
levels – A pair of upper levels, ie. warn and crit. This information is only used for visualization by the graphing system. It does not affect the service state.
boundaries – Additional information on the value domain for the graphing system.

If you create a Metric in this way, you may want to consider using check_levels().

Example

>>> my_metric = Metric("used_slots_percent", 23.0, levels=(80, 90), boundaries=(0, 100))

class OIDBytes(value: str)

Class to indicate that the OIDs value should be provided as list of integers

Parameters:: oid – The OID to fetch

Example

>>> _ = OIDBytes("2.1")

class OIDCached(value: str)

Class to indicate that the OIDs value should be cached

Parameters:: oid – The OID to fetch

Example

>>> _ = OIDCached("2.1")

class OIDEnd

Class to indicate the end of the OID string should be provided

When specifying an OID in an SNMPTree object, the parse function will be handed the corresponding value of that OID. If you use OIDEnd() instead, the parse function will be given the tailing portion of the OID (the part that you not already know).

class Result(*, state: State, summary: str, details: str | None = None)

class Result(*, state: State, notice: str, details: str | None = None)

A result to be yielded by check functions

This is the class responsible for creating service output and setting the state of a service.

Parameters:

state – The resulting state of the service.
summary – The text to be displayed in the services summary view.
notice – A text that will only be shown in the summary if state is not OK.
details – The alternative text that will be displayed in the details view. Defaults to the value of summary or notice.

Note

You must specify exactly one of the arguments summary and notice!

When yielding more than one result, Checkmk will not only aggregate the texts, but also compute the worst state for the service and highlight the individual non-OK states in the output. You should always match the state to the output, and yield subresults:

>>> def my_check_function() -> None:
...     # the back end will comput the worst overall state:
...     yield Result(state=State.CRIT, summary="All the foos are broken")
...     yield Result(state=State.OK, summary="All the bars are fine")
>>>
>>> # run function to make sure we have a working example
>>> _ = list(my_check_function())

The notice keyword has the special property that it will only be displayed in the summary if the state passed to _this_ Result instance is not OK. Otherwise we assume it is sufficient to show the information in the details view:

>>> def my_check_function() -> None:
...     count = 23
...     yield Result(
...         state=State.WARN if count <= 42 else State.OK,
...         notice=f"Things: {count}",  # only appear in summary if count drops below 43
...         details=f"We currently have this many things: {count}",
...     )
>>>
>>> # run function to make sure we have a working example
>>> _ = list(my_check_function())

If you find yourself computing the state by comparing a metric to some thresholds, you probably should be using check_levels()!

class RuleSetType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Indicate the type of the rule set

Discovery and host label functions may either use all rules of a rule set matching the current host, or the merged rules.

class Service(*, item: str | None = None, parameters: Mapping[str, object] | None = None, labels: Sequence[ServiceLabel] | None = None)

Class representing services that the discover function yields

Parameters:

item – The item of the service
parameters – The determined discovery parameters for this service
labels – A list of labels attached to this service

Example

>>> my_drive_service = Service(
...    item="disc_name",
...    parameters={},
... )

class ServiceLabel(name: str, value: str)

Representing a service label in Checkmk

This class creates a service label that can be passed to a ‘Service’ object. It can be used in the discovery function to create a new label like this:

>>> my_label = ServiceLabel("my_key", "my_value")

class SNMPTree(base: str, oids: Sequence[str | _OIDSpecTuple])

Specify an OID table to fetch

For every SNMPTree that is specified, the parse function will be handed a list of lists with the values of the corresponding OIDs.

Parameters:

base – The OID base string, starting with a dot.
oids – A list of OID specifications.

Example

>>> _ = SNMPTree(
...     base=".1.2.3.4.5.6",
...     oids=[
...         OIDEnd(),  # I want the end oids of every entry
...         "7.8",  # just a regular entry
...         OIDCached("123"),  # this is HUGE, please cache it
...         OIDBytes("42"),  # I expect bytes, give me a list of integers
...     ],
... )

class State(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

States of check results

classmethod best(*args)

Returns the best of all passed states

You can pass an arbitrary number of arguments, and the return value will be the “best” of them, where

OK -> WARN -> UNKNOWN -> CRIT

Parameters:: args (State | int) – Any number of one of State.OK, State.WARN, State.CRIT, State.UNKNOWN
Return type:: State
Returns:: The best of the input states, one of State.OK, State.WARN, State.CRIT, State.UNKNOWN.

Examples

>>> State.best(State.OK, State.WARN, State.CRIT, State.UNKNOWN)
<State.OK: 0>
>>> State.best(0, 1, State.CRIT)
<State.OK: 0>

classmethod worst(*args)

Returns the worst of all passed states.

You can pass an arbitrary number of arguments, and the return value will be the “worst” of them, where

OK < WARN < UNKNOWN < CRIT

Parameters:: args (State | int) – Any number of one of State.OK, State.WARN, State.CRIT, State.UNKNOWN
Return type:: State
Returns:: The worst of the input States, one of State.OK, State.WARN, State.CRIT, State.UNKNOWN.

Examples

>>> State.worst(State.OK, State.WARN, State.CRIT, State.UNKNOWN)
<State.CRIT: 2>
>>> State.worst(0, 1, State.CRIT)
<State.CRIT: 2>

class TableRow(*, path: list[str], key_columns: Mapping[str, int | float | str | bool | None], inventory_columns: Mapping[str, int | float | str | bool | None] | None = None, status_columns: Mapping[str, int | float | str | bool | None] | None = None): TableRow to be written into a Table at a node in the HW/SW Inventory

exception GetRateError: The exception raised by get_rate(). If unhandled, this exception will make the service go stale.

cmk.agent_based.v2

New in this version

type_defs module is dissolved

check_levels signature changed

Registration is replaced by a discovery approach

Changed arguments and validation for Agent and SNMP sections

Removed wrapper for regex creation regex()

Added rendering function time_offset()

Example:

Removed wrapper for regex creation `regex()`

Added rendering function `time_offset()`