Merged out from trunkQPID-2519

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-2519@1187375 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 60 insertions, 8 deletions

diff --git a/cpp/src/qpid/broker/ThresholdAlerts.cpp b/cpp/src/qpid/broker/ThresholdAlerts.cpp
index 4f35884af8..3c9e210d4d 100644
--- a/cpp/src/qpid/broker/ThresholdAlerts.cpp
+++ b/cpp/src/qpid/broker/ThresholdAlerts.cpp
@@ -28,6 +28,52 @@
 
 namespace qpid {
 namespace broker {
+namespace {
+const qmf::org::apache::qpid::broker::EventQueueThresholdExceeded EVENT("dummy", 0, 0);
+bool isQMFv2(const boost::intrusive_ptr<Message> message)
+{
+    const qpid::framing::MessageProperties* props = message->getProperties<qpid::framing::MessageProperties>();
+    return props && props->getAppId() == "qmf2";
+}
+
+bool isThresholdEvent(const boost::intrusive_ptr<Message> message)
+{
+    if (message->getIsManagementMessage()) {
+        //is this a qmf event? if so is it a threshold event?
+        if (isQMFv2(message)) {
+            const qpid::framing::FieldTable* headers = message->getApplicationHeaders();
+            if (headers && headers->getAsString("qmf.content") == "_event") {
+                //decode as list
+                std::string content = message->getFrames().getContent();
+                qpid::types::Variant::List list;
+                qpid::amqp_0_10::ListCodec::decode(content, list);
+                if (list.empty() || list.front().getType() != qpid::types::VAR_MAP) return false;
+                qpid::types::Variant::Map map = list.front().asMap();
+                try {
+                    std::string eventName = map["_schema_id"].asMap()["_class_name"].asString();
+                    return eventName == EVENT.getEventName();
+                } catch (const std::exception& e) {
+                    QPID_LOG(error, "Error checking for recursive threshold alert: " << e.what());
+                }
+            }
+        } else {
+            std::string content = message->getFrames().getContent();
+            qpid::framing::Buffer buffer(const_cast<char*>(content.data()), content.size());
+            if (buffer.getOctet() == 'A' && buffer.getOctet() == 'M' && buffer.getOctet() == '2' && buffer.getOctet() == 'e') {
+                buffer.getLong();//sequence
+                std::string packageName;
+                buffer.getShortString(packageName);
+                if (packageName != EVENT.getPackageName()) return false;
+                std::string eventName;
+                buffer.getShortString(eventName);
+                return eventName == EVENT.getEventName();
+            }
+        }
+    }
+    return false;
+}
+}
+
 ThresholdAlerts::ThresholdAlerts(const std::string& n,
                                  qpid::management::ManagementAgent& a,
                                  const uint32_t ct,
@@ -44,8 +90,14 @@ void ThresholdAlerts::enqueued(const QueuedMessage& m)
     if ((countThreshold && count >= countThreshold) || (sizeThreshold && size >= sizeThreshold)) {
         if ((repeatInterval == 0 && lastAlert == qpid::sys::EPOCH)
             || qpid::sys::Duration(lastAlert, qpid::sys::now()) > repeatInterval) {
-            agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdExceeded(name, count, size));
+            //Note: Raising an event may result in messages being
+            //enqueued on queues; it may even be that this event
+            //causes a message to be enqueued on the queue we are
+            //tracking, and so we need to avoid recursing
+            if (isThresholdEvent(m.payload)) return;
             lastAlert = qpid::sys::now();
+            agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdExceeded(name, count, size));
+            QPID_LOG(info, "Threshold event triggered for " << name << ", count=" << count << ", size=" << size);
         }
     }
 }
@@ -75,12 +127,12 @@ void ThresholdAlerts::observe(Queue& queue, qpid::management::ManagementAgent& a
 }
 
 void ThresholdAlerts::observe(Queue& queue, qpid::management::ManagementAgent& agent,
-                              const qpid::framing::FieldTable& settings)
+                              const qpid::framing::FieldTable& settings, uint16_t limitRatio)
 
 {
     qpid::types::Variant::Map map;
     qpid::amqp_0_10::translate(settings, map);
-    observe(queue, agent, map);
+    observe(queue, agent, map, limitRatio);
 }
 
 template <class T>
@@ -118,19 +170,19 @@ class Option
 };
 
 void ThresholdAlerts::observe(Queue& queue, qpid::management::ManagementAgent& agent,
-                              const qpid::types::Variant::Map& settings)
+                              const qpid::types::Variant::Map& settings, uint16_t limitRatio)
 
 {
     //Note: aliases are keys defined by java broker
     Option<int64_t> repeatInterval("qpid.alert_repeat_gap", 60);
     repeatInterval.addAlias("x-qpid-minimum-alert-repeat-gap");
 
-    //If no explicit threshold settings were given use 80% of any
-    //limit from the policy.
+    //If no explicit threshold settings were given use specified
+    //percentage of any limit from the policy.
     const QueuePolicy* policy = queue.getPolicy();
-    Option<uint32_t> countThreshold("qpid.alert_count", (uint32_t) (policy ? policy->getMaxCount()*0.8 : 0));
+    Option<uint32_t> countThreshold("qpid.alert_count", (uint32_t) (policy && limitRatio ? (policy->getMaxCount()*limitRatio/100) : 0));
     countThreshold.addAlias("x-qpid-maximum-message-count");
-    Option<uint64_t> sizeThreshold("qpid.alert_size", (uint64_t) (policy ? policy->getMaxSize()*0.8 : 0));
+    Option<uint64_t> sizeThreshold("qpid.alert_size", (uint64_t) (policy && limitRatio ? (policy->getMaxSize()*limitRatio/100) : 0));
     sizeThreshold.addAlias("x-qpid-maximum-message-size");
 
     observe(queue, agent, countThreshold.get(settings), sizeThreshold.get(settings), repeatInterval.get(settings));