zeek_halfduplex_connections.py

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

# const
PLUGIN_NAME='zeek_halfduplex_connections'
SERVICE_NAME='Zeek Half duplex connections'
THRESHOLD_CRIT=5
THRESHOLD_WARN=3
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_halfduplex(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_halfduplex(section):
    yield Service()

def check_halfduplex(section):
  summary = f"{section['percentage']}% half duplex connections ({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_halfduplex
)

register.check_plugin(
  name=PLUGIN_NAME,
  service_name=SERVICE_NAME,
  discovery_function=discover_halfduplex,
  check_function=check_halfduplex
)