Implement support for agent-based checks.

Refs #4865

components/CMakeLists.txt

@@ -1,3 +1,4 @@
+add_subdirectory(agent)
 add_subdirectory(checker)
 add_subdirectory(cluster)
 add_subdirectory(compat)

components/agent/CMakeLists.txt

@@ -0,0 +1,36 @@
+# Icinga 2
+# Copyright (C) 2012-2014 Icinga Development Team (http://www.icinga.org)
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software Foundation
+# Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.
+
+mkclass_target(agentlistener.ti agentlistener.th)
+
+mkembedconfig_target(agent-type.conf agent-type.cpp)
+
+add_library(agent SHARED
+  agentchecktask.cpp agentlistener.cpp agentlistener.th
+  agent-type.cpp
+)
+
+target_link_libraries(agent ${Boost_LIBRARIES} base config icinga remote)
+
+set_target_properties (
+  agent PROPERTIES
+  INSTALL_RPATH ${CMAKE_INSTALL_FULL_LIBDIR}/icinga2
+  FOLDER Components
+)
+
+install(TARGETS agent RUNTIME DESTINATION ${CMAKE_INSTALL_SBINDIR} LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}/icinga2)
+

components/agent/agent-type.conf

@@ -0,0 +1,39 @@
+/******************************************************************************
+ * Icinga 2                                                                   *
+ * Copyright (C) 2012-2014 Icinga Development Team (http://www.icinga.org)    *
+ *                                                                            *
+ * This program is free software; you can redistribute it and/or              *
+ * modify it under the terms of the GNU General Public License                *
+ * as published by the Free Software Foundation; either version 2             *
+ * of the License, or (at your option) any later version.                     *
+ *                                                                            *
+ * This program is distributed in the hope that it will be useful,            *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
+ * GNU General Public License for more details.                               *
+ *                                                                            *
+ * You should have received a copy of the GNU General Public License          *
+ * along with this program; if not, write to the Free Software Foundation     *
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
+ ******************************************************************************/
+
+%type AgentListener {
+	%attribute %string "cert_path",
+	%require "cert_path",
+
+	%attribute %string "key_path",
+	%require "key_path",
+
+	%attribute %string "ca_path",
+	%require "ca_path",
+
+	%attribute %string "crl_path",
+
+	%attribute %string "bind_host",
+	%attribute %string "bind_port",
+	
+	%attribute %string "upstream_name",
+	%attribute %string "upstream_host",
+	%attribute %string "upstream_port",
+	%attribute %number "upstream_interval"
+}

components/agent/agentchecktask.cpp

@@ -0,0 +1,107 @@
+/******************************************************************************
+ * Icinga 2                                                                   *
+ * Copyright (C) 2012-2014 Icinga Development Team (http://www.icinga.org)    *
+ *                                                                            *
+ * This program is free software; you can redistribute it and/or              *
+ * modify it under the terms of the GNU General Public License                *
+ * as published by the Free Software Foundation; either version 2             *
+ * of the License, or (at your option) any later version.                     *
+ *                                                                            *
+ * This program is distributed in the hope that it will be useful,            *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
+ * GNU General Public License for more details.                               *
+ *                                                                            *
+ * You should have received a copy of the GNU General Public License          *
+ * along with this program; if not, write to the Free Software Foundation     *
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
+ ******************************************************************************/
+
+#include "agent/agentchecktask.h"
+#include "agent/agentlistener.h"
+#include "icinga/service.h"
+#include "icinga/checkcommand.h"
+#include "icinga/macroprocessor.h"
+#include "icinga/icingaapplication.h"
+#include "base/application.h"
+#include "base/objectlock.h"
+#include "base/convert.h"
+#include "base/utility.h"
+#include "base/initialize.h"
+#include "base/scriptfunction.h"
+#include "base/dynamictype.h"
+
+using namespace icinga;
+
+boost::mutex l_Mutex;
+std::map<Checkable::Ptr, double> l_PendingChecks;
+Timer::Ptr l_AgentTimer;
+
+INITIALIZE_ONCE(&AgentCheckTask::StaticInitialize);
+REGISTER_SCRIPTFUNCTION(AgentCheck, &AgentCheckTask::ScriptFunc);
+
+void AgentCheckTask::StaticInitialize(void)
+{
+	l_AgentTimer = make_shared<Timer>();
+	l_AgentTimer->OnTimerExpired.connect(boost::bind(&AgentCheckTask::AgentTimerHandler));
+	l_AgentTimer->SetInterval(60);
+	l_AgentTimer->Start();
+}
+
+void AgentCheckTask::AgentTimerHandler(void)
+{
+	boost::mutex::scoped_lock lock(l_Mutex);
+
+	std::map<Checkable::Ptr, double> newmap;
+	std::pair<Checkable::Ptr, double> kv;
+	
+	double now = Utility::GetTime();
+
+	BOOST_FOREACH(kv, l_PendingChecks) {
+		if (kv.second < now - 60 && kv.first->IsCheckPending()) {
+			CheckResult::Ptr cr = make_shared<CheckResult>();
+			cr->SetOutput("Agent isn't responding.");
+			cr->SetState(ServiceCritical);
+			kv.first->ProcessCheckResult(cr);
+		} else {
+			newmap.insert(kv);
+		}
+	}
+	
+	l_PendingChecks.swap(newmap);
+}
+
+void AgentCheckTask::ScriptFunc(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
+{
+	Host::Ptr host;
+	Service::Ptr service;
+	tie(host, service) = GetHostService(checkable);
+
+	MacroProcessor::ResolverList resolvers;
+	if (service)
+		resolvers.push_back(std::make_pair("service", service));
+	resolvers.push_back(std::make_pair("host", host));
+	resolvers.push_back(std::make_pair("command", checkable->GetCheckCommand()));
+	resolvers.push_back(std::make_pair("icinga", IcingaApplication::GetInstance()));
+
+	String agent_host = MacroProcessor::ResolveMacros("$agent_host$", resolvers, checkable->GetLastCheckResult());
+	String agent_port = MacroProcessor::ResolveMacros("$agent_port$", resolvers, checkable->GetLastCheckResult());
+
+	if (agent_host.IsEmpty() || agent_port.IsEmpty()) {
+		Log(LogWarning, "agent", "'agent_host' and 'agent_port' must be set for agent checks.");
+		return;
+	}
+	
+	std::pair<String, String> key = std::make_pair(agent_host, agent_port);
+	
+	double now = Utility::GetTime();
+	
+	{
+		boost::mutex::scoped_lock lock(l_Mutex);
+		l_PendingChecks[checkable] = now;
+	}
+	
+	BOOST_FOREACH(const AgentListener::Ptr& al, DynamicType::GetObjects<AgentListener>()) {
+		al->AddConnection(agent_host, agent_port);
+	}
+}

components/agent/agentchecktask.h

@@ -0,0 +1,48 @@
+/******************************************************************************
+ * Icinga 2                                                                   *
+ * Copyright (C) 2012-2014 Icinga Development Team (http://www.icinga.org)    *
+ *                                                                            *
+ * This program is free software; you can redistribute it and/or              *
+ * modify it under the terms of the GNU General Public License                *
+ * as published by the Free Software Foundation; either version 2             *
+ * of the License, or (at your option) any later version.                     *
+ *                                                                            *
+ * This program is distributed in the hope that it will be useful,            *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
+ * GNU General Public License for more details.                               *
+ *                                                                            *
+ * You should have received a copy of the GNU General Public License          *
+ * along with this program; if not, write to the Free Software Foundation     *
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
+ ******************************************************************************/
+
+#ifndef AGENTCHECKTASK_H
+#define AGENTCHECKTASK_H
+
+#include "icinga/service.h"
+#include "base/timer.h"
+
+namespace icinga
+{
+
+/**
+ * Agent check type.
+ *
+ * @ingroup methods
+ */
+class AgentCheckTask
+{
+public:
+	static void StaticInitialize(void);
+	static void ScriptFunc(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
+
+private:
+	AgentCheckTask(void);
+	
+	static void AgentTimerHandler(void);
+};
+
+}
+
+#endif /* AGENTCHECKTASK_H */

components/agent/agentlistener.cpp

@@ -0,0 +1,260 @@
+/******************************************************************************
+ * Icinga 2                                                                   *
+ * Copyright (C) 2012-2014 Icinga Development Team (http://www.icinga.org)    *
+ *                                                                            *
+ * This program is free software; you can redistribute it and/or              *
+ * modify it under the terms of the GNU General Public License                *
+ * as published by the Free Software Foundation; either version 2             *
+ * of the License, or (at your option) any later version.                     *
+ *                                                                            *
+ * This program is distributed in the hope that it will be useful,            *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
+ * GNU General Public License for more details.                               *
+ *                                                                            *
+ * You should have received a copy of the GNU General Public License          *
+ * along with this program; if not, write to the Free Software Foundation     *
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
+ ******************************************************************************/
+
+#include "agent/agentlistener.h"
+#include "remote/jsonrpc.h"
+#include "icinga/icingaapplication.h"
+#include "base/netstring.h"
+#include "base/dynamictype.h"
+#include "base/logger_fwd.h"
+#include "base/objectlock.h"
+#include "base/networkstream.h"
+#include "base/application.h"
+#include "base/context.h"
+
+using namespace icinga;
+
+REGISTER_TYPE(AgentListener);
+
+/**
+ * Starts the component.
+ */
+void AgentListener::Start(void)
+{
+	DynamicObject::Start();
+
+	/* set up SSL context */
+	shared_ptr<X509> cert = GetX509Certificate(GetCertPath());
+	SetIdentity(GetCertificateCN(cert));
+	Log(LogInformation, "agent", "My identity: " + GetIdentity());
+
+	m_SSLContext = MakeSSLContext(GetCertPath(), GetKeyPath(), GetCaPath());
+
+	if (!GetCrlPath().IsEmpty())
+		AddCRLToSSLContext(m_SSLContext, GetCrlPath());
+
+	/* create the primary JSON-RPC listener */
+	if (!GetBindPort().IsEmpty())
+		AddListener(GetBindPort());
+		
+	m_AgentTimer = make_shared<Timer>();
+	m_AgentTimer->OnTimerExpired.connect(boost::bind(&AgentListener::AgentTimerHandler, this));
+	m_AgentTimer->SetInterval(GetUpstreamInterval());
+	m_AgentTimer->Start();
+}
+
+shared_ptr<SSL_CTX> AgentListener::GetSSLContext(void) const
+{
+	return m_SSLContext;
+}
+
+/**
+ * Creates a new JSON-RPC listener on the specified port.
+ *
+ * @param service The port to listen on.
+ */
+void AgentListener::AddListener(const String& service)
+{
+	ObjectLock olock(this);
+
+	shared_ptr<SSL_CTX> sslContext = m_SSLContext;
+
+	if (!sslContext)
+		BOOST_THROW_EXCEPTION(std::logic_error("SSL context is required for AddListener()"));
+
+	std::ostringstream s;
+	s << "Adding new listener: port " << service;
+	Log(LogInformation, "agent", s.str());
+
+	TcpSocket::Ptr server = make_shared<TcpSocket>();
+	server->Bind(service, AF_INET6);
+
+	boost::thread thread(boost::bind(&AgentListener::ListenerThreadProc, this, server));
+	thread.detach();
+
+	m_Servers.insert(server);
+}
+
+void AgentListener::ListenerThreadProc(const Socket::Ptr& server)
+{
+	Utility::SetThreadName("Cluster Listener");
+
+	server->Listen();
+
+	for (;;) {
+		Socket::Ptr client = server->Accept();
+
+		Utility::QueueAsyncCallback(boost::bind(&AgentListener::NewClientHandler, this, client, TlsRoleServer));
+	}
+}
+
+/**
+ * Creates a new JSON-RPC client and connects to the specified host and port.
+ *
+ * @param node The remote host.
+ * @param service The remote port.
+ */
+void AgentListener::AddConnection(const String& node, const String& service) {
+	{
+		ObjectLock olock(this);
+
+		shared_ptr<SSL_CTX> sslContext = m_SSLContext;
+
+		if (!sslContext)
+			BOOST_THROW_EXCEPTION(std::logic_error("SSL context is required for AddConnection()"));
+	}
+
+	TcpSocket::Ptr client = make_shared<TcpSocket>();
+
+	client->Connect(node, service);
+	Utility::QueueAsyncCallback(boost::bind(&AgentListener::NewClientHandler, this, client, TlsRoleClient));
+}
+
+/**
+ * Processes a new client connection.
+ *
+ * @param client The new client.
+ */
+void AgentListener::NewClientHandler(const Socket::Ptr& client, TlsRole role)
+{
+	CONTEXT("Handling new agent client connection");
+
+	NetworkStream::Ptr netStream = make_shared<NetworkStream>(client);
+
+	TlsStream::Ptr tlsStream = make_shared<TlsStream>(netStream, role, m_SSLContext);
+	tlsStream->Handshake();
+
+	shared_ptr<X509> cert = tlsStream->GetPeerCertificate();
+	String identity = GetCertificateCN(cert);
+
+	Log(LogInformation, "agent", "New client connection for identity '" + identity + "'");
+
+	if (identity != GetUpstreamName()) {
+		Dictionary::Ptr request = make_shared<Dictionary>();
+		request->Set("method", "get_crs");
+		JsonRpc::SendMessage(tlsStream, request);
+	}
+
+	Dictionary::Ptr message;
+
+	try {
+		message = JsonRpc::ReadMessage(tlsStream);
+	} catch (const std::exception& ex) {
+		Log(LogWarning, "agent", "Error while reading JSON-RPC message for agent '" + identity + "': " + DiagnosticInformation(ex));
+
+		return;
+	}
+
+	MessageHandler(tlsStream, identity, message);
+	
+	tlsStream->Close();
+}
+
+void AgentListener::MessageHandler(const TlsStream::Ptr& sender, const String& identity, const Dictionary::Ptr& message)
+{
+	CONTEXT("Processing agent message of type '" + message->Get("method") + "'");
+	
+	String method = message->Get("method");
+	
+	if (identity == GetUpstreamName()) {
+		if (method == "get_crs") {
+			Dictionary::Ptr services = make_shared<Dictionary>();
+
+			Host::Ptr host = Host::GetByName("localhost");
+			
+			if (!host)
+				Log(LogWarning, "agent", "Agent doesn't have any services for 'localhost'.");
+			else {
+				BOOST_FOREACH(const Service::Ptr& service, host->GetServices()) {
+					services->Set(service->GetShortName(), Serialize(service->GetLastCheckResult()));
+				}
+			}
+			
+			Dictionary::Ptr params = make_shared<Dictionary>();
+			params->Set("services", services);
+			params->Set("host", Serialize(host->GetLastCheckResult()));
+			
+			Dictionary::Ptr request = make_shared<Dictionary>();
+			request->Set("method", "push_crs");
+			request->Set("params", params);
+			
+			JsonRpc::SendMessage(sender, request);
+		}
+	}
+	
+	if (method == "push_crs") {
+		Host::Ptr host = Host::GetByName(identity);
+		
+		if (!host) {
+			Log(LogWarning, "agent", "Ignoring check results for host '" + identity + "'.");
+			return;
+		}
+		
+		Dictionary::Ptr params = message->Get("params");
+		
+		if (!params)
+			return;
+
+		Value hostcr = Deserialize(params->Get("host"), true);
+		
+		if (!hostcr.IsObjectType<CheckResult>()) {
+			Log(LogWarning, "agent", "Ignoring invalid check result for host '" + identity + "'.");
+		} else {
+			CheckResult::Ptr cr = hostcr;
+			host->ProcessCheckResult(cr);
+		}
+
+		Dictionary::Ptr services = params->Get("services");
+		
+		if (!services)
+			return;
+		
+		Dictionary::Pair kv;
+		
+		BOOST_FOREACH(kv, services) {
+			Service::Ptr service = host->GetServiceByShortName(kv.first);
+			
+			if (!service) {
+				Log(LogWarning, "agent", "Ignoring check result for service '" + kv.first + "' on host '" + identity + "'.");
+				continue;
+			}
+			
+			Value servicecr = Deserialize(kv.second, true);
+			
+			if (!servicecr.IsObjectType<CheckResult>()) {
+				Log(LogWarning, "agent", "Ignoring invalid check result for service '" + kv.first + "' on host '" + identity + "'.");
+				continue;
+			}
+			
+			CheckResult::Ptr cr = servicecr;
+			service->ProcessCheckResult(cr);
+		}
+	}
+}
+
+void AgentListener::AgentTimerHandler(void)
+{
+	String host = GetUpstreamHost();
+	String port = GetUpstreamPort();
+
+	if (host.IsEmpty() || port.IsEmpty())
+		return;
+	
+	AddConnection(host, port);
+}

components/agent/agentlistener.h

@@ -0,0 +1,70 @@
+/******************************************************************************
+ * Icinga 2                                                                   *
+ * Copyright (C) 2012-2014 Icinga Development Team (http://www.icinga.org)    *
+ *                                                                            *
+ * This program is free software; you can redistribute it and/or              *
+ * modify it under the terms of the GNU General Public License                *
+ * as published by the Free Software Foundation; either version 2             *
+ * of the License, or (at your option) any later version.                     *
+ *                                                                            *
+ * This program is distributed in the hope that it will be useful,            *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
+ * GNU General Public License for more details.                               *
+ *                                                                            *
+ * You should have received a copy of the GNU General Public License          *
+ * along with this program; if not, write to the Free Software Foundation     *
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
+ ******************************************************************************/
+
+#ifndef AGENTLISTENER_H
+#define AGENTLISTENER_H
+
+#include "agent/agentlistener.th"
+#include "base/dynamicobject.h"
+#include "base/timer.h"
+#include "base/array.h"
+#include "base/tcpsocket.h"
+#include "base/tlsstream.h"
+#include "base/utility.h"
+#include "base/tlsutility.h"
+#include "icinga/service.h"
+
+namespace icinga
+{
+
+/**
+ * @ingroup agent
+ */
+class AgentListener : public ObjectImpl<AgentListener>
+{
+public:
+	DECLARE_PTR_TYPEDEFS(AgentListener);
+	DECLARE_TYPENAME(AgentListener);
+
+	virtual void Start(void);
+
+	shared_ptr<SSL_CTX> GetSSLContext(void) const;
+
+private:
+	shared_ptr<SSL_CTX> m_SSLContext;
+	std::set<TcpSocket::Ptr> m_Servers;
+	Timer::Ptr m_Timer;
+
+	Timer::Ptr m_AgentTimer;
+	void AgentTimerHandler(void);
+
+	void AddListener(const String& service);
+	void AddConnection(const String& node, const String& service);
+
+	void NewClientHandler(const Socket::Ptr& client, TlsRole role);
+	void ListenerThreadProc(const Socket::Ptr& server);
+
+	void MessageHandler(const TlsStream::Ptr& sender, const String& identity, const Dictionary::Ptr& message);
+	
+	friend class AgentCheckTask;
+};
+
+}
+
+#endif /* AGENTLISTENER_H */

components/agent/agentlistener.ti

@@ -0,0 +1,24 @@
+#include "base/dynamicobject.h"
+#include "base/application.h"
+
+namespace icinga
+{
+
+class AgentListener : DynamicObject
+{
+	[config] String cert_path;
+	[config] String key_path;
+	[config] String ca_path;
+	[config] String crl_path;
+	[config] String bind_host;
+	[config] String bind_port;
+	[config] String upstream_host;
+	[config] String upstream_port;
+	[config] String upstream_name;
+	[config] int upstream_interval {
+		default {{{ return 60; }}}
+	};
+	String identity;
+};
+
+}

itl/command-common.conf

@@ -284,3 +284,8 @@ object CheckCommand "snmp-extend"{
 
 	vars.community = "public"
 }
+
+object CheckCommand "agent" {
+	import "agent-check-command"
+}
+

itl/command.conf

@@ -31,6 +31,10 @@ template CheckCommand "plugin-check-command" {
 	methods.execute = "PluginCheck"
 }
 
+template CheckCommand "agent-check-command" {
+	methods.execute = "AgentCheck"
+}
+
 template NotificationCommand "plugin-notification-command" {
 	methods.execute = "PluginNotification"
 }

lib/icinga/checkable-check.cpp

@@ -429,6 +429,12 @@ void Checkable::ProcessCheckResult(const CheckResult::Ptr& cr, const String& aut
 		OnNotificationsRequested(GetSelf(), recovery ? NotificationRecovery : NotificationProblem, cr, "", "");
 }
 
+bool Checkable::IsCheckPending(void) const
+{
+	ObjectLock olock(this);
+	return m_CheckRunning;
+}
+
 void Checkable::ExecuteCheck(void)
 {
 	CONTEXT("Executing check for object '" + GetName() + "'");

lib/icinga/checkable.h

@@ -157,6 +157,8 @@ public:
 	int GetModifiedAttributes(void) const;
 	void SetModifiedAttributes(int flags);
 
+	bool IsCheckPending(void) const;
+
 	static double CalculateExecutionTime(const CheckResult::Ptr& cr);
 	static double CalculateLatency(const CheckResult::Ptr& cr);