Skip to content

Fibre Channel

Fibre Channel HBA and remote port management via sysfs (/sys/class/fc_host/, /sys/class/fc_remote_ports/).

sts.fc

Fibre Channel device management.

Handles FC/FCoE host discovery, WWN handling, remote port management, and transport type detection via sysfs.

FcDevice pydantic-model

Bases: ScsiDevice

Fibre Channel device representation.

Note

Construction performs no I/O. Call discover() to populate attributes that require system access (sysfs, subprocess calls).

Example
device = FcDevice(name='sda').discover()  # Discovers other values
device = FcDevice(wwn='10:00:5c:b9:01:c1:ec:71').discover()  # Discovers device from WWN
Show JSON schema:
{
  "additionalProperties": false,
  "description": "Fibre Channel device representation.\n\nNote:\n    Construction performs no I/O. Call `discover()` to populate\n    attributes that require system access (sysfs, subprocess calls).\n\nExample:\n    ```python\n    device = FcDevice(name='sda').discover()  # Discovers other values\n    device = FcDevice(wwn='10:00:5c:b9:01:c1:ec:71').discover()  # Discovers device from WWN\n    ```",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "scsi_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Scsi Id"
    },
    "host_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Host Id"
    },
    "wwn": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Wwn"
    },
    "transport_type": {
      "anyOf": [
        {
          "enum": [
            "FC",
            "FCoE"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Transport Type"
    }
  },
  "title": "FcDevice",
  "type": "object"
}

Fields:

  • name (str | None)
  • path (PathOrStr | None)
  • size (int | None)
  • model (str | None)
  • scsi_id (str | None)
  • host_id (str | None)
  • wwn (str | None)
  • transport_type (TransportType | None)

Validators:

  • _derive_fields
Source code in sts_libs/src/sts/fc.py
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
class FcDevice(ScsiDevice):
    """Fibre Channel device representation.

    Note:
        Construction performs no I/O. Call `discover()` to populate
        attributes that require system access (sysfs, subprocess calls).

    Example:
        ```python
        device = FcDevice(name='sda').discover()  # Discovers other values
        device = FcDevice(wwn='10:00:5c:b9:01:c1:ec:71').discover()  # Discovers device from WWN
        ```
    """

    # Optional parameters from parent classes
    name: str | None = None
    path: PathOrStr | None = None
    size: int | None = None
    model: str | None = None
    scsi_id: str | None = None  # SCSI address (H:C:T:L)
    host_id: str | None = None  # Local HBA ID

    # Optional parameters for this class
    wwn: str | None = None  # World Wide Name
    transport_type: TransportType | None = None  # FC or FCoE

    # Internal fields
    _remote_ports: list[str] = PrivateAttr(default_factory=list)

    # Sysfs paths for FC information
    HOST_PATH: ClassVar[Path] = Path('/sys/class/fc_host')
    REMOTE_PORT_PATH: ClassVar[Path] = Path('/sys/class/fc_remote_ports')

    def discover(self) -> Self:
        """Discover FC device attributes from the system.

        Populates fields that require I/O (lsscsi, sysfs) which
        the constructor no longer performs eagerly:
        1. SCSI attributes, including host_id, via ScsiDevice.discover()
        2. WWN from host information
        3. Transport type from host

        Raises:
            DeviceError: If discovered WWN format is invalid
        """
        super().discover()

        # Discover WWN from host information
        if not self.wwn and self.host_id:
            self.wwn = self.get_host_wwn(self.host_id)

        # Discover transport type from host
        if not self.transport_type and self.host_id:
            self.transport_type = self.get_host_transport_type(self.host_id)

        # Validate WWN format if provided
        if self.wwn and not self.is_valid_wwn(self.wwn):
            raise DeviceError(f'Invalid WWN format: {self.wwn}')

        return self

    @staticmethod
    def is_valid_wwn(wwn: str) -> bool:
        """Check if WWN is valid (8 colon-separated hex pairs)."""
        return bool(WWN_PATTERN.match(wwn.lower()))

    @staticmethod
    def standardize_wwn(wwn: str) -> str | None:
        """Standardize WWN to lowercase colon-separated format, or None if invalid.

        Handles ``0x`` prefixes, missing colons, and mixed case.

        Example:
            ```python
            FcDevice.standardize_wwn('500A0981894B8DC5')
            '50:0a:09:81:89:4b:8d:c5'
            ```
        """
        if not wwn:
            return None

        # Remove 0x and : characters
        wwn = wwn.lower().replace('0x', '').replace(':', '')

        # Add : every 2 characters
        wwn = ':'.join(wwn[i : i + 2] for i in range(0, len(wwn), 2))

        return wwn if FcDevice.is_valid_wwn(wwn) else None

    @property
    def remote_ports(self) -> list[str]:
        """FC remote ports connected to this host (format: ``rport-H:B-R``)."""
        if not self._remote_ports and self.host_id:
            result = run(f'ls {self.REMOTE_PORT_PATH} | grep rport-{self.host_id}')
            if result.succeeded:
                self._remote_ports = result.stdout.splitlines()
        return self._remote_ports

    def get_remote_port_wwn(self, port: str) -> str | None:
        """Get standardized WWN of a remote port from sysfs."""
        result = run(f'cat {self.REMOTE_PORT_PATH}/{port}/port_name')
        if result.failed:
            return None
        return self.standardize_wwn(result.stdout.strip())

    def get_remote_port_param(self, port: str, param: str) -> str | None:
        """Get remote port sysfs parameter (e.g. ``dev_loss_tmo``, ``port_state``)."""
        result = run(f'cat {self.REMOTE_PORT_PATH}/{port}/{param}')
        if result.failed:
            return None
        return result.stdout.strip()

    def set_remote_port_param(self, port: str, param: str, value: str) -> bool:
        """Set remote port sysfs parameter."""
        result = run(f'echo {value} > {self.REMOTE_PORT_PATH}/{port}/{param}')
        return result.succeeded

    @classmethod
    def get_hosts(cls) -> list[str]:
        """Get list of FC host IDs from sysfs."""
        result = run(f'ls {cls.HOST_PATH}')
        if result.failed:
            return []
        return [h.removeprefix('host') for h in result.stdout.splitlines()]

    @classmethod
    def get_host_wwn(cls, host_id: str) -> str | None:
        """Get standardized WWN for an FC host from sysfs."""
        result = run(f'cat {cls.HOST_PATH}/host{host_id}/port_name')
        if result.failed:
            return None
        return cls.standardize_wwn(result.stdout.strip())

    @classmethod
    def get_host_transport_type(cls, host_id: str) -> TransportType | None:
        """Determine FC vs FCoE via model name, symbolic name, or driver."""
        # Common model to transport type mapping
        model_map: dict[str, TransportType] = {
            'QLE2462': 'FC',  # QLogic 4Gb FC
            'QLE2772': 'FC',  # QLogic 32Gb FC
            'QLE8262': 'FCoE',  # QLogic 10Gb FCoE
            'QLE8362': 'FCoE',  # QLogic 10Gb FCoE
            'CN1000Q': 'FCoE',  # Cavium/QLogic FCoE
            'QLogic-1020': 'FCoE',  # QLogic FCoE
            '554FLR-SFP+': 'FCoE',  # HP FCoE
            'Intel 82599': 'FCoE',  # Intel FCoE
        }

        # Try to get model from sysfs
        result = run(f'cat {cls.HOST_PATH}/host{host_id}/model_name')
        if result.succeeded:
            model = result.stdout.strip()
            if model in model_map:
                return model_map[model]

        # Try to get from symbolic name
        result = run(f'cat {cls.HOST_PATH}/host{host_id}/symbolic_name')
        if result.succeeded:
            name = result.stdout.lower()
            if 'fibre channel' in name:
                return 'FC'
            if 'fcoe' in name:
                return 'FCoE'

        # Try to get from driver
        result = run(f'cat {cls.HOST_PATH}/host{host_id}/driver_name')
        if result.succeeded:
            driver = result.stdout.strip()
            if driver in {'bnx2fc', 'qedf'}:  # FCoE drivers
                return 'FCoE'

        return None

    @classmethod
    def get_all(cls) -> list[FcDevice]:
        """Discover all FC devices by enumerating hosts and their targets."""
        # Get all FC hosts
        hosts = cls.get_hosts()

        # Get all targets for each host
        all_targets: list[tuple[str, str]] = []
        for host_id in hosts:
            result = run(f'ls -1 /sys/class/fc_host/host{host_id}/device/target*')
            if result.succeeded:
                all_targets.extend((target, host_id) for target in result.stdout.splitlines())

        # Extract device names from target paths
        device_info: list[tuple[str, str]] = []
        for target, host_id in all_targets:
            name = Path(target).name
            if name:
                device_info.append((name, host_id))
            else:
                logger.warning(f'Invalid target path: {target}')

        # Create device objects with host information
        return [cls(name=name, host_id=host_id).discover() for name, host_id in device_info]

remote_ports property

FC remote ports connected to this host (format: rport-H:B-R).

discover()

Discover FC device attributes from the system.

Populates fields that require I/O (lsscsi, sysfs) which the constructor no longer performs eagerly: 1. SCSI attributes, including host_id, via ScsiDevice.discover() 2. WWN from host information 3. Transport type from host

Raises:

Type Description
DeviceError

If discovered WWN format is invalid

Source code in sts_libs/src/sts/fc.py
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def discover(self) -> Self:
    """Discover FC device attributes from the system.

    Populates fields that require I/O (lsscsi, sysfs) which
    the constructor no longer performs eagerly:
    1. SCSI attributes, including host_id, via ScsiDevice.discover()
    2. WWN from host information
    3. Transport type from host

    Raises:
        DeviceError: If discovered WWN format is invalid
    """
    super().discover()

    # Discover WWN from host information
    if not self.wwn and self.host_id:
        self.wwn = self.get_host_wwn(self.host_id)

    # Discover transport type from host
    if not self.transport_type and self.host_id:
        self.transport_type = self.get_host_transport_type(self.host_id)

    # Validate WWN format if provided
    if self.wwn and not self.is_valid_wwn(self.wwn):
        raise DeviceError(f'Invalid WWN format: {self.wwn}')

    return self

get_all() classmethod

Discover all FC devices by enumerating hosts and their targets.

Source code in sts_libs/src/sts/fc.py
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
@classmethod
def get_all(cls) -> list[FcDevice]:
    """Discover all FC devices by enumerating hosts and their targets."""
    # Get all FC hosts
    hosts = cls.get_hosts()

    # Get all targets for each host
    all_targets: list[tuple[str, str]] = []
    for host_id in hosts:
        result = run(f'ls -1 /sys/class/fc_host/host{host_id}/device/target*')
        if result.succeeded:
            all_targets.extend((target, host_id) for target in result.stdout.splitlines())

    # Extract device names from target paths
    device_info: list[tuple[str, str]] = []
    for target, host_id in all_targets:
        name = Path(target).name
        if name:
            device_info.append((name, host_id))
        else:
            logger.warning(f'Invalid target path: {target}')

    # Create device objects with host information
    return [cls(name=name, host_id=host_id).discover() for name, host_id in device_info]

get_host_transport_type(host_id) classmethod

Determine FC vs FCoE via model name, symbolic name, or driver.

Source code in sts_libs/src/sts/fc.py
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
@classmethod
def get_host_transport_type(cls, host_id: str) -> TransportType | None:
    """Determine FC vs FCoE via model name, symbolic name, or driver."""
    # Common model to transport type mapping
    model_map: dict[str, TransportType] = {
        'QLE2462': 'FC',  # QLogic 4Gb FC
        'QLE2772': 'FC',  # QLogic 32Gb FC
        'QLE8262': 'FCoE',  # QLogic 10Gb FCoE
        'QLE8362': 'FCoE',  # QLogic 10Gb FCoE
        'CN1000Q': 'FCoE',  # Cavium/QLogic FCoE
        'QLogic-1020': 'FCoE',  # QLogic FCoE
        '554FLR-SFP+': 'FCoE',  # HP FCoE
        'Intel 82599': 'FCoE',  # Intel FCoE
    }

    # Try to get model from sysfs
    result = run(f'cat {cls.HOST_PATH}/host{host_id}/model_name')
    if result.succeeded:
        model = result.stdout.strip()
        if model in model_map:
            return model_map[model]

    # Try to get from symbolic name
    result = run(f'cat {cls.HOST_PATH}/host{host_id}/symbolic_name')
    if result.succeeded:
        name = result.stdout.lower()
        if 'fibre channel' in name:
            return 'FC'
        if 'fcoe' in name:
            return 'FCoE'

    # Try to get from driver
    result = run(f'cat {cls.HOST_PATH}/host{host_id}/driver_name')
    if result.succeeded:
        driver = result.stdout.strip()
        if driver in {'bnx2fc', 'qedf'}:  # FCoE drivers
            return 'FCoE'

    return None

get_host_wwn(host_id) classmethod

Get standardized WWN for an FC host from sysfs.

Source code in sts_libs/src/sts/fc.py
163
164
165
166
167
168
169
@classmethod
def get_host_wwn(cls, host_id: str) -> str | None:
    """Get standardized WWN for an FC host from sysfs."""
    result = run(f'cat {cls.HOST_PATH}/host{host_id}/port_name')
    if result.failed:
        return None
    return cls.standardize_wwn(result.stdout.strip())

get_hosts() classmethod

Get list of FC host IDs from sysfs.

Source code in sts_libs/src/sts/fc.py
155
156
157
158
159
160
161
@classmethod
def get_hosts(cls) -> list[str]:
    """Get list of FC host IDs from sysfs."""
    result = run(f'ls {cls.HOST_PATH}')
    if result.failed:
        return []
    return [h.removeprefix('host') for h in result.stdout.splitlines()]

get_remote_port_param(port, param)

Get remote port sysfs parameter (e.g. dev_loss_tmo, port_state).

Source code in sts_libs/src/sts/fc.py
143
144
145
146
147
148
def get_remote_port_param(self, port: str, param: str) -> str | None:
    """Get remote port sysfs parameter (e.g. ``dev_loss_tmo``, ``port_state``)."""
    result = run(f'cat {self.REMOTE_PORT_PATH}/{port}/{param}')
    if result.failed:
        return None
    return result.stdout.strip()

get_remote_port_wwn(port)

Get standardized WWN of a remote port from sysfs.

Source code in sts_libs/src/sts/fc.py
136
137
138
139
140
141
def get_remote_port_wwn(self, port: str) -> str | None:
    """Get standardized WWN of a remote port from sysfs."""
    result = run(f'cat {self.REMOTE_PORT_PATH}/{port}/port_name')
    if result.failed:
        return None
    return self.standardize_wwn(result.stdout.strip())

is_valid_wwn(wwn) staticmethod

Check if WWN is valid (8 colon-separated hex pairs).

Source code in sts_libs/src/sts/fc.py
 99
100
101
102
@staticmethod
def is_valid_wwn(wwn: str) -> bool:
    """Check if WWN is valid (8 colon-separated hex pairs)."""
    return bool(WWN_PATTERN.match(wwn.lower()))

set_remote_port_param(port, param, value)

Set remote port sysfs parameter.

Source code in sts_libs/src/sts/fc.py
150
151
152
153
def set_remote_port_param(self, port: str, param: str, value: str) -> bool:
    """Set remote port sysfs parameter."""
    result = run(f'echo {value} > {self.REMOTE_PORT_PATH}/{port}/{param}')
    return result.succeeded

standardize_wwn(wwn) staticmethod

Standardize WWN to lowercase colon-separated format, or None if invalid.

Handles 0x prefixes, missing colons, and mixed case.

Example
FcDevice.standardize_wwn('500A0981894B8DC5')
'50:0a:09:81:89:4b:8d:c5'
Source code in sts_libs/src/sts/fc.py
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@staticmethod
def standardize_wwn(wwn: str) -> str | None:
    """Standardize WWN to lowercase colon-separated format, or None if invalid.

    Handles ``0x`` prefixes, missing colons, and mixed case.

    Example:
        ```python
        FcDevice.standardize_wwn('500A0981894B8DC5')
        '50:0a:09:81:89:4b:8d:c5'
        ```
    """
    if not wwn:
        return None

    # Remove 0x and : characters
    wwn = wwn.lower().replace('0x', '').replace(':', '')

    # Add : every 2 characters
    wwn = ':'.join(wwn[i : i + 2] for i in range(0, len(wwn), 2))

    return wwn if FcDevice.is_valid_wwn(wwn) else None