Collection of CheckMK checks (see https://checkmk.com/). All checks and plugins are provided as is. Absolutely no warranty. Send any comments to thl-cmk[at]outlook[dot]com

Skip to content
Snippets Groups Projects
Commit 1a9cfa61 authored by thl-cmk's avatar thl-cmk :flag_na:
Browse files

update project

parent 549b8877
No related branches found
No related tags found
No related merge requests found
[PACKAGE]: ../../raw/master/create_topology_data-0.0.6-20231017.mkp "create_topology_data-0.0.6-20231017.mkp"
[PACKAGE]: ../../raw/master/create_topology_data-0.0.6-20231018.mkp "create_topology_data-0.0.6-20231018.mkp"
# PoC for creating topology_data.json from inventory data
This script creates the topology data file needed for the [Checkmk Exchange Network visualization](https://forum.checkmk.com/u/schnetz) plugin by Andreas Boesl and [schnetz](https://exchange.checkmk.com/u/schnetz). For more information see [Checkmk forum Network Visualization](https://forum.checkmk.com/t/network-visualization/41680).
......
......@@ -15,6 +15,7 @@
# added option --check-user-data-only
# added SEED_DEVICES to user data, option -s/--sed-devices is now optional
# refactoring
# 2023-10-17: fixed --make-default not working
#
# PoC for creating topology_data.json from inventory data
......@@ -50,15 +51,14 @@ from pathlib import Path
from typing import Dict, List, Any, Tuple
from time import strftime, time_ns
from create_topology_utils import (
parse_arguments,
save_data_to_file,
remove_old_data,
get_data_from_toml,
is_mac_address,
is_equal_with_default,
merge_topologies,
save_topology,
# get_data_form_live_status, # no valid json at the moment
)
from create_topology_classes import (
InventoryColumns,
......@@ -68,7 +68,7 @@ from create_topology_classes import (
)
CREATE_TOPOLOGY_VERSION = '0.0.6-202310117'
CREATE_TOPOLOGY_VERSION = '0.0.6-202310118'
ITEMS = {}
MAC_TABLE: Dict[str, Interface] = {}
......@@ -151,6 +151,20 @@ def add_static_connections(host: str, data: Dict) -> Dict:
return data
# def get_inventory_data_lq(host: str, path: List[str], debug: bool = False, ) -> List[Dict[str, str]] | None:
# query = f'GET hosts\nColumns: mk_inventory\nOutputFormat: json\nFilter: host_name = {host}\n'
# data = get_data_form_live_status(query=query)
#
# if data:
# for m in path:
# try:
# data = data[m]
# except KeyError:
# return []
# return data
# return []
def get_inventory_data(host: str, path: List[str], debug: bool = False, ) -> List[Dict[str, str]] | None:
inventory_file = Path(f'{SETTINGS.omd_root}/{SETTINGS.inventory_path}/{host}')
data = []
......@@ -181,6 +195,7 @@ def create_mac_to_interface_table(debug: bool = False):
if host.startswith('.'):
continue
if_table = get_inventory_data(host=host, path=SETTINGS.path_to_if_table, debug=debug)
# if_table = get_inventory_data_lq(host=host, path=SETTINGS.path_to_if_table, debug=debug)
if if_table and if_table != []:
if debug:
print(f'host: {host}, interfaces: {len(if_table)}')
......@@ -209,40 +224,6 @@ def create_mac_to_interface_table(debug: bool = False):
continue
def save_topology(
data: dict,
base_directory: str,
output_directory: str,
dont_compare: bool,
make_default: bool,
topology_file_name: str,
):
path = f'{base_directory}/{output_directory}'
if dont_compare:
save_data_to_file(
data=data,
path=path,
file=topology_file_name,
make_default=make_default,
)
else:
if not is_equal_with_default(
data=data,
file=f'{base_directory}/default/{topology_file_name}'
):
save_data_to_file(
data=data,
path=path,
file=topology_file_name,
make_default=make_default,
)
else:
print(
'Topology matches default topology, not saved! Use "--dont-compare" to save identical topologies.'
)
def create_device_from_inv(
host: str,
inv_data: List[Dict[str, str]],
......@@ -364,7 +345,7 @@ def create_topology(
continue
topo_data = get_inventory_data(host=device, path=path_in_inventory, debug=debug)
# topo_data = get_inventory_data_lq(host=device, path=path_in_inventory, debug=debug)
topology_data.update(
create_device_from_inv(
host=device,
......@@ -409,7 +390,7 @@ if __name__ == '__main__':
STATIC_CONNECTIONS = create_static_connections(connections=user_data.get('STATIC_CONNECTIONS', []))
# not yet used, need more test data
# create_mac_to_interface_table()
create_mac_to_interface_table()
topology = create_topology(
seed_devices=list(set(SETTINGS.seed_devices + user_data.get('SEED_DEVICES', []))),
......
......@@ -2,7 +2,7 @@
# -*- coding: utf-8 -*-
#
# License: GNU General Public License v2
import json.decoder
# Author: thl-cmk[at]outlook[dot]com
# URL : https://thl-cmk.hopto.org
# Date : 2023-10-12
......@@ -29,7 +29,8 @@
# --uppercase
from os import environ
from json import dumps
from json import dumps, loads
import socket
from ast import literal_eval
from time import time as now_time
from tomllib import loads as toml_loads
......@@ -53,6 +54,41 @@ lldp_columns = 'system_name,local_port_num,port_id'
lldp_label = 'inv_LLDP'
user_data_file = 'create_topology_data.toml'
# gets no valid json at the moment :-(
# def get_data_form_live_status(query: str):
# address = f'{environ.get("OMD_ROOT")}/tmp/run/live'
# family = socket.AF_INET if type(address) is tuple else socket.AF_UNIX
# sock = socket.socket(family, socket.SOCK_STREAM)
# sock.connect(address)
# sock.sendall(query.encode())
# sock.shutdown(socket.SHUT_WR)
# chunks = []
# while len(chunks) == 0 or chunks[-1] != "":
# chunks.append(sock.recv(4096).decode())
# sock.close()
# if len(chunks):
# reply = loads("".join(chunks))
# try:
# reply = reply[0][0]
# reply = reply.replace("'", '"')
# reply = reply.replace(' False', ' false')
# reply = reply.replace(' True', ' true')
# reply = reply.replace(' None', ' null')
# reply = reply.replace('"\\\\"', '"')
# reply = reply.replace('\\\\""', '"')
# except IndexError:
# return None
# if reply:
# # print(reply)
# try:
# reply = loads(reply)
# except json.decoder.JSONDecodeError as e:
# print(reply)
# print(e)
# print(query)
# exit()
# return reply
def get_data_from_toml(file: str, debug: bool = False) -> Dict:
data = {}
......@@ -130,14 +166,48 @@ def save_data_to_file(data: Dict, path: str, file: str, make_default: bool):
Returns:
None
"""
print(f'make_default: {make_default}')
path_file = f'{path}/{file}'
save_file = Path(f'{path_file}')
save_file.parent.mkdir(exist_ok=True, parents=True)
save_file.write_text(dumps(data))
parent_path = Path(f'{path_file}').parent
if make_default:
parent_path = Path(f'{path}').parent
Path(f'{parent_path}/default').unlink(missing_ok=True)
Path(f'{parent_path}/default').symlink_to(target=Path(path_file), target_is_directory=True)
Path(f'{parent_path}/default').symlink_to(target=Path(path), target_is_directory=True)
def save_topology(
data: dict,
base_directory: str,
output_directory: str,
dont_compare: bool,
make_default: bool,
topology_file_name: str,
):
path = f'{base_directory}/{output_directory}'
def _save():
save_data_to_file(
data=data,
path=path,
file=topology_file_name,
make_default=make_default,
)
if dont_compare:
_save()
else:
if not is_equal_with_default(
data=data,
file=f'{base_directory}/default/{topology_file_name}'
):
_save()
else:
print(
'Topology matches default topology, not saved! Use "--dont-compare" to save identical topologies.'
)
def is_mac_address(mac_address: str, debug: bool = False) -> bool:
......
File added
......@@ -45,7 +45,7 @@
'topology_data/create_topology_data.toml_clean']},
'name': 'create_topology_data',
'title': 'Create network topology data',
'version': '0.0.6-20231017',
'version': '0.0.6-20231018',
'version.min_required': '2.2.0p1',
'version.packaged': '2.2.0p11',
'version.usable_until': '2.3.0p1'}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment