From 6c618459083062830eff6b53d3e7d3445a86a7b9 Mon Sep 17 00:00:00 2001 From: Bartosz Bezak Date: Fri, 8 May 2026 16:59:25 +0200 Subject: [PATCH] batching: Fix missed batch result watches A batch result can be written to etcd while the waiting caller still does not receive the watch event. This has been seen with Cumulus 5 trunk configuration using many VLAN subports, where the result key was written but the waiting caller still timed out waiting for the watch event. When this happens, the caller waits until the full batch timeout and fails even though the result key exists. Use shorter watch waits and check etcd between them so missed watch events do not cause false batch timeouts. Conflicts: networking_generic_switch/batching.py 2025.1 has no "import threading"; added only "import time". Assisted-By: openai codex gpt 5.5 Closes-Bug: #2152047 Change-Id: Ifa6b4a3ec447712c0ccf9fe0b79f8236be0ce33e Signed-off-by: Bartosz Bezak (cherry picked from commit 230667610a59312207baa7410a0cc7f69f2c3d89) --- networking_generic_switch/batching.py | 58 ++++++++++++++----- .../tests/unit/test_batching.py | 57 +++++++++++++++++- ...batch-result-watches-8f5929b1e0ff1c77.yaml | 7 +++ 3 files changed, 106 insertions(+), 16 deletions(-) create mode 100644 releasenotes/notes/fix-missed-batch-result-watches-8f5929b1e0ff1c77.yaml diff --git a/networking_generic_switch/batching.py b/networking_generic_switch/batching.py index 5ec3c8f2..2fb981ab 100644 --- a/networking_generic_switch/batching.py +++ b/networking_generic_switch/batching.py @@ -14,6 +14,7 @@ import atexit import json +import time import etcd3gw from etcd3gw import exceptions as etcd3gw_exc @@ -29,6 +30,7 @@ from networking_generic_switch import exceptions as exc SHUTDOWN_TIMEOUT = 60 +RESULT_WATCH_TIMEOUT = 10 LOG = logging.getLogger(__name__) @@ -152,21 +154,45 @@ def wait_for_result(self, item, timeout): unsuccessful """ result_key = self.RESULT_ITEM_KEY % (self.switch_name, item.uuid) - try: - event = self.client.watch_once(result_key, timeout=timeout, - start_revision=item.create_revision) - except etcd3gw_exc.WatchTimedOut: - raise exc.GenericSwitchBatchError( - device=self.switch_name, - error="Timed out waiting for result key: %s" % result_key) + deadline = time.monotonic() + timeout + result_dict = None + start_revision = item.create_revision + # NOTE(bbezak): A result can exist even if the watch event was missed. + while time.monotonic() < deadline: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + watch_timeout = min(RESULT_WATCH_TIMEOUT, remaining) + try: + event = self.client.watch_once( + result_key, timeout=watch_timeout, + start_revision=start_revision) + except etcd3gw_exc.WatchTimedOut: + result_dict = self._get_and_delete_result(result_key) + if result_dict is not None: + break + # NOTE(bbezak): The result key does not exist now; avoid + # retrying from a revision that may be compacted. + start_revision = None + LOG.debug("Result key not ready yet: %s", result_key) + continue + + LOG.debug("got event: %s", event) + if event["kv"]["version"] == 0: + raise exc.GenericSwitchBatchError( + device=self.switch_name, + error="Output key was deleted, perhaps lease expired") + # TODO(johngarbutt) check we have the create event and result? + result_dict = self._get_and_delete_result(result_key) + break + + if result_dict is None: + result_dict = self._get_and_delete_result(result_key) + if result_dict is None: + raise exc.GenericSwitchBatchError( + device=self.switch_name, + error="Timed out waiting for result key: %s" % result_key) - LOG.debug("got event: %s", event) - if event["kv"]["version"] == 0: - raise exc.GenericSwitchBatchError( - device=self.switch_name, - error="Output key was deleted, perhaps lease expired") - # TODO(johngarbutt) check we have the create event and result? - result_dict = self._get_and_delete_result(result_key) LOG.debug("got result: %s", result_dict) if "result" in result_dict: return result_dict["result"] @@ -176,7 +202,7 @@ def wait_for_result(self, item, timeout): error=result_dict["error"]) def _get_and_delete_result(self, result_key): - # called when watch event says the result key should exist + # Consume the result if it exists, otherwise return None. txn = { 'compare': [], 'success': [{ @@ -194,6 +220,8 @@ def _get_and_delete_result(self, result_key): device=self.switch_name, error="Unable to find result: %s" % result_key) delete_response = result['responses'][0]['response_delete_range'] + if not delete_response.get('prev_kvs'): + return None raw_value = delete_response['prev_kvs'][0]['value'] result_dict = json.loads(_decode(raw_value)) LOG.debug("fetched and deleted result for: %s", result_key) diff --git a/networking_generic_switch/tests/unit/test_batching.py b/networking_generic_switch/tests/unit/test_batching.py index 49c9f08f..5b6060cb 100644 --- a/networking_generic_switch/tests/unit/test_batching.py +++ b/networking_generic_switch/tests/unit/test_batching.py @@ -15,6 +15,7 @@ from unittest import mock from etcd3gw.exceptions import Etcd3Exception +from etcd3gw.exceptions import WatchTimedOut from etcd3gw.utils import _encode from etcd3gw.utils import _increment_last_byte import fixtures @@ -110,9 +111,49 @@ def test_wait_for_result(self, mock_get): self.assertEqual("result1", result) result_key = '/ngs/batch/switch1/output/uuid' self.client.watch_once.assert_called_once_with( - result_key, timeout=43, start_revision=42) + result_key, timeout=batching.RESULT_WATCH_TIMEOUT, + start_revision=42) mock_get.assert_called_once_with(self.queue, result_key) + @mock.patch.object(batching.SwitchQueue, "_get_and_delete_result", + autospec=True) + def test_wait_for_result_after_watch_timeout(self, mock_get): + self.client.watch_once.side_effect = WatchTimedOut + mock_get.return_value = {"result": "result1"} + item = batching.SwitchQueueItem("uuid", 42) + + result = self.queue.wait_for_result(item, 43) + + self.assertEqual("result1", result) + result_key = '/ngs/batch/switch1/output/uuid' + self.client.watch_once.assert_called_once_with( + result_key, timeout=batching.RESULT_WATCH_TIMEOUT, + start_revision=42) + mock_get.assert_called_once_with(self.queue, result_key) + + @mock.patch.object(batching.SwitchQueue, "_get_and_delete_result", + autospec=True) + def test_wait_for_result_after_multiple_watch_timeouts(self, mock_get): + self.client.watch_once.side_effect = WatchTimedOut + mock_get.side_effect = [None, {"result": "result1"}] + item = batching.SwitchQueueItem("uuid", 42) + + result = self.queue.wait_for_result(item, 43) + + self.assertEqual("result1", result) + result_key = '/ngs/batch/switch1/output/uuid' + first_call = mock.call( + result_key, timeout=batching.RESULT_WATCH_TIMEOUT, + start_revision=42) + second_call = mock.call( + result_key, timeout=batching.RESULT_WATCH_TIMEOUT, + start_revision=None) + self.client.watch_once.assert_has_calls([first_call, second_call]) + self.assertEqual(2, self.client.watch_once.call_count) + mock_get.assert_has_calls([mock.call(self.queue, result_key), + mock.call(self.queue, result_key)]) + self.assertEqual(2, mock_get.call_count) + def test_get_and_delete_result(self): self.client.transaction.return_value = { "succeeded": True, @@ -141,6 +182,20 @@ def test_get_and_delete_result(self): } self.client.transaction.assert_called_once_with(expected_txn) + def test_get_and_delete_result_missing(self): + self.client.transaction.return_value = { + "succeeded": True, + "responses": [{ + "response_delete_range": { + "prev_kvs": [] + } + }] + } + + result = self.queue._get_and_delete_result(b"result_key") + + self.assertIsNone(result) + def test_get_and_delete_result_failure(self): self.client.transaction.return_value = {"succeeded": False} diff --git a/releasenotes/notes/fix-missed-batch-result-watches-8f5929b1e0ff1c77.yaml b/releasenotes/notes/fix-missed-batch-result-watches-8f5929b1e0ff1c77.yaml new file mode 100644 index 00000000..2ef627c2 --- /dev/null +++ b/releasenotes/notes/fix-missed-batch-result-watches-8f5929b1e0ff1c77.yaml @@ -0,0 +1,7 @@ +--- +fixes: + - | + Fixes false batch result timeouts when the result key exists in etcd but + the watch event was missed. See `LP#2152047 + `__ for + details.