NeoAnd/netbox-zabbix-sync
1
0
Fork 0
mirror of https://github.com/TheNetworkGuy/netbox-zabbix-sync.git synced 2025-07-14 01:41:25 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

Moved Inventory mapping logic to tools module

Browse Source
This commit is contained in:
Raymond Kuiper 2025-02-14 09:46:55 +01:00
parent 6d4e250b23
commit 1b831a2d39
2 changed files with 63 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
modules/device.py
View File

@ -13,6 +13,7 @@ from modules.exceptions import (SyncInventoryError, TemplateError, SyncExternalE
from modules.interface import ZabbixInterface from modules.interface import ZabbixInterface
from modules.usermacros import ZabbixUsermacros from modules.usermacros import ZabbixUsermacros
from modules.hostgroups import Hostgroup from modules.hostgroups import Hostgroup
from modules.tools import field_mapper
from pprint import pprint from pprint import pprint
try: try:
@ -195,31 +196,32 @@ class PhysicalDevice():
self.inventory = {} self.inventory = {}
if inventory_sync and self.inventory_mode in [0,1]: if inventory_sync and self.inventory_mode in [0,1]:
self.logger.debug(f"Host {self.name}: Starting inventory mapper") self.logger.debug(f"Host {self.name}: Starting inventory mapper")
# Let's build an inventory dict for each property in the inventory_map self.inventory = field_mapper(self.name, self._inventory_map(), nbdevice, self.logger)
for nb_inv_field, zbx_inv_field in self._inventory_map().items(): # # Let's build an inventory dict for each property in the inventory_map
field_list = nb_inv_field.split("/") # convert str to list based on delimiter # for nb_inv_field, zbx_inv_field in self._inventory_map().items():
# start at the base of the dict... # field_list = nb_inv_field.split("/") # convert str to list based on delimiter
value = nbdevice # # start at the base of the dict...
# ... and step through the dict till we find the needed value # value = nbdevice
for item in field_list: # # ... and step through the dict till we find the needed value
value = value[item] if value else None # for item in field_list:
# Check if the result is usable and expected # value = value[item] if value else None
# We want to apply any int or float 0 values, # # Check if the result is usable and expected
# even if python thinks those are empty. # # We want to apply any int or float 0 values,
if ((value and isinstance(value, int | float | str )) or # # even if python thinks those are empty.
(isinstance(value, int | float) and int(value) ==0)): # if ((value and isinstance(value, int | float | str )) or
self.inventory[zbx_inv_field] = str(value) # (isinstance(value, int | float) and int(value) ==0)):
elif not value: # self.inventory[zbx_inv_field] = str(value)
# empty value should just be an empty string for API compatibility # elif not value:
self.logger.debug(f"Host {self.name}: NetBox inventory lookup for " # # empty value should just be an empty string for API compatibility
f"'{nb_inv_field}' returned an empty value") # self.logger.debug(f"Host {self.name}: NetBox inventory lookup for "
self.inventory[zbx_inv_field] = "" # f"'{nb_inv_field}' returned an empty value")
else: # self.inventory[zbx_inv_field] = ""
# Value is not a string or numeral, probably not what the user expected. # else:
self.logger.error(f"Host {self.name}: Inventory lookup for '{nb_inv_field}'" # # Value is not a string or numeral, probably not what the user expected.
" returned an unexpected type: it will be skipped.") # self.logger.error(f"Host {self.name}: Inventory lookup for '{nb_inv_field}'"
self.logger.debug(f"Host {self.name}: Inventory mapping complete. " # " returned an unexpected type: it will be skipped.")
f"Mapped {len(list(filter(None, self.inventory.values())))} field(s)") # self.logger.debug(f"Host {self.name}: Inventory mapping complete. "
# f"Mapped {len(list(filter(None, self.inventory.values())))} field(s)")
return True return True
def isCluster(self): def isCluster(self):

36
modules/tools.py
View File

@ -1,3 +1,5 @@
from logging import getLogger
"""A collection of tools used by several classes""" """A collection of tools used by several classes"""
def convert_recordset(recordset): def convert_recordset(recordset):
""" Converts netbox RedcordSet to list of dicts. """ """ Converts netbox RedcordSet to list of dicts. """
@ -42,3 +44,37 @@ def proxy_prepper(proxy_list, proxy_group_list):
group["monitored_by"] = 2 group["monitored_by"] = 2
output.append(group) output.append(group)
return output return output
def field_mapper(host, mapper, nbdevice, logger):
"""
Maps NetBox field data to Zabbix properties.
Used for Inventory, Usermacros and Tag mappings.
"""
data={}
# Let's build an dict for each property in the map
for nb_field, zbx_field in mapper.items():
field_list = nb_field.split("/") # convert str to list based on delimiter
# start at the base of the dict...
value = nbdevice
# ... and step through the dict till we find the needed value
for item in field_list:
value = value[item] if value else None
# Check if the result is usable and expected
# We want to apply any int or float 0 values,
# even if python thinks those are empty.
if ((value and isinstance(value, int | float | str )) or
(isinstance(value, int | float) and int(value) ==0)):
data[zbx_field] = str(value)
elif not value:
# empty value should just be an empty string for API compatibility
logger.debug(f"Host {host}: NetBox lookup for "
f"'{nb_field}' returned an empty value")
data[zbx_field] = ""
else:
# Value is not a string or numeral, probably not what the user expected.
logger.error(f"Host {host}: Lookup for '{nb_field}'"
" returned an unexpected type: it will be skipped.")
logger.debug(f"Host {host}: Field mapping complete."
f"Mapped {len(list(filter(None, data.values())))} field(s)")
return data