Added serialization support for check results.

Bugfixes.

base/socket.cpp

@@ -329,7 +329,7 @@ void Socket::ReadThreadProc(void)
 			HandleException();
 
 		if (WantsToWrite())
-			; /* notify Write thread */
+			m_WriteCV.notify_all(); /* notify Write thread */
 	}
 }
 
@@ -342,17 +342,18 @@ void Socket::WriteThreadProc(void)
 
 		FD_ZERO(&writefds);
 
-		int fd = GetFD();
-
 		while (!WantsToWrite()) {
+			m_WriteCV.timed_wait(lock, boost::posix_time::seconds(1));
+
 			if (GetFD() == INVALID_SOCKET)
 				return;
-
-			lock.unlock();
-			Sleep(500);
-			lock.lock();
 		}
 
+		int fd = GetFD();
+
+		if (fd == INVALID_SOCKET)
+			return;
+
 		FD_SET(fd, &writefds);
 
 		lock.unlock();

base/socket.h

@@ -83,6 +83,8 @@ private:
 	thread m_ReadThread;
 	thread m_WriteThread;
 
+	condition_variable m_WriteCV;
+
 	void ReadThreadProc(void);
 	void WriteThreadProc(void);
 

components/checker/checkercomponent.cpp

@@ -117,7 +117,7 @@ void CheckerComponent::ResultTimerHandler(void)
 		CheckResult result = task->GetResult();
 		Application::Log(LogDebug, "checker", "Got result for service '" + service.GetName() + "'");
 
-		long latency = result.EndTime - result.StartTime;
+		long latency = result.GetEndTime() - result.GetStartTime();
 		avg_latency += latency;
 
 		if (min_latency == -1 || latency < min_latency)
@@ -128,9 +128,22 @@ void CheckerComponent::ResultTimerHandler(void)
 
 		results++;
 
-		if (result.State != StateOK)
+		if (result.GetState() != StateOK)
 			failed++;
 
+		RequestMessage rm;
+		rm.SetMethod("checker::CheckResult");
+
+		MessagePart params;
+		params.SetProperty("service", service.GetName());
+		params.SetProperty("result", result.GetDictionary());
+
+		rm.SetParams(params);
+
+		GetEndpointManager()->SendMulticastMessage(m_CheckerEndpoint, rm);
+
+		service.ApplyCheckResult(result);
+
 		service.SetNextCheck(now + service.GetCheckInterval());
 		m_PendingServices.erase(service.GetConfigObject());
 		m_Services.push(service);

components/delegation/delegationcomponent.cpp

@@ -41,6 +41,7 @@ void DelegationComponent::Start(void)
 	m_DelegationEndpoint = boost::make_shared<VirtualEndpoint>();
 	m_DelegationEndpoint->RegisterPublication("checker::AssignService");
 	m_DelegationEndpoint->RegisterPublication("checker::ClearServices");
+	m_DelegationEndpoint->RegisterSubscription("checker::CheckResult");
 	GetEndpointManager()->RegisterEndpoint(m_DelegationEndpoint);
 
 	GetEndpointManager()->OnNewEndpoint.connect(bind(&DelegationComponent::NewEndpointHandler, this, _2));

icinga/Makefile.am

@@ -5,6 +5,8 @@ pkglib_LTLIBRARIES =  \
 	libicinga.la
 
 libicinga_la_SOURCES =  \
+	checkresult.cpp \
+	checkresult.h \
 	checktask.cpp \
 	checktask.h \
 	configobjectadapter.cpp \

icinga/checkresult.cpp

@@ -0,0 +1,76 @@
+#include "i2-icinga.h"
+
+using namespace icinga;
+
+CheckResult::CheckResult(void)
+	: m_Data(boost::make_shared<Dictionary>())
+{ }
+
+CheckResult::CheckResult(const Dictionary::Ptr& dictionary)
+	: m_Data(dictionary)
+{ }
+
+Dictionary::Ptr CheckResult::GetDictionary(void) const
+{
+	return m_Data;
+}
+
+void CheckResult::SetStartTime(time_t ts)
+{
+	m_Data->SetProperty("start_time", static_cast<long>(ts));
+}
+
+time_t CheckResult::GetStartTime(void) const
+{
+	long value = 0;
+	m_Data->GetProperty("start_time", &value);
+	return static_cast<time_t>(value);
+}
+
+void CheckResult::SetEndTime(time_t ts)
+{
+	m_Data->SetProperty("end_time", static_cast<long>(ts));
+}
+
+time_t CheckResult::GetEndTime(void) const
+{
+	long value = 0;
+	m_Data->GetProperty("end_time", &value);
+	return static_cast<time_t>(value);
+}
+
+void CheckResult::SetState(CheckState state)
+{
+	m_Data->SetProperty("state", static_cast<long>(state));
+}
+
+CheckState CheckResult::GetState(void) const
+{
+	long value = StateUnknown;
+	m_Data->GetProperty("state", &value);
+	return static_cast<CheckState>(value);
+}
+
+void CheckResult::SetOutput(string output)
+{
+	m_Data->SetProperty("output", output);
+}
+
+string CheckResult::GetOutput(void) const
+{
+	string value;
+	m_Data->GetProperty("output", &value);
+	return value;
+}
+
+void CheckResult::SetPerformanceData(const Dictionary::Ptr& pd)
+{
+	m_Data->SetProperty("performance_data", pd);
+}
+
+Dictionary::Ptr CheckResult::GetPerformanceData(void) const
+{
+	Dictionary::Ptr value;
+	m_Data->GetProperty("performance_data", &value);
+	return value;
+}

icinga/checkresult.h

@@ -0,0 +1,46 @@
+#ifndef CHECKRESULT_H
+#define CHECKRESULT_H
+
+namespace icinga
+{
+
+enum CheckState
+{
+	StateOK,
+	StateWarning,
+	StateCritical,
+	StateUnreachable,
+	StateUncheckable,
+	StateUnknown
+};
+
+struct CheckResult
+{
+public:
+	CheckResult(void);
+	CheckResult(const Dictionary::Ptr& dictionary);
+
+	Dictionary::Ptr GetDictionary(void) const;
+
+	void SetStartTime(time_t ts);
+	time_t GetStartTime(void) const;
+
+	void SetEndTime(time_t ts);
+	time_t GetEndTime(void) const;
+
+	void SetState(CheckState state);
+	CheckState GetState(void) const;
+
+	void SetOutput(string output);
+	string GetOutput(void) const;
+
+	void SetPerformanceData(const Dictionary::Ptr& pd);
+	Dictionary::Ptr GetPerformanceData(void) const;
+
+private:
+	Dictionary::Ptr m_Data;
+};
+
+}
+
+#endif /* CHECKRESULT_H */

icinga/checktask.h

@@ -4,26 +4,6 @@
 namespace icinga
 {
 
-enum CheckState
-{
-	StateOK,
-	StateWarning,
-	StateCritical,
-	StateUnreachable,
-	StateUncheckable,
-	StateUnknown
-};
-
-struct CheckResult
-{
-	time_t StartTime;
-	time_t EndTime;
-
-	CheckState State;
-	string Output;
-	Dictionary::Ptr PerformanceData;
-};
-
 struct CheckTaskType;
 
 class I2_ICINGA_API CheckTask : public Object

icinga/i2-icinga.h

@@ -53,6 +53,7 @@ using boost::unique_future;
 #include "service.h"
 
 #include "macroprocessor.h"
+#include "checkresult.h"
 #include "checktask.h"
 #include "nagioschecktask.h"
 

icinga/nagioschecktask.cpp

@@ -20,7 +20,10 @@ NagiosCheckTask::NagiosCheckTask(const Service& service)
 
 void NagiosCheckTask::Enqueue(void)
 {
-	time(&m_Result.StartTime);
+	time_t now;
+	time(&now);
+	m_Result.SetStartTime(now);
+
 	m_PendingTasks.push_back(GetSelf());
 }
 
@@ -44,7 +47,6 @@ void NagiosCheckTask::CheckThreadProc(void)
 	mutex::scoped_lock lock(m_Mutex);
 
 	map<int, NagiosCheckTask::Ptr> tasks;
-	const int maxTasks = 128;
 
 	for (;;) {
 		while (m_Tasks.empty() || tasks.size() >= MaxChecksPerThread) {
@@ -94,7 +96,7 @@ void NagiosCheckTask::CheckThreadProc(void)
 			lock.lock();
 		}
 
-		while (!m_Tasks.empty() && tasks.size() < maxTasks) {
+		while (!m_Tasks.empty() && tasks.size() < MaxChecksPerThread) {
 			NagiosCheckTask::Ptr task = m_Tasks.front();
 			m_Tasks.pop_front();
 			if (!task->InitTask()) {
@@ -140,8 +142,9 @@ bool NagiosCheckTask::RunTask(void)
 	if (!feof(m_FP))
 		return true;
 
-	m_Result.Output = m_OutputStream.str();
+	string output = m_OutputStream.str();
-	boost::algorithm::trim(m_Result.Output);
+	boost::algorithm::trim(output);
+	m_Result.SetOutput(output);
 
 	int status, exitcode;
 #ifdef _MSC_VER
@@ -162,28 +165,36 @@ bool NagiosCheckTask::RunTask(void)
 		exitcode = status;
 #endif /* _MSC_VER */
 
+		CheckState state;
+
 		switch (exitcode) {
 			case 0:
-				m_Result.State = StateOK;
+				state = StateOK;
 				break;
 			case 1:
-				m_Result.State = StateWarning;
+				state = StateWarning;
 				break;
 			case 2:
-				m_Result.State = StateCritical;
+				state = StateCritical;
 				break;
 			default:
-				m_Result.State = StateUnknown;
+				state = StateUnknown;
 				break;
 		}
+
+		m_Result.SetState(state);
 #ifndef _MSC_VER
 	} else if (WIFSIGNALED(status)) {
-		m_Result.Output = "Process was terminated by signal " + WTERMSIG(status);
+		stringstream outputbuf;
-		m_Result.State = StateUnknown;
+		outputbuf << "Process was terminated by signal " << WTERMSIG(status);
+		m_Result.SetOutput(outputbuf.str());
+		m_Result.SetState(StateUnknown);
 	}
 #endif /* _MSC_VER */
 
-	time(&m_Result.EndTime);
+	time_t now;
+	time(&now);
+	m_Result.SetEndTime(now);
 
 	return false;
 }

icinga/service.cpp

@@ -95,3 +95,7 @@ string Service::GetChecker(void) const
 	return value;
 }
 
+void Service::ApplyCheckResult(const CheckResult& cr)
+{
+
+}

icinga/service.h

@@ -4,6 +4,8 @@
 namespace icinga
 {
 
+struct CheckResult;
+
 class I2_ICINGA_API Service : public ConfigObjectAdapter
 {
 public:
@@ -24,6 +26,8 @@ public:
 	time_t GetNextCheck(void);
 	void SetChecker(string checker);
 	string GetChecker(void) const;
+
+	void ApplyCheckResult(const CheckResult& cr);
 };
 
 }