bool forced = service->GetForceNextCheck();
bool check = true;
+ if (!service->HasAuthority("checker")) {
+ Log(LogDebug, "checker", "Skipping check for service '" + service->GetName() + "': not authoritative");
+ check = false;
+ }
+
if (!forced) {
if (!service->GetEnableActiveChecks()) {
Log(LogDebug, "checker", "Skipping check for service '" + service->GetName() + "': active checks are disabled");
Service::OnAcknowledgementSet.connect(boost::bind(&ClusterComponent::AcknowledgementSetHandler, this, _1, _2, _3, _4, _5, _6));
Service::OnAcknowledgementCleared.connect(boost::bind(&ClusterComponent::AcknowledgementClearedHandler, this, _1, _2));
+ DynamicObject::OnCheckAuthority.connect(boost::bind(&ClusterComponent::CheckAuthorityHandler, this, _1, _2, _3));
+
Endpoint::OnMessageReceived.connect(boost::bind(&ClusterComponent::MessageHandler, this, _1, _2));
}
}
}
+void ClusterComponent::CheckAuthorityHandler(const DynamicObject::Ptr& object, const String& type, bool& result)
+{
+ Array::Ptr authorities = object->GetAuthorities();
+ std::vector<String> endpoints;
+
+ if ((type == "checker" && DynamicType::GetByName("CheckerComponent")) ||
+ (type == "notification" && DynamicType::GetByName("NotificationComponent")))
+ endpoints.push_back(GetIdentity());
+
+ BOOST_FOREACH(const Endpoint::Ptr& endpoint, DynamicType::GetObjects<Endpoint>()) {
+ bool match = false;
+
+ if (!endpoint->IsConnected())
+ continue;
+
+ if (authorities) {
+ BOOST_FOREACH(const String& authority, authorities) {
+ if (Utility::Match(authority, endpoint->GetName())) {
+ match = true;
+
+ break;
+ }
+ }
+ } else {
+ match = true;
+ }
+
+ if (match)
+ endpoints.push_back(endpoint->GetName());
+ }
+
+ std::sort(endpoints.begin(), endpoints.end());
+
+ String key = object->GetType()->GetName() + "\t" + object->GetName();
+ unsigned long hash = Utility::SDBM(key);
+ unsigned long index = hash % endpoints.size();
+
+ Log(LogDebug, "cluster", "Authority for object '" + object->GetName() + "' of type '" + object->GetType()->GetName() + "' is '" + endpoints[index] + "'.");
+
+ result = (endpoints[index] == GetIdentity());
+}
+
void ClusterComponent::InternalSerialize(const Dictionary::Ptr& bag, int attributeTypes) const
{
DynamicObject::InternalSerialize(bag, attributeTypes);
void AcknowledgementSetHandler(const Service::Ptr& service, const String& author, const String& comment, AcknowledgementType type, double expiry, const String& authority);
void AcknowledgementClearedHandler(const Service::Ptr& service, const String& authority);
void MessageHandler(const Endpoint::Ptr& sender, const Dictionary::Ptr& message);
+ void CheckAuthorityHandler(const DynamicObject::Ptr& object, const String& type, bool& result);
};
}
m_RemoteLogPosition = ts;
}
+Dictionary::Ptr Endpoint::GetFeatures(void) const
+{
+ return m_Features;
+}
+
+void Endpoint::SetFeatures(const Dictionary::Ptr& features)
+{
+ m_Features = features;
+}
+
+bool Endpoint::HasFeature(const String& type) const
+{
+ Dictionary::Ptr features = GetFeatures();
+
+ if (!features)
+ return false;
+
+ return features->Get(type);
+}
+
void Endpoint::InternalSerialize(const Dictionary::Ptr& bag, int attributeTypes) const
{
DynamicObject::InternalSerialize(bag, attributeTypes);
bag->Set("seen", m_Seen);
bag->Set("local_log_position", m_LocalLogPosition);
bag->Set("remote_log_position", m_RemoteLogPosition);
+ bag->Set("features", m_Features);
}
}
m_Seen = bag->Get("seen");
m_LocalLogPosition = bag->Get("local_log_position");
m_RemoteLogPosition = bag->Get("remote_log_position");
+ m_Features = bag->Get("features");
}
}
double GetRemoteLogPosition(void) const;
void SetRemoteLogPosition(double ts);
+ Dictionary::Ptr GetFeatures(void) const;
+ void SetFeatures(const Dictionary::Ptr& features);
+
+ bool HasFeature(const String& type) const;
+
protected:
virtual void InternalSerialize(const Dictionary::Ptr& bag, int attributeTypes) const;
virtual void InternalDeserialize(const Dictionary::Ptr& bag, int attributeTypes);
double m_Seen;
double m_LocalLogPosition;
double m_RemoteLogPosition;
+ Dictionary::Ptr m_Features;
void MessageThreadProc(const Stream::Ptr& stream);
};
boost::signals2::signal<void (const DynamicObject::Ptr&)> DynamicObject::OnStarted;
boost::signals2::signal<void (const DynamicObject::Ptr&)> DynamicObject::OnStopped;
boost::signals2::signal<void (const DynamicObject::Ptr&)> DynamicObject::OnStateChanged;
+boost::signals2::signal<void (const DynamicObject::Ptr&, const String&, bool&)> DynamicObject::OnCheckAuthority;
DynamicObject::DynamicObject(void)
: m_Active(false)
bag->Set("__type", m_Type);
bag->Set("methods", m_Methods);
bag->Set("custom", m_Custom);
+ bag->Set("authorities", m_Authorities);
}
bag->Set("extensions", m_Extensions);
m_Type = bag->Get("__type");
m_Methods = bag->Get("methods");
m_Custom = bag->Get("custom");
+ m_Authorities = bag->Get("authorities");
}
m_Extensions = bag->Get("extensions");
return m_Active;
}
+Array::Ptr DynamicObject::GetAuthorities(void) const
+{
+ return m_Authorities;
+}
+
+bool DynamicObject::HasAuthority(const String& type)
+{
+ bool result = true;
+ OnCheckAuthority(GetSelf(), type, result);
+ return result;
+}
+
void DynamicObject::SetExtension(const String& key, const Object::Ptr& object)
{
Dictionary::Ptr extensions = m_Extensions;
#include "base/i2-base.h"
#include "base/object.h"
#include "base/dictionary.h"
+#include "base/array.h"
#include <boost/signals2.hpp>
#include <map>
#include <set>
static boost::signals2::signal<void (const DynamicObject::Ptr&)> OnStarted;
static boost::signals2::signal<void (const DynamicObject::Ptr&)> OnStopped;
static boost::signals2::signal<void (const DynamicObject::Ptr&)> OnStateChanged;
+ static boost::signals2::signal<void (const DynamicObject::Ptr&, const String&, bool&)> OnCheckAuthority;
Value InvokeMethod(const String& method, const std::vector<Value>& arguments);
bool IsActive(void) const;
+ Array::Ptr GetAuthorities(void) const;
+ bool HasAuthority(const String& type);
+
void SetExtension(const String& key, const Object::Ptr& object);
Object::Ptr GetExtension(const String& key);
void ClearExtension(const String& key);
Dictionary::Ptr m_Extensions;
Dictionary::Ptr m_Methods;
Dictionary::Ptr m_Custom;
+ Array::Ptr m_Authorities;
bool m_Active;
return *name;
}
+
+unsigned long Utility::SDBM(const String& str)
+{
+ unsigned long hash = 0;
+ int c;
+
+ BOOST_FOREACH(char c, str) {
+ hash = c + (hash << 6) + (hash << 16) - hash;
+ }
+
+ return hash;
+}
static void SetThreadName(const String& name, bool os = true);
static String GetThreadName(void);
+ static unsigned long SDBM(const String& str);
+
private:
Utility(void);
%attribute dictionary "custom" {
%attribute string "*"
+ },
+
+ %attribute array "authorities" {
+ %attribute string "*"
}
}