cluster: Implement authority checks.

components/checker/checkercomponent.cpp

@@ -106,6 +106,11 @@ void CheckerComponent::CheckThreadProc(void)
 		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");

components/cluster/clustercomponent.cpp

@@ -85,6 +85,8 @@ void ClusterComponent::Start(void)
 	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));
 }
 
@@ -1119,6 +1121,48 @@ void ClusterComponent::MessageHandler(const Endpoint::Ptr& sender, const Diction
 	}
 }
 
+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);

components/cluster/clustercomponent.h

@@ -111,6 +111,7 @@ private:
 	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);
 };
 
 }

components/cluster/endpoint.cpp

@@ -168,6 +168,26 @@ void Endpoint::SetRemoteLogPosition(double ts)
 	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);
@@ -183,6 +203,7 @@ void Endpoint::InternalSerialize(const Dictionary::Ptr& bag, int 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);
 	}
 }
 
@@ -201,5 +222,6 @@ void Endpoint::InternalDeserialize(const Dictionary::Ptr& bag, int attributeType
 		m_Seen = bag->Get("seen");
 		m_LocalLogPosition = bag->Get("local_log_position");
 		m_RemoteLogPosition = bag->Get("remote_log_position");
+		m_Features = bag->Get("features");
 	}
 }

components/cluster/endpoint.h

@@ -65,6 +65,11 @@ public:
 	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);
@@ -80,6 +85,7 @@ private:
 	double m_Seen;
 	double m_LocalLogPosition;
 	double m_RemoteLogPosition;
+	Dictionary::Ptr m_Features;
 
 	void MessageThreadProc(const Stream::Ptr& stream);
 };

lib/base/dynamicobject.cpp

@@ -41,6 +41,7 @@ using namespace icinga;
 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)
@@ -82,6 +83,7 @@ void DynamicObject::InternalSerialize(const Dictionary::Ptr& bag, int attributeT
 		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);
@@ -98,6 +100,7 @@ void DynamicObject::InternalDeserialize(const Dictionary::Ptr& bag, int attribut
 		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");
@@ -118,6 +121,18 @@ bool DynamicObject::IsActive(void) const
 	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;

lib/base/dynamicobject.h

@@ -23,6 +23,7 @@
 #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>
@@ -62,6 +63,7 @@ public:
 	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);
 
@@ -70,6 +72,9 @@ public:
 
 	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);
@@ -111,6 +116,7 @@ private:
 	Dictionary::Ptr m_Extensions;
 	Dictionary::Ptr m_Methods;
 	Dictionary::Ptr m_Custom;
+	Array::Ptr m_Authorities;
 
 	bool m_Active;
 

lib/base/utility.cpp

@@ -551,3 +551,15 @@ String Utility::GetThreadName(void)
 
 	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;
+}

lib/base/utility.h

@@ -95,6 +95,8 @@ public:
 	static void SetThreadName(const String& name, bool os = true);
 	static String GetThreadName(void);
 
+	static unsigned long SDBM(const String& str);
+
 private:
 	Utility(void);
 

lib/config/base-type.conf

@@ -28,6 +28,10 @@ type DynamicObject {
 
 	%attribute dictionary "custom" {
 		%attribute string "*"
+	},
+
+	%attribute array "authorities" {
+		%attribute string "*"
 	}
 }