Implement support for hosts in the agent component.

Refs #4865

components/agent/agentchecktask.cpp

@@ -84,24 +84,41 @@ void AgentCheckTask::ScriptFunc(const Checkable::Ptr& checkable, const CheckResu
 	resolvers.push_back(std::make_pair("command", checkable->GetCheckCommand()));
 	resolvers.push_back(std::make_pair("icinga", IcingaApplication::GetInstance()));
 
+	String agent_identity = MacroProcessor::ResolveMacros("$agent_identity$", resolvers, checkable->GetLastCheckResult());
 	String agent_host = MacroProcessor::ResolveMacros("$agent_host$", resolvers, checkable->GetLastCheckResult());
-	String agent_port = MacroProcessor::ResolveMacros("$agent_port$", resolvers, checkable->GetLastCheckResult());
+	String agent_service = MacroProcessor::ResolveMacros("$agent_service$", resolvers, checkable->GetLastCheckResult());
 
-	if (agent_host.IsEmpty() || agent_port.IsEmpty()) {
+	if (agent_identity.IsEmpty() || agent_host.IsEmpty()) {
-		Log(LogWarning, "agent", "'agent_host' and 'agent_port' must be set for agent checks.");
+		Log(LogWarning, "agent", "'agent_name' and 'agent_host' must be set for agent checks.");
 		return;
 	}
 
-	std::pair<String, String> key = std::make_pair(agent_host, agent_port);
+	String agent_peer_host = MacroProcessor::ResolveMacros("$agent_peer_host$", resolvers, checkable->GetLastCheckResult());
+	String agent_peer_port = MacroProcessor::ResolveMacros("$agent_peer_port$", resolvers, checkable->GetLastCheckResult());
 	
 	double now = Utility::GetTime();
 	
+	BOOST_FOREACH(const AgentListener::Ptr& al, DynamicType::GetObjects<AgentListener>()) {
+		double seen = al->GetAgentSeen(agent_identity);
+
+		if (seen < now - 300)
+			continue;
+
+		CheckResult::Ptr cr = al->GetCheckResult(agent_identity, agent_host, agent_service);
+
+		if (cr) {
+			checkable->ProcessCheckResult(cr);
+			return;
+		}
+	}
+	
 	{
 		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);
+		if (!agent_peer_host.IsEmpty() && !agent_peer_port.IsEmpty())
+			al->AddConnection(agent_peer_host, agent_peer_port);
 	}
 }

components/agent/agentlistener.cpp

@@ -40,6 +40,8 @@ void AgentListener::Start(void)
 {
 	DynamicObject::Start();
 
+	m_Results = make_shared<Dictionary>();
+
 	/* set up SSL context */
 	shared_ptr<X509> cert = GetX509Certificate(GetCertPath());
 	SetIdentity(GetCertificateCN(cert));
@@ -58,6 +60,7 @@ void AgentListener::Start(void)
 	m_AgentTimer->OnTimerExpired.connect(boost::bind(&AgentListener::AgentTimerHandler, this));
 	m_AgentTimer->SetInterval(GetUpstreamInterval());
 	m_AgentTimer->Start();
+	m_AgentTimer->Reschedule(0);
 }
 
 shared_ptr<SSL_CTX> AgentListener::GetSSLContext(void) const
@@ -181,21 +184,28 @@ void AgentListener::MessageHandler(const TlsStream::Ptr& sender, const String& i
 
 	if (identity == GetUpstreamName()) {
 		if (method == "get_crs") {
+			Dictionary::Ptr hosts = make_shared<Dictionary>();
+
+			BOOST_FOREACH(const Host::Ptr& host, DynamicType::GetObjects<Host>()) {
+				Dictionary::Ptr hostInfo = make_shared<Dictionary>();
+
+				hostInfo->Set("cr", Serialize(host->GetLastCheckResult()));
+
 				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 serviceInfo = make_shared<Dictionary>();
+					serviceInfo->Set("cr", Serialize(service->GetLastCheckResult()));
+					services->Set(service->GetShortName(), serviceInfo);
 				}
+
+				hostInfo->Set("services", services);
+
+				hosts->Set(host->GetName(), hostInfo);
 			}
 
 			Dictionary::Ptr params = make_shared<Dictionary>();
-			params->Set("services", services);
+			params->Set("hosts", hosts);
-			params->Set("host", Serialize(host->GetLastCheckResult()));
 
 			Dictionary::Ptr request = make_shared<Dictionary>();
 			request->Set("method", "push_crs");
@@ -206,14 +216,18 @@ void AgentListener::MessageHandler(const TlsStream::Ptr& sender, const String& i
 	}
 
 	if (method == "push_crs") {
-		Dictionary::Ptr params = message->Get("params");
+		Value paramsv = message->Get("params");
 
-		if (!params)
+		if (paramsv.IsEmpty() || !paramsv.IsObjectType<Dictionary>())
 			return;
 
+		Dictionary::Ptr params = paramsv;
+
+		params->Set("seen", Utility::GetTime());
+
 		Dictionary::Ptr inventoryDescr = make_shared<Dictionary>();
 		inventoryDescr->Set("identity", identity);
-		inventoryDescr->Set("crs", params);
+		inventoryDescr->Set("params", params);
 
 		String inventoryFile = GetInventoryDir() + SHA256(identity);
 		String inventoryTempFile = inventoryFile + ".tmp";
@@ -233,47 +247,69 @@ void AgentListener::MessageHandler(const TlsStream::Ptr& sender, const String& i
 			    << boost::errinfo_file_name(inventoryTempFile));
 		}
 
-		Host::Ptr host = Host::GetByName(identity);
+		m_Results->Set(identity, params);
-
+	}
-		if (!host) {
-			Log(LogWarning, "agent", "Ignoring check results for host '" + identity + "'.");
-			return;
 }
 
-		Value hostcr = Deserialize(params->Get("host"), true);
+double  AgentListener::GetAgentSeen(const String& agentIdentity)
+{
+	Dictionary::Ptr agentparams = m_Results->Get(agentIdentity);
 
-		if (!hostcr.IsObjectType<CheckResult>()) {
+	if (!agentparams)
-			Log(LogWarning, "agent", "Ignoring invalid check result for host '" + identity + "'.");
+		return 0;
+
+	return agentparams->Get("seen");
+}
+
+CheckResult::Ptr AgentListener::GetCheckResult(const String& agentIdentity, const String& hostName, const String& serviceName)
+{
+	Dictionary::Ptr agentparams = m_Results->Get(agentIdentity);
+
+	if (!agentparams)
+		return CheckResult::Ptr();
+
+	Value hostsv = agentparams->Get("hosts");
+
+	if (hostsv.IsEmpty() || !hostsv.IsObjectType<Dictionary>())
+		return CheckResult::Ptr();
+
+	Dictionary::Ptr hosts = hostsv;
+
+	Value hostv = hosts->Get(hostName);
+
+	if (hostv.IsEmpty() || !hostv.IsObjectType<Dictionary>())
+		return CheckResult::Ptr();
+
+	Dictionary::Ptr host = hostv;
+
+	if (serviceName.IsEmpty()) {
+		Value hostcrv = Deserialize(host->Get("cr"));
+
+		if (hostcrv.IsEmpty() || !hostcrv.IsObjectType<CheckResult>())
+			return CheckResult::Ptr();
+
+		return hostcrv;
 	} else {
-			CheckResult::Ptr cr = hostcr;
+		Value servicesv = host->Get("services");
-			host->ProcessCheckResult(cr);
-		}
 
-		Dictionary::Ptr services = params->Get("services");
+		if (servicesv.IsEmpty() || !servicesv.IsObjectType<Dictionary>())
+			return CheckResult::Ptr();
 
-		if (!services)
+		Dictionary::Ptr services = servicesv;
-			return;
 
-		Dictionary::Pair kv;
+		Value servicev = services->Get(serviceName);
 
-		BOOST_FOREACH(kv, services) {
+		if (servicev.IsEmpty() || !servicev.IsObjectType<Dictionary>())
-			Service::Ptr service = host->GetServiceByShortName(kv.first);
+			return CheckResult::Ptr();
 
-			if (!service) {
+		Dictionary::Ptr service = servicev;
-				Log(LogWarning, "agent", "Ignoring check result for service '" + kv.first + "' on host '" + identity + "'.");
-				continue;
-			}
 
-			Value servicecr = Deserialize(kv.second, true);
+		Value servicecrv = Deserialize(service->Get("cr"));
 
-			if (!servicecr.IsObjectType<CheckResult>()) {
+		if (servicecrv.IsEmpty() || !servicecrv.IsObjectType<CheckResult>())
-				Log(LogWarning, "agent", "Ignoring invalid check result for service '" + kv.first + "' on host '" + identity + "'.");
+			return CheckResult::Ptr();
-				continue;
-			}
 
-			CheckResult::Ptr cr = servicecr;
+		return servicecrv;
-			service->ProcessCheckResult(cr);
-		}
 	}
 }
 

components/agent/agentlistener.h

@@ -46,11 +46,16 @@ public:
 
 	shared_ptr<SSL_CTX> GetSSLContext(void) const;
 
+	double GetAgentSeen(const String& agentIdentity);
+	CheckResult::Ptr GetCheckResult(const String& agentIdentity, const String& hostName, const String& serviceName);
+
 private:
 	shared_ptr<SSL_CTX> m_SSLContext;
 	std::set<TcpSocket::Ptr> m_Servers;
 	Timer::Ptr m_Timer;
 
+	Dictionary::Ptr m_Results;
+
 	Timer::Ptr m_AgentTimer;
 	void AgentTimerHandler(void);
 

contrib/make-agent-config.py

@@ -20,12 +20,15 @@
 #
 # template Host "agent-host" {
 #   check_command = "agent"
-#   vars.agent_host = "$address$"
+#   vars.agent_host = "$host.name$"
-#   vars.agent_port = 7000
+#   vars.agent_service = ""
+#   vars.agent_peer_host = "$address$"
+#   vars.agent_peer_port = 7000
 # }
 #
 # template Service "agent-service" {
 #   check_command = "agent"
+#   vars.agent_service = "$service.name$"
 #}
 
 import subprocess, json
@@ -33,18 +36,28 @@ import subprocess, json
 inventory_json = subprocess.check_output(["icinga2-list-agents", "--batch"])
 inventory = json.loads(inventory_json)
 
-for host, hostinfo in inventory.items():
+for agent, agent_info in inventory.items():
-    print "object Host \"%s\" {" % (host)
+    for host, host_info in agent_info["hosts"].items():
-    print "  import \"agent-host\""
+        if host == "localhost":
+            host_name = agent
+        else:
+            host_name = host
 
-    if "peer" in hostinfo:
+        print "object Host \"%s\" {" % (host_name)
-        print "  vars.agent_host = \"%s\"" % (hostinfo["peer"]["agent_host"])
+        print "  import \"agent-host\""
-        print "  vars.agent_port = \"%s\"" % (hostinfo["peer"]["agent_port"])
+        print "  vars.agent_identity = \"%s\"" % (agent)
+
+        if host != host_name:
+            print "  vars.agent_host = \"%s\"" % (host)
+
+        if "peer" in agent_info:
+            print "  vars.agent_peer_host = \"%s\"" % (agent_info["peer"]["agent_host"])
+            print "  vars.agent_peer_port = \"%s\"" % (agent_info["peer"]["agent_port"])
 
         print "}"
         print ""
 
-    for service in hostinfo["services"]:
+        for service in host_info["services"]:
             print "object Service \"%s\" {" % (service)
             print "  import \"agent-service\""
             print "  host_name = \"%s\"" % (host)

lib/icinga/host.cpp

@@ -147,29 +147,21 @@ HostState Host::CalculateState(ServiceState state)
 
 HostState Host::GetState(void) const
 {
-	ASSERT(!OwnsLock());
-
 	return CalculateState(GetStateRaw());
 }
 
 HostState Host::GetLastState(void) const
 {
-	ASSERT(!OwnsLock());
-
 	return CalculateState(GetLastStateRaw());
 }
 
 HostState Host::GetLastHardState(void) const
 {
-	ASSERT(!OwnsLock());
-
 	return CalculateState(GetLastHardStateRaw());
 }
 
 double Host::GetLastStateUp(void) const
 {
-	ASSERT(!OwnsLock());
-
 	if (GetLastStateOK() > GetLastStateWarning())
 		return GetLastStateOK();
 	else
@@ -178,8 +170,6 @@ double Host::GetLastStateUp(void) const
 
 double Host::GetLastStateDown(void) const
 {
-	ASSERT(!OwnsLock());
-
 	return GetLastStateCritical();
 }
 

tools/icinga2-discover-agent.cmake

@@ -109,7 +109,7 @@ class NetstringParser(object):
 # along with this program; if not, write to the Free Software Foundation
 # Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.
 
-import socket, ssl, sys, json, os, hashlib
+import socket, ssl, sys, json, os, hashlib, time
 
 def warning(*objs):
     print(*objs, file=sys.stderr)
@@ -180,11 +180,12 @@ if method != "push_crs":
     warning("Agent did not return any check results. Make sure you're using the master certificate.")
     sys.exit(1)
 
-params = response['params']
+params = response["params"]
+params["seen"] = time.time()
 
 inventory_file = "@CMAKE_INSTALL_FULL_LOCALSTATEDIR@/lib/icinga2/agent/inventory/" + hashlib.sha256(cn).hexdigest()
 fp = open(inventory_file, "w")
-inventory_info = { "identity": cn, "crs": params }
+inventory_info = { "identity": cn, "params": params }
 json.dump(inventory_info, fp)
 fp.close()
 

tools/icinga2-list-agents.cmake

@@ -17,6 +17,7 @@
 # Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.
 
 import sys, os, json
+from datetime import datetime
 
 inventory_dir = "@CMAKE_INSTALL_FULL_LOCALSTATEDIR@/lib/icinga2/agent/inventory/"
 
@@ -32,7 +33,11 @@ for root, dirs, files in os.walk(inventory_dir):
         fp.close()
 
         inventory[inventory_info["identity"]] = {}
-        inventory[inventory_info["identity"]]["services"] = inventory_info["crs"]["services"].keys()
+        inventory[inventory_info["identity"]]["seen"] = inventory_info["params"]["seen"]
+        inventory[inventory_info["identity"]]["hosts"] = {}
+
+        for host, host_info in inventory_info["params"]["hosts"].items():
+            inventory[inventory_info["identity"]]["hosts"][host] = { "services": host_info["services"].keys() }
 
         try:
             fp = open(root + file + ".peer", "r")
@@ -46,14 +51,17 @@ for root, dirs, files in os.walk(inventory_dir):
 if len(sys.argv) > 1 and sys.argv[1] == "--batch":
     json.dump(inventory, sys.stdout)
 else:
-    for host, host_info in inventory.items():
+    for agent, agent_info in inventory.items():
-        if "peer" in host_info:
+        if "peer" in agent_info:
-            peer_info = host_info["peer"]
+            peer_info = agent_info["peer"]
-            peer_addr = " (%s:%s)" % (peer_info["agent_host"], peer_info["agent_port"])
+            peer_addr = "peer address: %s:%s, " % (peer_info["agent_host"], peer_info["agent_port"])
         else:
-            peer_addr = ""
+            peer_addr = "no peer address"
 
-        print "* %s%s" % (host, peer_addr)
+        print "* %s (%slast seen: %s)" % (agent, peer_addr, datetime.fromtimestamp(agent_info["seen"]))
+
+        for host, host_info in agent_info["hosts"].items():
+            print "    * %s" % (host)
 
             for service in host_info["services"]:
                 print "        * %s" % (service)