Add SRv6 support in Bgpcfgd

<!--
     Please make sure you've read and understood our contributing guidelines:
     https://github.com/Azure/SONiC/blob/gh-pages/CONTRIBUTING.md

     ** Make sure all your commits include a signature generated with `git commit -s` **

     If this is a bug fix, make sure your description includes "fixes #xxxx", or
     "closes #xxxx" or "resolves #xxxx"

     Please provide the following information:
-->

#### 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](sonic-net/sonic-buildimage#21380)

#### 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

<!--
If PR needs to be backported, then the PR must be tested against the base branch and the earliest backport release branch and provide tested image version on these two branches. For example, if the PR is requested for master, 202211 and 202012, then the requester needs to provide test results on master and 202012.
-->

#### Which release branch to backport (provide reason below if selected)

<!--
- Note we only backport fixes to a release branch, *not* features!
- Please also provide a reason for the backporting below.
- e.g.
- [x] 202006
-->

- [ ] 201811
- [ ] 201911
- [ ] 202006
- [ ] 202012
- [ ] 202106
- [ ] 202111
- [ ] 202205
- [ ] 202211
- [ ] 202305

#### Tested branch (Please provide the tested image version)

<!--
- Please provide tested image version
- e.g.
- [x] 20201231.100
-->

- [ ] <!-- image version 1 -->
- [ ] <!-- image version 2 -->

#### Description for the changelog
<!--
Write a short (one line) summary that describes the changes in this
pull request for inclusion in the changelog:
-->

<!--
 Ensure to add label/tag for the feature raised. example - PR#2174 under sonic-utilities repo. where, Generic Config and Update feature has been labelled as GCU.
-->

#### Link to config_db schema for YANG module changes
<!--
Provide a link to config_db schema for the table for which YANG model
is defined
Link should point to correct section on https://github.com/Azure/sonic-buildimage/blob/master/src/sonic-yang-models/doc/Configuration.md
-->

[SRv6 static config HLD](sonic-net/SONiC#1860)

#### A picture of a cute animal (not mandatory but encouraged)

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::")