mirror/netbox-sync
1
0
Fork 0
mirror of https://github.com/bb-Ricardo/netbox-sync.git synced 2026-05-08 11:00:12 -05:00
Code Issues Packages Projects Releases Wiki Activity

cluster site now optional, added scope type option. incomplete

Browse Source
This commit is contained in:
Noah
2025-08-21 13:27:53 +10:00
committed by Andrew Foster
parent 1320d8105c
commit 71a8dbf4d9
3 changed files with 177 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/netbox/object_classes.py
+47 -33
View File
@@ -1272,6 +1272,20 @@ class NetBoxObject:
if isinstance(this_site, dict):
return this_site.get("name")
# def get_scope_type(self, data=None):
# this_data_set = data
# if this_data_set is None:
# this_data_set = self.data
# return this_data_set.get("scope_type")
# def get_scope_id(self, data=None):
# this_data_set = data
# if this_data_set is None:
# this_data_set = self.data
# return this_data_set.get("scope_id")
class NBObjectList(list):
@@ -1424,39 +1438,39 @@ class NBTenant(NetBoxObject):
super().__init__(*args, **kwargs)
# class NBLocation(NetBoxObject):
# name = "location"
# api_path = "dcim/locations"
# object_type = "dcim.location"
# primary_key = "name"
# prune = False
# read_only = True
#
# def __init__(self, *args, **kwargs):
# self.data_model = {
# "name": 100,
# "slug": 100,
# "site": NBSite,
# "tags": NBTagList
# }
# super().__init__(*args, **kwargs)
#
#
# class NBRegion(NetBoxObject):
# name = "region"
# api_path = "dcim/regions"
# object_type = "dcim.region"
# primary_key = "name"
# prune = False
# read_only = True
#
# def __init__(self, *args, **kwargs):
# self.data_model = {
# "name": 100,
# "slug": 100,
# "tags": NBTagList
# }
# super().__init__(*args, **kwargs)
class NBLocation(NetBoxObject):
name = "location"
api_path = "dcim/locations"
object_type = "dcim.location"
primary_key = "name"
prune = False
read_only = True
def __init__(self, *args, **kwargs):
self.data_model = {
"name": 100,
"slug": 100,
"site": NBSite,
"tags": NBTagList
}
super().__init__(*args, **kwargs)
class NBRegion(NetBoxObject):
name = "region"
api_path = "dcim/regions"
object_type = "dcim.region"
primary_key = "name"
prune = False
read_only = True
def __init__(self, *args, **kwargs):
self.data_model = {
"name": 100,
"slug": 100,
"tags": NBTagList
}
super().__init__(*args, **kwargs)
class NBSite(NetBoxObject):
module/sources/vmware/config.py
+18
View File
@@ -143,6 +143,24 @@ class VMWareConfig(ConfigBase):
description="""Same as cluster site but on host level.
If unset it will fall back to cluster_site_relation""",
config_example="nyc02.* = New York, ffm01.* = Frankfurt"),
ConfigOption("cluster_scope_type_relation",
str,
description="""This option defines the scope type for a cluster.
The scope type can be 'site', 'site-group', 'location' or 'region'.
This is done with a comma separated key = value list.
key: defines a cluster name as regex
value: defines the NetBox scope type name (use quotes if name contains commas)
""",
config_example="Cluster_NYC = site, Cluster_FFM = sitegroup, Cluster_BER = location"),
ConfigOption("cluster_scope_id_relation",
str,
description="""This option defines the scope id for a cluster.
The scope id is the NetBox ID of the scope type.
This is done with a comma separated key = value list.
key: defines a cluster name as regex
value: defines the NetBox scope id (use quotes if name contains commas)
""",
config_example="Cluster_NYC = New York, Cluster_FFM.* = Data Centers, Cluster_BER = Building 1"),
ConfigOption("cluster_tenant_relation",
str,
description="""\
module/sources/vmware/connection.py
+112 -8
View File
@@ -478,11 +478,14 @@ class VMWareHandler(SourceBase):
site_name = self.get_site_name(NBCluster, cluster_name)
if site_name is not None:
log.debug2(f"Found a matching cluster site for {object_name}, using site '{site_name}'")
else:
site_name = self.site_name
log.debug(f"No site relation for {type(object_name)}: '{object_name}' found, using default site '{site_name}'")
# set default site name
if site_name is None:
site_name = self.site_name
log.debug(f"No site relation for '{object_name}' found, using default site '{site_name}'")
# if site_name is None:
# site_name = self.site_name
# log.debug(f"No site relation for '{object_name}' found, using default site '{site_name}'")
# set the site for cluster to None if None-keyword ("<NONE>") is set via cluster_site_relation
if object_type == NBCluster and site_name == "<NONE>":
@@ -490,7 +493,94 @@ class VMWareHandler(SourceBase):
log.debug2(f"Site relation for '{object_name}' set to None")
return site_name
def get_scope_type(self, object_type, object_name):
"""
Retrieve the scope_type for a NBCluster instance by object name or from the config option
cluster_scope_type_relation
Note: Only NBCluster is supported as the object_type.
Parameters
----------
object_type: object type
The NetBox object type (must be NBCluster).
object_name: str
The name of the object to look up.
Returns
-------
str or None: scope type if one is found, otherwise None
"""
# Validate object type
if object_type != NBCluster:
raise ValueError(f"Object type must be '{NBCluster.name}'.")
relation_name = "cluster_scope_type_relation"
scope_type = self.get_object_relation(object_name, relation_name)
object_instance = self.inventory.get_by_data(object_type, data={"name": object_name})
if object_instance is None:
log.debug2(f"No {object_type.name} found with name '{object_name}'.")
return None
if scope_type is None:
scope_type = object_instance.data_model.get("scope_type")
if scope_type is None:
log.debug2(f"No scope type found for {object_instance.get_display_name()}.")
return None
if type(scope_type) is not str:
scope_type_list = scope_type
scope_type = scope_type_list[0] if len(scope_type_list) > 0 else None
return scope_type
def get_scope_id(self, object_type, object_name):
"""
Retrieve the scope_id for a NBCluster instance by object name or from the config option
cluster_scope_id_relation
Note: Only NBCluster is supported as the object_type.
Parameters
----------
object_type: type
The NetBox object type (must be NBCluster).
object_name: str
The name of the object to look up.
Returns
-------
str or None: scope id if one is found, otherwise None
"""
# Validate object type
if object_type != NBCluster:
raise ValueError(f"Object type must be '{NBCluster.name}'.")
relation_name = "cluster_scope_type_relation"
scope_id = self.get_object_relation(object_name, relation_name)
object_instance = self.inventory.get_by_data(object_type, data={"name": object_name})
if object_instance is None:
log.debug2(f"No {object_type.name} found with name '{object_name}'.") # changed log level to warning for testing
return None
if scope_id is None:
scope_id = object_instance.data_model.get("scope_id")
if scope_id is None:
log.debug2(f"No scope id found for {object_instance.get_display_name()}.") # changed log level to warning for testing
return None
if type(scope_id) is not str:
log.debug(f"scope_id is type: {type(scope_id)}, not str")
return None
return scope_id
def get_object_based_on_macs(self, object_type, mac_list=None):
"""
Try to find a NetBox object based on list of MAC addresses.
@@ -1379,7 +1469,14 @@ class VMWareHandler(SourceBase):
self.settings.cluster_exclude_filter) is False:
return
site_name = self.get_site_name(NBCluster, full_cluster_name)
scope_type = self.get_scope_type(NBCluster, full_cluster_name)
if scope_type is None:
scope_type = self.get_scope_type(NBCluster, name)
if scope_type == "dcim.site":
site_name = self.get_site_name(NBCluster, full_cluster_name)
scope_id = self.get_scope_id(NBCluster, full_cluster_name)
if scope_id is None:
scope_id = self.get_scope_id(NBCluster, name)
data = {
"name": name,
@@ -1388,11 +1485,18 @@ class VMWareHandler(SourceBase):
}
if version.parse(self.inventory.netbox_api_version) >= version.parse("4.2.0"):
if site_name is not None:
data["scope_id"] = {"name": site_name}
data["scope_type"] = "dcim.site"
# four scope types here (dcim.site, dcim.location, dcim.region, dcim.sitegroup)
if scope_type is not None:
data["scope_type"] = scope_type
data["scope_id"] = {"name": scope_id}
log.debug(f"Cluster '{full_cluster_name}' (or {name}) has scope type '{scope_type}' "
f"and scope id '{scope_id}'.")
else:
log.debug(f"Cluster '{full_cluster_name}' has no scope type or scope id.")
else:
data["site"] = {"name": site_name}
# old verison has site only (# TODO: required??) --> optional (tested in netbox versions 4.1.11 and 3.7.1)
if site_name is not None:
data["site"] = {"name": site_name}
tenant_name = self.get_object_relation(full_cluster_name, "cluster_tenant_relation")
if tenant_name is not None: