From 18dc684896dd4c0e15434b06b53f2afa901adb8a Mon Sep 17 00:00:00 2001

From: Tomas Jelinek <tojeline@redhat.com>

Date: Mon, 12 Jan 2015 15:30:33 +0100

Subject: [PATCH] stop cluster nodes in parallel

---

 pcs/cluster.py | 78 +++++++++++++++++++++++++++++++++++++++++++++++++---------

 pcs/pcs.py     | 19 +++++++++++++-

 pcs/utils.py   | 40 ++++++++++++++++++++++++------

 pcsd/pcs.rb    |  4 +--

 pcsd/pcsd.rb   |  2 +-

 pcsd/remote.rb | 19 +++++++++++---

 6 files changed, 137 insertions(+), 25 deletions(-)

diff --git a/pcs/cluster.py b/pcs/cluster.py

index 9730e55..5e94389 100644

--- a/pcs/cluster.py

+++ b/pcs/cluster.py

@@ -18,6 +18,7 @@ import datetime

 import commands

 import json

 import xml.dom.minidom

+import threading

 pcs_dir = os.path.dirname(os.path.realpath(__file__))

 COROSYNC_CONFIG_TEMPLATE = pcs_dir + "/corosync.conf.template"

@@ -125,9 +126,7 @@ def sync_start(partial_argv, nodes):

     for node in nodes:

         utils.setCorosyncConfig(node,config)

     print "Starting cluster on nodes: " + ", ".join(nodes) + "..."

-

-    for node in nodes:

-        utils.startCluster(node)

+    start_cluster_nodes(nodes)

 def sync(partial_argv,nodes):

     argv = partial_argv[:]

@@ -536,16 +535,29 @@ def start_cluster_all():

     start_cluster_nodes(utils.getNodesFromCorosyncConf())

 def start_cluster_nodes(nodes):

-    error_list = utils.map_for_error_list(utils.startCluster, nodes)

-    if len(error_list) > 0:

+    threads = dict()

+    for node in nodes:

+        threads[node] = NodeStartThread(node)

+    error_list = utils.run_node_threads(threads)

+    if error_list:

         utils.err("unable to start all nodes\n" + "\n".join(error_list))

 def stop_cluster_all():

     stop_cluster_nodes(utils.getNodesFromCorosyncConf())

 def stop_cluster_nodes(nodes):

-    error_list = utils.map_for_error_list(utils.stopCluster, nodes)

-    if len(error_list) > 0:

+    threads = dict()

+    for node in nodes:

+        threads[node] = NodeStopPacemakerThread(node)

+    error_list = utils.run_node_threads(threads)

+    if error_list:

+        utils.err("unable to stop all nodes\n" + "\n".join(error_list))

+

+    threads = dict()

+    for node in nodes:

+        threads[node] = NodeStopCorosyncThread(node)

+    error_list = utils.run_node_threads(threads)

+    if error_list:

         utils.err("unable to stop all nodes\n" + "\n".join(error_list))

 def node_standby(argv,standby=True):

@@ -610,27 +622,44 @@ def disable_cluster_nodes(nodes):

 def destroy_cluster(argv):

     if len(argv) > 0:

-        error_list = utils.map_for_error_list(utils.destroyCluster, argv)

-        if len(error_list) > 0:

+        threads = dict()

+        for node in argv:

+            threads[node] = NodeDestroyThread(node)

+        error_list = utils.run_node_threads(threads)

+        if error_list:

             utils.err("unable to destroy cluster\n" + "\n".join(error_list))

-        return

 def stop_cluster(argv):

     if len(argv) > 0:

         stop_cluster_nodes(argv)

         return

-    print "Stopping Cluster..."

+    stop_all = (

+        "--pacemaker" not in utils.pcs_options

+        and

+        "--corosync" not in utils.pcs_options

+    )

+    if stop_all or "--pacemaker" in utils.pcs_options:

+        stop_cluster_pacemaker()

+    if stop_all or "--corosync" in utils.pcs_options:

+        stop_cluster_corosync()

+

+def stop_cluster_pacemaker():

+    print "Stopping Cluster (pacemaker)...",

     output, retval = utils.run(["service", "pacemaker","stop"])

     if retval != 0:

         print output,

         utils.err("unable to stop pacemaker")

+

+def stop_cluster_corosync():

     if utils.is_rhel6():

+        print "Stopping Cluster (cman)...",

         output, retval = utils.run(["service", "cman","stop"])

         if retval != 0:

             print output,

             utils.err("unable to stop cman")

     else:

+        print "Stopping Cluster (corosync)...",

         output, retval = utils.run(["service", "corosync","stop"])

         if retval != 0:

             print output,

@@ -1200,3 +1229,30 @@ def cluster_quorum_unblock(argv):

     utils.set_cib_property("startup-fencing", startup_fencing)

     print "Waiting for nodes cancelled"

+class NodeActionThread(threading.Thread):

+    def __init__(self, node):

+        super(NodeActionThread, self).__init__()

+        self.node = node

+        self.retval = 0

+        self.output = ""

+

+class NodeStartThread(NodeActionThread):

+    def run(self):

+        self.retval, self.output = utils.startCluster(self.node, quiet=True)

+

+class NodeStopPacemakerThread(NodeActionThread):

+    def run(self):

+        self.retval, self.output = utils.stopCluster(

+            self.node, quiet=True, pacemaker=True, corosync=False

+        )

+

+class NodeStopCorosyncThread(NodeActionThread):

+    def run(self):

+        self.retval, self.output = utils.stopCluster(

+            self.node, quiet=True, pacemaker=False, corosync=True

+        )

+

+class NodeDestroyThread(NodeActionThread):

+    def run(self):

+        self.retval, self.output = utils.destroyCluster(self.node, quiet=True)

+

diff --git a/pcs/pcs.py b/pcs/pcs.py

index b2c3f4b..a0c0df5 100755

--- a/pcs/pcs.py

+++ b/pcs/pcs.py

@@ -54,7 +54,24 @@ def main(argv):

                 pcs_short_options_with_args.append(prev_char)

             prev_char = c

-        pcs_long_options = ["local","start","all","clone","master","force","corosync_conf=", "defaults","debug","version","help","fullhelp","off","from=","to=", "name=", "wait", "group=","groups","full","enable","node=","nodesc","transport=", "addr0=","addr1=","bcast0=","bcast1=","mcast0=","mcast1=","mcastport0=","mcastport1=","ttl0=","ttl1=","rrpmode=", "broadcast0", "broadcast1","wait_for_all=","auto_tie_breaker=","last_man_standing=", "last_man_standing_window=","no-default-ops","ipv6","token=", "token_coefficient=", "consensus=", "miss_count_const=", "fail_recv_const=","join=", "disabled", "after=", "before=", "autocorrect", "interactive", "autodelete"]

+        pcs_long_options = [

+            "debug", "version", "help", "fullhelp",

+            "force", "autocorrect", "interactive", "autodelete",

+            "all", "full", "groups", "local", "wait", "config",

+            "start", "enable", "disabled", "off",

+            "pacemaker", "corosync",

+            "no-default-ops", "defaults", "nodesc",

+            "clone", "master", "name=", "group=", "node=",

+            "from=", "to=", "after=", "before=",

+            "transport=", "rrpmode=", "ipv6",

+            "addr0=", "bcast0=", "mcast0=", "mcastport0=", "ttl0=", "broadcast0",

+            "addr1=", "bcast1=", "mcast1=", "mcastport1=", "ttl1=", "broadcast1",

+            "wait_for_all=", "auto_tie_breaker=", "last_man_standing=",

+            "last_man_standing_window=",

+            "token=", "token_coefficient=", "consensus=", "join=",

+            "miss_count_const=", "fail_recv_const=",

+            "corosync_conf=", "cluster_conf=",

+        ]

         # pull out negative number arguments and add them back after getopt

         prev_arg = ""

         for arg in argv:

diff --git a/pcs/utils.py b/pcs/utils.py

index 724519a..1f41ae0 100644

--- a/pcs/utils.py

+++ b/pcs/utils.py

@@ -208,11 +208,19 @@ def setCorosyncConfig(node,config):

         if status != 0:

             err("Unable to set corosync config")

-def startCluster(node):

-    return sendHTTPRequest(node, 'remote/cluster_start', None, False, True)

-

-def stopCluster(node):

-    return sendHTTPRequest(node, 'remote/cluster_stop', None, False, True)

+def startCluster(node, quiet=False):

+    return sendHTTPRequest(node, 'remote/cluster_start', None, False, not quiet)

+

+def stopCluster(node, quiet=False, pacemaker=True, corosync=True):

+    if (pacemaker and corosync) or (not pacemaker and not corosync):

+        data = None

+    elif pacemaker:

+        data = {"component": "pacemaker"}

+    elif corosync:

+        data = {"component": "corosync"}

+    if data:

+        data = urllib.urlencode(data)

+    return sendHTTPRequest(node, 'remote/cluster_stop', data, False, not quiet)

 def enableCluster(node):

     return sendHTTPRequest(node, 'remote/cluster_enable', None, False, True)

@@ -220,8 +228,8 @@ def enableCluster(node):

 def disableCluster(node):

     return sendHTTPRequest(node, 'remote/cluster_disable', None, False, True)

-def destroyCluster(node):

-    return sendHTTPRequest(node, 'remote/cluster_destroy')

+def destroyCluster(node, quiet=False):

+    return sendHTTPRequest(node, 'remote/cluster_destroy', None, not quiet, not quiet)

 def restoreConfig(node, tarball_data):

     data = urllib.urlencode({"tarball": tarball_data})

@@ -730,6 +738,24 @@ def map_for_error_list(callab, iterab):

             error_list.append(err)

     return error_list

+def run_node_threads(node_threads):

+    error_list = []

+    for node, thread in node_threads.items():

+        thread.daemon = True

+        thread.start()

+    while node_threads:

+        for node in node_threads.keys():

+            thread = node_threads[node]

+            thread.join(1)

+            if thread.is_alive():

+                continue

+            output = node + ": " + thread.output.strip()

+            print output

+            if thread.retval != 0:

+                error_list.append(output)

+            del node_threads[node]

+    return error_list

+

 # Check is something exists in the CIB, if it does return it, if not, return

 #  an empty string

 def does_exist(xpath_query):

diff --git a/pcsd/pcs.rb b/pcsd/pcs.rb

index a1acfdc..3fad833 100644

--- a/pcsd/pcs.rb

+++ b/pcsd/pcs.rb

@@ -283,7 +283,7 @@ def send_cluster_request_with_token(cluster_name, request, post=false, data={},

   return code,out

 end

-def send_request_with_token(node,request, post=false, data={}, remote=true, raw_data = nil)

+def send_request_with_token(node, request, post=false, data={}, remote=true, raw_data=nil, timeout=30)

   start = Time.now

   begin

     retval, token = get_node_token(node)

@@ -312,7 +312,7 @@ def send_request_with_token(node,request, post=false, data={}, remote=true, raw_

     myhttp.use_ssl = true

     myhttp.verify_mode = OpenSSL::SSL::VERIFY_NONE

     res = myhttp.start do |http|

-      http.read_timeout = 30 

+      http.read_timeout = timeout

       http.request(req)

     end

     return res.code.to_i, res.body

diff --git a/pcsd/pcsd.rb b/pcsd/pcsd.rb

index c653ae2..94fdae2 100644

--- a/pcsd/pcsd.rb

+++ b/pcsd/pcsd.rb

@@ -428,7 +428,7 @@ if not DISABLE_GUI

     }

     $logger.info("Sending setup cluster request for: " + @cluster_name + " to: " + @nodes[0])

-    code,out = send_request_with_token(@nodes[0], "setup_cluster", true, {:clustername => @cluster_name, :nodes => @nodes_rrp.join(';'), :options => options.to_json})

+    code,out = send_request_with_token(@nodes[0], "setup_cluster", true, {:clustername => @cluster_name, :nodes => @nodes_rrp.join(';'), :options => options.to_json}, true, nil, 60)

     if code == 200

       pcs_config.clusters << Cluster.new(@cluster_name, @nodes)

diff --git a/pcsd/remote.rb b/pcsd/remote.rb

index 9709941..2245d47 100644

--- a/pcsd/remote.rb

+++ b/pcsd/remote.rb

@@ -151,10 +151,23 @@ end

 def cluster_stop(params)

   if params[:name]

-    code, response = send_request_with_token(params[:name], 'cluster_stop', true)

+    params_without_name = params.reject {|key, value|

+      key == "name" or key == :name

+    }

+    code, response = send_request_with_token(

+      params[:name], 'cluster_stop', true, params_without_name

+    )

   else

-    $logger.info "Starting Daemons"

-    output =  `#{PCS} cluster stop`

+    options = ""

+    if params.has_key?("component")

+      if params["component"].downcase == "pacemaker"

+        options = "--pacemaker"

+      elsif params["component"].downcase == "corosync"

+        options = "--corosync"

+      end

+    end

+    $logger.info "Stopping Daemons #{options}"

+    output =  `#{PCS} cluster stop #{options}`

     $logger.debug output

     return output

   end

-- 

1.9.1