# DevNet Lab 9 -- Configure Open vSwitch using the Python ovsdbapp library [toc] --- > Copyright (c) 2026 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 >Changelog: >February 4, 2026: Null checks were added for `self.ovs` and the return values of `.execute()` across all scripts to resolve Pylance type safety warnings. ### Scenario The purpose of this lab is to provide an initial illustration of basic switch fabric programmability. In the context of our private cloud infrastructure, we deploy hypervisors using an Open vSwitch distribution switch called dsw-host. Additionally, a large number of tap interfaces are provisioned and declared as switch ports. When students begin their first labs, they are assigned a set of these tap interfaces on which 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. This is a 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 you will analyze the conditions required to set up a development communication channel between your 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 VM The question now is how to access this socket from the DevNet virtual machine, which is our development system. OpenSSH's `LocalForward` feature allows you to securely forward Unix domain sockets from the hypervisor to the DevNet VM. This provides secure access to the `ovsdb-server` service by creating a local socket that proxies requests to the remote hypervisor 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 is a good security practice because it limits exposure of the service to necessary development and configuration times. 1. Here is an example `ssh` command that uses `-f` to fork a new process in the background and `-N` to specify that no remote commands should be executed. ```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 must keep the SSH connection open between the DevNet virtual machine and the hypervisor while programming. However, this is not too inconvenient since we have to evaluate the results of the Python scripts on the hypervisor anyway. ::: ## Part 2: Connect to the Open vSwitch database and retrieve switch port configuration In this part, you will start by setting up a Python virtual environment that contains the **ovsdbapp** module code. Then, you will write your first two simple scripts to verify interaction with the `ovsdb-server` service on the hypervisor. ### Step 1: Configure ovsdbapp on the DevNet VM After initializing a new Git repository for this lab, start by creating the `pyproject.toml` and `.gitignore` files. The `pyproject.toml` defines `ovsdbapp`, `tabulate`, and `PyYAML` as the main libraries of this lab Python virtual environment. ```bash cat << EOF >pyproject.toml [project] name = "Lab09" version = "0.1.0" description = "Configure Open vSwitch using the Python ovsdbapp library" requires-python = ">=3.13" dependencies = ["ovsdbapp", "tabulate", "PyYAML"] EOF ``` The `.gitignore` file defines caches and binaries that should not be included in the version control system. ```bash cat << EOF >.gitignore .venv/ __pycache__/ *.pyc .pytest_cache EOF ``` :::info For a detailed introduction to UV installation and configuration, refer to [Lab 5 – Explore Python virtual environments with UV](https://md.inetdoc.net/s/f4P_Oy4yo). ::: Install the libraries and set up the virtual environment. ```bash uv lock --upgrade ``` ```bash= Using CPython 3.13.12 Resolved 11 packages in 607ms ``` ```bash uv sync ``` ```bash= Using CPython 3.13.12 Creating virtual environment at: .venv Resolved 11 packages in 0.83ms Built ovs==3.7.1 Prepared 7 packages in 1.10s Installed 9 packages in 15ms + fixtures==4.3.2 + netaddr==1.3.0 + ovs==3.7.1 + ovsdbapp==2.18.0 + pbr==7.0.3 + pyyaml==6.0.3 + setuptools==82.0.1 + sortedcontainers==2.4.0 + tabulate==0.10.0 ``` If your IDE does not activate the new virtual environment automatically, you can do so manually. ```bash source .venv/bin/activate ``` Open a new terminal and verify the `python` command belongs to the newly created virtual environment. ```bash command -v python ``` ```bash= /home/etu/labs/lab09/.venv/bin/python ``` You are now ready to start coding. ### Step 2: Start a first connection to the OvS database Here is a Python script 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: """Manage OVSDB connection and simple read operations.""" # Class constants OVS_CONNECT_SOCK = "unix:/tmp/ovs-forwarded.sock" OVSDB_CONNECT_TIMEOUT = 30 def __init__(self): """Initialize the manager without opening a connection.""" self.conn = None self.ovs = None def connect(self): """Establish a connection to Open vSwitch via Unix socket.""" helper = idlutils.get_schema_helper(self.OVS_CONNECT_SOCK, "Open_vSwitch") helper.register_all() idl = connection.OvsdbIdl(self.OVS_CONNECT_SOCK, helper) self.conn = connection.Connection(idl=idl, timeout=self.OVSDB_CONNECT_TIMEOUT) self.ovs = ovs_impl_idl.OvsdbIdl(self.conn) return self def list_switches(self): """Return bridge names (equivalent to ovs-vsctl list-br).""" if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") return self.ovs.list_br().execute() or [] def main(): print("Attempting to connect to remote Open vSwitch server via Unix socket...") try: switch_list = OVSDBManager().connect().list_switches() except Exception as e: print(f"Connection error: {e}") return 1 print(f"Switches found ({len(switch_list)}):") for sw in switch_list: print(f" - {sw}") if switch_list: print("Connection and switch list retrieval successful!") return 0 print("Connection successful, but no switches found.") return 1 if __name__ == "__main__": sys.exit(main()) ``` Open the SSH connection that will establish the Unix domain socket proxy in a terminal session. ```bash ssh -fN hypervisor_name ``` Next, run the Python script to test the first ovsdb-server query, which lists existing switches. ```bash python 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! ``` This script lists the existing virtual switches as a basic connectivity test. It separates connection setup (`connect()`) from data retrieval (`list_switches()`), which keeps the code reusable and easy to extend. * The `connect()` method uses `idlutils.get_schema_helper()` and Connection to build an `OvsdbIdl` instance, and then wraps it in the higher-level `ovs_impl_idl.OvsdbIdl` helper. * The `list_switches()` method mirrors `ovs-vsctl list-br`, returning an empty list if no bridges are found, to simplify consumer logic. * For readability, the `main()` function uses the fluent pattern `OVSDBManager().connect().list_switches()` and prints user-friendly status messages. It also uses broad exception handling to report connection errors and set an appropriate exit status. :::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. 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 the port belongs to and retrieves its detailed attributes. 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: """Manage OVSDB connection and simple read operations.""" # Class constants OVS_CONNECT_SOCK = "unix:/tmp/ovs-forwarded.sock" OVSDB_CONNECT_TIMEOUT = 30 def __init__(self): """Initialize the manager without opening a connection.""" self.conn = None self.ovs = None def connect(self): """Establish a connection to Open vSwitch via Unix socket.""" helper = idlutils.get_schema_helper(self.OVS_CONNECT_SOCK, "Open_vSwitch") helper.register_all() idl = connection.OvsdbIdl(self.OVS_CONNECT_SOCK, helper) self.conn = connection.Connection(idl=idl, timeout=self.OVSDB_CONNECT_TIMEOUT) self.ovs = ovs_impl_idl.OvsdbIdl(self.conn) return self def list_switches(self): """Return bridge names (equivalent to ovs-vsctl list-br).""" if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") return self.ovs.list_br().execute() or [] 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> """ if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") if bridge_name not in self.list_switches(): return [] ports = self.ovs.list_ports(bridge_name).execute() or [] if port_name not in ports: return [] return self.ovs.db_find("Port", ("name", "=", port_name)).execute() or [] def get_bridge_for_port(self, port_name): """ Determines which bridge a port belongs to CLI equivalent: ovs-vsctl port-to-br <port_name> """ if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") for bridge in self.list_switches(): ports = self.ovs.list_ports(bridge).execute() or [] if port_name in ports: return bridge return None def main(): print("Attempting to connect to remote Open vSwitch server via Unix socket...") try: manager = OVSDBManager().connect() port_name = input("\nEnter the port name to examine (e.g. tap123): ") bridge = manager.get_bridge_for_port(port_name) except Exception as e: print(f"Connection error: {e}") return 1 if bridge is None: print(f"Error: Port '{port_name}' does not exist on any switch.") return 1 if bridge != SWITCH_NAME: print( f"Error: Port '{port_name}' belongs to '{bridge}', not to '{SWITCH_NAME}'." ) return 1 print(f"Confirmed: Port '{port_name}' belongs to '{SWITCH_NAME}'.") port_records = manager.list_port_attributes(port_name, bridge) if not port_records: print("Failed to retrieve port attributes.") return 1 print(f"\nAttributes for port '{port_name}':") for record in port_records: table_data = sorted( [[k, str(v)] for k, v in record.items()], key=lambda x: x[0] ) print(tabulate(table_data, headers=["Attribute", "Value"], tablefmt="fancy")) return 0 if __name__ == "__main__": sys.exit(main()) ``` Here is a concise documentation of the methods and the logic of the main function in the script: __init__(self) : Initializes the `OVSDBManager` without opening a connection, setting `conn` and `ovs` to None so `connect()` must be called explicitly before any database operation. connect(self) : Creates the `IDL` helper, opens the **OVSDB** Unix-socket connection, and builds the `OvsdbIdl` wrapper, storing it in `self.ovs` and returning self for fluent usage. list_switches(self) : Returns the list of bridge names from **OVSDB*, equivalent to `ovs-vsctl list-br`, or an empty list if no bridges are found. list_port_attributes(self, port_name, bridge_name="dsw-host") : Checks connectivity, validates that the bridge and port exist, then queries the Port table and returns the matching records or an empty list. get_bridge_for_port(self, port_name) : Searches all known bridges and returns the name of the first bridge that contains the given port, or None if the port is not found. main() : Connects to **OVSDB**, prompts for a port name, finds its bridge, validates it belongs to **dsw-host**, then retrieves and tabulates the port’s attributes or reports an error. Here is an example of the code execution: ```bash python 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): tap100 Confirmed: Port 'tap100' belongs to 'dsw-host'. Attributes for port 'tap100': Attribute Value ----------------- ---------------------------------------------- _uuid 67832620-377c-4e55-85c8-ec2ea5fea32b bond_active_slave [] bond_downdelay 0 bond_fake_iface False bond_mode [] bond_updelay 0 cvlans [] external_ids {} fake_bridge False interfaces [UUID('04fb88b7-c260-424a-b141-e1e2b23236fb')] lacp [] mac [] name tap100 other_config {} protected False qos [] rstp_statistics {} rstp_status {} statistics {} status {} tag [] trunks [52, 600, 601] 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. ```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 SWITCH_NAME = "dsw-host" class OVSDBManager: """Manage OVSDB connection and cached read operations.""" OVS_CONNECT_SOCK = "unix:/tmp/ovs-forwarded.sock" OVSDB_CONNECT_TIMEOUT = 30 def __init__(self): """Initialize the manager without opening a connection.""" self.conn = None self.ovs = None self._port_cache = {} def connect(self): """Establish a connection to Open vSwitch via Unix socket.""" helper = idlutils.get_schema_helper(self.OVS_CONNECT_SOCK, "Open_vSwitch") helper.register_all() idl = connection.OvsdbIdl(self.OVS_CONNECT_SOCK, helper) self.conn = connection.Connection(idl=idl, timeout=self.OVSDB_CONNECT_TIMEOUT) self.ovs = ovs_impl_idl.OvsdbIdl(self.conn) return self def list_switches(self): """Return bridge names (equivalent to ovs-vsctl list-br).""" if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") return self.ovs.list_br().execute() or [] def _get_port_details(self, port_name, force_refresh=False): """Return port details and cache them to avoid repeated OVSDB queries.""" if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") if not force_refresh and port_name in self._port_cache: return self._port_cache[port_name] records = self.ovs.db_find("Port", ("name", "=", port_name)).execute() if not records: return None self._port_cache[port_name] = records[0] return self._port_cache[port_name] def get_bridge_for_port(self, port_name): """Return the bridge name that owns the given port, or None.""" if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") for bridge in self.list_switches(): ports = self.ovs.list_ports(bridge).execute() or [] if port_name in ports: return bridge return None def get_port_mode(self, port_name): """Return vlan_mode value ('access' or 'trunk') for the port.""" record = self._get_port_details(port_name) if not record: return None return record.get("vlan_mode") def get_port_access_vlan(self, port_name): """Return access VLAN ID for the port, or None.""" record = self._get_port_details(port_name) if not record: return None if self.get_port_mode(port_name) == "access" and record.get("tag"): return record["tag"] return None def get_port_trunk_vlan_list(self, port_name): """Return trunk VLAN list for the port, [] if all are allowed, or None.""" record = self._get_port_details(port_name) if not record: return None if self.get_port_mode(port_name) != "trunk": return None return record.get("trunks") or [] def main(): print("Attempting to connect to remote Open vSwitch server via Unix socket...") try: manager = OVSDBManager().connect() port_name = input("\nEnter the port name to examine (e.g. tap123): ") bridge = manager.get_bridge_for_port(port_name) except Exception as e: print(f"Connection error: {e}") return 1 if bridge is None: print(f"Error: Port '{port_name}' does not exist on any switch.") return 1 if bridge != SWITCH_NAME: print( f"Error: Port '{port_name}' belongs to '{bridge}', not to '{SWITCH_NAME}'." ) return 1 print(f"Confirmed: Port '{port_name}' belongs to '{SWITCH_NAME}'.") port_mode = manager.get_port_mode(port_name) if port_mode is None: print(f"Error: vlan_mode attribute not found for port '{port_name}'.") return 1 print(f"\nPort mode: {port_mode}") if port_mode == "access": vlan_id = manager.get_port_access_vlan(port_name) if vlan_id is None: print("No VLAN configured for this access port") else: print(f"Access VLAN ID: {vlan_id}") return 0 if 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") return 0 print("Unable to determine port mode") return 1 if __name__ == "__main__": sys.exit(main()) ``` Here is a short documentation for the new methods introduced in the `03_ovsdb_list_port_mode.py` script. _get_port_details(self, port_name, force_refresh=False) : Returns and caches detailed attributes for the given port, optionally bypassing the cache to refresh data from **OVSDB** when force_refresh is enabled. get_port_mode(self, port_name) : Returns the **vlan_mode** value ('access' or 'trunk') for the specified port, or None if the port is not found or the attribute is missing. get_port_access_vlan(self, port_name) : Returns the VLAN tag configured on an access port, or None if the port is not in access mode, has no tag, or the record cannot be retrieved. get_port_trunk_vlan_list(self, port_name) : Returns the list of allowed VLAN IDs for a trunk port, an empty list if all VLANs are allowed, or None if the port is not in trunk mode or missing. Rather than displaying all port attributes, the main function provides a focused output displaying only relevant VLAN information based on the port's operating mode. This targeted approach provides more meaningful insight into network segmentation configurations and maintains better performance through data caching and optimized database interactions. ```bash python 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 3: Declare switch configuration using a YAML file In this part, you will transition to applying a new configuration to the switch ports. You will use a 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, you only need a starting point to illustrate the configuration of a switch with a few ports and their attributes. Here is your 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 into the new script named `04_ovsdb_apply_config.py`. ```python= class ConfigLoader: """Load and expose YAML configuration content.""" def __init__(self, config_file): self.config_file = config_file self.config = None def load_config(self): 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 or {} except Exception as e: print(f"Error loading configuration: {e}") return None def get_switches(self): if not self.config: return [] return self.config.get("ovs", {}).get("switches", []) def print_config_summary(self): """Print a readable summary of YAML content.""" 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") 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 brief description of the `ConfigLoader` class methods. __init__(self, config_file) : Initializes the loader with the YAML file path and an empty internal configuration state. load_config(self) : Validates that the YAML file exists, loads and parses its content safely, and stores the resulting configuration dictionary or returns None on error. get_switches(self) : Returns the list of switch configuration entries from the loaded data, or an empty list if no configuration or switches are defined. print_config_summary(self) : Prints a human-readable summary showing how many switches and ports are defined, including a small table summarizing VLAN mode and VLAN information for each port. ### Step 3: Edit the OVSDBManager class This step introduces an enhanced version of the `OVSDBManager` class that now adds connection auto-initialization, bridge and port caching, and higher-level configuration methods, instead of only providing basic, on-demand read operations as in the previous part. ```python= class OVSDBManager: """Manage OVSDB connection and apply switch/port changes.""" OVS_CONNECT_SOCK = "unix:/tmp/ovs-forwarded.sock" OVSDB_CONNECT_TIMEOUT = 30 def __init__(self): self.conn = None self.ovs = None self._port_cache = {} self._bridges_cache = None def connect(self): """Establish a connection once and reuse it.""" helper = idlutils.get_schema_helper(self.OVS_CONNECT_SOCK, "Open_vSwitch") helper.register_all() idl = connection.OvsdbIdl(self.OVS_CONNECT_SOCK, helper) self.conn = connection.Connection(idl=idl, timeout=self.OVSDB_CONNECT_TIMEOUT) self.ovs = ovs_impl_idl.OvsdbIdl(self.conn) return self def _get_bridges(self, force_refresh=False): if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") if not force_refresh and self._bridges_cache is not None: return self._bridges_cache self._bridges_cache = self.ovs.list_br().execute() or [] return self._bridges_cache def _get_port_details(self, port_name, force_refresh=False): if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") if not force_refresh and port_name in self._port_cache: return self._port_cache[port_name] records = self.ovs.db_find("Port", ("name", "=", port_name)).execute() if not records: return None self._port_cache[port_name] = records[0] return self._port_cache[port_name] def _clear_cache(self, port_name=None, clear_bridges=False): if port_name: self._port_cache.pop(port_name, None) else: self._port_cache.clear() if clear_bridges: self._bridges_cache = None def _port_in_bridge(self, port_name, bridge_name): if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") ovs = self.ovs ports = ovs.list_ports(bridge_name).execute() or [] return port_name in ports def apply_port_config(self, port_config, bridge_name=SWITCH_NAME): try: if self.ovs is None: raise RuntimeError("Not connected to OVSDB. Call connect() first.") if bridge_name not in self._get_bridges(): print(f"Error: Bridge '{bridge_name}' does not exist.") return False port_name = port_config.get("name") if not port_name: print("Error: Port configuration missing 'name' field.") return False if not self._port_in_bridge(port_name, bridge_name): print( f"Error: Port '{port_name}' does not exist on bridge '{bridge_name}'." ) return False self._clear_cache(port_name) 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() if vlan_mode == "access": self.ovs.db_set("Port", port_name, ("trunks", [])).execute() elif vlan_mode == "trunk": self.ovs.db_set("Port", port_name, ("tag", [])).execute() if vlan_mode == "access" and "tag" in port_config: print( f"Setting port '{port_name}' access VLAN tag to {port_config['tag']}" ) self.ovs.db_set( "Port", port_name, ("tag", port_config["tag"]) ).execute() if vlan_mode == "trunk" and "trunks" in port_config: print( f"Setting port '{port_name}' trunk VLANs to {port_config['trunks']}" ) self.ovs.db_set( "Port", port_name, ("trunks", port_config["trunks"]) ).execute() self._get_port_details(port_name, force_refresh=True) return True except Exception as e: print(f"Error applying port configuration: {e}") return False def apply_switch_config(self, switch_config): try: switch_name = switch_config.get("name") if not switch_name: print("Error: Switch configuration missing 'name' field.") return False if switch_name not in self._get_bridges(force_refresh=True): print(f"Error: Switch '{switch_name}' does not exist.") return False ports = switch_config.get("ports", []) success_count = sum( 1 for port_config in ports if self.apply_port_config(port_config, switch_name) ) print( f"Applied configuration to {success_count} out of {len(ports)} ports." ) 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 documentation of the new private methods of the `OVSDBManager` Class. __init__(self) : Initializes connection and cache attributes, and optionally auto-connects to **OVSDB**, printing a warning if automatic connection fails so users know they must call connect() manually. _get_bridges(self, force_refresh=False) : Returns the list of bridge names, using a cached copy unless force_refresh is requested, and handles errors by logging a message and returning an empty list. _get_port_details(self, port_name, force_refresh=False) : Returns and caches detailed attributes for a given port, refreshing from **OVSDB** when requested and logging an informative message if the port does not exist. _clear_cache(self, port_name=None, clear_bridges=False) : Clears cached data for a specific port or all ports, and optionally resets the bridge cache to force fresh reads on subsequent operations. Here is a concise documentation of the two key configuration methods. apply_port_config(self, port_config, bridge_name=SWITCH_NAME) : Applies VLAN-related settings from a single port configuration entry, validating bridge and port existence, updating vlan_mode, tag, and trunks in **OVSDB**, refreshing caches, and reporting any configuration errors. apply_switch_config(self, switch_config) : Validates the target switch, iterates over all declared ports, calls apply_port_config() for each, prints how many ports were successfully updated, clears caches, and returns True only if all ports were applied. ### Step 4: Edit the script main function The `main()` function 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 path as a positional argument. It also supports a `--dry-run` flag, allowing administrators to preview and validate intended network changes without applying them to the switch. This provides a safe way to verify configurations before committing them 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 import sys import yaml from tabulate import tabulate ``` Here is a copy of the edited `main()` function code. ```python= def main(): 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: config_loader = ConfigLoader(args.config_file) if config_loader.load_config() is None: print("Failed to load configuration.") return 1 config_loader.print_config_summary() if args.dry_run: print("\nDry run completed. No changes were made.") return 0 manager = OVSDBManager().connect() switches = config_loader.get_switches() overall_success = all( manager.apply_switch_config(sw_cfg) for sw_cfg in switches ) if overall_success: print("\nConfiguration applied successfully!") return 0 print("\nConfiguration applied with errors.") return 1 except Exception as e: print(f"Error: {e}") return 1 if __name__ == "__main__": sys.exit(main()) ``` ### Step 5: Test the `--dry-run` flag Here is a sample run of the `04_ovsdb_apply_config.py` script with the `--dry-run` flag ```bash python 04_ovsdb_apply_config.py switch_config.yaml --dry-run ``` ```bash= Configuration loaded: 1 switch(es) defined - Switch 'dsw-host': 3 port(s) defined Port Mode VLAN Config ------ ------ -------------------- tapXX2 access VLAN 52 tapXX5 access VLAN 410 tapXX6 trunk VLANs [52, 410, 420] Dry run completed. No changes were made. ``` The output shows that the YAML declarations were parsed correctly. ### Step 6: Apply the switch ports configuration Now that our Python code is complete, we are ready to apply the declared switch ports configuration. ```bash python 04_ovsdb_apply_config.py switch_config.yaml ``` ```bash= Configuration loaded: 1 switch(es) defined - Switch 'dsw-host': 3 port(s) defined Port Mode VLAN Config ------ ------ -------------------- tapXX2 access VLAN 52 tapXX5 access VLAN 410 tapXX6 trunk VLANs [52, 410, 420] Setting port 'tapXX2' VLAN mode to 'access' Setting port 'tapXX2' access VLAN tag to 52 Setting port 'tapXX5' VLAN mode to 'access' Setting port 'tapXX5' access VLAN tag to 410 Setting port 'tapXX6' VLAN mode to 'trunk' Setting port 'tapXX6' trunk VLANs to [52, 410, 420] Applied configuration to 3 out of 3 ports. Configuration applied successfully! ``` ### Step 7: 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 tapXX5 ``` ```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 : tapXX5 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 tapXX6 ``` ```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 : tapXX6 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.