mfd-dcb 1.18.0

A Python module providing cross-platform Data Center Bridging (DCB) configuration and management APIs for Windows, Linux, and FreeBSD systems.

[!IMPORTANT] This project is under development. All source code and features on the main branch are for the purpose of testing or evaluation and not production ready.

MFD DCB

A Python module providing cross-platform Data Center Bridging (DCB) configuration and management APIs for Windows, Linux, and FreeBSD systems.

Usage

from mfd_dcb import Dcb
from mfd_connect import RPyCConnection
from mfd_typing import DeviceID
conn = RPyCConnection(ip="x.x.x.x")
dcb_obj = Dcb(connection=conn)

print(dcb_obj.is_willing())
print(dcb_obj.set_willing(enable=False, is_40g_or_later=True))
print(dcb_obj.set_policy('test1', 1, "-IPSrcPrefixMatchCondition x.x.x.x -IPDstPortStartMatchCondition 5001 -IPDstPortEndMatchCondition 5001"))
print(dcb_obj.remove_policy('test1'))
print(dcb_obj.get_policies())
print(dcb_obj.verify_policy(dcb_obj.verify_policy({'test1': {'PriorityValue': '1'}})))
print(dcb_obj.is_qos_enabled("SLOT 4 Port 1"))
print(dcb_obj.set_qos("SLOT 4 Port 1", True))
print(dcb_obj.set_default_config())
print(dcb_obj.set_pfc([1, 1, 1, 1, 1, 1, 1, 1]))
print(dcb_obj.set_ets(['1', '2', '3', '4', '5', '6', '7'], [5, 7, 10, 15, 17, 20, 23]))
print(dcb_obj.get_dcb("SLOT 4 Port 1"))
print(dcb_obj.get_ets("SLOT 4 Port 1"))
print(dcb_obj.get_pfc("SLOT 4 Port 1"))
print(dcb_obj.get_app("SLOT 4 Port 1"))
print(dcb_obj.get_remote_ets("SLOT 4 Port 1"))
print(dcb_obj.get_remote_pfc("SLOT 4 Port 1"))
print(dcb_obj.get_remote_app("SLOT 4 Port 1"))
print(dcb_obj.set_dcb_operational_value("Intel(R) Ethernet Network Adapter E810-C-Q2", "DCB", "Enabled"))
print(dcb_obj.get_dcb_operational_value("Intel(R) Ethernet Network Adapter E810-C-Q2", "DCB Version"))

val = dcb_obj.get_dcb("SLOT 4 Port 1")
print(dcb_obj.verify_dcb("SLOT 4 Port 1", val))

print(dcb_obj.verify_ets("SLOT 4 Port 1", {'ETS': {'0': {'TSA': 'ETS', 'Bandwidth': 1000, 'Priorities': [0, 1, 2, 3, 4, 5, 6, 7]}}}))
print(dcb_obj.verify_pfc("SLOT 4 Port 1", {'PFC': [True, False, True, False, True, False, True, False]}))
print(dcb_obj.verify_app("SLOT 4 Port 1", {'APP': {'0x8906': {'Priority': 3, 'Protocol': 'Ethertype'}, '3260': {'Priority': 34, 'Protocol': 'TCP'}}}))
print(dcb_obj.pfc_bool_to_str(True))
print(dcb_obj.remove_dcb_all_user_priorities())
print(dcb_obj.get_all_dcb_operational_values("Intel(R) Ethernet Network Adapter E810-C-Q2"))
print(dcb_obj.get_dcb_status("Intel(R) Ethernet Network Adapter E810-C-Q2"))
print(dcb_obj.set_dcb_status("Intel(R) Ethernet Network Adapter E810-C-Q2", enable_dcb=False, enable_willing_mode=False))
print(dcb_obj.get_pfc_enabled_bits("SLOT 4 Port 1", "Intel(R) Ethernet Network Adapter E810-C-Q2", True))
print(dcb_obj.get_pfc_port_statistics("Intel(R) Ethernet Network Adapter E810-C-Q2", 1))
print(dcb_obj.get_pfc_counters("Intel(R) Ethernet Network Adapter E810-C-Q2"))
print(dcb_obj.set_dcb_user_priority_to_tcp_port("10.10.10.10"))
print(dcb_obj.get_dcb_max_supported_traffic_class(True, DeviceID(0x10FB)))
print(dcb_obj.set_dcb(True, "SLOT 4 Port 1"))

Implemented methods

# Below APIs supported in WINDOWS
'is_willing(self) -> bool' - Get Willing status of DCB configuration on the server
'set_willing(self, enable: bool, is_40g_or_later: bool, adapter_name: str = "", mode: str = "ieee") -> None' - Turn ON or OFF Willing status of DCB configuration on the server
'set_policy(self, name: str, priority: int, parameters: str = "") -> None' - Create Quality of Service policy
'remove_policy(self, name: str) -> None' - Remove Quality of Servicy policy with a given name
'get_policies(self) -> Dict' - Get all available Quality of Service policies
'verify_policy(self, expected_policies: Dict, policies: Dict = None) -> bool' - Verify if expected NetQosPolicies are correct
'is_qos_enabled(self, adapter_name: str, from_cache: bool = False) -> bool' - Get QoS Enabled status
'set_qos(self, adapter_name: str, enable: bool) -> None' - Turn ON or OFF Quality of Service on the adapter
'set_default_config(self) -> None' - Revert to default DCB settings - all User Priorities in Traffic Class 0 with bandwidth 100% and PFC disabled
'set_pfc(self, pfc_per_priority: list, mode: str = "ieee") -> None' - Turn ON or OFF Priority Flow Control for each User Priority
'set_ets(self, user_priority_mapping: list, bandwidth_per_traffic_class: list, mode: str = "ieee") -> None' - Set ETS configuration based on User Priority mapping to Traffic Classes and given bandwidth
'get_dcb(self, adapter_name: str) -> Dict' - Get complete DCB configuration for ETS, PFC and APP TLVs, local and remote
'get_ets(self, adapter_name: str, from_cache: bool = False) -> Dict' - Get local ETS TLV configuration of the adapter
'get_pfc(self, adapter_name: str, from_cache: bool = False) -> Dict' - Get local PFC TLV configuration of the adapter
'get_app(self, adapter_name: str, from_cache: bool = False) -> Dict' - Get local APP TLV configuration of the adapter
'get_remote_ets(self, adapter_name: str, from_cache: bool = False) -> Dict' - Get remote ETS TLV configuration of the adapter
'get_remote_pfc(self, adapter_name: str, from_cache: bool = False) -> Dict' - Get remote PFC TLV configuration of the adapter
'get_remote_app(self, adapter_name: str, from_cache: bool = False) -> Dict' - Get remote APP TLV configuration of the adapter
'set_dcb_operational_value(self, adapter_description: str, dcb_setting: str, dcb_value: str) -> None' - Set operational value for a given DCB setting
'get_dcb_operational_value(self, adapter_description: str, dcb_setting: str) -> str' - Retrieve the operational value for a given DCB setting
'verify_dcb(self, adapter_name: str, dcb_map: Dict[str, Dict[str, str]], dcb_config: Optional[Dict[str, str]] = None) -> bool' - Verify if DCB configuration is correct
'verify_ets(self, adapter_name: str, dcb_map: Dict[str, Dict[str, str]], ets_config: Optional[Dict[str, Dict[str, str]]] = None) -> bool' - Verify if ETS configuration is correct
'verify_pfc(self, adapter_name: str, dcb_map: Dict[str, List[bool]], pfc_config: Optional[Dict[str, List[bool]]] = None) -> bool' - Verify if PFC configuration is correct
'verify_app(self, adapter_name: str, dcb_map: Dict[str, Dict[str, str]], app_config: Optional[Dict[str, Dict[str, str]]] = None) -> bool' - Verify if APP TLV configuration is correct
'pfc_bool_to_str(self, enabled: bool) -> str' - Convert bool value to string enabled/disabled
'remove_dcb_all_user_priorities(self) -> None' - Remove all DCB user priorities
'get_all_dcb_operational_values(self, adapter_description: str) -> Dict[str, str]' - Retrieves the operational value for a given DCB setting
'get_dcb_status(self, adapter_description: str) -> tuple' - Get DCB status (enabled or disabled) and willing mode
'set_dcb_status(self, adapter_description: str, *, enable_dcb: bool, enable_willing_mode: bool) -> bool' - Set DCB option and willing mode
'get_pfc_enabled_bits(self, adapter_name: str, adapter_description: str, is_40g_or_later: bool -> Optional[tuple]' - Get remote and local PFC enable bits for all user priorities
'get_pfc_port_statistics(self, adapter_description: str, priority: int) -> Dict[str, int]' - Get PFC RX/TX XOFF, XON counters for the port for a given priority
'get_pfc_counters(self, adapter_description: str) -> Dict[int, Dict[str, int]]' - Get PFC counters values for all User Priorities
'set_dcb_user_priority_to_tcp_port(self, tested_ip: str, user_priority: int = 1, tcp_port: int = 5000) -> None' - Set user priority to port
'get_dcb_max_supported_traffic_class(self, is_40g_or_later: bool, device_id: "DeviceID") -> int' - Get maximum number of traffic class that the current adapter support
'set_dcb(self, enabled: bool, adapter_name: str) -> None' - Set dcb feature: enabled/disabled

Usage LinuxDcb

from mfd_dcb import Dcb
from mfd_dcb import LinuxDcb
from mfd_connect import SSHConnection
conn = SSHConnection(username="xxx", password="xxx", ip="x.x.x.x")
dcb_obj = LinuxDcb(connection=conn)

user_priority_mapping = [(0, 1), (3, 2), (5, 4), (6, 7)]
bandwidth_per_traffic_class = [25, 25, 25, 25]
print(dcb_obj.set_ets(user_priority_mapping, bandwidth_per_traffic_class, interface_name="eth4", mode="ieee"))
pfc_per_priority = [False, True, True, False, False, False, False, False]
print(dcb_obj.set_pfc(interface_name="eth4",pfc_per_priority=pfc_per_priority))
print(dcb_obj.restart_lldpad())
print(dcb_obj.set_dcbx_mode(interface_name="eth4", mode="ieee"))
print(dcb_obj.is_sw_dcb_mode_enabled(interface_name="eth4"))
print(dcb_obj.is_sw_dcb_mode_enabled(interface_name="eth4", dcbx_mode="ieee"))
print(dcb_obj.set_willing(interface_name="eth4", enable=True, mode="ieee", is_fwlldp_enabled=False))
print(dcb_obj.set_willing(interface_name="eth4", enable=True, mode="ieee", is_fwlldp_enabled=True))
print(dcb_obj.set_willing(interface_name="eth4", enable=True, mode="cee", is_fwlldp_enabled=True))
print(dcb_obj.set_willing(interface_name="eth4", enable=False, mode="cee", is_fwlldp_enabled=False))
print(dcb_obj.get_pfc_counters(interface_name="enp59s0f1"))
print(dcb_obj.get_pfc_counters(interface_name="enp59s0f1", is_40g_adapter=True))
print(dcb_obj.get_pfc_counters(interface_name="enp94s0f0", is_10g_adapter=True))
print(dcb_obj.remove_lldpad_conf())
print(dcb_obj.get_dcb(interface_name="eth4", tool_name="dcbnl"))
print(dcb_obj.get_ets(interface_name="eth4", tool_name="dcbnl"))
print(dcb_obj.get_app(interface_name="eth4", tool_name="dcbnl"))
print(dcb_obj.get_pfc(interface_name="eth4", tool_name="dcbnl"))
print(dcb_obj.get_dcb(interface_name="eth4", tool_name="dcbtool"))
print(dcb_obj.get_ets(interface_name="eth4", tool_name="dcbtool"))
print(dcb_obj.get_app(interface_name="eth4", tool_name="dcbtool"))
print(dcb_obj.get_pfc(interface_name="eth4", tool_name="dcbtool"))
DCB_MAP = {
            "LOCAL_ETS": {
                "0": {"TSA": "ETS", "Bandwidth": 40, "Priorities": [0, 1, 2, 5, 6, 7]},
                "1": {"TSA": "ETS", "Bandwidth": 30, "Priorities": [3]},
                "2": {"TSA": "ETS", "Bandwidth": 30, "Priorities": [4]},
            },
            "LOCAL_PFC": [False, False, False, True, True, False, False, False],
            "LOCAL_APP": {"3260": {"Priority": 4, "Protocol": "TCP"}},
        }
print(dcb_obj.verify_dcb(interface_name="eth4", dcb_map=DCB_MAP))
print(dcb_obj.verify_ets(interface_name="eth4", dcb_map=DCB_MAP))
print(dcb_obj.verify_pfc(interface_name"eth4", dcb_map=DCB_MAP))
print(dcb_obj.verify_app(interface_name"eth4", dcb_map=DCB_MAP))

Implemented methods

# Below APIs supported in LINUX
'set_ets(self, user_priority_mapping, bandwidth_per_traffic_class, mode="ieee", interface_name: str) -> None:' - Set ETS configuration based on User Priority mapping to Traffic Classes and given bandwidth.

'set_pfc(self, interface_name: str, pfc_per_priority: list = [], mode: str = "ieee") -> None:' - Turn ON or OFF Priority Flow Control for each User Priority.

'restart_lldpad(self, action: str = "default") -> None:' - Restart/stop lldpad service and lldp agent daemon

'set_sw_dcb(self, interface_name: str, enable: bool = False, dcbx_mode: str = "ieee") -> None:' - Turn ON or OFF software DCB

'set_dcbx_mode(self, interface_name: str, mode: str = "", fw_lldp: bool = False) -> None:' - Set adapter DCB to given DCBX mode.

'is_sw_dcb_mode_enabled(self, interface_name: str, dcbx_mode: str) -> bool:' - Check software DCB status.
'set_willing(self, interface_name: str, enable: bool, mode: str, is_fwlldp_enabled: bool) - >None:' - Turn ON or OFF Willing of remote DCB configuration on the server.
'get_pfc_counters(self, interface_name: str, is_10g_adapter: bool = False, is_40g_adapter: bool = False) -> Dict[str, Dict[str, str]]:' - Get PFC counters values for all User Priorities.
'remove_lldpad_conf(self) -> None' - Remove lldpad conf file to restore settings.
'get_dcb(self, interface_name: str, tool_name: str = dcbnl) -> Dict[str, Dict[str, str]] - Get complete DCB configuration for ETS, PFC and APP TLVs, local and remote.'
'get_ets(self, interface_name: str, tool_name: str = dcbnl, from_cache: bool = False) -> Dict[str, Dict[str, Any]]:' - Get local ETS TLV configuration of the adapter.
'get_tool(self, interface_name: str, tool_name: str = dcbnl) -> str:' - Returns particular tool based on tool name.
'get_app(self, interface_name: str, tool_name: str = dcbnl, from_cache: bool = False) -> Dict[str, Dict[str, Any]]:' - Get local APP TLV configuration of the adapter.
'def get_pfc(self, interface_name: str, tool_name: str = dcbnl) -> Dict[str, Dict[str, Any]]:' - Get local PFC TLV configuration of the adapter.
'get_dcb(self, interface_name: str, tool_name: str = dcbtool) -> Dict[str, Dict[str, str]] - Get complete DCB configuration for ETS, PFC and APP TLVs, local and remote.'
'get_ets(self, interface_name: str, tool_name: str = dcbtool, from_cache: bool = False) -> Dict[str, Dict[str, Any]]:' - Get local ETS TLV configuration of the adapter.
'get_tool(self, interface_name: str, tool_name: str = dcbtool) -> str:' - Returns particular tool based on tool name.
'def get_pfc(self, interface_name: str, tool_name: str = dcbtool) -> Dict[str, Dict[str, Any]]:' - Get local PFC TLV configuration of the adapter.
'verify_dcb(self, interface_name: str, dcb_map: dict, dcb_config: dict | None = None, retry: int | None = 6, interval: int | None = 10, switch: object | None = None, switch_port: str | None = None) -> bool:' - Verify if DCB configuration is correct
'verify_ets(self, interface_name: str, dcb_map: dict, ets_config: dict | None = None) -> bool:' - Verify if ETS configuration is correct
'verify_pfc(self, interface_name: str, dcb_map: dict, pfc_config: dict | None = None) -> bool:' - Verify if PFC configuration is correct
'verify_app(self, interface_name: str, dcb_map: dict, app_config: dict | None = None) -> bool:' - Verify if APP TLV configuration is correct.

Required Tools for LinuxDcb

• lldptool v1.1.0
• dcbtool v1.1.0
• dcbnl.py - DCB_TOOL_PATH_LNX = r"/tmp/tools/dcb"' dcbnl.py or dcbgetset to query the active dcb configuration on an interface ./dcbnl.py $interface Netlink message generation/parsing script is a standalone script which is not maintained by Intel.

Usage FreeBsdDcb

from mfd_dcb import Dcb
from mfd_dcb import FreeBsdDcb, DSCPMap
from mfd_connect import RPyCConnection
conn = RPyCConnection(ip="x.x.x.x")
dcb_obj = FreeBsdDcb(connection=conn)

dscpmap = [0] * 64
dcb_obj.dscp_apply_map(interface_name="ice0", dscpmap=dscpmap)
print(dcb_obj.dscp_read_map(interface_name="ice0"))
dscpmap[0]=7
dcb_obj.dscp_apply_map(interface_name="ice0", dscpmap=dscpmap)
assert dcb_obj.dscp_verify_map(interface_name="ice0", dscpmap) is True
dcb_obj.set_pfc_mode("ice0", PfcMode.DSCP)
print(dcb_obj.get_pfc_mode("ice0"))
dcb_obj.set_ets_min_rate("ice0", [100, 0, 0, 0, 0, 0, 0, 0])
print(dcb_obj.get_ets_min_rate("ice0"))

Implemented methods

# Below APIs supported in FreeBSD
'dscp_apply_map(self, interface_name: str, dscpmap: List[int]) -> None:' - Configure DSCP2TC map on the interface.
'dscp_read_map(self, interface_name: str) -> List[int]:' - Read DSCP2TC map from the interface.
'dscp_verify_map(self, interface_name: str, dscpmap: List[int]) -> bool:' - Check that the interface has the same DSCP2TC map.
'set_pfc_mode(self, interface_name: str, mode: PfcMode) -> None:' - Enable/disable DSCP mode on the interface.
'get_pfc_mode(self, interface_name: str) -> PfcMode:' - Get the current PFC mode on the interface.
'set_ets_min_rate(self, interface_name: str, ets: List[int]) -> None:' - Configure ETS on an interface.
'get_ets_min_rate(self, interface_name: str) -> List[int]:' - Read current ETS bandwidth table.

OS supported:

Here is a place to write what OSes support your MFD module:

• WINDOWS
• LINUX
• FREEBSD

Issue reporting

If you encounter any bugs or have suggestions for improvements, you're welcome to contribute directly or open an issue here.