Newer
Older
#!/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 abc import abstractmethod
from ast import literal_eval
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.__topology_save_path = 'var/topology_data'
self.__topology_file_name = 'network_data.json'
self.__path_to_if_table = 'networking,interfaces'
'user_data_file': f'{self.__omd_root}/local/bin/topology_data/{USER_DATA_FILE}',
'keep_domain': False,
'lowercase': False,
'uppercase': False,
'debug': False,
'version': False,
# 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:
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.')
@property
def backend(self) -> str:
return self.__settings['backend']
@property
def version(self) -> bool:
return self.__settings['version']
def layers(self) -> List[str]:
return self.__settings['layers']
@property
def min_age(self) -> int:
if self.__settings['min_age']:
@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']
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:
@property
def user_data_file(self) -> str:
return self.__settings['user_data_file']
@property
def seed_devices(self) -> List[str]:
if 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
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)
if host not in self.__cache.keys():
self.__cache[host]: Dict[str, Any] = {}
self.__fill_cache(host=host)
try:
return None
def add_inventory_prefetch_path(self, path: str):
self.__inventory_pre_fetch_list = list(set(self.__inventory_pre_fetch_list + [path]))
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
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