QPID-4632 - Improvement to queue threshold alerting

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1454601 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 67 insertions, 78 deletions

diff --git a/qpid/cpp/src/qpid/broker/QueueSettings.cpp b/qpid/cpp/src/qpid/broker/QueueSettings.cpp
index b92a81bcf3..cb2b11621a 100644
--- a/qpid/cpp/src/qpid/broker/QueueSettings.cpp
+++ b/qpid/cpp/src/qpid/broker/QueueSettings.cpp
@@ -47,6 +47,10 @@ const std::string AUTO_DELETE_TIMEOUT("qpid.auto_delete_timeout");
 const std::string ALERT_REPEAT_GAP("qpid.alert_repeat_gap");
 const std::string ALERT_COUNT("qpid.alert_count");
 const std::string ALERT_SIZE("qpid.alert_size");
+const std::string ALERT_COUNT_UP("qpid.alert_count_up");
+const std::string ALERT_SIZE_UP("qpid.alert_size_up");
+const std::string ALERT_COUNT_DOWN("qpid.alert_count_down");
+const std::string ALERT_SIZE_DOWN("qpid.alert_size_down");
 const std::string PRIORITIES("qpid.priorities");
 const std::string FAIRSHARE("qpid.fairshare");
 const std::string FAIRSHARE_ALIAS("x-qpid-fairshare");
@@ -165,12 +169,18 @@ bool QueueSettings::handle(const std::string& key, const qpid::types::Variant& v
     } else if (key == ALERT_REPEAT_GAP) {
         alertRepeatInterval = value;
         return true;
-    } else if (key == ALERT_COUNT) {
+    } else if ((key == ALERT_COUNT) || (key == ALERT_COUNT_UP)) {
         alertThreshold.setCount(value);
         return true;
-    } else if (key == ALERT_SIZE) {
+    } else if ((key == ALERT_SIZE) || (key == ALERT_SIZE_UP)) {
         alertThreshold.setSize(value);
         return true;
+    } else if (key == ALERT_COUNT_DOWN) {
+        alertThresholdDown.setCount(value);
+        return true;
+    } else if (key == ALERT_SIZE_DOWN) {
+        alertThresholdDown.setSize(value);
+        return true;
     } else if (key == MAX_FILE_COUNT && value.asUint64() > 0) {
         maxFileCount = value.asUint64();
         return false; // 'handle' here and also pass to store
diff --git a/qpid/cpp/src/qpid/broker/QueueSettings.h b/qpid/cpp/src/qpid/broker/QueueSettings.h
index 62d34db5cb..ec012c578f 100644
--- a/qpid/cpp/src/qpid/broker/QueueSettings.h
+++ b/qpid/cpp/src/qpid/broker/QueueSettings.h
@@ -71,6 +71,7 @@ struct QueueSettings
 
     //threshold events:
     QueueDepth alertThreshold;
+    QueueDepth alertThresholdDown;
     int64_t alertRepeatInterval;
 
     //file limits checked by Acl and shared with storeSettings
diff --git a/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp b/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp
index 9b4e948e4f..345b9d89d5 100644
--- a/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp
+++ b/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp
@@ -21,84 +21,44 @@
 #include "qpid/broker/ThresholdAlerts.h"
 #include "qpid/broker/Queue.h"
 #include "qpid/broker/Message.h"
-#include "qpid/broker/amqp_0_10/MessageTransfer.h"
-#include "qpid/amqp_0_10/Codecs.h"
 #include "qpid/log/Statement.h"
 #include "qpid/management/ManagementAgent.h"
+#include "qmf/org/apache/qpid/broker/EventQueueThresholdCrossedUpward.h"
+#include "qmf/org/apache/qpid/broker/EventQueueThresholdCrossedDownward.h"
 #include "qmf/org/apache/qpid/broker/EventQueueThresholdExceeded.h"
 
 namespace qpid {
 namespace broker {
-namespace {
-const qmf::org::apache::qpid::broker::EventQueueThresholdExceeded EVENT("dummy", 0, 0);
-bool isQMFv2(const Message& message)
-{
-    const qpid::framing::MessageProperties* props = qpid::broker::amqp_0_10::MessageTransfer::get(message).getProperties<qpid::framing::MessageProperties>();
-    return props && props->getAppId() == "qmf2";
-}
-
-bool isThresholdEvent(const Message& message)
-{
-    if (message.getIsManagementMessage()) {
-        //is this a qmf event? if so is it a threshold event?
-        if (isQMFv2(message)) {
-            if (message.getPropertyAsString("qmf.content") == "_event") {
-                //decode as list
-                std::string content = qpid::broker::amqp_0_10::MessageTransfer::get(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 = qpid::broker::amqp_0_10::MessageTransfer::get(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,
-                                 const uint64_t st,
-                                 const long repeat)
-    : name(n), agent(a), countThreshold(ct), sizeThreshold(st),
-      repeatInterval(repeat ? repeat*qpid::sys::TIME_SEC : 0),
-      count(0), size(0), lastAlert(qpid::sys::EPOCH) {}
+                                 const uint32_t ctu,
+                                 const uint32_t ctd,
+                                 const uint64_t stu,
+                                 const uint64_t std,
+                                 const bool bw)
+    : name(n), agent(a),
+      countThreshold(ctu), countThresholdDown(ctd),
+      sizeThreshold(stu), sizeThresholdDown(std),
+      count(0), size(0), countGoingUp(true), sizeGoingUp(true), backwardCompat(bw)  {}
 
 void ThresholdAlerts::enqueued(const Message& m)
 {
     size += m.getContentSize();
     ++count;
-    if ((countThreshold && count >= countThreshold) || (sizeThreshold && size >= sizeThreshold)) {
-        if ((repeatInterval == 0 && lastAlert == qpid::sys::EPOCH)
-            || qpid::sys::Duration(lastAlert, qpid::sys::now()) > repeatInterval) {
-            //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)) return;
-            lastAlert = qpid::sys::now();
+
+    if (sizeGoingUp && sizeThreshold && size >= sizeThreshold) {
+        sizeGoingUp = false;
+        agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdCrossedUpward(name, count, size));
+        if (backwardCompat)
+            agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdExceeded(name, count, size));
+    }
+
+    if (countGoingUp && countThreshold && count >= countThreshold) {
+        countGoingUp = false;
+        agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdCrossedUpward(name, count, size));
+        if (backwardCompat)
             agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdExceeded(name, count, size));
-            QPID_LOG(info, "Threshold event triggered for " << name << ", count=" << count << ", size=" << size);
-        }
     }
 }
 
@@ -106,21 +66,32 @@ void ThresholdAlerts::dequeued(const Message& m)
 {
     size -= m.getContentSize();
     --count;
-    if ((countThreshold && count < countThreshold) || (sizeThreshold && size < sizeThreshold)) {
-        lastAlert = qpid::sys::EPOCH;
+
+    if (!sizeGoingUp && sizeThreshold && size <= sizeThresholdDown) {
+        sizeGoingUp = true;
+        agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdCrossedDownward(name, count, size));
+    }
+
+    if (!countGoingUp && countThreshold && count <= countThresholdDown) {
+        countGoingUp = true;
+        agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdCrossedDownward(name, count, size));
     }
 }
 
 
 
 void ThresholdAlerts::observe(Queue& queue, qpid::management::ManagementAgent& agent,
-                              const uint64_t countThreshold,
-                              const uint64_t sizeThreshold,
-                              const long repeatInterval)
+                              const uint64_t ctu,
+                              const uint64_t _ctd,
+                              const uint64_t stu,
+                              const uint64_t _std)
 {
-    if (countThreshold || sizeThreshold) {
+    if (ctu || stu) {
+        uint64_t ctd = (_ctd == 0 || _ctd >= ctu) ? ctu >> 1 : _ctd;
+        uint64_t std = (_std == 0 || _std >= stu) ? stu >> 1 : _std;
+
         boost::shared_ptr<QueueObserver> observer(
-            new ThresholdAlerts(queue.getName(), agent, countThreshold, sizeThreshold, repeatInterval)
+        new ThresholdAlerts(queue.getName(), agent, ctu, ctd, stu, std, (_ctd == 0 && _std == 0))
         );
         queue.addObserver(observer);
     }
@@ -133,8 +104,10 @@ void ThresholdAlerts::observe(Queue& queue, qpid::management::ManagementAgent& a
     //percentage of any limit from the policy.
     uint32_t countThreshold = settings.alertThreshold.hasCount() ? settings.alertThreshold.getCount() : (settings.maxDepth.getCount()*limitRatio/100);
     uint32_t sizeThreshold = settings.alertThreshold.hasSize() ? settings.alertThreshold.getSize() : (settings.maxDepth.getSize()*limitRatio/100);
+    uint32_t countThresholdDown = settings.alertThresholdDown.hasCount() ? settings.alertThresholdDown.getCount() : 0;
+    uint32_t sizeThresholdDown = settings.alertThresholdDown.hasSize() ? settings.alertThresholdDown.getSize() : 0;
 
-    observe(queue, agent, countThreshold, sizeThreshold, settings.alertRepeatInterval);
+    observe(queue, agent, countThreshold, countThresholdDown , sizeThreshold, sizeThresholdDown);
 }
 
 }}
diff --git a/qpid/cpp/src/qpid/broker/ThresholdAlerts.h b/qpid/cpp/src/qpid/broker/ThresholdAlerts.h
index 4f985522e2..a8ff5270f3 100644
--- a/qpid/cpp/src/qpid/broker/ThresholdAlerts.h
+++ b/qpid/cpp/src/qpid/broker/ThresholdAlerts.h
@@ -22,7 +22,6 @@
  *
  */
 #include "qpid/broker/QueueObserver.h"
-#include "qpid/sys/Time.h"
 #include "qpid/types/Variant.h"
 #include <string>
 
@@ -44,8 +43,10 @@ class ThresholdAlerts : public QueueObserver
     ThresholdAlerts(const std::string& name,
                     qpid::management::ManagementAgent& agent,
                     const uint32_t countThreshold,
+                    const uint32_t countThresholdDown,
                     const uint64_t sizeThreshold,
-                    const long repeatInterval);
+                    const uint64_t sizeThresholdDown,
+                    const bool backwardCompat);
     void enqueued(const Message&);
     void dequeued(const Message&);
     void acquired(const Message&) {};
@@ -53,19 +54,23 @@ class ThresholdAlerts : public QueueObserver
 
     static void observe(Queue& queue, qpid::management::ManagementAgent& agent,
                         const uint64_t countThreshold,
+                        const uint64_t countThresholdDown,
                         const uint64_t sizeThreshold,
-                        const long repeatInterval);
+                        const uint64_t sizeThresholdDown);
     static void observe(Queue& queue, qpid::management::ManagementAgent& agent,
                         const QueueSettings& settings, uint16_t limitRatio);
   private:
     const std::string name;
     qpid::management::ManagementAgent& agent;
     const uint32_t countThreshold;
+    const uint32_t countThresholdDown;
     const uint64_t sizeThreshold;
-    const qpid::sys::Duration repeatInterval;
+    const uint64_t sizeThresholdDown;
     uint64_t count;
     uint64_t size;
-    qpid::sys::AbsTime lastAlert;
+    bool countGoingUp;
+    bool sizeGoingUp;
+    bool backwardCompat;
 };
 }} // namespace qpid::broker