Optimize lag_keepalive by crafting the LACPDU packet ourselves (sonic-net#3170)

* Optimize lag_keepalive by crafting the LACPDU packet ourselves

Instead of waiting for a LACPDU packet to be sent and capturing that
(which involves waiting roughly 30 seconds), get the necessary
information from teamd and craft it ourselves. This means that the
60-second wait in making sure that a LACPDU packet is captured and the
keepalive script is ready can be largely eliminated (this has been
reduced to 10 seconds to make sure the script has a chance to craft the
packets and send some LACPDUs).

Signed-off-by: Saikrishna Arcot <[email protected]>

* Fix pre-commit errors

Signed-off-by: Saikrishna Arcot <[email protected]>

* Keep a socket open, and reuse that for sending LACPDUs

Signed-off-by: Saikrishna Arcot <[email protected]>

* Add logic to fork into background after collecting information

Signed-off-by: Saikrishna Arcot <[email protected]>

* Start lag_keepalive before OA pause, and fork after building packets

Start lag_keepalive before pausing orchagent, so that there's less of a
delay between when orchagent is paused and when kexec happens, and so
that fewer events/changes aren't handled by orchagent.

Additionally, add an option into the lag_keepalive script to fork into
the background after generating the LACPDUs and opening sockets, but
before sending the actual packets. This serves as a sort-of error check
to make sure that it is at least able to send LACPDU packets, and didn't
bail out early.

Signed-off-by: Saikrishna Arcot <[email protected]>

---------

Signed-off-by: Saikrishna Arcot <[email protected]>

Author: saiarcot895

scripts/fast-reboot

@@ -744,6 +744,13 @@ fi
 
 init_warm_reboot_states
 
+# start sending LACPDUs to keep the LAGs refreshed
+# the process will die in 30s
+debug "Starting lag_keepalive to send LACPDUs ..."
+timeout 30 python3 ${LAG_KEEPALIVE_SCRIPT} --fork-into-background
+# give the lag_keepalive script a chance to send some LACPDUs
+sleep 5
+
 setup_control_plane_assistant
 
 TEAMD_INCREASE_RETRY_COUNT=0
@@ -813,13 +820,6 @@ fi
 # disable trap-handlers which were set before
 trap '' EXIT HUP INT QUIT TERM KILL ABRT ALRM
 
-# start sending LACPDUs to keep the LAGs refreshed
-# this is a non-blocking call, and the process will die in 300s
-debug "Starting lag_keepalive to send LACPDUs ..."
-timeout 300 python ${LAG_KEEPALIVE_SCRIPT} &
-# give the lag_keepalive script a chance to get ready (30s) and collect one lacpdu before going down (30s)
-sleep 60
-
 if [ -x ${LOG_SSD_HEALTH} ]; then
     debug "Collecting logs to check ssd health before ${REBOOT_TYPE}..."
     ${LOG_SSD_HEALTH}

scripts/lag_keepalive.py

@@ -1,13 +1,22 @@
 #!/usr/bin/env python3
 
+import argparse
+import json
+import os
 from scapy.config import conf
 conf.ipv6_enabled = False
-from scapy.all import sendp, sniff
-from swsscommon.swsscommon import ConfigDBConnector
-import time, threading, traceback
-import syslog
+from scapy.layers.l2 import Ether  # noqa: E402
+from scapy.sendrecv import sendp  # noqa: E402
+import scapy.contrib.lacp  # noqa: E402
+import subprocess  # noqa: E402
+import syslog  # noqa: E402
+import time  # noqa: E402
+import traceback  # noqa: E402
+from swsscommon.swsscommon import ConfigDBConnector  # noqa: E402
 
 SYSLOG_ID = 'lag_keepalive'
+SLOW_PROTOCOL_MAC_ADDRESS = "01:80:c2:00:00:02"
+LACP_ETHERTYPE = 0x8809
 
 
 def log_info(msg):
@@ -22,67 +31,92 @@ def log_error(msg):
     syslog.closelog()
 
 
-def sniff_lacpdu(device_mac, lag_member, lag_member_to_packet):
-    sniffed_packet = sniff(iface=lag_member,
-        filter="ether proto 0x8809 and ether src {}".format(device_mac),
-        count=1, timeout=30)
-    lag_member_to_packet[lag_member] = sniffed_packet
+def getCmdOutput(cmd):
+    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)
+    return proc.communicate()[0], proc.returncode
+
+
+def get_port_channel_config(portChannelName):
+    (processStdout, _) = getCmdOutput(["teamdctl", portChannelName, "state", "dump"])
+    return json.loads(processStdout)
+
+
+def craft_lacp_packet(portChannelConfig, portName):
+    portConfig = portChannelConfig["ports"][portName]
+    actorConfig = portConfig["runner"]["actor_lacpdu_info"]
+    partnerConfig = portConfig["runner"]["partner_lacpdu_info"]
+    l2 = Ether(dst=SLOW_PROTOCOL_MAC_ADDRESS, src=portConfig["ifinfo"]["dev_addr"], type=LACP_ETHERTYPE)
+    l3 = scapy.contrib.lacp.SlowProtocol(subtype=0x01)
+    l4 = scapy.contrib.lacp.LACP()
+    l4.version = 0x1
+    l4.actor_system_priority = actorConfig["system_priority"]
+    l4.actor_system = actorConfig["system"]
+    l4.actor_key = actorConfig["key"]
+    l4.actor_port_priority = actorConfig["port_priority"]
+    l4.actor_port_number = actorConfig["port"]
+    l4.actor_state = actorConfig["state"]
+    l4.partner_system_priority = partnerConfig["system_priority"]
+    l4.partner_system = partnerConfig["system"]
+    l4.partner_key = partnerConfig["key"]
+    l4.partner_port_priority = partnerConfig["port_priority"]
+    l4.partner_port_number = partnerConfig["port"]
+    l4.partner_state = partnerConfig["state"]
+    packet = l2 / l3 / l4
+    return packet
 
 
 def get_lacpdu_per_lag_member():
     appDB = ConfigDBConnector()
     appDB.db_connect('APPL_DB')
     appDB_lag_info = appDB.get_keys('LAG_MEMBER_TABLE')
-    configDB = ConfigDBConnector()
-    configDB.db_connect('CONFIG_DB')
-    device_mac = configDB.get(configDB.CONFIG_DB,  "DEVICE_METADATA|localhost", "mac")
-    hwsku = configDB.get(configDB.CONFIG_DB,  "DEVICE_METADATA|localhost", "hwsku")
     active_lag_members = list()
     lag_member_to_packet = dict()
-    sniffer_threads = list()
     for lag_entry in appDB_lag_info:
         lag_name = str(lag_entry[0])
-        oper_status = appDB.get(appDB.APPL_DB,"LAG_TABLE:{}".format(lag_name), "oper_status")
+        oper_status = appDB.get(appDB.APPL_DB, "LAG_TABLE:{}".format(lag_name), "oper_status")
         if oper_status == "up":
             # only apply the workaround for active lags
             lag_member = str(lag_entry[1])
             active_lag_members.append(lag_member)
-            # use threading to capture lacpdus from several lag members simultaneously
-            sniffer_thread = threading.Thread(target=sniff_lacpdu,
-                args=(device_mac, lag_member, lag_member_to_packet))
-            sniffer_thread.start()
-            sniffer_threads.append(sniffer_thread)
-
-    # sniff for lacpdu should finish in <= 30s. sniff timeout is also set to 30s
-    for sniffer in sniffer_threads:
-        sniffer.join(timeout=30)
-    
+            # craft lacpdu packets for each lag member based on config
+            port_channel_config = get_port_channel_config(lag_name)
+            packet = craft_lacp_packet(port_channel_config, lag_member)
+            socket = conf.L2socket(iface=lag_member)
+            lag_member_to_packet[lag_member] = (socket, packet)
+
     return active_lag_members, lag_member_to_packet
 
 
 def lag_keepalive(lag_member_to_packet):
-    while True:
-        for lag_member, packet in lag_member_to_packet.items():
+    num_iterations = 300
+    current_iteration = 0
+    while current_iteration < num_iterations:
+        for lag_member, (socket, packet) in lag_member_to_packet.items():
             try:
-                sendp(packet, iface=lag_member, verbose=False)
+                sendp(packet, socket=socket, verbose=False)
             except Exception:
                 # log failure and continue to send lacpdu
                 traceback_msg = traceback.format_exc()
                 log_error("Failed to send LACPDU packet from interface {} with error: {}".format(
                     lag_member, traceback_msg))
                 continue
         log_info("sent LACPDU packets via {}".format(lag_member_to_packet.keys()))
+        current_iteration += 1
         time.sleep(1)
 
 
 def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--fork-into-background', action='store_true')
+    args = parser.parse_args()
+
     while True:
         try:
             active_lag_members, lag_member_to_packet = get_lacpdu_per_lag_member()
             if len(active_lag_members) != len(lag_member_to_packet.keys()):
-                log_error("Failed to capture LACPDU packets for some lag members. " +\
-                "Active lag members: {}. LACPDUs captured for: {}".format(
-                    active_lag_members, lag_member_to_packet.keys()))
+                log_error("Failed to craft LACPDU packets for some lag members. " +
+                          "Active lag members: {}. LACPDUs crafted for: {}".format(
+                              active_lag_members, lag_member_to_packet.keys()))
 
             log_info("ready to send LACPDU packets via {}".format(lag_member_to_packet.keys()))
         except Exception:
@@ -94,9 +128,15 @@ def main():
         # if no exceptions are thrown, break from loop as LACPDUs are ready to be sent
         break
 
+    if args.fork_into_background:
+        pid = os.fork()
+        if pid != 0:   # The parent process
+            os._exit(0)   # Exit parent of the child process
+
     if lag_member_to_packet:
-        # start an infinite loop to keep sending lacpdus from lag member ports
+        # start a loop to keep sending lacpdus from lag member ports
         lag_keepalive(lag_member_to_packet)
 
+
 if __name__ == "__main__":
     main()