Allow including config elements from ZooKeeper [#CLICKHOUSE-2794]

dbms/include/DB/Common/ConfigProcessor.h

@@ -1,6 +1,8 @@
 #pragma once
 
 #include <string>
+#include <unordered_set>
+
 #include <Poco/DOM/Document.h>
 #include <Poco/DOM/DOMParser.h>
 #include <Poco/DOM/DOMWriter.h>
@@ -12,8 +14,15 @@
 #include <Poco/DirectoryIterator.h>
 #include <Poco/ConsoleChannel.h>
 #include <Poco/Util/AbstractConfiguration.h>
+
 #include <common/logger_useful.h>
 
+
+namespace zkutil
+{
+	class ZooKeeperNodeCache;
+}
+
 using ConfigurationPtr = Poco::AutoPtr<Poco::Util::AbstractConfiguration>;
 using XMLDocumentPtr = Poco::AutoPtr<Poco::XML::Document>;
 
@@ -25,6 +34,8 @@ public:
 	/// log_to_console нужно использовать, если система логгирования еще не инициализирована.
 	ConfigProcessor(bool throw_on_bad_incl = false, bool log_to_console = false, const Substitutions & substitutions = Substitutions());
 
+	~ConfigProcessor();
+
 	/** Выполняет подстановки в конфиге и возвращает XML-документ.
 	  *
 	  * Пусть в качестве path передана "/path/file.xml"
@@ -37,12 +48,34 @@ public:
 	  * 3) Заменяем элементы вида "<foo incl="bar"/>" на "<foo>содержимое элемента yandex.bar из metrika.xml</foo>"
 	  * 4) Заменяет "<layer/>" на "<layer>номер слоя из имени хоста</layer>"
 	  */
-	XMLDocumentPtr processConfig(const std::string & path);
+	XMLDocumentPtr processConfig(
+			const std::string & path,
+			bool * has_zk_includes = nullptr,
+			zkutil::ZooKeeperNodeCache * zk_node_cache = nullptr);
+
+
+	struct LoadedConfig
+	{
+		ConfigurationPtr configuration;
+		bool has_zk_includes;
+		bool loaded_from_preprocessed;
+		bool preprocessed_written;
+	};
 
 	/** Делает processConfig и создает из результата Poco::Util::XMLConfiguration.
 	  * Еще сохраняет результат в файл по пути, полученному из path приписыванием строки "-preprocessed" к имени файла.
 	  */
-	ConfigurationPtr loadConfig(const std::string & path);
+
+	/// If allow_zk_includes is true, expects that the configuration xml can contain from_zk nodes.
+	/// If the xml contains them, set has_zk_includes to true and don't write config-preprocessed.xml,
+	/// expecting that config would be reloaded with zookeeper later.
+
+	LoadedConfig loadConfig(const std::string & path, bool allow_zk_includes = false);
+
+	LoadedConfig loadConfigWithZooKeeperIncludes(
+			const std::string & path,
+			zkutil::ZooKeeperNodeCache & zk_node_cache,
+			bool fallback_to_preprocessed = false);
 
 public:
 
@@ -51,21 +84,31 @@ public:
 	static Files getConfigMergeFiles(const std::string & config_path);
 
 private:
+	bool throw_on_bad_incl;
+
 	Logger * log;
 	Poco::AutoPtr<Poco::Channel> channel_ptr;
-	bool throw_on_bad_incl;
+
 	Substitutions substitutions;
 
-	using DocumentPtr = XMLDocumentPtr;
+	Poco::AutoPtr<Poco::XML::NamePool> name_pool;
+	Poco::XML::DOMParser dom_parser;
+
+private:
 	using NodePtr = Poco::AutoPtr<Poco::XML::Node>;
 
-	void mergeRecursive(DocumentPtr config, Poco::XML::Node * config_node, Poco::XML::Node * with_node);
+	void mergeRecursive(XMLDocumentPtr config, Poco::XML::Node * config_node, Poco::XML::Node * with_node);
 
-	void merge(DocumentPtr config, DocumentPtr with);
+	void merge(XMLDocumentPtr config, XMLDocumentPtr with);
 
 	std::string layerFromHost();
 
-	void doIncludesRecursive(DocumentPtr config, DocumentPtr include_from, Poco::XML::Node * node);
+	void doIncludesRecursive(
+			XMLDocumentPtr config,
+			XMLDocumentPtr include_from,
+			Poco::XML::Node * node,
+			zkutil::ZooKeeperNodeCache * zk_node_cache,
+			std::unordered_set<std::string> & contributing_zk_paths);
 
-	void doIncludes(DocumentPtr config, DocumentPtr include_from);
+	void savePreprocessedConfig(const XMLDocumentPtr & config, const std::string & preprocessed_path);
 };

dbms/src/Common/ConfigProcessor.cpp

@@ -3,12 +3,15 @@
 #include <cerrno>
 #include <cstring>
 #include <iostream>
+#include <functional>
 
 #include <Poco/DOM/Text.h>
 #include <Poco/DOM/Attr.h>
 #include <Poco/DOM/Comment.h>
 #include <Poco/Util/XMLConfiguration.h>
 
+#include <zkutil/ZooKeeperNodeCache.h>
+
 using namespace Poco::XML;
 
 
@@ -37,7 +40,13 @@ static std::string numberFromHost(const std::string & s)
 }
 
 ConfigProcessor::ConfigProcessor(bool throw_on_bad_incl_, bool log_to_console, const Substitutions & substitutions_)
-	: throw_on_bad_incl(throw_on_bad_incl_), substitutions(substitutions_)
+	: throw_on_bad_incl(throw_on_bad_incl_)
+	, substitutions(substitutions_)
+	/// We need larger name pool to allow to support vast amount of users in users.xml files for ClickHouse.
+	/// Size is prime because Poco::XML::NamePool uses bad (inefficient, low quality)
+	///  hash function internally, and its size was prime by default.
+	, name_pool(new Poco::XML::NamePool(65521))
+	, dom_parser(name_pool)
 {
 	if (log_to_console && Logger::has("ConfigProcessor") == nullptr)
 	{
@@ -50,6 +59,13 @@ ConfigProcessor::ConfigProcessor(bool throw_on_bad_incl_, bool log_to_console, c
 	}
 }
 
+ConfigProcessor::~ConfigProcessor()
+{
+	if (channel_ptr) /// This means we have created a new console logger in the constructor.
+		Logger::destroy("ConfigProcessor");
+}
+
+
 /// Вектор из имени элемента и отсортированного списка имен и значений атрибутов (кроме атрибутов replace и remove).
 /// Взаимно однозначно задает имя элемента и список его атрибутов. Нужен, чтобы сравнивать элементы.
 using ElementIdentifier = std::vector<std::string>;
@@ -67,7 +83,7 @@ static ElementIdentifier getElementIdentifier(Node * element)
 	{
 		Node * node = attrs->item(i);
 		std::string name = node->nodeName();
-		if (name == "replace" || name == "remove" || name == "incl")
+		if (name == "replace" || name == "remove" || name == "incl" || name == "from_zk")
 			continue;
 		std::string value = node->nodeValue();
 		attrs_kv.push_back(std::make_pair(name, value));
@@ -104,7 +120,14 @@ static bool allWhitespace(const std::string & s)
 	return s.find_first_not_of(" \t\n\r") == std::string::npos;
 }
 
-void ConfigProcessor::mergeRecursive(DocumentPtr config, Node * config_root, Node * with_root)
+static std::string preprocessedConfigPath(const std::string & path)
+{
+	Poco::Path preprocessed_path(path);
+	preprocessed_path.setBaseName(preprocessed_path.getBaseName() + "-preprocessed");
+	return preprocessed_path.toString();
+}
+
+void ConfigProcessor::mergeRecursive(XMLDocumentPtr config, Node * config_root, Node * with_root)
 {
 	NodeListPtr with_nodes = with_root->childNodes();
 	using ElementsByIdentifier = std::multimap<ElementIdentifier, Node *>;
@@ -137,7 +160,7 @@ void ConfigProcessor::mergeRecursive(DocumentPtr config, Node * config_root, Nod
 			bool replace = with_element->hasAttribute("replace");
 
 			if (remove && replace)
-				throw Poco::Exception("remove and replace attributes on the same element");
+				throw Poco::Exception("both remove and replace attributes set for element <" + with_node->nodeName() + ">");
 
 			ElementsByIdentifier::iterator it = config_element_by_id.find(getElementIdentifier(with_node));
 
@@ -171,7 +194,7 @@ void ConfigProcessor::mergeRecursive(DocumentPtr config, Node * config_root, Nod
 	}
 }
 
-void ConfigProcessor::merge(DocumentPtr config, DocumentPtr with)
+void ConfigProcessor::merge(XMLDocumentPtr config, XMLDocumentPtr with)
 {
 	mergeRecursive(config, getRootNode(&*config), getRootNode(&*with));
 }
@@ -189,7 +212,12 @@ std::string ConfigProcessor::layerFromHost()
 	return layer;
 }
 
-void ConfigProcessor::doIncludesRecursive(DocumentPtr config, DocumentPtr include_from, Node * node)
+void ConfigProcessor::doIncludesRecursive(
+		XMLDocumentPtr config,
+		XMLDocumentPtr include_from,
+		Node * node,
+		zkutil::ZooKeeperNodeCache * zk_node_cache,
+		std::unordered_set<std::string> & contributing_zk_paths)
 {
 	if (node->nodeType() == Node::TEXT_NODE)
 	{
@@ -226,62 +254,93 @@ void ConfigProcessor::doIncludesRecursive(DocumentPtr config, DocumentPtr includ
 
 	NamedNodeMapPtr attributes = node->attributes();
 	Node * incl_attribute = attributes->getNamedItem("incl");
+	Node * from_zk_attribute = attributes->getNamedItem("from_zk");
+
+	if (incl_attribute && from_zk_attribute)
+		throw Poco::Exception("both incl and from_zk attributes set for element <" + node->nodeName() + ">");
 
 	/// Заменять имеющееся значение, а не добавлять к нему.
 	bool replace = attributes->getNamedItem("replace");
 
-	if (incl_attribute)
+	auto process_include = [&](const Node * include_attr, const std::function<Node * (const std::string &)> & get_node, const char * error_msg)
 	{
-		std::string name = incl_attribute->getNodeValue();
-		Node * included_node = include_from ? include_from->getNodeByPath("yandex/" + name) : nullptr;
-		if (!included_node)
+		std::string name = include_attr->getNodeValue();
+		Node * node_to_include = get_node(name);
+		if (!node_to_include)
 		{
 			if (attributes->getNamedItem("optional"))
 				node->parentNode()->removeChild(node);
 			else if (throw_on_bad_incl)
-				throw Poco::Exception("Include not found: " + name);
+				throw Poco::Exception(error_msg + name);
 			else
-				LOG_WARNING(log, "Include not found: " << name);
+				LOG_WARNING(log, error_msg << name);
 		}
 		else
 		{
+			Element * element = dynamic_cast<Element *>(node);
+
+			element->removeAttribute("incl");
+			element->removeAttribute("from_zk");
+
 			if (replace)
+			{
 				while (Node * child = node->firstChild())
 					node->removeChild(child);
 
-			NodeListPtr children = included_node->childNodes();
+				element->removeAttribute("replace");
+			}
+
+			NodeListPtr children = node_to_include->childNodes();
 			for (size_t i = 0; i < children->length(); ++i)
 			{
 				NodePtr new_node = config->importNode(children->item(i), true);
 				node->appendChild(new_node);
 			}
 
-			Element * element = dynamic_cast<Element *>(node);
-			element->removeAttribute("incl");
-
-			if (replace)
-				element->removeAttribute("replace");
-
-			NamedNodeMapPtr from_attrs = included_node->attributes();
+			NamedNodeMapPtr from_attrs = node_to_include->attributes();
 			for (size_t i = 0; i < from_attrs->length(); ++i)
 			{
 				element->setAttributeNode(dynamic_cast<Attr *>(config->importNode(from_attrs->item(i), true)));
 			}
 		}
+	};
+
+	auto get_incl_node = [&](const std::string & name)
+	{
+		return include_from ? include_from->getNodeByPath("yandex/" + name) : nullptr;
+	};
+	if (incl_attribute)
+		process_include(incl_attribute, get_incl_node, "Include not found: ");
+
+	if (from_zk_attribute)
+	{
+		contributing_zk_paths.insert(from_zk_attribute->getNodeValue());
+
+		if (zk_node_cache)
+		{
+			XMLDocumentPtr zk_document;
+			auto get_zk_node = [&](const std::string & name) -> Node *
+			{
+				std::experimental::optional<std::string> contents = zk_node_cache->get(name);
+				if (!contents)
+					return nullptr;
+
+				/// Enclose contents into a fake <from_zk> tag to allow pure text substitutions.
+				zk_document = dom_parser.parseString("<from_zk>" + contents.value() + "</from_zk>");
+				return getRootNode(zk_document.get());
+			};
+
+			process_include(from_zk_attribute, get_zk_node, "Could not get ZooKeeper node: ");
+		}
 	}
 
 	NodeListPtr children = node->childNodes();
 	for (size_t i = 0; i < children->length(); ++i)
 	{
-		doIncludesRecursive(config, include_from, children->item(i));
+		doIncludesRecursive(config, include_from, children->item(i), zk_node_cache, contributing_zk_paths);
 	}
 }
 
-void ConfigProcessor::doIncludes(DocumentPtr config, DocumentPtr include_from)
-{
-	doIncludesRecursive(config, include_from, getRootNode(&*config));
-}
-
 ConfigProcessor::Files ConfigProcessor::getConfigMergeFiles(const std::string & config_path)
 {
 	Files res;
@@ -314,15 +373,12 @@ ConfigProcessor::Files ConfigProcessor::getConfigMergeFiles(const std::string &
 	return res;
 }
 
-XMLDocumentPtr ConfigProcessor::processConfig(const std::string & path_str)
+XMLDocumentPtr ConfigProcessor::processConfig(
+		const std::string & path_str,
+		bool * has_zk_includes,
+		zkutil::ZooKeeperNodeCache * zk_node_cache)
 {
-	/// We need larger name pool to allow to support vast amount of users in users.xml files for ClickHouse.
-	/// Size is prime because Poco::XML::NamePool uses bad (inefficient, low quality)
-	///  hash function internally, and its size was prime by default.
-	Poco::AutoPtr<Poco::XML::NamePool> name_pool(new Poco::XML::NamePool(65521));
-	Poco::XML::DOMParser dom_parser(name_pool);
-
-	DocumentPtr config = dom_parser.parse(path_str);
+	XMLDocumentPtr config = dom_parser.parse(path_str);
 
 	std::vector<std::string> contributing_files;
 	contributing_files.push_back(path_str);
@@ -331,7 +387,7 @@ XMLDocumentPtr ConfigProcessor::processConfig(const std::string & path_str)
 	{
 		try
 		{
-			DocumentPtr with = dom_parser.parse(merge_file);
+			XMLDocumentPtr with = dom_parser.parse(merge_file);
 			merge(config, with);
 			contributing_files.push_back(merge_file);
 		}
@@ -341,10 +397,11 @@ XMLDocumentPtr ConfigProcessor::processConfig(const std::string & path_str)
 		}
 	}
 
+	std::unordered_set<std::string> contributing_zk_paths;
 	try
 	{
 		Node * node = config->getNodeByPath("yandex/include_from");
-		DocumentPtr include_from;
+		XMLDocumentPtr include_from;
 		std::string include_from_path;
 		if (node)
 		{
@@ -362,13 +419,16 @@ XMLDocumentPtr ConfigProcessor::processConfig(const std::string & path_str)
 			include_from = dom_parser.parse(include_from_path);
 		}
 
-		doIncludes(config, include_from);
+		doIncludesRecursive(config, include_from, getRootNode(config.get()), zk_node_cache, contributing_zk_paths);
 	}
 	catch (Poco::Exception & e)
 	{
-		throw Poco::Exception("Failed to preprocess config: " + e.displayText());
+		throw Poco::Exception("Failed to preprocess config `" + path_str + "': " + e.displayText(), e);
 	}
 
+	if (has_zk_includes)
+		*has_zk_includes = !contributing_zk_paths.empty();
+
 	std::stringstream comment;
 	comment <<     " This file was generated automatically.\n";
 	comment << "     Do not edit it: it is likely to be discarded and generated again before it's read next time.\n";
@@ -377,7 +437,14 @@ XMLDocumentPtr ConfigProcessor::processConfig(const std::string & path_str)
 	{
 		comment << "\n       " << path;
 	}
-	comment<<"      ";
+	if (zk_node_cache && !contributing_zk_paths.empty())
+	{
+		comment << "\n     ZooKeeper nodes used to generate this file:";
+		for (const std::string & path : contributing_zk_paths)
+			comment << "\n       " << path;
+	}
+
+	comment << "      ";
 	NodePtr new_node = config->createTextNode("\n\n");
 	config->insertBefore(new_node, config->firstChild());
 	new_node = config->createComment(comment.str());
@@ -386,20 +453,74 @@ XMLDocumentPtr ConfigProcessor::processConfig(const std::string & path_str)
 	return config;
 }
 
-ConfigurationPtr ConfigProcessor::loadConfig(const std::string & path)
+ConfigProcessor::LoadedConfig ConfigProcessor::loadConfig(const std::string & path, bool allow_zk_includes)
 {
-	DocumentPtr res = processConfig(path);
+	bool has_zk_includes;
+	XMLDocumentPtr config_xml = processConfig(path, &has_zk_includes);
 
-	Poco::Path preprocessed_path(path);
-	preprocessed_path.setBaseName(preprocessed_path.getBaseName() + "-preprocessed");
+	if (has_zk_includes && !allow_zk_includes)
+		throw Poco::Exception("Error while loading config `" + path + "': from_zk includes are not allowed!");
+
+	bool preprocessed_written = false;
+	if (!has_zk_includes)
+	{
+		savePreprocessedConfig(config_xml, preprocessedConfigPath(path));
+		preprocessed_written = true;
+	}
+
+	ConfigurationPtr configuration(new Poco::Util::XMLConfiguration(config_xml));
+
+	return LoadedConfig{configuration, has_zk_includes, /* loaded_from_preprocessed = */ false, preprocessed_written};
+}
+
+ConfigProcessor::LoadedConfig ConfigProcessor::loadConfigWithZooKeeperIncludes(
+		const std::string & path,
+		zkutil::ZooKeeperNodeCache & zk_node_cache,
+		bool fallback_to_preprocessed)
+{
+	std::string preprocessed_path = preprocessedConfigPath(path);
+
+	XMLDocumentPtr config_xml;
+	bool has_zk_includes;
+	bool processed_successfully = false;
 	try
 	{
-		DOMWriter().writeNode(preprocessed_path.toString(), res);
+		config_xml = processConfig(path, &has_zk_includes, &zk_node_cache);
+		processed_successfully = true;
+	}
+	catch (const Poco::Exception & ex)
+	{
+		if (!fallback_to_preprocessed)
+			throw;
+
+		const auto * zk_exception = dynamic_cast<const zkutil::KeeperException *>(ex.nested());
+		if (!zk_exception)
+			throw;
+
+		LOG_WARNING(
+				log,
+				"Error while processing from_zk config includes: " + zk_exception->message() +
+				". Config will be loaded from preprocessed file: " + preprocessed_path);
+
+		config_xml = dom_parser.parse(preprocessed_path);
+	}
+
+	if (processed_successfully)
+		savePreprocessedConfig(config_xml, preprocessed_path);
+
+	ConfigurationPtr configuration(new Poco::Util::XMLConfiguration(config_xml));
+
+	return LoadedConfig{configuration, has_zk_includes, !processed_successfully, processed_successfully};
+}
+
+void ConfigProcessor::savePreprocessedConfig(const XMLDocumentPtr & config, const std::string & preprocessed_path)
+{
+	try
+	{
+		DOMWriter().writeNode(preprocessed_path, config);
 	}
 	catch (Poco::Exception & e)
 	{
-		LOG_WARNING(log, "Couldn't save preprocessed config to " << preprocessed_path.toString() << ": " << e.displayText());
+		LOG_WARNING(log, "Couldn't save preprocessed config to " << preprocessed_path << ": " << e.displayText());
 	}
-
-	return new Poco::Util::XMLConfiguration(res);
 }

dbms/src/Interpreters/tests/CMakeLists.txt

@@ -51,4 +51,4 @@ target_link_libraries (in_join_subqueries_preprocessor dbms)
 add_check(in_join_subqueries_preprocessor)
 
 add_executable (users users.cpp)
-target_link_libraries (users dbms ${Boost_FILESYSTEM_LIBRARY})
+target_link_libraries (users dbms ${Boost_FILESYSTEM_LIBRARY} zkutil dbms)

dbms/src/Interpreters/tests/users.cpp

@@ -195,7 +195,7 @@ void runOneTest(size_t test_num, const TestDescriptor & test_descriptor)
 
 	try
 	{
-		config = ConfigProcessor{}.loadConfig(path_name);
+		config = ConfigProcessor{}.loadConfig(path_name).configuration;
 	}
 	catch (const Poco::Exception & ex)
 	{

dbms/src/Server/ConfigReloader.cpp

@@ -13,42 +13,31 @@
 namespace DB
 {
 
-namespace ErrorCodes { extern const int FILE_DOESNT_EXIST; }
-
-
 constexpr decltype(ConfigReloader::reload_interval) ConfigReloader::reload_interval;
 
-
-ConfigReloader::ConfigReloader(const std::string & main_config_path_, const std::string & users_config_path_,
-							   const std::string & include_from_path_, Context * context_)
-	: main_config_path(main_config_path_), users_config_path(users_config_path_),
-	include_from_path(include_from_path_), context(context_)
+ConfigReloader::ConfigReloader(
+		const std::string & path_,
+		const std::string & include_from_path_,
+		zkutil::ZooKeeperNodeCache && zk_node_cache_,
+		Updater && updater_,
+		bool already_loaded)
+	: path(path_), include_from_path(include_from_path_)
+	, zk_node_cache(std::move(zk_node_cache_))
+	, updater(std::move(updater_))
 {
-	/// If path to users' config isn't absolute, try guess its root (current) dir.
-	/// At first, try to find it in dir of main config, after will use current dir.
-	if (users_config_path.empty() || users_config_path[0] != '/')
-	{
-		std::string config_dir = Poco::Path(main_config_path).parent().toString();
-		if (Poco::File(config_dir + users_config_path).exists())
-			users_config_path = config_dir + users_config_path;
-	}
-
-	/// Setup users on server init
-	reloadIfNewer(false, true);
+	if (!already_loaded)
+		reloadIfNewer(/* force = */ true, /* throw_on_error = */ true, /* fallback_to_preprocessed = */ true);
 
 	thread = std::thread(&ConfigReloader::run, this);
 }
 
+
 ConfigReloader::~ConfigReloader()
 {
 	try
 	{
-		{
-			std::lock_guard<std::mutex> lock{mutex};
-			quit = true;
-		}
-
-		cond.notify_one();
+		quit = true;
+		zk_node_cache.getChangedEvent().set();
 		thread.join();
 	}
 	catch (...)
@@ -62,104 +51,106 @@ void ConfigReloader::run()
 {
 	setThreadName("ConfigReloader");
 
-	std::unique_lock<std::mutex> lock{mutex};
-
 	while (true)
 	{
-		if (cond.wait_for(lock, reload_interval, [this] { return quit; }))
-			break;
+		bool zk_changed = zk_node_cache.getChangedEvent().tryWait(std::chrono::milliseconds(reload_interval).count());
+		if (quit)
+			return;
 
-		reloadIfNewer(false, false);
+		reloadIfNewer(zk_changed, /* throw_on_error = */ false, /* fallback_to_preprocessed = */ false);
 	}
 }
 
+void ConfigReloader::reloadIfNewer(bool force, bool throw_on_error, bool fallback_to_preprocessed)
+{
+	FilesChangesTracker new_files = getNewFileList();
+	if (force || new_files.isDifferOrNewerThan(files))
+	{
+		ConfigProcessor::LoadedConfig loaded_config;
+		try
+		{
+			LOG_DEBUG(log, "Loading config `" << path << "'");
 
-ConfigReloader::FilesChangesTracker ConfigReloader::getFileListFor(const std::string & root_config_path)
+			loaded_config = ConfigProcessor().loadConfig(path, /* allow_zk_includes = */ true);
+			if (loaded_config.has_zk_includes)
+				loaded_config = ConfigProcessor().loadConfigWithZooKeeperIncludes(
+						path, zk_node_cache, fallback_to_preprocessed);
+		}
+		catch (...)
+		{
+			if (throw_on_error)
+				throw;
+
+			tryLogCurrentException(log, "Error loading config from `" + path + "'");
+			return;
+		}
+
+		/** We should remember last modification time if and only if config was sucessfully loaded
+		 * Otherwise a race condition could occur during config files update:
+		 *  File is contain raw (and non-valid) data, therefore config is not applied.
+		 *  When file has been written (and contain valid data), we don't load new data since modification time remains the same.
+		 */
+		if (!loaded_config.loaded_from_preprocessed)
+			files = std::move(new_files);
+
+		try
+		{
+			updater(loaded_config.configuration);
+		}
+		catch (...)
+		{
+			if (throw_on_error)
+				throw;
+			tryLogCurrentException(log, "Error updating configuration from `" + path + "' config.");
+		}
+	}
+}
+
+struct ConfigReloader::FileWithTimestamp
+{
+	std::string path;
+	time_t modification_time;
+
+	FileWithTimestamp(const std::string & path_, time_t modification_time_)
+		: path(path_), modification_time(modification_time_) {}
+
+	bool operator < (const FileWithTimestamp & rhs) const
+	{
+		return path < rhs.path;
+	}
+
+	static bool isTheSame(const FileWithTimestamp & lhs, const FileWithTimestamp & rhs)
+	{
+		return (lhs.modification_time == rhs.modification_time) && (lhs.path == rhs.path);
+	}
+};
+
+
+void ConfigReloader::FilesChangesTracker::addIfExists(const std::string & path)
+{
+	if (!path.empty() && Poco::File(path).exists())
+	{
+		files.emplace(path, Poco::File(path).getLastModified().epochTime());
+	}
+}
+
+bool ConfigReloader::FilesChangesTracker::isDifferOrNewerThan(const FilesChangesTracker & rhs)
+{
+	return (files.size() != rhs.files.size()) ||
+		!std::equal(files.begin(), files.end(), rhs.files.begin(), FileWithTimestamp::isTheSame);
+}
+
+ConfigReloader::FilesChangesTracker ConfigReloader::getNewFileList() const
 {
 	FilesChangesTracker file_list;
 
-	file_list.addIfExists(root_config_path);
+	file_list.addIfExists(path);
 	file_list.addIfExists(include_from_path);
 
-	for (const auto & path : ConfigProcessor::getConfigMergeFiles(root_config_path))
-		file_list.addIfExists(path);
+	for (const auto & merge_path : ConfigProcessor::getConfigMergeFiles(path))
+		file_list.addIfExists(merge_path);
 
- 	return file_list;
+	return file_list;
 }
 
-
-ConfigurationPtr ConfigReloader::loadConfigFor(const std::string & root_config_path, bool throw_on_error)
-{
-	ConfigurationPtr config;
-
-	LOG_DEBUG(log, "Loading config '" << root_config_path << "'");
-
-	try
-	{
-		config = ConfigProcessor().loadConfig(root_config_path);
-	}
-	catch (...)
-	{
-		if (throw_on_error)
-			throw;
-
-		tryLogCurrentException(log, "Error loading config from '" + root_config_path + "' ");
-		return nullptr;
-	}
-
-	return config;
-}
-
-
-void ConfigReloader::reloadIfNewer(bool force_main, bool force_users)
-{
-	FilesChangesTracker main_config_files = getFileListFor(main_config_path);
-	if (force_main || main_config_files.isDifferOrNewerThan(last_main_config_files))
-	{
-		ConfigurationPtr config = loadConfigFor(main_config_path, force_main);
-		if (config)
-		{
-			/** We should remember last modification time if and only if config was sucessfully loaded
-			  * Otherwise a race condition could occur during config files update:
-			  *  File is contain raw (and non-valid) data, therefore config is not applied.
-			  *  When file has been written (and contain valid data), we don't load new data since modification time remains the same.
-			  */
-			last_main_config_files = std::move(main_config_files);
-
-			try
-			{
-				context->setClustersConfig(config);
-			}
-			catch (...)
-			{
-				if (force_main)
-					throw;
-				tryLogCurrentException(log, "Error updating remote_servers config from '" + main_config_path + "' ");
-			}
-		}
-	}
-
-	FilesChangesTracker users_config_files = getFileListFor(users_config_path);
-	if (force_users || users_config_files.isDifferOrNewerThan(last_users_config_files))
-	{
-		ConfigurationPtr config = loadConfigFor(users_config_path, force_users);
-		if (config)
-		{
-			last_users_config_files = std::move(users_config_files);
-
-			try
-			{
-				context->setUsersConfig(config);
-			}
-			catch (...)
-			{
-				if (force_users)
-					throw;
-				tryLogCurrentException(log, "Error updating users config from '" + users_config_path + "' ");
-			}
-		}
-	}
-}
-
-
 }

dbms/src/Server/ConfigReloader.h

@@ -1,6 +1,8 @@
 #pragma once
 
 #include <DB/Common/ConfigProcessor.h>
+#include <zkutil/Common.h>
+#include <zkutil/ZooKeeperNodeCache.h>
 
 #include <time.h>
 #include <string>
@@ -19,89 +21,57 @@ class Context;
 
 /** Every two seconds checks configuration files for update.
   * If configuration is changed, then config will be reloaded by ConfigProcessor
-  *  and the reloaded config will be applied via setUsersConfig() and setClusters() methods of Context.
-  * So, ConfigReloader actually reloads only <users> and <remote_servers> "tags".
-  * Also, it doesn't take into account changes of --config-file, <users_config> and <include_from> parameters.
+  *  and the reloaded config will be applied via Updater functor.
+  * It doesn't take into account changes of --config-file, <users_config> and <include_from> parameters.
   */
 class ConfigReloader
 {
 public:
-	/** main_config_path is usually /path/to/.../clickhouse-server/config.xml (i.e. --config-file value)
-	  * users_config_path is usually /path/to/.../clickhouse-server/users.xml (i.e. value of <users_config> tag)
-	  * include_from_path is usually /path/to/.../etc/metrika.xml (i.e. value of <include_from> tag)
+	using Updater = std::function<void(ConfigurationPtr)>;
+
+	/** include_from_path is usually /etc/metrika.xml (i.e. value of <include_from> tag)
 	  */
-	ConfigReloader(const std::string & main_config_path_, const std::string & users_config_path_, const std::string & include_from_path_, Context * context_);
+	ConfigReloader(
+			const std::string & path,
+			const std::string & include_from_path,
+			zkutil::ZooKeeperNodeCache && zk_node_cache,
+			Updater && updater,
+			bool already_loaded);
 
 	~ConfigReloader();
 
 private:
+	void run();
 
-	struct FileWithTimestamp
-	{
-		std::string path;
-		time_t modification_time;
+	void reloadIfNewer(bool force, bool throw_on_error, bool fallback_to_preprocessed);
 
-		FileWithTimestamp(const std::string & path_, time_t modification_time_)
-			: path(path_), modification_time(modification_time_) {}
-
-		bool operator < (const FileWithTimestamp & rhs) const
-		{
-			return path < rhs.path;
-		}
-
-		static bool isTheSame(const FileWithTimestamp & lhs, const FileWithTimestamp & rhs)
-		{
-			return (lhs.modification_time == rhs.modification_time) && (lhs.path == rhs.path);
-		}
-	};
+	struct FileWithTimestamp;
 
 	struct FilesChangesTracker
 	{
 		std::set<FileWithTimestamp> files;
 
-		void addIfExists(const std::string & path)
-		{
-			if (!path.empty() && Poco::File(path).exists())
-			{
-				files.emplace(path, Poco::File(path).getLastModified().epochTime());
-			}
-		}
-
-		bool isDifferOrNewerThan(const FilesChangesTracker & rhs)
-		{
-			return (files.size() != rhs.files.size()) ||
-					!std::equal(files.begin(), files.end(), rhs.files.begin(), FileWithTimestamp::isTheSame);
-		}
+		void addIfExists(const std::string & path);
+		bool isDifferOrNewerThan(const FilesChangesTracker & rhs);
 	};
 
-private:
-
-	/// Make sense to separate this function on two threads
-	void reloadIfNewer(bool force_main, bool force_users);
-	void run();
-
-	FilesChangesTracker getFileListFor(const std::string & root_config_path);
-	ConfigurationPtr loadConfigFor(const std::string & root_config_path, bool throw_error);
+	FilesChangesTracker getNewFileList() const;
 
 private:
 
 	static constexpr auto reload_interval = std::chrono::seconds(2);
 
-	std::string main_config_path;
-	std::string users_config_path;
-	std::string include_from_path;
-
-	Context * context;
-
-	FilesChangesTracker last_main_config_files;
-	FilesChangesTracker last_users_config_files;
-
-	bool quit {false};
-	std::mutex mutex;
-	std::condition_variable cond;
-	std::thread thread;
-
 	Poco::Logger * log = &Logger::get("ConfigReloader");
+
+	std::string path;
+	std::string include_from_path;
+	FilesChangesTracker files;
+	zkutil::ZooKeeperNodeCache zk_node_cache;
+
+	Updater updater;
+
+	std::atomic<bool> quit{false};
+	std::thread thread;
 };
 
 }

dbms/src/Server/LocalServer.cpp

@@ -243,7 +243,9 @@ try
 	/// Load config files if exists
 	if (config().has("config-file") || Poco::File("config.xml").exists())
 	{
-		ConfigurationPtr processed_config = ConfigProcessor(false, true).loadConfig(config().getString("config-file", "config.xml"));
+		ConfigurationPtr processed_config = ConfigProcessor(false, true)
+			.loadConfig(config().getString("config-file", "config.xml"))
+			.configuration;
 		config().add(processed_config.duplicate(), PRIO_DEFAULT, false);
 	}
 
@@ -438,7 +440,7 @@ void LocalServer::setupUsers()
 	if (config().has("users_config") || config().has("config-file") || Poco::File("config.xml").exists())
 	{
 		auto users_config_path = config().getString("users_config", config().getString("config-file", "config.xml"));
-		users_config = ConfigProcessor().loadConfig(users_config_path);
+		users_config = ConfigProcessor().loadConfig(users_config_path).configuration;
 	}
 	else
 	{

dbms/src/Server/MetricsTransmitter.cpp

@@ -3,6 +3,7 @@
 #include <daemon/BaseDaemon.h>
 #include <DB/Common/setThreadName.h>
 #include <DB/Common/CurrentMetrics.h>
+#include <DB/Common/Exception.h>
 #include <DB/Interpreters/AsynchronousMetrics.h>
 
 

dbms/src/Server/Server.cpp

@@ -209,6 +209,23 @@ int Server::main(const std::vector<std::string> & args)
 	global_context->setGlobalContext(*global_context);
 	global_context->setApplicationType(Context::ApplicationType::SERVER);
 
+	bool has_zookeeper = false;
+	if (config().has("zookeeper"))
+	{
+		global_context->setZooKeeper(std::make_shared<zkutil::ZooKeeper>(config(), "zookeeper"));
+		has_zookeeper = true;
+	}
+
+	zkutil::ZooKeeperNodeCache main_config_zk_node_cache([&] { return global_context->getZooKeeper(); });
+	if (loaded_config.has_zk_includes)
+	{
+		auto old_configuration = loaded_config.configuration;
+		loaded_config = ConfigProcessor().loadConfigWithZooKeeperIncludes(
+				config_path, main_config_zk_node_cache, /* fallback_to_preprocessed = */ true);
+		config().removeConfiguration(old_configuration.get());
+		config().add(loaded_config.configuration.duplicate(), PRIO_DEFAULT, false);
+	}
+
 	std::string path = getCanonicalPath(config().getString("path"));
 	std::string default_database = config().getString("default_database", "default");
 
@@ -278,13 +295,6 @@ int Server::main(const std::vector<std::string> & args)
 	Poco::File(path + "flags/").createDirectories();
 	global_context->setFlagsPath(path + "flags/");
 
-	bool has_zookeeper = false;
-	if (config().has("zookeeper"))
-	{
-		global_context->setZooKeeper(std::make_shared<zkutil::ZooKeeper>(config(), "zookeeper"));
-		has_zookeeper = true;
-	}
-
 	if (config().has("interserver_http_port"))
 	{
 		String this_host = config().getString("interserver_http_host", "");
@@ -309,11 +319,29 @@ int Server::main(const std::vector<std::string> & args)
 	if (config().has("macros"))
 		global_context->setMacros(Macros(config(), "macros"));
 
-	/// Initialize automatic config updater
-	std::string main_config_path = config().getString("config-file", "config.xml");
-	std::string users_config_path = config().getString("users_config", main_config_path);
+	/// Initialize main config reloader.
 	std::string include_from_path = config().getString("include_from", "/etc/metrika.xml");
-	auto config_reloader = std::make_unique<ConfigReloader>(main_config_path, users_config_path, include_from_path, global_context.get());
+	auto main_config_reloader = std::make_unique<ConfigReloader>(
+			config_path, include_from_path,
+			std::move(main_config_zk_node_cache),
+			[&](ConfigurationPtr config) { global_context->setClustersConfig(config); },
+			/* already_loaded = */ true);
+
+	/// Initialize users config reloader.
+	std::string users_config_path = config().getString("users_config", config_path);
+	/// If path to users' config isn't absolute, try guess its root (current) dir.
+	/// At first, try to find it in dir of main config, after will use current dir.
+	if (users_config_path.empty() || users_config_path[0] != '/')
+	{
+		std::string config_dir = Poco::Path(config_path).parent().toString();
+		if (Poco::File(config_dir + users_config_path).exists())
+			users_config_path = config_dir + users_config_path;
+	}
+	auto users_config_reloader = std::make_unique<ConfigReloader>(
+			users_config_path, include_from_path,
+			zkutil::ZooKeeperNodeCache([&] { return global_context->getZooKeeper(); }),
+			[&](ConfigurationPtr config) { global_context->setUsersConfig(config); },
+			/* already_loaded = */ false);
 
 	/// Limit on total number of coucurrently executed queries.
 	global_context->getProcessList().setMaxSize(config().getInt("max_concurrent_queries", 0));
@@ -535,7 +563,8 @@ int Server::main(const std::vector<std::string> & args)
 
 			LOG_DEBUG(log, "Closed all connections.");
 
-			config_reloader.reset();
+			main_config_reloader.reset();
+			users_config_reloader.reset();
 		});
 
 		/// try to load dictionaries immediately, throw on error and die

libs/libdaemon/include/daemon/BaseDaemon.h

@@ -6,6 +6,9 @@
 #include <iostream>
 #include <memory>
 #include <functional>
+#include <experimental/optional>
+#include <mutex>
+#include <condition_variable>
 
 #include <Poco/Process.h>
 #include <Poco/ThreadPool.h>
@@ -23,9 +26,7 @@
 
 #include <daemon/GraphiteWriter.h>
 
-#include <experimental/optional>
-#include <zkutil/ZooKeeperHolder.h>
-
+#include <DB/Common/ConfigProcessor.h>
 
 namespace Poco { class TaskManager; }
 
@@ -207,6 +208,9 @@ protected:
 	std::condition_variable signal_event;
 	std::atomic_size_t terminate_signals_counter{0};
 	std::atomic_size_t sigint_signals_counter{0};
+
+	std::string config_path;
+	ConfigProcessor::LoadedConfig loaded_config;
 };
 
 

libs/libdaemon/src/BaseDaemon.cpp

@@ -480,8 +480,9 @@ void BaseDaemon::reloadConfiguration()
 	  * При этом, параметры логгирования, заданные в командной строке, не игнорируются.
 	  */
 	std::string log_command_line_option = config().getString("logger.log", "");
-	ConfigurationPtr processed_config = ConfigProcessor(false, true).loadConfig(config().getString("config-file", "config.xml"));
-	config().add(processed_config.duplicate(), PRIO_DEFAULT, false);
+	config_path = config().getString("config-file", "config.xml");
+	loaded_config = ConfigProcessor(false, true).loadConfig(config_path, /* allow_zk_includes = */ true);
+	config().add(loaded_config.configuration.duplicate(), PRIO_DEFAULT, false);
 	log_to_console = !config().getBool("application.runAsDaemon", false) && log_command_line_option.empty();
 }
 

libs/libzkutil/CMakeLists.txt

@@ -9,17 +9,19 @@ add_library(zkutil
 	src/SingleBarrier.cpp
 	src/RWLock.cpp
 	src/ZooKeeperHolder.cpp
+	src/ZooKeeperNodeCache.cpp
 
 	include/zkutil/Increment.h
-    include/zkutil/LeaderElection.h
-    include/zkutil/KeeperException.h
-    include/zkutil/Common.h
-    include/zkutil/Lock.h
-    include/zkutil/SingleBarrier.h
-    include/zkutil/RWLock.h
-    include/zkutil/ZooKeeper.h
-    include/zkutil/Types.h
-    include/zkutil/ZooKeeperHolder.h)
+	include/zkutil/LeaderElection.h
+	include/zkutil/KeeperException.h
+	include/zkutil/Common.h
+	include/zkutil/Lock.h
+	include/zkutil/SingleBarrier.h
+	include/zkutil/RWLock.h
+	include/zkutil/ZooKeeper.h
+	include/zkutil/Types.h
+	include/zkutil/ZooKeeperHolder.h
+	include/zkutil/ZooKeeperNodeCache.h)
 
 find_package (Threads)
 

libs/libzkutil/include/zkutil/Types.h

@@ -123,4 +123,13 @@ namespace CreateMode
 
 using EventPtr = std::shared_ptr<Poco::Event>;
 
+class ZooKeeper;
+
+/// Callback to call when the watch fires.
+/// zookeeper - zookeeper session to which the fired watch belongs
+/// type - event type, one of the *_EVENT constants from zookeeper.h
+/// state - session connection state, one of the *_STATE constants from zookeeper.h
+/// path - znode path to which the change happened. if event == ZOO_SESSION_EVENT it is either NULL or empty string.
+using WatchCallback = std::function<void(ZooKeeper & zookeeper, int type, int state, const char * path)>;
+
 }

libs/libzkutil/include/zkutil/ZooKeeper.h

@@ -31,7 +31,7 @@ const UInt32 DEFAULT_SESSION_TIMEOUT = 30000;
 const UInt32 MEDIUM_SESSION_TIMEOUT = 120000;
 const UInt32 BIG_SESSION_TIMEOUT = 600000;
 
-struct WatchWithEvent;
+struct WatchContext;
 
 
 /** Сессия в ZooKeeper. Интерфейс существенно отличается от обычного API ZooKeeper.
@@ -151,14 +151,17 @@ public:
 	  */
 	int32_t tryRemoveEphemeralNodeWithRetries(const std::string & path, int32_t version = -1, size_t * attempt = nullptr);
 
-	bool exists(const std::string & path, Stat * stat = nullptr, EventPtr watch = nullptr);
+	bool exists(const std::string & path, Stat * stat = nullptr, const EventPtr & watch = nullptr);
+	bool existsWatch(const std::string & path, Stat * stat, const WatchCallback & watch_callback);
 
-	std::string get(const std::string & path, Stat * stat = nullptr, EventPtr watch = nullptr);
+	std::string get(const std::string & path, Stat * stat = nullptr, const EventPtr & watch = nullptr);
 
 	/** Не бросает исключение при следующих ошибках:
 	  *  - Такой ноды нет. В таком случае возвращает false.
 	  */
-	bool tryGet(const std::string & path, std::string & res, Stat * stat = nullptr, EventPtr watch = nullptr, int * code = nullptr);
+	bool tryGet(const std::string & path, std::string & res, Stat * stat = nullptr, const EventPtr & watch = nullptr, int * code = nullptr);
+
+	bool tryGetWatch(const std::string & path, std::string & res, Stat * stat, const WatchCallback & watch_callback, int * code = nullptr);
 
 	void set(const std::string & path, const std::string & data,
 			int32_t version = -1, Stat * stat = nullptr);
@@ -175,14 +178,14 @@ public:
 
 	Strings getChildren(const std::string & path,
 						Stat * stat = nullptr,
-						EventPtr watch = nullptr);
+						const EventPtr & watch = nullptr);
 
 	/** Не бросает исключение при следующих ошибках:
 	  *  - Такой ноды нет.
 	  */
 	int32_t tryGetChildren(const std::string & path, Strings & res,
 						Stat * stat = nullptr,
-						EventPtr watch = nullptr);
+						const EventPtr & watch = nullptr);
 
 	/** Транзакционно выполняет несколько операций. При любой ошибке бросает исключение.
 	  */
@@ -331,15 +334,17 @@ public:
 	zhandle_t * getHandle() { return impl; }
 
 private:
-	friend struct WatchWithEvent;
+	friend struct WatchContext;
 	friend class EphemeralNodeHolder;
 
 	void init(const std::string & hosts, int32_t session_timeout_ms);
 	void removeChildrenRecursive(const std::string & path);
 	void tryRemoveChildrenRecursive(const std::string & path);
-	void * watchForEvent(EventPtr event);
-	watcher_fn callbackForEvent(EventPtr event);
-	static void processEvent(zhandle_t * zh, int type, int state, const char * path, void * watcherCtx);
+
+	static WatchCallback callbackForEvent(const EventPtr & event);
+	WatchContext * createContext(WatchCallback && callback);
+	static void destroyContext(WatchContext * context);
+	static void processCallback(zhandle_t * zh, int type, int state, const char * path, void * watcher_ctx);
 
 	template <class T>
 	int32_t retry(T && operation, size_t * attempt = nullptr)
@@ -367,14 +372,11 @@ private:
 	/// методы не бросают исключений, а возвращают коды ошибок
 	int32_t createImpl(const std::string & path, const std::string & data, int32_t mode, std::string & path_created);
 	int32_t removeImpl(const std::string & path, int32_t version = -1);
-	int32_t getImpl(const std::string & path, std::string & res, Stat * stat = nullptr, EventPtr watch = nullptr);
-	int32_t setImpl(const std::string & path, const std::string & data,
-							int32_t version = -1, Stat * stat = nullptr);
-	int32_t getChildrenImpl(const std::string & path, Strings & res,
-						Stat * stat = nullptr,
-						EventPtr watch = nullptr);
+	int32_t getImpl(const std::string & path, std::string & res, Stat * stat, WatchCallback watch_callback);
+	int32_t setImpl(const std::string & path, const std::string & data, int32_t version = -1, Stat * stat = nullptr);
+	int32_t getChildrenImpl(const std::string & path, Strings & res, Stat * stat, WatchCallback watch_callback);
 	int32_t multiImpl(const Ops & ops, OpResultsPtr * out_results = nullptr);
-	int32_t existsImpl(const std::string & path, Stat * stat_, EventPtr watch = nullptr);
+	int32_t existsImpl(const std::string & path, Stat * stat_, WatchCallback watch_callback);
 
 	std::string hosts;
 	int32_t session_timeout_ms;
@@ -383,7 +385,7 @@ private:
 	ACLPtr default_acl;
 	zhandle_t * impl;
 
-	std::unordered_set<WatchWithEvent *> watch_store;
+	std::unordered_set<WatchContext *> watch_context_store;
 
 	/// Количество попыток повторить операцию чтения при OperationTimeout, ConnectionLoss
 	static constexpr size_t retry_num = 3;

libs/libzkutil/include/zkutil/ZooKeeperNodeCache.h

@@ -0,0 +1,57 @@
+#pragma once
+
+#include <unordered_map>
+#include <unordered_set>
+#include <mutex>
+#include <memory>
+#include <experimental/optional>
+#include <Poco/Event.h>
+#include <zkutil/ZooKeeper.h>
+#include <zkutil/Common.h>
+
+namespace DB
+{
+	namespace ErrorCodes
+	{
+		extern const int NO_ZOOKEEPER;
+	}
+}
+
+namespace zkutil
+{
+
+/// This class allows querying the contents of ZooKeeper nodes and caching the results.
+/// Watches are set for cached nodes and for nodes that were nonexistent at the time of query.
+/// After a watch fires, a notification is generated for the change event.
+/// NOTE: methods of this class are not thread-safe.
+class ZooKeeperNodeCache
+{
+public:
+	ZooKeeperNodeCache(GetZooKeeper get_zookeeper);
+
+	ZooKeeperNodeCache(const ZooKeeperNodeCache &) = delete;
+	ZooKeeperNodeCache(ZooKeeperNodeCache &&) = default;
+
+	std::experimental::optional<std::string> get(const std::string & path);
+
+	Poco::Event & getChangedEvent() { return context->changed_event; }
+
+private:
+	GetZooKeeper get_zookeeper;
+
+	struct Context
+	{
+		Poco::Event changed_event;
+
+		std::mutex mutex;
+		zkutil::ZooKeeperPtr zookeeper;
+		std::unordered_set<std::string> invalidated_paths;
+	};
+
+	std::shared_ptr<Context> context;
+
+	std::unordered_set<std::string> nonexistent_nodes;
+	std::unordered_map<std::string, std::string> node_cache;
+};
+
+}

libs/libzkutil/src/ZooKeeper.cpp

@@ -43,42 +43,30 @@ void check(int32_t code, const std::string path = "")
 	}
 }
 
-struct WatchWithEvent
+struct WatchContext
 {
-	/// существует все время существования WatchWithEvent
+	/// существует все время существования WatchContext
 	ZooKeeper & zk;
-	EventPtr event;
+	WatchCallback callback;
 	CurrentMetrics::Increment metric_increment{CurrentMetrics::ZooKeeperWatch};
 
-	WatchWithEvent(ZooKeeper & zk_, EventPtr event_) : zk(zk_), event(event_) {}
+	WatchContext(ZooKeeper & zk_, WatchCallback callback_) : zk(zk_), callback(std::move(callback_)) {}
 
-	void process(zhandle_t * zh, int32_t event_type, int32_t state, const char * path)
+	void process(int32_t event_type, int32_t state, const char * path)
 	{
-		if (event)
-		{
-			event->set();
-			event = nullptr;
-		}
+		if (callback)
+			callback(zk, event_type, state, path);
 	}
 };
 
-void ZooKeeper::processEvent(zhandle_t * zh, int type, int state, const char * path, void * watcherCtx)
+void ZooKeeper::processCallback(zhandle_t * zh, int type, int state, const char * path, void * watcher_ctx)
 {
-	if (watcherCtx)
-	{
-		WatchWithEvent * watch = static_cast<WatchWithEvent *>(watcherCtx);
-		watch->process(zh, type, state, path);
+	WatchContext * context = static_cast<WatchContext *>(watcher_ctx);
+	context->process(type, state, path);
 
-		/// Гарантируется, что не-ZOO_SESSION_EVENT событие придет ровно один раз (https://issues.apache.org/jira/browse/ZOOKEEPER-890).
-		if (type != ZOO_SESSION_EVENT)
-		{
-			{
-				std::lock_guard<std::mutex> lock(watch->zk.mutex);
-				watch->zk.watch_store.erase(watch);
-			}
-			delete watch;
-		}
-	}
+	/// Гарантируется, что не-ZOO_SESSION_EVENT событие придет ровно один раз (https://issues.apache.org/jira/browse/ZOOKEEPER-890).
+	if (type != ZOO_SESSION_EVENT)
+		destroyContext(context);
 }
 
 void ZooKeeper::init(const std::string & hosts_, int32_t session_timeout_ms_)
@@ -147,40 +135,62 @@ ZooKeeper::ZooKeeper(const Poco::Util::AbstractConfiguration & config, const std
 	init(args.hosts, args.session_timeout_ms);
 }
 
-void * ZooKeeper::watchForEvent(EventPtr event)
+WatchCallback ZooKeeper::callbackForEvent(const EventPtr & event)
 {
+	WatchCallback callback;
 	if (event)
 	{
-		WatchWithEvent * res = new WatchWithEvent(*this, event);
+		callback = [e=event](ZooKeeper &, int, int, const char *) mutable
+		{
+			if (e)
+			{
+				e->set();
+				e.reset(); /// The event is set only once, even if the callback can fire multiple times due to session events.
+			}
+		};
+	}
+	return callback;
+}
+
+WatchContext * ZooKeeper::createContext(WatchCallback && callback)
+{
+	if (callback)
+	{
+		WatchContext * res = new WatchContext(*this, std::move(callback));
 		{
 			std::lock_guard<std::mutex> lock(mutex);
-			watch_store.insert(res);
-			if (watch_store.size() % 10000 == 0)
+			watch_context_store.insert(res);
+			if (watch_context_store.size() % 10000 == 0)
 			{
-				LOG_ERROR(log, "There are " << watch_store.size() << " active watches. There must be a leak somewhere.");
+				LOG_ERROR(log, "There are " << watch_context_store.size() << " active watches. There must be a leak somewhere.");
 			}
 		}
 		return res;
 	}
 	else
-	{
 		return nullptr;
-	}
 }
 
-watcher_fn ZooKeeper::callbackForEvent(EventPtr event)
+void ZooKeeper::destroyContext(WatchContext * context)
 {
-	return event ? processEvent : nullptr;
+	if (context)
+	{
+		std::lock_guard<std::mutex> lock(context->zk.mutex);
+		context->zk.watch_context_store.erase(context);
+	}
+	delete context;
 }
 
 int32_t ZooKeeper::getChildrenImpl(const std::string & path, Strings & res,
 					Stat * stat_,
-					EventPtr watch)
+					WatchCallback watch_callback)
 {
 	String_vector strings;
 	int code;
 	Stat stat;
-	code = zoo_wget_children2(impl, path.c_str(), callbackForEvent(watch), watchForEvent(watch), &strings, &stat);
+	watcher_fn watcher = watch_callback ? processCallback : nullptr;
+	WatchContext * context = createContext(std::move(watch_callback));
+	code = zoo_wget_children2(impl, path.c_str(), watcher, context, &strings, &stat);
 	ProfileEvents::increment(ProfileEvents::ZooKeeperGetChildren);
 	ProfileEvents::increment(ProfileEvents::ZooKeeperTransactions);
 
@@ -193,11 +203,16 @@ int32_t ZooKeeper::getChildrenImpl(const std::string & path, Strings & res,
 			res[i] = std::string(strings.data[i]);
 		deallocate_String_vector(&strings);
 	}
+	else
+	{
+		/// The call was unsuccessful, so the watch was not set. Destroy the context.
+		destroyContext(context);
+	}
 
 	return code;
 }
 Strings ZooKeeper::getChildren(
-	const std::string & path, Stat * stat, EventPtr watch)
+	const std::string & path, Stat * stat, const EventPtr & watch)
 {
 	Strings res;
 	check(tryGetChildren(path, res, stat, watch), path);
@@ -205,9 +220,9 @@ Strings ZooKeeper::getChildren(
 }
 
 int32_t ZooKeeper::tryGetChildren(const std::string & path, Strings & res,
-								Stat * stat_, EventPtr watch)
+								Stat * stat_, const EventPtr & watch)
 {
-	int32_t code = retry(std::bind(&ZooKeeper::getChildrenImpl, this, std::ref(path), std::ref(res), stat_, watch));
+	int32_t code = retry(std::bind(&ZooKeeper::getChildrenImpl, this, std::ref(path), std::ref(res), stat_, callbackForEvent(watch)));
 
 	if (!(	code == ZOK ||
 			code == ZNONODE))
@@ -356,11 +371,13 @@ int32_t ZooKeeper::tryRemoveEphemeralNodeWithRetries(const std::string & path, i
 	}
 }
 
-int32_t ZooKeeper::existsImpl(const std::string & path, Stat * stat_, EventPtr watch)
+int32_t ZooKeeper::existsImpl(const std::string & path, Stat * stat_, WatchCallback watch_callback)
 {
 	int32_t code;
 	Stat stat;
-	code = zoo_wexists(impl, path.c_str(), callbackForEvent(watch), watchForEvent(watch), &stat);
+	watcher_fn watcher = watch_callback ? processCallback : nullptr;
+	WatchContext * context = createContext(std::move(watch_callback));
+	code = zoo_wexists(impl, path.c_str(), watcher, context, &stat);
 	ProfileEvents::increment(ProfileEvents::ZooKeeperExists);
 	ProfileEvents::increment(ProfileEvents::ZooKeeperTransactions);
 
@@ -369,13 +386,18 @@ int32_t ZooKeeper::existsImpl(const std::string & path, Stat * stat_, EventPtr w
 		if (stat_)
 			*stat_ = stat;
 	}
+	if (code != ZOK && code != ZNONODE)
+	{
+		/// The call was unsuccessful, so the watch was not set. Destroy the context.
+		destroyContext(context);
+	}
 
 	return code;
 }
 
-bool ZooKeeper::exists(const std::string & path, Stat * stat_, EventPtr watch)
+bool ZooKeeper::exists(const std::string & path, Stat * stat_, const EventPtr & watch)
 {
-	int32_t code = retry(std::bind(&ZooKeeper::existsImpl, this, path, stat_, watch));
+	int32_t code = retry(std::bind(&ZooKeeper::existsImpl, this, path, stat_, callbackForEvent(watch)));
 
 	if (!(	code == ZOK ||
 			code == ZNONODE))
@@ -385,13 +407,28 @@ bool ZooKeeper::exists(const std::string & path, Stat * stat_, EventPtr watch)
 	return true;
 }
 
-int32_t ZooKeeper::getImpl(const std::string & path, std::string & res, Stat * stat_, EventPtr watch)
+bool ZooKeeper::existsWatch(const std::string & path, Stat * stat_, const WatchCallback & watch_callback)
+{
+	int32_t code = retry(std::bind(&ZooKeeper::existsImpl, this, path, stat_, watch_callback));
+
+	if (!(	code == ZOK ||
+			code == ZNONODE))
+		throw KeeperException(code, path);
+	if (code == ZNONODE)
+		return false;
+	return true;
+}
+
+int32_t ZooKeeper::getImpl(const std::string & path, std::string & res, Stat * stat_, WatchCallback watch_callback)
 {
 	char buffer[MAX_NODE_SIZE];
 	int buffer_len = MAX_NODE_SIZE;
 	int32_t code;
 	Stat stat;
-	code = zoo_wget(impl, path.c_str(), callbackForEvent(watch), watchForEvent(watch), buffer, &buffer_len, &stat);
+	watcher_fn watcher = watch_callback ? processCallback : nullptr;
+	WatchContext * context = createContext(std::move(watch_callback));
+
+	code = zoo_wget(impl, path.c_str(), watcher, context, buffer, &buffer_len, &stat);
 	ProfileEvents::increment(ProfileEvents::ZooKeeperGet);
 	ProfileEvents::increment(ProfileEvents::ZooKeeperTransactions);
 
@@ -405,10 +442,15 @@ int32_t ZooKeeper::getImpl(const std::string & path, std::string & res, Stat * s
 		else
 			res.assign(buffer, buffer_len);
 	}
+	else
+	{
+		/// The call was unsuccessful, so the watch was not set. Destroy the context.
+		destroyContext(context);
+	}
 	return code;
 }
 
-std::string ZooKeeper::get(const std::string & path, Stat * stat, EventPtr watch)
+std::string ZooKeeper::get(const std::string & path, Stat * stat, const EventPtr & watch)
 {
 	int code;
 	std::string res;
@@ -418,9 +460,23 @@ std::string ZooKeeper::get(const std::string & path, Stat * stat, EventPtr watch
 		throw KeeperException("Can't get data for node " + path + ": node doesn't exist", code);
 }
 
-bool ZooKeeper::tryGet(const std::string & path, std::string & res, Stat * stat_, EventPtr watch, int * return_code)
+bool ZooKeeper::tryGet(const std::string & path, std::string & res, Stat * stat_, const EventPtr & watch, int * return_code)
 {
-	int32_t code = retry(std::bind(&ZooKeeper::getImpl, this, std::ref(path), std::ref(res), stat_, watch));
+	int32_t code = retry(std::bind(&ZooKeeper::getImpl, this, std::ref(path), std::ref(res), stat_, callbackForEvent(watch)));
+
+	if (!(code == ZOK ||
+			code == ZNONODE))
+		throw KeeperException(code, path);
+
+	if (return_code)
+		*return_code = code;
+
+	return code == ZOK;
+}
+
+bool ZooKeeper::tryGetWatch(const std::string & path, std::string & res, Stat * stat_, const WatchCallback & watch_callback, int * return_code)
+{
+	int32_t code = retry(std::bind(&ZooKeeper::getImpl, this, std::ref(path), std::ref(res), stat_, watch_callback));
 
 	if (!(code == ZOK ||
 			code == ZNONODE))
@@ -626,11 +682,11 @@ ZooKeeper::~ZooKeeper()
 		LOG_ERROR(&Logger::get("~ZooKeeper"), "Failed to close ZooKeeper session: " << zerror(code));
 	}
 
-	LOG_INFO(&Logger::get("~ZooKeeper"), "Removing " << watch_store.size() << " watches");
+	LOG_INFO(&Logger::get("~ZooKeeper"), "Removing " << watch_context_store.size() << " watches");
 
-	/// удаляем WatchWithEvent которые уже никогда не будут обработаны
-	for (WatchWithEvent * watch : watch_store)
-		delete watch;
+	/// удаляем WatchContext которые уже никогда не будут обработаны
+	for (WatchContext * context : watch_context_store)
+		delete context;
 
 	LOG_INFO(&Logger::get("~ZooKeeper"), "Removed watches");
 }

libs/libzkutil/src/ZooKeeperNodeCache.cpp

@@ -0,0 +1,99 @@
+#include <zkutil/ZooKeeperNodeCache.h>
+
+namespace zkutil
+{
+
+ZooKeeperNodeCache::ZooKeeperNodeCache(GetZooKeeper get_zookeeper_)
+	: get_zookeeper(std::move(get_zookeeper_))
+	, context(std::make_shared<Context>())
+{
+}
+
+std::experimental::optional<std::string> ZooKeeperNodeCache::get(const std::string & path)
+{
+	zkutil::ZooKeeperPtr zookeeper;
+	std::unordered_set<std::string> invalidated_paths;
+	{
+		std::lock_guard<std::mutex> lock(context->mutex);
+
+		if (!context->zookeeper)
+		{
+			/// Possibly, there was a previous session and it has expired. Clear the cache.
+			nonexistent_nodes.clear();
+			node_cache.clear();
+
+			context->zookeeper = get_zookeeper();
+		}
+		zookeeper = context->zookeeper;
+
+		invalidated_paths.swap(context->invalidated_paths);
+	}
+
+	if (!zookeeper)
+		throw DB::Exception("Could not get znode: `" + path + "'. ZooKeeper not configured.", DB::ErrorCodes::NO_ZOOKEEPER);
+
+	for (const auto & path : invalidated_paths)
+	{
+		nonexistent_nodes.erase(path);
+		node_cache.erase(path);
+	}
+
+	if (nonexistent_nodes.count(path))
+		return std::experimental::nullopt;
+
+	auto watch_callback = [context=context](zkutil::ZooKeeper & zookeeper, int type, int state, const char * path)
+	{
+		if (!(type != ZOO_SESSION_EVENT || state == ZOO_EXPIRED_SESSION_STATE))
+			return;
+
+		bool changed = false;
+		{
+			std::lock_guard<std::mutex> lock(context->mutex);
+			if (&zookeeper != context->zookeeper.get())
+				return;
+
+			if (type != ZOO_SESSION_EVENT)
+				changed = context->invalidated_paths.emplace(path).second;
+			else if (state == ZOO_EXPIRED_SESSION_STATE)
+			{
+				context->zookeeper = nullptr;
+				context->invalidated_paths.clear();
+				changed = true;
+			}
+		}
+		if (changed)
+			context->changed_event.set();
+	};
+
+	std::string contents;
+
+	auto cache_it = node_cache.find(path);
+	if (cache_it != node_cache.end())
+	{
+		return cache_it->second;
+	}
+
+	if (zookeeper->tryGetWatch(path, contents, /* stat = */nullptr, watch_callback))
+	{
+		node_cache.emplace(path, contents);
+		return contents;
+	}
+
+	/// Node doesn't exist. Create a watch on node creation.
+	nonexistent_nodes.insert(path);
+
+	if (!zookeeper->existsWatch(path, /* stat = */nullptr, watch_callback))
+		return std::experimental::nullopt;
+
+	/// Node was created between the two previous calls, try again. Watch is already set.
+	if (zookeeper->tryGet(path, contents))
+	{
+		nonexistent_nodes.erase(path);
+		node_cache.emplace(path, contents);
+		return contents;
+	}
+
+	return std::experimental::nullopt;
+}
+
+}

libs/libzkutil/src/tests/zk_many_watches_reconnect.cpp

@@ -24,7 +24,7 @@ int main(int argc, char ** argv)
 		}
 
 		ConfigProcessor processor(false, true);
-		auto config = processor.loadConfig(argv[1]);
+		auto config = processor.loadConfig(argv[1]).configuration;
 		zkutil::ZooKeeper zk(*config, "zookeeper");
 		zkutil::EventPtr watch = std::make_shared<Poco::Event>();
 

utils/config-processor/CMakeLists.txt

@@ -1,4 +1,4 @@
 add_executable (config-processor config-processor.cpp)
-target_link_libraries (config-processor daemon)
+target_link_libraries (config-processor dbms zkutil dbms)
 
 INSTALL(TARGETS config-processor RUNTIME DESTINATION bin COMPONENT config-processor)