test_wowza.py

#!/usr/bin/env python3

"""
Criticality: Normal
Checks that the streaming server (Wowza) is running correctly.
"""

from pathlib import Path
import re
import subprocess  # nosec: B404
import sys

from defusedxml.ElementTree import parse
from psutil import net_connections
from packaging.version import parse as parse_version
from lxml import etree

sys.path.append(str(Path(__file__).parents[1].resolve()))

# pylint: disable=wrong-import-position
from utilities import logging as lg  # noqa: E402

LATEST_VERSION = "4.8.5"
WOWZA_TUNE_FILE = "/usr/local/WowzaStreamingEngine/conf/Tune.xml"
WOWZA_RECOMMENDED_HEAP_SIZE = 2000


def main():
    """Run all checks and exits with corresponding exit code."""

    print("Checking Wowza settings:")
    warnings = 0
    errors = 0

    # check if wowza is installed
    if not check_installed():
        exit(2)

    # check wowza version
    check_warn, check_err = check_version()
    if check_err:
        errors += 1
    elif check_warn:
        warnings += 1

    # check wowza heap size
    check_warn, check_err = check_heap_size()
    if check_err:
        errors += 1
    elif check_warn:
        warnings += 1

    # check that wowza is running
    check_warn, check_err = check_running()
    if check_err:
        errors += 1
    elif check_warn:
        warnings += 1

    # check that wowza is listening
    check_warn, check_err = check_listening()
    if check_err:
        errors += 1
    elif check_warn:
        warnings += 1

    if errors:
        exit(1)
    elif warnings:
        exit(3)

    exit(0)


def check_installed() -> bool:
    """Check that Wowza is installed.

    :return: Exit return codes
    :rtype: bool
    """

    cmd = "dpkg --get-selections 'wowza*'"
    out = subprocess.getoutput(cmd)
    state = out.split()[-1]

    if state != "install":
        lg.info("not installed, skip test")
        return False

    return True


def check_version() -> tuple:
    """Check the Wowza version installed.

    :return: Exit return codes
    :rtype: bool
    """

    warnings = 0
    errors = 0

    cmd = "dpkg --get-selections 'wowza*'"
    out = subprocess.getoutput(cmd)
    version = None

    for line in out.split("\n"):
        if line.split()[-1] == "install":
            if version:
                lg.error("many Wowza versions installed, keep only latest")
                errors += 1
            version = ".".join(re.findall(r"\d", line))

    if not version:
        lg.error("cannot find Wowza version")
        errors += 1

    if parse_version(version) < parse_version(LATEST_VERSION):
        lg.info(
            "using outdated version: {0} < {1} (recommended)"
            .format(version, LATEST_VERSION)
        )
    elif parse_version(version) > parse_version(LATEST_VERSION):
        lg.success(
            "using newer version than the recommended: {0} > {1} (recommended)"
            .format(version, LATEST_VERSION)
        )
    else:
        lg.success("using the recommended version: {0}".format(version))

    return warnings, errors


def check_heap_size() -> tuple:
    """Check the heap size configured.

    :return: Exit return codes
    :rtype: bool
    """

    warnings = 0
    errors = 0

    # Configured wowza heap size extraction
    tune_xml = etree.parse(WOWZA_TUNE_FILE)
    heap_size = tune_xml.find('Tune/HeapSize').text[0:-1]

    if int(heap_size) < WOWZA_RECOMMENDED_HEAP_SIZE:
        lg.error(
            "not using recommended heap size: {0}M < {1}M (recommended)"
            .format(int(heap_size), recommended_heap_size)
        )
        errors += 1
    else:
        lg.success(
            "using recommended heap size or above: {0}M"
            .format(int(heap_size))
        )

    return warnings, errors


def check_running() -> tuple:
    """Check that Wowza is running.

    :return: Exit return codes
    :rtype: bool
    """

    warnings = 0
    errors = 0

    cmd = "systemctl status WowzaStreamingEngine"
    out = subprocess.getoutput(cmd)
    if "Active: active (running)" not in out:
        lg.error("service not running")
        errors += 1
    else:
        lg.success("service running")

    return warnings, errors


def check_listening() -> tuple:
    """Check that Wowza is listening on configured port.

    :return: Exit return codes
    :rtype: bool
    """

    warnings = 0
    errors = 0

    # get port number in Wowza config
    conf = parse("/usr/local/WowzaStreamingEngine/conf/VHost.xml").getroot()
    port = conf.findall("VHost/HostPortList/HostPort/Port")[0].text

    # get listening ports
    listening = set(
        c.laddr[1] for c in net_connections(kind="inet") if c.status == "LISTEN"
    )

    # check that system is listening on this port
    if int(port) not in listening:
        lg.error("not listening on port {}".format(port))
        errors += 1
    else:
        lg.success("listening on port {}".format(port))

    return warnings, errors


if __name__ == "__main__":
    main()