Merge pull request #10013 from Icinga/broken-runtime-config-sync

Fix broken runtime config sync

lib/remote/apilistener-configsync.cpp

@@ -8,6 +8,7 @@
 #include "base/json.hpp"
 #include "base/convert.hpp"
 #include "config/vmops.hpp"
+#include "remote/configobjectslock.hpp"
 #include <fstream>
 
 using namespace icinga;
@@ -104,6 +105,11 @@ Value ApiListener::ConfigUpdateObjectAPIHandler(const MessageOrigin::Ptr& origin
 		return Empty;
 	}
 
+	// Wait for the object name to become available for processing and block it immediately.
+	// Doing so guarantees that only one (create/update/delete) cluster event or API request of a
+	// given object is being processed at any given time.
+	ObjectNameLock objectNameLock(ptype, objName);
+
 	ConfigObject::Ptr object = ctype->GetObject(objName);
 
 	String config = params->Get("config");
@@ -258,6 +264,11 @@ Value ApiListener::ConfigDeleteObjectAPIHandler(const MessageOrigin::Ptr& origin
 		return Empty;
 	}
 
+	// Wait for the object name to become available for processing and block it immediately.
+	// Doing so guarantees that only one (create/update/delete) cluster event or API request of a
+	// given object is being processed at any given time.
+	ObjectNameLock objectNameLock(ptype, objName);
+
 	ConfigObject::Ptr object = ctype->GetObject(objName);
 
 	if (!object) {

lib/remote/configobjectslock.cpp

@@ -1,13 +1,16 @@
 /* Icinga 2 | (c) 2022 Icinga GmbH | GPLv2+ */
 
-#ifndef _WIN32
-
-#include "base/shared-memory.hpp"
 #include "remote/configobjectslock.hpp"
+
+#ifndef _WIN32
+#include "base/shared-memory.hpp"
 #include <boost/interprocess/sync/lock_options.hpp>
+#endif /* _WIN32 */
 
 using namespace icinga;
 
+#ifndef _WIN32
+
 // On *nix one process may write config objects while another is loading the config, so this uses IPC.
 static SharedMemory<boost::interprocess::interprocess_sharable_mutex> l_ConfigObjectsMutex;
 
@@ -22,3 +25,37 @@ ConfigObjectsSharedLock::ConfigObjectsSharedLock(std::try_to_lock_t)
 }
 
 #endif /* _WIN32 */
+
+std::mutex ObjectNameLock::m_Mutex;
+std::condition_variable ObjectNameLock::m_CV;
+std::map<Type*, std::set<String>> ObjectNameLock::m_LockedObjectNames;
+
+/**
+ * Locks the specified object name of the given type and unlocks it upon destruction of the instance of this class.
+ *
+ * If it is already locked, the call blocks until the lock is released.
+ *
+ * @param Type::Ptr ptype The type of the object you want to lock
+ * @param String objName The object name you want to lock
+ */
+ObjectNameLock::ObjectNameLock(const Type::Ptr& ptype, const String& objName): m_ObjectName{objName}, m_Type{ptype}
+{
+	std::unique_lock<std::mutex> lock(m_Mutex);
+	m_CV.wait(lock, [this]{
+		auto& locked = m_LockedObjectNames[m_Type.get()];
+		return locked.find(m_ObjectName) == locked.end();
+	});
+
+	// Add the object name to the locked list to block all other threads that try
+	// to process a message affecting the same object.
+	m_LockedObjectNames[ptype.get()].emplace(objName);
+}
+
+ObjectNameLock::~ObjectNameLock()
+{
+	{
+		std::unique_lock<std::mutex> lock(m_Mutex);
+		m_LockedObjectNames[m_Type.get()].erase(m_ObjectName);
+	}
+	m_CV.notify_all();
+}

lib/remote/configobjectslock.hpp

@@ -2,7 +2,12 @@
 
 #pragma once
 
+#include "base/type.hpp"
+#include "base/string.hpp"
+#include <condition_variable>
+#include <map>
 #include <mutex>
+#include <set>
 
 #ifndef _WIN32
 #include <boost/interprocess/sync/interprocess_sharable_mutex.hpp>
@@ -69,4 +74,29 @@ private:
 
 #endif /* _WIN32 */
 
+
+/**
+ * Allows you to easily lock/unlock a specific object of a given type by its name.
+ *
+ * That way, locking an object "this" of type Host does not affect an object "this" of
+ * type "Service" nor an object "other" of type "Host".
+ *
+ * @ingroup remote
+ */
+class ObjectNameLock
+{
+public:
+	ObjectNameLock(const Type::Ptr& ptype, const String& objName);
+
+	~ObjectNameLock();
+
+private:
+	String m_ObjectName;
+	Type::Ptr m_Type;
+
+	static std::mutex m_Mutex;
+	static std::condition_variable m_CV;
+	static std::map<Type*, std::set<String>> m_LockedObjectNames;
+};
+
 }

lib/remote/configobjectutility.cpp

@@ -5,7 +5,9 @@
 #include "remote/apilistener.hpp"
 #include "config/configcompiler.hpp"
 #include "config/configitem.hpp"
+#include "base/atomic-file.hpp"
 #include "base/configwriter.hpp"
+#include "base/defer.hpp"
 #include "base/exception.hpp"
 #include "base/dependencygraph.hpp"
 #include "base/tlsutility.hpp"
@@ -198,11 +200,16 @@ bool ConfigObjectUtility::CreateObject(const Type::Ptr& type, const String& full
 		return false;
 	}
 
+	// AtomicFile doesn't create not yet existing directories, so we have to do it by ourselves.
 	Utility::MkDirP(Utility::DirName(path), 0700);
 
-	std::ofstream fp(path.CStr(), std::ofstream::out | std::ostream::trunc);
+	AtomicFile::Write(path, 0644, config);
-	fp << config;
+
-	fp.close();
+	// Remove the just created config file in all the error cases and if the object creation
+	// succeeds the deferred callback will be cancelled.
+	Defer removeConfigPath([&path]{
+		Utility::Remove(path);
+	});
 
 	std::unique_ptr<Expression> expr = ConfigCompiler::CompileFile(path, String(), "_api");
 
@@ -227,8 +234,6 @@ bool ConfigObjectUtility::CreateObject(const Type::Ptr& type, const String& full
 				Log(LogNotice, "ConfigObjectUtility")
 					<< "Failed to commit config item '" << fullName << "'. Aborting and removing config path '" << path << "'.";
 
-				Utility::Remove(path);
-
 				for (const boost::exception_ptr& ex : upq.GetExceptions()) {
 					errors->Add(DiagnosticInformation(ex, false));
 
@@ -250,8 +255,6 @@ bool ConfigObjectUtility::CreateObject(const Type::Ptr& type, const String& full
 				Log(LogNotice, "ConfigObjectUtility")
 					<< "Failed to activate config object '" << fullName << "'. Aborting and removing config path '" << path << "'.";
 
-				Utility::Remove(path);
-
 				for (const boost::exception_ptr& ex : upq.GetExceptions()) {
 					errors->Add(DiagnosticInformation(ex, false));
 
@@ -275,16 +278,16 @@ bool ConfigObjectUtility::CreateObject(const Type::Ptr& type, const String& full
 		ConfigObject::Ptr obj = ctype->GetObject(fullName);
 
 		if (obj) {
+			// Object is successfully created and activated, so don't remove its config.
+			removeConfigPath.Cancel();
+
 			Log(LogInformation, "ConfigObjectUtility")
 				<< "Created and activated object '" << fullName << "' of type '" << type->GetName() << "'.";
 		} else {
 			Log(LogNotice, "ConfigObjectUtility")
 				<< "Object '" << fullName << "' was not created but ignored due to errors.";
 		}
-
 	} catch (const std::exception& ex) {
-		Utility::Remove(path);
-
 		if (errors)
 			errors->Add(DiagnosticInformation(ex, false));
 

lib/remote/createobjecthandler.cpp

@@ -124,6 +124,9 @@ bool CreateObjectHandler::HandleRequest(
 		return true;
 	}
 
+	// Lock the object name of the given type to prevent from being created concurrently.
+	ObjectNameLock objectNameLock(type, name);
+
 	if (!ConfigObjectUtility::CreateObject(type, name, config, errors, diagnosticInformation)) {
 		result1->Set("errors", errors);
 		result1->Set("code", 500);

lib/remote/deleteobjecthandler.cpp

@@ -84,6 +84,9 @@ bool DeleteObjectHandler::HandleRequest(
 		Array::Ptr errors = new Array();
 		Array::Ptr diagnosticInformation = new Array();
 
+		// Lock the object name of the given type to prevent from being modified/deleted concurrently.
+		ObjectNameLock objectNameLock(type, obj->GetName());
+
 		if (!ConfigObjectUtility::DeleteObject(obj, cascade, errors, diagnosticInformation)) {
 			code = 500;
 			status = "Object could not be deleted.";

lib/remote/modifyobjecthandler.cpp

@@ -112,6 +112,9 @@ bool ModifyObjectHandler::HandleRequest(
 
 		String key;
 
+		// Lock the object name of the given type to prevent from being modified/deleted concurrently.
+		ObjectNameLock objectNameLock(type, obj->GetName());
+
 		try {
 			if (restoreAttrs) {
 				ObjectLock oLock (restoreAttrs);