Tree: support offline gateways (#260)

With this change, WorkerTree is now able to safely repropagate commands
when the communication channel to a specific gateway cannot be
established at all.  Such offline gateways were already marked as
unreachable, but the 'relaunch' code was missing. A new
WorkerTree._relaunch() method has been added for that purpose.

Note that if a gateway fails once a propagation channel is "set up", the
remote commands cannot be safely repropagated at this point and so will
be seen as failed, but this was already the case.

Closes #260

Change-Id: I2be52b01638de95f58cedc0f2671f02cfffe8297

Author: thiell

lib/ClusterShell/CLI/Error.py

@@ -44,6 +44,7 @@
 from ClusterShell.NodeUtils import GroupSourceNoUpcall
 from ClusterShell.NodeSet import NodeSetExternalError, NodeSetParseError
 from ClusterShell.NodeSet import RangeSetParseError
+from ClusterShell.Propagation import RouteResolvingError
 from ClusterShell.Topology import TopologyError
 from ClusterShell.Worker.EngineClient import EngineClientError
 from ClusterShell.Worker.Worker import WorkerError
@@ -59,6 +60,7 @@
                   GroupResolverSourceError,
                   GroupSourceError,
                   GroupSourceNoUpcall,
+                  RouteResolvingError,
                   TopologyError,
                   TypeError,
                   IOError,
@@ -93,7 +95,7 @@ def handle_generic_error(excobj, prog=os.path.basename(sys.argv[0])):
         print(msgfmt % (prog, exc, exc.group_source.name), file=sys.stderr)
     except GroupSourceError as exc:
         print("%s: Group error: %s" % (prog, exc), file=sys.stderr)
-    except TopologyError as exc:
+    except (RouteResolvingError, TopologyError) as exc:
         print("%s: TREE MODE: %s" % (prog, exc), file=sys.stderr)
     except configparser.Error as exc:
         print("%s: %s" % (prog, exc), file=sys.stderr)

lib/ClusterShell/Communication.py

@@ -204,11 +204,10 @@ def _open(self):
     def _close(self):
         """close an already opened channel"""
         send_endtag = self.opened
-        # set to False before sending tag for state test purposes
-        self.opened = self.setup = False
         if send_endtag:
             XMLGenerator(self.worker, encoding=ENCODING).endElement('channel')
         self.worker.abort()
+        self.opened = self.setup = False
 
     def ev_start(self, worker):
         """connection established. Open higher level channel"""

lib/ClusterShell/Propagation.py

@@ -1,7 +1,7 @@
 #
 # Copyright (C) 2010-2016 CEA/DAM
 # Copyright (C) 2010-2011 Henri Doreau <[email protected]>
-# Copyright (C) 2015-2017 Stephane Thiell <[email protected]>
+# Copyright (C) 2015-2018 Stephane Thiell <[email protected]>
 #
 # This file is part of ClusterShell.
 #
@@ -41,6 +41,7 @@
 class RouteResolvingError(Exception):
     """error raised on invalid conditions during routing operations"""
 
+
 class PropagationTreeRouter(object):
     """performs routes resolving operations within a propagation tree.
     This object provides a next_hop method, that will look for the best
@@ -392,9 +393,13 @@ def ev_close(self, worker, timedout):
         self.logger.debug("ev_close gateway=%s %s", gateway, self)
         self.logger.debug("ev_close rc=%s", self._rc) # may be None
 
-        if self._rc: # got explicit error code
-            # ev_routing?
-            self.logger.debug("unreachable gateway %s", gateway)
-            worker.task.router.mark_unreachable(gateway)
-            self.logger.debug("worker.task.gateways=%s", worker.task.gateways)
-            # TODO: find best gateway, update TreeWorker counters, relaunch...
+        if self._rc:
+            self.logger.debug("error on gateway %s (setup=%s)", gateway,
+                              self.setup)
+            self.task.router.mark_unreachable(gateway)
+            self.logger.debug("gateway %s now set as unreachable", gateway)
+
+            if not self.setup:
+                # channel was not set up: we can safely repropagate commands
+                for mw in set(self.task.gateways[gateway][1]):
+                    mw._relaunch(gateway)

lib/ClusterShell/Worker/Tree.py

@@ -334,7 +334,7 @@ def _copy_remote(self, source, dest, targets, gateway, timeout, reverse):
 
         self._target_count += len(targets)
 
-        self.gwtargets[str(gateway)] = targets.copy()
+        self.gwtargets.setdefault(str(gateway), NodeSet()).add(targets)
 
         # tar commands are built here and launched on targets
         if reverse:
@@ -360,13 +360,32 @@ def _execute_remote(self, cmd, targets, gateway, timeout):
 
         self._target_count += len(targets)
 
-        self.gwtargets[str(gateway)] = targets.copy()
+        self.gwtargets.setdefault(str(gateway), NodeSet()).add(targets)
 
         pchan = self.task._pchannel(gateway, self)
         pchan.shell(nodes=targets, command=cmd, worker=self, timeout=timeout,
                     stderr=self.stderr, gw_invoke_cmd=self.invoke_gateway,
                     remote=self.remote)
 
+    def _relaunch(self, previous_gateway):
+        """Redistribute and relaunch commands on targets that were running
+        on previous_gateway (which is probably marked unreacheable by now)
+
+        NOTE: Relaunch is always called after failed remote execution, so
+        previous_gateway must be defined. However, it is not guaranted that
+        the relaunch is going to be performed using gateways (that's a feature).
+        """
+        targets = self.gwtargets[previous_gateway].copy()
+        self.logger.debug("_relaunch on targets %s from previous_gateway %s",
+                          targets, previous_gateway)
+
+        for target in targets:
+            self.gwtargets[previous_gateway].remove(target)
+
+        self._check_fini(previous_gateway)
+        self._target_count -= len(targets)
+        self._launch(targets)
+
     def _engine_clients(self):
         """
         Access underlying engine clients.

tests/TreeGatewayTest.py

@@ -157,8 +157,6 @@ def test_basic_noop(self):
 
         self.channel_send_stop()
         self.recvxml(EndMessage)
-        self.assertEqual(self.chan.opened, False)
-        self.assertEqual(self.chan.setup, False)
         # ending tag should abort gateway worker without delay
         self.gateway.wait()
         self.gateway.close()
@@ -179,8 +177,6 @@ def test_channel_err_dup(self):
 
         # gateway should terminate channel session
         msg = self.recvxml(EndMessage)
-        self.assertEqual(self.chan.opened, False)
-        self.assertEqual(self.chan.setup, False)
         self.gateway.wait()
         self.gateway.close()
 
@@ -217,10 +213,6 @@ def _check_channel_err(self, sendmsg, errback, openchan=True,
         else:
             self.recvxml()
 
-        # flags should be reset
-        self.assertEqual(self.chan.opened, False)
-        self.assertEqual(self.chan.setup, False)
-
         # gateway task should exit properly
         self.gateway.wait()
         self.gateway.close()

tests/TreeTaskTest.py

@@ -7,6 +7,7 @@
 from textwrap import dedent
 import unittest
 
+from ClusterShell.Propagation import RouteResolvingError
 from ClusterShell.Task import task_self
 from ClusterShell.Topology import TopologyError
 
@@ -33,14 +34,11 @@ def test_shell_auto_tree_dummy(self):
         task = task_self()
         task.set_default("auto_tree", True)
         task.TOPOLOGY_CONFIGS = [topofile.name]
-        task.run("/bin/hostname", nodes="dummy-node", stderr=True)
-        # FIXME gateway errors are not yet being handled correctly
+
+        self.assertRaises(RouteResolvingError, task.run, "/bin/hostname",
+                          nodes="dummy-node", stderr=True)
+        # FIXME: should probably be None
         self.assertEqual(task.max_retcode(), 255)
-        # XXX correct results would be:
-        #self.assertEqual(task.max_retcode(), None)
-        #expected = "Name or service not known"
-        #if not task.node_error("dummy-node").endswith(expected):
-        #    self.assertEqual(task.node_error("dummy-node"), expected)
 
     def test_shell_auto_tree_noconf(self):
         """test task shell auto tree [no topology.conf]"""