diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 18e99e1..ccc6f55 100644 --- a/.github/workflows/tests.yaml +++ b/.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 }} + diff --git a/evpn_connector/bgp/client.py b/evpn_connector/bgp/client.py index dc32ba1..f6f0590 100644 --- a/evpn_connector/bgp/client.py +++ b/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( diff --git a/evpn_connector/cmd/evpn.py b/evpn_connector/cmd/evpn.py index 68b0b0d..fed287a 100644 --- a/evpn_connector/cmd/evpn.py +++ b/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,6 +71,10 @@ 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, @@ -77,6 +82,7 @@ def main(): 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,8 @@ 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, + fail_static_min_peers=CONF.gobgp.fail_static_min_peers, configs_dir=CONF.daemon.configs_dir, router_mac_type5=CONF.gobgp.router_mac_type5, anycast_status_file=CONF.anycast.anycast_status_file, diff --git a/evpn_connector/common/conf_opts.py b/evpn_connector/common/conf_opts.py index 4c022d1..d51f1e7 100644 --- a/evpn_connector/common/conf_opts.py +++ b/evpn_connector/common/conf_opts.py @@ -87,12 +87,52 @@ 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 too few " + "configured BGP peers are 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." + ), + ), + cfg.IntOpt( + name="fail_static_min_peers", + default=0, + min=0, + help=( + "How many configured BGP peers must be ESTABLISHED for the " + "RIB to be trusted for flow deletion. 0 means all of them, " + "which is right for peers that each feed their own routes; " + "set it to 1 with redundant route reflectors, where one " + "session still carries the whole RIB. Values above the " + "number of configured peers mean all of them." + ), + ), ] 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", diff --git a/evpn_connector/common/constants.py b/evpn_connector/common/constants.py index 1f76a97..89677de 100644 --- a/evpn_connector/common/constants.py +++ b/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" diff --git a/evpn_connector/ovs/client.py b/evpn_connector/ovs/client.py index 105deea..d5f85df 100644 --- a/evpn_connector/ovs/client.py +++ b/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): diff --git a/evpn_connector/service/evpn.py b/evpn_connector/service/evpn.py index 789e1e0..635aed9 100644 --- a/evpn_connector/service/evpn.py +++ b/evpn_connector/service/evpn.py @@ -77,6 +77,8 @@ def __init__( event_type=None, error_event_type=None, policy_enabled=True, + fail_static=True, + fail_static_min_peers=0, ): super(EvpnConnectorService, self).__init__( step_period=step_period, @@ -100,6 +102,14 @@ 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 enough 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.fail_static_min_peers = fail_static_min_peers + self._last_good_flows = set() + self._peers_healthy = True def _setup(self): super(EvpnConnectorService, self)._setup() @@ -342,6 +352,93 @@ 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 at least fail_static_min_peers of the configured + BGP peers are ESTABLISHED, all of them if the option is 0. 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. + + A peer list that cannot be read is not degradation but a local + gobgp fault, indistinguishable here from a broken fabric, so it + is raised: the step stops before syncing and flows stay as they + are, which is what fail-static would have done anyway, and it is + reported as an error instead of a held state. + """ + total, established = self.gobgp_client.count_peers() + if total == 0: + self._log_peer_health(False, "no peers are configured") + return False + # More required than configured means all of them. + required = min(self.fail_static_min_peers or total, total) + if established < required: + self._log_peer_health( + False, + "only %d of %d peers established, %d required" + % (established, total, required), + ) + return False + self._log_peer_health( + True, "%d of %d peers established" % (established, total) + ) + return True + + def _apply_fail_static(self, target_flows, metrics): + """Decide which flows to sync, retaining flows on peer loss. + + While enough BGP peers are 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 +672,13 @@ def _step(self): duration_metrics["prep_ovs_time"], ) + uni_flows = uni_flows.union(bum_flows) + + flows_to_sync = self._apply_fail_static(uni_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", diff --git a/evpn_connector/service/objects.py b/evpn_connector/service/objects.py index b60c571..bc2b43b 100644 --- a/evpn_connector/service/objects.py +++ b/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, diff --git a/evpn_connector/tests/unit/test_evpn_objects.py b/evpn_connector/tests/unit/test_evpn_objects.py index 2a3d650..70cd387 100644 --- a/evpn_connector/tests/unit/test_evpn_objects.py +++ b/evpn_connector/tests/unit/test_evpn_objects.py @@ -16,11 +16,20 @@ import sys +import pytest + from evpn_connector.common import constants from evpn_connector.service import objects class TestEvpnConnectorObjects(object): + @pytest.fixture(autouse=True) + def gbp_off_afterwards(self): + # It is process-wide state; leaving it on would build every + # later flow in this run as a GBP one. + yield + objects.set_gbp(False) + def test_ovs_flow(self): match1 = "table=1,priority=100" match2 = "table=1,priority=200" @@ -333,3 +342,85 @@ def test_vnet_from_ce(self): assert len(vnet_vnis) == len(vnets) == 2 assert expected_vnets == vnets assert expected_vnis == vnet_vnis + + def _gbp_objects(self): + ce = objects.ClientEdge( + mac="d8:6b:5c:cd:97:ee", + vni=1, + ip="192.168.1.1", + rt=objects.RouteTarget(targets=[(65001, 100)]), + as_number=65001, + ofport=1, + port_type="vxlan", + tag=100, + next_hop="10.0.0.2", + ) + + prefix = objects.ClientEdgePrefix( + prefix="10.1.1.0", + prefix_len=24, + mac="d8:6b:5c:cd:97:ee", + router_mac="d8:6b:5c:cd:97:ef", + vni=1, + rt=objects.RouteTarget(targets=[(65001, 100)]), + as_number=65001, + ofport=1, + port_type="vxlan", + tag=100, + next_hop="10.0.0.2", + ) + + vnet = objects.VirtNet( + vni=1, + next_hop="10.0.0.2", + rt=objects.RouteTarget(targets=[(65001, 100)]), + as_number=65001, + ) + return ce, prefix, vnet + + def test_the_identity_is_carried_only_at_the_tunnel(self): + """The identity is put on the wire and taken off it, once each. + + Every path out has to carry it (a guest reached by a /32, or by + flooding, is otherwise a hole), and only tunnel ingress may take it + off: local traffic never had a header, and reading one would erase + its sender's mark. + """ + objects.set_gbp(True) + ce, prefix, vnet = self._gbp_objects() + + assert constants.MARK_TO_GBP in ce.ovs_output() + assert constants.MARK_TO_GBP in prefix.ovs_output() + assert constants.MARK_TO_GBP in vnet.ovs_output() + + # Both flows that a tunnel ingress can hit: the Type 2 one, and + # VirtNet's stand-in for a missing Type 2 announce. + for obj in (ce, vnet): + assert constants.GBP_TO_MARK in obj._ovs_to_out_table_action( + local=False + ) + assert constants.GBP_TO_MARK not in obj._ovs_to_out_table_action( + local=True + ) + + def test_no_identity_is_carried_when_gbp_is_off(self): + """Off by default means the flows are exactly the ones from before. + + OVS will not mix GBP and non-GBP tunnels on one UDP port, so a + fabric that has not asked for this must be left as it was. + """ + objects.set_gbp(False) + ce, prefix, vnet = self._gbp_objects() + + for flow in ( + ce.ovs_output(), + prefix.ovs_output(), + vnet.ovs_output(), + ce._ovs_to_out_table_action(local=False), + ce._ovs_to_out_table_action(local=True), + vnet._ovs_to_out_table_action(local=False), + vnet._ovs_to_out_table_action(local=True), + ): + assert constants.MARK_TO_GBP not in flow + assert constants.GBP_TO_MARK not in flow + assert ",," not in flow and not flow.endswith(",") diff --git a/evpn_connector/tests/unit/test_evpn_service.py b/evpn_connector/tests/unit/test_evpn_service.py index 7f41dd7..d7c9caf 100644 --- a/evpn_connector/tests/unit/test_evpn_service.py +++ b/evpn_connector/tests/unit/test_evpn_service.py @@ -17,6 +17,7 @@ import json import mock import os +import pytest import shutil import tempfile @@ -42,10 +43,10 @@ def test_rt2lst_as_local_ovveride(self): == expected ) - def setup(self): + def setup_method(self, method): self.temp_dir = tempfile.mkdtemp() - def teardown(self): + def teardown_method(self, method): shutil.rmtree(self.temp_dir) def test_read_client_configs_with_no_folder(self): @@ -311,3 +312,190 @@ def test_read_l3_anycast_client_configs(self): assert expected_pr in res_pr assert expected_any_pr1 in res_pr assert expected_any_pr2 in res_pr + + +class TestFailStatic(object): + def _make_service(self, fail_static=True, fail_static_min_peers=0): + return evpn.EvpnConnectorService( + source_ip="10.10.10.1", + as_number=1, + configs_dir="", + gobgp_client=mock.MagicMock(), + ovs_client=mock.MagicMock(), + sender=mock.MagicMock(), + vxlan_udp_port=4789, + router_mac_type5="11:22:33:44:55:66", + anycast_status_file="/tmp/anycast_status_file", + anycast_check_ofport=65277, + anycast_check_mac="12:34:56:78:90:aa", + fail_static=fail_static, + fail_static_min_peers=fail_static_min_peers, + ) + + def test_peers_healthy_all_established(self): + service = self._make_service() + service.gobgp_client.count_peers.return_value = (2, 2) + + assert service._upstream_peers_healthy() is True + + def test_peers_degraded_no_peers(self): + """An empty peer list is a restarted gobgp, not a healthy node. + + It is indistinguishable from every session having been lost, and + it comes with an empty RIB, which is precisely the set of flows + that must not be trusted for deletion. + """ + service = self._make_service() + service.gobgp_client.count_peers.return_value = (0, 0) + + assert service._upstream_peers_healthy() is False + + def test_apply_without_a_snapshot_changes_nothing(self): + """A node that never had peers is not held back by fail-static.""" + service = self._make_service() + service.gobgp_client.count_peers.return_value = (0, 0) + metrics = {} + + result = service._apply_fail_static({"flow-a"}, metrics) + + assert result == {"flow-a"} + assert metrics["fail_static_retained_cnt"] == 0 + + def test_peers_degraded_partial(self): + service = self._make_service() + service.gobgp_client.count_peers.return_value = (2, 1) + + assert service._upstream_peers_healthy() is False + + def test_min_peers_allows_a_partial_fabric(self): + service = self._make_service(fail_static_min_peers=1) + service.gobgp_client.count_peers.return_value = (3, 1) + + assert service._upstream_peers_healthy() is True + + def test_min_peers_still_degrades_below_the_threshold(self): + service = self._make_service(fail_static_min_peers=2) + service.gobgp_client.count_peers.return_value = (3, 1) + + assert service._upstream_peers_healthy() is False + + def test_min_peers_above_configured_means_all_of_them(self): + service = self._make_service(fail_static_min_peers=5) + service.gobgp_client.count_peers.return_value = (2, 2) + + assert service._upstream_peers_healthy() is True + + def test_unreadable_peers_raise(self): + """A local gobgp fault is an error, not a held state. + + It is indistinguishable from a broken fabric, and the step that + raises never reaches sync_flows, so flows are retained anyway. + """ + service = self._make_service() + service.gobgp_client.count_peers.side_effect = RuntimeError("boom") + + with pytest.raises(RuntimeError): + service._upstream_peers_healthy() + + def test_apply_raises_and_keeps_the_snapshot_on_error(self): + service = self._make_service() + metrics = {} + service.gobgp_client.count_peers.return_value = (1, 1) + service._apply_fail_static({"flow-a"}, metrics) + service.gobgp_client.count_peers.side_effect = RuntimeError("boom") + + with pytest.raises(RuntimeError): + service._apply_fail_static({"flow-b"}, metrics) + + assert service._last_good_flows == {"flow-a"} + + def test_apply_healthy_snapshots_and_passes_through(self): + service = self._make_service() + service.gobgp_client.count_peers.return_value = (1, 1) + metrics = {} + target = {"flow-a", "flow-b"} + + result = service._apply_fail_static(target, metrics) + + assert result == target + assert service._last_good_flows == target + assert metrics["fail_static_active"] == 0 + assert metrics["fail_static_retained_cnt"] == 0 + + def test_apply_degraded_retains_last_good(self): + service = self._make_service() + metrics = {} + # Healthy step snapshots two flows + service.gobgp_client.count_peers.return_value = (1, 1) + service._apply_fail_static({"flow-a", "flow-remote"}, metrics) + # Peer lost: remote flow vanished from the computed set + service.gobgp_client.count_peers.return_value = (1, 0) + + result = service._apply_fail_static({"flow-a"}, metrics) + + assert result == {"flow-a", "flow-remote"} + # Snapshot must not be overwritten by the degraded set + assert service._last_good_flows == {"flow-a", "flow-remote"} + assert metrics["fail_static_active"] == 1 + assert metrics["fail_static_retained_cnt"] == 1 + + def test_apply_degraded_still_applies_local_changes(self): + service = self._make_service() + metrics = {} + service.gobgp_client.count_peers.return_value = (1, 1) + service._apply_fail_static({"flow-remote"}, metrics) + service.gobgp_client.count_peers.return_value = (1, 0) + + result = service._apply_fail_static({"flow-new-local"}, metrics) + + assert result == {"flow-remote", "flow-new-local"} + + def test_apply_recovery_drops_stale_flows(self): + service = self._make_service() + metrics = {} + service.gobgp_client.count_peers.return_value = (1, 1) + service._apply_fail_static({"flow-a", "flow-remote"}, metrics) + service.gobgp_client.count_peers.return_value = (1, 0) + service._apply_fail_static({"flow-a"}, metrics) + # Peer is back; RIB is authoritative again + service.gobgp_client.count_peers.return_value = (1, 1) + + result = service._apply_fail_static({"flow-a"}, metrics) + + assert result == {"flow-a"} + assert service._last_good_flows == {"flow-a"} + assert metrics["fail_static_active"] == 0 + + def test_apply_degraded_keeps_the_fresh_action_for_a_known_match(self): + """What makes retention safe: a stale flow never wins a match. + + Flows are equal by match alone, and a union keeps the side it + started from, so a match that is still computed is applied with + its current action and only matches that vanished are retained. + """ + service = self._make_service() + metrics = {} + stale = objects.OvsFlow( + match="table=0,priority=10,in_port=1", action="action=output:5" + ) + service.gobgp_client.count_peers.return_value = (1, 1) + service._apply_fail_static({stale}, metrics) + fresh = objects.OvsFlow( + match="table=0,priority=10,in_port=1", action="action=output:9" + ) + service.gobgp_client.count_peers.return_value = (1, 0) + + result = service._apply_fail_static({fresh}, metrics) + + assert [flow.to_string() for flow in result] == [fresh.to_string()] + assert metrics["fail_static_retained_cnt"] == 0 + + def test_disabled_passes_through_when_degraded(self): + service = self._make_service(fail_static=False) + metrics = {} + service.gobgp_client.count_peers.return_value = (1, 0) + + result = service._apply_fail_static({"flow-a"}, metrics) + + assert result == {"flow-a"} + service.gobgp_client.count_peers.assert_not_called() diff --git a/evpn_connector/tests/unit/test_ovs_client.py b/evpn_connector/tests/unit/test_ovs_client.py index 1fc51a3..7d3fc22 100644 --- a/evpn_connector/tests/unit/test_ovs_client.py +++ b/evpn_connector/tests/unit/test_ovs_client.py @@ -88,10 +88,59 @@ def test_create_tun_port(self, mock_shell_run): 'options:local_ip="%s"' % local_ip, "options:dst_port=%d" % vxlan_udp_port, "ofport_request=%s" % vxlan_ofport, + "--", + "remove", + "Interface", + "vxlan_out", + "options", + "exts", ], enable_sudo=False, ) + def test_create_tun_port_with_gbp(self, mock_shell_run): + """With gbp on, the tunnel gains the extension and nothing else. + + OVS will not mix GBP and non-GBP tunnels on one UDP port, so this is + a property of a whole fabric, which is why it is opt-in. + """ + ovs_client = client.OvSClient( + "test_sw", "/tmp/flows.txt", vxlan_ofport=10, gbp=True + ) + + ovs_client.create_tun_port( + vxlan_source_ip="1.2.3.4", vxlan_udp_port=3423 + ) + + command = mock_shell_run.call_args[1]["command"] + assert "options:exts=gbp" in command + assert "remove" not in command + + def test_create_tun_port_without_gbp_unmakes_it(self, mock_shell_run): + """An existing tunnel is taken back, or the flag is one-way. + + The port outlives the daemon: --may-exist means turning gbp off + would otherwise leave a GBP tunnel behind forever. + """ + ovs_client = client.OvSClient( + "test_sw", "/tmp/flows.txt", vxlan_ofport=10, gbp=False + ) + + ovs_client.create_tun_port( + vxlan_source_ip="1.2.3.4", vxlan_udp_port=3423 + ) + + command = mock_shell_run.call_args[1]["command"] + assert "options:exts=gbp" not in command + assert command[-6:] == [ + "--", + "remove", + "Interface", + "vxlan_out", + "options", + "exts", + ] + def test_sync_flows(self, mock_shell_run): file_name = "/tmp/flows.txt" switch_name = "test_sw" diff --git a/requirements.txt b/requirements.txt index 7ad7ae3..0e972e8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,16 +1,19 @@ -pbr>=1.10.0,<=5.8.1 # Apache-2.0 +pbr>=1.10.0,<=5.8.1 # Apache-2.0 (capped by loopster) +setuptools<81 # MIT (pbr 5.8.1 imports pkg_resources, gone in 81) six>=1.9.0,<=1.16.0 # MIT -oslo.config==3.22.0 # Apache-2.0 -#grpcio===1.41.1 # Apache-2.0 -grpcio===1.22.0; python_version<'3.8' # Apache-2.0 -grpcio===1.26.0; python_version>='3.8' # Apache-2.0 -protobuf===3.9.0; python_version<'3.8' # BSD -protobuf===3.14.0; python_version>='3.8' # BSD -netaddr>=0.7.18,<=0.8.0 # BSD +oslo.config==3.22.0; python_version<'3.10' # Apache-2.0 +oslo.config>=9.4.0,<11.0.0; python_version>='3.10' # Apache-2.0 +grpcio===1.39.0; python_version<'3.10' # Apache-2.0 +grpcio>=1.83.0,<2.0.0; python_version>='3.10' # Apache-2.0 +protobuf===3.17.3; python_version<'3.10' # BSD +protobuf>=7.35.1,<8.0.0; python_version>='3.10' # BSD +netaddr>=0.7.18,<=0.8.0; python_version<'3.10' # BSD +netaddr>=1.0.0,<2.0.0; python_version>='3.10' # BSD enum34===1.1.6; python_version<'3.8' subprocess32===3.2.6;python_version=='2.7' # PSF sentry-sdk==1.5.2;python_version<'3.8' # BSD -sentry-sdk==1.6.0;python_version>='3.8' # BSD +sentry-sdk==1.6.0;python_version>='3.8' and python_version<'3.10' # BSD +sentry-sdk>=2.0.0,<3.0.0; python_version>='3.10' # BSD loopster>=2.14.4,<3.0.0 # Apache-2.0 obsender>=5.0.0,<6.0.0 # Apache-2.0 pyyaml>=6.0 # MIT diff --git a/test-requirements.txt b/test-requirements.txt index 8f12f67..b1f8ae4 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1,7 +1,9 @@ hacking<3 # Apache-2.0 typing-extensions<4.2.0;python_version=='3.6' # PSF-2.0 pytest-timer -pytest==4.6.1 +pytest==4.6.1; python_version<'3.10' +pytest>=8.0.0; python_version>='3.10' coverage>=4.0 # Apache-2.0 -mock==3.0.5 +mock==3.0.5; python_version<'3.10' +mock>=5.0.0; python_version>='3.10' flake8>=3.8.4 # Flake8 License (MIT) diff --git a/tox.ini b/tox.ini index 9dc635e..2352bff 100644 --- a/tox.ini +++ b/tox.ini @@ -1,6 +1,6 @@ [tox] envlist = pep8 - begin,py27,py38,py313,end + begin,py27,py38,py313,py314,end skipsdist = True minversion = 2.0 @@ -14,7 +14,7 @@ deps = -r{toxinidir}/requirements.txt setenv = PYTHONDONTWRITEBYTECODE = 1 commands = - py27,py38: coverage run -m pytest {posargs} --timer-top-n=10 {[base]project_name}/tests/unit + py27,py38,py313,py314: coverage run -m pytest {posargs} --timer-top-n=10 {[base]project_name}/tests/unit functional: pytest {posargs} --timer-top-n=10 {[base]project_name}/tests/functional [testenv:pep8]