Fresh python support, GBP support, fail-static mode

.github/workflows/tests.yaml

@@ -24,9 +24,13 @@ jobs:
   tests:
     runs-on: ubuntu-24.04
     strategy:
-      fail-fast: true
+      fail-fast: false
       matrix:
-        python-version: ["3.8"]
+        include:
+          - python-version: "3.8"
+            toxenv: py38
+          - python-version: "3.14"
+            toxenv: py314
     steps:
       - uses: actions/checkout@v3
       - uses: actions/setup-python@v5
@@ -40,7 +44,5 @@ jobs:
         run: pip install tox
       - name: Unit tests
         run: |
-          tox -e ${{ matrix.python-version }}
-      # - name: Functional tests
-      #   run: |
-      #     tox -e ${{ matrix.python-version }}-functional
+          tox -e ${{ matrix.toxenv }}
+

evpn_connector/bgp/client.py

@@ -762,6 +762,24 @@ def list_peer(self, peer_address):
         resp = self._extract_response_from_grpc(list_peers_channel)
         return resp
 
+    def count_peers(self):
+        """Return (total, established) counts of configured BGP peers.
+
+        Used by fail-static: a peer that is configured but not
+        ESTABLISHED means the local RIB no longer reflects the true
+        remote topology (e.g. the route reflector is gone), so flows
+        derived from it must not be trusted for deletion.
+        """
+        peers = self.list_peer("")
+        if not peers:
+            return 0, 0
+        established = sum(
+            1
+            for resp in peers
+            if resp.peer.state.session_state == gobgp_pb2.PeerState.ESTABLISHED
+        )
+        return len(peers), established
+
     def reset_peer(self, address, soft, direction):
         self.stub.ResetPeer(
             gobgp_pb2.ResetPeerRequest(

evpn_connector/cmd/evpn.py

@@ -30,6 +30,7 @@
 from evpn_connector.common import sentry
 from evpn_connector.ovs import client as ovs_client
 from evpn_connector.service import evpn
+from evpn_connector.service import objects as evpnobj
 
 
 OBSENDER_APP_NAME = constants.GLOBAL_SERVICE_NAME
@@ -70,13 +71,18 @@ def main():
         router_mac_type5=CONF.gobgp.router_mac_type5,
     )
 
+    # The tunnel and the flows have to agree on it, so it is read once
+    # and handed to both.
+    evpnobj.set_gbp(CONF.ovs.gbp)
+
     # Init ovs client
     shell_ovs_client = ovs_client.OvSClient(
         sw_name=CONF.ovs.switch_name,
         tmp_flow_file_path=CONF.ovs.tmp_flow_file_path,
         enable_sudo=CONF.ovs.enable_sudo,
         ovsvsctl_bin=CONF.ovs.ovs_vsctl_bin_path,
         ovsofctl_bin=CONF.ovs.ovs_ofctl_bin_path,
+        gbp=CONF.ovs.gbp,
     )
 
     # Start service
@@ -87,6 +93,7 @@ def main():
         vxlan_udp_port=CONF.ovs.vxlan_udp_port,
         as_number=CONF.gobgp.as_number,
         policy_enabled=CONF.gobgp.policy_enabled,
+        fail_static=CONF.gobgp.fail_static,
         configs_dir=CONF.daemon.configs_dir,
         router_mac_type5=CONF.gobgp.router_mac_type5,
         anycast_status_file=CONF.anycast.anycast_status_file,

evpn_connector/common/conf_opts.py

@@ -87,12 +87,39 @@
         default=constants.TYPE_5_DEFAULT_ROUTER_MAC_EXTENDED,
         help="Value of RouterMacExtended Ext Communities Attr for Type5",
     ),
+    cfg.BoolOpt(
+        name="fail_static",
+        default=True,
+        help=(
+            "Retain the last-known-good OVS flows while any configured "
+            "BGP peer is not ESTABLISHED, or none is configured at all, "
+            "instead of withdrawing flows for routes that vanished from "
+            "the RIB because a session (e.g. the route reflector) was "
+            "lost. Existing connectivity keeps working, and local "
+            "additions and changes still land; only deletions wait for "
+            "the sessions to come back."
+        ),
+    ),
 ]
 
 ovs_opts = [
     cfg.StrOpt(
         name="switch_name", required=True, help="OpenvSwitch switch name"
     ),
+    cfg.BoolOpt(
+        name="gbp",
+        default=False,
+        help=(
+            "Carry the sender's group across the fabric in VXLAN-GBP's "
+            "16-bit Group Policy ID, copying it to and from the low bits "
+            "of the skb mark at the tunnel, which reserves mark bits "
+            "0..15 for it. An id arriving from the wire is taken as it "
+            "comes, so the underlay has to be trusted for it to mean "
+            "anything. Off by default: it makes the tunnel a GBP one, "
+            "and OVS will not mix GBP and non-GBP tunnels on one UDP "
+            "port, so it is a property of a whole fabric."
+        ),
+    ),
     cfg.StrOpt(
         name="tmp_flow_file_path",
         default="/tmp/evpn_tmp_flow_file",

evpn_connector/common/constants.py

@@ -51,6 +51,11 @@
 # Traffic direction in REG1
 REG_FROM_REMOTE = 0
 REG_FROM_LOCAL = 1
+
+# A sender's group: VXLAN-GBP's 16-bit id on the wire, the low 16 bits of the
+# skb mark on the host (a tunnel field does not survive a patch port).
+GBP_TO_MARK = "move:NXM_NX_TUN_GBP_ID[]->NXM_NX_PKT_MARK[0..15]"
+MARK_TO_GBP = "move:NXM_NX_PKT_MARK[0..15]->NXM_NX_TUN_GBP_ID[]"
 # ECMP Multipath hash algorithm (for details see man 7 ovs-actions: multipath)
 ECMP_HASH_ALGORITHM = "symmetric_l3l4+udp"
 

evpn_connector/ovs/client.py

@@ -40,8 +40,10 @@ def __init__(
         enable_sudo=False,
         ovsvsctl_bin=constants.OVSVSCTL_BIN,
         ovsofctl_bin=constants.OVSOFCTL_BIN,
+        gbp=False,
     ):
         self.vxlan_ofport = vxlan_ofport or constants.VXLAN_PORT_OFPORT
+        self.gbp = gbp
         self.sw_name = sw_name
         self.enable_sudo = enable_sudo
         self.tmp_flow_file_path = tmp_flow_file_path
@@ -87,6 +89,12 @@ def create_tun_port(
             "options:dst_port={}".format(vxlan_udp_port),
             "ofport_request={}".format(self.vxlan_ofport),
         ]
+        if self.gbp:
+            cmd.append("options:exts=gbp")
+        else:
+            # Turning it off has to unmake a GBP tunnel too, or a fabric
+            # cannot be moved back; removing an absent key is a no-op.
+            cmd += ["--", "remove", "Interface", port_name, "options", "exts"]
         return shell.runsh(command=cmd, enable_sudo=self.enable_sudo)
 
     def sync_flows(self, flows):

evpn_connector/service/evpn.py

@@ -77,6 +77,7 @@ def __init__(
         event_type=None,
         error_event_type=None,
         policy_enabled=True,
+        fail_static=True,
     ):
         super(EvpnConnectorService, self).__init__(
             step_period=step_period,
@@ -100,6 +101,12 @@ def __init__(
         self.anycast_check_ofport = anycast_check_ofport
         self.anycast_check_mac = anycast_check_mac
         self.anycast_used = False
+        # Fail-static: the last flow set synced while all BGP peers were
+        # established, retained if a peer (e.g. the RR) is later lost so
+        # that a RIB emptied by session loss does not tear down flows.
+        self.fail_static = fail_static
+        self._last_good_flows = set()
+        self._peers_healthy = True
 
     def _setup(self):
         super(EvpnConnectorService, self)._setup()
@@ -342,6 +349,88 @@ def update_peer(
             )
         self.need_reset_peers = False
 
+    def _log_peer_health(self, healthy, reason):
+        # A step is short and this is checked on every one of them, so
+        # only a change of state is worth a line.
+        if healthy == self._peers_healthy:
+            return
+        self._peers_healthy = healthy
+        if healthy:
+            LOG.info("BGP peers healthy again: %s", reason)
+        else:
+            LOG.warning(
+                "BGP peers degraded (%s); holding last-known flows "
+                "(fail-static)",
+                reason,
+            )
+
+    def _upstream_peers_healthy(self):
+        """Whether the RIB can be trusted for flow deletion.
+
+        Healthy means every configured BGP peer is ESTABLISHED. An empty
+        peer list is degraded too: that is what a restarted gobgp looks
+        like before it has read its config, and it is exactly the moment
+        the RIB is empty for a reason that has nothing to do with the
+        fabric. A node that genuinely has no peers loses nothing by it,
+        its snapshot never fills, so retaining it is a no-op. If the peer
+        list cannot be read, assume degraded as well.
+        """
+        try:
+            total, established = self.gobgp_client.count_peers()
+        except Exception as err:
+            self._log_peer_health(
+                False, "failed to read peer states: %s" % str(err)
+            )
+            return False
+        if total == 0:
+            self._log_peer_health(False, "no peers are configured")
+            return False
+        if established < total:
+            self._log_peer_health(
+                False,
+                "only %d of %d peers established" % (established, total),
+            )
+            return False
+        self._log_peer_health(True, "all %d peers established" % total)
+        return True
+
+    def _apply_fail_static(self, target_flows, metrics):
+        """Decide which flows to sync, retaining flows on peer loss.
+
+        While every BGP peer is established the RIB is authoritative:
+        the computed set is applied as-is and snapshotted as
+        last-known-good. If a peer is lost, routes it fed vanish from
+        the RIB; rather than let `replace-flows` delete them, the fresh
+        set is unioned with the snapshot, so existing connectivity
+        survives until the session returns.
+
+        A union only adds. Additions and changes still land: a flow is
+        equal to another by its match alone, and union keeps the fresh
+        side, so a match that is still computed keeps its new action.
+        Deletions do not land: a flow that vanished from the computed
+        set is held until the peers are back, and a port whose ofport is
+        then reused would be reached by the stale flow. That is the
+        price of not tearing down a fabric over a lost session, and the
+        reason this is scoped to peer loss rather than left on always.
+        """
+        if not self.fail_static or self._upstream_peers_healthy():
+            self._last_good_flows = target_flows
+            metrics["fail_static_active"] = 0
+            metrics["fail_static_retained_cnt"] = 0
+            return target_flows
+
+        flows_to_sync = target_flows.union(self._last_good_flows)
+        metrics["fail_static_active"] = 1
+        metrics["fail_static_retained_cnt"] = len(flows_to_sync) - len(
+            target_flows
+        )
+        LOG.debug(
+            "Fail-static active: retaining %d flows on top of %d computed",
+            metrics["fail_static_retained_cnt"],
+            len(target_flows),
+        )
+        return flows_to_sync
+
     def get_anycast_status(self):
         if not os.path.isfile(self.anycast_status_file):
             LOG.warning(
@@ -575,9 +664,13 @@ def _step(self):
             duration_metrics["prep_ovs_time"],
         )
 
+        flows_to_sync = self._apply_fail_static(
+            uni_flows.union(bum_flows), metrics
+        )
+
         start_time = time.time()
         LOG.debug("Sync flows in ovs")
-        self.ovs_client.sync_flows(uni_flows.union(bum_flows))
+        self.ovs_client.sync_flows(flows_to_sync)
         duration_metrics["sync_ovs_time"] = time.time() - start_time
         LOG.info(
             "Sync ovs flows done for %0.4f sec",

evpn_connector/service/objects.py

@@ -27,6 +27,21 @@
 
 LOG = logging.getLogger(__name__)
 
+# Whether the fabric carries a sender's group. A property of the whole
+# switch rather than of one flow, wired once at start-up so that the
+# tunnel and the flows cannot disagree about it.
+_GBP_ENABLED = False
+
+
+def set_gbp(enabled):
+    global _GBP_ENABLED
+    _GBP_ENABLED = enabled
+
+
+def _gbp(action):
+    """The GBP move with its separator, or nothing when gbp is off."""
+    return "%s," % action if _GBP_ENABLED else ""
+
 
 class BaseObj(object):
     def __init__(self):
@@ -296,6 +311,11 @@ def _ovs_to_out_table_action(self, local=False):
         reg1_value = constants.REG_FROM_REMOTE
         if local:
             reg1_value = constants.REG_FROM_LOCAL
+        elif self.port_type == constants.EVPN_EDGE_TYPE_VXLAN:
+            # Off the wire on the switched path: this is the flow a Type 2
+            # announce installs, and VirtNet's only stands in for it when
+            # the announce is missing.
+            res += _gbp(constants.GBP_TO_MARK)
 
         res += "set_field:%d->reg0,set_field:%d->reg1,resubmit(,%d)" % (
             self.vni,
@@ -316,7 +336,10 @@ def ovs_output(self, for_group=False):
                 self.ofport,
             )
         elif self.port_type == constants.EVPN_EDGE_TYPE_VXLAN:
-            res += "set_field:%s->tun_id,set_field:%s->tun_dst,output:%d" % (
+            # Onto the wire: switched and routed alike, or a guest reached
+            # by a /32 is a hole in whatever the identity protects.
+            res += "%sset_field:%s->tun_id,set_field:%s->tun_dst,output:%d" % (
+                _gbp(constants.MARK_TO_GBP),
                 self.tun_id,
                 self.next_hop,
                 self.ofport,
@@ -519,7 +542,10 @@ def ovs_output(self, local=False):
                 self.ofport,
             )
         elif self.port_type == constants.EVPN_EDGE_TYPE_VXLAN:
-            res += "set_field:%s->tun_id,set_field:%s->tun_dst,output:%d" % (
+            # Onto the wire: switched and routed alike, or a guest reached
+            # by a /32 is a hole in whatever the identity protects.
+            res += "%sset_field:%s->tun_id,set_field:%s->tun_dst,output:%d" % (
+                _gbp(constants.MARK_TO_GBP),
                 self.tun_id,
                 self.next_hop,
                 self.ofport,
@@ -1058,7 +1084,10 @@ def import_announces(self, local_ce_prefixes, remote_ce_prefixes):
 
     @property
     def _ovs_to_out_table_action(self):
-        return "action=set_field:%d->reg2,resubmit(,%d)" % (
+        # Off the wire on the routed path, which is what cross-host guest
+        # traffic actually takes.
+        return "action=%sset_field:%d->reg2,resubmit(,%d)" % (
+            _gbp(constants.GBP_TO_MARK),
             self.vrf_number,
             constants.OUTPUT_TABLE_NUM,
         )
@@ -1152,6 +1181,10 @@ def _ovs_to_out_table_action(self, local=False):
         reg1_value = constants.REG_FROM_REMOTE
         if local:
             reg1_value = constants.REG_FROM_LOCAL
+        else:
+            # Off the wire, only on tunnel ingress: local traffic never had
+            # a header, and reading one would erase its sender's mark.
+            res += _gbp(constants.GBP_TO_MARK)
 
         res += "set_field:%d->reg0,set_field:%d->reg1,resubmit(,%d)" % (
             self.vni,
@@ -1164,7 +1197,9 @@ def _ovs_to_out_table_action(self, local=False):
     def ovs_output(
         self, for_group=False, tun_ofport=constants.VXLAN_PORT_OFPORT
     ):
-        return "set_field:%s->tun_id,set_field:%s->tun_dst,output:%d" % (
+        # The flooded path carries it too.
+        return "%sset_field:%s->tun_id,set_field:%s->tun_dst,output:%d" % (
+            _gbp(constants.MARK_TO_GBP),
             self.tun_id,
             self.next_hop,
             tun_ofport,