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
    try:
        tune_xml = etree.parse(WOWZA_TUNE_FILE)
        value = tune_xml.find('Tune/HeapSize').text[0:-1]
        heap_size = int(value)
    except Exception as e:
        lg.info(
            'failed to get heap size value: {0}'
            .format(e)
        )
    else:
        if heap_size < WOWZA_RECOMMENDED_HEAP_SIZE:
            lg.error(
                'not using recommended heap size: {0}M < {1}M (recommended)'
                .format(heap_size, WOWZA_RECOMMENDED_HEAP_SIZE)
            )
            errors += 1
        else:
            lg.success(
                'using recommended heap size or above: {0}M'
                .format(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()