Browse Source

Update MAC typing

Remove dead function.  Make mac_t and mac_str_t strong types that
require casting to and from; propagate type checking outwards.

Change-Id: If5c82665bdea849d437677b79a08bc967b9f35dd
changes/54/752554/6
Ian Wells 1 year ago
parent
commit
618457ad76
  1. 4
      mypy.cfg
  2. 19
      networking_vpp/agent/gpe.py
  3. 11
      networking_vpp/agent/server.py
  4. 40
      networking_vpp/agent/vpp.py

4
mypy.cfg

@ -3,6 +3,10 @@ ignore_missing_imports = True
# Some modules have been written to stricter standards, and we don't want to regress.
# (In general we're building type strength out from vpp.py.)
[mypy-networking_vpp.agent.gpe]
check_untyped_defs = True
[mypy-networking_vpp.agent.vpp]
check_untyped_defs = True
disallow_untyped_calls = True

19
networking_vpp/agent/gpe.py

@ -18,6 +18,7 @@ import ipaddress
import re
import sys
from networking_vpp.agent.vpp import mac_str_t
from networking_vpp import constants as nvpp_const
from networking_vpp import etcdutils
@ -25,6 +26,8 @@ from oslo_config import cfg
from oslo_log import log as logging
from oslo_serialization import jsonutils
from typing import Dict
import neutron_lib.constants as n_const
LOG = logging.getLogger(__name__)
@ -65,7 +68,7 @@ class GPEForwarder(object):
# Will be set when we ensure GPE link
self.gpe_underlay_addr = None
# keeps track of gpe locators and mapping info
self.gpe_map = {'remote_map': {}}
self.gpe_map: Dict[str, dict] = {'remote_map': {}}
def ensure_gpe_link(self):
"""Ensures that the GPE uplink interface is present and configured.
@ -149,7 +152,9 @@ class GPEForwarder(object):
# Clear IPv6 NDP Entries
self.vpp.clear_lisp_ndp_entries(bridge_domain)
def ensure_remote_gpe_mapping(self, vni, mac, ip, remote_ip: str) -> None:
def ensure_remote_gpe_mapping(self, vni,
mac: mac_str_t, ip,
remote_ip: str) -> None:
"""Ensures a remote GPE mapping
A remote GPE mapping contains a remote mac-address of the instance,
@ -192,7 +197,7 @@ class GPEForwarder(object):
bridge_domain,
ip6)
def delete_remote_gpe_mapping(self, vni, mac, ip=None):
def delete_remote_gpe_mapping(self, vni, mac: mac_str_t, ip=None):
"""Delete a remote GPE vni to mac mapping."""
if (mac, vni) in self.gpe_map['remote_map']:
self.vpp.del_lisp_remote_mac(mac, vni)
@ -213,7 +218,7 @@ class GPEForwarder(object):
bridge_domain,
ip6)
def add_local_gpe_mapping(self, vni, mac):
def add_local_gpe_mapping(self, vni, mac: mac_str_t):
"""Add a local GPE mapping between a mac and vni."""
lset_mapping = self.gpe_map[gpe_lset_name]
# If a remote map exists, clear it as local map takes precedence
@ -229,7 +234,7 @@ class GPEForwarder(object):
self.vpp.add_lisp_local_mac(mac, vni, gpe_lset_name)
lset_mapping['local_map'][mac] = vni
def delete_local_gpe_mapping(self, vni, mac):
def delete_local_gpe_mapping(self, vni, mac: mac_str_t):
lset_mapping = self.gpe_map[gpe_lset_name]
if mac in lset_mapping['local_map']:
self.vpp.del_lisp_local_mac(mac, vni, gpe_lset_name)
@ -373,7 +378,7 @@ class GpeListener(object):
hostname = m.group(1)
ip = m.group(2)
data = jsonutils.loads(child.value)
mac = data["mac"]
mac = mac_str_t(data["mac"])
remote_ip = data["host"]
if self.is_valid_remote_map(vni, hostname):
self.gpe.ensure_remote_gpe_mapping(vni, mac, ip,
@ -405,7 +410,7 @@ class GpeListener(object):
data = router_ports[port]
vxlan_bound = data['net_type'] == n_const.TYPE_VXLAN
seg_id = data['segmentation_id']
mac_addr = data['mac_address']
mac_addr = mac_str_t(data['mac_address'])
ip_addr = data['gateway_ip']
# Master --> Backup state transition
# Delete local GPE mapping as we no longer own this mac-address

11
networking_vpp/agent/server.py

@ -1703,16 +1703,9 @@ class VPPForwarder(object):
return sub_if
def _get_loopback_mac(self, loopback_idx):
"""Returns the mac address of the loopback interface."""
loopback_mac = self.vpp.get_ifidx_mac_address(loopback_idx)
LOG.debug("mac address %s of the router BVI loopback idx: %s",
loopback_mac, loopback_idx)
return loopback_mac
def ensure_bridge_bvi(self,
bridge_idx: vpp.br_idx_t,
mac_address: str = None) -> vpp.if_idx_t:
mac_address: vpp.mac_str_t = None) -> vpp.if_idx_t:
"""Ensure a BVI loopback interface for the bridge."""
bvi_if_idx = self.vpp.get_bridge_bvi(bridge_idx)
if not bvi_if_idx:
@ -1773,7 +1766,7 @@ class VPPForwarder(object):
# Ensure a BVI (i.e. A loopback) for the bridge domain
loopback_idx = self.vpp.get_bridge_bvi(bridge_idx)
# Create a loopback BVI interface
loopback_mac = router_data['loopback_mac']
loopback_mac = vpp.mac_str_t(router_data['loopback_mac'])
if loopback_idx is None:
# Create the loopback interface, but don't bring it UP yet
loopback_idx = self.ensure_bridge_bvi(bridge_idx, loopback_mac)

40
networking_vpp/agent/vpp.py

@ -47,7 +47,8 @@ macip_acl_idx_t = NewType('macip_acl_idx_t', int)
macip_acl_rule_t = dict # an acl rule definition
ip_addr_t = bytes
ip_addr_str_t = str
mac_t = str
mac_t = NewType('mac_t', bytes)
mac_str_t = NewType('mac_str_t', str)
vni_t = int
lisp_dp_table_t = int
lisp_ls_idx_t = NewType('lisp_ls_idx_t', int)
@ -59,12 +60,15 @@ route_path_t = dict
DEFAULT_VRF = vrf_idx_t(0)
def mac_to_bytes(mac: str) -> bytes:
def mac_to_bytes(mac: mac_str_t) -> mac_t:
# py3 note:
# TODO(onong): PAPI has introduced a new macaddress object which seemingly
# takes care of conversion to/from MAC addr to string.
# TODO(onong): move to common file in phase 2
return bytes.fromhex(mac.replace(':', ''))
return mac_t(bytes.fromhex(mac.replace(':', '')))
DEFAULT_MAC = mac_to_bytes(mac_str_t('00:00:00:00:00:00'))
def fix_string(s: bytes) -> str:
@ -318,11 +322,11 @@ class VPPInterface(object):
return iface['sw_if_idx']
return None
def get_ifidx_mac_address(self, ifidx: if_idx_t) -> Optional[bytes]:
def get_ifidx_mac_address(self, ifidx: if_idx_t) -> Optional[mac_t]:
for iface in self.get_interfaces():
if iface['sw_if_idx'] == ifidx:
return iface['mac']
return mac_t(iface['mac'].mac_binary)
return None
def get_ifidx_by_tag(self, tag: str) -> Optional[if_idx_t]:
@ -353,13 +357,13 @@ class VPPInterface(object):
########################################
def create_tap(self, ifname: str, mac: Optional[str] = None,
def create_tap(self, ifname: str, mac: Optional[mac_str_t] = None,
tag: str = "") -> if_idx_t:
if mac is not None:
mac_bytes = mac_to_bytes(mac)
use_random_mac = False
else:
mac_bytes = mac_to_bytes('00:00:00:00:00:00')
mac_bytes = DEFAULT_MAC
use_random_mac = True
# Note(onong): In VPP 20.01, the following API changes have happened:
@ -722,7 +726,7 @@ class VPPInterface(object):
########################################
def create_loopback(self, mac_address_in: Optional[str] = None) \
def create_loopback(self, mac_address_in: Optional[mac_str_t] = None) \
-> if_idx_t:
# Create a loopback interface to act as a BVI
if mac_address_in is not None:
@ -1362,7 +1366,7 @@ class VPPInterface(object):
dp_table=bridge_domain,
is_l2=True)
def add_lisp_local_mac(self, mac: mac_t, vni: vni_t,
def add_lisp_local_mac(self, mac: mac_str_t, vni: vni_t,
locator_set_name: str) -> None:
"""Add a local mac address to VNI association in LISP"""
# Note(onong): In 20.05, eid and eid_type are subsumed in a new type,
@ -1375,7 +1379,7 @@ class VPPInterface(object):
locator_set_name=locator_set_name,
vni=vni)
def del_lisp_local_mac(self, mac: mac_t, vni: vni_t,
def del_lisp_local_mac(self, mac: mac_str_t, vni: vni_t,
locator_set_name: str) -> None:
"""Delete a local mac address to VNI association in LISP"""
# Note(onong): In 20.05, eid and eid_type are subsumed in a new type,
@ -1388,7 +1392,7 @@ class VPPInterface(object):
locator_set_name=locator_set_name,
vni=vni)
def add_lisp_remote_mac(self, mac: mac_t, vni: vni_t,
def add_lisp_remote_mac(self, mac: mac_str_t, vni: vni_t,
remote_ip: IPAddress) -> None:
"""Add a LISP entry for a remote mac address to the underlay IP.
@ -1420,7 +1424,7 @@ class VPPInterface(object):
rloc_num=1,
is_src_dst=False)
def del_lisp_remote_mac(self, mac: mac_t, vni: vni_t) -> None:
def del_lisp_remote_mac(self, mac: mac_str_t, vni: vni_t) -> None:
"""Delete a LISP entry for a remote mac address.
Deletes all underlay IPs along with the eid.
@ -1479,7 +1483,7 @@ class VPPInterface(object):
locator_set_name=locator_set_name,
sw_if_index=sw_if_index)
def add_lisp_arp_entry(self, mac: mac_t, bridge_domain: br_idx_t,
def add_lisp_arp_entry(self, mac: mac_str_t, bridge_domain: br_idx_t,
ipv4_address: ip_addr_t) -> None:
"""Adds a static ARP entry to LISP.
@ -1494,7 +1498,7 @@ class VPPInterface(object):
bd=bridge_domain
)
def add_lisp_ndp_entry(self, mac: mac_t, bridge_domain: br_idx_t,
def add_lisp_ndp_entry(self, mac: mac_str_t, bridge_domain: br_idx_t,
ipv6_address: ip_addr_t) -> None:
"""Adds a static IPv6 NDP entry to LISP.
@ -1509,7 +1513,7 @@ class VPPInterface(object):
bd=bridge_domain
)
def del_lisp_arp_entry(self, mac: mac_t, bridge_domain: br_idx_t,
def del_lisp_arp_entry(self, mac: mac_str_t, bridge_domain: br_idx_t,
ipv4_address: ip_addr_t) -> None:
"""Removes a static ARP entry from LISP.
@ -1524,7 +1528,7 @@ class VPPInterface(object):
bd=bridge_domain
)
def del_lisp_ndp_entry(self, mac: mac_t, bridge_domain: br_idx_t,
def del_lisp_ndp_entry(self, mac: mac_str_t, bridge_domain: br_idx_t,
ipv6_address: ip_addr_t) -> None:
"""Removes a static IPv6 NDP entry from LISP.
@ -1539,7 +1543,7 @@ class VPPInterface(object):
bd=bridge_domain
)
def replace_lisp_arp_entry(self, mac: mac_t, bridge_domain: br_idx_t,
def replace_lisp_arp_entry(self, mac: mac_str_t, bridge_domain: br_idx_t,
ipv4_address: bytes) -> None:
"""Replaces the LISP ARP entry in a bridge domain for the IP address.
@ -1558,7 +1562,7 @@ class VPPInterface(object):
# Add the new ARP entry
self.add_lisp_arp_entry(mac, bridge_domain, ipv4_address)
def replace_lisp_ndp_entry(self, mac: mac_t, bridge_domain: br_idx_t,
def replace_lisp_ndp_entry(self, mac: mac_str_t, bridge_domain: br_idx_t,
ipv6_address: ip_addr_t) -> None:
"""Replaces the LISP NDP entry in a bridge domain for the v6 address.

Loading…
Cancel
Save