#!/usr/bin/env python3 # # Copyright VyOS maintainers and contributors # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 or later as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import sys import json import typing from tabulate import tabulate from vyos.vpp import VPPControl from vyos.vpp.utils import vpp_iface_name_transform from vyos.configquery import ConfigTreeQuery import vyos.opmode NO_INDEX = 0xFFFFFFFF class VPPShow: RX_STATES = { 0: 'INITIALIZE', 1: 'PORT_DISABLED', 2: 'EXPIRED', 3: 'LACP_DISABLED', 4: 'DEFAULTED', 5: 'CURRENT', } TX_STATES = {0: 'TRANSMIT'} MUX_STATES = { 0: 'DETACHED', 1: 'WAITING', 2: 'ATTACHED', 3: 'COLLECTING_DISTRIBUTING', } PTX_STATES = {0: 'NO_PERIODIC', 1: 'FAST', 2: 'SLOW', 3: 'PERIODIC_TX'} BOND_MODE = { 1: 'round-robin', 2: 'active-backup', 3: 'xor', 4: 'broadcast', 5: 'lacp', } BOND_LB = { 0: 'layer2', 1: 'layer3+4', 2: 'layer2+3', 3: 'round-robin', 4: 'broadcast', 5: 'active-backup', } def __init__(self): self.config = ConfigTreeQuery() self.vpp = VPPControl() # ----------------------------- # IPFIX Interfaces # ----------------------------- def _get_ipfix_interfaces_raw(self) -> typing.List[dict]: interfaces = self.vpp.api.flowprobe_interface_dump() index_map = { i.sw_if_index: i.interface_name for i in self.vpp.api.sw_interface_dump() } return [ { 'interface': index_map.get(e.sw_if_index, f'if{e.sw_if_index}'), 'sw_if_index': e.sw_if_index, 'which': e.which.name.replace('FLOWPROBE_WHICH_', '').lower(), 'direction': e.direction.name.replace( 'FLOWPROBE_DIRECTION_', '' ).lower(), } for e in interfaces ] def _show_ipfix_interfaces_formatted(self, data: typing.List[dict]) -> str: if not data: return 'No flowprobe interfaces configured.' table_data = [ { 'Interface': d['interface'], 'VppIfIndex': d['sw_if_index'], 'Flow-variant': d['which'], 'Direction': d['direction'], } for d in data ] return tabulate(table_data, headers='keys', tablefmt='simple') def ipfix_interfaces(self, raw: bool): base = ['vpp', 'ipfix', 'interface'] if not self.config.exists(base): raise vyos.opmode.UnconfiguredSubsystem( 'vpp ipfix interface is not configured' ) data = self._get_ipfix_interfaces_raw() return data if raw else self._show_ipfix_interfaces_formatted(data) # ----------------------------- # IPFIX Collectors # ----------------------------- def _get_ipfix_collectors_raw(self) -> typing.List[dict]: _, collectors = self.vpp.api.ipfix_all_exporter_get() return [ { 'collector_address': str(c.collector_address), 'collector_port': c.collector_port, 'src_address': str(c.src_address), 'vrf_id': c.vrf_id, 'path_mtu': c.path_mtu, 'template_interval': c.template_interval, 'udp_checksum': bool(c.udp_checksum), } for c in collectors ] def _show_ipfix_collectors_formatted(self, data: typing.List[dict]) -> str: if not data: return 'No IPFIX collectors configured.' table_data = [ { 'Collector': f"{d['collector_address']}:{d['collector_port']}", 'Source': d['src_address'], 'VRF': d['vrf_id'], 'MTU': d['path_mtu'], 'Template Intvl': d['template_interval'], 'UDP Cksum': 'on' if d['udp_checksum'] else 'off', } for d in data ] return tabulate(table_data, headers='keys', tablefmt='simple') def ipfix_collectors(self, raw: bool): base = ['vpp', 'ipfix', 'collector'] if not self.config.exists(base): raise vyos.opmode.UnconfiguredSubsystem( 'vpp ipfix collector is not configured' ) data = self._get_ipfix_collectors_raw() return data if raw else self._show_ipfix_collectors_formatted(data) # ----------------------------- # IPFIX table # ----------------------------- def _get_ipfix_table_raw(self): # VPP does not have API call to get this data data = self.vpp.cli_cmd('show flowprobe table') return [data.reply] def _show_ipfix_table_formatted(self) -> str: data = self.vpp.cli_cmd('show flowprobe table') return data.reply def ipfix_table(self, raw: bool): base = ['vpp', 'ipfix', 'collector'] if not self.config.exists(base): raise vyos.opmode.UnconfiguredSubsystem( 'vpp ipfix collector is not configured' ) data = self._get_ipfix_table_raw() return data if raw else self._show_ipfix_table_formatted() # ----------------------------- # Bonding information # ----------------------------- def _get_raw_output(self, data_dump: typing.List[dict]) -> list[dict]: data = [json.loads(json.dumps(d._asdict(), default=str)) for d in data_dump] return data def _get_lacp_raw(self, ifname: typing.Optional[str]) -> list[dict]: lacp_dump = self.vpp.api.sw_interface_lacp_dump() data = self._get_raw_output(lacp_dump) if ifname: res = next((d for d in data if d['interface_name'] == ifname), None) if not res: raise vyos.opmode.IncorrectValue( f'Interface {ifname} is not a member of any LACP bond' ) data = [res] return data def _get_lacp_info_formatted(self, data): def bit(x, n): return (x >> n) & 1 def bits_to_str(x): return ' '.join(f'{bit(x, n):3d}' for n in range(7, -1, -1)) # Headers (exactly like VPP) print(f'{"":55} {"actor state":32} {"partner state":32}') print( 'interface name'.ljust(26) + 'sw_if_index'.ljust(13) + 'bond interface'.ljust(17) + 'exp/def/dis/col/syn/agg/tim/act'.ljust(33) + 'exp/def/dis/col/syn/agg/tim/act'.ljust(32) ) for d in data: iface = d['interface_name'] sw_if = str(d['sw_if_index']) bond_if = d['bond_interface_name'] actor_bits = bits_to_str(d['actor_state']) partner_bits = bits_to_str(d['partner_state']) print( f'{iface:25} {sw_if:12} {bond_if:16} {actor_bits:32} {partner_bits:32}' ) # LAG ID formatting lag_line = ( f' LAG ID: ' f'[({d["actor_system_priority"]:04x},{d["actor_system"].replace(":", "-")},' f'{d["actor_key"]:04x},{d["actor_port_priority"]:04x},{d["actor_port_number"]:04x}), ' f'({d["partner_system_priority"]:04x},{d["partner_system"].replace(":", "-")},' f'{d["partner_key"]:04x},{d["partner_port_priority"]:04x},{d["partner_port_number"]:04x})]' ) print(lag_line) # State machine line print( f' RX-state: {self.RX_STATES[d["rx_state"]]}, ' f'TX-state: {self.TX_STATES[d["tx_state"]]}, ' f'MUX-state: {self.MUX_STATES[d["mux_state"]]}, ' f'PTX-state: {self.PTX_STATES[d["ptx_state"]]}' ) def _get_bond_raw(self, index: typing.Optional[str]) -> list[dict]: bond_dump = self.vpp.api.sw_bond_interface_dump(sw_if_index=index) result = [] for bond in bond_dump: bond_info = { 'interface_name': bond.interface_name, 'sw_if_index': bond.sw_if_index, 'mode': self.BOND_MODE[bond.mode], 'hash_policy': self.BOND_LB[bond.lb], 'active_members': bond.active_members, 'members': {}, } members = self.vpp.api.sw_member_interface_dump( sw_if_index=bond.sw_if_index ) for member in members: bond_info['members'][member.interface_name] = { 'sw_if_index': member.sw_if_index, 'is_passive': member.is_passive, 'is_long_timeout': member.is_long_timeout, 'is_local_numa': member.is_local_numa, 'weight': member.weight, } result.append(bond_info) return result def _show_bond_info_formatted(self, data: typing.List[dict]) -> str: table_data = [ { 'Interface': d['interface_name'], 'Mode': d['mode'], 'Hash': d['hash_policy'], 'Members': '\n'.join(sorted(d['members'].keys())), 'Active members': d['active_members'], } for d in data ] return tabulate(table_data, headers='keys', tablefmt='simple', numalign='left') def lacp_info(self, raw: bool, ifname: typing.Optional[str]): data = self._get_lacp_raw(ifname) if not data: raise vyos.opmode.DataUnavailable( 'No VPP interface is configured with LACP (802.3ad) mode' ) if raw: return data return self._get_lacp_info_formatted(data) def lacp_details(self, raw: bool, ifname: typing.Optional[str]) -> str: # Check if interface is a part of any LACP bond self._get_lacp_raw(ifname) # VPP does not have API call to get this data cmd_command = f'show lacp{f" {ifname}" if ifname else ""} details' data = self.vpp.cli_cmd(cmd_command) if raw: return [data.reply] return data.reply def bond_info(self, raw: bool, ifname: typing.Optional[str]) -> str: index = NO_INDEX if ifname: if not ifname.startswith('vppbond') or not ifname[7:].isdigit(): raise vyos.opmode.IncorrectValue( f'"{ifname}" is not a valid bonding interface name (expected vppbondN)' ) ifname_vpp = vpp_iface_name_transform(ifname) index = self.vpp.get_sw_if_index(ifname_vpp) if index is None: raise vyos.opmode.IncorrectValue( f'Bonding interface {ifname} does not exist in VPP' ) data = self._get_bond_raw(index) return data if raw else self._show_bond_info_formatted(data) def bond_details(self, raw: bool) -> str: # VPP API call is not so informative -> use CLI command cmd_command = 'show bond details' data = self.vpp.cli_cmd(cmd_command) return [data.reply] if raw else data.reply # ----------------------------- # Bridge-domain information # ----------------------------- def _parse_bridge_id(self, ifname: typing.Optional[str]) -> typing.Optional[int]: if ifname is None: return None if not ifname.startswith('vppbr') and not ifname[5:].isdigit(): raise vyos.opmode.IncorrectValue( f'"{ifname}" is not a valid bridge interface name (expected vppbrN)' ) if not self.config.exists(['interfaces', 'vpp', 'bridge', ifname]): raise vyos.opmode.IncorrectValue( f'Bridge interface {ifname} does not exist' ) return int(ifname[5:]) def _get_bridge_domain_raw( self, bd_id: typing.Optional[int] = None ) -> typing.List[dict]: # Dump bridge domains domains = self.vpp.api.bridge_domain_dump( bd_id=bd_id if bd_id is not None else NO_INDEX ) result = [] for d in domains: domain_info = { 'bd_id': d.bd_id, 'learning': bool(d.learn), 'forward': bool(d.forward), 'uu_flood': bool(d.uu_flood), 'flood': bool(d.flood), 'arp_term': bool(d.arp_term), 'arp_ufwd': bool(d.arp_ufwd), 'mac_age': d.mac_age, 'bvi_interface': d.bvi_sw_if_index, 'n_sw_ifs': d.n_sw_ifs, 'members': [ { 'ifname': self.vpp.get_interface_name(m.sw_if_index), 'sw_if_index': m.sw_if_index, 'shg': m.shg, } for m in d.sw_if_details ], } result.append(domain_info) result.sort(key=lambda x: x['bd_id']) return result def _show_bridge_domain_formatted(self, data: typing.List[dict]) -> str: if not data: return 'No bridge domains configured.' table_data = [ { 'BD-ID': d['bd_id'], 'Age(min)': 'off' if d['mac_age'] == 0 else d['mac_age'], 'Learning': 'on' if d['learning'] else 'off', 'U-Forwrd': 'on' if d['forward'] else 'off', 'UU-Flood': 'flood' if d['uu_flood'] else 'drop', 'Flooding': 'on' if d['flood'] else 'off', 'ARP-Term': 'on' if d['arp_term'] else 'off', 'arp-ufwd': 'on' if d['arp_ufwd'] else 'off', 'BVI-Intf': ( self.vpp.get_interface_name(d['bvi_interface']) if d['bvi_interface'] != NO_INDEX else 'N/A' ), } for d in data ] return tabulate(table_data, headers='keys', tablefmt='simple', numalign='left') def bridge_domain(self, raw: bool, ifname: typing.Optional[str] = None): bd_id = self._parse_bridge_id(ifname) data = self._get_bridge_domain_raw(bd_id) return data if raw else self._show_bridge_domain_formatted(data) def bridge_domain_details(self, raw: bool, ifname: typing.List): bd_id = self._parse_bridge_id(ifname) # VPP API call is not so informative -> use CLI command cmd_command = f'show bridge-domain {bd_id} detail' data = self.vpp.cli_cmd(cmd_command) if raw: return [data.reply] return data.reply # ----------------------------- # Runtime table # ----------------------------- def _get_runtime_raw(self): # VPP does not have API call to get this data data = self.vpp.cli_cmd('show runtime') return [data.reply] def _show_runtime_formatted(self) -> str: data = self.vpp.cli_cmd('show runtime') return data.reply def runtime(self, raw: bool): data = self._get_runtime_raw() return data if raw else self._show_runtime_formatted() # ----------------------------- # Interfaces mode # ----------------------------- def mode(self, raw: bool): # VPP does not have API call to get this data data = self.vpp.cli_cmd('show mode') return [data.reply] if raw else data.reply # ----------------------------- # VyOS IPFIX op-mode entries # ----------------------------- @vyos.opmode.verify_cli_exists(['vpp', 'ipfix', 'interface']) def show_ipfix_interfaces(raw: bool): return VPPShow().ipfix_interfaces(raw) @vyos.opmode.verify_cli_exists(['vpp', 'ipfix', 'collector']) def show_ipfix_collectors(raw: bool): return VPPShow().ipfix_collectors(raw) @vyos.opmode.verify_cli_exists(['vpp', 'ipfix']) def show_ipfix_table(raw: bool): return VPPShow().ipfix_table(raw) # ----------------------------- # VPP Bonding information # ----------------------------- @vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bonding']) def show_lacp(raw: bool, ifname: typing.Optional[str]): return VPPShow().lacp_info(raw, ifname) @vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bonding']) def show_lacp_details(raw: bool, ifname: typing.Optional[str]): return VPPShow().lacp_details(raw, ifname) @vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bonding']) def show_bond(raw: bool, ifname: typing.Optional[str]): return VPPShow().bond_info(raw, ifname) @vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bonding']) def show_bond_details(raw: bool): return VPPShow().bond_details(raw) # ----------------------------- # Bridge op-mode entry # ----------------------------- @vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bridge']) def show_bridge(raw: bool, ifname: typing.Optional[str] = None): return VPPShow().bridge_domain(raw, ifname) @vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bridge']) def show_bridge_details(raw: bool, ifname: typing.Optional[str] = None): return VPPShow().bridge_domain_details(raw, ifname) # ----------------------------- # show runtime # ----------------------------- @vyos.opmode.verify_cli_exists(['vpp']) def show_runtime(raw: bool): return VPPShow().runtime(raw) # ----------------------------- # show mode # ----------------------------- @vyos.opmode.verify_cli_exists(['vpp']) def show_mode(raw: bool): return VPPShow().mode(raw) if __name__ == '__main__': try: res = vyos.opmode.run(sys.modules[__name__]) if res: print(res) except (ValueError, vyos.opmode.Error) as e: print(e) sys.exit(1)