create_topology_utils.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# License: GNU General Public License v2

# Author: thl-cmk[at]outlook[dot]com
# URL   : https://thl-cmk.hopto.org
# Date  : 2023-10-12
# File  : create_topology_utils.py

#
# options used
# -c --inventory-columns
# -d --data-source
# -m --make-default
# -o --output-directory
# -p --path-in-inventory
# -s --seed-devices
# --time-format
# --keep-domain
# --lowercase
# --uppercase
# --version
# --debug

from os import environ
from pathlib import Path
from typing import List, Dict
from argparse import (
    Namespace as arg_Namespace,
    ArgumentParser,
    RawTextHelpFormatter,
)

script = '~/local/bin/network-topology/create_topology_data.py'
sample_seeds = 'Core01 Core02'
cdp_path = 'networking,cdp_cache'
cdp_columns = 'device_id,local_port,device_port'
cdp_label = 'inv_CDP'
lldp_path = 'networking,lldp_cache'
lldp_columns = 'system_name,local_port_num,port_id'
lldp_label = 'inv_LLDP'


def rm_tree(root: Path):
    # safety
    if not str(root).startswith(f'{environ["OMD_ROOT"]}/var/topology_data'):
        print(f"WARNING: bad path to remove, {str(root)}, don\'t delete it.")
        return
    for p in root.iterdir():
        if p.is_dir():
            rm_tree(p)
        else:
            p.unlink()
    root.rmdir()


def is_dict_keys_equal(dict1: Dict, dict2: Dict) -> bool:
    keys1 = list(dict1.keys())
    keys2 = list(dict2.keys())
    keys1.sort()
    keys2.sort()
    return keys1 == keys2


def is_is_list_of_str__equal(list1: List[str], list2: List[str]) -> bool:
    return list1.sort() == list2.sort()


def compare_dicts(dict1: Dict, dict2: Dict) -> bool:
    # check top level keys
    if not is_dict_keys_equal(dict1, dict2):
        return False

    for key, value in dict1.items():
        _type = type(value)
        if _type == dict:
            if not compare_dicts(value, dict2[key]):
                return False
        elif _type == list:
            if not is_is_list_of_str__equal(value, dict2[key]):
                return False
        elif _type == str:
            if not value == dict2[key]:
                return False
        else:
            return False

    return True


def parse_arguments(create_topology_version: str) -> arg_Namespace:
    parser = ArgumentParser(
        prog='create_topology_data.py',
        description='This script creates the topology data file needed for the Checkmk "network_visualization" '
                    'plugin by Andreas Boesl and schnetz.\n'
                    'For more information see https://forum.checkmk.com/t/network-visualization/41680 and '
                    'https://exchange.checkmk.com/p/network-visualization\n'
                    '\n'
                    'The inventory data could be created with my inventory plugins:\n'
                    'CDP: https://thl-cmk.hopto.org/gitlab/checkmk/vendor-independent/inventory/inv_cdp_cache\n'
                    'LLDP: https://thl-cmk.hopto.org/gitlab/checkmk/vendor-independent/inventory/inv_lldp_cache\n'
                    '\n'
                    f'\nVersion: {create_topology_version} | Written by: thl-cmk, for more information '
                    f'see: https://thl-cmk.hopto.org',
        formatter_class=RawTextHelpFormatter,
        epilog='Usage:\n'
               'for CDP (the default):\n'
               f'{script} -s {sample_seeds} -m\n'
               f'or\n'
               f'{script} -s {sample_seeds} -m -p "{cdp_path}" -c "{cdp_columns}" -d {cdp_label}\n\n'
               'for LLDP:\n'
               f'{script} -s {sample_seeds} -m --lldp\n'
               f'or\n'
               f'{script} -s {sample_seeds} -m -p "{lldp_path}" -c "{lldp_columns}" -d {lldp_label}\n',
    )
    command_group = parser.add_mutually_exclusive_group()
    parser.add_argument(
        '-c', '--inventory-columns', type=str,
        help=f'Columns used from the inventory data. I.e. "{cdp_columns}"\n'
             'NOTE: the columns must be in the order: neighbour, local_port, neighbour_port',
    )
    parser.add_argument(
        '-d', '--data-source', type=str,
        help=f'The source from which the topology data originates. I.e. {cdp_label} '
             'for CDP data from the inventory. NOTE: right now this only an unused label.',
    )
    parser.add_argument(
        '-m', '--make-default', default=False, action='store_const', const=True,
        help='Set the created topology data as default',
    )
    parser.add_argument(
        '-o', '--output-directory', type=str,
        help='Directory name where to save the topology data. I.e.: my_topology. Default is the actual date/time '
             'in "--time-format" format.\n'
             'NOTE: the directory is a sub directory under "~/var/topology_data/"',
    )
    parser.add_argument(
        '-p', '--path-in-inventory', type=str,
        help=f'Checkmk inventory path to the topology data. I.e. "{cdp_path}"',
    )
    parser.add_argument(
        '-s', '--seed-devices', type=str, nargs='+',
        help=f'List of devices to start the topology discovery from. I.e. {sample_seeds}',
    )
    parser.add_argument(
        '-v', '--version', default=False, action='store_const', const=True,
        help='Print version of this script and exit',
    )
    parser.add_argument(
        '--debug', default=False, action='store_const', const=True,
        help='Print debug information',
    )
    parser.add_argument(
        '--dont-compare', default=False, action='store_const', const=True,
        help='Do not compare the actual topology data with the default topology data. By default, the actual\n'
             'topology is compared with the default topology. If the data matches, the actual topology is not saved.\n'
             'So, if you run this tool in a cron job, a new topology will be created only if there was a change,\n'
             'unless you use "--dont-compare".'
    )
    parser.add_argument(
        '--keep-domain', default=False, action='store_const', const=True,
        help='Do not remove the domain name from the neighbor name',
    )
    parser.add_argument(
        '--keep-max', type=int,
        help=f'Number of topologies to keep. The oldest topologies above keep max will be deleted.\n'
             f'The default topologies will be always kept. The minimum value for --keep-max is 1.',
    )
    parser.add_argument(
        '--lldp', default=False, action='store_const', const=True,
        help=f'Set data source to {lldp_label}, inventory path to "{lldp_path}" and columns to "{lldp_columns}"',
    )
    command_group.add_argument(
        '--lowercase', default=False, action='store_const', const=True,
        help='Change neighbour names to all lower case',
    )
    parser.add_argument(
        '--min-age', type=int,
        help=f'The minimum number of days before a topology is deleted by "--keep-max". Topologies that are not '
             f'older than 1 days are always kept.',
    )
    parser.add_argument(
        '--time-format', type=str,
        help='Format string to render the time. (default: %%Y-%%m-%%dT%%H:%%M:%%S.%%m)',
    )
    command_group.add_argument(
        '--uppercase', default=False, action='store_const', const=True,
        help='Change neighbour names to all upper case',
    )
    return parser.parse_args()