NeoAnd/netbox-zabbix-sync
1
0
Fork 0
mirror of https://github.com/TheNetworkGuy/netbox-zabbix-sync.git synced 2025-07-13 15:24:48 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #6 from q1x/new-ghcr-workflow

Browse Source 
New ghcr workflow
This commit is contained in:
Raymond Kuiper 2025-02-20 11:29:16 +01:00 committed by GitHub
commit 48a04c58e3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 702 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
.github/workflows/publish-image.yml vendored
View File

@ -1,46 +1,53 @@
name: Publish Docker image to GHCR on a new version
name: Build and Push Docker Image
permissions:
contents: read
packages: write
on:
push:
branches:
- main
- dockertest
# tags:
# - [0-9]+.*
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}
release:
types: [published]
pull_request:
types: [opened, synchronize]
jobs:
test_quality:
uses: ./.github/workflows/quality.yml
build_and_publish:
build:
runs-on: ubuntu-latest
steps:
- name: Checkout sources
uses: actions/checkout@v4
- name: Log in to the container registry
uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GHCR_PAT }}
- name: Extract metadata (tags, labels)
id: meta
uses: docker/metadata-action@v5
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
tags: |
type=semver,pattern={{ version }}
type=ref,event=branch
type=raw,value=latest,enable=${{ github.ref == format('refs/heads/{0}', 'master') }}
type=sha
- name: Build and push Docker image
uses: docker/build-push-action@v5
with:
context: .
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
- name: Checkout repository
uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@6524bf65af31da8d45b59e8c27de4bd072b392f5
- name: Login to GitHub Container Registry
uses: docker/login-action@9780b0c442fbb1117ed29e0efdff1e18412f7567
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Extract metadata
id: meta
uses: docker/metadata-action@369eb591f429131d6889c46b94e711f089e6ca96
with:
images: ghcr.io/${{ github.repository }}
tags: |
type=ref,event=branch
type=ref,event=pr
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}}
- name: Build and push Docker image
uses: docker/build-push-action@ca877d9245402d1537745e0e356eab47c3520991
with:
context: .
file: ./Dockerfile
push: true
platforms: linux/amd64,linux/arm64
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
annotations: |
index:org.opencontainers.image.description=Python script to synchronise NetBox devices to Zabbix.

2
.gitignore vendored
View File

@ -1,6 +1,8 @@
*.log
.venv
config.py
Pipfile
Pipfile.lock
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]

227
README.md
View File

@ -7,7 +7,7 @@ A script to create, update and delete Zabbix hosts using NetBox device objects.
To pull the latest stable version to your local cache, use the following docker
pull command:
```
```bash
docker pull ghcr.io/thenetworkguy/netbox-zabbix-sync:main
```
@ -15,7 +15,7 @@ Make sure to specify the needed environment variables for the script to work
(see [here](#set-environment-variables)) on the command line or use an
[env file](https://docs.docker.com/reference/cli/docker/container/run/#env).
```
```bash
docker run -d -t -i -e ZABBIX_HOST='https://zabbix.local' \
-e ZABBIX_TOKEN='othersecrettoken' \
-e NETBOX_HOST='https://netbox.local' \
@ -30,7 +30,7 @@ The image uses the default `config.py` for it's configuration, you can use a
volume mount in the docker run command to override with your own config file if
needed (see [config file](#config-file)):
```
```bash
docker run -d -t -i -v $(pwd)/config.py:/opt/netbox-zabbix/config.py ...
```
@ -38,7 +38,7 @@ docker run -d -t -i -v $(pwd)/config.py:/opt/netbox-zabbix/config.py ...
### Cloning the repository
```
```bash
git clone https://github.com/TheNetworkGuy/netbox-zabbix-sync.git
```
@ -66,7 +66,7 @@ cp config.py.example config.py
Set the following environment variables:
```
```bash
ZABBIX_HOST="https://zabbix.local"
ZABBIX_USER="username"
ZABBIX_PASS="Password"
@ -77,7 +77,7 @@ NETBOX_TOKEN="secrettoken"
Or, you can use a Zabbix API token to login instead of using a username and
password. In that case `ZABBIX_USER` and `ZABBIX_PASS` will be ignored.
```
```bash
ZABBIX_TOKEN=othersecrettoken
```
@ -183,9 +183,9 @@ used:
| cluster | VM cluster name |
| cluster_type | VM cluster type |
You can specify the value sperated by a "/" like so:
You can specify the value seperated by a "/" like so:
```
```python
hostgroup_format = "tenant/site/dev_location/role"
```
@ -232,7 +232,7 @@ have a relationship with a tenant.
- Device_role: PDU
- Site: HQ-AMS
```
```python
hostgroup_format = "site/tenant/device_role"
```
@ -245,7 +245,7 @@ generated for both hosts:
The same logic applies to custom fields being used in the HG format:
```
```python
hostgroup_format = "site/mycustomfieldname"
```
@ -292,15 +292,18 @@ You can set the inventory mode to "disabled", "manual" or "automatic" with the
[Zabbix Manual](https://www.zabbix.com/documentation/current/en/manual/config/hosts/inventory#building-inventory)
for more information about the modes.
Use the `inventory_map` variable to map which NetBox properties are used in
Use the `device_inventory_map` variable to map which NetBox properties are used in
which Zabbix Inventory fields. For nested properties, you can use the '/'
seperator. For example, the following map will assign the custom field
'mycustomfield' to the 'alias' Zabbix inventory field:
```
For Virtual Machines, use `vm_inventory_map`.
```python
inventory_sync = True
inventory_mode = "manual"
inventory_map = { "custom_fields/mycustomfield/name": "alias"}
device_inventory_map = {"custom_fields/mycustomfield/name": "alias"}
vm_inventory_map = {"custom_fields/mycustomfield/name": "alias"}
```
See `config.py.example` for an extensive example map. Any Zabix Inventory fields
@ -321,14 +324,14 @@ sticking to the custom field.
You can change the behaviour in the config file. By default this setting is
false but you can set it to true to use config context:
```
```python
templates_config_context = True
```
After that make sure that for each host there is at least one template defined
in the config context in this format:
```
```json
{
"zabbix": {
"templates": [
@ -346,10 +349,196 @@ added benefit of overwriting the template should a device in NetBox have a
device specific context defined. In this case the device specific context
template(s) will take priority over the device type custom field template.
```
```python
templates_config_context_overrule = True
```
### Tags
This script can sync host tags to your Zabbix hosts for use in filtering,
SLA calculations and event correlation.
Tags can be synced from the following sources:
1. NetBox device/vm tags
2. NetBox config ontext
3. NetBox fields
Syncing tags will override any tags that were set manually on the host,
making NetBox the single source-of-truth for managing tags.
To enable syncing, turn on tag_sync in the config file.
By default, this script will modify tag names and tag values to lowercase.
You can change this behaviour by setting tag_lower to False.
```python
tag_sync = True
tag_lower = True
```
#### Device tags
As NetBox doesn't follow the tag/value pattern for tags, we will need a tag
name set to register the netwbox tags.
By default the tag name is "NetBox", but you can change this to whatever you want.
The value for the tag can be choosen from 'name', 'display' or 'slug'.
```python
tag_name = 'NetBox'
tag_value = 'name'
```
#### Config context
You can supply custom tags via config context by adding the following:
```json
{
"zabbix": {
"tags": [
{
"MyTagName": "MyTagValue"
},
{
"environment": "production"
}
],
}
}
```
This will allow you to assign tags based on the config context rules.
#### NetBox Field
NetBox field can also be used as input for tags, just like inventory and usermacros.
To enable syncing from fields, make sure to configure a `device_tag_map` and/or a `vm_tag_map`.
```python
device_tag_map = {"site/name": "site",
"rack/name": "rack",
"platform/name": "target"}
vm_tag_map = {"site/name": "site",
"cluster/name": "cluster",
"platform/name": "target"}
```
To turn off field syncing, set the maps to empty dictionaries:
```python
device_tag_map = {}
vm_tag_map = {}
```
### Usermacros
You can choose to use NetBox as a source for Host usermacros by
enabling the following option in the configuration file:
```python
usermacro_sync = True
```
Please be advised that enabling this option will _clear_ any usermacros
manually set on the managed hosts and override them with the usermacros
from NetBox.
There are two NetBox sources that can be used to populate usermacros:
1. NetBox config context
2. NetBox fields
#### Config context
By defining a dictionary `usermacros` within the `zabbix` key in
config context, you can dynamically assign usermacro values based on
anything that you can target based on
[config contexts](https://netboxlabs.com/docs/netbox/en/stable/features/context-data/)
within NetBox.
Through this method, it is possible to define the following types of usermacros:
1. Text
2. Secret
3. Vault
The default macro type is text if no `type` and `value` have been set.
It is also possible to create usermacros with
[context](https://www.zabbix.com/documentation/7.0/en/manual/config/macros/user_macros_context).
Examples:
```json
{
"zabbix": {
"usermacros": {
"{$USER_MACRO}": "test value",
"{$CONTEXT_MACRO:\"test\"}": "test value",
"{$CONTEXT_REGEX_MACRO:regex:\".*\"}": "test value",
"{$SECRET_MACRO}": {
"type": "secret",
"value": "PaSsPhRaSe"
},
"{$VAULT_MACRO}": {
"type": "vault",
"value": "secret/vmware:password"
},
"{$USER_MACRO2}": {
"type": "text",
"value": "another test value"
}
}
}
}
```
Please be aware that secret usermacros are only synced _once_ by default.
This is the default behaviour because Zabbix API won't return the value of
secrets so the script cannot compare the values with the ones set in NetBox.
If you update a secret usermacro value, just remove the value from the host
in Zabbix and the new value will be synced during the next run.
Alternatively, you can set the following option in the config file:
```python
usermacro_sync = "full"
```
This will force a full usermacro sync on every run on hosts that have secret usermacros set.
That way, you will know for sure the secret values are always up to date.
Keep in mind that NetBox (and the log output of this script) will show your secrets
in plain text. If true secrecy is required, consider switching to
[vault](https://www.zabbix.com/documentation/current/en/manual/config/macros/secret_macros#vault-secret)
usermacros.
#### Netbox Fields
To use NetBox fields as a source for usermacros, you will need to set up usermacro maps
for devices and/or virtual machines in the configuration file.
This method only supports `text` type usermacros.
For example:
```python
usermacro_sync = True
device_usermacro_map = {"serial": "{$HW_SERIAL}",
"role/name": "{$DEV_ROLE}",
"url": "{$NB_URL}",
"id": "{$NB_ID}"}
vm_usermacro_map = {"memory": "{$TOTAL_MEMORY}",
"role/name": "{$DEV_ROLE}",
"url": "{$NB_URL}",
"id": "{$NB_ID}"}
```
## Permissions
### NetBox
@ -518,9 +707,13 @@ environment. For example, you could:
}
```
I would recommend using macros for sensitive data such as community strings
I would recommend using usermacros for sensitive data such as community strings
since the data in NetBox is plain-text.
> **_NOTE:_** Not all SNMP data is required for a working configuration.
> [The following parameters are allowed](https://www.zabbix.com/documentation/current/manual/api/reference/hostinterface/object#details_tag "The following parameters are allowed")but
> are not all required, depending on your environment.

85
config.py.example
View File

@ -80,19 +80,74 @@ inventory_sync = False
# For nested properties, you can use the '/' seperator.
# For example, the following map will assign the custom field 'mycustomfield' to the 'alias' Zabbix inventory field:
#
# inventory_map = { "custom_fields/mycustomfield/name": "alias"}
# device_inventory_map = { "custom_fields/mycustomfield/name": "alias"}
#
# The following map should provide some nice defaults:
inventory_map = { "asset_tag": "asset_tag",
"virtual_chassis/name": "chassis",
"status/label": "deployment_status",
"location/name": "location",
"latitude": "location_lat",
"longitude": "location_lon",
"comments": "notes",
"name": "name",
"rack/name": "site_rack",
"serial": "serialno_a",
"device_type/model": "type",
"device_type/manufacturer/name": "vendor",
"oob_ip/address": "oob_ip" }
# The following maps should provide some nice defaults:
device_inventory_map = { "asset_tag": "asset_tag",
"virtual_chassis/name": "chassis",
"status/label": "deployment_status",
"location/name": "location",
"latitude": "location_lat",
"longitude": "location_lon",
"comments": "notes",
"name": "name",
"rack/name": "site_rack",
"serial": "serialno_a",
"device_type/model": "type",
"device_type/manufacturer/name": "vendor",
"oob_ip/address": "oob_ip" }
# We also support inventory mapping on Virtual Machines.
vm_inventory_map = { "status/label": "deployment_status",
"comments": "notes",
"name": "name" }
# To allow syncing of usermacros from NetBox, set to True.
# this will enable both field mapping and config context usermacros.
#
# If set to "full", it will force the update of secret usermacros every run.
# Please see the README.md for more information.
usermacro_sync = False
# device usermacro_map to map NetBox fields to usermacros.
device_usermacro_map = {"serial": "{$HW_SERIAL}",
"role/name": "{$DEV_ROLE}",
"url": "{$NB_URL}",
"id": "{$NB_ID}"}
# virtual machine usermacro_map to map NetBox fields to usermacros.
vm_usermacro_map = {"memory": "{$TOTAL_MEMORY}",
"role/name": "{$DEV_ROLE}",
"url": "{$NB_URL}",
"id": "{$NB_ID}"}
# To sync host tags to Zabbix, set to True.
tag_sync = False
# Setting tag_lower to True will lower capital letters ain tag names and values
# This is more inline with the Zabbix way of working with tags.
#
# You can however set this to False to ensure capital letters are synced to Zabbix tags.
tag_lower = True
# We can sync NetBox device/VM tags to Zabbix, but as NetBox tags don't follow the key/value
# pattern, we need to specify a tag name to register the NetBox tags in Zabbix.
#
#
#
# If tag_name is set to False, we won't sync NetBox device/VM tags to Zabbix.
tag_name = 'NetBox'
# We can choose to use 'name', 'slug' or 'display' NetBox tag properties as a value in Zabbix.
# 'name'is used by default.
tag_value = "name"
# device tag_map to map NetBox fields to host tags.
device_tag_map = {"site/name": "site",
"rack/name": "rack",
"platform/name": "target"}
# Virtual machine tag_map to map NetBox fields to host tags.
vm_tag_map = {"site/name": "site",
"cluster/name": "cluster",
"platform/name": "target"}

119
modules/device.py
View File

@ -1,16 +1,21 @@
#!/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
"""
from os import sys
from re import search
from copy import deepcopy
from logging import getLogger
from zabbix_utils import APIRequestError
from modules.exceptions import (SyncInventoryError, TemplateError, SyncExternalError,
InterfaceConfigError, JournalError)
from modules.interface import ZabbixInterface
from modules.usermacros import ZabbixUsermacros
from modules.tags import ZabbixTags
from modules.hostgroups import Hostgroup
from modules.tools import field_mapper, remove_duplicates
try:
from config import (
template_cf, device_cf,
@ -18,7 +23,14 @@ try:
traverse_regions,
inventory_sync,
inventory_mode,
inventory_map
device_inventory_map,
usermacro_sync,
device_usermacro_map,
tag_sync,
tag_lower,
tag_name,
tag_value,
device_tag_map
)
except ModuleNotFoundError:
print("Configuration file config.py not found in main directory."
@ -53,6 +65,8 @@ class PhysicalDevice():
self.nb_journals = nb_journal_class
self.inventory_mode = -1
self.inventory = {}
self.usermacros = {}
self.tags = {}
self.logger = logger if logger else getLogger(__name__)
self._setBasics()
@ -62,6 +76,18 @@ class PhysicalDevice():
def __str__(self):
return self.__repr__()
def _inventory_map(self):
""" Use device inventory maps """
return device_inventory_map
def _usermacro_map(self):
""" Use device inventory maps """
return device_usermacro_map
def _tag_map(self):
""" Use device host tag maps """
return device_tag_map
def _setBasics(self):
"""
Sets basic information like IP address.
@ -180,31 +206,7 @@ class PhysicalDevice():
self.inventory = {}
if inventory_sync and self.inventory_mode in [0,1]:
self.logger.debug(f"Host {self.name}: Starting inventory mapper")
# Let's build an inventory dict for each property in the inventory_map
for nb_inv_field, zbx_inv_field in 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)")
self.inventory = field_mapper(self.name, self._inventory_map(), nbdevice, self.logger)
return True
def isCluster(self):
@ -358,6 +360,34 @@ class PhysicalDevice():
self.logger.warning(message)
raise SyncInventoryError(message) from e
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 set_tags(self):
"""
Generates Host Tags
"""
tags = ZabbixTags(self.nb, self._tag_map(),
tag_sync, tag_lower, tag_name=tag_name,
tag_value=tag_value, logger=self.logger,
host=self.name)
if tags.sync is False:
self.tags = []
self.tags = tags.generate()
return True
def setProxy(self, proxy_list):
"""
Sets proxy or proxy group if this
@ -427,7 +457,9 @@ class PhysicalDevice():
"templates": templateids,
"description": description,
"inventory_mode": self.inventory_mode,
"inventory": self.inventory
"inventory": self.inventory,
"macros": self.usermacros,
"tags": self.tags
}
# If a Zabbix proxy or Zabbix Proxy group has been defined
if self.zbxproxy:
@ -542,7 +574,10 @@ class PhysicalDevice():
selectGroups=["groupid"],
selectHostGroups=["groupid"],
selectParentTemplates=["templateid"],
selectInventory=list(inventory_map.values()))
selectInventory=list(self._inventory_map().values()),
selectMacros=["macro","value","type","description"],
selectTags=["tag","value"]
)
if len(host) > 1:
e = (f"Got {len(host)} results for Zabbix hosts "
f"with ID {self.zabbix_id} - hostname {self.name}.")
@ -591,7 +626,6 @@ 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})
@ -664,6 +698,31 @@ class PhysicalDevice():
self.logger.warning(f"Host {self.name}: inventory OUT of sync.")
self.updateZabbixHost(inventory=self.inventory)
# Check host usermacros
if usermacro_sync:
macros_filtered = []
# Do not re-sync secret usermacros unless sync is set to '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')
macros_filtered.append(m)
if host['macros'] == self.usermacros or host['macros'] == macros_filtered:
self.logger.debug(f"Host {self.name}: usermacros in-sync.")
else:
self.logger.warning(f"Host {self.name}: usermacros OUT of sync.")
self.updateZabbixHost(macros=self.usermacros)
# Check host usermacros
if tag_sync:
if remove_duplicates(host['tags'],sortkey='tag') == self.tags:
self.logger.debug(f"Host {self.name}: tags in-sync.")
else:
self.logger.warning(f"Host {self.name}: tags OUT of sync.")
self.updateZabbixHost(tags=self.tags)
# If only 1 interface has been found
# pylint: disable=too-many-nested-blocks
if len(host['interfaces']) == 1:

3
modules/exceptions.py
View File

@ -31,3 +31,6 @@ class HostgroupError(SyncError):
class TemplateError(SyncError):
""" Class TemplateError """
class UsermacroError(SyncError):
""" Class UsermacroError """

117
modules/tags.py Normal file
View File

@ -0,0 +1,117 @@
#!/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 logging import getLogger
from modules.tools import field_mapper, remove_duplicates
class ZabbixTags():
"""Class that represents a Zabbix interface."""
def __init__(self, nb, tag_map, tag_sync, tag_lower=True,
tag_name=None, tag_value=None, logger=None, host=None):
self.nb = nb
self.name = host if host else nb.name
self.tag_map = tag_map
self.logger = logger if logger else getLogger(__name__)
self.tags = {}
self.lower = tag_lower
self.tag_name = tag_name
self.tag_value = tag_value
self.tag_sync = tag_sync
self.sync = False
self._set_config()
def __repr__(self):
return self.name
def __str__(self):
return self.__repr__()
def _set_config(self):
"""
Setup class
"""
if self.tag_sync:
self.sync = True
return True
def validate_tag(self, tag_name):
"""
Validates tag name
"""
if tag_name and isinstance(tag_name, str) and len(tag_name)<=256:
return True
return False
def validate_value(self, tag_value):
"""
Validates tag value
"""
if tag_value and isinstance(tag_value, str) and len(tag_value)<=256:
return True
return False
def render_tag(self, tag_name, tag_value):
"""
Renders a tag
"""
tag={}
if self.validate_tag(tag_name):
if self.lower:
tag['tag'] = tag_name.lower()
else:
tag['tag'] = tag_name
else:
self.logger.error(f'Tag {tag_name} is not a valid tag name, skipping.')
return False
if self.validate_value(tag_value):
if self.lower:
tag['value'] = tag_value.lower()
else:
tag['value'] = tag_value
else:
self.logger.error(f'Tag {tag_name} has an invalid value: \'{tag_value}\', skipping.')
return False
return tag
def generate(self):
"""
Generate full set of Usermacros
"""
# pylint: disable=too-many-branches
tags=[]
# Parse the field mapper for tags
if self.tag_map:
self.logger.debug(f"Host {self.nb.name}: Starting tag mapper")
field_tags = field_mapper(self.nb.name, self.tag_map, self.nb, self.logger)
for tag, value in field_tags.items():
t = self.render_tag(tag, value)
if t:
tags.append(t)
# Parse NetBox config context for tags
if ("zabbix" in self.nb.config_context and "tags" in self.nb.config_context['zabbix']
and isinstance(self.nb.config_context['zabbix']['tags'], list)):
for tag in self.nb.config_context['zabbix']['tags']:
if isinstance(tag, dict):
for tagname, value in tag.items():
t = self.render_tag(tagname, value)
if t:
tags.append(t)
# Pull in NetBox device tags if tag_name is set
if self.tag_name and isinstance(self.tag_name, str):
for tag in self.nb.tags:
if self.tag_value.lower() in ['display', 'name', 'slug']:
value = tag[self.tag_value]
else:
value = tag['name']
t = self.render_tag(self.tag_name, value)
if t:
tags.append(t)
return remove_duplicates(tags, sortkey='tag')

45
modules/tools.py
View File

@ -1,4 +1,5 @@
"""A collection of tools used by several classes"""
def convert_recordset(recordset):
""" Converts netbox RedcordSet to list of dicts. """
recordlist = []
@ -42,3 +43,47 @@ def proxy_prepper(proxy_list, proxy_group_list):
group["monitored_by"] = 2
output.append(group)
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
def remove_duplicates(input_list, sortkey=None):
"""
Removes duplicate entries from a list and sorts the list
"""
output_list = []
if isinstance(input_list, list):
output_list = [dict(t) for t in {tuple(d.items()) for d in input_list}]
if sortkey and isinstance(sortkey, str):
output_list.sort(key=lambda x: x[sortkey])
return output_list

101
modules/usermacros.py Normal file
View File

@ -0,0 +1,101 @@
#!/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 modules.tools import field_mapper
class ZabbixUsermacros():
"""Class that represents a Zabbix interface."""
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._set_config()
def __repr__(self):
return self.name
def __str__(self):
return self.__repr__()
def _set_config(self):
"""
Setup class
"""
if str(self.usermacro_sync).lower() == "full":
self.sync = True
self.force_sync = True
elif self.usermacro_sync:
self.sync = True
return True
def validate_macro(self, macro_name):
"""
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
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)):
macro['description'] = macro_properties['description']
else:
macro['description'] = ""
elif isinstance(macro_properties, str):
macro['value'] = macro_properties
macro['type'] = str(0)
macro['description'] = ""
else:
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=[]
# 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)
# 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

15
modules/virtual_machine.py
View File

@ -9,6 +9,9 @@ from modules.interface import ZabbixInterface
from modules.exceptions import TemplateError, InterfaceConfigError, SyncInventoryError
try:
from config import (
vm_inventory_map,
vm_usermacro_map,
vm_tag_map,
traverse_site_groups,
traverse_regions
)
@ -24,6 +27,18 @@ class VirtualMachine(PhysicalDevice):
self.hostgroup = None
self.zbx_template_names = None
def _inventory_map(self):
""" use VM inventory maps """
return vm_inventory_map
def _usermacro_map(self):
""" use VM usermacro maps """
return vm_usermacro_map
def _tag_map(self):
""" use VM tag maps """
return vm_tag_map
def set_hostgroup(self, hg_format, nb_site_groups, nb_regions):
"""Set the hostgroup for this device"""
# Create new Hostgroup instance

7
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:
@ -171,6 +171,9 @@ def main(arguments):
# Check if a valid hostgroup has been found for this VM.
if not vm.hostgroup:
continue
vm.set_inventory(nb_vm)
vm.set_usermacros()
vm.set_tags()
# Checks if device is in cleanup state
if vm.status in zabbix_device_removal:
if vm.zabbix_id:
@ -224,6 +227,8 @@ def main(arguments):
if not device.hostgroup:
continue
device.set_inventory(nb_device)
device.set_usermacros()
device.set_tags()
# Checks if device is part of cluster.
# Requires clustering variable
if device.isCluster() and clustering: