fix watching with a specified resource version

The watch code reset the version to the last found in the
response.
When you first list existing objects and then start watching from that
resource version the existing versions are older than the version you
wanted and the watch starts from the wrong version after the first
restart.
This leads to for example already deleted objects ending in the stream
again.

Fix this by not resetting to an older version than the one specified in
the watch.
It does not handle overflows of the resource version but they are 64 bit
integers so they should not realistically overflow even in the most loaded
clusters.

Closes kubernetes-client/python#700

Author: juliantaylor

watch/watch.py

@@ -83,13 +83,14 @@ def unmarshal_event(self, data, return_type):
             obj = SimpleNamespace(data=json.dumps(js['raw_object']))
             js['object'] = self._api_client.deserialize(obj, return_type)
             if hasattr(js['object'], 'metadata'):
-                self.resource_version = js['object'].metadata.resource_version
+                self.resource_version = int(
+                    js['object'].metadata.resource_version)
             # For custom objects that we don't have model defined, json
             # deserialization results in dictionary
             elif (isinstance(js['object'], dict) and 'metadata' in js['object']
                   and 'resourceVersion' in js['object']['metadata']):
-                self.resource_version = js['object']['metadata'][
-                    'resourceVersion']
+                self.resource_version = int(
+                    js['object']['metadata']['resourceVersion'])
         return js
 
     def stream(self, func, *args, **kwargs):
@@ -122,6 +123,7 @@ def stream(self, func, *args, **kwargs):
         return_type = self.get_return_type(func)
         kwargs['watch'] = True
         kwargs['_preload_content'] = False
+        min_resource_version = int(kwargs.get('resource_version', 0))
 
         timeouts = ('timeout_seconds' in kwargs)
         while True:
@@ -132,7 +134,13 @@ def stream(self, func, *args, **kwargs):
                     if self._stop:
                         break
             finally:
-                kwargs['resource_version'] = self.resource_version
+                # if the existing objects are older than the requested version
+                # continue to watch from the requested resource version
+                # does not handle overflow though that should take a few
+                # hundred years
+                kwargs['resource_version'] = max(
+                    self.resource_version, min_resource_version
+                )
                 resp.close()
                 resp.release_conn()
 

watch/watch_test.py

@@ -49,7 +49,7 @@ def test_watch_with_decode(self):
             # make sure decoder worked and updated Watch.resource_version
             self.assertEqual(
                 "%d" % count, e['object'].metadata.resource_version)
-            self.assertEqual("%d" % count, w.resource_version)
+            self.assertEqual(count, w.resource_version)
             count += 1
             # make sure we can stop the watch and the last event with won't be
             # returned
@@ -62,6 +62,42 @@ def test_watch_with_decode(self):
         fake_resp.close.assert_called_once()
         fake_resp.release_conn.assert_called_once()
 
+    def test_watch_resource_version_set(self):
+        #
+        fake_resp = Mock()
+        fake_resp.close = Mock()
+        fake_resp.release_conn = Mock()
+        values = [
+            '{"type": "ADDED", "object": {"metadata": {"name": "test1",'
+            '"resourceVersion": "1"}, "spec": {}, "status": {}}}\n',
+            '{"type": "ADDED", "object": {"metadata": {"name": "test2",'
+            '"resourceVersion": "2"}, "spec": {}, "sta',
+            'tus": {}}}\n'
+            '{"type": "ADDED", "object": {"metadata": {"name": "test3",'
+            '"resourceVersion": "3"}, "spec": {}, "status": {}}}\n'
+        ]
+        fake_resp.read_chunked = Mock(
+            return_value=values)
+
+        fake_api = Mock()
+        fake_api.get_namespaces = Mock(return_value=fake_resp)
+        fake_api.get_namespaces.__doc__ = ':return: V1NamespaceList'
+
+        w = Watch()
+        count = 1
+        # ensure we keep our requested resource version when the existing
+        # versions are older than the requested version
+        # needed for the list existing objects, then watch from there use case
+        for e in w.stream(fake_api.get_namespaces, resource_version=5):
+            count += 1
+            if count % 3 == 0:
+                fake_api.get_namespaces.assert_called_once_with(
+                    _preload_content=False, watch=True, resource_version=5)
+                fake_api.get_namespaces.reset_mock()
+            # returned
+            if count == len(values) * 3:
+                w.stop()
+
     def test_watch_stream_twice(self):
         w = Watch(float)
         for step in ['first', 'second']:
@@ -148,7 +184,7 @@ def test_unmarshal_with_custom_object(self):
         # Watch.resource_version
         self.assertTrue(isinstance(event['object'], dict))
         self.assertEqual("1", event['object']['metadata']['resourceVersion'])
-        self.assertEqual("1", w.resource_version)
+        self.assertEqual(1, w.resource_version)
 
     def test_watch_with_exception(self):
         fake_resp = Mock()