diff options
Diffstat (limited to 'cpp/src/qpid/broker/ThresholdAlerts.cpp')
| -rw-r--r-- | cpp/src/qpid/broker/ThresholdAlerts.cpp | 191 |
1 files changed, 0 insertions, 191 deletions
diff --git a/cpp/src/qpid/broker/ThresholdAlerts.cpp b/cpp/src/qpid/broker/ThresholdAlerts.cpp deleted file mode 100644 index 3c9e210d4d..0000000000 --- a/cpp/src/qpid/broker/ThresholdAlerts.cpp +++ /dev/null @@ -1,191 +0,0 @@ -/* - * - * 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)); -} - -}} |