mgr/rook: Implement fallback for CentOS 7

`kubernetes-client` requires `urllib3` > 1.15, but CentOS 7 ships 1.10

Signed-off-by: Sebastian Wagner <sebastian.wagner@suse.com>

diff --git a/src/pybind/mgr/rook/rook_cluster.py b/src/pybind/mgr/rook/rook_cluster.py
index 3ffbecd5271259cbfc5e4a6a1fa8e8e62578c3cc..adc508d89a5790ee765a61adb80f51f8a765a0e5 100644 (file)
--- a/src/pybind/mgr/rook/rook_cluster.py
+++ b/src/pybind/mgr/rook/rook_cluster.py
@@ -15,6 +15,8 @@ from six.moves.urllib.parse import urljoin  # pylint: disable=import-error
 
 # Optional kubernetes imports to enable MgrModule.can_run
 # to behave cleanly.
+from urllib3.exceptions import ProtocolError
+
 from mgr_util import merge_dicts
 
 try:
@@ -37,6 +39,18 @@ except ImportError:
 log = logging.getLogger(__name__)
 
 
+def _urllib3_supports_read_chunked():
+    # There is a bug in CentOS 7 as it ships a urllib3 which is lower
+    # than required by kubernetes-client
+    try:
+        from urllib3.response import HTTPResponse
+        return hasattr(HTTPResponse, 'read_chunked')
+    except ImportError:
+        return False
+
+
+_urllib3_supports_read_chunked = _urllib3_supports_read_chunked()
+
 class ApplyException(orchestrator.OrchestratorError):
     """
     For failures to update the Rook CRDs, usually indicating
@@ -73,6 +87,8 @@ class KubernetesResource(object):
         self._items = dict()
         self.thread = None  # type: threading.Thread
         self.exception = None
+        if not _urllib3_supports_read_chunked:
+            logging.info('urllib3 is too old. Fallback to full fetches')
 
     def _fetch(self):
         """ Execute the requested api method as a one-off fetch"""
@@ -95,10 +111,11 @@ class KubernetesResource(object):
             self.exception = None
             raise e  # Propagate the exception to the user.
         if not self.thread or not self.thread.is_alive():
-            # Start a thread which will use the kubernetes watch client against a resource
             resource_version = self._fetch()
-            log.debug("Attaching resource watcher for k8s {}".format(self.api_func))
-            self.thread = self._watch(resource_version)  # type: threading.Thread
+            if _urllib3_supports_read_chunked:
+                # Start a thread which will use the kubernetes watch client against a resource
+                log.debug("Attaching resource watcher for k8s {}".format(self.api_func))
+                self.thread = self._watch(resource_version)  # type: threading.Thread
 
         return self._items.values()
 
@@ -135,16 +152,15 @@ class KubernetesResource(object):
                     raise ApiException(str(event))
                 else:
                     raise KeyError('Unknown watch event {}'.format(event['type']))
-
+        except ProtocolError as e:
+            if 'Connection broken' in str(e):
+                log.info('Connection reset.')
+                return
+            raise
         except ApiException as e:
             log.exception('K8s API failed. {}'.format(self.api_func))
             self.exception = e
             raise
-        except AttributeError as e:
-            log.exception(
-                "Unable to attach watcher - incompatible urllib3? ({})".format(self.api_func))
-            self.exception = e
-            raise
         except Exception as e:
             log.exception("Watcher failed. ({})".format(self.api_func))
             self.exception = e