[config, show] Add CLI commands for Path Tracing Midpoint

config/main.py

@@ -111,6 +111,9 @@
 
 DOM_CONFIG_SUPPORTED_SUBPORTS = ['0', '1']
 
+SWITCH_CAPABILITY = "SWITCH_CAPABILITY|switch"
+SWITCH_CAPABILITY_TABLE_PATH_TRACING_CAPABLE = "PATH_TRACING_CAPABLE"
+
 asic_type = None
 
 DSCP_RANGE = click.IntRange(min=0, max=63)
@@ -4828,6 +4831,93 @@ def loopback_action(ctx, interface_name, action):
     table_name = get_interface_table_name(interface_name)
     config_db.mod_entry(table_name, interface_name, {"loopback_action": action})
 
+
+def is_path_tracing_supported(ctx):
+    if ctx.obj['namespace'] is DEFAULT_NAMESPACE:
+        state_db = SonicV2Connector(host='127.0.0.1')
+    else:
+        state_db = SonicV2Connector(use_unix_socket_path=True, namespace=ctx.obj['namespace'])
+    state_db.connect(state_db.STATE_DB, False)
+    supported = state_db.get(state_db.STATE_DB, SWITCH_CAPABILITY, SWITCH_CAPABILITY_TABLE_PATH_TRACING_CAPABLE)
+    return supported
+
+
+#
+# 'path-tracing' subgroup ('config interface path-tracing ...')
+#
+@interface.group(cls=clicommon.AbbreviationGroup, name='path-tracing')
+@click.pass_context
+def path_tracing(ctx):
+    """Set Path Tracing attributes"""
+    pass
+
+
+#
+# 'add' subcommand
+#
+@path_tracing.command('add')
+@click.argument('interface_name', metavar='<interface_name>', required=True)
+@click.option(
+    '--interface-id', metavar='<interface_id>', required=True,
+    type=click.IntRange(1, 4095), help='Path Tracing Interface ID'
+)
+@click.option(
+    '--ts-template', metavar='<ts_template>', required=False,
+    type=click.Choice(['template1', 'template2', 'template3', 'template4']),
+    default='template3', help='Path Tracing Timestamp Template'
+)
+@click.option('-v', '--verbose', is_flag=True, help="Enable verbose output")
+@click.pass_context
+def add_path_tracing(ctx, interface_name, interface_id, ts_template, verbose):
+    """Set Path Tracing parameters"""
+    # Get the config_db connector
+    config_db = ctx.obj['config_db']
+
+    if clicommon.get_interface_naming_mode() == "alias":
+        interface_name = interface_alias_to_name(config_db, interface_name)
+        if interface_name is None:
+            ctx.fail("'interface_name' is None!")
+
+    log.log_info(
+        "'interface path-tracing add {} {} {}' executing...".format(
+            interface_name, interface_id, ts_template)
+    )
+
+    if not is_path_tracing_supported(ctx):
+        raise Exception("Path Tracing is not supported yet.")
+
+    config_db.mod_entry('PORT', interface_name, {'pt_interface_id': interface_id})
+    config_db.mod_entry('PORT', interface_name, {'pt_timestamp_template': ts_template})
+
+
+#
+# 'del' subcommand
+#
+@path_tracing.command('del')
+@click.argument('interface_name', metavar='<interface_name>', required=True)
+@click.option('-v', '--verbose', is_flag=True, help="Enable verbose output")
+@click.pass_context
+def del_path_tracing(ctx, interface_name, verbose):
+    """Delete Path Tracing parameters"""
+    # Get the config_db connector
+    config_db = ctx.obj['config_db']
+
+    if clicommon.get_interface_naming_mode() == "alias":
+        interface_name = interface_alias_to_name(config_db, interface_name)
+        if interface_name is None:
+            ctx.fail("'interface_name' is None!")
+
+    log.log_info(
+        "'interface path-tracing del {}' executing...".format(interface_name)
+    )
+
+    if not is_path_tracing_supported(ctx):
+        raise Exception("Path Tracing is not supported yet.")
+
+    config_db.mod_entry('PORT', interface_name, {'pt_interface_id': None})
+    config_db.mod_entry('PORT', interface_name, {'pt_timestamp_template': None})
+
+
 #
 # buffer commands and utilities
 #

scripts/intfutil

@@ -61,6 +61,9 @@ PORT_LINK_TRAINING_STATUS = 'link_training_status'
 VLAN_SUB_INTERFACE_SEPARATOR = "."
 VLAN_SUB_INTERFACE_TYPE = "802.1q-encapsulation"
 
+PORT_PATH_TRACING_INTERFACE_ID = "pt_interface_id"
+PORT_PATH_TRACING_TIMESTAMP_TEMPLATE = "pt_timestamp_template"
+
 SUB_PORT = "subport"
 
 def get_frontpanel_port_list(config_db):
@@ -867,10 +870,79 @@ class IntfFecStatus(object):
                 table.append((key, oper_fec, admin_fec))
         return table
 
+
+# ========================== interface path-tracing logic ==========================
+
+header_path_tracing = ['Interface', 'Alias', 'Oper', 'Admin', 'PT Interface ID', 'PT Timestamp Template']
+
+class IntfPathTracing(object):
+
+    def __init__(self, intf_name, namespace_option, display_option):
+        """
+        Class constructor method
+        :param self:
+        :param intf_name: string of interface
+        :return:
+        """
+        self.db = None
+        self.config_db = None
+        self.intf_name = intf_name
+        self.table = []
+        self.multi_asic = multi_asic_util.MultiAsic(
+            display_option, namespace_option)
+
+        if intf_name is not None and intf_name == SUB_PORT:
+            self.intf_name = None
+
+    def display_intf_path_tracing(self):
+        self.get_intf_path_tracing()
+
+        # Sorting and tabulating the result table.
+        sorted_table = natsorted(self.table)
+        print(tabulate(sorted_table, header_path_tracing, tablefmt="simple", stralign='right'))
+
+    def generate_intf_path_tracing(self):
+        """
+            Generate interface path-tracing output
+        """
+
+        i = {}
+        table = []
+        key = []
+
+        intf_fs = parse_interface_in_filter(self.intf_name)
+        #
+        # Iterate through all the keys and append port's associated state to
+        # the result table.
+        #
+        for i in self.appl_db_keys:
+            key = re.split(':', i, maxsplit=1)[-1].strip()
+            if key in self.front_panel_ports_list:
+                if self.multi_asic.skip_display(constants.PORT_OBJ, key):
+                    continue
+
+                if self.intf_name is None or key in intf_fs:
+                    table.append((key,
+                        appl_db_port_status_get(self.db, key, PORT_ALIAS),
+                        appl_db_port_status_get(self.db, key, PORT_OPER_STATUS),
+                        appl_db_port_status_get(self.db, key, PORT_ADMIN_STATUS),
+                        appl_db_port_status_get(self.db, key, PORT_PATH_TRACING_INTERFACE_ID),
+                        appl_db_port_status_get(self.db, key, PORT_PATH_TRACING_TIMESTAMP_TEMPLATE)))
+        return table
+
+    @multi_asic_util.run_on_multi_asic
+    def get_intf_path_tracing(self):
+        self.front_panel_ports_list = get_frontpanel_port_list(self.config_db)
+        self.appl_db_keys = appl_db_keys_get(self.db, self.front_panel_ports_list, None)
+
+        if self.appl_db_keys:
+            self.table += self.generate_intf_path_tracing()
+
+
 def main():
     parser = argparse.ArgumentParser(description='Display Interface information',
                                      formatter_class=argparse.RawTextHelpFormatter)
-    parser.add_argument('-c', '--command', type=str, help='get interface status or description or auto negotiation status or tpid', default=None)
+    parser.add_argument('-c', '--command', type=str, help='get interface status or description or auto negotiation status or tpid or path-tracing information', default=None)
     parser.add_argument('-i', '--interface', type=str, help='interface information for specific port: Ethernet0', default=None)
     parser = multi_asic_util.multi_asic_args(parser)
     args = parser.parse_args()
@@ -893,6 +965,9 @@ def main():
     elif args.command == "fec":
         interface_fec_status = IntfFecStatus(args.interface, args.namespace, args.display)
         interface_fec_status.display_fec_status()
+    elif args.command == "path-tracing":
+        interface_path_tracing = IntfPathTracing(args.interface, args.namespace, args.display)
+        interface_path_tracing.display_intf_path_tracing()
 
     sys.exit(0)
 

show/interfaces/__init__.py

@@ -870,3 +870,28 @@ def tablelize(keys):
 
     header = ['Interface', 'Mode']
     click.echo(tabulate(tablelize(keys), header, tablefmt="simple", stralign='left'))
+
+
+# 'path-tracing' subcommand ("show interfaces path-tracing")
+@interfaces.command("path-tracing")
+@click.argument('interfacename', required=False)
+@multi_asic_util.multi_asic_click_options
+@click.option('--verbose', is_flag=True, help="Enable verbose output")
+def path_tracing(interfacename, namespace, display, verbose):
+    """Show Interface Path Tracing information"""
+
+    ctx = click.get_current_context()
+
+    cmd = ['intfutil', '-c', 'path-tracing']
+
+    if interfacename is not None:
+        interfacename = try_convert_interfacename_from_alias(ctx, interfacename)
+
+        cmd += ['-i', str(interfacename)]
+    else:
+        cmd += ['-d', str(display)]
+
+    if namespace is not None:
+        cmd += ['-n', str(namespace)]
+
+    clicommon.run_command(cmd, display_cmd=verbose)

tests/config_int_path_tracing_test.py

@@ -0,0 +1,57 @@
+import os
+import config.main as config
+from click.testing import CliRunner
+from utilities_common.db import Db
+
+
+class TestConfigInterfacePathTracing(object):
+    def test_interface_path_tracing_check(self):
+        runner = CliRunner()
+        db = Db()
+
+        obj = {'config_db': db.cfgdb, 'namespace': ''}
+        result = runner.invoke(config.config.commands["interface"].commands["path-tracing"].commands["add"],
+                               ["Ethernet0", "--interface-id", "129"], obj=obj)
+        assert result.exit_code == 0
+
+        obj = {'config_db': db.cfgdb, 'namespace': ''}
+        result = runner.invoke(config.config.commands["interface"].commands["path-tracing"].commands["add"],
+                               ["Ethernet0", "--interface-id", "129", "--ts-template", "template2"], obj=obj)
+        assert result.exit_code == 0
+
+        obj = {'config_db': db.cfgdb, 'namespace': ''}
+        result = runner.invoke(config.config.commands["interface"].commands["path-tracing"].commands["del"],
+                               ["Ethernet0"], obj=obj)
+        assert result.exit_code == 0
+
+        os.environ['SONIC_CLI_IFACE_MODE'] = "alias"
+        result = runner.invoke(config.config.commands["interface"].commands["path-tracing"].commands["add"],
+                               ["Ethernet0", "--interface-id", "129"], obj=obj)
+        os.environ['SONIC_CLI_IFACE_MODE'] = "default"
+        assert result.exit_code == 0
+
+        os.environ['SONIC_CLI_IFACE_MODE'] = "alias"
+        result = runner.invoke(config.config.commands["interface"].commands["path-tracing"].commands["add"],
+                               ["Ethernet0", "--interface-id", "129", "--ts-template", "template2"], obj=obj)
+        os.environ['SONIC_CLI_IFACE_MODE'] = "default"
+        assert result.exit_code == 0
+
+        os.environ['SONIC_CLI_IFACE_MODE'] = "alias"
+        result = runner.invoke(config.config.commands["interface"].commands["path-tracing"].commands["del"],
+                               ["Ethernet0"], obj=obj)
+        os.environ['SONIC_CLI_IFACE_MODE'] = "default"
+        assert result.exit_code == 0
+
+    def test_interface_invalid_path_tracing_check(self):
+        runner = CliRunner()
+        db = Db()
+
+        obj = {'config_db': db.cfgdb, 'namespace': ''}
+        result = runner.invoke(config.config.commands["interface"].commands["path-tracing"].commands["add"],
+                               ["Ethernet0", "--interface-id", "4096"], obj=obj)
+        assert "Error: Invalid value" in result.output
+
+        obj = {'config_db': db.cfgdb, 'namespace': ''}
+        result = runner.invoke(config.config.commands["interface"].commands["path-tracing"].commands["add"],
+                               ["Ethernet0", "--interface-id", "129", "--ts-template", "template5"], obj=obj)
+        assert "Error: Invalid value" in result.output

tests/intfutil_test.py

@@ -127,6 +127,27 @@
 Ethernet124          rs         auto
 """
 
+show_interfaces_path_tracing_output = """\
+  Interface      Alias    Oper    Admin    PT Interface ID    PT Timestamp Template
+-----------  ---------  ------  -------  -----------------  -----------------------
+  Ethernet0  Ethernet0    down       up                129                template3
+ Ethernet16       etp5      up       up                130                template1
+ Ethernet24       etp6      up       up                131                template2
+ Ethernet28       etp8      up       up                132                template3
+ Ethernet32       etp9      up       up                133                template4
+ Ethernet36      etp10      up       up                N/A                      N/A
+Ethernet112      etp29      up       up                N/A                      N/A
+Ethernet116      etp30      up       up                N/A                      N/A
+Ethernet120      etp31      up       up                N/A                      N/A
+Ethernet124      etp32      up       up                N/A                      N/A
+"""
+
+show_interfaces_path_tracing_Ethernet0_output = """\
+  Interface      Alias    Oper    Admin    PT Interface ID    PT Timestamp Template
+-----------  ---------  ------  -------  -----------------  -----------------------
+  Ethernet0  Ethernet0    down       up                129                template3
+"""
+
 class TestIntfutil(TestCase):
     @classmethod
     def setup_class(cls):
@@ -360,6 +381,31 @@ def test_show_interfaces_fec_status(self):
         assert result.exit_code == 0
         assert result.output == show_interface_fec_status_output
 
+    def test_show_interfaces_path_tracing_status(self):
+        result = self.runner.invoke(show.cli.commands["interfaces"].commands["path-tracing"], [])
+        print(result.exit_code)
+        print(result.output)
+        assert result.exit_code == 0
+        assert result.output == show_interfaces_path_tracing_output
+
+        # Test 'intfutil -c path-tracing'
+        output = subprocess.check_output(['intfutil', '-c', 'path-tracing'], stderr=subprocess.STDOUT, text=True)
+        print(output)
+        assert result.output == show_interfaces_path_tracing_output
+
+    def test_show_interfaces_path_tracing_Ethernet0_status(self):
+        result = self.runner.invoke(show.cli.commands["interfaces"].commands["path-tracing"], ["Ethernet0"])
+        print(result.exit_code)
+        print(result.output)
+        assert result.exit_code == 0
+        assert result.output == show_interfaces_path_tracing_Ethernet0_output
+
+        # Test 'intfutil -c path-tracing'
+        output = subprocess.check_output(['intfutil', '-c', 'path-tracing', '-i', 'Ethernet0'],
+                                         stderr=subprocess.STDOUT, text=True)
+        print(output)
+        assert result.output == show_interfaces_path_tracing_Ethernet0_output
+
     @classmethod
     def teardown_class(cls):
         print("TEARDOWN")