#!/usr/bin/env python3
# MIT License
#
# Copyright (c) 2020 FABRIC Testbed
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# Author: Paul Ruth (pruth@renci.org)
"""
Methods to work with FABRIC components_.
.. _components: https://learn.fabric-testbed.net/knowledge-base/glossary/#component
You normally would not create :class:`Component` objects directly with
a constructor call; they are created when you invoke
:py:func:`fabrictestbed_extensions.fablib.node.Node.add_component()`,
like so::
node.add_component(model='NVME_P4510', name="nvme1")
node.add_component(model='NIC_Basic', name="nic1")
"""
from __future__ import annotations
import json
import time
from typing import TYPE_CHECKING, Optional, Union
from fim.user import ComponentType
from fabrictestbed_extensions.fablib.constants import Constants
from fabrictestbed_extensions.fablib.exceptions import (
ResourceNotFoundError,
SliceStateError,
)
from fabrictestbed_extensions.fablib.template_mixin import TemplateMixin
from fabrictestbed_extensions.utils.utils import Utils
if TYPE_CHECKING:
from fabrictestbed_extensions.fablib.slice import Slice
from fabrictestbed_extensions.fablib.node import Node
from fabrictestbed_extensions.fablib.interface import Interface
import logging
from typing import List
from fabrictestbed.slice_editor import Component as FimComponent
from fabrictestbed.slice_editor import ComponentModelType, Flags, Labels, UserData
from tabulate import tabulate
log = logging.getLogger("fablib")
[docs]
class Component(TemplateMixin):
"""Represents a hardware component attached to a FABRIC node.
Components (NICs, GPUs, FPGAs, NVMe, etc.) extend a node's capabilities and
are typically created via :meth:`Node.add_component()
<fabrictestbed_extensions.fablib.node.Node.add_component>` rather than direct
instantiation.
:cvar dict component_model_map: Mapping of component model names to FIM types.
:cvar dict component_configure_commands: Component-specific configuration commands.
"""
_show_title = "Component"
component_model_map = {
Constants.CMP_NIC_Basic: ComponentModelType.SharedNIC_ConnectX_6,
Constants.CMP_NIC_BlueField2_ConnectX_6: ComponentModelType.SmartNIC_BlueField_2_ConnectX_6,
Constants.CMP_NIC_ConnectX_6: ComponentModelType.SmartNIC_ConnectX_6,
Constants.CMP_NIC_ConnectX_5: ComponentModelType.SmartNIC_ConnectX_5,
Constants.CMP_NIC_ConnectX_7_100: ComponentModelType.SmartNIC_ConnectX_7_100,
Constants.CMP_NIC_ConnectX_7_400: ComponentModelType.SmartNIC_ConnectX_7_400,
Constants.CMP_NIC_P4: Constants.P4_DedicatedPort,
Constants.CMP_NVME_P4510: ComponentModelType.NVME_P4510,
Constants.CMP_GPU_TeslaT4: ComponentModelType.GPU_Tesla_T4,
Constants.CMP_GPU_RTX6000: ComponentModelType.GPU_RTX6000,
Constants.CMP_GPU_A40: ComponentModelType.GPU_A40,
Constants.CMP_GPU_A30: ComponentModelType.GPU_A30,
Constants.CMP_NIC_OpenStack: ComponentModelType.SharedNIC_OpenStack_vNIC,
Constants.CMP_FPGA_Xilinx_U280: ComponentModelType.FPGA_Xilinx_U280,
Constants.CMP_FPGA_Xilinx_SN1022: ComponentModelType.FPGA_Xilinx_SN1022,
}
component_configure_commands = {
Constants.CMP_NIC_ConnectX_7_100: [
"sudo ip addr add 192.168.100.1/24 dev tmfifo_net0",
"sudo ip link set tmfifo_net0 up",
"sudo bfb-install --bfb /opt/bf-bundle/bf-bundle-2.9.1-40_24.11_ubuntu-22.04_prod.bfb --rshim rshim0",
],
Constants.CMP_NIC_ConnectX_7_400: [
"sudo ip addr add 192.168.100.1/24 dev tmfifo_net0",
"sudo ip link set tmfifo_net0 up",
"sudo bfb-install --bfb /opt/bf-bundle/bf-bundle-2.9.1-40_24.11_ubuntu-22.04_prod.bfb --rshim rshim0",
],
Constants.CMP_NIC_BlueField2_ConnectX_6: [
"sudo ip addr add 192.168.100.1/24 dev tmfifo_net0",
"sudo ip link set tmfifo_net0 up",
"sudo bfb-install --bfb /opt/bf-bundle/bf-bundle-2.9.1-40_24.11_ubuntu-22.04_prod.bfb --rshim rshim0",
],
}
[docs]
def __str__(self):
"""
Creates a tabulated string describing the properties of the component.
Intended for printing component information.
:return: Tabulated string of component information
:rtype: String
"""
table = [
["Name", self.get_name()],
["Details", self.get_details()],
["Disk", self.get_disk()],
["Units", self.get_unit()],
["PCI Address", self.get_pci_addr()],
["Model", self.get_model()],
["Type", self.get_type()],
]
return tabulate(table)
[docs]
@staticmethod
def get_pretty_name_dict():
"""
Returns the mapping used when rendering table headers.
"""
return {
"name": "Name",
"short_name": "Short Name",
"details": "Details",
"disk": "Disk",
"units": "Units",
"pci_address": "PCI Address",
"model": "Model",
"type": "Type",
"dev": "Device",
"node": "Node",
"numa": "Numa Node",
}
[docs]
def toDict(self, skip: Optional[List[str]] = None):
"""
Returns the component attributes as a dictionary.
Results are cached. Cache is invalidated when ``_invalidate_cache()``
is called.
:param skip: list of keys to exclude
:type skip: List[str]
:return: component attributes as dictionary
:rtype: dict
"""
if skip is None:
skip = []
if self._cached_dict is None:
d = {}
d["name"] = str(self.get_name())
d["short_name"] = str(self.get_short_name())
d["details"] = str(self.get_details())
d["disk"] = str(self.get_disk())
d["units"] = str(self.get_unit())
d["pci_address"] = str(self.get_pci_addr())
d["model"] = str(self.get_model())
d["type"] = str(self.get_type())
d["dev"] = str(self.get_device_name())
d["node"] = str(self.get_node().get_name()) if self.get_node() else ""
d["numa"] = str(self.get_numa_node())
self._cached_dict = d
if not skip:
return dict(self._cached_dict)
return {k: v for k, v in self._cached_dict.items() if k not in skip}
[docs]
def generate_template_context(self, skip: Optional[List[str]] = None):
"""
Generate the base template context for this component.
Creates a dictionary context suitable for Jinja2 template rendering,
including component attributes and an empty interfaces list.
:param skip: list of keys to exclude
:type skip: List[str]
:return: Template context dictionary with component attributes
:rtype: dict
"""
context = self.toDict(skip=skip)
context["interfaces"] = []
return context
[docs]
def list_interfaces(
self,
fields=None,
output=None,
quiet=False,
filter_function=None,
refresh: bool = False,
):
"""
Lists all the interfaces in the component with their attributes.
There are several output options: "text", "pandas", and "json" that determine the format of the
output that is returned and (optionally) displayed/printed.
output: 'text': string formatted with tabular
'pandas': pandas dataframe
'json': string in json format
fields: json output will include all available fields/columns.
Example: fields=['Name','MAC']
filter_function: A lambda function to filter data by field values.
Example: filter_function=lambda s: s['Node'] == 'Node1'
:param output: output format
:type output: str
:param fields: list of fields (table columns) to show
:type fields: List[str]
:param quiet: True to specify printing/display
:type quiet: bool
:param filter_function: lambda function
:type filter_function: lambda
:param refresh: Refresh the interface object with latest Fim info
:type refresh: bool
:return: table in format specified by output parameter
:rtype: Object
"""
ifaces = []
for iface in self.get_interfaces(refresh=refresh):
ifaces.append(iface.get_name())
name_filter = lambda s: s["Name"] in set(ifaces)
if filter_function is not None:
filter_function = lambda x: filter_function(x) + name_filter(x)
else:
filter_function = name_filter
return self.get_slice().list_interfaces(
fields=fields,
output=output,
quiet=quiet,
filter_function=filter_function,
refresh=refresh,
)
[docs]
@staticmethod
def calculate_name(node: Node = None, name: str = None) -> str:
"""
Not intended for API use
"""
# Hack to make it possile to find interfaces
return f"{node.get_name()}-{name}"
[docs]
@staticmethod
def new_component(
node: Node = None, model: str = None, name: str = None, user_data: dict = {}
):
"""
Not intended for API use
Creates a new FIM component on the fablib node inputted.
:param node: the fablib node to build the component on
:type node: Node
:param model: the name of the component type to build
:type model: str
:param name: the name of the new component
:type name: str
:return: the new fablib compoent
:rtype: Component
"""
# Hack to make it possile to find interfaces
name = Component.calculate_name(node=node, name=name)
component = Component(
node=node,
fim_component=node.fim_node.add_component(
model_type=Component.component_model_map[model], name=name
),
)
component.set_user_data(user_data)
component.get_interfaces(refresh=True)
return component
def __init__(self, node: Node = None, fim_component: FimComponent = None):
"""
Typically invoked when you add a component to a ``Node``.
.. note ::
``Component`` constructer is not meant to be directly used.
:param node: the fablib node to build the component on
:type node: Node
:param fim_component: the FIM component this object represents
:type fim_component: FIMComponent
"""
super().__init__()
self.fim_component = fim_component
self.fim_model = None
self.node = node
self.interfaces = {}
# Dict for toDict() / template rendering
self.dict = {
"name": "",
"short_name": "",
"details": "",
"disk": "",
"units": "",
"pci_address": "",
"model": "",
"type": "",
"dev": "",
"node": "",
"numa": "",
}
# V2 specific: cached FIM properties
self._cached_details: Optional[str] = None
self._cached_numa_node: Optional[str] = None
self._cached_disk: Optional[int] = None
self._cached_unit: Optional[int] = None
self._cached_bdf: Optional[str] = None
self._cached_fim_model: Optional[str] = None
self._cached_fim_type: Optional[str] = None
self._cached_device_name: Optional[str] = None
def _invalidate_cache(self):
"""Invalidate all cached properties."""
super(Component, self)._invalidate_cache()
self._cached_details = None
self._cached_numa_node = None
self._cached_disk = None
self._cached_unit = None
self._cached_bdf = None
self._cached_fim_model = None
self._cached_fim_type = None
self._cached_device_name = None
self.interfaces = {}
self.dict = {
"name": "",
"short_name": "",
"details": "",
"disk": "",
"units": "",
"pci_address": "",
"model": "",
"type": "",
"dev": "",
"node": self.get_node().get_name() if self.get_node() else "",
"numa": "",
}
[docs]
def update(self, fim_component: Optional[FimComponent] = None):
"""
Update the component with new FIM data.
:param fim_component: The new FIM component data
:type fim_component: FimComponent
"""
if fim_component:
self.fim_component = fim_component
self._invalidate_cache()
self._fim_dirty = False
[docs]
def get_interfaces(
self, include_subs: bool = True, refresh: bool = False, output: str = "list"
) -> Union[dict[str, Interface], list[Interface]]:
"""
Gets the interfaces attached to this fablib component's FABRIC component.
Results are cached. Use refresh=True to force reload from FIM.
:param include_subs: Flag indicating if sub interfaces should be included
:type include_subs: bool
:param refresh: Refresh the interface object with latest Fim info
:type refresh: bool
:param output: Specify how the return type is expected; Possible values: list or dict
:type output: str
:return: a list or dict of the interfaces on this component.
:rtype: Union[dict[str, Interface], list[Interface]]
"""
from fabrictestbed_extensions.fablib.interface import Interface
if self.interfaces and not refresh and not self._fim_dirty:
if output == "dict":
return self.interfaces
return list(self.interfaces.values())
self.interfaces = {}
for fim_interface in self.get_fim().interface_list:
iface = Interface(component=self, fim_interface=fim_interface)
self.interfaces[iface.get_name()] = iface
if include_subs:
child_interfaces = iface.get_interfaces(refresh=refresh, output="dict")
if child_interfaces and len(child_interfaces):
self.interfaces.update(child_interfaces)
if output == "dict":
return self.interfaces
return list(self.interfaces.values())
[docs]
def get_interface(
self, name: str = None, network_name: str = None, refresh: bool = False
) -> Optional[Interface]:
"""
Gets a particular interface attached to this component.
Accepts either the interface name or a network_name. If a network name
is used, returns the interface connected to that network. If both name
and network_name are provided, name takes precedence.
:param name: the name of the interface to search for
:type name: str
:param network_name: network name to search for
:type network_name: str
:param refresh: Refresh interface objects with latest FIM info
:type refresh: bool
:return: the particular interface
:rtype: Interface
:raises Exception: if interface is not found
"""
interfaces = self.get_interfaces(refresh=refresh, output="dict")
if name is not None:
interface = interfaces.get(name)
if interface is not None:
return interface
elif network_name is not None:
for interface in interfaces.values():
if (
interface is not None
and interface.get_network() is not None
and interface.get_network().get_name() == network_name
):
return interface
raise ResourceNotFoundError(f"Interface not found: {name or network_name}")
[docs]
def get_slice(self) -> Optional[Slice]:
"""
Gets the fablib slice associated with this component's node.
:return: the slice this component is on
:rtype: Slice
"""
if self.get_node():
return self.node.get_slice()
else:
return None
[docs]
def get_node(self) -> Node:
"""
Gets the fablib node this component is associated with.
:return: the node this component is on
:rtype: Node
"""
return self.node
[docs]
def get_site(self) -> str:
"""
Gets the name of the site this component's node is on.
:return: the site name this node is on
:rtype: String
"""
return self.node.get_site()
[docs]
def get_short_name(self):
"""
Gets the short name of the component.
"""
# strip of the extra parts of the name added by fim
if not self.dict["short_name"]:
self.dict["short_name"] = self.get_name()[
len(f"{self.get_node().get_name()}-") :
]
return self.dict["short_name"]
[docs]
def get_details(self) -> str:
"""
Not intended for API use
"""
if self._cached_details is None:
try:
if self.get_fim():
self._cached_details = self.get_fim().details
self.dict["details"] = self._cached_details
except Exception:
self._cached_details = None
return self._cached_details
[docs]
def get_numa_node(self) -> str:
"""
Get the Numa Node assigned to the device
"""
if self._cached_numa_node is None:
try:
if self.fim_component:
label_allocations = self.fim_component.get_property(
pname="label_allocations"
)
if label_allocations:
if label_allocations.numa:
if isinstance(label_allocations.numa, str):
self._cached_numa_node = label_allocations.numa
self.dict["numa"] = self._cached_numa_node
elif isinstance(label_allocations.numa, list):
self._cached_numa_node = label_allocations.numa[0]
self.dict["numa"] = self._cached_numa_node
except Exception:
self._cached_numa_node = None
return self._cached_numa_node if self._cached_numa_node else ""
[docs]
def get_disk(self) -> int:
"""
Gets the amount of disk space on this component.
:return: this component's disk space
:rtype: int
"""
if self._cached_disk is None:
try:
if self.fim_component:
capacity_allocations = self.fim_component.get_property(
pname="capacity_allocations"
)
if capacity_allocations:
if capacity_allocations.disk:
self._cached_disk = capacity_allocations.disk
self.dict["disk"] = self._cached_disk
except Exception:
self._cached_disk = None
return self._cached_disk if self._cached_disk else 0
[docs]
def get_unit(self) -> int:
"""
Get unit count for this component.
:return: unit
:rtype: int
"""
if self._cached_unit is None:
try:
if self.fim_component:
capacity_allocations = self.fim_component.get_property(
pname="capacity_allocations"
)
if capacity_allocations:
if capacity_allocations.unit:
self._cached_unit = capacity_allocations.unit
except Exception:
self._cached_unit = None
return self._cached_unit if self._cached_unit else 0
[docs]
def get_pci_addr(self) -> str:
"""
Get the PIC device ID for this component.
:return: PCI device ID
:rtype: String
"""
if self._cached_bdf is None:
try:
if self.fim_component:
label_allocations = self.fim_component.get_property(
pname="label_allocations"
)
if label_allocations:
if label_allocations.bdf:
self._cached_bdf = label_allocations.bdf
except Exception:
self._cached_bdf = None
return self._cached_bdf
[docs]
def get_model(self) -> str:
"""
Get FABlib model name for this component.
:return: FABlib model name
:rtype: String
"""
fim_model = str(self.get_fim_model()).replace("-", "_").replace(" ", "")
component_type = self.get_type()
prefix_map = {
ComponentType.SmartNIC: "NIC_",
ComponentType.NVME: "NVME_",
ComponentType.GPU: "GPU_",
ComponentType.FPGA: "FPGA_",
ComponentType.Storage: "Storage_",
}
if component_type == ComponentType.SharedNIC:
return Constants.CMP_NIC_Basic
prefix = prefix_map.get(component_type)
if prefix:
return f"{prefix}{fim_model}"
else:
raise ValueError(f"Unsupported component type: {component_type}")
[docs]
def get_reservation_id(self) -> Optional[str]:
"""
Get reservation ID for this component.
:return: reservation ID
:rtype: String
"""
if self.get_node():
return self.get_node().get_reservation_id()
return None
[docs]
def get_reservation_state(self) -> Optional[str]:
"""
Get reservation state for this component.
:return: reservation state
:rtype: String
"""
if self.get_node():
return self.get_node().get_reservation_state()
return None
[docs]
def get_error_message(self) -> Optional[str]:
"""
Get error message for this component.
:return: reservation state
:rtype: String
"""
if self.get_node():
return self.get_node().get_error_message()
return None
[docs]
def get_fim_model(self) -> str:
"""
Not for API use
"""
if self._cached_fim_model is None:
try:
if self.fim_component:
self._cached_fim_model = self.fim_component.model
except Exception:
self._cached_fim_model = None
return self._cached_fim_model
[docs]
def get_type(self) -> str:
"""
Not for API use
Gets the type of this component.
:return: the type of component
:rtype: str
"""
if self._cached_fim_type is None:
try:
if self.fim_component:
self._cached_fim_type = self.fim_component.type
except Exception:
self._cached_fim_type = None
return self._cached_fim_type
[docs]
def get_device_name(self) -> str:
"""
Not for API use
"""
if self._cached_device_name is None:
try:
if self.fim_component:
label_allocations = self.fim_component.get_property(
pname="label_allocations"
)
if label_allocations:
if label_allocations.device_name:
self._cached_device_name = label_allocations.device_name
except Exception:
self._cached_device_name = None
return self._cached_device_name
[docs]
@staticmethod
def new_storage(node: Node, name: str, auto_mount: bool = False):
"""
Not intended for API use
Creates a new FIM component on the fablib node inputted.
:param node: the fablib node to build the component on
:param name: the name of the new component
:param auto_mount: True - mount the storage; False - do not mount
:return: the new fablib compoent
:rtype: Component
"""
# Hack to make it possile to find interfaces
fim_component = node.fim_node.add_storage(
name=name,
labels=Labels(local_name=name),
flags=Flags(auto_mount=auto_mount),
)
return Component(node=node, fim_component=fim_component)
[docs]
def get_fim(self):
"""
Gets the component's FABRIC Information Model (fim) object.
This method is used to access data at a lower level than
FABlib.
"""
return self.fim_component
[docs]
def get_fim_component(self) -> FimComponent:
"""
Not recommended for most users.
Gets the FABRIC component this fablib component represents. This method
is used to access data at a lower level than FABlib.
:return: the FABRIC component on this component
:rtype: FIMComponent
"""
return self.get_fim()
[docs]
def delete(self):
"""
Remove the component from the slice/node.
"""
if self.get_interfaces(refresh=True):
for interface in self.get_interfaces():
interface.delete()
self.get_slice().get_fim_topology().nodes[
self.get_node().get_name()
].remove_component(name=self.get_name())
# Invalidate parent node's component cache so subsequent
# get_components() calls don't return the deleted component
self.get_node().components = {}
self.get_node()._fim_dirty = True