diff options
Diffstat (limited to 'qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp')
| -rw-r--r-- | qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp | 191 |
1 files changed, 191 insertions, 0 deletions
diff --git a/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp b/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp new file mode 100644 index 0000000000..3c9e210d4d --- /dev/null +++ b/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp @@ -0,0 +1,191 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +#include "qpid/broker/ThresholdAlerts.h" +#include "qpid/broker/Queue.h" +#include "qpid/broker/QueuedMessage.h" +#include "qpid/amqp_0_10/Codecs.h" +#include "qpid/log/Statement.h" +#include "qpid/management/ManagementAgent.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 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, + 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) {} + +void ThresholdAlerts::enqueued(const QueuedMessage& m) +{ + size += m.payload->contentSize(); + ++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.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); + } + } +} + +void ThresholdAlerts::dequeued(const QueuedMessage& m) +{ + size -= m.payload->contentSize(); + --count; + if ((countThreshold && count < countThreshold) || (sizeThreshold && size < sizeThreshold)) { + lastAlert = qpid::sys::EPOCH; + } +} + + + +void ThresholdAlerts::observe(Queue& queue, qpid::management::ManagementAgent& agent, + const uint64_t countThreshold, + const uint64_t sizeThreshold, + const long repeatInterval) +{ + if (countThreshold || sizeThreshold) { + boost::shared_ptr<QueueObserver> observer( + new ThresholdAlerts(queue.getName(), agent, countThreshold, sizeThreshold, repeatInterval) + ); + queue.addObserver(observer); + } +} + +void ThresholdAlerts::observe(Queue& queue, qpid::management::ManagementAgent& agent, + const qpid::framing::FieldTable& settings, uint16_t limitRatio) + +{ + qpid::types::Variant::Map map; + qpid::amqp_0_10::translate(settings, map); + observe(queue, agent, map, limitRatio); +} + +template <class T> +class Option +{ + public: + Option(const std::string& name, T d) : defaultValue(d) { names.push_back(name); } + void addAlias(const std::string& name) { names.push_back(name); } + T get(const qpid::types::Variant::Map& settings) const + { + T value(defaultValue); + for (std::vector<std::string>::const_iterator i = names.begin(); i != names.end(); ++i) { + if (get(settings, *i, value)) break; + } + return value; + } + private: + std::vector<std::string> names; + T defaultValue; + + bool get(const qpid::types::Variant::Map& settings, const std::string& name, T& value) const + { + qpid::types::Variant::Map::const_iterator i = settings.find(name); + if (i != settings.end()) { + try { + value = (T) i->second; + } catch (const qpid::types::InvalidConversion&) { + QPID_LOG(warning, "Bad value for" << name << ": " << i->second); + } + return true; + } else { + return false; + } + } +}; + +void ThresholdAlerts::observe(Queue& queue, qpid::management::ManagementAgent& agent, + 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 specified + //percentage of any limit from the policy. + const QueuePolicy* policy = queue.getPolicy(); + 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 && 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)); +} + +}} |