icinga2/lib/icingadb/icingadb.hpp

// SPDX-FileCopyrightText: 2012 Icinga GmbH <https://icinga.com>
// SPDX-License-Identifier: GPL-2.0-or-later

#ifndef ICINGADB_H
#define ICINGADB_H

#include "icingadb/icingadb-ti.hpp"
#include "icingadb/redisconnection.hpp"
#include "base/atomic.hpp"
#include "base/bulker.hpp"
#include "base/timer.hpp"
#include "base/workqueue.hpp"
#include "icinga/customvarobject.hpp"
#include "icinga/checkable.hpp"
#include "icinga/command.hpp"
#include "icinga/service.hpp"
#include "icinga/downtime.hpp"
#include "remote/messageorigin.hpp"
#include <boost/multi_index_container.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/sequenced_index.hpp>
#include <atomic>
#include <chrono>
#include <future>
#include <memory>
#include <mutex>
#include <set>
#include <unordered_map>
#include <unordered_set>
#include <utility>

namespace icinga
{

#define CONFIG_REDIS_KEY_PREFIX "icinga:"
#define CHECKSUM_REDIS_KEY_PREFIX CONFIG_REDIS_KEY_PREFIX "checksum:"

namespace icingadb::task_queue
{

/**
 * Dirty bits for config/state changes.
 *
 * These are used to mark objects as "dirty" in order to trigger appropriate updates in Redis.
 * Each bit represents a different type of change that requires a specific action to be taken.
 *
 * @ingroup icingadb
 */
enum DirtyBits : uint32_t
{
	ConfigUpdate  = 1<<0, // Trigger a Redis config update for the object.
	ConfigDelete  = 1<<1, // Send a deletion command for the object to Redis.
	VolatileState = 1<<2, // Send a volatile state update to Redis (affects only checkables).
	RuntimeState  = 1<<3, // Send a runtime state update to Redis (affects only checkables).
	NextUpdate    = 1<<4, // Update the `icinga:nextupdate:{host,service}` Redis keys (affects only checkables).

	FullState = VolatileState | RuntimeState, // A combination of all (non-dependency) state-related dirty bits.

	// All valid dirty bits combined used for masking input values.
	DirtyBitsAll = ConfigUpdate | ConfigDelete | FullState | NextUpdate
};


/**
 * A pending configuration object item.
 *
 * This struct represents a pending item in the queue that is associated with a configuration object.
 * It contains a pointer to the configuration object and the dirty bits indicating the type of updates
 * required for that object in Redis.
 *
 * @ingroup icingadb
 */
struct PendingConfigItem
{
	ConfigObject::Ptr Object;
	uint32_t DirtyBits;

	PendingConfigItem(const ConfigObject::Ptr& obj, uint32_t bits);

	[[nodiscard]] ConfigObject* GetQueueLookupKey() const
	{
		return Object.get();
	}
};

/**
 * A pending dependency group state item.
 *
 * This struct represents a pending item in the queue that is associated with a dependency group.
 * It contains a pointer to the dependency group for which state updates are required.
 *
 * @ingroup icingadb
 */
struct PendingDependencyGroupStateItem
{
	DependencyGroup::Ptr DepGroup;

	explicit PendingDependencyGroupStateItem(const DependencyGroup::Ptr& depGroup);

	[[nodiscard]] DependencyGroup* GetQueueLookupKey() const
	{
		return DepGroup.get();
	}
};

/**
 * A pending dependency edge item.
 *
 * This struct represents a pending dependency child registration into a dependency group.
 * It contains a pointer to the dependency group and the checkable child being registered.
 *
 * @ingroup icingadb
 */
struct PendingDependencyEdgeItem
{
	DependencyGroup::Ptr DepGroup;
	Checkable::Ptr Child;

	PendingDependencyEdgeItem(const DependencyGroup::Ptr& depGroup, const Checkable::Ptr& child);

	[[nodiscard]] std::pair<DependencyGroup*, Checkable*> GetQueueLookupKey() const
	{
		return {DepGroup.get(), Child.get()};
	}
};

/**
 * A pending relations deletion item.
 *
 * This struct represents a pending item in the queue that is associated with the deletion of relations
 * in Redis. It contains a map of Redis keys from which the relation identified by the @c ID field should
 * be deleted. The @c ID field represents the unique identifier of the relation to be deleted, and the
 * @c Relations map specifies the Redis keys and whether to delete the corresponding checksum keys.
 *
 * @ingroup icingadb
 */
struct RelationsDeletionItem
{
	std::string ID;
	// Set of Redis keys from which to delete a relation, along with their checksums (if any).
	using RelationsKeySet = std::set<std::pair<RedisConnection::QueryArg, RedisConnection::QueryArg>>;
	RelationsKeySet Relations;

	RelationsDeletionItem(const String& id, const RelationsKeySet& relations);

	[[nodiscard]] std::string_view GetQueueLookupKey() const
	{
		return ID;
	}
};

/**
 * A pending queue item.
 *
 * This struct represents a generic pending item in the queue. The @c EnqueueTime field records the
 * time when the item was added to the queue, which can be useful for tracking how long an item waits before
 * being processed. This base struct wraps one of the available more specific pending item types that operate
 * on different kinds of objects, such as configuration objects or dependency groups.
 *
 * @ingroup icingadb
 */
struct PendingQueueItem
{
	using ItemVariant = std::variant<
		PendingConfigItem,
		PendingDependencyGroupStateItem,
		PendingDependencyEdgeItem,
		RelationsDeletionItem
	>;

	ItemVariant Item;
	std::chrono::steady_clock::time_point EnqueueTime{std::chrono::steady_clock::now()};

	template<typename T,
		// don't hide the default copy and move constructors
		typename = std::enable_if_t<!std::is_same_v<std::decay_t<T>, PendingQueueItem>>
	>
	explicit PendingQueueItem(T&& item) : Item(std::forward<T>(item)) {}
};

template<typename T>
struct KeyExtractor /* not implemented, only the specialization for std::variant is used */;

/**
 * Helper to use @c PendingQueueItem in @c boost::multi_index_container.
 *
 * It implements a key extractor that @c multi_index_container will invoke for each element to create an index.
 * Every individual type in @c PendingQueueItem::ItemVariant implements a method @c GetQueueLookupKey() that will
 * be invoked by this key extractor. The return value from the individual type is then returned as part of a second
 * variant type (its element types are automatically deduced from the individual @c GetQueueLookupKey() methods).
 * @c multi_index_container will then just use the standard comparison operators on it for building the index.
 *
 * @ingroup icingadb
 */
template<typename ...Ts>
struct KeyExtractor<std::variant<Ts...>> {
	using result_type = std::variant<decltype(std::declval<Ts>().GetQueueLookupKey())...>;

	result_type operator()(const PendingQueueItem& queueItem) const
	{
		return std::visit([](const auto& innerItem) -> result_type {
			return innerItem.GetQueueLookupKey();
		}, queueItem.Item);
	};
};

// A multi-index container for managing pending items with unique IDs and maintaining insertion order.
// The first index is an ordered unique index based on the pending item key, allowing for efficient
// lookups and ensuring uniqueness of items. The second index is a sequenced index that maintains the
// order of insertion, enabling FIFO processing of pending items.
using PendingItemsSet = boost::multi_index_container<
	PendingQueueItem,
	boost::multi_index::indexed_by<
		boost::multi_index::ordered_unique<KeyExtractor<PendingQueueItem::ItemVariant>>,
		boost::multi_index::sequenced<>
	>
>;

} // namespace icingadb::task_queue

/**
 * @ingroup icingadb
 */
class IcingaDB : public ObjectImpl<IcingaDB>
{
public:
	DECLARE_OBJECT(IcingaDB);
	DECLARE_OBJECTNAME(IcingaDB);

	IcingaDB();

	static void ConfigStaticInitialize();

	void Validate(int types, const ValidationUtils& utils) override;
	virtual void Start(bool runtimeCreated) override;
	virtual void Stop(bool runtimeRemoved) override;

	String GetEnvironmentId() const override;

	inline RedisConnection::Ptr GetConnection() const
	{
		return m_RconLocked.load();
	}

	void LoadPendingItemsStats(const Array::Ptr& perfdata) const;

	template<class T>
	static void AddKvsToMap(const Array::Ptr& kvs, T& map)
	{
		Value* key = nullptr;
		ObjectLock oLock (kvs);

		for (auto& kv : kvs) {
			if (key) {
				map.emplace(std::move(*key), std::move(kv));
				key = nullptr;
			} else {
				key = &kv;
			}
		}
	}

	// Represents any pair of Redis keys that belong together, e.g, @c {icinga:host:state, icinga:checksum:host:state}.
	struct QueryArgPair
	{
		std::string_view ObjectKey;
		std::string_view ChecksumKey;
	};

	struct CmdArgEnvRedisKeys
	{
		std::string_view ArgObjectKey; // Represents `icinga:{command_type}:argument`.
		std::string_view ArgChecksumKey; // Represents `icinga:checksum:{command_type}:argument`.
		std::string_view EnvObjectKey; // Represents `icinga:{command_type}:envvar`.
		std::string_view EnvChecksumKey; // Represents `icinga:checksum:{command_type}:envvar`.
	};

protected:
	void ValidateTlsProtocolmin(const Lazy<String>& lvalue, const ValidationUtils& utils) override;
	void ValidateConnectTimeout(const Lazy<double>& lvalue, const ValidationUtils& utils) override;
	RedisConnInfo::ConstPtr GetRedisConnInfo() const;

private:
	class DumpedGlobals
	{
	public:
		void Reset();
		bool IsNew(const String& id);

	private:
		std::set<String> m_Ids;
		std::mutex m_Mutex;
	};

	void OnConnectedHandler();

	void PublishStatsTimerHandler();
	void PublishStats();

	/* config & status dump */
	void UpdateAllConfigObjects();
	std::vector<std::vector<intrusive_ptr<ConfigObject>>> ChunkObjects(std::vector<intrusive_ptr<ConfigObject>> objects, size_t chunkSize);
	void InsertCheckableDependencies(const Checkable::Ptr& checkable, std::map<RedisConnection::QueryArg, RedisConnection::Query>& hMSets,
		std::vector<Dictionary::Ptr>* runtimeUpdates, const DependencyGroup::Ptr& onlyDependencyGroup = nullptr);
	void InsertObjectDependencies(const ConfigObject::Ptr& object,
		std::map<RedisConnection::QueryArg, RedisConnection::Query>& hMSets, std::vector<Dictionary::Ptr>& runtimeUpdates, bool runtimeUpdate);
	void UpdateDependenciesState(const Checkable::Ptr& checkable, const DependencyGroup::Ptr& dependencyGroup) const;
	void UpdateState(const Checkable::Ptr& checkable, uint32_t mode);
	void CreateConfigUpdate(const ConfigObject::Ptr& object, const QueryArgPair& redisKeyPair,
		std::map<RedisConnection::QueryArg, RedisConnection::Query>& hMSets, std::vector<Dictionary::Ptr>& runtimeUpdates, bool runtimeUpdate);
	void SendConfigDelete(const ConfigObject::Ptr& object);
	void SendStateChange(const ConfigObject::Ptr& object, const CheckResult::Ptr& cr, StateType type);
	void AddObjectDataToRuntimeUpdates(std::vector<Dictionary::Ptr>& runtimeUpdates,
		String objectKey, String redisKey, const Dictionary::Ptr& data);
	void DeleteRelationship(const String& id, RedisConnection::QueryArg redisObjKey, RedisConnection::QueryArg redisChecksumKey = "");
	void DeleteState(const String& id, RedisConnection::QueryArg redisObjKey, RedisConnection::QueryArg redisChecksumKey = "") const;

	void SendSentNotification(
		const Notification::Ptr& notification, const Checkable::Ptr& checkable, const std::set<User::Ptr>& users,
		NotificationType type, const CheckResult::Ptr& cr, const String& author, const String& text, double sendTime
	);

	void SendStartedDowntime(const Downtime::Ptr& downtime);
	void SendRemovedDowntime(const Downtime::Ptr& downtime);
	void SendAddedComment(const Comment::Ptr& comment);
	void SendRemovedComment(const Comment::Ptr& comment);
	void SendFlappingChange(const Checkable::Ptr& checkable, double changeTime, double flappingLastChange);
	void SendAcknowledgementSet(const Checkable::Ptr& checkable, const String& author, const String& comment, AcknowledgementType type, bool persistent, double changeTime, double expiry);
	void SendAcknowledgementCleared(const Checkable::Ptr& checkable, const String& removedBy, double changeTime, double ackLastChange);
	void SendNotificationUsersChanged(const Notification::Ptr& notification, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	void SendNotificationUserGroupsChanged(const Notification::Ptr& notification, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	void SendTimePeriodRangesChanged(const TimePeriod::Ptr& timeperiod, const Dictionary::Ptr& oldValues, const Dictionary::Ptr& newValues);
	void SendTimePeriodIncludesChanged(const TimePeriod::Ptr& timeperiod, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	void SendTimePeriodExcludesChanged(const TimePeriod::Ptr& timeperiod, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	template<class T>
	void SendGroupsChanged(const ConfigObject::Ptr& command, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	void SendCommandEnvChanged(const ConfigObject::Ptr& command, const CmdArgEnvRedisKeys& cmdRedisKeys, const Dictionary::Ptr& oldValues, const Dictionary::Ptr& newValues);
	void SendCommandArgumentsChanged(const ConfigObject::Ptr& command, const CmdArgEnvRedisKeys& cmdRedisKeys, const Dictionary::Ptr& oldValues, const Dictionary::Ptr& newValues);
	void SendCustomVarsChanged(const ConfigObject::Ptr& object, const Dictionary::Ptr& oldValues, const Dictionary::Ptr& newValues);

	void ForwardHistoryEntries();

	Dictionary::Ptr SerializeState(const Checkable::Ptr& checkable);

	/* Stats */
	static Dictionary::Ptr GetStats();

	/* utilities */
	static void DeleteKeys(const RedisConnection::Ptr& conn, const std::vector<RedisConnection::QueryArg>& keys);
	static std::vector<RedisConnection::QueryArg> GetTypeOverwriteKeys(const Type::Ptr& type, bool skipChecksums = false);
	static void AddDataToHmSets(std::map<RedisConnection::QueryArg, RedisConnection::Query>& hMSets, const RedisConnection::QueryArg& redisKey, const String& id, const Dictionary::Ptr& data);
	static bool IsStateKey(const RedisConnection::QueryArg& key);
	static String FormatCheckSumBinary(const String& str);
	static String FormatCommandLine(const Value& commandLine);
	static QueryArgPair GetCheckableStateKeys(const Type::Ptr& type);
	static CmdArgEnvRedisKeys GetCmdEnvArgKeys(const Type::Ptr& cmdType);
	static long long TimestampToMilliseconds(double timestamp);
	static String IcingaToStreamValue(const Value& value);
	static std::vector<Value> GetArrayDeletedValues(const Array::Ptr& arrayOld, const Array::Ptr& arrayNew);
	static std::vector<String> GetDictionaryDeletedKeys(const Dictionary::Ptr& dictOld, const Dictionary::Ptr& dictNew);

	static String GetObjectIdentifier(const ConfigObject::Ptr& object);
	static String CalcEventID(const char* eventType, const ConfigObject::Ptr& object, double eventTime = 0, NotificationType nt = NotificationType(0));
	static int StateFilterToRedisValue(int filter);
	static int TypeFilterToRedisValue(int filter);
	static const char* GetNotificationTypeByEnum(NotificationType type);
	static String CommentTypeToString(CommentType 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 Checkable::Ptr& child, const DependencyGroup::Ptr& redundancyGroup);
	static String GetDependencyEdgeStateId(const DependencyGroup::Ptr& dependencyGroup, const Dependency::Ptr& dep);

	static String HashValue(const Value& value);
	static String HashValue(const Value& value, const std::set<String>& propertiesBlacklist, bool propertiesWhitelist = false);

	static String GetLowerCaseTypeNameDB(const ConfigObject::Ptr& obj);
	static bool PrepareObject(const ConfigObject::Ptr& object, Dictionary::Ptr& attributes);

	static void ReachabilityChangeHandler(const std::set<Checkable::Ptr>& children);
	static void StateChangeHandler(const ConfigObject::Ptr& object, const CheckResult::Ptr& cr, StateType type);
	static void VersionChangedHandler(const ConfigObject::Ptr& object);
	static void DowntimeStartedHandler(const Downtime::Ptr& downtime);
	static void DowntimeRemovedHandler(const Downtime::Ptr& downtime);

	static void NotificationSentToAllUsersHandler(
		const Notification::Ptr& notification, const Checkable::Ptr& checkable, const std::set<User::Ptr>& users,
		NotificationType type, const CheckResult::Ptr& cr, const String& author, const String& text
	);

	static void CommentAddedHandler(const Comment::Ptr& comment);
	static void CommentRemovedHandler(const Comment::Ptr& comment);
	static void FlappingChangeHandler(const Checkable::Ptr& checkable, double changeTime);
	static void NewCheckResultHandler(const Checkable::Ptr& checkable);
	static void NextCheckChangedHandler(const Checkable::Ptr& checkable);
	static void DependencyGroupChildRegisteredHandler(const Checkable::Ptr& child, const DependencyGroup::Ptr& dependencyGroup);
	static void DependencyGroupChildRemovedHandler(const DependencyGroup::Ptr& dependencyGroup, const std::vector<Dependency::Ptr>& dependencies, bool removeGroup);
	static void HostProblemChangedHandler(const Service::Ptr& service);
	static void AcknowledgementSetHandler(const Checkable::Ptr& checkable, const String& author, const String& comment, AcknowledgementType type, bool persistent, double changeTime, double expiry);
	static void AcknowledgementClearedHandler(const Checkable::Ptr& checkable, const String& removedBy, double changeTime);
	static void NotificationUsersChangedHandler(const Notification::Ptr& notification, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	static void NotificationUserGroupsChangedHandler(const Notification::Ptr& notification, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	static void TimePeriodRangesChangedHandler(const TimePeriod::Ptr& timeperiod, const Dictionary::Ptr& oldValues, const Dictionary::Ptr& newValues);
	static void TimePeriodIncludesChangedHandler(const TimePeriod::Ptr& timeperiod, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	static void TimePeriodExcludesChangedHandler(const TimePeriod::Ptr& timeperiod, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	static void UserGroupsChangedHandler(const User::Ptr& user, const Array::Ptr&, const Array::Ptr& newValues);
	static void HostGroupsChangedHandler(const Host::Ptr& host, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	static void ServiceGroupsChangedHandler(const Service::Ptr& service, const Array::Ptr& oldValues, const Array::Ptr& newValues);
	static void CommandEnvChangedHandler(const Command::Ptr& command, const Dictionary::Ptr& oldValues, const Dictionary::Ptr& newValues);
	static void CommandArgumentsChangedHandler(const Command::Ptr& command, const Dictionary::Ptr& oldValues, const Dictionary::Ptr& newValues);
	static void CustomVarsChangedHandler(const ConfigObject::Ptr& object, const Dictionary::Ptr& oldValues, const Dictionary::Ptr& newValues);

	static void ExecuteRedisTransaction(const RedisConnection::Ptr& rcon,
		std::map<RedisConnection::QueryArg, RedisConnection::Query>& hMSets, const std::vector<Dictionary::Ptr>& runtimeUpdates);

	void AssertOnWorkQueue();

	void ExceptionHandler(std::exception_ptr exp);

	using SyncableTypeInfo = std::pair<const Type::Ptr, QueryArgPair>;
	static std::vector<SyncableTypeInfo> GetSyncableTypes();
	static const QueryArgPair& GetSyncableTypeRedisKeys(const Type::Ptr& type);

	static void InitEnvironmentId();
	static void PersistEnvironmentId();

	Timer::Ptr m_StatsTimer;
	WorkQueue m_WorkQueue{0, 1, LogNotice};

	std::future<void> m_HistoryThread;
	Bulker<RedisConnection::Query> m_HistoryBulker {4096, std::chrono::milliseconds(250)};

	bool m_ConfigDumpInProgress;
	std::atomic_bool m_ConfigDumpDone{false};

	/**
	 * The primary Redis connection used to send history and heartbeat queries.
	 *
	 * This connection is used exclusively for sending history and heartbeat queries to Redis. It ensures that
	 * history and heartbeat operations do not interfere with other Redis operations. Also, it is the leader for
	 * all other Redis connections including @c m_RconWorker, and is the only source of truth for all IcingaDB Redis
	 * related connection statistics.
	 *
	 * Note: This will still be shared with the icingadb check command, as that command also sends
	 * only XREAD queries which are similar in nature to history/heartbeat queries.
	 */
	RedisConnection::Ptr m_Rcon;
	// m_RconLocked contains a copy of the value in m_Rcon where all accesses are guarded by a mutex to
	// allow safe concurrent access like from the icingadb check command. It's a copy to still allow fast access
	// without additional synchronization to m_Rcon within the IcingaDB feature itself.
	Locked<RedisConnection::Ptr> m_RconLocked;
	/**
	 * A Redis connection for config and state updates.
	 *
	 * This connection is used for all non-history and non-heartbeat related queries to Redis.
	 * It is a child of @c m_Rcon, meaning it forwards all its connection stats to @c m_Rcon as well.
	 */
	RedisConnection::Ptr m_RconWorker;
	std::unordered_map<ConfigType*, RedisConnection::Ptr> m_Rcons;
	std::atomic_size_t m_PendingRcons;

	struct {
		DumpedGlobals CustomVar, ActionUrl, NotesUrl, IconImage, DependencyGroup;
	} m_DumpedGlobals;

	// m_EnvironmentId is shared across all IcingaDB objects (typically there is at most one, but it is perfectly fine
	// to have multiple ones). It is initialized once (synchronized using m_EnvironmentIdInitMutex). After successful
	// initialization, the value is read-only and can be accessed without further synchronization.
	static String m_EnvironmentId;
	static std::mutex m_EnvironmentIdInitMutex;

	std::thread m_PendingItemsThread; // The background worker thread (consumer of m_PendingItems).
	icingadb::task_queue::PendingItemsSet m_PendingItems; // Container for pending items with dirty bits (access protected by m_PendingItemsMutex).
	mutable std::mutex m_PendingItemsMutex; // Mutex to protect access to m_PendingItems.
	std::condition_variable m_PendingItemsCV; // Condition variable to forcefully wake up the worker thread.

	void PendingItemsThreadProc();

	void ProcessQueueItem(const icingadb::task_queue::PendingConfigItem& item);
	void ProcessQueueItem(const icingadb::task_queue::PendingDependencyGroupStateItem& item) const;
	void ProcessQueueItem(const icingadb::task_queue::PendingDependencyEdgeItem& item);
	void ProcessQueueItem(const icingadb::task_queue::RelationsDeletionItem& item);

	void EnqueueConfigObject(const ConfigObject::Ptr& object, uint32_t bits);
	void EnqueueDependencyGroupStateUpdate(const DependencyGroup::Ptr& depGroup);
	void EnqueueDependencyChildRegistered(const DependencyGroup::Ptr& depGroup, const Checkable::Ptr& child);
	void EnqueueDependencyChildRemoved(const DependencyGroup::Ptr& depGroup, const std::vector<Dependency::Ptr>& dependencies, bool removeGroup);
	void EnqueueRelationsDeletion(const String& id, icingadb::task_queue::RelationsDeletionItem::RelationsKeySet relations);
};
}

#endif /* ICINGADB_H */