diff --git a/gso/__init__.py b/gso/__init__.py
index 464bffa920d1ab17883cdb90bfe1a4d31f0c83a9..072559de81c3b3a25900bbff4fc9c905f95adcc9 100644
--- a/gso/__init__.py
+++ b/gso/__init__.py
@@ -1,7 +1,10 @@
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")
diff --git a/gso/cli/netbox.py b/gso/cli/netbox.py
new file mode 100644
index 0000000000000000000000000000000000000000..00f8f399287412b8cf13964d875e51089bcb9630
--- /dev/null
+++ b/gso/cli/netbox.py
@@ -0,0 +1,40 @@
+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.")
diff --git a/gso/services/netbox_client.py b/gso/services/netbox_client.py
index 974bb7ba9d51578178e3979714c4da3ec0c33f62..eb69515c6d593f4071b4fd0f8ea04b5f5066f581 100644
--- a/gso/services/netbox_client.py
+++ b/gso/services/netbox_client.py
@@ -1,266 +1,338 @@
-# 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")))
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index be23b4159398d38f8255fdd0584a66add973717e..ba48e89e090e74949e0b246f3e68efb90f36a56d 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -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
diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py
index 413e84b901994da275bad375433fbf5024cbe956..a564e48876ca06fedf90beeb6c4bbee875614fd7 100644
--- a/gso/workflows/router/terminate_router.py
+++ b/gso/workflows/router/terminate_router.py
@@ -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