Skip to content
Snippets Groups Projects

Add iBGP workflow and LibreNMS client

Merged Karel van Klink requested to merge feature/add-ibgp-workflow into develop
All threads resolved!
4 files
+ 42
67
Compare changes
  • Side-by-side
  • Inline
Files
4
"""
The LibreNMS module interacts with the LibreNMS instance when
"""The LibreNMS module interacts with the LibreNMS instance when
- Creating a device.
- Validating the input of a device.
- Terminating a device.
"""
import json
import logging
import requests
from gso import settings
from gso import settings
logger = logging.getLogger(__name__)
class CfgStruct(object):
class CfgStruct:
pass
def _get_cfg():
"""
Internal function to retrieve all needed configuration.
"""
"""Internal function to retrieve all needed configuration."""
oss = settings.load_oss_params()
cfg = CfgStruct()
# Hack for later ease: 1st setattr will fill in the inner's dict
setattr(cfg, "_hack", "")
cfg._hack = ""
# Update inner dict
cfg.__dict__.update(oss.MONITORING)
assert cfg.__dict__ is not None
@@ -51,8 +49,7 @@ def _get_cfg():
def validate_device(fqdn: str):
"""
Function that validates the existence of a device in LibreNMS.
"""Function that validates the existence of a device in LibreNMS.
:param FQDN of the device to validate.
"""
@@ -63,10 +60,8 @@ def validate_device(fqdn: str):
CFG.url_devices, headers=CFG.headers)
assert nms_result is not None
device_id = list(map(
lambda x: x.get("device_id"),
filter(lambda x: x.get("hostname") == fqdn,
nms_result.json().get("devices"))))
device_id = [x.get("device_id") for x in filter(lambda x: x.get("hostname") == fqdn,
nms_result.json().get("devices"))]
if len(device_id) != 1 or device_id[0] is None:
error_msg = f"Device with FQDN={fqdn} is not registered in LibreNMS"
@@ -96,8 +91,7 @@ def validate_device(fqdn: str):
def register_device(fqdn: str):
"""
Function that registers a new device in LibreNMS.
"""Function that registers a new device in LibreNMS.
:param FQDN of the device to register.
"""
@@ -115,7 +109,7 @@ def register_device(fqdn: str):
}
if CFG.SNMP.version == "v2c":
device_data.update({
"community": CFG.SNMP.V2.community
"community": CFG.SNMP.V2.community,
})
elif CFG.SNMP.version == "v3":
@@ -138,8 +132,7 @@ def register_device(fqdn: str):
def deregister_device(fqdn: str):
"""
Function that reregisters a device from LibreNMS.
"""Function that reregisters a device from LibreNMS.
:param FQDN of the device to deregister.
"""
@@ -149,10 +142,8 @@ def deregister_device(fqdn: str):
nms_result = requests.get(
CFG.url_devices, headers=CFG.headers)
assert nms_result is not None
device_id = list(map(
lambda x: x.get("device_id"),
filter(lambda x: x.get("hostname") == fqdn,
nms_result.json().get("devices"))))
device_id = [x.get("device_id") for x in filter(lambda x: x.get("hostname") == fqdn,
nms_result.json().get("devices"))]
if len(device_id) != 1:
return
device_id = device_id[0]
@@ -160,7 +151,7 @@ def deregister_device(fqdn: str):
# https://docs.librenms.org/API/Devices/#endpoint-categories
device_data = {
"field": "disabled",
"data": "1"
"data": "1",
}
url_device = f"{CFG.url_devices}/{device_id}"
logger.debug(f"Connecting to URL: {url_device}"
Loading