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

Netbox integration including intial CLI for populating base data and ...

Netbox integration including intial CLI for populating base data and  Integrating router creation/termination with NetBox.
parent 53cbe8a2
No related branches found
No related tags found
1 merge request!77Netbox integration including intial CLI for populating base data and ...
from typer import Typer
from gso.cli import netbox
def load_gso_cli(app: Typer) -> None:
from gso.cli import import_sites
app.add_typer(import_sites.app, name="import_sites")
app.add_typer(netbox.app, name="netbox-cli")
import typer
from pynetbox import RequestError
from gso.services.netbox_client import NetBoxClient, create_device_role, create_device_site
app: typer.Typer = typer.Typer()
@app.command()
def netbox_initial_setup() -> None:
"""Initial setup of NetBox.
It includes:
- Creating a default site (GEANT)
- Creating device roles (Router)
"""
typer.echo("Initial setup of NetBox ...")
typer.echo("Connecting to NetBox ...")
try:
nbclient = NetBoxClient().connect()
except RequestError as e:
typer.echo(f"Error connecting to NetBox: {e}")
return
typer.echo("Creating GEANT site ...")
try:
create_device_site(nbclient, "GEANT", "geant")
typer.echo("Site created successfully.")
except RequestError as e:
typer.echo(f"Error creating site: {e}")
typer.echo("Creating Router device role ...")
try:
create_device_role(nbclient, "router", "router")
typer.echo("Device role created successfully.")
except RequestError as e:
typer.echo(f"Error creating device role: {e}")
typer.echo("NetBox initial setup completed successfully.")
# mypy: ignore-errors
"""
This module contains all methods
to communicate with the
NetBox API endpoint: Data Center Infrastructure Main (dcim)
"""
import pynetbox
import pydantic
# Define device models
class Manufacturer(pydantic.BaseModel):
"""
The device manufacturer.
Is mandatory to create a device in netbox.
Is needed to define the device type.
"""
name: str
slug: str
class DeviceType(pydantic.BaseModel):
"""
The device type for a device.
Is mandatory to create a device.
The manufacturer should be created first to get the manufacturer id,
which is defined here as int.
"""
manufacturer: int
model: str
slug: str
class DeviceRole(pydantic.BaseModel):
"""
The role of a device in netbox.
Is mandatory to create a device.
"""
name: str
slug: str
class Site(pydantic.BaseModel):
"""
The site where to place the device.
Is mandatory to create a device.
"""
name: str
slug: str
# An exception for not found search
class NotFoundError(Exception):
"""Exception raised for not found search."""
pass
# An exception on a workflow error
class WorkflowStateException(Exception):
"""Exception raised on problems during workflow."""
pass
def connect(api, token):
"""
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/
"""
return pynetbox.api(api, token=token)
def get_all_devices(nbclient):
return list(nbclient.dcim.devices.all())
# We need sometimes a specific device
# for creating devices
def get_device_by_name(nbclient, device_name):
device = nbclient.dcim.devices.get(name=device_name)
if device is None:
raise NotFoundError(f"Device: {device_name} not found")
else:
return device
# 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)
# Create device type
device_type = nbclient.dcim.device_types.create(device_type)
return dict(device_type)
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)
def create_device_site(nbclient, name: str, slug: str) -> dict:
device_site = Site(**{'name': name, 'slug': slug})
device_site = dict(device_site)
device_site = nbclient.dcim.sites.create(device_site)
return dict(device_site)
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(nbclient,
fqdn: str,
model: str,
device_role: str,
site: str) -> dict:
""" Creates a device and
returns the new device as a dict
"""
# Get device type id
device_type = nbclient.dcim.device_types.get(model=model)
# Get device role id
device_role = nbclient.dcim.device_roles.get(name=device_role)
# Get site id
device_site = nbclient.dcim.sites.get(name=site)
# Create new device
new_device = nbclient.dcim.devices.create(name=fqdn,
device_type=device_type.id,
role=device_role.id,
site=device_site.id)
return dict(new_device)
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)
# Assign interface to lag
iface.lag = lag.id
# Update interface
updated_iface = nbclient.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)
# Reserve interface by enabling it
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# 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
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)
# 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.")
# Check if interface is reserved
if interface.mark_connected:
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already allocated.")
# allocate interface by mark as connected
interface.mark_connected = True
interface.save()
return dict(interface)
if __name__ == "__main__":
print(dict(create_device_manufacturer("Juniper", "juniper")))
"""
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
from gso.products import Router
# Define device models
class Manufacturer(pydantic.BaseModel):
"""
The device manufacturer.
Is mandatory to create a device in netbox.
Is needed to define the device type.
"""
name: str
slug: str
class DeviceType(pydantic.BaseModel):
"""
The device type for a device.
Is mandatory to create a device.
The manufacturer should be created first to get the manufacturer id,
which is defined here as int.
"""
manufacturer: int
model: str
slug: str
class DeviceRole(pydantic.BaseModel):
"""
The role of a device in netbox.
Is mandatory to create a device.
"""
name: str
slug: str
class Site(pydantic.BaseModel):
"""
The site where to place the device.
Is mandatory to create a device.
"""
name: str
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:
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)
# An exception for not found search
class NotFoundError(Exception):
"""Exception raised for not found search."""
pass
# An exception on a workflow error
class WorkflowStateException(Exception):
"""Exception raised on problems during workflow."""
pass
class NetBoxClient:
token = "6762782659eba4eb2a490716093dba9f2fc31b36"
api = "http://localhost:8000/"
def connect(self):
"""
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/
"""
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)
# Create device type
device_type = nbclient.dcim.device_types.create(device_type)
return dict(device_type)
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)
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)
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(router_name: str, site_tier: str) -> dict:
# fqdn: str, model: str, device_role: str, site: str
nbclient = NetBoxClient().connect()
# 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 role id
device_role = nbclient.dcim.device_roles.get(name="router")
# Get site id
device_site = nbclient.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",
)
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)
# Assign interface to lag
iface.lag = lag.id
# Update interface
updated_iface = nbclient.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)
# Reserve interface by enabling it
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# 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
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)
# 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.")
# Check if interface is reserved
if interface.mark_connected:
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already allocated.")
# 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."""
nbclient = NetBoxClient().connect()
router_name = Router.from_subscription(router_id).router.router_fqdn
device = get_device_by_name(nbclient, 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")
]
# 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
if __name__ == "__main__":
print(dict(create_device_manufacturer("Juniper", "juniper")))
......@@ -15,7 +15,7 @@ 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
from gso.services import infoblox, provisioning_proxy, subscriptions, netbox_client
from gso.services.provisioning_proxy import pp_interaction
from gso.workflows.utils import customer_selector, iso_from_ipv4
......@@ -145,12 +145,21 @@ def provision_router_dry(subscription: RouterProvisioning, process_id: UUIDstr,
def provision_router_real(subscription: RouterProvisioning, process_id: UUIDstr, tt_number: str) -> State:
provisioning_proxy.provision_router(subscription, process_id, tt_number, False)
return {
return ({
"subscription": subscription,
"label_text": (
"Deployment of base config for a new router. Deployment is being taken care of by the"
" provisioning proxy, please wait for the results to come back before continuing."
),
})
@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)
return {
"subscription": subscription
}
......@@ -211,6 +220,7 @@ def create_router() -> StepList:
>> pp_interaction(provision_router_real, 3)
>> verify_ipam_loopback
>> should_allocate_ias(verify_ipam_ias)
>> create_netbox_device
>> set_status(SubscriptionLifecycle.ACTIVE)
>> resync
>> done
......
......@@ -11,6 +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
logger = logging.getLogger(__name__)
......@@ -58,6 +59,12 @@ def remove_config_from_router() -> None:
pass
@step("Remove Device from NetBox")
def remove_device_from_netbox(subscription: Router) -> dict[str, Router]:
netbox_client.delete_device(subscription.router.router_fqdn)
return {"subscription": subscription}
@workflow(
"Terminate router",
initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
......@@ -78,6 +85,7 @@ def terminate_router() -> StepList:
>> unsync
>> run_ipam_steps(ipam_steps)
>> run_config_steps(remove_config_from_router)
>> remove_device_from_netbox
>> set_status(SubscriptionLifecycle.TERMINATED)
>> resync
>> done
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment