[sflow] Added sflow dropped packet notification support

config/main.py

@@ -8786,6 +8786,38 @@ def global_sample_direction(ctx, direction):
     except ValueError as e:
         ctx.fail("Invalid ConfigDB. Error: {}".format(e))
 
+def is_sflow_mirror_on_drop_supported():
+    state_db = SonicV2Connector(use_unix_socket_path=True)
+    state_db.connect(state_db.STATE_DB, False)
+    entry_name="SWITCH_CAPABILITY|switch"
+    supported = state_db.get(state_db.STATE_DB, entry_name, "MIRROR_ON_DROP_CAPABLE")
+    return supported
+
+#
+# 'sflow' command ('config sflow drop-monitor-limit ...')
+#
+@sflow.command('drop-monitor-limit')
+@click.argument('limit', metavar='<packet_per_second>', required=True, type=int)
+@click.pass_context
+def drop_monitor_limit(ctx, limit):
+    """Enable and rate limit dropped packet notifications. (1-500, 0 to disable)"""
+    if ADHOC_VALIDATION:
+        if is_sflow_mirror_on_drop_supported() == 'false':
+            ctx.fail("Drop monitor is not supported on this platform")
+
+    if limit not in range(0, 501):
+        ctx.fail("Drop monitor limit must be between 1-500 (0 to disable)")
+
+    config_db = ctx.obj['db']
+    sflow_tbl = config_db.get_table('SFLOW')
+
+    if not sflow_tbl:
+        sflow_tbl = {'global': {'admin_state': 'down'}}
+
+    sflow_tbl['global']['drop_monitor_limit'] = limit
+    config_db.mod_entry('SFLOW', 'global', sflow_tbl['global'])
+
+
 def is_valid_sample_rate(rate):
     return rate.isdigit() and int(rate) in range(256, 8388608 + 1)
 

doc/Command-Reference.md

@@ -10958,6 +10958,22 @@ This command takes global sflow sample direction. If not configured, default is
   ```
   admin@sonic:~# sudo config sflow sample-direction tx
   ```
+**config sflow drop-monitor-limit**
+
+This command is used to enable or disable the drop notification and configure the maximum rate at which such notifications are sent. By default, drop monitoring is disabled.
+
+- Usage:
+  ```
+  config sflow drop-monitor-limit <limit>
+  ```
+
+  - Parameters:
+    - limit: specify the maximum number of drop notifications to be sent per second. The valid range is 0 to 500. 0 means disable drop monitoring.
+
+- Example:
+  ```
+  admin@sonic:~# sudo config sflow drop-monitor-limit 100
+  ```
 **config sflow interface**
 
 Enable/disable sflow at an interface level. By default, sflow is enabled on all interfaces at the interface level. Use this command to explicitly disable sFlow for a specific interface. An interface is sampled if sflow is enabled globally as well as at the interface level. Note that this configuration deals only with sFlow flow samples and not counter samples.

show/sflow.py

@@ -70,17 +70,22 @@ def show_sflow_global(config_db):
         if ('sample_direction' in sflow_info['global']):
             global_sample_dir = sflow_info['global']['sample_direction']
 
+    ljust_len = 35
+
     click.echo("\nsFlow Global Information:")
-    click.echo("  sFlow Admin State:".ljust(30) + "{}".format(global_admin_state))
-    click.echo("  sFlow Sample Direction:".ljust(30) + "{}".format(global_sample_dir))
+    click.echo("  sFlow Admin State:".ljust(ljust_len) + "{}".format(global_admin_state))
+    click.echo("  sFlow Sample Direction:".ljust(ljust_len) + "{}".format(global_sample_dir))
 
-    click.echo("  sFlow Polling Interval:".ljust(30), nl=False)
+    click.echo("  sFlow Polling Interval:".ljust(ljust_len), nl=False)
     if (sflow_info and 'polling_interval' in sflow_info['global']):
         click.echo("{}".format(sflow_info['global']['polling_interval']))
     else:
         click.echo("default")
 
-    click.echo("  sFlow AgentID:".ljust(30), nl=False)
+    drop_monitor_limit = sflow_info.get('global', {}).get('drop_monitor_limit', 0) if sflow_info else 0
+    click.echo("  sFlow Drop Notification Limit:".ljust(ljust_len) + "{}".format(drop_monitor_limit))
+
+    click.echo("  sFlow AgentID:".ljust(ljust_len), nl=False)
     if (sflow_info and 'agent_id' in sflow_info['global']):
         click.echo("{}".format(sflow_info['global']['agent_id']))
     else:
@@ -91,7 +96,7 @@ def show_sflow_global(config_db):
     for collector_name in sorted(list(sflow_info.keys())):
         vrf_name = (sflow_info[collector_name]['collector_vrf']
                     if 'collector_vrf' in sflow_info[collector_name] else 'default')
-        click.echo("    Name: {}".format(collector_name).ljust(30) +
+        click.echo("    Name: {}".format(collector_name).ljust(ljust_len) +
                    "IP addr: {} ".format(sflow_info[collector_name]['collector_ip']).ljust(25) +
                    "UDP port: {}".format(sflow_info[collector_name]['collector_port']).ljust(17) +
                    "VRF: {}".format(vrf_name))

tests/sflow_test.py

@@ -17,14 +17,15 @@
 # Expected output for 'show sflow'
 show_sflow_output = """
 sFlow Global Information:
-  sFlow Admin State:          up
-  sFlow Sample Direction:     both
-  sFlow Polling Interval:     0
-  sFlow AgentID:              default
+  sFlow Admin State:               up
+  sFlow Sample Direction:          both
+  sFlow Polling Interval:          0
+  sFlow Drop Notification Limit:   0
+  sFlow AgentID:                   default
 
   2 Collectors configured:
-    Name: prod                IP addr: fe80::6e82:6aff:fe1e:cd8e UDP port: 6343   VRF: mgmt
-    Name: ser5                IP addr: 172.21.35.15    UDP port: 6343   VRF: default
+    Name: prod                     IP addr: fe80::6e82:6aff:fe1e:cd8e UDP port: 6343   VRF: mgmt
+    Name: ser5                     IP addr: 172.21.35.15    UDP port: 6343   VRF: default
 """
 
 # Expected output for 'show sflow interface'
@@ -63,7 +64,7 @@ def test_show_sflow_intf(self):
         assert result.exit_code == 0
         assert result.output == show_sflow_intf_output
 
-    @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
+    @patch("config.validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
     @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError))
     def test_config_sflow_disable_enable_yang_validation(self):
         db = Db()
@@ -92,8 +93,8 @@ def test_config_sflow_disable_enable(self):
         # change the output
         global show_sflow_output
         show_sflow_output_local = show_sflow_output.replace(
-            'Admin State:          up',
-            'Admin State:          down')
+            'Admin State:               up',
+            'Admin State:               down')
 
         # run show and check
         result = runner.invoke(show.cli.commands["sflow"], [], obj=db)
@@ -133,8 +134,8 @@ def test_config_sflow_agent_id(self):
         # change the output
         global show_sflow_output
         show_sflow_output_local = show_sflow_output.replace(
-                'sFlow AgentID:              default',
-                'sFlow AgentID:              Ethernet0')
+                'sFlow AgentID:                   default',
+                'sFlow AgentID:                   Ethernet0')
 
         # run show and check
         result = runner.invoke(show.cli.commands["sflow"], [], obj=db)
@@ -171,10 +172,10 @@ def test_config_sflow_collector(self):
         global show_sflow_output
         show_sflow_output_local = show_sflow_output.replace(
     "2 Collectors configured:\n\
-    Name: prod                IP addr: fe80::6e82:6aff:fe1e:cd8e UDP port: 6343   VRF: mgmt\n\
-    Name: ser5                IP addr: 172.21.35.15    UDP port: 6343   VRF: default",
+    Name: prod                     IP addr: fe80::6e82:6aff:fe1e:cd8e UDP port: 6343   VRF: mgmt\n\
+    Name: ser5                     IP addr: 172.21.35.15    UDP port: 6343   VRF: default",
     "1 Collectors configured:\n\
-    Name: ser5                IP addr: 172.21.35.15    UDP port: 6343   VRF: default")
+    Name: ser5                     IP addr: 172.21.35.15    UDP port: 6343   VRF: default")
 
         # run show and check
         result = runner.invoke(show.cli.commands["sflow"], [], obj=db)
@@ -196,7 +197,7 @@ def test_config_sflow_collector(self):
 
         return
 
-    @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
+    @patch("config.validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
     @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError))
     @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_set_entry", mock.Mock(side_effect=JsonPatchConflict))
     def test_config_sflow_collector_invalid_yang_validation(self):
@@ -214,8 +215,8 @@ def test_config_sflow_collector_invalid_yang_validation(self):
             commands["collector"].commands["add"],
             ["prod", "fe80::6e82:6aff:fe1e:cd8e", "--vrf", "mgmt"], obj=obj)
         assert "Invalid ConfigDB. Error" in result.output
-    
-    @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
+
+    @patch("config.validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
     @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError))
     def test_config_sflow_polling_interval_yang_validation(self):
         db = Db()
@@ -250,8 +251,8 @@ def test_config_sflow_polling_interval(self):
         # change the expected output
         global show_sflow_output
         show_sflow_output_local = show_sflow_output.replace(
-            'sFlow Polling Interval:     0',
-            'sFlow Polling Interval:     20')
+            'sFlow Polling Interval:          0',
+            'sFlow Polling Interval:          20')
 
         # run show and check
         result = runner.invoke(show.cli.commands["sflow"], [], obj=db)
@@ -268,7 +269,7 @@ def test_config_sflow_polling_interval(self):
         return
 
     @patch("config.main.ConfigDBConnector.get_table", mock.Mock(return_value={'Ethernet1': {'admin_state': 'sample_state'}}))
-    @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
+    @patch("config.validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
     @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError))
     def test_config_sflow_existing_intf_enable_yang_validation(self):
         db = Db()
@@ -280,15 +281,15 @@ def test_config_sflow_existing_intf_enable_yang_validation(self):
             commands["interface"].commands["enable"], ["Ethernet1"], obj=obj)
         print(result.exit_code, result.output)
         assert "Invalid ConfigDB. Error" in result.output
-    
+
     @patch("config.main.ConfigDBConnector.get_table", mock.Mock(return_value=None))
-    @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
+    @patch("config.validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
     @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError))
     def test_config_sflow_nonexistent_intf_enable_yang_validation(self):
         db = Db()
         runner = CliRunner()
         obj = {'db':db.cfgdb}
- 
+
         result = runner.invoke(config.config.commands["sflow"].
             commands["interface"].commands["enable"], ["Ethernet1"], obj=obj)
         print(result.exit_code, result.output)
@@ -383,7 +384,7 @@ def test_config_enable_all_intf(self):
                                commands["enable"], ["all"], obj=obj)
         print(result.exit_code, result.output)
         assert result.exit_code == 0
-        
+
         # verify in configDb
         sflowSession = db.cfgdb.get_table('SFLOW_SESSION')
         assert sflowSession["all"]["admin_state"] == "up"
@@ -399,7 +400,7 @@ def test_config_sflow_intf_sample_rate_default(self):
         result_out1 = runner.invoke(show.cli.commands["sflow"].commands["interface"], [], obj=Db())
         print(result_out1.exit_code, result_out1.output)
         assert result_out1.exit_code == 0
-        
+
         # set sample-rate to 2500
         result = runner.invoke(config.config.commands["sflow"].
             commands["interface"].commands["sample-rate"],
@@ -446,12 +447,12 @@ def test_config_sflow_sample_direction(self):
         # change the output
         global show_sflow_output
         show_sflow_output_tx = show_sflow_output.replace(
-            'Sample Direction:     both',
-            'Sample Direction:     tx')
+            'Sample Direction:          both',
+            'Sample Direction:          tx')
 
         show_sflow_output_rx = show_sflow_output.replace(
-            'Sample Direction:     both',
-            'Sample Direction:     rx')
+            'Sample Direction:          both',
+            'Sample Direction:          rx')
 
         # run show and check
         result = runner.invoke(show.cli.commands["sflow"], [], obj=db)
@@ -473,6 +474,51 @@ def test_config_sflow_sample_direction(self):
 
         return
 
+    def test_config_sflow_drop_monitor_limit(self):
+        # config sflow drop-monitor-limit <limit>
+        db = Db()
+        runner = CliRunner()
+        obj = {'db':db.cfgdb}
+
+        # drop-monitor-limit : Invalid
+        result = runner.invoke(config.config.commands["sflow"].commands["drop-monitor-limit"], ["NA"], obj=obj)
+        print(result.output)
+        expected = "Error: Invalid value for \"<packet_per_second>\": NA is not a valid integer"
+        assert expected in result.output
+
+        #disable
+        result = runner.invoke(config.config.commands["sflow"].commands["drop-monitor-limit"], ["0"], obj=obj)
+        print(result.exit_code, result.output)
+        assert result.exit_code == 0
+
+        # change the output
+        global show_sflow_output
+
+        show_sflow_output_drop_monitor_disable = show_sflow_output
+
+        # run show and check
+        result = runner.invoke(show.cli.commands["sflow"], [], obj=db)
+        print(result.exit_code, result.output, show_sflow_output_drop_monitor_disable)
+        assert result.exit_code == 0
+        assert result.output == show_sflow_output_drop_monitor_disable
+
+        # enable drop-monitor-limit
+        result = runner.invoke(config.config.commands["sflow"].commands["drop-monitor-limit"], ["100"], obj=obj)
+        print(result.exit_code, result.output)
+        assert result.exit_code == 0
+
+        show_sflow_output_drop_monitor_enable = show_sflow_output.replace(
+            'sFlow Drop Notification Limit:   0',
+            'sFlow Drop Notification Limit:   100')
+
+        # run show and check
+        result = runner.invoke(show.cli.commands["sflow"], [], obj=db)
+        print(result.exit_code, result.output)
+        assert result.exit_code == 0
+        assert result.output == show_sflow_output_drop_monitor_enable
+
+        return
+
     def test_config_all_intf_sample_direction(self):
         db = Db()
         runner = CliRunner()