QPID-4632 - Improvement to queue threshold alerting

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

6 files changed, 220 insertions, 92 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
 
diff --git a/qpid/specs/management-schema.xml b/qpid/specs/management-schema.xml
index 678c1a6476..b3514e1f70 100644
--- a/qpid/specs/management-schema.xml
+++ b/qpid/specs/management-schema.xml
@@ -523,6 +523,10 @@
   <event name="unbind"            sev="inform" args="rhost, user, exName, qName, key"/>
   <event name="subscribe"         sev="inform" args="rhost, user, qName, dest, excl, args"/>
   <event name="unsubscribe"       sev="inform" args="rhost, user, dest"/>
-  <event name="queueThresholdExceeded"       sev="warn" args="qName, msgDepth, byteDepth"/>
+  <event name="queueThresholdCrossedUpward"   sev="inform" args="qName, msgDepth, byteDepth"/>
+  <event name="queueThresholdCrossedDownward" sev="inform" args="qName, msgDepth, byteDepth"/>
+
+  <!-- The following are deprecated -->
+  <event name="queueThresholdExceeded" sev="warn" args="qName, msgDepth, byteDepth"/>
 </schema>
 
diff --git a/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py b/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py
index 0ad2295822..a2ffddb7ee 100644
--- a/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py
+++ b/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py
@@ -32,46 +32,181 @@ class ThresholdTests (Base):
     def setup_session(self):
         return self.conn.session()
 
-    def do_threshold_test(self, key, value, messages):
-        rcv = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdExceeded.#")
-        snd = self.ssn.sender("ttq; {create:always, node: {x-declare:{auto_delete:True,exclusive:True,arguments:{'%s':%s}}}}" % (key, value))
+    def enqueue(self, snd, count):
+        for i in range(count):
+            m = Message("msg-%d" % i)
+            snd.send(m)
+
+    def dequeue(self, rcv, count):
+        for i in range(count):
+            m = rcv.fetch(timeout=1)
+        self.ssn.acknowledge()
+
+    def check_events(self, rcv, count):
+        for i in range(count):
+            m = rcv.fetch(timeout=0)
+        try:
+            m = rcv.fetch(timeout=0)
+            assert False
+        except:
+            pass
+
+    def do_threshold_test(self, args, messages, drain_count, bw_compat=None):
+        astr = ''
+        first = True
+        for key, value in args.items():
+            if first:
+                first = None
+            else:
+                astr += ','
+            astr += "'%s':%s" % (key, value)
+        rcvUp = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedUpward.#")
+        rcvDn = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedDownward.#")
+        rcvBw = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdExceeded.#")
+        snd = self.ssn.sender("ttq; {create:always, node: {x-declare:{auto_delete:True,exclusive:True,arguments:{%s}}}}" % astr)
+        rcv = self.ssn.receiver("ttq")
         size = 0
         count = 0
         for m in messages:
             snd.send(m)
             count = count + 1
             size = size + len(m.content)
-        event = rcv.fetch(timeout=1)
+        event = rcvUp.fetch(timeout=1)
         schema = event.content[0]["_schema_id"]
-        assert schema["_class_name"] == "queueThresholdExceeded"
+        assert schema["_class_name"] == "queueThresholdCrossedUpward"
         values = event.content[0]["_values"]
         assert values["qName"] == "ttq"
         assert values["msgDepth"] == count, "msgDepth %s, expected %s" % (values["msgDepth"], count)
         assert values["byteDepth"] == size, "byteDepth %s, expected %s" % (values["byteDepth"], size)
+        if bw_compat:
+            event = rcvBw.fetch(timeout=0)
+
+        try:
+            event = rcvUp.fetch(timeout=0)
+            assert False
+        except:
+            pass
+
+        if drain_count > 0:
+            for i in range(drain_count):
+                m = rcv.fetch(timeout=1)
+                self.ssn.acknowledge()
+                count -= 1
+                size -= len(m.content)
+            event = rcvDn.fetch(timeout=1)
+            schema = event.content[0]["_schema_id"]
+            assert schema["_class_name"] == "queueThresholdCrossedDownward"
+            values = event.content[0]["_values"]
+            assert values["qName"] == "ttq"
+            assert values["msgDepth"] == count, "msgDepth %s, expected %s" % (values["msgDepth"], count)
+            assert values["byteDepth"] == size, "byteDepth %s, expected %s" % (values["byteDepth"], size)
+            try:
+                event = rcvUp.fetch(timeout=0)
+                assert False
+            except:
+                pass
 
     def test_alert_count(self):
-        self.do_threshold_test("qpid.alert_count", 5, [Message("msg-%s" % i) for i in range(5)])
+        a = {'qpid.alert_count':5, 'qpid.alert_count_down':3}
+        self.do_threshold_test(a, [Message("msg-%s" % i) for i in range(5)], 2)
 
     def test_alert_size(self):
-        self.do_threshold_test("qpid.alert_size", 25, [Message("msg-%s" % i) for i in range(5)])
+        a = {'qpid.alert_size_up':25,'qpid.alert_size_down':15}
+        self.do_threshold_test(a, [Message("msg-%s" % i) for i in range(5)], 2)
 
     def test_alert_count_alias(self):
-        self.do_threshold_test("x-qpid-maximum-message-count", 10, [Message("msg-%s" % i) for i in range(10)])
+        a = {'x-qpid-maximum-message-count':10}
+        self.do_threshold_test(a, [Message("msg-%s" % i) for i in range(10)], 0, True)
 
     def test_alert_size_alias(self):
-        self.do_threshold_test("x-qpid-maximum-message-size", 15, [Message("msg-%s" % i) for i in range(3)])
+        a = {'x-qpid-maximum-message-size':15}
+        self.do_threshold_test(a, [Message("msg-%s" % i) for i in range(3)], 0, True)
 
     def test_alert_on_alert_queue(self):
-        rcv = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdExceeded.#; {link:{x-declare:{arguments:{'qpid.alert_count':1}}}}")
+        rcv = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedUpward.#; {link:{x-declare:{arguments:{'qpid.alert_count':1}}}}")
         rcvQMFv1 = self.ssn.receiver("qpid.management/console.event.#; {link:{x-declare:{arguments:{'qpid.alert_count':1}}}}")
         snd = self.ssn.sender("ttq; {create:always, node: {x-declare:{auto_delete:True,exclusive:True,arguments:{'qpid.alert_count':1}}}}")
         snd.send(Message("my-message"))
         queues = []
-        for i in range(2):
-            event = rcv.fetch()
+        for i in range(3):
+            event = rcv.fetch(timeout=1)
             schema = event.content[0]["_schema_id"]
-            assert schema["_class_name"] == "queueThresholdExceeded"
+            assert schema["_class_name"] == "queueThresholdCrossedUpward"
             values = event.content[0]["_values"]
             queues.append(values["qName"])
         assert "ttq" in queues, "expected event for ttq (%s)" % (queues)
 
+    def test_hysteresis(self):
+        astr  = "'qpid.alert_count_up':10,'qpid.alert_count_down':5"
+        rcvUp = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedUpward.#")
+        rcvDn = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedDownward.#")
+        snd   = self.ssn.sender("thq; {create:always, node: {x-declare:{auto_delete:True,exclusive:True,arguments:{%s}}}}" % astr)
+        rcv   = self.ssn.receiver("thq")
+
+        rcvUp.capacity = 5
+        rcvDn.capacity = 5
+        rcv.capacity   = 5
+
+        self.enqueue(snd, 8)         # depth = 8
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.dequeue(rcv, 6)         # depth = 2
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.enqueue(snd, 8)         # depth = 10
+        self.check_events(rcvUp, 1)
+        self.check_events(rcvDn, 0)
+
+        self.dequeue(rcv, 1)         # depth = 9
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.enqueue(snd, 1)         # depth = 10
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.enqueue(snd, 10)        # depth = 20
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.dequeue(rcv, 5)         # depth = 15
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.dequeue(rcv, 12)        # depth = 3
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 1)
+
+        self.dequeue(rcv, 1)         # depth = 2
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.enqueue(snd, 6)         # depth = 8
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.enqueue(snd, 6)         # depth = 14
+        self.check_events(rcvUp, 1)
+        self.check_events(rcvDn, 0)
+
+        self.dequeue(rcv, 9)         # depth = 5
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 1)
+
+        self.enqueue(snd, 1)         # depth = 6
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.dequeue(rcv, 1)         # depth = 5
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+        self.dequeue(rcv, 5)         # depth = 0
+        self.check_events(rcvUp, 0)
+        self.check_events(rcvDn, 0)
+
+
+
+