IcingaDB: Sync dependencies states to Redis

lib/icinga/dependency.hpp

@@ -180,6 +180,15 @@ protected:
 
 private:
 	mutable std::mutex m_Mutex;
+	/**
+	 * This identifier is used by Icinga DB to cache the unique hash of this dependency group.
+	 *
+	 * For redundancy groups, once Icinga DB sets this identifier, it will never change again for the lifetime
+	 * of the object. For non-redundant dependency groups, this identifier is (mis)used to cache the shared edge
+	 * state ID of the group. Specifically, non-redundant dependency groups are irrelevant for Icinga DB, so since
+	 * this field isn't going to be used for anything else, we use it to cache the computed shared edge state ID.
+	 * Likewise, if that gets set, it will never change again for the lifetime of the object as well.
+	 */
 	String m_IcingaDBIdentifier;
 	String m_RedundancyGroupName;
 	MembersMap m_Members;

lib/icingadb/icingadb-objects.cpp

@@ -217,7 +217,9 @@ void IcingaDB::UpdateAllConfigObjects()
 		// This allows us to wait on both types to be dumped before we send a config dump done signal for those keys.
 		m_PrefixConfigObject + "dependency:node",
 		m_PrefixConfigObject + "dependency:edge",
+		m_PrefixConfigObject + "dependency:edge:state",
 		m_PrefixConfigObject + "redundancygroup",
+		m_PrefixConfigObject + "redundancygroup:state",
 	};
 	DeleteKeys(m_Rcon, globalKeys, Prio::Config);
 	DeleteKeys(m_Rcon, {"icinga:nextupdate:host", "icinga:nextupdate:service"}, Prio::Config);
@@ -1344,6 +1346,90 @@ void IcingaDB::UpdateState(const Checkable::Ptr& checkable, StateUpdate mode)
 	}
 }
 
+/**
+ * Send dependencies state information of the given Checkable to Redis.
+ *
+ * If the dependencyGroup parameter is set, only the dependencies state of that group are sent. Otherwise, all
+ * dependency groups of the provided Checkable are processed.
+ *
+ * @param checkable The Checkable you want to send the dependencies state update for
+ * @param onlyDependencyGroup If set, send state updates only for this dependency group and its dependencies.
+ */
+void IcingaDB::UpdateDependenciesState(const Checkable::Ptr& checkable, const DependencyGroup::Ptr& onlyDependencyGroup) const
+{
+	if (!m_Rcon || !m_Rcon->IsConnected()) {
+		return;
+	}
+
+	std::vector<DependencyGroup::Ptr> dependencyGroups{onlyDependencyGroup};
+	if (!onlyDependencyGroup) {
+		dependencyGroups = checkable->GetDependencyGroups();
+		if (dependencyGroups.empty()) {
+			return;
+		}
+	}
+
+	RedisConnection::Queries streamStates;
+	auto addDependencyStateToStream([this, &streamStates](const String& redisKey, const Dictionary::Ptr& stateAttrs) {
+		RedisConnection::Query xAdd{
+			"XADD", "icinga:runtime:state", "MAXLEN", "~", "1000000", "*", "runtime_type", "upsert",
+			"redis_key", redisKey
+		};
+		ObjectLock olock(stateAttrs);
+		for (auto& [key, value] : stateAttrs) {
+			xAdd.emplace_back(key);
+			xAdd.emplace_back(IcingaToStreamValue(value));
+		}
+		streamStates.emplace_back(std::move(xAdd));
+	});
+
+	for (auto& dependencyGroup : dependencyGroups) {
+		bool isRedundancyGroup(dependencyGroup->IsRedundancyGroup());
+		if (isRedundancyGroup && dependencyGroup->GetIcingaDBIdentifier().IsEmpty()) {
+			// Way too soon! The Icinga DB hash will be set during the initial config dump, but this state
+			// update seems to occur way too early. So, we've to skip it for now and wait for the next one.
+			// The m_ConfigDumpInProgress flag is probably still set to true at this point!
+			continue;
+		}
+
+		auto dependencies(dependencyGroup->GetDependenciesForChild(checkable.get()));
+		std::sort(dependencies.begin(), dependencies.end(), [](const Dependency::Ptr& lhs, const Dependency::Ptr& rhs) {
+			return lhs->GetParent() < rhs->GetParent();
+		});
+		for (auto it(dependencies.begin()); it != dependencies.end(); /* no increment */) {
+			const auto& dependency(*it);
+
+			Dictionary::Ptr stateAttrs;
+			// Note: The following loop is intended to cover some possible special cases but may not occur in practice
+			// that often. That is, having two or more dependency objects that point to the same parent Checkable.
+			// So, traverse all those duplicates and merge their relevant state information into a single edge.
+			for (; it != dependencies.end() && (*it)->GetParent() == dependency->GetParent(); ++it) {
+				if (!stateAttrs || stateAttrs->Get("failed") == false) {
+					stateAttrs = SerializeDependencyEdgeState(dependencyGroup, *it);
+				}
+			}
+
+			addDependencyStateToStream(m_PrefixConfigObject + "dependency:edge:state", stateAttrs);
+		}
+
+		if (isRedundancyGroup) {
+			Dictionary::Ptr stateAttrs(SerializeRedundancyGroupState(dependencyGroup));
+
+			Dictionary::Ptr sharedGroupState(stateAttrs->ShallowClone());
+			sharedGroupState->Remove("redundancy_group_id");
+			sharedGroupState->Remove("is_reachable");
+			sharedGroupState->Remove("last_state_change");
+
+			addDependencyStateToStream(m_PrefixConfigObject + "redundancygroup:state", stateAttrs);
+			addDependencyStateToStream(m_PrefixConfigObject + "dependency:edge:state", sharedGroupState);
+		}
+	}
+
+	if (!streamStates.empty()) {
+		m_Rcon->FireAndForgetQueries(std::move(streamStates), Prio::RuntimeStateStream, {0, 1});
+	}
+}
+
 // Used to update a single object, used for runtime updates
 void IcingaDB::SendConfigUpdate(const ConfigObject::Ptr& object, bool runtimeUpdate)
 {
@@ -2933,6 +3019,7 @@ void IcingaDB::ReachabilityChangeHandler(const std::set<Checkable::Ptr>& childre
 	for (const IcingaDB::Ptr& rw : ConfigType::GetObjectsByType<IcingaDB>()) {
 		for (auto& checkable : children) {
 			rw->UpdateState(checkable, StateUpdate::Full);
+			rw->UpdateDependenciesState(checkable);
 		}
 	}
 }

lib/icingadb/icingadb-utility.cpp

@@ -159,6 +159,65 @@ Dictionary::Ptr IcingaDB::SerializeVars(const Dictionary::Ptr& vars)
 	return res;
 }
 
+/**
+ * Serialize a dependency edge state for Icinga DB
+ *
+ * @param dependencyGroup The state of the group the dependency is part of.
+ * @param dep The dependency object to serialize.
+ *
+ * @return A dictionary with the serialized state.
+ */
+Dictionary::Ptr IcingaDB::SerializeDependencyEdgeState(const DependencyGroup::Ptr& dependencyGroup, const Dependency::Ptr& dep)
+{
+	String edgeStateId;
+	// The edge state ID is computed a bit differently depending on whether this is for a redundancy group or not.
+	// For redundancy groups, the state ID is supposed to represent the connection state between the redundancy group
+	// and the parent Checkable of the given dependency. Hence, the outcome will always be different for each parent
+	// Checkable of the redundancy group.
+	if (dependencyGroup->IsRedundancyGroup()) {
+		edgeStateId = HashValue(new Array{
+			dependencyGroup->GetIcingaDBIdentifier(),
+			GetObjectIdentifier(dep->GetParent()),
+		});
+	} else if (dependencyGroup->GetIcingaDBIdentifier().IsEmpty()) {
+		// For non-redundant dependency groups, on the other hand, all dependency objects within that group will
+		// always have the same parent Checkable. Likewise, the state ID will be always the same as well it doesn't
+		// matter which dependency object is used to compute it. Therefore, it's sufficient to compute it only once
+		// and all the other dependency objects can reuse the cached state ID.
+		edgeStateId = HashValue(new Array{dependencyGroup->GetCompositeKey(), GetObjectIdentifier(dep->GetParent())});
+		dependencyGroup->SetIcingaDBIdentifier(edgeStateId);
+	} else {
+		// Use the already computed state ID for the dependency group.
+		edgeStateId = dependencyGroup->GetIcingaDBIdentifier();
+	}
+
+	return new Dictionary{
+		{"id", std::move(edgeStateId)},
+		{"environment_id", m_EnvironmentId},
+		{"failed", !dep->IsAvailable(DependencyState) || !dep->GetParent()->IsReachable()}
+	};
+}
+
+/**
+ * Serialize the provided redundancy group state attributes.
+ *
+ * @param redundancyGroup The redundancy group object to serialize the state for.
+ *
+ * @return A dictionary with the serialized redundancy group state.
+ */
+Dictionary::Ptr IcingaDB::SerializeRedundancyGroupState(const DependencyGroup::Ptr& redundancyGroup)
+{
+	auto state(redundancyGroup->GetState());
+	return new Dictionary{
+		{"id", redundancyGroup->GetIcingaDBIdentifier()},
+		{"environment_id", m_EnvironmentId},
+		{"redundancy_group_id", redundancyGroup->GetIcingaDBIdentifier()},
+		{"failed", !state.Reachable || !state.OK},
+		{"is_reachable", state.Reachable},
+		{"last_state_change", TimestampToMilliseconds(Utility::GetTime())},
+	};
+}
+
 const char* IcingaDB::GetNotificationTypeByEnum(NotificationType type)
 {
 	switch (type) {

lib/icingadb/icingadb.hpp

@@ -114,6 +114,7 @@ private:
 			std::vector<Dictionary::Ptr>* runtimeUpdates);
 	void InsertObjectDependencies(const ConfigObject::Ptr& object, const String typeName, std::map<String, std::vector<String>>& hMSets,
 			std::vector<Dictionary::Ptr>& runtimeUpdates, bool runtimeUpdate);
+	void UpdateDependenciesState(const Checkable::Ptr& checkable, const DependencyGroup::Ptr& onlyDependencyGroup = nullptr) const;
 	void UpdateState(const Checkable::Ptr& checkable, StateUpdate mode);
 	void SendConfigUpdate(const ConfigObject::Ptr& object, bool runtimeUpdate);
 	void CreateConfigUpdate(const ConfigObject::Ptr& object, const String type, std::map<String, std::vector<String>>& hMSets,
@@ -169,6 +170,8 @@ private:
 	static String CalcEventID(const char* eventType, const ConfigObject::Ptr& object, double eventTime = 0, NotificationType nt = NotificationType(0));
 	static const char* GetNotificationTypeByEnum(NotificationType type);
 	static Dictionary::Ptr SerializeVars(const Dictionary::Ptr& vars);
+	static Dictionary::Ptr SerializeDependencyEdgeState(const DependencyGroup::Ptr& dependencyGroup, const Dependency::Ptr& dep);
+	static Dictionary::Ptr SerializeRedundancyGroupState(const DependencyGroup::Ptr& redundancyGroup);
 
 	static String HashValue(const Value& value);
 	static String HashValue(const Value& value, const std::set<String>& propertiesBlacklist, bool propertiesWhitelist = false);