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

Full usermacro support

Browse Source
This commit is contained in:
Raymond Kuiper 2025-02-14 15:18:26 +01:00
parent 1b831a2d39
commit eea7df660a
5 changed files with 73 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
modules/device.py
View File

@ -1,5 +1,5 @@
#!/usr/bin/env python3
# pylint: disable=invalid-name, logging-not-lazy, too-many-locals, logging-fstring-interpolation, too-many-lines
# pylint: disable=invalid-name, logging-not-lazy, too-many-locals, logging-fstring-interpolation, too-many-lines, too-many-public-methods
"""
Device specific handeling for NetBox to Zabbix
"""
@ -9,12 +9,11 @@ from copy import deepcopy
from logging import getLogger
from zabbix_utils import APIRequestError
from modules.exceptions import (SyncInventoryError, TemplateError, SyncExternalError,
InterfaceConfigError, JournalError, UsermacroError)
InterfaceConfigError, JournalError)
from modules.interface import ZabbixInterface
from modules.usermacros import ZabbixUsermacros
from modules.hostgroups import Hostgroup
from modules.tools import field_mapper
from pprint import pprint
try:
from config import (
@ -73,7 +72,7 @@ class PhysicalDevice():
def _inventory_map(self):
""" Use device inventory maps """
return device_inventory_map
def _usermacro_map(self):
""" Use device inventory maps """
return device_usermacro_map
@ -197,31 +196,6 @@ class PhysicalDevice():
if inventory_sync and self.inventory_mode in [0,1]:
self.logger.debug(f"Host {self.name}: Starting inventory mapper")
self.inventory = field_mapper(self.name, self._inventory_map(), nbdevice, self.logger)
# # Let's build an inventory dict for each property in the inventory_map
# for nb_inv_field, zbx_inv_field in self._inventory_map().items():
# field_list = nb_inv_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)):
# self.inventory[zbx_inv_field] = str(value)
# elif not value:
# # empty value should just be an empty string for API compatibility
# self.logger.debug(f"Host {self.name}: NetBox inventory lookup for "
# f"'{nb_inv_field}' returned an empty value")
# self.inventory[zbx_inv_field] = ""
# else:
# # Value is not a string or numeral, probably not what the user expected.
# self.logger.error(f"Host {self.name}: Inventory lookup for '{nb_inv_field}'"
# " returned an unexpected type: it will be skipped.")
# self.logger.debug(f"Host {self.name}: Inventory mapping complete. "
# f"Mapped {len(list(filter(None, self.inventory.values())))} field(s)")
return True
def isCluster(self):
@ -375,14 +349,19 @@ class PhysicalDevice():
self.logger.warning(message)
raise SyncInventoryError(message) from e
def setUsermacros(self):
# Initiate Usermacros class
macros = ZabbixUsermacros(self.nb.config_context, self._usermacro_map())
if macros.sync == False:
return []
else:
return macros.generate()
def set_usermacros(self):
"""
Generates Usermacros
"""
macros = ZabbixUsermacros(self.nb, self._usermacro_map(),
usermacro_sync, logger=self.logger,
host=self.name)
if macros.sync is False:
self.usermacros = []
self.usermacros = macros.generate()
return True
def setProxy(self, proxy_list):
"""
Sets proxy or proxy group if this
@ -443,8 +422,6 @@ class PhysicalDevice():
groups = [{"groupid": self.group_id}]
# Set Zabbix proxy if defined
self.setProxy(proxies)
# Set usermacros
self.usermacros = self.setUsermacros()
# Set basic data for host creation
create_data = {"host": self.name,
"name": self.visible_name,
@ -571,7 +548,7 @@ class PhysicalDevice():
selectHostGroups=["groupid"],
selectParentTemplates=["templateid"],
selectInventory=list(self._inventory_map().values()),
selectMacros=["macro","value","type","description"]
selectMacros=["macro","value","type","description"]
)
if len(host) > 1:
e = (f"Got {len(host)} results for Zabbix hosts "
@ -621,9 +598,9 @@ class PhysicalDevice():
if group["groupid"] == self.group_id:
self.logger.debug(f"Host {self.name}: hostgroup in-sync.")
break
else:
self.logger.warning(f"Host {self.name}: hostgroup OUT of sync.")
self.updateZabbixHost(groups={'groupid': self.group_id})
else:
self.logger.warning(f"Host {self.name}: hostgroup OUT of sync.")
self.updateZabbixHost(groups={'groupid': self.group_id})
if int(host["status"]) == self.zabbix_state:
self.logger.debug(f"Host {self.name}: status in-sync.")
@ -697,14 +674,13 @@ class PhysicalDevice():
# Check host usermacros
if usermacro_sync:
macros_filtered = []
self.usermacros = self.setUsermacros()
# Do not re-sync secret usermacros unless sync is set to 'full'
if not str(usermacro_sync).lower() == "full":
if str(usermacro_sync).lower() != "full":
for m in deepcopy(self.usermacros):
if m['type'] == str(1):
# Remove the value as the api doesn't return it
# this will allow us to only update usermacros that don't exist
m.pop('value')
# Remove the value as the api doesn't return it
# this will allow us to only update usermacros that don't exist
m.pop('value')
macros_filtered.append(m)
if host['macros'] == self.usermacros or host['macros'] == macros_filtered:
self.logger.debug(f"Host {self.name}: usermacros in-sync.")

8
modules/tools.py
View File

@ -1,6 +1,5 @@
from logging import getLogger
"""A collection of tools used by several classes"""
def convert_recordset(recordset):
""" Converts netbox RedcordSet to list of dicts. """
recordlist = []
@ -45,7 +44,6 @@ def proxy_prepper(proxy_list, proxy_group_list):
output.append(group)
return output
def field_mapper(host, mapper, nbdevice, logger):
"""
Maps NetBox field data to Zabbix properties.
@ -75,6 +73,6 @@ def field_mapper(host, mapper, nbdevice, logger):
# 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)")
logger.debug(f"Host {host}: Field mapping complete. "
f"Mapped {len(list(filter(None, data.values())))} field(s)")
return data

75
modules/usermacros.py
View File

@ -1,71 +1,71 @@
#!/usr/bin/env python3
# pylint: disable=too-many-instance-attributes, too-many-arguments, too-many-positional-arguments, logging-fstring-interpolation
"""
All of the Zabbix Usermacro related configuration
"""
from re import match
from logging import getLogger
from zabbix_utils import APIRequestError
from modules.exceptions import UsermacroError
from pprint import pprint
try:
from config import (
usermacro_sync,
)
except ModuleNotFoundError:
print("Configuration file config.py not found in main directory."
"Please create the file or rename the config.py.example file to config.py.")
sys.exit(0)
from modules.tools import field_mapper
class ZabbixUsermacros():
"""Class that represents a Zabbix interface."""
def __init__(self, context, usermacro_map, logger=None):
self.context = context
def __init__(self, nb, usermacro_map, usermacro_sync, logger=None, host=None):
self.nb = nb
self.name = host if host else nb.name
self.usermacro_map = usermacro_map
self.logger = logger if logger else getLogger(__name__)
self.usermacros = {}
self.usermacro_sync = usermacro_sync
self.sync = False
self.force_sync = False
self._setConfig()
self._set_config()
def __repr__(self):
return self.name
def __str__(self):
return self.__repr__()
def _setConfig(self):
if str(usermacro_sync).lower() == "full":
def _set_config(self):
"""
Setup class
"""
if str(self.usermacro_sync).lower() == "full":
self.sync = True
self.force_sync = True
elif usermacro_sync:
elif self.usermacro_sync:
self.sync = True
return True
def validate_macro(self, macro_name):
pattern = '\{\$[A-Z0-9\._]*(\:.*)?\}'
"""
Validates usermacro name
"""
pattern = r'\{\$[A-Z0-9\._]*(\:.*)?\}'
return match(pattern, macro_name)
def render_macro(self, macro_name, macro_properties):
"""
Renders a full usermacro from partial input
"""
macro={}
macrotypes={'text': 0, 'secret': 1, 'vault': 2}
if self.validate_macro(macro_name):
macro['macro'] = str(macro_name)
if isinstance(macro_properties, dict):
if not 'value' in macro_properties:
self.logger.error(f'Usermacro {macro_name} has no value, skipping.')
return False
else:
macro['value'] = macro_properties['value']
self.logger.error(f'Usermacro {macro_name} has no value, skipping.')
return False
macro['value'] = macro_properties['value']
if 'type' in macro_properties and macro_properties['type'].lower() in macrotypes:
macro['type'] = str(macrotypes[macro_properties['type']])
else:
macro['type'] = str(0)
if 'description' in macro_properties and isinstance(macro_properties['description'], str):
if ('description' in macro_properties and
isinstance(macro_properties['description'], str)):
macro['description'] = macro_properties['description']
else:
macro['description'] = ""
@ -78,13 +78,24 @@ class ZabbixUsermacros():
self.logger.error(f'Usermacro {macro_name} is not a valid usermacro name, skipping.')
return False
return macro
def generate(self):
"""
Generate full set of Usermacros
"""
macros=[]
if "zabbix" in self.context and "usermacros" in self.context['zabbix']:
for macro, properties in self.context['zabbix']['usermacros'].items():
m = self.render_macro(macro, properties)
pprint(m)
# Parse the field mapper for usermacros
if self.usermacro_map:
self.logger.debug(f"Host {self.nb.name}: Starting usermacro mapper")
field_macros = field_mapper(self.nb.name, self.usermacro_map, self.nb, self.logger)
for macro, value in field_macros.items():
m = self.render_macro(macro, value)
if m:
macros.append(m)
macros.append(m)
# Parse NetBox config context for usermacros
if "zabbix" in self.nb.config_context and "usermacros" in self.nb.config_context['zabbix']:
for macro, properties in self.nb.config_context['zabbix']['usermacros'].items():
m = self.render_macro(macro, properties)
if m:
macros.append(m)
return macros

2
modules/virtual_machine.py
View File

@ -9,8 +9,6 @@ from modules.interface import ZabbixInterface
from modules.exceptions import TemplateError, InterfaceConfigError, SyncInventoryError
try:
from config import (
inventory_sync,
inventory_mode,
vm_inventory_map,
traverse_site_groups,
traverse_regions

4
netbox_zabbix_sync.py
View File

@ -161,7 +161,7 @@ def main(arguments):
try:
vm = VirtualMachine(nb_vm, zabbix, netbox_journals, nb_version,
create_journal, logger)
logger.debug(f"Host {vm.name}: started operations on VM.")
logger.debug(f"Host {vm.name}: Started operations on VM.")
vm.set_vm_template()
# Check if a valid template has been found for this VM.
if not vm.zbx_template_names:
@ -172,6 +172,7 @@ def main(arguments):
if not vm.hostgroup:
continue
vm.set_inventory(nb_vm)
vm.set_usermacros()
# Checks if device is in cleanup state
if vm.status in zabbix_device_removal:
if vm.zabbix_id:
@ -225,6 +226,7 @@ def main(arguments):
if not device.hostgroup:
continue
device.set_inventory(nb_device)
device.set_usermacros()
# Checks if device is part of cluster.
# Requires clustering variable
if device.isCluster() and clustering: