[Bgpcfgd]: Add SRv6 support in Bgpcfgd (#21156)

Why I did it
There is a motivation to add capabilities in SONiC that allows static configuration of SRv6 network.

Work item tracking
Microsoft ADO (number only): 30251795

How I did it
I added a SRv6 manager in Bgpcfgd that subscribes to SRV6_MY_LOCATORS and SRV6_MY_SIDS in CONFIG_DB and programs the changes to FRR's configuration.

Note: this change depends on the availability and implementation details of the following FRR patch FRR SRv6 Static SID CLI

How to verify it
- Run unit tests
- Build an image that contains this change and the relevant FRR CLI support.
- Test the Image on a virtual device or physical device

Author: BYGX-wcr

src/sonic-bgpcfgd/bgpcfgd/main.py

@@ -22,6 +22,7 @@
 from .managers_rm import RouteMapMgr
 from .managers_device_global import DeviceGlobalCfgMgr
 from .managers_chassis_app_db import ChassisAppDbMgr
+from .managers_srv6 import SRv6Mgr
 from .static_rt_timer import StaticRouteTimer
 from .runner import Runner, signal_handler
 from .template import TemplateFabric
@@ -75,6 +76,9 @@ def do_work():
         RouteMapMgr(common_objs, "APPL_DB", swsscommon.APP_BGP_PROFILE_TABLE_NAME),
         # Device Global Manager
         DeviceGlobalCfgMgr(common_objs, "CONFIG_DB", swsscommon.CFG_BGP_DEVICE_GLOBAL_TABLE_NAME),
+        # SRv6 Manager
+        SRv6Mgr(common_objs, "CONFIG_DB", "SRV6_MY_SIDS"),
+        SRv6Mgr(common_objs, "CONFIG_DB", "SRV6_MY_LOCATORS")
     ]
 
     if device_info.is_chassis():

src/sonic-bgpcfgd/bgpcfgd/managers_srv6.py

@@ -0,0 +1,136 @@
+from .log import log_err, log_debug, log_warn
+from .manager import Manager
+from ipaddress import IPv6Address
+from swsscommon import swsscommon
+
+supported_SRv6_behaviors = {
+    'uN',
+    'uDT46',
+}
+
+DEFAULT_VRF = "default"
+SRV6_MY_SIDS_TABLE_NAME = "SRV6_MY_SIDS"
+
+class SRv6Mgr(Manager):
+    """ This class updates SRv6 configurations when SRV6_MY_SID_TABLE table is updated """
+    def __init__(self, common_objs, db, table):
+        """
+        Initialize the object
+        :param common_objs: common object dictionary
+        :param db: name of the db
+        :param table: name of the table in the db
+        """
+        super(SRv6Mgr, self).__init__(
+            common_objs,
+            [],
+            db,
+            table,
+        )
+
+    def set_handler(self, key, data):
+        if self.table_name == SRV6_MY_SIDS_TABLE_NAME:
+            return self.sids_set_handler(key, data)
+        else:
+            return self.locators_set_handler(key, data)
+
+    def locators_set_handler(self, key, data):
+        locator_name = key
+
+        locator = Locator(locator_name, data)
+        cmd_list = ["segment-routing", "srv6"]
+        cmd_list += ['locators',
+                     'locator {}'.format(locator_name),
+                     'prefix {} block-len {} node-len {} func-bits {}'.format(
+                     locator.prefix,
+                     locator.block_len, locator.node_len, locator.func_len),
+                     "behavior usid"
+        ]
+
+        self.cfg_mgr.push_list(cmd_list)
+        log_debug("{} SRv6 static configuration {}|{} is scheduled for updates. {}".format(self.db_name, self.table_name, key, str(cmd_list)))
+
+        self.directory.put(self.db_name, self.table_name, key, locator)
+        return True
+
+    def sids_set_handler(self, key, data):
+        locator_name = key.split("|")[0]
+        ip_addr = key.split("|")[1].lower()
+        key = "{}|{}".format(locator_name, ip_addr)
+
+        if not self.directory.path_exist(self.db_name, "SRV6_MY_LOCATORS", locator_name):
+            log_err("Found a SRv6 SID config entry with a locator that does not exist: {} | {}".format(key, data))
+            return False
+        
+        locator = self.directory.get(self.db_name, "SRV6_MY_LOCATORS", locator_name)
+
+        if 'action' not in data:
+            log_err("Found a SRv6 SID config entry that does not specify action: {} | {}".format(key, data))
+            return False
+        
+        if data['action'] not in supported_SRv6_behaviors:
+            log_err("Found a SRv6 SID config entry associated with unsupported action: {} | {}".format(key, data))
+            return False
+        
+        sid = SID(locator_name, ip_addr, data) # the information in data will be parsed into SID's attributes
+
+        cmd_list = ['segment-routing', 'srv6', 'static-sids']
+        sid_cmd = 'sid {}/{} locator {} behavior {}'.format(ip_addr, locator.block_len + locator.node_len + locator.func_len, locator_name, sid.action)
+        if sid.decap_vrf != DEFAULT_VRF:
+            sid_cmd += ' vrf {}'.format(sid.decap_vrf)
+        cmd_list.append(sid_cmd)
+
+        self.cfg_mgr.push_list(cmd_list)
+        log_debug("{} SRv6 static configuration {}|{} is scheduled for updates. {}".format(self.db_name, self.table_name, key, str(cmd_list)))
+
+        self.directory.put(self.db_name, self.table_name, key, (sid, sid_cmd))
+        return True
+
+    def del_handler(self, key):
+        if self.table_name == SRV6_MY_SIDS_TABLE_NAME:
+            self.sids_del_handler(key)
+        else:
+            self.locators_del_handler(key)
+
+    def locators_del_handler(self, key):
+        locator_name = key
+        cmd_list = ['segment-routing', 'srv6', 'locators', 'no locator {}'.format(locator_name)]
+
+        self.cfg_mgr.push_list(cmd_list)
+        log_debug("{} SRv6 static configuration {}|{} is scheduled for updates. {}".format(self.db_name, self.table_name, key, str(cmd_list)))
+        self.directory.remove(self.db_name, self.table_name, key)
+
+    def sids_del_handler(self, key):
+        locator_name = key.split("|")[0]
+        ip_addr = key.split("|")[1].lower()
+        key = "{}|{}".format(locator_name, ip_addr)
+
+        if not self.directory.path_exist(self.db_name, self.table_name, key):
+            log_warn("Encountered a config deletion with a SRv6 SID that does not exist: {}".format(key))
+            return
+
+        _, sid_cmd = self.directory.get(self.db_name, self.table_name, key)
+        cmd_list = ['segment-routing', 'srv6', "static-sids"]
+        no_sid_cmd = 'no ' + sid_cmd
+        cmd_list.append(no_sid_cmd)
+
+        self.cfg_mgr.push_list(cmd_list)
+        log_debug("{} SRv6 static configuration {}|{} is scheduled for updates. {}".format(self.db_name, self.table_name, key, str(cmd_list)))
+        self.directory.remove(self.db_name, self.table_name, key)
+
+class Locator:
+    def __init__(self, name, data):
+        self.name = name
+        self.block_len = int(data['block_len'] if 'block_len' in data else 32)
+        self.node_len = int(data['node_len'] if 'node_len' in data else 16)
+        self.func_len = int(data['func_len'] if 'func_len' in data else 16)
+        self.arg_len = int(data['arg_len'] if 'arg_len' in data else 0)
+        self.prefix = data['prefix'].lower() + "/{}".format(self.block_len + self.node_len)
+
+class SID:
+    def __init__(self, locator, ip_addr, data):
+        self.locator_name = locator
+        self.ip_addr = ip_addr
+        
+        self.action = data['action']
+        self.decap_vrf = data['decap_vrf'] if 'decap_vrf' in data else DEFAULT_VRF
+        self.adj = data['adj'].split(',') if 'adj' in data else []

src/sonic-bgpcfgd/tests/test_srv6.py

@@ -0,0 +1,163 @@
+from unittest.mock import MagicMock, patch
+
+from bgpcfgd.directory import Directory
+from bgpcfgd.template import TemplateFabric
+from bgpcfgd.managers_srv6 import SRv6Mgr
+
+def constructor():
+    cfg_mgr = MagicMock()
+
+    common_objs = {
+        'directory': Directory(),
+        'cfg_mgr':   cfg_mgr,
+        'tf':        TemplateFabric(),
+        'constants': {},
+    }
+
+    loc_mgr = SRv6Mgr(common_objs, "CONFIG_DB", "SRV6_MY_LOCATORS")
+    sid_mgr = SRv6Mgr(common_objs, "CONFIG_DB", "SRV6_MY_SIDS")
+
+    return loc_mgr, sid_mgr
+
+def op_test(mgr: SRv6Mgr, op, args, expected_ret, expected_cmds):
+    op_test.push_list_called = False
+    def push_list_checker(cmds):
+        op_test.push_list_called = True
+        assert len(cmds) == len(expected_cmds)
+        for i in range(len(expected_cmds)):
+            assert cmds[i].lower() == expected_cmds[i].lower()
+        return True
+    mgr.cfg_mgr.push_list = push_list_checker
+
+    if op == 'SET':
+        ret = mgr.set_handler(*args)
+        mgr.cfg_mgr.push_list = MagicMock()
+        assert expected_ret == ret
+    elif op == 'DEL':
+        mgr.del_handler(*args)
+        mgr.cfg_mgr.push_list = MagicMock()
+    else:
+        mgr.cfg_mgr.push_list = MagicMock()
+        assert False, "Unexpected operation {}".format(op)
+
+    if expected_ret and expected_cmds:
+        assert op_test.push_list_called, "cfg_mgr.push_list wasn't called"
+    else:
+        assert not op_test.push_list_called, "cfg_mgr.push_list was called"
+
+def test_locator_add():
+    loc_mgr, _ = constructor()
+
+    op_test(loc_mgr, 'SET', ("loc1", {
+        'prefix': 'fcbb:bbbb:1::'
+    }), expected_ret=True, expected_cmds=[
+        'segment-routing',
+        'srv6',
+        'locators',
+        'locator loc1',
+        'prefix fcbb:bbbb:1::/48 block-len 32 node-len 16 func-bits 16',
+        'behavior usid'
+    ])
+
+    assert loc_mgr.directory.path_exist(loc_mgr.db_name, loc_mgr.table_name, "loc1")
+
+def test_locator_del():
+    loc_mgr, _ = constructor()
+    loc_mgr.set_handler("loc1", {'prefix': 'fcbb:bbbb:1::'})
+
+    op_test(loc_mgr, 'DEL', ("loc1",), expected_ret=True, expected_cmds=[
+        'segment-routing',
+        'srv6',
+        'locators',
+        'no locator loc1'
+    ])
+
+    assert not loc_mgr.directory.path_exist(loc_mgr.db_name, loc_mgr.table_name, "loc1")
+
+def test_uN_add():
+    loc_mgr, sid_mgr = constructor()
+    assert loc_mgr.set_handler("loc1", {'prefix': 'fcbb:bbbb:1::'})
+
+    op_test(sid_mgr, 'SET', ("loc1|FCBB:BBBB:1:F1::", {
+        'action': 'uN'
+    }), expected_ret=True, expected_cmds=[
+        'segment-routing',
+        'srv6',
+        'static-sids',
+        'sid fcbb:bbbb:1:f1::/64 locator loc1 behavior uN'
+    ])
+
+    assert sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f1::")
+
+def test_uDT46_add_vrf1():
+    loc_mgr, sid_mgr = constructor()
+    assert loc_mgr.set_handler("loc1", {'prefix': 'fcbb:bbbb:1::'})
+
+    op_test(sid_mgr, 'SET', ("loc1|FCBB:BBBB:1:F2::", {
+        'action': 'uDT46',
+        'decap_vrf': 'Vrf1'
+    }), expected_ret=True, expected_cmds=[
+        'segment-routing',
+        'srv6',
+        'static-sids',
+        'sid fcbb:bbbb:1:f2::/64 locator loc1 behavior uDT46 vrf Vrf1'
+    ])
+
+    assert sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f2::")
+
+def test_uN_del():
+    loc_mgr, sid_mgr = constructor()
+    assert loc_mgr.set_handler("loc1", {'prefix': 'fcbb:bbbb:1::'})
+    
+    # add uN function first
+    assert sid_mgr.set_handler("loc1|FCBB:BBBB:1:F1::", {
+        'action': 'uN'
+    })
+
+    # test the deletion
+    op_test(sid_mgr, 'DEL', ("loc1|FCBB:BBBB:1:F1::",),
+            expected_ret=True, expected_cmds=[
+            'segment-routing',
+            'srv6',
+            'static-sids',
+            'no sid fcbb:bbbb:1:f1::/64 locator loc1 behavior uN'
+    ])
+
+    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f1::")
+
+def test_uDT46_del_vrf1():
+    loc_mgr, sid_mgr = constructor()
+    assert loc_mgr.set_handler("loc1", {'prefix': 'fcbb:bbbb:1::'})
+    
+    # add a uN action first to make the uDT46 action not the last function
+    assert sid_mgr.set_handler("loc1|FCBB:BBBB:1:F1::", {
+        'action': 'uN'
+    })
+
+    # add the uDT46 action
+    assert sid_mgr.set_handler("loc1|FCBB:BBBB:1:F2::", {
+        'action': 'uDT46',
+        "decap_vrf": "Vrf1"
+    })
+
+    # test the deletion of uDT46
+    op_test(sid_mgr, 'DEL', ("loc1|FCBB:BBBB:1:F2::",),
+            expected_ret=True, expected_cmds=[
+            'segment-routing',
+            'srv6',
+            'static-sids',
+            'no sid fcbb:bbbb:1:f2::/64 locator loc1 behavior uDT46 vrf Vrf1'
+    ])
+
+    assert sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f1::")
+    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f2::")
+
+def test_invalid_add():
+    _, sid_mgr = constructor()
+
+    # test the addition of a SID with a non-existent locator
+    op_test(sid_mgr, 'SET', ("loc2|FCBB:BBBB:21:F1::", {
+        'action': 'uN'
+    }), expected_ret=False, expected_cmds=[])
+
+    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc2|fcbb:bbbb:21:f1::")