batching: Fix missed batch result watches

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)

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}
 

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
+    <https://bugs.launchpad.net/networking-generic-switch/+bug/2152047>`__ for
+    details.