1313from datetime import datetime
1414from ipaddress import ip_interface
1515from queue import Queue
16+ from threading import Lock , Event , Thread
1617
1718from swsscommon .swsscommon import ConfigDBConnector , SonicV2Connector , \
1819 DBConnector , Select , SubscriberStateTable
2930
3031STATE_DB = 'STATE_DB'
3132APPL_DB = 'APPL_DB'
33+ COUNTERS_DB = 'COUNTERS_DB'
34+ TUNNEL_PKT_COUNTER_TEMPLATE = 'COUNTERS{}IPINIP_TUNNEL_CPU_PKTS'
35+ COUNTER_KEY = 'RX_COUNT'
3236PORTCHANNEL_INTERFACE_TABLE = 'PORTCHANNEL_INTERFACE'
3337TUNNEL_TABLE = 'TUNNEL'
3438PEER_SWITCH_TABLE = 'PEER_SWITCH'
@@ -69,13 +73,18 @@ def __init__(self):
6973 self .config_db .connect ()
7074 self .state_db = SonicV2Connector ()
7175 self .state_db .connect (STATE_DB )
76+ self .counters_db = SonicV2Connector ()
77+ self .counters_db .connect (COUNTERS_DB )
78+ counters_db_separator = self .counters_db .get_db_separator (COUNTERS_DB )
79+ self .tunnel_counter_table = TUNNEL_PKT_COUNTER_TEMPLATE .format (counters_db_separator )
7280 self ._portchannel_intfs = None
7381 self .up_portchannels = None
7482 self .netlink_api = IPRoute ()
7583 self .sniffer = None
7684 self .self_ip = ''
7785 self .packet_filter = ''
7886 self .sniff_intfs = set ()
87+ self .pending_cmds = Queue ()
7988
8089 global portchannel_intfs
8190 portchannel_intfs = [name for name , _ in self .portchannel_intfs ]
@@ -304,6 +313,27 @@ def start_sniffer(self):
304313 while not hasattr (self .sniffer , 'stop_cb' ):
305314 time .sleep (0.1 )
306315
316+ def write_count_to_db (self ):
317+ while True :
318+ # use a set to automatically deduplicate destination IPs
319+ to_run = set ()
320+
321+ to_run .add (tuple (self .pending_cmds .get ()))
322+ pkt_count = 1
323+ while not self .pending_cmds .empty () and len (to_run ) < 100 :
324+ to_run .add (tuple (self .pending_cmds .get ()))
325+ # we should always count each packet, but only ping for each unique IP
326+ pkt_count += 1
327+
328+ for cmds in to_run :
329+ logger .log_info ("Running command '{}'" .format (' ' .join (cmds )))
330+ subprocess .run (cmds , stdout = subprocess .DEVNULL )
331+ try :
332+ curr_count = int (self .counters_db .get (COUNTERS_DB , self .tunnel_counter_table , COUNTER_KEY ))
333+ except TypeError :
334+ curr_count = 0
335+ self .counters_db .set (COUNTERS_DB , self .tunnel_counter_table , COUNTER_KEY , str (curr_count + pkt_count ))
336+
307337 def ping_inner_dst (self , packet ):
308338 """
309339 Pings the inner destination IP for an encapsulated packet
@@ -319,8 +349,7 @@ def ping_inner_dst(self, packet):
319349 cmds .append ('-6' )
320350 dst_ip = packet [IP ].payload [inner_packet_type ].dst
321351 cmds .append (dst_ip )
322- logger .log_info ("Running command '{}'" .format (' ' .join (cmds )))
323- subprocess .run (cmds , stdout = subprocess .DEVNULL )
352+ self .pending_cmds .put (cmds )
324353
325354 def listen_for_tunnel_pkts (self ):
326355 """
@@ -339,7 +368,6 @@ def listen_for_tunnel_pkts(self):
339368 logger .log_notice ('Starting tunnel packet handler for {}'
340369 .format (self .packet_filter ))
341370
342-
343371 app_db = DBConnector (APPL_DB , 0 )
344372 lag_table = SubscriberStateTable (app_db , LAG_TABLE )
345373 sel = Select ()
@@ -355,7 +383,7 @@ def listen_for_tunnel_pkts(self):
355383 elif rc == Select .ERROR :
356384 raise Exception ("Select() error" )
357385 else :
358- lag , op , fvs = lag_table .pop ()
386+ lag , _ , fvs = lag_table .pop ()
359387 if self .sniffer_restart_required (lag , fvs ):
360388 self .sniffer .stop ()
361389 start = datetime .now ()
@@ -374,6 +402,8 @@ def run(self):
374402 Entry point for the TunnelPacketHandler class
375403 """
376404 self .wait_for_portchannels ()
405+ db_thread = Thread (target = self .write_count_to_db , daemon = True )
406+ db_thread .start ()
377407 self .listen_for_tunnel_pkts ()
378408
379409
0 commit comments