create_topology_classes.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_classes.py


from os import environ
from pathlib import Path
from time import strftime
from typing import Dict, List, Any, NamedTuple
from enum import Enum, unique
from abc import abstractmethod
from ast import literal_eval

from create_topology_utils import (
    CREATE_TOPOLOGY_VERSION,
    PATH_INTERFACES,
    USER_DATA_FILE,
    CACHE_INTERFACES_ITEM,
    SCRIPT,
    get_data_form_live_status,
    get_table_from_inventory,
    ExitCodes,
    get_data_from_toml,
    OMD_ROOT,
)


@unique
class CacheItems(Enum):
    inventory = 'inventory'
    interfaces = 'interfaces'


class InventoryColumns(NamedTuple):
    neighbour: str
    local_port: str
    neighbour_port: str


class StaticConnection(NamedTuple):
    host: str
    local_port: str
    neighbour_port: str
    neighbour: str
    label: str


class Settings:
    def __init__(
            self,
            cli_args: Dict[str, Any],
    ):
        self.__omd_root = environ['OMD_ROOT']
        self.__topology_save_path = 'var/topology_data'
        self.__topology_file_name = 'network_data.json'
        self.__path_to_if_table = 'networking,interfaces'
        self.__settings = {
            'layers': [],
            'seed_devices': None,
            'time_format': '%Y-%m-%dT%H:%M:%S.%m',
            'user_data_file': f'{self.__omd_root}/local/bin/topology_data/{USER_DATA_FILE}',
            'output_directory': None,
            'default': False,
            'keep_domain': False,
            'lowercase': False,
            'uppercase': False,
            'debug': False,
            'version': False,
            'dont_compare': False,
            'check_user_data_only': False,
            'keep': None,
            'min_age': None,
            'backend': 'LIVESTATUS',
        }
        # args in the form {'s, __seed_devices': 'CORE01', 'p, __path_in_inventory': None, ... }}
        # we will remove 's, __'
        self.__args = ({k.split(',')[-1].strip(' ').strip('_'): v for k, v in cli_args.items() if v})
        self.__settings.update(self.__args)

        if self.version:
            print(f'{Path(SCRIPT).name} version: {CREATE_TOPOLOGY_VERSION}')
            exit(code=ExitCodes.OK.value)

        if self.check_user_data_only:
            _user_data = get_data_from_toml(file=self.user_data_file)
            print(f'Could read/parse the user data from {self.user_data_file}')
            exit(code=ExitCodes.OK.value)

        if self.layers:
            layers = list(set(self.layers))
            if len(layers) != len(self.layers):
                print(f'-l/--layers options must be unique. Don~\'t use any layer more than once.')
                exit(code=ExitCodes.BAD_OPTION_LIST.value)

    @property
    def backend(self) -> str:
        return self.__settings['backend']

    @property
    def version(self) -> bool:
        return self.__settings['version']

    @property
    def layers(self) -> List[str]:
        return self.__settings['layers']

    @property
    def keep(self) -> int | None:
        if self.__settings['keep']:
            return max(self.__settings['keep'], 0)

    @property
    def min_age(self) -> int:
        if self.__settings['min_age']:
            return max(self.__settings['min_age'], 0)
        else:
            return 0

    @property
    def check_user_data_only(self) -> bool:
        return self.__settings['check_user_data_only']

    @property
    def debug(self) -> bool:
        return self.__settings['debug']

    @property
    def dont_compare(self) -> bool:
        return self.__settings['dont_compare']

    @property
    def default(self) -> bool:
        return self.__settings['default']

    @property
    def keep_domain(self) -> bool:
        return self.__settings['keep_domain']

    @property
    def uppercase(self) -> bool:
        return self.__settings['uppercase']

    @property
    def lowercase(self) -> bool:
        return self.__settings['lowercase']

    @property
    def time_format(self) -> str:
        return self.__settings['time_format']

    @property
    def omd_root(self) -> str:
        return self.__omd_root

    @property
    def user_data_file(self) -> str:
        return self.__settings['user_data_file']

    @property
    def topology_save_path(self) -> str:
        return self.__topology_save_path

    @property
    def topology_file_name(self) -> str:
        return self.__topology_file_name

    @property
    def seed_devices(self) -> List[str]:
        if self.__settings['seed_devices']:
            return self.__settings['seed_devices']
        else:
            return []

    @property
    def output_directory(self) -> str:
        if not self.__settings['output_directory']:
            return f'{strftime(self.__settings["time_format"])}'
        else:
            return self.__settings['output_directory']


class HostCache:
    def __init__(self):
        self.__cache = {}
        self.__inventory_pre_fetch_list: List[str] = [
            PATH_INTERFACES,
        ]

    @abstractmethod
    def get_inventory_data(self, host: str, debug: bool = False) -> Dict[str, str] | None:
        """
        Args:
            host: the host name to return the inventory data for
            debug: enable debug output

        Returns:
            the inventory data as dictionary
        """

    @abstractmethod
    def get_interface_items(self, host: str, debug: bool = False) -> List:
        """

        Args:
            host:  the host name to return the interface items
            debug: enable debug output

        Returns:
            list of the interface items
        """

    def __fill_cache(self, host: str):
        # pre fill inventory data
        inventory = self.get_inventory_data(host=host)
        if inventory:
            self.__cache[host][CacheItems.inventory.value] = {}
            self.__cache[host][CacheItems.inventory.value].update({
                entry: get_table_from_inventory(
                    inventory=inventory,
                    path=entry
                ) for entry in self.__inventory_pre_fetch_list
            })
        else:
            self.__cache[host][CacheItems.inventory.value] = None
        self.__cache[host][CacheItems.interfaces.value] = {}
        self.__cache[host][CacheItems.interfaces.value][CACHE_INTERFACES_ITEM] = self.get_interface_items(host)

    def get_data(self, host: str, item: CacheItems, path: str):
        if host not in self.__cache.keys():
            self.__cache[host]: Dict[str, Any] = {}
            self.__fill_cache(host=host)
        try:
            return self.__cache[host][item.value][path]
        except (KeyError, TypeError) as _e:
            return None

    def add_inventory_prefetch_path(self, path: str):
        self.__inventory_pre_fetch_list = list(set(self.__inventory_pre_fetch_list + [path]))


class HostCacheLiveStatus(HostCache):
    def get_inventory_data(self, host: str, debug: bool = False) -> Dict[str, str] | None:
        query = f'GET hosts\nColumns: mk_inventory\nOutputFormat: python3\nFilter: host_name = {host}\n'
        data = get_data_form_live_status(query=query)
        if data:
            try:
                data = literal_eval(data[0][0].decode('utf-8'))
            except SyntaxError as e:
                if debug:
                    print(f'data: |{data}|')
                    print(f'type: {type(data)}')
                    print(f'exception: {e}')
                return
            return data

    def get_interface_items(self, host: str, debug: bool = False) -> List:
        """
        Sample data from lq query, we keep only the item (description without "Interface" ).
        [
         ['C9540-7-1', 'Interface Vlan-999', 'check_mk-if64'],
         ['C9540-7-1', 'Interface Vlan-998', 'check_mk-if64'],
         ['C9540-7-1', 'Interface Vlan-997', 'check_mk-if64'],
         ['C9540-7-1', 'Interface Vlan-996', 'check_mk-if64'],
         ['C9540-7-1', 'Interface Vlan-8', 'check_mk-if64'],
         ['C9540-7-1', 'Interface Te2/0/2', 'check_mk-if64'],
         ['C9540-7-1', 'Interface Te2/0/30', 'check_mk-if64']
         ]
        Args:
            host:
            debug:

        Returns:

        """
        query = (
            'GET services\n'
            'Columns: host_name description check_command\n'
            'Filter: description ~ Interface\n'
            f'Filter: host_name = {host}\n'
            'OutputFormat: python3\n'
        )
        data = get_data_form_live_status(query=query)
        items = []
        for host, description, check_command in data:
            items.append(description[10:])  # remove 'Interface ' from description

        if debug:
            print(f'Interfaces items found: {len(items)} an host {host}')
        return items


class HostCacheFileSystem(HostCache):
    def get_inventory_data(self, host: str, debug: bool = False) -> Dict[str, str] | None:
        __inventory_path = 'var/check_mk/inventory'
        inventory_file = Path(f'{OMD_ROOT}/{__inventory_path}/{host}')
        if inventory_file.exists():
            data = literal_eval(inventory_file.read_text())
            return data
        else:
            return None

    def get_interface_items(self, host: str, debug: bool = False) -> List:
        """
        Sample autochecks data we keep only the item
        [
         {'check_plugin_name': 'if64', 'item': 'Fa0', 'parameters': {'discovered_oper_status': ['2'], ...}},
         {'check_plugin_name': 'if64', 'item': 'Fa1/0/1', 'parameters': {'discovered_oper_status': ['1'], ...}},
         {'check_plugin_name': 'if64', 'item': 'Fa1/0/10', 'parameters': {'discovered_oper_status': ['2'], ...}},
         {'check_plugin_name': 'if64', 'item': 'Fa1/0/11', 'parameters': {'discovered_oper_status': ['2'], ...}},
         {'check_plugin_name': 'if64', 'item': 'Fa1/0/12', 'parameters': {'discovered_oper_status': ['1'], ...}}
        ]

        Args:
            host:
            debug:

        Returns:

        """
        __autochecks_path = 'var/check_mk/autochecks'
        autochecks_file = Path(f'{OMD_ROOT}/{__autochecks_path}/{host}.mk')
        __data = []
        if autochecks_file.exists():
            data: List[Dict[str, str]] = literal_eval(autochecks_file.read_text())
            for service in data:
                if service['check_plugin_name'] in ['if64']:
                    __data.append(service['item'])
        else:
            if debug:
                print(f'Device: {host}: not found in auto checks path!')
            return []

        return __data