Skip to content
Snippets Groups Projects
Commit f2faa18c authored by Neda Moeini's avatar Neda Moeini Committed by Neda Moeini
Browse files

Improved Netbox codebase and added credential to config file.

parent 8022ed0d
Branches
Tags
1 merge request!77Netbox integration including intial CLI for populating base data and ...
......@@ -2,9 +2,9 @@
"GENERAL": {
"public_hostname": "https://gap.geant.org"
},
"RESOURCE_MANAGEMENT": {
"api": "https://127.0.0.1:8000"
"token": "TOKEN"
"NETBOX": {
"api": "https://127.0.0.1:8000",
"token": "TOKEN"
},
"IPAM": {
"INFOBLOX": {
......
......@@ -3,14 +3,15 @@ This module contains all methods
to communicate with the
NetBox API endpoint: Data Center Infrastructure Main (dcim)
"""
from typing import Optional
from uuid import UUID
import pynetbox
import pydantic
from pydantic import BaseModel
import pynetbox
from gso.products import Router
from gso.settings import NetBoxParams
from gso.utils.device_info import TierInfo, FEASIBLE_LAG_RANGE
from gso.utils.exceptions import NotFoundError, WorkflowStateException
# Define device models
......@@ -58,281 +59,188 @@ class Site(pydantic.BaseModel):
slug: str
class ModuleInfo(BaseModel):
device_type: str
module_bays_slots: list[int]
module_type: str
breakout_interfaces_per_slot: list[int]
total_10g_interfaces: int
class TierInfo:
class NetBoxClient:
def __init__(self):
self.Tier1 = ModuleInfo(
device_type="7750-SR7s",
module_bays_slots=[1, 2],
module_type="XCM2s-XMA2s-36p-800g",
breakout_interfaces_per_slot=[36, 35, 34, 33],
total_10g_interfaces=80,
)
self.Tier2 = ModuleInfo(
device_type="7750-SR7s",
module_bays_slots=[1, 2],
module_type="XCM2s-XMA2s-36p-400g",
breakout_interfaces_per_slot=[36, 35, 34, 33],
total_10g_interfaces=60,
)
def get_module_by_name(self, name: str) -> Optional[ModuleInfo]:
if name == "Tier1":
return self.Tier1
elif name == "Tier2":
return self.Tier2
else:
return None
FEASIBLE_LAG_RANGE = range(1, 11)
self.netbox = pynetbox.api(NetBoxParams.api, NetBoxParams.token)
# An exception for not found search
class NotFoundError(Exception):
"""Exception raised for not found search."""
pass
def get_all_devices(self):
return list(self.netbox.dcim.devices.all())
def get_device_by_name(self, device_name: str):
"""Returns the device object by name from netbox, or None if not found."""
return self.netbox.dcim.devices.get(name=device_name)
# An exception on a workflow error
class WorkflowStateException(Exception):
"""Exception raised on problems during workflow."""
pass
def get_interfaces_by_device(self, device_name: str, speed: str):
"""Get all interfaces of a device by name and speed."""
device = self.get_device_by_name(device_name)
return list(
self.netbox.dcim.interfaces.filter(device_id=device.id, enabled=False, mark_connected=False, speed=speed)
)
class NetBoxClient:
token = "6762782659eba4eb2a490716093dba9f2fc31b36"
api = "http://localhost:8000/"
def connect(self):
def create_interface(self, iface_name: str, type: str, speed: str, device_name: str) -> dict:
"""
Creates a netbox client to communicate
with the NetBox endpoints
The responses of the client are returned mostly as Records or RecordSet.
For more details see the dcim.py file in the
pynetbox package.
To create the client a main URL to NetBox is required. Something like:
api = 'http://127.0.0.1:8001'
To create a token, login to the NetBox web frontend or
post request directly a token from the NetBox API:
http://127.0.0.1:8001/api/users/tokens/provision/
Creates a new interface on
a device, where device is defined by name.
The type parameter can be 1000base-t, 10gbase-t, lag...
For more details on type definidtion have a look in
choises.py in the netbox API implementation in module dcim.
Returns the new interface object as dict.
"""
device = self.get_device_by_name(device_name)
return pynetbox.api(self.api, self.token)
def get_all_devices(nbclient):
return list(nbclient.dcim.devices.all())
def get_device_by_name(nbclient, device_name: str) -> dict | None:
""" Returns the device object by name from netbox, or None if not found."""
return nbclient.dcim.devices.get(name=device_name)
# get all interfaces for a device
def get_interfaces_by_device(nbclient, device_name: str, speed: str):
device = get_device_by_name(nbclient, device_name)
return list(nbclient.dcim.interfaces.filter(device_id=device.id,
enabled=False,
mark_connected=False,
speed=speed
))
# Create a interface
def create_interface(nbclient,
iface_name: str,
type: str,
speed: str,
device_name: str) -> dict:
"""
Creates a new interface on
a device, where device is defined by name.
The type parameter can be 1000base-t, 10gbase-t, lag...
For more details on type definidtion have a look in
choises.py in the netbox API implementation in module dcim.
Returns the new interface object as dict.
"""
device = get_device_by_name(nbclient, device_name)
new_iface = nbclient.dcim.interfaces.create(name=iface_name,
type=type,
speed=speed,
enabled=False,
mark_connected=False,
device=device.id)
return dict(new_iface)
def create_device_type(nbclient, manufacturer: str, model: str, slug: str) -> dict:
# First get manufacturer id
manufacturer_id = int(nbclient.dcim.manufacturers.get(name=manufacturer).id)
device_type = DeviceType(**{"manufacturer": manufacturer_id, "model": model, "slug": slug})
# Convert the format from DeviceType to dict
device_type = dict(device_type)
new_iface = self.netbox.dcim.interfaces.create(
name=iface_name, type=type, speed=speed, enabled=False, mark_connected=False, device=device.id
)
# Create device type
device_type = nbclient.dcim.device_types.create(device_type)
return dict(device_type)
return dict(new_iface)
def create_device_type(self, manufacturer: str, model: str, slug: str) -> dict:
"""Creates a new device type in netbox."""
def create_device_role(nbclient, name: str, slug: str) -> dict:
device_role = DeviceRole(**{"name": name, "slug": slug})
device_role = dict(device_role)
device_role = nbclient.dcim.device_roles.create(device_role)
return dict(device_role)
# First get manufacturer id
manufacturer_id = int(self.netbox.dcim.manufacturers.get(name=manufacturer).id)
device_type = DeviceType(**{"manufacturer": manufacturer_id, "model": model, "slug": slug})
# Convert the format from DeviceType to dict
device_type = dict(device_type)
def create_device_site(nbclient, name, slug) -> dict:
device_site = Site(**{"name": name, "slug": slug})
device_site = nbclient.dcim.sites.create(dict(device_site))
return dict(device_site)
# Create device type
device_type = self.netbox.dcim.device_types.create(device_type)
return dict(device_type)
def create_device_role(self, name: str, slug: str) -> dict:
device_role = DeviceRole(**{"name": name, "slug": slug})
device_role = dict(device_role)
device_role = self.netbox.dcim.device_roles.create(device_role)
return dict(device_role)
def create_device_manufacturer(nbclient, name: str, slug: str) -> dict:
device_manufacturer = Manufacturer(**{"name": name, "slug": slug})
device_manufacturer = dict(device_manufacturer)
device_manufacturer = nbclient.dcim.manufacturers.create(device_manufacturer)
return dict(device_manufacturer)
def create_device_site(self, name, slug) -> dict:
device_site = Site(**{"name": name, "slug": slug})
device_site = self.netbox.dcim.sites.create(dict(device_site))
return dict(device_site)
def create_device_manufacturer(self, name: str, slug: str) -> dict:
device_manufacturer = Manufacturer(**{"name": name, "slug": slug})
device_manufacturer = dict(device_manufacturer)
device_manufacturer = self.netbox.dcim.manufacturers.create(device_manufacturer)
return dict(device_manufacturer)
def create_device(router_name: str, site_tier: str) -> dict:
# fqdn: str, model: str, device_role: str, site: str
nbclient = NetBoxClient().connect()
def create_device(self, router_name: str, site_tier: str) -> dict:
"""Creates a new device in netbox."""
# Get device type id
tier_info = TierInfo().get_module_by_name(f"Tier{site_tier}")
device_type = nbclient.dcim.device_types.get(model=tier_info.device_type)
# Get device type id
tier_info = TierInfo().get_module_by_name(f"Tier{site_tier}")
device_type = self.netbox.dcim.device_types.get(model=tier_info.device_type)
# Get device role id
device_role = nbclient.dcim.device_roles.get(name="router")
# Get device role id
device_role = self.netbox.dcim.device_roles.get(name="router")
# Get site id
device_site = nbclient.dcim.sites.get(name="Amsterdam")
# Get site id
device_site = self.netbox.dcim.sites.get(name="Amsterdam")
# Create new device
device = nbclient.dcim.devices.create(
name=router_name, device_type=device_type.id, role=device_role.id, site=device_site.id
)
module_bays = list(nbclient.dcim.module_bays.filter(device_id=device.id))
card_type = nbclient.dcim.module_types.get(model=tier_info.module_type)
for module_bay in module_bays:
nbclient.dcim.modules.create(
device=device.id,
module_bay=module_bay.id,
module_type=card_type.id,
status="active",
comments="Installed via pynetbox",
# Create new device
device = self.netbox.dcim.devices.create(
name=router_name, device_type=device_type.id, role=device_role.id, site=device_site.id
)
module_bays = list(self.netbox.dcim.module_bays.filter(device_id=device.id))
card_type = self.netbox.dcim.module_types.get(model=tier_info.module_type)
for module_bay in module_bays:
self.netbox.dcim.modules.create(
device=device.id,
module_bay=module_bay.id,
module_type=card_type.id,
status="active",
comments="Installed via pynetbox",
)
return dict(device)
def delete_device(self, router_name: str):
self.netbox.dcim.devices.get(name=router_name).delete()
return
def attach_interface_to_lag(self, device_name: str, lag_name: str, iface_name: str) -> dict:
"""
Assign a given interface to a lag.
Returns the lag object with the assignend interfaces
"""
# Get device id
device = self.get_device_by_name(device_name)
return dict(device)
def delete_device(router_name: str):
nbclient = NetBoxClient().connect()
nbclient.dcim.devices.get(name=router_name).delete()
return
def attach_interface_to_lag(nbclient, device_name: str, lag_name: str, iface_name: str) -> dict:
"""
Assign a given interface to a lag.
Returns the lag object with the assignend interfaces
"""
# Get device id
device = get_device_by_name(nbclient, device_name)
# Now get interface for device
iface = nbclient.dcim.interfaces.get(name=iface_name, device_id=device.id)
# Get lag
lag = nbclient.dcim.interfaces.get(name=lag_name, device_id=device.id)
# Now get interface for device
iface = self.netbox.dcim.interfaces.get(name=iface_name, device_id=device.id)
# Assign interface to lag
iface.lag = lag.id
# Get lag
lag = self.netbox.dcim.interfaces.get(name=lag_name, device_id=device.id)
# Update interface
updated_iface = nbclient.dcim.interfaces.update(iface)
# Assign interface to lag
iface.lag = lag.id
return dict(updated_iface)
# Update interface
updated_iface = self.netbox.dcim.interfaces.update(iface)
return dict(updated_iface)
def reserve_interface(nbclient, device_name: str, iface_name: str) -> dict:
# First get interface from device
device = get_device_by_name(nbclient, device_name)
interface = nbclient.dcim.interfaces.get(device_id=device.id,
name=iface_name)
def reserve_interface(self, device_name: str, iface_name: str) -> dict:
"""Reserve an interface by enabling it."""
# Reserve interface by enabling it
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# First get interface from device
device = self.get_device_by_name(device_name)
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
# Check if interface is reserved
if interface.enabled:
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already reserved.")
# Reserve interface by enabling it
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# Reserve interface by enabling it
interface.enabled = True
interface.save()
# Check if interface is reserved
if interface.enabled:
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already reserved.")
return dict(interface)
# Reserve interface by enabling it
interface.enabled = True
interface.save()
return dict(interface)
def allocate_interface(nbclient, device_name: str, iface_name: str) -> dict:
# First get interface from device
device = get_device_by_name(nbclient, device_name)
interface = nbclient.dcim.interfaces.get(device_id=device.id,
name=iface_name)
def allocate_interface(self, device_name: str, iface_name: str) -> dict:
"""Allocate an interface by marking it as connected."""
# allocate interface by marking it as connected
# Check if interface is available
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
device = self.get_device_by_name(device_name)
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
# Check if interface is reserved
if interface.mark_connected:
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already allocated.")
# Check if interface is available
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# allocate interface by mark as connected
interface.mark_connected = True
interface.save()
# Check if interface is reserved
if interface.mark_connected:
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already allocated.")
return dict(interface)
# allocate interface by mark as connected
interface.mark_connected = True
interface.save()
return dict(interface)
def get_available_lags(router_id: UUID) -> list[str]:
"""Returns all available lags not assigned to a device."""
def get_available_lags(self, router_id: UUID) -> list[str]:
"""Returns all available lags not assigned to a device."""
nbclient = NetBoxClient().connect()
router_name = Router.from_subscription(router_id).router.router_fqdn
device = get_device_by_name(nbclient, router_name)
router_name = Router.from_subscription(router_id).router.router_fqdn
device = self.get_device_by_name(router_name)
# Get the existing lag interfaces for the device
lag_interface_names = [
interface["name"] for interface in nbclient.dcim.interfaces.filter(device_name=device.id, type="lag")
]
# Get the existing lag interfaces for the device
lag_interface_names = [
interface["name"] for interface in self.netbox.dcim.interfaces.filter(device_name=device.id, type="lag")
]
# Generate all feasible lags
all_feasible_lags = [f"LAG-{i}" for i in FEASIBLE_LAG_RANGE]
# Generate all feasible lags
all_feasible_lags = [f"LAG-{i}" for i in FEASIBLE_LAG_RANGE]
# Find available lags not assigned to the device
available_lags = [lag for lag in all_feasible_lags if lag not in lag_interface_names]
return available_lags
# Find available lags not assigned to the device
available_lags = [lag for lag in all_feasible_lags if lag not in lag_interface_names]
return available_lags
if __name__ == "__main__":
print(dict(create_device_manufacturer("Juniper", "juniper")))
# if __name__ == "__main__":
# print(dict(create_device_manufacturer("Juniper", "juniper")))
......@@ -79,10 +79,11 @@ class ProvisioningProxyParams(BaseSettings):
api_version: int
class ResourceManagementParams(BaseSettings):
"""TO DO: resource management parameters."""
class NetBoxParams(BaseSettings):
"""Parameters for NetBox."""
todo: str
token: str
api: str
class OSSParams(BaseSettings):
......@@ -90,7 +91,7 @@ class OSSParams(BaseSettings):
GENERAL: GeneralParams
IPAM: IPAMParams
RESOURCE_MANAGEMENT: ResourceManagementParams
NETBOX: NetBoxParams
PROVISIONING_PROXY: ProvisioningProxyParams
......
from typing import Optional
from pydantic import BaseModel
class ModuleInfo(BaseModel):
device_type: str
module_bays_slots: list[int]
module_type: str
breakout_interfaces_per_slot: list[int]
total_10g_interfaces: int
class TierInfo:
def __init__(self):
self.Tier1 = ModuleInfo(
device_type="7750-SR7s",
module_bays_slots=[1, 2],
module_type="XCM2s-XMA2s-36p-800g",
breakout_interfaces_per_slot=[36, 35, 34, 33],
total_10g_interfaces=80,
)
self.Tier2 = ModuleInfo(
device_type="7750-SR7s",
module_bays_slots=[1, 2],
module_type="XCM2s-XMA2s-36p-400g",
breakout_interfaces_per_slot=[36, 35, 34, 33],
total_10g_interfaces=60,
)
def get_module_by_name(self, name: str) -> Optional[ModuleInfo]:
if name == "Tier1":
return self.Tier1
elif name == "Tier2":
return self.Tier2
else:
return None
FEASIBLE_LAG_RANGE = range(1, 11)
class NotFoundError(Exception):
"""Exception raised for not found search."""
pass
class WorkflowStateException(Exception):
"""Exception raised on problems during workflow."""
pass
......@@ -15,7 +15,8 @@ from gso.products.product_blocks.router import RouterRole, RouterVendor, generat
from gso.products.product_types.router import RouterInactive, RouterProvisioning
from gso.products.product_types.site import Site
from gso.products.shared import PortNumber
from gso.services import infoblox, provisioning_proxy, subscriptions, netbox_client
from gso.services import infoblox, provisioning_proxy, subscriptions
from gso.services.netbox_client import NetBoxClient
from gso.services.provisioning_proxy import pp_interaction
from gso.workflows.utils import customer_selector, iso_from_ipv4
......@@ -156,7 +157,7 @@ def provision_router_real(subscription: RouterProvisioning, process_id: UUIDstr,
@step("Create NetBox Device")
def create_netbox_device(subscription: RouterProvisioning) -> State:
netbox_client.create_device(subscription.router.router_fqdn, subscription.router.router_site.site_tier)
NetBoxClient().create_device(subscription.router.router_fqdn, subscription.router.router_site.site_tier)
return {
"subscription": subscription
......
......@@ -11,7 +11,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.router import Router
from gso.services import infoblox
from gso.services import netbox_client
from gso.services.netbox_client import NetBoxClient
logger = logging.getLogger(__name__)
......@@ -61,7 +61,7 @@ def remove_config_from_router() -> None:
@step("Remove Device from NetBox")
def remove_device_from_netbox(subscription: Router) -> dict[str, Router]:
netbox_client.delete_device(subscription.router.router_fqdn)
NetBoxClient().delete_device(subscription.router.router_fqdn)
return {"subscription": subscription}
......
......@@ -6,12 +6,7 @@ import click
import pandas as pd
from gso.services import netbox_client as nc
# Define url and token for netbox API
# TODO move to config file
token = "428b68e7cb7afc6e7441879f21c3de6583081e40"
api = 'http://127.0.0.1:8000'
nbclient = nc.connect(api, token)
from gso.services.netbox_client import NetBoxClient
def convert_to_table(data, fields):
......@@ -41,7 +36,7 @@ def create():
@click.option('--site', default="Amsterdam", help='Device role')
def device(fqdn: str, model: str, role: str, site: str):
click.echo(f"Creating device: fqdn={fqdn}, model={model}, role={role}")
new_device = nc.create_device(nbclient, fqdn, model, role, site)
new_device = NetBoxClient().create_device(fqdn, model)
print(new_device)
......@@ -52,7 +47,7 @@ def device(fqdn: str, model: str, role: str, site: str):
@click.option('--fqdn', help='Device where to create interface')
def interface(name: str, type: str, speed: str, fqdn: str):
click.echo(f"Creating interface: name={name}, speed={speed}, fqdn={fqdn}")
new_interface = nc.create_interface(nbclient, name, type, speed, fqdn)
new_interface = NetBoxClient().create_interface(name, type, speed, fqdn)
print(new_interface)
......@@ -61,7 +56,7 @@ def interface(name: str, type: str, speed: str, fqdn: str):
@click.option('--slug', help='Short name for manufacturer')
def manufacturer(name: str, slug: str):
click.echo(f"Creating manufacturer: name={name}")
manufacturer = nc.create_device_manufacturer(nbclient, name, slug)
manufacturer = NetBoxClient().create_device_manufacturer(name, slug)
print(manufacturer)
......@@ -71,7 +66,7 @@ def manufacturer(name: str, slug: str):
@click.option('--slug', help='Short name for manufacturer')
def device_type(manufacturer: str, model: str, slug: str):
click.echo(f"Creating device type: manufacturer={manufacturer} model = {model}")
device_type = nc.create_device_type(nbclient, manufacturer, model, slug)
device_type = NetBoxClient().create_device_type(manufacturer, model, slug)
print(device_type)
......@@ -80,7 +75,7 @@ def device_type(manufacturer: str, model: str, slug: str):
@click.option('--slug', help='Short name for device role')
def device_role(name: str, slug: str):
click.echo(f"Creating device role: name={name}")
device_role = nc.create_device_role(nbclient, name, slug)
device_role = NetBoxClient().create_device_role(name, slug)
print(device_role)
......@@ -89,7 +84,7 @@ def device_role(name: str, slug: str):
@click.option('--slug', help='Short name for device site')
def device_site(name: str, slug: str):
click.echo(f"Creating device site: name={name}")
device_site = nc.create_device_site(nbclient, name, slug)
device_site = NetBoxClient().create_device_site(name, slug)
print(device_site)
......@@ -112,7 +107,7 @@ def list():
@click.option('--speed', default="1000", help='Interface speed to list interfaces (default 1000=1G)')
def interfaces(fqdn: str, speed: str):
click.echo(f"Listing all interfaces for: device with fqdn={fqdn}, speed={speed}")
interface_list = nc.get_interfaces_by_device(nbclient, fqdn, speed)
interface_list = NetBoxClient().get_interfaces_by_device(fqdn, speed)
display_fields = ["name", "enabled", "mark_connected", "custom_fields", "lag", "speed"]
iface_list = []
for iface in interface_list:
......@@ -125,7 +120,7 @@ def interfaces(fqdn: str, speed: str):
@list.command()
def devices():
click.echo("Listing all devices:")
device_list = nc.get_all_devices(nbclient)
device_list = NetBoxClient().get_all_devices()
display_fields = ["name", "device_type"]
devices = []
for device in device_list:
......@@ -151,7 +146,7 @@ def attach():
@click.option('--lag', help='LAG name to attach interface')
def interface(fqdn: str, iface: str, lag: str):
click.echo(f"Attaching interface to lag: device ={fqdn}, interface name={iface} to lag={lag}")
new_iface = nc.attach_interface_to_lag(nbclient, fqdn, lag, iface)
new_iface = NetBoxClient().attach_interface_to_lag(fqdn, lag, iface)
print(new_iface)
......@@ -169,7 +164,7 @@ def reserve():
@click.option('--iface', help='Interface name to reserve')
def interface(fqdn: str, iface: str):
click.echo(f"Reserving interface: device ={fqdn}, interface name={iface}")
reserved_iface = nc.reserve_interface(nbclient, fqdn, iface)
reserved_iface = NetBoxClient().reserve_interface(fqdn, iface)
print(reserved_iface )
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment