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.