Author: aconway
Date: Thu Mar 1 19:47:08 2012
New Revision: 1295759
URL: http://svn.apache.org/viewvc?rev=1295759&view=rev
Log:
QPID-3603: Added "ready" command to qpid-ha, minor improvements.
Modified:
qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
qpid/trunk/qpid/tools/src/py/qpid-ha
qpid/trunk/qpid/tools/src/py/qpidtoollibs/broker.py
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp?rev=1295759&r1=1295758&r2=1295759&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp Thu Mar 1 19:47:08 2012
@@ -44,8 +44,10 @@ using namespace std;
namespace {
-const std::string PRIMARY="primary";
+const std::string STANDALONE="standalone";
+const std::string CATCH_UP="catch-up";
const std::string BACKUP="backup";
+const std::string PRIMARY="primary";
} // namespace
@@ -65,12 +67,12 @@ HaBroker::HaBroker(broker::Broker& b, co
ManagementAgent* ma = broker.getManagementAgent();
if (!ma)
throw Exception("Cannot start HA: management is disabled");
- if (ma) {
- _qmf::Package packageInit(ma);
- mgmtObject = new _qmf::HaBroker(ma, this, "ha-broker");
- mgmtObject->set_status(BACKUP);
- ma->addObject(mgmtObject);
- }
+ _qmf::Package packageInit(ma);
+ mgmtObject = new _qmf::HaBroker(ma, this, "ha-broker");
+ // FIXME aconway 2012-03-01: should start in catch-up state and move to
backup
+ // only when caught up.
+ mgmtObject->set_status(BACKUP);
+ ma->addObject(mgmtObject);
sys::Mutex::ScopedLock l(lock);
if (!settings.clientUrl.empty()) setClientUrl(Url(settings.clientUrl), l);
if (!settings.brokerUrl.empty()) setBrokerUrl(Url(settings.brokerUrl), l);
Modified: qpid/trunk/qpid/tools/src/py/qpid-ha
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/tools/src/py/qpid-ha?rev=1295759&r1=1295758&r2=1295759&view=diff
==============================================================================
--- qpid/trunk/qpid/tools/src/py/qpid-ha (original)
+++ qpid/trunk/qpid/tools/src/py/qpid-ha Thu Mar 1 19:47:08 2012
@@ -19,7 +19,7 @@
# under the License.
#
-import qmf.console, optparse, sys
+import qmf.console, optparse, sys, time
from qpid.management import managementChannel, managementClient
from qpid.messaging import Connection
from qpid.messaging import Message as QpidMessage
@@ -44,32 +44,26 @@ class Command:
self.op=optparse.OptionParser(usage)
self.op.add_option("-b", "--broker", metavar="<url>", help="Connect to
broker at <url>")
- def execute(self, command):
- opts, args = self.op.parse_args(command)
+ def execute(self):
+ opts, args = self.op.parse_args()
if len(args) != len(self.args)+1:
self.op.print_help()
- print "Error: wrong number of arguments"
- return
+ raise Exception("Wrong number of arguments")
broker = opts.broker or "localhost:5672"
connection = Connection.establish(broker,
client_properties={"qpid.ha-admin":1})
- try: self.do_execute(BrokerAgent(connection), opts, args)
+ qmf_broker = BrokerAgent(connection)
+ ha_broker = qmf_broker.getHaBroker()
+ if not ha_broker: raise Exception("HA module is not loaded on broker
at %s"%broker)
+ try: return self.do_execute(qmf_broker, ha_broker, opts, args)
finally: connection.close()
def do_execute(self, qmf_broker, opts, args):
raise Exception("Command '%s' is not yet implemented"%self.name)
-def print_all_help(name):
- print "usage: %s <command> [<arguments>]\n\nCommands are:\n"%name
- for c in Command.commands:
- help = Command.commands[c].help
- print " %-12s %s."%(c, help.split(".")[0])
- print "\nFor help with a command: %s <command> --help\n"%name
-
-
class PromoteCmd(Command):
def __init__(self):
Command.__init__(self, "promote","Promote broker from backup to
primary")
- def do_execute(self, qmf_broker, opts, args):
+ def do_execute(self, qmf_broker, ha_broker, opts, args):
qmf_broker._method("promote", {}, HA_BROKER)
PromoteCmd()
@@ -77,14 +71,27 @@ class ReadyCmd(Command):
def __init__(self):
Command.__init__(self, "ready", "Test if a backup broker is
ready.\nReturn 0 if broker is a ready backup, non-0 otherwise.")
self.op.add_option(
- "--wait", type="int", metavar="<seconds>",
+ "--wait", type="int", metavar="<seconds>", default=None,
help="Wait up to <seconds> for broker to be ready. 0 means wait
forever.")
+ def do_execute(self, qmf_broker, ha_broker, opts, args):
+ if (ha_broker.status == "backup"): return
+ if (ha_broker.status != "catch-up"):
+ raise Exception("Broker is not a backup, status is
'%s'"%ha_broker.status)
+ if (opts.wait is None): return 1
+ delay = 0.1
+ timeout = time.time() + opts.wait
+ while opts.wait == 0 or time.time() < timeout:
+ time.sleep(delay)
+ delay = min(2*delay, 1)
+ ha_broker = qmf_broker.getHaBroker()
+ if (ha_broker.status == "backup"): return
+ return 1
ReadyCmd()
class ReplicateCmd(Command):
def __init__(self):
Command.__init__(self, "replicate", "Set up replication from <queue>
on <remote-broker> to <queue> on the current broker.", ["<queue>",
"<remote-broker>"])
- def do_execute(self, qmf_broker, opts, args):
+ def do_execute(self, qmf_broker, ha_broker, opts, args):
qmf_broker._method("replicate", {"broker":args[1], "queue":args[2]},
HA_BROKER)
ReplicateCmd()
@@ -95,38 +102,55 @@ class SetCmd(Command):
self.op.add_option(optname, metavar=metavar, type=type, help=help,
action="store")
add("--brokers", "<url>", "string", "HA brokers use <url> to connect
to each other")
add("--public-brokers", "<url>", "string", "Clients use <url> to
connect to HA brokers")
- add("--backups", "<n>", "int", "Expect <n> backups to be running")
+ add("--backups", "<n>", "int", "Expect <n> backups to be running"),
- def do_execute(self, qmf_broker, opts, args):
+ def do_execute(self, qmf_broker, ha_broker, opts, args):
if (opts.brokers): qmf_broker._method("setBrokers",
{"url":opts.brokers}, HA_BROKER)
if (opts.public_brokers): qmf_broker._method("setPublicBrokers",
{"url":opts.public_brokers}, HA_BROKER)
if (opts.backups): qmf_broker._method("setExpectedBackups",
{"expectedBackups":opts.backups}, HA_BROKER)
+
SetCmd()
class QueryCmd(Command):
def __init__(self):
Command.__init__(self, "query", "Print HA configuration settings")
- def do_execute(self, qmf_broker, opts, args):
- hb = qmf_broker.getHaBroker()
+ def do_execute(self, qmf_broker, ha_broker, opts, args):
+ hb = ha_broker
for x in [("Status:", hb.status),
("Brokers URL:", hb.brokers),
- ("Public URL:", hb.publicBrokers)]:
- print "%-16s%s"%(x[0], x[1])
-
+ ("Public URL:", hb.publicBrokers),
+ ("Expected Backups:", hb.expectedBackups)
+ ]:
+ print "%-20s %s"%(x[0], x[1])
QueryCmd()
+def print_usage(name):
+ print "usage: %s <command> [<arguments>]\n\nCommands are:\n"%name
+ for name, command in Command.commands.iteritems():
+ help = command.help
+ print " %-12s %s."%(name, help.split(".")[0])
+ print "\nFor help with a command: %s <command> --help\n"%name
+
+def find_command(args):
+ """Find a command among the arguments and options"""
+ for arg in args:
+ if arg in Command.commands:
+ return Command.commands[arg]
+ return None
+
def main(argv):
try:
- command=argv[1:]
- if command and command[0] == "--help-all":
+ args=argv[1:]
+ if args and args[0] == "--help-all":
for c in Command.commands.itervalues():
c.op.print_help(); print
return 1
- if not command or not command[0] in Command.commands:
- print_all_help(argv[0]);
+ command = find_command(args)
+ if not command:
+ print_usage(argv[0]);
return 1;
- Command.commands[command[0]].execute(command)
+ if command.execute(): return 1
except Exception, e:
print e
return 1
Modified: qpid/trunk/qpid/tools/src/py/qpidtoollibs/broker.py
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/tools/src/py/qpidtoollibs/broker.py?rev=1295759&r1=1295758&r2=1295759&view=diff
==============================================================================
--- qpid/trunk/qpid/tools/src/py/qpidtoollibs/broker.py (original)
+++ qpid/trunk/qpid/tools/src/py/qpidtoollibs/broker.py Thu Mar 1 19:47:08 2012
@@ -122,13 +122,18 @@ class BrokerAgent(object):
for item in items:
objs.append(cls(self, item))
return objs
-
+
def _getBrokerObject(self, cls, name):
obj = self._doNameQuery(cls.__name__.lower(), name)
if obj:
return cls(self, obj)
return None
+ def _getSingleObject(self, cls):
+ objects = self._getAllBrokerObjects(cls)
+ if objects: return objects[0]
+ return None
+
def getBroker(self):
"""
Get the Broker object that contains broker-scope statistics and operations.
@@ -138,16 +143,14 @@ class BrokerAgent(object):
# of a bug that used to be in the broker whereby by-name queries did not
return the
# object timestamps.
#
- brokers = self._getAllBrokerObjects(Broker)
- if brokers:
- return brokers[0]
- return None
+ return self._getSingleObject(Broker)
+
def getCluster(self):
- return self._getAllBrokerObjects(Cluster)[0]
+ return self._getSingleObject(Cluster)
def getHaBroker(self):
- return self._getAllBrokerObjects(HaBroker)[0]
+ return self._getSingleObject(HaBroker)
def getAllConnections(self):
return self._getAllBrokerObjects(Connection)
@@ -186,10 +189,7 @@ class BrokerAgent(object):
return self._getAllBrokerObjects(Link)
def getAcl(self):
- objects = self._getAllBrokerObjects(Acl)
- if len(objects) > 0:
- return objects[0]
- return None # Acl module not loaded
+ return self._getSingleObject(Acl)
def echo(self, sequence, body):
"""Request a response to test the path to the management broker"""
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscr...@qpid.apache.org
