Hello guys, maybe someone can help me with this, because I’m not sure how to proceed from here. So we’ve got a custom extension, written years ago for version 1.4. This extension monitors our emergency power module.
I attached the screenshot of the extension package, there are also man pages included, but that’s not my priority. The power module got analogue and digital sensors. The analogue sensors monitor the tank level and motor temperature, the digital ones the overall status of the module itself.
I want to upgrade my CheckMK instance from version 2.2 to 2.3 (and then to 2.4 after the initial step to 2.3) but I can’t update to 2.3 because of the new API.
So I’ve ported the analogue script from python2 to python3 and tried to import the new API modules, which works to some extend, but every time I try to update my instance, it just tells me ‘name’… Also the analogue sensor checks no longer work in the monitoring (Item not found in monitoring data). Here’s the original code:
#!/usr/bin/python
# -*- encoding: utf-8; py-indent-offset: 4 -*-
import collections
from cmk.base.plugins.agent_based.utils.backport import check_levels_backport
from cmk.base.plugins.agent_based.utils.my_temperature import check_temperature
AnalogSensor = collections.namedtuple('AnalogSensor', 'description maximum minimum voltage')
# tank defaults
TANK_RADIUS = float(37.5) # cm
TANK_LENGTH = float(120) # cm
def parse_tcw241_analog(info):
"""
parse info data and create list of namedtuples for 4 analog sensors.
expected info structure:
list of 4 analog sensors:
[description, maximum , minimum]
..
[description, maximum , minimum]
list of analog voltages:
[voltage1, voltage2, voltage3, voltage4]
converted to list structure:
[AnalogSensor(description maximum minimum voltage)]
:param info: parsed snmp data
:return: list of namedtuples for analog sensors
"""
try:
sensor_parameter, voltages = info[0], info[1][0]
except IndexError:
return {}
info_dict = {}
for (oid, description, maximum, minimum), voltage in zip(sensor_parameter, voltages):
try:
sensor_voltage = float(voltage) / 1000.0
sensor_maximum = float(maximum) / 1000.0
sensor_minimum = float(minimum) / 1000.0
except ValueError:
continue
if sensor_minimum < 1 or sensor_maximum < 1:
continue
info_dict[oid.split(".")[0]] = AnalogSensor(description=str(description),
maximum=sensor_maximum,
minimum=sensor_minimum,
voltage=sensor_voltage)
return info_dict
def inventory_tcw241_analog(parsed):
for key in parsed:
sensor = parsed.get(key)
yield sensor.description, {}
def check_tcw241_analog(item, _params, parsed):
"""
Check sensor data if value is in range
:param item: sensor number
:param _params: <not used>
:param sensor: analog sensor data
:return: status
"""
for key in parsed:
sensor = parsed.get(key)
if item != sensor.description:
continue
# special case
if sensor.description.startswith("Tank"):
volume = calculate_volume(tank_radius=TANK_RADIUS,
tank_length=TANK_LENGTH,
tank_fill_level=sensor.voltage)
yield check_levels_backport(volume,
"volume", (None, None, sensor.maximum, sensor.minimum),
unit="l",
infoname="[%s]" % sensor.description)
elif sensor.description.endswith("Temp"):
yield check_temperature(sensor.voltage,
(sensor.maximum, sensor.minimum ),
unique_name="[%s]" % sensor.description,
dev_unit="c",
#dev_levels = (sensor.maximum, sensor.minimum ),
dev_levels_lower = (sensor.maximum - 65, sensor.minimum - 60))
else:
yield check_levels_backport(sensor.voltage,
"voltage", (sensor.minimum, sensor.maximum),
unit="V",
infoname="[%s]" % sensor.description)
def calculate_volume(tank_radius, tank_length, tank_fill_level):
"""
Calculates volume in liter based on
https://iwer.info/article/Mathematisches/Volumen-liegender-Zylinder/index.html
:param tank_radius: radius of cylinder
:param tank_length: length cylinder of
:param fill_level: height of fill level of cylinder
:return: volume
"""
minuend = math.pow(tank_radius, 2) * math.acos((tank_radius - tank_fill_level) / tank_radius)
subtrahend = (tank_radius - tank_fill_level) * math.sqrt(2 * tank_radius * tank_fill_level -
math.pow(tank_fill_level, 2))
difference = minuend - subtrahend
volume = tank_length * difference
return round(volume / 1000, 3)
check_info["teracom_tcw241_analog"] = {
"parse_function": parse_tcw241_analog,
"check_function": check_tcw241_analog,
"inventory_function": inventory_tcw241_analog,
"service_description": "Analog Sensor %s",
"has_perfdata": True,
#"includes": ["backport.include", "temperature.include"],
'snmp_scan_function': lambda oid: "Teracom" in oid(".1.3.6.1.2.1.1.1.0"),
'snmp_info': [
(
".1.3.6.1.4.1.38783.3.2.2.2",
[
"1", # Voltage 1
"2", # Voltage 2
"3", # Voltage 3
"4" # Voltage 4
],
[
OID_END,
"1", # Voltage description
"2", # Voltage maximum x1000 in Integer format
"3" # Voltage minimum x1000 in Integer format
]),
(
".1.3.6.1.4.1.38783.3.3.2",
[
"1.0", # Voltage 1 x1000 in Integer format
"2.0", # Voltage 2 x1000 in Integer format
"3.0", # Voltage 3 x1000 in Integer format
"4.0" # Voltage 4 x1000 in Integer format
])
]
}
And here my modified attempt (which isn’t complete right now):
#!/usr/bin/env python3
# -*- encoding: utf-8; py-indent-offset: 4 -*-
import math
import collections
from cmk.base.plugins.agent_based.agent_based_api.v1 import (
SNMPTree,
OIDEnd,
Result,
State,
Metric,
check_levels,
Service,
)
from cmk.base.plugins.agent_based.agent_based_api.v1.register import (
snmp_section,
check_plugin,
inventory_plugin,
)
AnalogSensor = collections.namedtuple('AnalogSensor', 'description maximum minimum voltage')
# tank defaults
TANK_RADIUS = 37.5 # cm
TANK_LENGTH = 120.0 # cm
def calculate_volume(tank_radius, tank_length, tank_fill_level):
minuend = math.pow(tank_radius, 2) * math.acos((tank_radius - tank_fill_level) / tank_radius)
subtrahend = (tank_radius - tank_fill_level) * math.sqrt(2 * tank_radius * tank_fill_level - math.pow(tank_fill_level, 2))
difference = minuend - subtrahend
volume = tank_length * difference
return round(volume / 1000, 3)
def parse_tcw241_analog(string_table):
try:
sensor_parameter, voltages = string_tables[0], string_tables[1][0]
except IndexError:
return {}
info_dict = {}
for (oid, description, maximum, minimum), voltage in zip(sensor_parameter, voltages):
try:
sensor_voltage = float(voltage) / 1000.0
sensor_maximum = float(maximum) / 1000.0
sensor_minimum = float(minimum) / 1000.0
except ValueError:
continue
if sensor_minimum < 1 or sensor_maximum < 1:
continue
info_dict[oid.split(".")[0]] = AnalogSensor(
description=str(description),
maximum=sensor_maximum,
minimum=sensor_minimum,
voltage=sensor_voltage,
)
return info_dict
@snmp_section(
name="teracom_tcw241_analog",
detect=SNMPTree(".1.3.6.1.2.1.1.1.0", ["0"]),
fetch=[
SNMPTree(".1.3.6.1.4.1.38783.3.2.2.2", [OIDEnd(), "1", "2", "3"]),
SNMPTree(".1.3.6.1.4.1.38783.3.3.2", ["1.0", "2.0", "3.0", "4.0"]),
],
parse_function=parse_tcw241_analog,
)
def section_tcw241_analog():
pass
@check_plugin(
name="teracom_tcw241_analog",
service_name="Analog Sensor %s",
discovery_function=lambda section: [Service(item=sensor.description) for sensor in section.values()],
check_function=lambda item, section: check_tcw241_analog(item, section),
)
def check_tcw241_analog(item, section):
for sensor in section.values():
if item != sensor.description:
continue
if sensor.description.startswith("Tank"):
volume = calculate_volume(TANK_RADIUS, TANK_LENGTH, sensor.voltage)
yield from check_levels(
value=volume,
levels_upper=(sensor.maximum, sensor.minimum),
metric_name="volume",
label=f"[{sensor.description}]",
boundaries=(0, None),
)
elif sensor.description.endswith("Temp"):
yield from check_levels(
value=sensor.voltage,
levels_lower=(sensor.maximum - 65, sensor.minimum - 60),
metric_name="temperature",
label=f"[{sensor.description}]",
boundaries=(0, 100),
)
else:
yield from check_levels(
value=sensor.voltage,
levels_upper=(sensor.maximum, sensor.minimum),
metric_name="voltage",
label=f"[{sensor.description}]",
boundaries=(0, 10),
)
@inventory_plugin(
name="teracom_tcw241_analog",
inventory_function=lambda section: ((sensor.description, {}) for sensor in section.values()),
)
def inventory_tcw241_analog(section):
pass
Did somebody of you did something similar and could share his code for me? Or put me in the right direction which new API modules I’d need to make this work. And also, is the path for the extensions still correct? 
Every kind of help is very much appreciated! Best regards! 