zeek_packet_loss.py

from .agent_based_api.v1 import register, Result, State, Service
import re

# const
PLUGIN_NAME='zeek_packet_loss'
SERVICE_NAME='Zeek packet loss'
THRESHOLD_CRIT=50.0
THRESHOLD_WARN=30.0
REGEX='(?P<percentage>\d{1,3}\.\d{1,2})%,\s+(?P<bad_conn_count>\d+) out of (?P<total_conn_count>\d+)'

def strip_ansi(source):
  COLOUR_CODE=r'\033\[(\d|;)+?m'
  return re.sub(COLOUR_CODE, '', source)

def parse_packet_loss(string_table):
  for string in string_table:
    clean_string = strip_ansi(str(' '.join(string)))
    match = re.compile(REGEX).match(clean_string)
    if match:
      return_dict = {}
      return_dict['percentage'] = float(match.group('percentage'))
      return_dict['bad_conn_count'] = match.group('bad_conn_count')
      return_dict['total_conn_count'] = match.group('total_conn_count')
      return return_dict

def discover_packet_loss(section):
  yield Service()

def check_packet_loss(section):
  summary = f"{section['percentage']}% packet loss ({section['bad_conn_count']} out of {section['total_conn_count']})"
  if section['percentage'] >= THRESHOLD_CRIT: yield Result(state=State.CRIT, summary=summary); return
  if section['percentage'] >= THRESHOLD_WARN: yield Result(state=State.WARN, summary=summary); return
  yield Result(state=State.OK, summary=summary); return

register.agent_section(
  name=PLUGIN_NAME,
  parse_function=parse_packet_loss
)

register.check_plugin(
  name=PLUGIN_NAME,
  service_name=SERVICE_NAME,
  discovery_function=discover_packet_loss,
  check_function=check_packet_loss
)