# DevNet Lab 9 -- Configure Open vSwitch using the Python ovsdbapp library
[toc]
---
> Copyright (c) 2025 Philippe Latu.
Permission is granted to copy, distribute and/or modify this document under the
terms of the GNU Free Documentation License, Version 1.3 or any later version
published by the Free Software Foundation; with no Invariant Sections, no
Front-Cover Texts, and no Back-Cover Texts. A copy of the license is included
in the section entitled "GNU Free Documentation License".
https://inetdoc.net
### Scenario
The purpose of this lab is to provide an initial illustration of atomic switch fabric programmability.
In the context of the private cloud infrastructure we are using, hypervisors are deployed using an Open vSwitch distribution switch called dsw-host. In addition, a large number of tap interfaces are also provisioned and declared as switch ports. When students start their first labs, they are preassigned sets of these tap interfaces to run virtual machines or routers.
The activities in this lab will show how to connect to the Open vSwitch database server service named ovsdb-server and configure existing switch ports from a Python script code. This is a very first step into the world of network programmability.
### Objectives
The primary objectives of this lab are to equip students with hands-on experience in network programmability using Open vSwitch and Python. By the end of this lab, students will be able to:
- Connect to the Open vSwitch Database Server
: Establish a secure connection to the OVSDB server using Python scripts, enabling interaction with the Open vSwitch configuration database.
- Retrieve and Manage Switch Port Configurations
: Use Python to list existing switches, retrieve detailed attributes of specific ports, and determine their VLAN configurations (access or trunk mode).
- Apply Declarative Configuration
: Load and apply network configurations from a YAML file to switch ports, demonstrating a DevOps approach to infrastructure management.
- Verify Applied Configurations
: Manually verify the applied configurations on the hypervisor to ensure consistency between the declared state and the actual network setup.
## Part 1: Setup the lab environment
In this part we analyse the conditions for setting up a development communication channel between our Python code and the ovsdb-server service running on the hypervisor.
### Step 1: Identify the access conditions for the ovsdb-server service
We first have to connect to the hypervisor in order to identify the ovsdb-server process attributes.
```bash
ps aux | grep ovsdb-server
```
```bash=
root 2050 0.0 0.0 38856 23248 ? S<s mars23 0:58 ovsdb-server /etc/openvswitch/conf.db -vconsole:emer -vsyslog:err -vfile:info --remote=punix:/var/run/openvswitch/db.sock --private-key=db:Open_vSwitch,SSL,private_key --certificate=db:Open_vSwitch,SSL,certificate --bootstrap-ca-cert=db:Open_vSwitch,SSL,ca_cert --no-chdir --log-file=/var/log/openvswitch/ovsdb-server.log --pidfile=/var/run/openvswitch/ovsdb-server.pid --detach
```
From this somewhat long line, we can identify the connection socket to the database service.
```bash
--remote=punix:/var/run/openvswitch/db.sock
```
Then we can determine who has access to that socket by looking at the permissions in the socket file.
```bash
ls -lAh /var/run/openvswitch/db.sock
```
```bash=
srwxrwx--- 1 root kvm 0 23 mars 07:49 /var/run/openvswitch/db.sock
```
The main point here is that members of the kvm system group have full access to the ovsdb-server socket. It is this access granted by group membership that allows us to program switching functions.
### Step 2: Access the ovsdb-server from the DevNet development system
The question now is how to access this socket from the DevNet virtual machine, which is our development system.
OpenSSH's `LocalForward` feature allows us to securely forward Unix domain sockets from the hypervisor to the DevNet VM, providing secure access to the ovsdb-server service running on the hypervisor by creating a local socket that proxies requests to the remote socket.
Here is a snippet of the SSH client configuration file `~/.ssh/config` that shows how to configure the `LocalForward` feature in our context.
```
Host hypervisor_name
HostName fe80::VVVV:1%%enp0s1
User etudianttest
Port 2222
StreamLocalBindUnlink yes
LocalForward /tmp/ovs-forwarded.sock /var/run/openvswitch/db.sock
```
- `/tmp/ovs-forwarded.sock` : local Unix domain socket on the development system.
- `/var/run/openvswitch/db.sock` : remote Unix domain socket on the hypervisor running the ovsdb-server service.
This lab setup can be considered a good security practice, as it limits the exposure of the service to only the necessary development or configuration times.
1. Here is an example `ssh` command that forks a new process with `-f` and doesn't execute a remote command with `-N`.
```bash
ssh -fN hypervisor_name
```
2. Here is the instruction to close the SSH tunnel by killing the process started by the above command.
```bash
pkill -fu $USER "ssh -fN"
```
:::info
The main limitation is that we need to keep the SSH connection open between the DevNet system and the hypervisor while we are programming. This is not too annoying, as we may have to evaluate the results of the Python scripts on the hypervisor.
:::
## Part 2: Connect to the Open vSwitch database and retrieve switch port configuration
In this part, we start by setting up the Python virtual environment that contains the ovsdbapp module code. Then we code two first simple scripts that prove we are able to interact with the ovsd-server service on the hypervisor.
### Step 1: Configure ovsdbapp on the DevNet VM
We start by creating the working directory and creating a Python virtual environment.
1. Make the `$HOME/labs/lab09` directory for example and navigate to this folder
```bash
mkdir -p $HOME/labs/lab09 && cd $HOME/labs/lab09
```
2. Install **ovsdbapp** Python virtual environement
We choose to install `ovsdbapp` in a Python virtual environment to take advantage of the latest release.
We start by creating a `requirements.txt` file.
```bash
cat << EOF > requirements.txt
ovsdbapp
tabulate
PyYAML
EOF
```
Then we install the tools in a virtual environment called `ovsdb`.
```bash
python3 -m venv ovsdb
source ./ovsdb/bin/activate
pip3 install -r requirements.txt
```
### Step 2: Start a first connection to the OvS database
Here is Python script code named `01_ovsdb_connect.py` that provides an object-oriented interface to communicate with the Open vSwitch Database (OVSDB) via a Unix socket forwarded through SSH.
In this script, the `OVSDBManager` class encapsulates connection handling and database operations, allowing Open vSwitch network configurations to be managed programmatically.
This first example allows existing virtual switches to be listed.
```python=
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
from ovsdbapp.backend.ovs_idl import connection, idlutils
from ovsdbapp.schema.open_vswitch import impl_idl as ovs_impl_idl
class OVSDBManager:
"""Class for managing OVSDB connections and operations"""
# Class constants
OVS_CONNECTION = "unix:/tmp/ovs-forwarded.sock"
OVSDB_CONNECTION_TIMEOUT = 30
def __init__(self, auto_connect=True):
"""Initialize the OVSDB manager and establish connection if auto_connect is True"""
self.conn = None
self.ovs = None
if auto_connect:
try:
self.connect()
except Exception as e:
print(f"Warning: Failed to connect automatically: {e}")
print(
"You'll need to call connect() manually before using other methods."
)
def connect(self):
"""Establish a connection to Open vSwitch via Unix socket"""
helper = idlutils.get_schema_helper(self.OVS_CONNECTION, "Open_vSwitch")
helper.register_all()
idl = connection.OvsdbIdl(self.OVS_CONNECTION, helper)
self.conn = connection.Connection(
idl=idl, timeout=self.OVSDB_CONNECTION_TIMEOUT
)
self.ovs = ovs_impl_idl.OvsdbIdl(self.conn)
return self.ovs
def print_switch_list(self):
"""
Lists existing switches in Open vSwitch database
CLI equivalent: ovs-vsctl list-br
"""
try:
# Get all existing bridges
sw_list = self.ovs.list_br().execute()
print(f"Switches found ({len(sw_list)}):")
for sw in sw_list:
print(f" - {sw}")
return sw_list
except Exception as e:
print(f"Error retrieving bridges: {e}")
return []
def main():
print("Attempting to connect to remote Open vSwitch server via Unix socket...")
try:
# Create manager with automatic connection
manager = OVSDBManager()
# List existing bridges
switch_list = manager.print_switch_list()
# Display success message
if switch_list:
print("Connection and switch list retrieval successful!")
sys.exit(0) # Exit with success code
else:
print("Connection successful, but no switches found.")
sys.exit(1) # Exit with error code
except Exception as e:
print(f"Connection error: {e}")
sys.exit(1) # Exit with error code
if __name__ == "__main__":
main()
```
Open the SSH connection that will establish the Unix domain socket proxy in a terminal session.
```bash
ssh hypervisor_name
```
Next, run the Pyton script to test the first ovsdb-server query, which lists existing switches.
```bash
python3 01_ovsdb_connect.py
```
```bash=
Attempting to connect to remote Open vSwitch server via Unix socket...
Switches found (1):
- dsw-host
Connection and switch list retrieval successful!
```
:::success
We now have a verified communication channel between the DevNet system and the Open vSwitch hypervisor database.
:::
### Step 3: Get the attributes of a switch port
This new Python script is an extension of the previous step code. As in the first version, it is designed to connect to an Open vSwitch Database (OVSDB) where the switch named "dsw-host" is already configured with multiple tap interfaces defined as its ports.
This new version of the script prompts the user for a tap interface name. The code then determines which bridge that particular port belongs to and retrieves detailed attributes of that particular port.
Here is a copy of the `02_ovsdb_list_port.py` script code:
```python=
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
from ovsdbapp.backend.ovs_idl import connection, idlutils
from ovsdbapp.schema.open_vswitch import impl_idl as ovs_impl_idl
from tabulate import tabulate # Import tabulate module for formatted output
# Define constant
SWITCH_NAME = "dsw-host"
class OVSDBManager:
"""Class for managing OVSDB connections and operations"""
# Class constants
OVS_CONNECTION = "unix:/tmp/ovs-forwarded.sock"
OVSDB_CONNECTION_TIMEOUT = 30
def __init__(self, auto_connect=True):
"""Initialize the OVSDB manager and establish connection if auto_connect is True"""
self.conn = None
self.ovs = None
if auto_connect:
try:
self.connect()
except Exception as e:
print(f"Warning: Failed to connect automatically: {e}")
print(
"You'll need to call connect() manually before using other methods."
)
def connect(self):
"""Establish a connection to Open vSwitch via Unix socket"""
helper = idlutils.get_schema_helper(self.OVS_CONNECTION, "Open_vSwitch")
helper.register_all()
idl = connection.OvsdbIdl(self.OVS_CONNECTION, helper)
self.conn = connection.Connection(
idl=idl, timeout=self.OVSDB_CONNECTION_TIMEOUT
)
self.ovs = ovs_impl_idl.OvsdbIdl(self.conn)
return self.ovs
def print_switch_list(self):
"""
Lists existing switches in Open vSwitch database
CLI equivalent: ovs-vsctl list-br
"""
try:
# Get all existing bridges
sw_list = self.ovs.list_br().execute()
print(f"Switches found ({len(sw_list)}):")
for sw in sw_list:
print(f" - {sw}")
return sw_list
except Exception as e:
print(f"Error retrieving bridges: {e}")
return []
def list_port_attributes(self, port_name, bridge_name=SWITCH_NAME):
"""
Lists all attributes for a specific port on a given bridge
CLI equivalent: ovs-vsctl list port <port_name>
"""
try:
# Check if the bridge exists
bridges = self.ovs.list_br().execute()
if bridge_name not in bridges:
print(f"Error: Bridge '{bridge_name}' does not exist.")
return None
# Check if the port exists on the bridge
ports = self.ovs.list_ports(bridge_name).execute()
if port_name not in ports:
print(
f"Error: Port '{port_name}' does not exist on bridge '{bridge_name}'."
)
return None
# Get port details using direct OVSDB command
cmd = self.ovs.db_find("Port", ("name", "=", port_name))
port_records = cmd.execute()
if not port_records:
print(f"No details found for port '{port_name}'.")
return None
print(f"\nAttributes for port '{port_name}':")
# Prepare data for tabulate
for record in port_records:
# Convert record to a list of [key, value] pairs and sort alphabetically by key
table_data = sorted(
[[k, str(v)] for k, v in record.items()], key=lambda x: x[0]
)
# Display using tabulate with fancy_grid format
print(
tabulate(
table_data,
headers=["Attribute", "Value"],
tablefmt="fancy",
)
)
return port_records
except Exception as e:
print(f"Error retrieving port attributes: {e}")
return None
def get_bridge_for_port(self, port_name):
"""
Determines which bridge a port belongs to
CLI equivalent: ovs-vsctl port-to-br <port_name>
"""
try:
# Get all bridges
bridges = self.ovs.list_br().execute()
# Check each bridge for the port
for bridge in bridges:
ports = self.ovs.list_ports(bridge).execute()
if port_name in ports:
return bridge
# Port not found on any bridge
return None
except Exception as e:
print(f"Error determining bridge for port: {e}")
return None
def main():
print("Attempting to connect to remote Open vSwitch server via Unix socket...")
try:
# Create manager with automatic connection
manager = OVSDBManager()
# Ask user for port name to examine
port_name = input("\nEnter the port name to examine (e.g. tap123): ")
# Check which bridge this port belongs to
bridge = manager.get_bridge_for_port(port_name)
if bridge is None:
print(f"Error: Port '{port_name}' does not exist on any switch.")
sys.exit(1) # Exit with error code
elif bridge != SWITCH_NAME:
print(
f"Error: Port '{port_name}' belongs to '{bridge}', not to '{SWITCH_NAME}'."
)
sys.exit(1) # Exit with error code
else:
print(f"Confirmed: Port '{port_name}' belongs to '{SWITCH_NAME}'.")
# Get detailed port attributes
port_records = manager.list_port_attributes(port_name)
if port_records:
sys.exit(0) # Exit with success code
else:
print("Failed to retrieve port attributes.")
sys.exit(1) # Exit with error code
except Exception as e:
print(f"Connection error: {e}")
sys.exit(1) # Exit with error code
if __name__ == "__main__":
main()
```
Here is a concise documentation of the methods and the logic of the main function in the script:
__init__(self, auto_connect=True)
: Initializes the OVSDB manager. As auto_connect is True, it attempts to establish a connection to the OVSDB server automatically.
connect(self)
: Establishes a connection to the Open vSwitch server via a Unix socket.
print_switch_list(self)
: Lists all existing switches in the Open vSwitch database. This method is not used here.
list_port_attributes(self, port_name, bridge_name="dsw-host")
: Lists all attributes for a specific port on the `dsw-host` given bridge.
get_bridge_for_port(self, port_name)
: Determines which bridge a port belongs to.
The `main()` function of the script creates an instance of the OVSDB manager, prompts the user for a port name, checks which bridge the port belongs to, and then lists the detailed attributes of the port if it belongs to `dsw-host`.
Here is an example of the code execution:
```bash
python3 02_ovsdb_list_port.py
```
```bash=
Attempting to connect to remote Open vSwitch server via Unix socket...
Enter the port name to examine (e.g. tap123): tap700
Confirmed: Port 'tap700' belongs to 'dsw-host'.
Attributes for port 'tap700':
Attribute Value
----------------- ----------------------------------------------
_uuid 0a79920b-9e45-4cea-8292-5779ade33d2d
bond_active_slave []
bond_downdelay 0
bond_fake_iface False
bond_mode []
bond_updelay 0
cvlans []
external_ids {}
fake_bridge False
interfaces [UUID('30b469a1-b5a6-4c23-8d8f-0d863b6bff1d')]
lacp []
mac []
name tap700
other_config {}
protected False
qos []
rstp_statistics {}
rstp_status {}
statistics {}
status {}
tag []
trunks []
vlan_mode trunk
```
### Step 4: Get `vlan_mode` and VLAN id(s) of a switch port
To go one step further, we want to be more specific about the `vlan_mode` attribute and the VLAN IDs depending on its value: `access` or `trunk`.
We want this new step script named `03_ovsdb_list_port_mode.py` to extend the functionality of `02_ovsdb_list_port.py` by implementing a more sophisticated approach to port information retrieval and VLAN configuration analysis.
We want this new step script named `03_ovsdb_list_port_mode.py` to extend the functionality of `02_ovsdb_list_port.py` by implementing a more sophisticated approach to port information retrieval and VLAN configuration analysis.
In order to avoid redundant database queries, we have to add two private methods to the OVSDBManager class:
`_get_port_details()`
: Collects all port attributes at once and stores the records in a private `_port_cache` dictionary.
`_clear_cache()`
: Clears records for a specific port or the entire dictionary.
Here is the code of these two new private methods to insert in the OVSDBManager class:
```python=
def _get_port_details(self, port_name, force_refresh=False):
"""
Private method that retrieves port details, using cache when available
Args:
port_name (str): Name of the port to get details for
force_refresh (bool): If True, ignore cache and fetch fresh data
CLI equivalent: ovs-vsctl list port <port_name>
Returns:
dict: Port details record or None if port not found
"""
try:
# Check if we already have this port's details in cache
if not force_refresh and port_name in self._port_cache:
return self._port_cache[port_name]
# Get port details via OVSDB command
cmd = self.ovs.db_find("Port", ("name", "=", port_name))
port_records = cmd.execute()
if not port_records:
print(f"No details found for port '{port_name}'.")
return None
# Cache the result for future use
self._port_cache[port_name] = port_records[0]
return self._port_cache[port_name]
except Exception as e:
print(f"Error retrieving port details: {e}")
return None
def _clear_cache(self, port_name=None):
"""
Private method that clears the port details cache
Args:
port_name (str, optional): If provided, clear only this port's cache
If None, clear the entire cache
"""
if port_name:
if port_name in self._port_cache:
del self._port_cache[port_name]
else:
self._port_cache.clear()
```
Once the private methods are in place, we can add specialized methods for VLAN configuration analysis:
`get_port_mode()`
: Determines if a port is in access or trunk mode by examining the vlan_mode attribute
`get_port_access_vlan()`
: Retrieves the VLAN tag for access ports
`get_port_trunk_vlan_list()`
: Retrieves the list of allowed VLANs for trunk ports
```python=
def get_port_mode(self, port_name):
"""
Determines if a port is in access or trunk mode by checking vlan_mode attribute
CLI equivalent: ovs-vsctl get port <port_name> vlan_mode
Returns:
"access" - if the port is in access mode
"trunk" - if the port is in trunk mode
None - if port not found or error
"""
# Get port details
record = self._get_port_details(port_name)
if not record:
return None
# Check the vlan_mode attribute to determine port mode
if "vlan_mode" in record:
return record["vlan_mode"]
else:
print(f"Error: vlan_mode attribute not found for port '{port_name}'.")
return None
def get_port_access_vlan(self, port_name):
"""
Returns the VLAN ID (tag) for a port in access mode
CLI equivalent: ovs-vsctl get port <port_name> tag
Returns:
VLAN ID (integer) if port is in access mode and has a VLAN
None if port is not in access mode or has no VLAN or not found
"""
# Get port details
record = self._get_port_details(port_name)
if not record:
return None
# Check if this is an access port with a VLAN tag
port_mode = self.get_port_mode(port_name)
if port_mode == "access" and "tag" in record and record["tag"]:
return record["tag"]
else:
return None
def get_port_trunk_vlan_list(self, port_name):
"""
Returns the list of allowed VLAN IDs for a port in trunk mode
CLI equivalent: ovs-vsctl get port <port_name> trunks
Returns:
List of VLAN IDs if port is in trunk mode
Empty list if port is in trunk mode but no VLANs are configured
None if port is not in trunk mode or not found
"""
# Get port details
record = self._get_port_details(port_name)
if not record:
return None
# Check if this is a trunk port
port_mode = self.get_port_mode(port_name)
if port_mode == "trunk":
if "trunks" in record and record["trunks"]:
# Return the list of allowed VLANs
return record["trunks"]
else:
# Trunk port but no specific VLANs configured (all allowed)
return []
else:
return None
```
Instead of displaying all port attributes, the main function provides a focused output that displays only the relevant VLAN information based on the port's operating mode. This targeted approach provides more meaningful insight into the network segmentation configuration, while maintaining better performance through data caching and optimized database interactions.
Here is a copy of the main function of the `03_ovsdb_list_port_mode.py` script:
```python=
def main():
print("Attempting to connect to remote Open vSwitch server via Unix socket...")
try:
# Create manager with automatic connection
manager = OVSDBManager()
# Ask user for port name to examine
port_name = input("\nEnter the port name to examine (e.g. tap123): ")
# Check which bridge this port belongs to
bridge = manager.get_bridge_for_port(port_name)
if bridge is None:
print(f"Error: Port '{port_name}' does not exist on any switch.")
sys.exit(1) # Exit with error code
elif bridge != SWITCH_NAME:
print(
f"Error: Port '{port_name}' belongs to '{bridge}', not to '{SWITCH_NAME}'."
)
sys.exit(1) # Exit with error code
else:
print(f"Confirmed: Port '{port_name}' belongs to '{SWITCH_NAME}'.")
# No need to explicitly call get_port_details anymore
# Get and display port mode
port_mode = manager.get_port_mode(port_name)
print(f"\nPort mode: {port_mode}")
if port_mode is None:
sys.exit(1) # Exit with error code if port mode cannot be determined
# Get and display VLAN information based on port mode
if port_mode == "access":
vlan_id = manager.get_port_access_vlan(port_name)
if vlan_id is not None:
print(f"Access VLAN ID: {vlan_id}")
else:
print("No VLAN configured for this access port")
elif port_mode == "trunk":
vlan_list = manager.get_port_trunk_vlan_list(port_name)
if vlan_list:
print(f"Allowed VLANs: {vlan_list}")
else:
print("All VLANs are allowed on this trunk port")
else:
print("Unable to determine port mode")
sys.exit(1) # Exit with error code
# If we reach here, everything was successful
sys.exit(0) # Exit with success code
except Exception as e:
print(f"Connection error: {e}")
sys.exit(1) # Exit with error code
```
Once all the script code is assembled, we are ready to run tests on different switch port configurations:
```bash
python3 03_ovsdb_list_port_mode.py
```
In the first case, the `tap7` switch port is in **access mode**.
```bash=
Attempting to connect to remote Open vSwitch server via Unix socket...
Enter the port name to examine (e.g. tap123): tap7
Confirmed: Port 'tap7' belongs to 'dsw-host'.
Port mode: access
Access VLAN ID: 52
```
In the second example, the `tap20` port is in **trunk mode**.
```bash=
Attempting to connect to remote Open vSwitch server via Unix socket...
Enter the port name to examine (e.g. tap123): tap20
Confirmed: Port 'tap20' belongs to 'dsw-host'.
Port mode: trunk
Allowed VLANs: [52, 1220, 1221]
```
## Part 2: Declare switch configuration using a YAML file
In this part, we have reached the point where we want to apply a new configuration to switch ports. We will to use the declarative method using a YAML file as the desired configuration state or source of truth.
Designing a source of truth is not an easy task. Here, we just need a starting point to illustrate the configuration of a switch with a few ports and their attributes.
Here is our very first YAML configuration file:
### Step 1: Build your own YAML switch configuration file
Create a new file named `switch_config.yaml` in the `$HOME/labs/lab09` directory.
```yaml=
---
ovs:
switches:
- name: dsw-host
ports:
- name: tapXX2
type: OVSPort
vlan_mode: access
tag: 2X8
- name: tapXX5
type: OVSPort
vlan_mode: access
tag: 4X0
- name: tapXX6
type: OVSPort
vlan_mode: trunk
trunks: [4XX, 5XX, 6XX]
```
:::warning
Make sure to use the tap interface and VLAN numbers assigned to you :zap:
Edit and replace all XX marks.
:::
### Step 2: Add a new ConfigLoader class
The `ConfigLoader` class we create in this step is a specialized component designed to parse, validate, and extract network configuration from YAML files, providing an interface for accessing switch and port configurations.
Start by copying the existing script code to a new file:
```bash
cp 03_ovsdb_list_port_mode.py 04_ovsdb_apply_config.py
```
Insert the following `ConfigLoader` class code to the new script named `04_ovsdb_apply_config.py`.
```python=
class ConfigLoader:
"""Class for loading and parsing YAML configuration files"""
def __init__(self, config_file):
"""
Initialize the configuration loader
Args:
config_file (str): Path to the YAML configuration file
"""
self.config_file = config_file
self.config = None
def load_config(self):
"""
Load the configuration from the YAML file
Returns:
dict: The loaded configuration or None if there was an error
"""
try:
if not os.path.exists(self.config_file):
print(f"Error: Configuration file '{self.config_file}' not found.")
return None
with open(self.config_file, "r") as file:
self.config = yaml.safe_load(file)
return self.config
except Exception as e:
print(f"Error loading configuration: {e}")
return None
def get_switches(self):
"""
Get the list of switches from the configuration
Returns:
list: List of switch configurations or empty list if not found
"""
if not self.config:
return []
return self.config.get("ovs", {}).get("switches", [])
def get_ports_for_switch(self, switch_name):
"""
Get the list of ports for a specific switch
Args:
switch_name (str): Name of the switch
Returns:
list: List of port configurations or empty list if not found
"""
switches = self.get_switches()
for switch in switches:
if switch.get("name") == switch_name:
return switch.get("ports", [])
return []
def print_config_summary(self):
"""Print a summary of the loaded configuration"""
if not self.config:
print("No configuration loaded.")
return
switches = self.get_switches()
print(f"Configuration loaded: {len(switches)} switch(es) defined")
for switch in switches:
switch_name = switch.get("name", "Unknown")
ports = switch.get("ports", [])
print(f"- Switch '{switch_name}': {len(ports)} port(s) defined")
# Print port summary in a table
if ports:
port_data = []
for port in ports:
mode = port.get("vlan_mode", "unknown")
if mode == "access":
vlan_info = f"VLAN {port.get('tag', 'None')}"
elif mode == "trunk":
vlan_info = f"VLANs {port.get('trunks', [])}"
else:
vlan_info = "No VLAN info"
port_data.append([port.get("name", "Unknown"), mode, vlan_info])
print(
tabulate(
port_data,
headers=["Port", "Mode", "VLAN Config"],
tablefmt="simple",
)
)
```
Here is a short description of the `ConfigLoader` class methods.
`__init__(self, config_file)`
: Initializes the loader with the path to a YAML configuration file and prepares the internal state.
`load_config(self)`
: Reads and parses the YAML file, handling file existence checks and parsing errors, returning the configuration data structure or None if errors occur.
`get_switches(self)`
: Extracts and returns the list of switch configurations from the loaded data, following the expected structure where switches are defined under the `ovs.switches` path.
`get_ports_for_switch(self, switch_name)`
: Retrieves the port configurations specific to a named switch, facilitating targeted access to port details.
`print_config_summary(self)`
: Generates and displays a human-readable summary of the configuration, including switch count, ports per switch, and a formatted table showing VLAN configurations for each port.
### Step 3: Edit the script main function
The `main()` must be edited to use the YAML configuration declaration file.
The command-line interface is managed using Python's argparse module, which parses the required configuration file paths as positional arguments and supports a `--dry-run` flag that allows administrators to validate configuration changes by displaying a summary of intended network changes without actually applying them to the switch.
This provides a preview mechanism before committing any changes to the production environment.
Therefore, we need to add Python module import statements at the top of the script code.
```python=
import argparse
import os.path
import sys
import yaml
```
Here is a copy of the edited `main()` function code.
```python=
def main():
# Parse command line arguments
parser = argparse.ArgumentParser(
description="Apply YAML configuration to Open vSwitch"
)
parser.add_argument("config_file", help="Path to the YAML configuration file")
parser.add_argument(
"--dry-run", action="store_true", help="Print configuration but do not apply it"
)
args = parser.parse_args()
try:
# Load configuration
config_loader = ConfigLoader(args.config_file)
config = config_loader.load_config()
if not config:
print("Failed to load configuration.")
sys.exit(1)
# Display configuration summary
config_loader.print_config_summary()
# If dry run, exit here
if args.dry_run:
print("\nDry run completed. No changes were made.")
sys.exit(0)
except Exception as e:
print(f"Error: {e}")
sys.exit(1)
```
### Step 4: Test the `--dry-run` flag
Here is a sample run of the `04_ovsdb_apply_config.py` script with the `--dry-run` flag
```bash
python3 04_ovsdb_apply_config.py switch_config.yml --dry-run
```
```bash=
Configuration loaded: 1 switch(es) defined
- Switch 'dsw-host': 3 port(s) defined
Port Mode VLAN Config
------ ------ --------------------
tap2 access VLAN 52
tap5 access VLAN 410
tap9 trunk VLANs [52, 410, 420]
Dry run completed. No changes were made.
```
The output shows that the YAML declarations were parsed correctly.
## Part 3: Apply switch ports configuration
We are now ready to apply the configuration parameters declared in the YAML file to the switch ports. To do this, we need to extend the functionality of the `OVSDBManager` class by transforming it from a pure information tool to a configuration management system.
The focus here is on a caching architecture that tracks both port details and bridge information, resulting in faster execution and fewer database queries. In addition, this implementation enforces proper VLAN configuration practices by automatically resolving conflicting settings when switching between access and trunk modes, ensuring configuration integrity across the virtual switch infrastructure.
### Step 1: Transform the OVSDBManager class code
Here is a new copy of the `OVSDBManager` class code that includes the extended caching features and the switch port configuration methods.
```python=
class OVSDBManager:
"""Class for managing OVSDB connections and operations"""
# Class constants
OVS_CONNECTION = "unix:/tmp/ovs-forwarded.sock"
OVSDB_CONNECTION_TIMEOUT = 30
def __init__(self, auto_connect=True):
"""Initialize the OVSDB manager and establish connection if auto_connect is True"""
self.conn = None
self.ovs = None
# Cache to store details of previously queried ports
self._port_cache = {}
# Cache to store list of bridges
self._bridges_cache = None
if auto_connect:
try:
self.connect()
except Exception as e:
print(f"Warning: Failed to connect automatically: {e}")
print(
"You'll need to call connect() manually before using other methods."
)
def connect(self):
"""Establish a connection to Open vSwitch via Unix socket"""
helper = idlutils.get_schema_helper(self.OVS_CONNECTION, "Open_vSwitch")
helper.register_all()
idl = connection.OvsdbIdl(self.OVS_CONNECTION, helper)
self.conn = connection.Connection(
idl=idl, timeout=self.OVSDB_CONNECTION_TIMEOUT
)
self.ovs = ovs_impl_idl.OvsdbIdl(self.conn)
return self.ovs
def _get_bridges(self, force_refresh=False):
"""
Private method that retrieves the list of bridges, using cache when available
Args:
force_refresh (bool): If True, ignore cache and fetch fresh data
CLI equivalent: ovs-vsctl list-br
Returns:
list: List of bridge names or empty list if error
"""
try:
# Check if we already have the bridges in cache
if not force_refresh and self._bridges_cache is not None:
return self._bridges_cache
# Get bridges from database
bridges = self.ovs.list_br().execute()
# Cache the result for future use
self._bridges_cache = bridges
return self._bridges_cache
except Exception as e:
print(f"Error retrieving bridges: {e}")
return []
def _get_port_details(self, port_name, force_refresh=False):
"""
Private method that retrieves port details, using cache when available
Args:
port_name (str): Name of the port to get details for
force_refresh (bool): If True, ignore cache and fetch fresh data
CLI equivalent: ovs-vsctl list port <port_name>
Returns:
dict: Port details record or None if port not found
"""
try:
# Check if we already have this port's details in cache
if not force_refresh and port_name in self._port_cache:
return self._port_cache[port_name]
# Get port details via OVSDB command
cmd = self.ovs.db_find("Port", ("name", "=", port_name))
port_records = cmd.execute()
if not port_records:
print(f"No details found for port '{port_name}'.")
return None
# Cache the result for future use
self._port_cache[port_name] = port_records[0]
return self._port_cache[port_name]
except Exception as e:
print(f"Error retrieving port details: {e}")
return None
def _clear_cache(self, port_name=None, clear_bridges=False):
"""
Private method that clears the caches
Args:
port_name (str, optional): If provided, clear only this port's cache
If None, clear all port caches
clear_bridges (bool): If True, clear the bridges cache
"""
if port_name:
if port_name in self._port_cache:
del self._port_cache[port_name]
else:
self._port_cache.clear()
if clear_bridges:
self._bridges_cache = None
def list_port_attributes(self, port_name, bridge_name=SWITCH_NAME):
"""
Lists all attributes for a specific port on a given bridge
CLI equivalent: ovs-vsctl list port <port_name>
"""
try:
# Check if the bridge exists
bridges = self._get_bridges()
if bridge_name not in bridges:
print(f"Error: Bridge '{bridge_name}' does not exist.")
return None
# Check if the port exists on the bridge
ports = self.ovs.list_ports(bridge_name).execute()
if port_name not in ports:
print(
f"Error: Port '{port_name}' does not exist on bridge '{bridge_name}'."
)
return None
# Get port details
record = self._get_port_details(port_name)
if not record:
return None
print(f"\nAttributes for port '{port_name}':")
# Prepare data for tabulate
# Convert record to a list of [key, value] pairs and sort alphabetically by key
table_data = sorted(
[[k, str(v)] for k, v in record.items()], key=lambda x: x[0]
)
# Display using tabulate with fancy_grid format
print(
tabulate(
table_data,
headers=["Attribute", "Value"],
tablefmt="fancy",
)
)
return [record] # Return in same format as before for compatibility
except Exception as e:
print(f"Error retrieving port attributes: {e}")
return None
def get_bridge_for_port(self, port_name):
"""
Determines which bridge a port belongs to
CLI equivalent: ovs-vsctl port-to-br <port_name>
"""
try:
# Get all bridges
bridges = self._get_bridges()
# Check each bridge for the port
for bridge in bridges:
ports = self.ovs.list_ports(bridge).execute()
if port_name in ports:
return bridge
# Port not found on any bridge
return None
except Exception as e:
print(f"Error determining bridge for port: {e}")
return None
def get_port_mode(self, port_name):
"""
Determines if a port is in access or trunk mode by checking vlan_mode attribute
CLI equivalent: ovs-vsctl get port <port_name> vlan_mode
Returns:
"access" - if the port is in access mode
"trunk" - if the port is in trunk mode
None - if port not found or error
"""
# Get port details
record = self._get_port_details(port_name)
if not record:
return None
# Check the vlan_mode attribute to determine port mode
if "vlan_mode" in record:
return record["vlan_mode"]
else:
print(f"Error: vlan_mode attribute not found for port '{port_name}'.")
return None
def get_port_access_vlan(self, port_name):
"""
Returns the VLAN ID (tag) for a port in access mode
CLI equivalent: ovs-vsctl get port <port_name> tag
Returns:
VLAN ID (integer) if port is in access mode and has a VLAN
None if port is not in access mode or has no VLAN or not found
"""
# Get port details
record = self._get_port_details(port_name)
if not record:
return None
# Check if this is an access port with a VLAN tag
port_mode = self.get_port_mode(port_name)
if port_mode == "access" and "tag" in record and record["tag"]:
return record["tag"]
else:
return None
def get_port_trunk_vlan_list(self, port_name):
"""
Returns the list of allowed VLAN IDs for a port in trunk mode
CLI equivalent: ovs-vsctl get port <port_name> trunks
Returns:
List of VLAN IDs if port is in trunk mode
Empty list if port is in trunk mode but no VLANs are configured
None if port is not in trunk mode or not found
"""
# Get port details
record = self._get_port_details(port_name)
if not record:
return None
# Check if this is a trunk port
port_mode = self.get_port_mode(port_name)
if port_mode == "trunk":
if "trunks" in record and record["trunks"]:
# Return the list of allowed VLANs
return record["trunks"]
else:
# Trunk port but no specific VLANs configured (all allowed)
return []
else:
return None
def apply_port_config(self, port_config, bridge_name=SWITCH_NAME):
"""
Apply configuration to a port
Args:
port_config (dict): Port configuration dictionary
bridge_name (str): Name of the bridge to which the port belongs
Returns:
bool: True if successful, False otherwise
"""
try:
port_name = port_config.get("name")
if not port_name:
print("Error: Port configuration missing 'name' field.")
return False
# Check if the bridge exists
bridges = self._get_bridges()
if bridge_name not in bridges:
print(f"Error: Bridge '{bridge_name}' does not exist.")
return False
# Verify the port belongs to the bridge
actual_bridge = self.get_bridge_for_port(port_name)
if actual_bridge != bridge_name:
print(
f"Error: Port '{port_name}' does not exist on bridge '{bridge_name}'."
)
return False
# Clear the cache for this port to ensure fresh data
self._clear_cache(port_name)
# Apply VLAN mode configuration
vlan_mode = port_config.get("vlan_mode")
if vlan_mode:
print(f"Setting port '{port_name}' VLAN mode to '{vlan_mode}'")
self.ovs.db_set("Port", port_name, ("vlan_mode", vlan_mode)).execute()
# Apply mode-specific cleanup
if vlan_mode == "access":
# If setting to access mode, clear trunks
self.ovs.db_set("Port", port_name, ("trunks", [])).execute()
elif vlan_mode == "trunk":
# If setting to trunk mode, clear tag
self.ovs.db_set("Port", port_name, ("tag", [])).execute()
# Apply VLAN tag for access ports
if vlan_mode == "access" and "tag" in port_config:
tag = port_config.get("tag")
print(f"Setting port '{port_name}' access VLAN tag to {tag}")
self.ovs.db_set("Port", port_name, ("tag", tag)).execute()
# Apply VLAN trunks for trunk ports
if vlan_mode == "trunk" and "trunks" in port_config:
trunks = port_config.get("trunks")
print(f"Setting port '{port_name}' trunk VLANs to {trunks}")
self.ovs.db_set("Port", port_name, ("trunks", trunks)).execute()
# Success
return True
except Exception as e:
print(f"Error applying port configuration: {e}")
return False
def apply_switch_config(self, switch_config):
"""
Apply configuration to a switch and its ports
Args:
switch_config (dict): Switch configuration dictionary
Returns:
bool: True if successful, False otherwise
"""
try:
switch_name = switch_config.get("name")
if not switch_name:
print("Error: Switch configuration missing 'name' field.")
return False
# Check if the switch exists
bridges = self._get_bridges(
force_refresh=True
) # Force refresh to get latest state
if switch_name not in bridges:
print(f"Error: Switch '{switch_name}' does not exist.")
return False
# Apply port configurations
ports = switch_config.get("ports", [])
success_count = 0
for port_config in ports:
if self.apply_port_config(port_config, switch_name):
success_count += 1
print(
f"Applied configuration to {success_count} out of {len(ports)} ports."
)
# After applying configurations, clear bridges cache to ensure fresh data next time
self._clear_cache(clear_bridges=True)
return success_count == len(ports)
except Exception as e:
print(f"Error applying switch configuration: {e}")
return False
```
Here is a short decsription of the new methods introduced in this step.
`_get_bridges(force_refresh=False)`
: A private method that retrieves the list of bridges with efficient caching behavior, reducing redundant database queries.
`_clear_cache(port_name=None, clear_bridges=False)`
: An enhanced cache management method that can selectively clear port cache entries and/or the bridge cache.
`apply_port_config(port_config, bridge_name)`
: Applies VLAN configuration to a specific port, including setting the VLAN mode, access VLAN tag, and trunk VLAN lists while enforcing proper configuration by clearing conflicting settings.
`apply_switch_config(switch_config)`
: Orchestrates the configuration of an entire switch and its ports by iterating through port configurations and applying them with appropriate verification.
### Step 2: Apply the switch ports configuration
Now that our Python code is complete, we are ready to apply the declared switch ports configuration.
```bash
python3 04_ovsdb_apply_config.py switch_config.yml
```
```bash=
Configuration loaded: 1 switch(es) defined
- Switch 'dsw-host': 3 port(s) defined
Port Mode VLAN Config
------ ------ --------------------
tap2 access VLAN 52
tap5 access VLAN 410
tap9 trunk VLANs [52, 410, 420]
Setting port 'tap2' VLAN mode to 'access'
Setting port 'tap2' access VLAN tag to 52
Setting port 'tap5' VLAN mode to 'access'
Setting port 'tap5' access VLAN tag to 410
Setting port 'tap9' VLAN mode to 'trunk'
Setting port 'tap9' trunk VLANs to [52, 410, 420]
Applied configuration to 3 out of 3 ports.
Configuration applied successfully!
```
### Step 3: Verify the Applied Configuration from the Hypervisor's SSH Connection
To complete this lab process, we need to manually verify the switch ports parameters from the hypervisor console. Therefore, we will list the attributes of two different ports: one in access mode and the other in trunk mode.
- The `tap5` switch port is configured in access mode and belongs to VLAN 410:
```bash
sudo ovs-vsctl list port tap5
```
```bash=
_uuid : 268a9e67-b9cb-4478-8410-2e99c3b43812
bond_active_slave : []
bond_downdelay : 0
bond_fake_iface : false
bond_mode : []
bond_updelay : 0
cvlans : []
external_ids : {}
fake_bridge : false
interfaces : [72d6b44a-0d32-48e7-a6ab-89e3edd41a45]
lacp : []
mac : []
name : tap5
other_config : {}
protected : false
qos : []
rstp_statistics : {}
rstp_status : {}
statistics : {}
status : {}
tag : 410
trunks : []
vlan_mode : access
```
- The `tap9` switch port is configured in trunk mode with three VLANs allowed:
```bash
sudo ovs-vsctl list port tap9
```
```bash=
_uuid : fc5b244a-8e43-4f64-83b1-bbd8176fa4ac
bond_active_slave : []
bond_downdelay : 0
bond_fake_iface : false
bond_mode : []
bond_updelay : 0
cvlans : []
external_ids : {}
fake_bridge : false
interfaces : [173e12f5-0cb1-4046-8ad9-45de33d7cad6]
lacp : []
mac : []
name : tap9
other_config : {}
protected : false
qos : []
rstp_statistics : {}
rstp_status : {}
statistics : {}
status : {}
tag : []
trunks : [52, 410, 420]
vlan_mode : trunk
```
## Conclusion
This lab has provided a foundational understanding of network programmability by leveraging Open vSwitch and Python. Students have successfully connected to the OVSDB server, managed switch port configurations, and applied declarative configurations using YAML files. These skills are crucial in a DevOps environment where infrastructure as code (IaC) is increasingly important.
The code and techniques presented here serve as a starting point for more advanced network programming tasks. As students progress, they will be able to build upon this foundation to program network flows in a more complex switch fabric. This could involve integrating OpenFlow controllers to dynamically manage traffic flows or exploring other network virtualization technologies like Open Virtual Network (OVN). The ability to automate and manage network configurations programmatically is essential for efficient and scalable network operations in modern data centers and cloud environments.