You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2012/08/03 14:14:12 UTC
svn commit: r1368910 [7/27] - in /qpid/branches/asyncstore: ./ bin/
cpp/bindings/qmf/ruby/ cpp/bindings/qmf2/ruby/ cpp/bindings/qpid/python/
cpp/bindings/qpid/ruby/ cpp/bindings/qpid/ruby/features/
cpp/bindings/qpid/ruby/features/step_definitions/ cpp/...
Modified: qpid/branches/asyncstore/cpp/src/qpid/ha/BrokerReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/ha/BrokerReplicator.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/ha/BrokerReplicator.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/ha/BrokerReplicator.cpp Fri Aug 3 12:13:32 2012
@@ -22,6 +22,7 @@
#include "HaBroker.h"
#include "QueueReplicator.h"
#include "qpid/broker/Broker.h"
+#include "qpid/broker/Connection.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/Link.h"
#include "qpid/framing/FieldTable.h"
@@ -37,8 +38,11 @@
#include "qmf/org/apache/qpid/broker/EventQueueDeclare.h"
#include "qmf/org/apache/qpid/broker/EventQueueDelete.h"
#include "qmf/org/apache/qpid/broker/EventSubscribe.h"
+#include "qmf/org/apache/qpid/ha/EventMembersUpdate.h"
#include <algorithm>
#include <sstream>
+#include <iostream>
+#include <assert.h>
namespace qpid {
namespace ha {
@@ -50,15 +54,16 @@ using qmf::org::apache::qpid::broker::Ev
using qmf::org::apache::qpid::broker::EventQueueDeclare;
using qmf::org::apache::qpid::broker::EventQueueDelete;
using qmf::org::apache::qpid::broker::EventSubscribe;
+using qmf::org::apache::qpid::ha::EventMembersUpdate;
using namespace framing;
using std::string;
+using std::ostream;
using types::Variant;
using namespace broker;
namespace {
const string QPID_CONFIGURATION_REPLICATOR("qpid.configuration-replicator");
-const string QPID_REPLICATE("qpid.replicate");
const string CLASS_NAME("_class_name");
const string EVENT("_event");
@@ -69,10 +74,13 @@ const string SCHEMA_ID("_schema_id");
const string VALUES("_values");
const string ALTEX("altEx");
+const string ALTEXCHANGE("altExchange");
const string ARGS("args");
const string ARGUMENTS("arguments");
const string AUTODEL("autoDel");
const string AUTODELETE("autoDelete");
+const string EXCL("excl");
+const string EXCLUSIVE("exclusive");
const string BIND("bind");
const string UNBIND("unbind");
const string BINDING("binding");
@@ -86,12 +94,12 @@ const string KEY("key");
const string NAME("name");
const string QNAME("qName");
const string QUEUE("queue");
-const string RHOST("rhost");
const string TYPE("type");
-const string USER("user");
const string HA_BROKER("habroker");
+const string PARTIAL("partial");
-const string AGENT_IND_EVENT_ORG_APACHE_QPID_BROKER("agent.ind.event.org_apache_qpid_broker.#");
+const string AGENT_EVENT_BROKER("agent.ind.event.org_apache_qpid_broker.#");
+const string AGENT_EVENT_HA("agent.ind.event.org_apache_qpid_ha.#");
const string QMF2("qmf2");
const string QMF_CONTENT("qmf.content");
const string QMF_DEFAULT_TOPIC("qmf.default.topic");
@@ -107,6 +115,7 @@ const string ORG_APACHE_QPID_HA("org.apa
const string QMF_DEFAULT_DIRECT("qmf.default.direct");
const string _QUERY_REQUEST("_query_request");
const string BROKER("broker");
+const string MEMBERS("members");
bool isQMFv2(const Message& message) {
const framing::MessageProperties* props = message.getProperties<framing::MessageProperties>();
@@ -117,7 +126,9 @@ template <class T> bool match(Variant::M
return T::match(schema[CLASS_NAME], schema[PACKAGE_NAME]);
}
-void sendQuery(const string& packageName, const string& className, const string& queueName, SessionHandler& sessionHandler) {
+void sendQuery(const string& packageName, const string& className, const string& queueName,
+ SessionHandler& sessionHandler)
+{
framing::AMQP_ServerProxy peer(sessionHandler.out);
Variant::Map request;
request[_WHAT] = OBJECT;
@@ -137,6 +148,7 @@ void sendQuery(const string& packageName
props->setAppId(QMF2);
props->getApplicationHeaders().setString(QMF_OPCODE, _QUERY_REQUEST);
headerBody.get<qpid::framing::DeliveryProperties>(true)->setRoutingKey(BROKER);
+ headerBody.get<qpid::framing::MessageProperties>(true)->setCorrelationId(className);
AMQFrame header(headerBody);
header.setBof(false);
header.setEof(false);
@@ -159,39 +171,23 @@ Variant::Map asMapVoid(const Variant& va
if (!value.isVoid()) return value.asMap();
else return Variant::Map();
}
-
} // namespace
-
-ReplicateLevel BrokerReplicator::replicateLevel(const std::string& str) {
- ReplicateLevel rl;
- if (qpid::ha::replicateLevel(str, rl)) return rl;
- else return haBroker.getSettings().replicateDefault;
-}
-
-ReplicateLevel BrokerReplicator::replicateLevel(const framing::FieldTable& f) {
- if (f.isSet(QPID_REPLICATE))
- return replicateLevel(f.getAsString(QPID_REPLICATE));
- else
- return haBroker.getSettings().replicateDefault;
-}
-
-ReplicateLevel BrokerReplicator::replicateLevel(const Variant::Map& m) {
- Variant::Map::const_iterator i = m.find(QPID_REPLICATE);
- if (i != m.end())
- return replicateLevel(i->second.asString());
- else
- return haBroker.getSettings().replicateDefault;
-}
-
-BrokerReplicator::~BrokerReplicator() {}
-
BrokerReplicator::BrokerReplicator(HaBroker& hb, const boost::shared_ptr<Link>& l)
: Exchange(QPID_CONFIGURATION_REPLICATOR),
- haBroker(hb), broker(hb.getBroker()), link(l)
-{
+ logPrefix("Backup: "), replicationTest(hb.getReplicationTest()),
+ haBroker(hb), broker(hb.getBroker()), link(l),
+ initialized(false),
+ alternates(hb.getBroker().getExchanges())
+{}
+
+void BrokerReplicator::initialize() {
+ // Can't do this in the constructor because we need a shared_ptr to this.
+ types::Uuid uuid(true);
+ const std::string name(QPID_CONFIGURATION_REPLICATOR + ".bridge." + uuid.str());
broker.getLinks().declare(
- link->getHost(), link->getPort(),
+ name, // name for bridge
+ *link, // parent
false, // durable
QPID_CONFIGURATION_REPLICATOR, // src
QPID_CONFIGURATION_REPLICATOR, // dest
@@ -202,21 +198,41 @@ BrokerReplicator::BrokerReplicator(HaBro
"", // excludes
false, // dynamic
0, // sync?
- boost::bind(&BrokerReplicator::initializeBridge, this, _1, _2)
+ // shared_ptr keeps this in memory until outstanding initializeBridge
+ // calls are run.
+ boost::bind(&BrokerReplicator::initializeBridge, shared_from_this(), _1, _2)
);
}
+BrokerReplicator::~BrokerReplicator() { }
+
// This is called in the connection IO thread when the bridge is started.
void BrokerReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionHandler) {
- framing::AMQP_ServerProxy peer(sessionHandler.out);
+ // Use the credentials of the outgoing Link connection for creating queues,
+ // exchanges etc. We know link->getConnection() is non-zero because we are
+ // being called in the connections thread context.
+ //
+ assert(link->getConnection());
+ userId = link->getConnection()->getUserId();
+ remoteHost = link->getConnection()->getUrl();
+
+ link->getRemoteAddress(primary);
string queueName = bridge.getQueueName();
+
+ QPID_LOG(info, logPrefix << (initialized ? "Connecting" : "Failing over")
+ << " to primary " << primary
+ << " status:" << printable(haBroker.getStatus()));
+ initialized = true;
+
+ framing::AMQP_ServerProxy peer(sessionHandler.out);
const qmf::org::apache::qpid::broker::ArgsLinkBridge& args(bridge.getArgs());
//declare and bind an event queue
FieldTable declareArgs;
- declareArgs.setString(QPID_REPLICATE, str(RL_NONE));
+ declareArgs.setString(QPID_REPLICATE, printable(NONE).str());
peer.getQueue().declare(queueName, "", false, false, true, true, declareArgs);
- peer.getExchange().bind(queueName, QMF_DEFAULT_TOPIC, AGENT_IND_EVENT_ORG_APACHE_QPID_BROKER, FieldTable());
+ peer.getExchange().bind(queueName, QMF_DEFAULT_TOPIC, AGENT_EVENT_BROKER, FieldTable());
+ peer.getExchange().bind(queueName, QMF_DEFAULT_TOPIC, AGENT_EVENT_HA, FieldTable());
//subscribe to the queue
peer.getMessage().subscribe(queueName, args.i_dest, 1, 0, false, "", 0, FieldTable());
peer.getMessage().flow(args.i_dest, 0, 0xFFFFFFFF);
@@ -228,23 +244,30 @@ void BrokerReplicator::initializeBridge(
sendQuery(ORG_APACHE_QPID_BROKER, QUEUE, queueName, sessionHandler);
sendQuery(ORG_APACHE_QPID_BROKER, EXCHANGE, queueName, sessionHandler);
sendQuery(ORG_APACHE_QPID_BROKER, BINDING, queueName, sessionHandler);
- QPID_LOG(debug, "HA: Backup configuration bridge: " << queueName);
}
void BrokerReplicator::route(Deliverable& msg) {
+ // We transition from JOINING->CATCHUP on the first message received from the primary.
+ // Until now we couldn't be sure if we had a good connection to the primary.
+ if (haBroker.getStatus() == JOINING) {
+ haBroker.setStatus(CATCHUP);
+ QPID_LOG(notice, logPrefix << "Connected to primary " << primary);
+ }
+
const framing::FieldTable* headers = msg.getMessage().getApplicationHeaders();
+ const MessageProperties* messageProperties = msg.getMessage().getProperties<MessageProperties>();
Variant::List list;
try {
- if (!isQMFv2(msg.getMessage()) || !headers)
+ if (!isQMFv2(msg.getMessage()) || !headers || !messageProperties)
throw Exception("Unexpected message, not QMF2 event or query response.");
// decode as list
string content = msg.getMessage().getFrames().getContent();
amqp_0_10::ListCodec::decode(content, list);
-
+ QPID_LOG(trace, "Broker replicator received: " << *messageProperties);
if (headers->getAsString(QMF_CONTENT) == EVENT) {
for (Variant::List::iterator i = list.begin(); i != list.end(); ++i) {
Variant::Map& map = i->asMap();
- QPID_LOG(trace, "HA: Backup received event: " << map);
+ QPID_LOG(trace, "Broker replicator event: " << map);
Variant::Map& schema = map[SCHEMA_ID].asMap();
Variant::Map& values = map[VALUES].asMap();
if (match<EventQueueDeclare>(schema)) doEventQueueDeclare(values);
@@ -253,12 +276,13 @@ void BrokerReplicator::route(Deliverable
else if (match<EventExchangeDelete>(schema)) doEventExchangeDelete(values);
else if (match<EventBind>(schema)) doEventBind(values);
else if (match<EventUnbind>(schema)) doEventUnbind(values);
+ else if (match<EventMembersUpdate>(schema)) doEventMembersUpdate(values);
}
} else if (headers->getAsString(QMF_OPCODE) == QUERY_RESPONSE) {
for (Variant::List::iterator i = list.begin(); i != list.end(); ++i) {
Variant::Map& map = i->asMap();
- QPID_LOG(trace, "HA: Backup received event: " << map);
- string type = map[SCHEMA_ID].asMap()[CLASS_NAME];
+ QPID_LOG(trace, "Broker replicator response: " << map);
+ string type = map[SCHEMA_ID].asMap()[CLASS_NAME].asString();
Variant::Map& values = map[VALUES].asMap();
framing::FieldTable args;
amqp_0_10::translate(asMapVoid(values[ARGUMENTS]), args);
@@ -267,85 +291,82 @@ void BrokerReplicator::route(Deliverable
else if (type == BINDING) doResponseBind(values);
else if (type == HA_BROKER) doResponseHaBroker(values);
}
+ if (messageProperties->getCorrelationId() == EXCHANGE && !headers->isSet(PARTIAL)) {
+ // We have received all of the exchange response.
+ alternates.clear();
+ }
}
} catch (const std::exception& e) {
- QPID_LOG(critical, "HA: Backup configuration failed: " << e.what()
+ QPID_LOG(critical, logPrefix << "Configuration failed: " << e.what()
<< ": while handling: " << list);
+ haBroker.shutdown();
throw;
}
}
+
void BrokerReplicator::doEventQueueDeclare(Variant::Map& values) {
- string name = values[QNAME].asString();
Variant::Map argsMap = asMapVoid(values[ARGS]);
- if (values[DISP] == CREATED && replicateLevel(argsMap)) {
+ bool autoDel = values[AUTODEL].asBool();
+ bool excl = values[EXCL].asBool();
+ if (values[DISP] == CREATED &&
+ replicationTest.isReplicated(CONFIGURATION, argsMap, autoDel, excl))
+ {
+ string name = values[QNAME].asString();
+ QPID_LOG(debug, logPrefix << "Queue declare event: " << name);
framing::FieldTable args;
amqp_0_10::translate(argsMap, args);
- std::pair<boost::shared_ptr<Queue>, bool> result =
- broker.createQueue(
- name,
- values[DURABLE].asBool(),
- values[AUTODEL].asBool(),
- 0 /*i.e. no owner regardless of exclusivity on master*/,
- values[ALTEX].asString(),
- args,
- values[USER].asString(),
- values[RHOST].asString());
- if (result.second) {
- QPID_LOG(debug, "HA: Backup queue declare event: " << name);
- startQueueReplicator(result.first);
- } else {
- // FIXME aconway 2011-12-02: what's the right way to handle this?
- // Should we delete the old & re-create form the event? Responses
- // may be old but events are always up-to-date.
- QPID_LOG(warning, "HA: Backup queue declare event, already exists: " << name);
+ // If we already have a queue with this name, replace it.
+ // The queue was definitely created on the primary.
+ if (broker.getQueues().find(name)) {
+ QPID_LOG(warning, logPrefix << "Replacing exsiting queue: " << name);
+ broker.getQueues().destroy(name);
+ stopQueueReplicator(name);
}
+ boost::shared_ptr<Queue> queue = createQueue(
+ name, values[DURABLE].asBool(), autoDel, args, values[ALTEX].asString());
+ assert(queue); // Should be created since we destroed the previous queue above.
+ if (queue) startQueueReplicator(queue);
}
}
+boost::shared_ptr<QueueReplicator> BrokerReplicator::findQueueReplicator(
+ const std::string& qname)
+{
+ string rname = QueueReplicator::replicatorName(qname);
+ boost::shared_ptr<broker::Exchange> ex = broker.getExchanges().find(rname);
+ return boost::dynamic_pointer_cast<QueueReplicator>(ex);
+}
+
void BrokerReplicator::doEventQueueDelete(Variant::Map& values) {
// The remote queue has already been deleted so replicator
// sessions may be closed by a "queue deleted" exception.
string name = values[QNAME].asString();
boost::shared_ptr<Queue> queue = broker.getQueues().find(name);
- if (!queue) {
- QPID_LOG(warning, "HA: Backup queue delete event, does not exist: " << name);
- } else if (!replicateLevel(queue->getSettings())) {
- QPID_LOG(warning, "HA: Backup queue delete event, not replicated: " << name);
- } else {
- string rname = QueueReplicator::replicatorName(name);
- boost::shared_ptr<broker::Exchange> ex = broker.getExchanges().find(rname);
- boost::shared_ptr<QueueReplicator> qr = boost::dynamic_pointer_cast<QueueReplicator>(ex);
- if (qr) qr->deactivate();
- // QueueReplicator's bridge is now queued for destruction but may not
- // actually be destroyed, deleting the exhange
- broker.getExchanges().destroy(rname);
- broker.deleteQueue(name, values[USER].asString(), values[RHOST].asString());
- QPID_LOG(debug, "HA: Backup queue delete event: " << name);
+ if (queue && replicationTest.replicateLevel(queue->getSettings())) {
+ QPID_LOG(debug, logPrefix << "Queue delete event: " << name);
+ stopQueueReplicator(name);
+ broker.deleteQueue(name, userId, remoteHost);
}
}
void BrokerReplicator::doEventExchangeDeclare(Variant::Map& values) {
Variant::Map argsMap(asMapVoid(values[ARGS]));
- if (values[DISP] == CREATED && replicateLevel(argsMap)) {
+ if (!replicationTest.replicateLevel(argsMap)) return; // Not a replicated exchange.
+ if (values[DISP] == CREATED && replicationTest.replicateLevel(argsMap)) {
string name = values[EXNAME].asString();
+ QPID_LOG(debug, logPrefix << "Exchange declare event: " << name);
framing::FieldTable args;
amqp_0_10::translate(argsMap, args);
- if (broker.createExchange(
- name,
- values[EXTYPE].asString(),
- values[DURABLE].asBool(),
- values[ALTEX].asString(),
- args,
- values[USER].asString(),
- values[RHOST].asString()).second)
- {
- QPID_LOG(debug, "HA: Backup exchange declare event: " << name);
- } else {
- // FIXME aconway 2011-11-22: should delete pre-existing exchange
- // and re-create from event. See comment in doEventQueueDeclare.
- QPID_LOG(debug, "HA: Backup exchange declare event, already exists: " << name);
+ // If we already have a exchange with this name, replace it.
+ // The exchange was definitely created on the primary.
+ if (broker.getExchanges().find(name)) {
+ broker.getExchanges().destroy(name);
+ QPID_LOG(warning, logPrefix << "Replaced exsiting exchange: " << name);
}
+ boost::shared_ptr<Exchange> exchange =
+ createExchange(name, values[EXTYPE].asString(), values[DURABLE].asBool(), args, values[ALTEX].asString());
+ assert(exchange);
}
}
@@ -353,15 +374,12 @@ void BrokerReplicator::doEventExchangeDe
string name = values[EXNAME].asString();
boost::shared_ptr<Exchange> exchange = broker.getExchanges().find(name);
if (!exchange) {
- QPID_LOG(warning, "HA: Backup exchange delete event, does not exist: " << name);
- } else if (!replicateLevel(exchange->getArgs())) {
- QPID_LOG(warning, "HA: Backup exchange delete event, not replicated: " << name);
+ QPID_LOG(warning, logPrefix << "Exchange delete event, does not exist: " << name);
+ } else if (!replicationTest.replicateLevel(exchange->getArgs())) {
+ QPID_LOG(warning, logPrefix << "Exchange delete event, not replicated: " << name);
} else {
- QPID_LOG(debug, "HA: Backup exchange delete event:" << name);
- broker.deleteExchange(
- name,
- values[USER].asString(),
- values[RHOST].asString());
+ QPID_LOG(debug, logPrefix << "Exchange delete event:" << name);
+ broker.deleteExchange(name, userId, remoteHost);
}
}
@@ -372,16 +390,16 @@ void BrokerReplicator::doEventBind(Varia
broker.getQueues().find(values[QNAME].asString());
// We only replicate binds for a replicated queue to replicated
// exchange that both exist locally.
- if (exchange && replicateLevel(exchange->getArgs()) &&
- queue && replicateLevel(queue->getSettings()))
+ if (exchange && replicationTest.replicateLevel(exchange->getArgs()) &&
+ queue && replicationTest.replicateLevel(queue->getSettings()))
{
framing::FieldTable args;
amqp_0_10::translate(asMapVoid(values[ARGS]), args);
string key = values[KEY].asString();
- exchange->bind(queue, key, &args);
- QPID_LOG(debug, "HA: Backup bind event: exchange=" << exchange->getName()
+ QPID_LOG(debug, logPrefix << "Bind event: exchange=" << exchange->getName()
<< " queue=" << queue->getName()
<< " key=" << key);
+ exchange->bind(queue, key, &args);
}
}
@@ -392,64 +410,71 @@ void BrokerReplicator::doEventUnbind(Var
broker.getQueues().find(values[QNAME].asString());
// We only replicate unbinds for a replicated queue to replicated
// exchange that both exist locally.
- if (exchange && replicateLevel(exchange->getArgs()) &&
- queue && replicateLevel(queue->getSettings()))
+ if (exchange && replicationTest.replicateLevel(exchange->getArgs()) &&
+ queue && replicationTest.replicateLevel(queue->getSettings()))
{
framing::FieldTable args;
amqp_0_10::translate(asMapVoid(values[ARGS]), args);
string key = values[KEY].asString();
- exchange->unbind(queue, key, &args);
- QPID_LOG(debug, "HA: Backup unbind event: exchange=" << exchange->getName()
+ QPID_LOG(debug, logPrefix << "Unbind event: exchange=" << exchange->getName()
<< " queue=" << queue->getName()
<< " key=" << key);
+ exchange->unbind(queue, key, &args);
}
}
+void BrokerReplicator::doEventMembersUpdate(Variant::Map& values) {
+ Variant::List members = values[MEMBERS].asList();
+ haBroker.setMembership(members);
+}
+
+namespace {
+
+// Get the alternate exchange from the exchange field of a queue or exchange response.
+static const string EXCHANGE_KEY_PREFIX("org.apache.qpid.broker:exchange:");
+
+string getAltExchange(const types::Variant& var) {
+ if (!var.isVoid()) {
+ management::ObjectId oid(var);
+ string key = oid.getV2Key();
+ if (key.find(EXCHANGE_KEY_PREFIX) != 0) throw Exception("Invalid exchange reference: "+key);
+ return key.substr(EXCHANGE_KEY_PREFIX.size());
+ }
+ else return string();
+}
+}
+
void BrokerReplicator::doResponseQueue(Variant::Map& values) {
Variant::Map argsMap(asMapVoid(values[ARGUMENTS]));
- if (!replicateLevel(argsMap)) return;
+ if (!replicationTest.isReplicated(
+ CONFIGURATION,
+ values[ARGUMENTS].asMap(),
+ values[AUTODELETE].asBool(),
+ values[EXCLUSIVE].asBool()))
+ return;
+ string name(values[NAME].asString());
+ QPID_LOG(debug, logPrefix << "Queue response: " << name);
framing::FieldTable args;
amqp_0_10::translate(argsMap, args);
- string name(values[NAME].asString());
- std::pair<boost::shared_ptr<Queue>, bool> result =
- broker.createQueue(
- name,
- values[DURABLE].asBool(),
- values[AUTODELETE].asBool(),
- 0 /*i.e. no owner regardless of exclusivity on master*/,
- ""/*TODO: need to include alternate-exchange*/,
- args,
- ""/*TODO: who is the user?*/,
- ""/*TODO: what should we use as connection id?*/);
- if (result.second) {
- QPID_LOG(debug, "HA: Backup queue response: " << name);
- startQueueReplicator(result.first);
- } else {
- // FIXME aconway 2011-11-22: Normal to find queue already
- // exists if we're failing over.
- QPID_LOG(warning, "HA: Backup queue response, already exists: " << name);
- }
+ boost::shared_ptr<Queue> queue =
+ createQueue(name, values[DURABLE].asBool(), values[AUTODELETE].asBool(), args,
+ getAltExchange(values[ALTEXCHANGE]));
+ // It is normal for the queue to already exist if we are failing over.
+ if (queue) startQueueReplicator(queue);
+ else QPID_LOG(debug, logPrefix << "Queue already replicated: " << name);
}
void BrokerReplicator::doResponseExchange(Variant::Map& values) {
Variant::Map argsMap(asMapVoid(values[ARGUMENTS]));
- if (!replicateLevel(argsMap)) return;
+ if (!replicationTest.replicateLevel(argsMap)) return;
+ string name = values[NAME].asString();
+ QPID_LOG(debug, logPrefix << "Exchange response: " << name);
framing::FieldTable args;
amqp_0_10::translate(argsMap, args);
- if (broker.createExchange(
- values[NAME].asString(),
- values[TYPE].asString(),
- values[DURABLE].asBool(),
- ""/*TODO: need to include alternate-exchange*/,
- args,
- ""/*TODO: who is the user?*/,
- ""/*TODO: what should we use as connection id?*/).second)
- {
- QPID_LOG(debug, "HA: Backup exchange response: " << values[NAME].asString());
- } else {
- QPID_LOG(warning, "HA: Backup exchange query, already exists: " <<
- values[QNAME].asString());
- }
+ boost::shared_ptr<Exchange> exchange = createExchange(
+ name, values[TYPE].asString(), values[DURABLE].asBool(), args,
+ getAltExchange(values[ALTEXCHANGE]));
+ QPID_LOG_IF(debug, !exchange, logPrefix << "Exchange already exists: " << name);
}
namespace {
@@ -480,16 +505,16 @@ void BrokerReplicator::doResponseBind(Va
boost::shared_ptr<Queue> queue = broker.getQueues().find(qName);
// Automatically replicate binding if queue and exchange exist and are replicated
- if (exchange && replicateLevel(exchange->getArgs()) &&
- queue && replicateLevel(queue->getSettings()))
+ if (exchange && replicationTest.replicateLevel(exchange->getArgs()) &&
+ queue && replicationTest.replicateLevel(queue->getSettings()))
{
+ string key = values[KEY].asString();
+ QPID_LOG(debug, logPrefix << "Bind response: exchange:" << exName
+ << " queue:" << qName
+ << " key:" << key);
framing::FieldTable args;
amqp_0_10::translate(asMapVoid(values[ARGUMENTS]), args);
- string key = values[KEY].asString();
exchange->bind(queue, key, &args);
- QPID_LOG(debug, "HA: Backup bind response: exchange=" << exchange->getName()
- << " queue=" << queue->getName()
- << " key=" << key);
}
}
@@ -500,30 +525,97 @@ const string REPLICATE_DEFAULT="replicat
// Received the ha-broker configuration object for the primary broker.
void BrokerReplicator::doResponseHaBroker(Variant::Map& values) {
try {
- ReplicateLevel mine = haBroker.getSettings().replicateDefault;
- ReplicateLevel primary = replicateLevel(values[REPLICATE_DEFAULT].asString());
- if (mine != primary) {
- std::ostringstream os;
- os << "Replicate default on backup (" << mine
- << ") does not match primary (" << primary << ")";
- haBroker.shutdown(os.str());
- }
+ QPID_LOG(trace, logPrefix << "HA Broker response: " << values);
+ ReplicateLevel mine = haBroker.getSettings().replicateDefault.get();
+ ReplicateLevel primary = replicationTest.replicateLevel(
+ values[REPLICATE_DEFAULT].asString());
+ if (mine != primary)
+ throw Exception(QPID_MSG("Replicate default on backup (" << mine
+ << ") does not match primary (" << primary << ")"));
+ haBroker.setMembership(values[MEMBERS].asList());
} catch (const std::exception& e) {
- std::ostringstream os;
- os << "Received invalid replicate default from primary: " << e.what();
- haBroker.shutdown(os.str());
+ QPID_LOG(critical, logPrefix << "Invalid HA Broker response: " << e.what()
+ << ": " << values);
+ haBroker.shutdown();
+ throw;
}
}
-void BrokerReplicator::startQueueReplicator(const boost::shared_ptr<Queue>& queue) {
- if (replicateLevel(queue->getSettings()) == RL_ALL) {
- boost::shared_ptr<QueueReplicator> qr(new QueueReplicator(queue, link));
+void BrokerReplicator::startQueueReplicator(const boost::shared_ptr<Queue>& queue)
+{
+ if (replicationTest.replicateLevel(queue->getSettings()) == ALL) {
+ boost::shared_ptr<QueueReplicator> qr(
+ new QueueReplicator(haBroker, queue, link));
if (!broker.getExchanges().registerExchange(qr))
throw Exception(QPID_MSG("Duplicate queue replicator " << qr->getName()));
qr->activate();
}
}
+void BrokerReplicator::stopQueueReplicator(const std::string& name) {
+ boost::shared_ptr<QueueReplicator> qr = findQueueReplicator(name);
+ if (qr) {
+ qr->deactivate();
+ // QueueReplicator's bridge is now queued for destruction but may not
+ // actually be destroyed.
+ broker.getExchanges().destroy(qr->getName());
+ }
+}
+
+boost::shared_ptr<Queue> BrokerReplicator::createQueue(
+ const std::string& name,
+ bool durable,
+ bool autodelete,
+ const qpid::framing::FieldTable& arguments,
+ const std::string& alternateExchange)
+{
+ std::pair<boost::shared_ptr<Queue>, bool> result =
+ broker.createQueue(
+ name,
+ durable,
+ autodelete,
+ 0, // no owner regardless of exclusivity on primary
+ string(), // Set alternate exchange below
+ arguments,
+ userId,
+ remoteHost);
+ if (result.second) {
+ if (!alternateExchange.empty()) {
+ alternates.setAlternate(
+ alternateExchange, boost::bind(&Queue::setAlternateExchange, result.first, _1));
+ }
+ return result.first;
+ }
+ else return boost::shared_ptr<Queue>();
+}
+
+boost::shared_ptr<Exchange> BrokerReplicator::createExchange(
+ const std::string& name,
+ const std::string& type,
+ bool durable,
+ const qpid::framing::FieldTable& args,
+ const std::string& alternateExchange)
+{
+ std::pair<boost::shared_ptr<Exchange>, bool> result =
+ broker.createExchange(
+ name,
+ type,
+ durable,
+ string(), // Set alternate exchange below
+ args,
+ userId,
+ remoteHost);
+ if (result.second) {
+ alternates.addExchange(result.first);
+ if (!alternateExchange.empty()) {
+ alternates.setAlternate(
+ alternateExchange, boost::bind(&Exchange::setAlternate, result.first, _1));
+ }
+ return result.first;
+ }
+ else return boost::shared_ptr<Exchange>();
+}
+
bool BrokerReplicator::bind(boost::shared_ptr<Queue>, const string&, const framing::FieldTable*) { return false; }
bool BrokerReplicator::unbind(boost::shared_ptr<Queue>, const string&, const framing::FieldTable*) { return false; }
bool BrokerReplicator::isBound(boost::shared_ptr<Queue>, const string* const, const framing::FieldTable* const) { return false; }
Modified: qpid/branches/asyncstore/cpp/src/qpid/ha/BrokerReplicator.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/ha/BrokerReplicator.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/ha/BrokerReplicator.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/ha/BrokerReplicator.h Fri Aug 3 12:13:32 2012
@@ -22,10 +22,15 @@
*
*/
-#include "ReplicateLevel.h"
+#include "types.h"
+#include "ReplicationTest.h"
+#include "AlternateExchangeSetter.h"
+#include "qpid/Address.h"
#include "qpid/broker/Exchange.h"
#include "qpid/types/Variant.h"
+#include "qpid/management/ManagementObject.h"
#include <boost/shared_ptr.hpp>
+#include <boost/enable_shared_from_this.hpp>
namespace qpid {
@@ -42,6 +47,7 @@ class FieldTable;
namespace ha {
class HaBroker;
+class QueueReplicator;
/**
* Replicate configuration on a backup broker.
@@ -51,28 +57,29 @@ class HaBroker;
* exchanges and bindings to replicate the primary.
* It also creates QueueReplicators for newly replicated queues.
*
- * THREAD SAFE: Has no mutable state.
+ * THREAD UNSAFE: Only called in Link connection thread, no need for locking.
*
*/
-class BrokerReplicator : public broker::Exchange
+class BrokerReplicator : public broker::Exchange,
+ public boost::enable_shared_from_this<BrokerReplicator>
{
public:
BrokerReplicator(HaBroker&, const boost::shared_ptr<broker::Link>&);
~BrokerReplicator();
- std::string getType() const;
+
+ void initialize();
// Exchange methods
+ std::string getType() const;
bool bind(boost::shared_ptr<broker::Queue>, const std::string&, const framing::FieldTable*);
bool unbind(boost::shared_ptr<broker::Queue>, const std::string&, const framing::FieldTable*);
void route(broker::Deliverable&);
bool isBound(boost::shared_ptr<broker::Queue>, const std::string* const, const framing::FieldTable* const);
private:
- void initializeBridge(broker::Bridge&, broker::SessionHandler&);
+ typedef boost::shared_ptr<QueueReplicator> QueueReplicatorPtr;
- ReplicateLevel replicateLevel(const std::string&);
- ReplicateLevel replicateLevel(const framing::FieldTable& args);
- ReplicateLevel replicateLevel(const types::Variant::Map& args);
+ void initializeBridge(broker::Bridge&, broker::SessionHandler&);
void doEventQueueDeclare(types::Variant::Map& values);
void doEventQueueDelete(types::Variant::Map& values);
@@ -80,17 +87,40 @@ class BrokerReplicator : public broker::
void doEventExchangeDelete(types::Variant::Map& values);
void doEventBind(types::Variant::Map&);
void doEventUnbind(types::Variant::Map&);
+ void doEventMembersUpdate(types::Variant::Map&);
void doResponseQueue(types::Variant::Map& values);
void doResponseExchange(types::Variant::Map& values);
void doResponseBind(types::Variant::Map& values);
void doResponseHaBroker(types::Variant::Map& values);
+ QueueReplicatorPtr findQueueReplicator(const std::string& qname);
void startQueueReplicator(const boost::shared_ptr<broker::Queue>&);
+ void stopQueueReplicator(const std::string& name);
+ boost::shared_ptr<broker::Queue> createQueue(
+ const std::string& name,
+ bool durable,
+ bool autodelete,
+ const qpid::framing::FieldTable& arguments,
+ const std::string& alternateExchange);
+
+ boost::shared_ptr<broker::Exchange> createExchange(
+ const std::string& name,
+ const std::string& type,
+ bool durable,
+ const qpid::framing::FieldTable& args,
+ const std::string& alternateExchange);
+
+ std::string logPrefix;
+ std::string userId, remoteHost;
+ ReplicationTest replicationTest;
HaBroker& haBroker;
broker::Broker& broker;
boost::shared_ptr<broker::Link> link;
+ bool initialized;
+ AlternateExchangeSetter alternates;
+ qpid::Address primary;
};
}} // namespace qpid::broker
Modified: qpid/branches/asyncstore/cpp/src/qpid/ha/HaBroker.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/ha/HaBroker.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/ha/HaBroker.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/ha/HaBroker.cpp Fri Aug 3 12:13:32 2012
@@ -19,22 +19,31 @@
*
*/
#include "Backup.h"
-#include "ConnectionExcluder.h"
+#include "BackupConnectionExcluder.h"
+#include "ConnectionObserver.h"
#include "HaBroker.h"
-#include "Settings.h"
+#include "Primary.h"
+#include "QueueReplicator.h"
#include "ReplicatingSubscription.h"
+#include "Settings.h"
+#include "qpid/amqp_0_10/Codecs.h"
#include "qpid/Exception.h"
#include "qpid/broker/Broker.h"
#include "qpid/broker/Link.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/SignalHandler.h"
+#include "qpid/framing/FieldTable.h"
#include "qpid/management/ManagementAgent.h"
+#include "qpid/sys/SystemInfo.h"
+#include "qpid/types/Uuid.h"
+#include "qpid/framing/Uuid.h"
#include "qmf/org/apache/qpid/ha/Package.h"
#include "qmf/org/apache/qpid/ha/ArgsHaBrokerReplicate.h"
-#include "qmf/org/apache/qpid/ha/ArgsHaBrokerSetBrokers.h"
-#include "qmf/org/apache/qpid/ha/ArgsHaBrokerSetPublicBrokers.h"
-#include "qmf/org/apache/qpid/ha/ArgsHaBrokerSetExpectedBackups.h"
+#include "qmf/org/apache/qpid/ha/ArgsHaBrokerSetBrokersUrl.h"
+#include "qmf/org/apache/qpid/ha/ArgsHaBrokerSetPublicUrl.h"
+#include "qmf/org/apache/qpid/ha/EventMembersUpdate.h"
#include "qpid/log/Statement.h"
+#include <boost/shared_ptr.hpp>
namespace qpid {
namespace ha {
@@ -42,91 +51,149 @@ namespace ha {
namespace _qmf = ::qmf::org::apache::qpid::ha;
using namespace management;
using namespace std;
+using types::Variant;
+using types::Uuid;
+using sys::Mutex;
-namespace {
-
-const std::string STANDALONE="standalone";
-const std::string CATCH_UP="catch-up";
-const std::string BACKUP="backup";
-const std::string PRIMARY="primary";
-
-} // namespace
-
-
+// Called in Plugin::earlyInitialize
HaBroker::HaBroker(broker::Broker& b, const Settings& s)
- : broker(b),
+ : logPrefix("Broker: "),
+ broker(b),
+ systemId(broker.getSystem()->getSystemId().data()),
settings(s),
- mgmtObject(0)
+ observer(new ConnectionObserver(*this, systemId)),
+ mgmtObject(0),
+ status(STANDALONE),
+ membership(systemId),
+ replicationTest(s.replicateDefault.get())
{
- // Register a factory for replicating subscriptions.
- broker.getConsumerFactories().add(
- boost::shared_ptr<ReplicatingSubscription::Factory>(
- new ReplicatingSubscription::Factory()));
+ // If we are joining a cluster we must start excluding clients now,
+ // otherwise there's a window for a client to connect before we get to
+ // initialize()
+ if (settings.cluster) {
+ QPID_LOG(debug, logPrefix << "Rejecting client connections.");
+ observer->setObserver(boost::shared_ptr<broker::ConnectionObserver>(
+ new BackupConnectionExcluder));
+ broker.getConnectionObservers().add(observer);
+ }
+}
- broker.getKnownBrokers = boost::bind(&HaBroker::getKnownBrokers, this);
+// Called in Plugin::initialize
+void HaBroker::initialize() {
+ // FIXME aconway 2012-07-19: assumes there's a TCP transport with a meaningful port.
+ brokerInfo = BrokerInfo(
+ broker.getSystem()->getNodeName(), broker.getPort(broker::Broker::TCP_TRANSPORT), systemId);
+
+ // Set up the management object.
ManagementAgent* ma = broker.getManagementAgent();
- if (!ma)
+ if (settings.cluster && !ma)
throw Exception("Cannot start HA: management is disabled");
_qmf::Package packageInit(ma);
mgmtObject = new _qmf::HaBroker(ma, this, "ha-broker");
- mgmtObject->set_status(settings.cluster ? BACKUP : STANDALONE);
- mgmtObject->set_replicateDefault(str(settings.replicateDefault));
+ mgmtObject->set_replicateDefault(settings.replicateDefault.str());
+ mgmtObject->set_systemId(systemId);
ma->addObject(mgmtObject);
- // NOTE: lock is not needed in a constructor but we created it just to pass
- // to the set functions.
- sys::Mutex::ScopedLock l(lock);
- if (!settings.clientUrl.empty()) setClientUrl(Url(settings.clientUrl), l);
- if (!settings.brokerUrl.empty()) setBrokerUrl(Url(settings.brokerUrl), l);
+ // Register a factory for replicating subscriptions.
+ broker.getConsumerFactories().add(
+ boost::shared_ptr<ReplicatingSubscription::Factory>(
+ new ReplicatingSubscription::Factory()));
+
+ // If we are in a cluster, start as backup in joining state.
+ if (settings.cluster) {
+ status = JOINING;
+ backup.reset(new Backup(*this, settings));
+ broker.getKnownBrokers = boost::bind(&HaBroker::getKnownBrokers, this);
+ }
+
+ if (!settings.clientUrl.empty()) setClientUrl(Url(settings.clientUrl));
+ if (!settings.brokerUrl.empty()) setBrokerUrl(Url(settings.brokerUrl));
+
+
+ QPID_LOG(notice, logPrefix << "Initializing: " << brokerInfo);
- // If we are in a cluster, we start in backup mode.
- if (settings.cluster) backup.reset(new Backup(*this, s));
+ // NOTE: lock is not needed in a constructor, but create one
+ // to pass to functions that have a ScopedLock parameter.
+ Mutex::ScopedLock l(lock);
+ statusChanged(l);
}
-HaBroker::~HaBroker() {}
+HaBroker::~HaBroker() {
+ QPID_LOG(notice, logPrefix << "Shut down: " << brokerInfo);
+ broker.getConnectionObservers().remove(observer);
+}
+
+void HaBroker::recover() {
+ auto_ptr<Backup> b;
+ {
+ Mutex::ScopedLock l(lock);
+ // No longer replicating, close link. Note: link must be closed before we
+ // setStatus(RECOVERING) as that will remove our broker info from the
+ // outgoing link properties so we won't recognize self-connects.
+ b = backup;
+ }
+ b.reset(); // Call destructor outside of lock.
+ BrokerInfo::Set backups;
+ {
+ Mutex::ScopedLock l(lock);
+ setStatus(RECOVERING, l);
+ backups = membership.otherBackups();
+ membership.reset(brokerInfo);
+ // Drop the lock, new Primary may call back on activate.
+ }
+ // Outside of lock, may call back on activate()
+ primary.reset(new Primary(*this, backups)); // Starts primary-ready check.
+}
+
+// Called back from Primary active check.
+void HaBroker::activate() { setStatus(ACTIVE); }
Manageable::status_t HaBroker::ManagementMethod (uint32_t methodId, Args& args, string&) {
- sys::Mutex::ScopedLock l(lock);
switch (methodId) {
case _qmf::HaBroker::METHOD_PROMOTE: {
- if (backup.get()) { // I am a backup
- // NOTE: resetting backup allows client connections, so any
- // primary state should be set up here before backup.reset()
- backup.reset();
- QPID_LOG(notice, "HA: Promoted to primary");
- mgmtObject->set_status(PRIMARY);
+ switch (getStatus()) {
+ case JOINING: recover(); break;
+ case CATCHUP:
+ QPID_LOG(error, logPrefix << "Still catching up, cannot be promoted.");
+ throw Exception("Still catching up, cannot be promoted.");
+ break;
+ case READY: recover(); break;
+ case RECOVERING: break;
+ case ACTIVE: break;
+ case STANDALONE: break;
}
break;
}
- case _qmf::HaBroker::METHOD_SETBROKERS: {
- setBrokerUrl(Url(dynamic_cast<_qmf::ArgsHaBrokerSetBrokers&>(args).i_url), l);
+ case _qmf::HaBroker::METHOD_SETBROKERSURL: {
+ setBrokerUrl(Url(dynamic_cast<_qmf::ArgsHaBrokerSetBrokersUrl&>(args).i_url));
break;
}
- case _qmf::HaBroker::METHOD_SETPUBLICBROKERS: {
- setClientUrl(Url(dynamic_cast<_qmf::ArgsHaBrokerSetPublicBrokers&>(args).i_url), l);
+ case _qmf::HaBroker::METHOD_SETPUBLICURL: {
+ setClientUrl(Url(dynamic_cast<_qmf::ArgsHaBrokerSetPublicUrl&>(args).i_url));
break;
}
- case _qmf::HaBroker::METHOD_SETEXPECTEDBACKUPS: {
- setExpectedBackups(dynamic_cast<_qmf::ArgsHaBrokerSetExpectedBackups&>(args).i_expectedBackups, l);
- break;
- }
case _qmf::HaBroker::METHOD_REPLICATE: {
_qmf::ArgsHaBrokerReplicate& bq_args =
dynamic_cast<_qmf::ArgsHaBrokerReplicate&>(args);
- QPID_LOG(debug, "HA replicating individual queue "<< bq_args.i_queue << " from " << bq_args.i_broker);
+ QPID_LOG(debug, logPrefix << "Replicate individual queue "
+ << bq_args.i_queue << " from " << bq_args.i_broker);
boost::shared_ptr<broker::Queue> queue = broker.getQueues().get(bq_args.i_queue);
Url url(bq_args.i_broker);
string protocol = url[0].protocol.empty() ? "tcp" : url[0].protocol;
+ Uuid uuid(true);
std::pair<broker::Link::shared_ptr, bool> result = broker.getLinks().declare(
+ broker::QPID_NAME_PREFIX + string("ha.link.") + uuid.str(),
url[0].host, url[0].port, protocol,
false, // durable
- settings.mechanism, settings.username, settings.password);
+ settings.mechanism, settings.username, settings.password,
+ false); // no amq.failover - don't want to use client URL.
boost::shared_ptr<broker::Link> link = result.first;
link->setUrl(url);
// Create a queue replicator
- boost::shared_ptr<QueueReplicator> qr(new QueueReplicator(queue, link));
+ boost::shared_ptr<QueueReplicator> qr(
+ new QueueReplicator(*this, queue, link));
qr->activate();
broker.getExchanges().registerExchange(qr);
break;
@@ -138,43 +205,146 @@ Manageable::status_t HaBroker::Managemen
return Manageable::STATUS_OK;
}
-void HaBroker::setClientUrl(const Url& url, const sys::Mutex::ScopedLock& l) {
+void HaBroker::setClientUrl(const Url& url) {
+ Mutex::ScopedLock l(lock);
if (url.empty()) throw Exception("Invalid empty URL for HA client failover");
clientUrl = url;
updateClientUrl(l);
}
-void HaBroker::updateClientUrl(const sys::Mutex::ScopedLock&) {
+void HaBroker::updateClientUrl(Mutex::ScopedLock&) {
Url url = clientUrl.empty() ? brokerUrl : clientUrl;
if (url.empty()) throw Url::Invalid("HA client URL is empty");
- mgmtObject->set_publicBrokers(url.str());
+ mgmtObject->set_publicUrl(url.str());
knownBrokers.clear();
knownBrokers.push_back(url);
- QPID_LOG(debug, "HA: Setting client URL to: " << url);
+ QPID_LOG(debug, logPrefix << "Setting client URL to: " << url);
}
-void HaBroker::setBrokerUrl(const Url& url, const sys::Mutex::ScopedLock& l) {
+void HaBroker::setBrokerUrl(const Url& url) {
+ Mutex::ScopedLock l(lock);
if (url.empty()) throw Url::Invalid("HA broker URL is empty");
- QPID_LOG(debug, "HA: Setting broker URL to: " << url);
brokerUrl = url;
- mgmtObject->set_brokers(brokerUrl.str());
+ mgmtObject->set_brokersUrl(brokerUrl.str());
if (backup.get()) backup->setBrokerUrl(brokerUrl);
// Updating broker URL also updates defaulted client URL:
if (clientUrl.empty()) updateClientUrl(l);
}
-void HaBroker::setExpectedBackups(size_t n, const sys::Mutex::ScopedLock&) {
- expectedBackups = n;
- mgmtObject->set_expectedBackups(n);
-}
-
std::vector<Url> HaBroker::getKnownBrokers() const {
+ Mutex::ScopedLock l(lock);
return knownBrokers;
}
-void HaBroker::shutdown(const std::string& message) {
- QPID_LOG(critical, "Shutting down: " << message);
+void HaBroker::shutdown() {
+ QPID_LOG(critical, logPrefix << "Critical error, shutting down.");
broker.shutdown();
}
+BrokerStatus HaBroker::getStatus() const {
+ Mutex::ScopedLock l(lock);
+ return status;
+}
+
+void HaBroker::setStatus(BrokerStatus newStatus) {
+ Mutex::ScopedLock l(lock);
+ setStatus(newStatus, l);
+}
+
+namespace {
+bool checkTransition(BrokerStatus from, BrokerStatus to) {
+ // Legal state transitions. Initial state is JOINING, ACTIVE is terminal.
+ static const BrokerStatus TRANSITIONS[][2] = {
+ { JOINING, CATCHUP }, // Connected to primary
+ { JOINING, RECOVERING }, // Chosen as initial primary.
+ { CATCHUP, READY }, // Caught up all queues, ready to take over.
+ { READY, RECOVERING }, // Chosen as new primary
+ { READY, CATCHUP }, // Timed out failing over, demoted to catch-up.
+ { RECOVERING, ACTIVE } // All expected backups are ready
+ };
+ static const size_t N = sizeof(TRANSITIONS)/sizeof(TRANSITIONS[0]);
+ for (size_t i = 0; i < N; ++i) {
+ if (TRANSITIONS[i][0] == from && TRANSITIONS[i][1] == to)
+ return true;
+ }
+ return false;
+}
+} // namespace
+
+void HaBroker::setStatus(BrokerStatus newStatus, Mutex::ScopedLock& l) {
+ QPID_LOG(info, logPrefix << "Status change: "
+ << printable(status) << " -> " << printable(newStatus));
+ bool legal = checkTransition(status, newStatus);
+ assert(legal);
+ if (!legal) {
+ QPID_LOG(critical, logPrefix << "Illegal state transition: "
+ << printable(status) << " -> " << printable(newStatus));
+ shutdown();
+ }
+ status = newStatus;
+ statusChanged(l);
+}
+
+void HaBroker::statusChanged(Mutex::ScopedLock& l) {
+ mgmtObject->set_status(printable(status).str());
+ brokerInfo.setStatus(status);
+ setLinkProperties(l);
+}
+
+void HaBroker::membershipUpdated(Mutex::ScopedLock&) {
+ Variant::List brokers = membership.asList();
+ mgmtObject->set_members(brokers);
+ broker.getManagementAgent()->raiseEvent(_qmf::EventMembersUpdate(brokers));
+}
+
+void HaBroker::setMembership(const Variant::List& brokers) {
+ Mutex::ScopedLock l(lock);
+ membership.assign(brokers);
+ QPID_LOG(info, logPrefix << "Membership update: " << membership);
+ BrokerInfo info;
+ // Update my status to what the primary says it is. The primary can toggle
+ // status between READY and CATCHUP based on the state of our subscriptions.
+ if (membership.get(systemId, info) && status != info.getStatus()) {
+ setStatus(info.getStatus(), l);
+ if (backup.get()) backup->setStatus(status);
+ }
+ membershipUpdated(l);
+}
+
+void HaBroker::resetMembership(const BrokerInfo& b) {
+ Mutex::ScopedLock l(lock);
+ membership.reset(b);
+ QPID_LOG(debug, logPrefix << "Membership reset to: " << membership);
+ membershipUpdated(l);
+}
+
+void HaBroker::addBroker(const BrokerInfo& b) {
+ Mutex::ScopedLock l(lock);
+ membership.add(b);
+ QPID_LOG(debug, logPrefix << "Membership add: " << b << " now: " << membership);
+ membershipUpdated(l);
+}
+
+void HaBroker::removeBroker(const Uuid& id) {
+ Mutex::ScopedLock l(lock);
+ membership.remove(id);
+ QPID_LOG(debug, logPrefix << "Membership remove: " << id << " now: " << membership);
+ membershipUpdated(l);
+}
+
+void HaBroker::setLinkProperties(Mutex::ScopedLock&) {
+ framing::FieldTable linkProperties = broker.getLinkClientProperties();
+ if (isBackup(status)) {
+ // If this is a backup then any outgoing links are backup
+ // links and need to be tagged.
+ linkProperties.setTable(ConnectionObserver::BACKUP_TAG, brokerInfo.asFieldTable());
+ }
+ else {
+ // If this is a primary then any outgoing links are federation links
+ // and should not be tagged.
+ linkProperties.erase(ConnectionObserver::BACKUP_TAG);
+ }
+ broker.setLinkClientProperties(linkProperties);
+}
+
}} // namespace qpid::ha
Modified: qpid/branches/asyncstore/cpp/src/qpid/ha/HaBroker.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/ha/HaBroker.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/ha/HaBroker.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/ha/HaBroker.h Fri Aug 3 12:13:32 2012
@@ -22,31 +22,54 @@
*
*/
+#include "BrokerInfo.h"
+#include "Membership.h"
+#include "types.h"
+#include "ReplicationTest.h"
#include "Settings.h"
#include "qpid/Url.h"
#include "qpid/sys/Mutex.h"
#include "qmf/org/apache/qpid/ha/HaBroker.h"
#include "qpid/management/Manageable.h"
+#include "qpid/types/Variant.h"
#include <memory>
+#include <set>
+#include <boost/shared_ptr.hpp>
namespace qpid {
+
+namespace types {
+class Variant;
+}
+
namespace broker {
class Broker;
+class Queue;
+}
+namespace framing {
+class FieldTable;
}
+
namespace ha {
class Backup;
+class ConnectionObserver;
+class Primary;
/**
- * HA state and actions associated with a broker.
+ * HA state and actions associated with a HA broker. Holds all the management info.
*
* THREAD SAFE: may be called in arbitrary broker IO or timer threads.
*/
class HaBroker : public management::Manageable
{
public:
+ /** HaBroker is constructed during earlyInitialize */
HaBroker(broker::Broker&, const Settings&);
~HaBroker();
+ /** Called during plugin initialization */
+ void initialize();
+
// Implement Manageable.
qpid::management::ManagementObject* GetManagementObject() const { return mgmtObject; }
management::Manageable::status_t ManagementMethod (
@@ -55,26 +78,57 @@ class HaBroker : public management::Mana
broker::Broker& getBroker() { return broker; }
const Settings& getSettings() const { return settings; }
- // Log a critical error message and shut down the broker.
- void shutdown(const std::string& message);
+ /** Shut down the broker. Caller should log a critical error message. */
+ void shutdown();
+
+ BrokerStatus getStatus() const;
+ void setStatus(BrokerStatus);
+ void activate();
+
+ Backup* getBackup() { return backup.get(); }
+ ReplicationTest getReplicationTest() const { return replicationTest; }
+
+ boost::shared_ptr<ConnectionObserver> getObserver() { return observer; }
+
+ const BrokerInfo& getBrokerInfo() const { return brokerInfo; }
+
+ void setMembership(const types::Variant::List&); // Set membership from list.
+ void resetMembership(const BrokerInfo& b); // Reset to contain just one member.
+ void addBroker(const BrokerInfo& b); // Add a broker to the membership.
+ void removeBroker(const types::Uuid& id); // Remove a broker from membership.
private:
- void setClientUrl(const Url&, const sys::Mutex::ScopedLock&);
- void setBrokerUrl(const Url&, const sys::Mutex::ScopedLock&);
- void setExpectedBackups(size_t, const sys::Mutex::ScopedLock&);
- void updateClientUrl(const sys::Mutex::ScopedLock&);
- bool isPrimary(const sys::Mutex::ScopedLock&) { return !backup.get(); }
+ void setClientUrl(const Url&);
+ void setBrokerUrl(const Url&);
+ void updateClientUrl(sys::Mutex::ScopedLock&);
+
+ bool isPrimary(sys::Mutex::ScopedLock&) { return !backup.get(); }
+
+ void setStatus(BrokerStatus, sys::Mutex::ScopedLock&);
+ void recover();
+ void statusChanged(sys::Mutex::ScopedLock&);
+ void setLinkProperties(sys::Mutex::ScopedLock&);
+
std::vector<Url> getKnownBrokers() const;
+ void membershipUpdated(sys::Mutex::ScopedLock&);
+
+ std::string logPrefix;
broker::Broker& broker;
+ types::Uuid systemId;
const Settings settings;
- sys::Mutex lock;
+ mutable sys::Mutex lock;
+ boost::shared_ptr<ConnectionObserver> observer; // Used by Backup and Primary
std::auto_ptr<Backup> backup;
+ std::auto_ptr<Primary> primary;
qmf::org::apache::qpid::ha::HaBroker* mgmtObject;
Url clientUrl, brokerUrl;
std::vector<Url> knownBrokers;
- size_t expectedBackups;
+ BrokerStatus status;
+ BrokerInfo brokerInfo;
+ Membership membership;
+ ReplicationTest replicationTest;
};
}} // namespace qpid::ha
Modified: qpid/branches/asyncstore/cpp/src/qpid/ha/HaPlugin.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/ha/HaPlugin.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/ha/HaPlugin.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/ha/HaPlugin.cpp Fri Aug 3 12:13:32 2012
@@ -20,7 +20,7 @@
#include "qpid/Plugin.h"
#include "qpid/Options.h"
#include "qpid/broker/Broker.h"
-
+#include <boost/bind.hpp>
namespace qpid {
namespace ha {
@@ -33,21 +33,21 @@ struct Options : public qpid::Options {
addOptions()
("ha-cluster", optValue(settings.cluster, "yes|no"),
"Join a HA active/passive cluster.")
- ("ha-brokers", optValue(settings.brokerUrl,"URL"),
- "URL that backup brokers use to connect and fail over.")
- ("ha-public-brokers", optValue(settings.clientUrl,"URL"),
- "URL that clients use to connect and fail over, defaults to ha-brokers.")
+ ("ha-brokers-url", optValue(settings.brokerUrl,"URL"),
+ "URL with address of each broker in the cluster.")
+ ("ha-public-url", optValue(settings.clientUrl,"URL"),
+ "URL advertized to clients to connect to the cluster.")
("ha-replicate",
optValue(settings.replicateDefault, "LEVEL"),
"Replication level for creating queues and exchanges if there is no qpid.replicate argument supplied. LEVEL is 'none', 'configuration' or 'all'")
- ("ha-expected-backups", optValue(settings.expectedBackups, "N"),
- "Number of backups expected to be active in the HA cluster.")
("ha-username", optValue(settings.username, "USER"),
"Username for connections between HA brokers")
("ha-password", optValue(settings.password, "PASS"),
"Password for connections between HA brokers")
("ha-mechanism", optValue(settings.mechanism, "MECH"),
"Authentication mechanism for connections between HA brokers")
+ ("ha-backup-timeout", optValue(settings.backupTimeout, "SECONDS"),
+ "Maximum time to wait for an expected backup to connect and become ready.")
;
}
};
@@ -62,14 +62,26 @@ struct HaPlugin : public Plugin {
Options* getOptions() { return &options; }
- void earlyInitialize(Plugin::Target& ) {}
+ void earlyInitialize(Plugin::Target& target) {
+ broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
+ if (broker) {
+ // Must create the HaBroker in earlyInitialize so it can set up its
+ // connection observer before clients start connecting.
+ haBroker.reset(new ha::HaBroker(*broker, settings));
+ broker->addFinalizer(boost::bind(&HaPlugin::finalize, this));
+ }
+ }
void initialize(Plugin::Target& target) {
broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
- if (broker) haBroker.reset(new ha::HaBroker(*broker, settings));
+ if (broker) haBroker->initialize();
+ }
+
+ void finalize() {
+ haBroker.reset();
}
};
-static HaPlugin instance; // Static initialization.
+HaPlugin instance; // Static initialization.
}} // namespace qpid::ha
Modified: qpid/branches/asyncstore/cpp/src/qpid/ha/QueueReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/ha/QueueReplicator.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/ha/QueueReplicator.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/ha/QueueReplicator.cpp Fri Aug 3 12:13:32 2012
@@ -19,6 +19,7 @@
*
*/
+#include "HaBroker.h"
#include "QueueReplicator.h"
#include "ReplicatingSubscription.h"
#include "qpid/broker/Bridge.h"
@@ -44,25 +45,39 @@ namespace ha {
using namespace broker;
using namespace framing;
-const std::string QueueReplicator::DEQUEUE_EVENT_KEY("qpid.dequeue-event");
-const std::string QueueReplicator::POSITION_EVENT_KEY("qpid.position-event");
+const std::string QPID_HA_EVENT_PREFIX("qpid.ha-event:");
+const std::string QueueReplicator::DEQUEUE_EVENT_KEY(QPID_HA_EVENT_PREFIX+"dequeue");
+const std::string QueueReplicator::POSITION_EVENT_KEY(QPID_HA_EVENT_PREFIX+"position");
std::string QueueReplicator::replicatorName(const std::string& queueName) {
return QPID_REPLICATOR_ + queueName;
}
-QueueReplicator::QueueReplicator(boost::shared_ptr<Queue> q, boost::shared_ptr<Link> l)
- : Exchange(replicatorName(q->getName()), 0, q->getBroker()), queue(q), link(l)
+bool QueueReplicator::isEventKey(const std::string key) {
+ const std::string& prefix = QPID_HA_EVENT_PREFIX;
+ bool ret = key.size() > prefix.size() && key.compare(0, prefix.size(), prefix) == 0;
+ return ret;
+}
+
+QueueReplicator::QueueReplicator(HaBroker& hb,
+ boost::shared_ptr<Queue> q,
+ boost::shared_ptr<Link> l)
+ : Exchange(replicatorName(q->getName()), 0, q->getBroker()),
+ haBroker(hb),
+ logPrefix("Backup queue "+q->getName()+": "),
+ queue(q), link(l), brokerInfo(hb.getBrokerInfo())
{
- logPrefix = "HA: Backup of " + queue->getName() + ": ";
- QPID_LOG(info, logPrefix << "Created");
+ Uuid uuid(true);
+ bridgeName = replicatorName(q->getName()) + std::string(".") + uuid.str();
}
// This must be separate from the constructor so we can call shared_from_this.
void QueueReplicator::activate() {
- // Note this may create a new bridge or use an existing one.
+ sys::Mutex::ScopedLock l(lock);
+ std::pair<Bridge::shared_ptr, bool> result =
queue->getBroker()->getLinks().declare(
- link->getHost(), link->getPort(),
+ bridgeName,
+ *link,
false, // durable
queue->getName(), // src
getName(), // dest
@@ -77,43 +92,47 @@ void QueueReplicator::activate() {
// before initializeBridge is called.
boost::bind(&QueueReplicator::initializeBridge, shared_from_this(), _1, _2)
);
+ bridge = result.first;
}
-QueueReplicator::~QueueReplicator() {}
+QueueReplicator::~QueueReplicator() { deactivate(); }
void QueueReplicator::deactivate() {
+ // destroy the route
sys::Mutex::ScopedLock l(lock);
- queue->getBroker()->getLinks().destroy(
- link->getHost(), link->getPort(), queue->getName(), getName(), string());
- QPID_LOG(debug, logPrefix << "Deactivated bridge " << bridgeName);
+ if (bridge) {
+ bridge->close();
+ bridge.reset();
+ QPID_LOG(debug, logPrefix << "Deactivated bridge " << bridgeName);
+ }
}
// Called in a broker connection thread when the bridge is created.
void QueueReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionHandler) {
sys::Mutex::ScopedLock l(lock);
- bridgeName = bridge.getName();
- framing::AMQP_ServerProxy peer(sessionHandler.out);
+ AMQP_ServerProxy peer(sessionHandler.out);
const qmf::org::apache::qpid::broker::ArgsLinkBridge& args(bridge.getArgs());
- framing::FieldTable settings;
-
- // FIXME aconway 2011-12-09: Failover optimization removed.
- // There was code here to re-use messages already on the backup
- // during fail-over. This optimization was removed to simplify
- // the logic till we get the basic replication stable, it
- // can be re-introduced later. Last revision with the optimization:
- // r1213258 | QPID-3603: Fix QueueReplicator subscription parameters.
-
- // Clear out any old messages, reset the queue to start replicating fresh.
- queue->purge();
- queue->setPosition(0);
-
+ FieldTable settings;
settings.setInt(ReplicatingSubscription::QPID_REPLICATING_SUBSCRIPTION, 1);
- // TODO aconway 2011-12-19: optimize.
- settings.setInt(QPID_SYNC_FREQUENCY, 1);
- peer.getMessage().subscribe(args.i_src, args.i_dest, 0/*accept-explicit*/, 1/*not-acquired*/, false/*exclusive*/, "", 0, settings);
+ settings.setInt(QPID_SYNC_FREQUENCY, 1); // FIXME aconway 2012-05-22: optimize?
+ settings.setInt(ReplicatingSubscription::QPID_BACK,
+ queue->getPosition());
+ settings.setTable(ReplicatingSubscription::QPID_BROKER_INFO,
+ brokerInfo.asFieldTable());
+ SequenceNumber front;
+ if (ReplicatingSubscription::getFront(*queue, front))
+ settings.setInt(ReplicatingSubscription::QPID_FRONT, front);
+ peer.getMessage().subscribe(
+ args.i_src, args.i_dest, 0/*accept-explicit*/, 1/*not-acquired*/,
+ false/*exclusive*/, "", 0, settings);
+ // FIXME aconway 2012-05-22: use a finite credit window?
peer.getMessage().flow(getName(), 0, 0xFFFFFFFF);
peer.getMessage().flow(getName(), 1, 0xFFFFFFFF);
- QPID_LOG(debug, logPrefix << "Activated bridge " << bridgeName);
+
+ qpid::Address primary;
+ link->getRemoteAddress(primary);
+ QPID_LOG(info, logPrefix << "Connected to " << primary << "(" << bridgeName << ")");
+ QPID_LOG(trace, logPrefix << "Subscription settings: " << settings);
}
namespace {
@@ -127,13 +146,11 @@ template <class T> T decodeContent(Messa
}
}
-void QueueReplicator::dequeue(SequenceNumber n, const sys::Mutex::ScopedLock&) {
+void QueueReplicator::dequeue(SequenceNumber n, sys::Mutex::ScopedLock&) {
// Thread safe: only calls thread safe Queue functions.
- if (queue->getPosition() >= n) { // Ignore messages we haven't reached yet
- QueuedMessage message;
- if (queue->acquireMessageAt(n, message))
- queue->dequeue(0, message);
- }
+ QueuedMessage message;
+ if (queue->acquireMessageAt(n, message))
+ queue->dequeue(0, message);
}
// Called in connection thread of the queues bridge to primary.
@@ -142,29 +159,33 @@ void QueueReplicator::route(Deliverable&
try {
const std::string& key = msg.getMessage().getRoutingKey();
sys::Mutex::ScopedLock l(lock);
- if (key == DEQUEUE_EVENT_KEY) {
+ if (!isEventKey(key)) {
+ msg.deliverTo(queue);
+ // We are on a backup so the queue is not modified except via this.
+ QPID_LOG(trace, logPrefix << "Enqueued message " << queue->getPosition());
+ }
+ else if (key == DEQUEUE_EVENT_KEY) {
SequenceSet dequeues = decodeContent<SequenceSet>(msg.getMessage());
QPID_LOG(trace, logPrefix << "Dequeue: " << dequeues);
//TODO: should be able to optimise the following
for (SequenceSet::iterator i = dequeues.begin(); i != dequeues.end(); i++)
dequeue(*i, l);
- } else if (key == POSITION_EVENT_KEY) {
+ }
+ else if (key == POSITION_EVENT_KEY) {
SequenceNumber position = decodeContent<SequenceNumber>(msg.getMessage());
QPID_LOG(trace, logPrefix << "Position moved from " << queue->getPosition()
<< " to " << position);
- if (queue->getPosition() > position) {
- throw Exception(
- QPID_MSG(logPrefix << "Invalid position update from "
- << queue->getPosition() << " to " << position));
- }
+ // Verify that there are no messages after the new position in the queue.
+ SequenceNumber next;
+ if (ReplicatingSubscription::getNext(*queue, position, next))
+ throw Exception("Invalid position move, preceeds messages");
queue->setPosition(position);
- } else {
- msg.deliverTo(queue);
- QPID_LOG(trace, logPrefix << "Enqueued message " << queue->getPosition());
}
+ // Ignore unknown event keys, may be introduced in later versions.
}
catch (const std::exception& e) {
QPID_LOG(critical, logPrefix << "Replication failed: " << e.what());
+ haBroker.shutdown();
throw;
}
}
Modified: qpid/branches/asyncstore/cpp/src/qpid/ha/QueueReplicator.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/ha/QueueReplicator.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/ha/QueueReplicator.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/ha/QueueReplicator.h Fri Aug 3 12:13:32 2012
@@ -21,6 +21,8 @@
* under the License.
*
*/
+
+#include "BrokerInfo.h"
#include "qpid/broker/Exchange.h"
#include "qpid/framing/SequenceSet.h"
#include <boost/enable_shared_from_this.hpp>
@@ -38,6 +40,7 @@ class Deliverable;
}
namespace ha {
+class HaBroker;
/**
* Exchange created on a backup broker to replicate a queue on the primary.
@@ -55,8 +58,13 @@ class QueueReplicator : public broker::E
static const std::string DEQUEUE_EVENT_KEY;
static const std::string POSITION_EVENT_KEY;
static std::string replicatorName(const std::string& queueName);
+ /** Test if a string is an event key */
+ static bool isEventKey(const std::string key);
+
+ QueueReplicator(HaBroker&,
+ boost::shared_ptr<broker::Queue> q,
+ boost::shared_ptr<broker::Link> l);
- QueueReplicator(boost::shared_ptr<broker::Queue> q, boost::shared_ptr<broker::Link> l);
~QueueReplicator();
void activate(); // Call after ctor
@@ -71,13 +79,16 @@ class QueueReplicator : public broker::E
private:
void initializeBridge(broker::Bridge& bridge, broker::SessionHandler& sessionHandler);
- void dequeue(framing::SequenceNumber, const sys::Mutex::ScopedLock&);
+ void dequeue(framing::SequenceNumber, sys::Mutex::ScopedLock&);
+ HaBroker& haBroker;
std::string logPrefix;
std::string bridgeName;
sys::Mutex lock;
boost::shared_ptr<broker::Queue> queue;
boost::shared_ptr<broker::Link> link;
+ boost::shared_ptr<broker::Bridge> bridge;
+ BrokerInfo brokerInfo;
};
}} // namespace qpid::ha
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org