#!/usr/bin/env python3 # pylint: disable=missing-module-docstring, unused-argument, # pylint: disable=missing-function-docstring, line-too-long import datetime import time from pprint import pprint from cmk.utils import debug # import necessary elements from API version 2 from cmk.agent_based.v2 import ( AgentSection, CheckPlugin, Service, State, Metric, Result, DiscoveryResult, CheckResult, 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_traefik_info(string_table): """the parse function""" parsed_data = {} for line in string_table: if line[0] == "Traefik_Version": parsed_data["version"] = line[1] elif line[0] == "Traefik_CodeName": parsed_data["codename"] = line[1] elif line[0] == "Traefik_StartDate": parsed_data["startdate"] = line[1] # convert timestamp into seconds since epoch dt = datetime.datetime.fromisoformat(line[1]) parsed_data["startdate_seconds_since_epoch"] = int(dt.timestamp()) elif line[0] == "Agent_Runtime": # convert seconds into ms parsed_data["agent_runtime"] = float(line[1]) * 1_000 elif line[0] == "Agent_Version": parsed_data["agent_version"] = line[1] return parsed_data def discover_traefik_info(section) -> DiscoveryResult: """the discover function""" yield Service() def check_traefik_info(params, section) -> CheckResult: """the check function""" if debug.enabled(): pprint(section) _level_type, levels = params["levels_traefik_agent_execution_time"] codename: str = section["codename"] version: str = section["version"] startdate: str = section["startdate"] startdate_seconds_since_epoch: int = section["startdate_seconds_since_epoch"] # calculate runtime of Traefik in seconds current_epoch_time: int = int(time.time()) runtime: int = current_epoch_time - startdate_seconds_since_epoch agent_version: str = section["agent_version"] agent_runtime: float = round(section["agent_runtime"], 1) state: State = get_state_upper(levels=levels, value=agent_runtime) yield Metric( name="traefik_agent_execution_time", value=agent_runtime, levels=levels, ) summary: str = f"Traefik version: {version}, code name: {codename}" details: str = f"Traefik start date: {startdate}\nRunning since: {render.timespan(runtime)}\n\nAgent version: {agent_version}\nAgent execution time: {agent_runtime}ms" yield Result(state=state, summary=summary, details=details) # create the new agent section, must begin with "agent_section_" # and must be an instance of "AgentSection" agent_section_traefik_info = AgentSection( # "name" must exactly match the section name within the agent output name="traefik_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_traefik_info, ) # create the new check plugin, must begin with "check_plugin_" # and must be an instance of "CheckPlugin" check_plugin_traefik_info = CheckPlugin( # "name" should be the same as the corresponding section within the agent output name="traefik_info", service_name="Traefik 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_traefik_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_traefik_info, # define the default parameters check_default_parameters={ "levels_traefik_agent_execution_time": ("fixed", (500.0, 750.0)) }, # connect to the ruleset where parameters can be defined # must match the name of the ruleset exactly check_ruleset_name="traefik_info", )