From 89d313008dae9172297830c9267f1e3a2b6ce041 Mon Sep 17 00:00:00 2001
From: Gunnar Beutner <gunnar.beutner@netways.de>
Date: Fri, 18 Oct 2013 09:11:21 +0200
Subject: [PATCH] Fix: GraphiteWriter: reconnect on broken socket.

Fixes #4883
---
 components/perfdata/graphitewriter.cpp | 24 ++++++++++++++++++------
 lib/base/bufferedstream.cpp            |  6 ++++++
 lib/base/networkstream.cpp             | 11 ++++++++++-
 lib/base/socket.cpp                    |  2 +-
 4 files changed, 35 insertions(+), 8 deletions(-)

diff --git a/components/perfdata/graphitewriter.cpp b/components/perfdata/graphitewriter.cpp
index ab01a2f07..9d990af36 100644
--- a/components/perfdata/graphitewriter.cpp
+++ b/components/perfdata/graphitewriter.cpp
@@ -37,6 +37,7 @@
 #include <boost/foreach.hpp>
 #include <boost/algorithm/string/split.hpp>
 #include <boost/algorithm/string/replace.hpp>
+#include <boost/exception/diagnostic_information.hpp>
 
 using namespace icinga;
 
@@ -198,17 +199,28 @@ void GraphiteWriter::AddServiceMetric(std::vector<String>& metrics, const Servic
 
 void GraphiteWriter::SendMetrics(const std::vector<String>& metrics)
 {
-	if (!m_Stream) {
-		Log(LogWarning, "perfdata", "GraphiteWriter not connected!");
-		return;
-	}
-
 	BOOST_FOREACH(const String& metric, metrics) {
 		if (metric.IsEmpty())
 			continue;
 
 		Log(LogDebug, "perfdata", "GraphiteWriter: Sending metric '" + metric + "'.");
-		m_Stream->Write(metric.CStr(), metric.GetLength());
+
+		ObjectLock olock(this);
+
+		if (!m_Stream)
+			return;
+
+		try {
+			m_Stream->Write(metric.CStr(), metric.GetLength());
+		} catch (const std::exception& ex) {
+			std::ostringstream msgbuf;
+			msgbuf << "Exception thrown while writing to the Graphite socket: " << std::endl
+                               << boost::diagnostic_information(ex);
+
+			Log(LogCritical, "base", msgbuf.str());
+
+			m_Stream.reset();
+		}
 	}
 }
 
diff --git a/lib/base/bufferedstream.cpp b/lib/base/bufferedstream.cpp
index 8277e32c0..682569cc5 100644
--- a/lib/base/bufferedstream.cpp
+++ b/lib/base/bufferedstream.cpp
@@ -151,6 +151,9 @@ size_t BufferedStream::Read(void *buffer, size_t count)
 	if (m_Exception)
 		boost::rethrow_exception(m_Exception);
 
+	if (m_Eof)
+		BOOST_THROW_EXCEPTION(std::invalid_argument("Tried to read from closed socket."));
+
 	return m_RecvQ->Read(buffer, count);
 }
 
@@ -169,6 +172,9 @@ void BufferedStream::Write(const void *buffer, size_t count)
 	if (m_Exception)
 		boost::rethrow_exception(m_Exception);
 
+	if (m_Eof)
+		BOOST_THROW_EXCEPTION(std::invalid_argument("Tried to write to closed socket."));
+
 	m_SendQ->Write(buffer, count);
 	m_WriteCV.notify_all();
 }
diff --git a/lib/base/networkstream.cpp b/lib/base/networkstream.cpp
index ea86ff2cd..d614886eb 100644
--- a/lib/base/networkstream.cpp
+++ b/lib/base/networkstream.cpp
@@ -45,6 +45,9 @@ size_t NetworkStream::Read(void *buffer, size_t count)
 {
 	size_t rc;
 
+	if (m_Eof)
+		BOOST_THROW_EXCEPTION(std::invalid_argument("Tried to read from closed socket."));
+
 	try {
 		rc = m_Socket->Read(buffer, count);
 	} catch (...) {
@@ -70,6 +73,9 @@ void NetworkStream::Write(const void *buffer, size_t count)
 {
 	size_t rc;
 
+	if (m_Eof)
+		BOOST_THROW_EXCEPTION(std::invalid_argument("Tried to write to closed socket."));
+
 	try {
 		rc = m_Socket->Write(buffer, count);
 	} catch (...) {
@@ -78,8 +84,11 @@ void NetworkStream::Write(const void *buffer, size_t count)
 		throw;
 	}
 
-	if (rc < count)
+	if (rc < count) {
+		m_Eof = true;
+
 		BOOST_THROW_EXCEPTION(std::runtime_error("Short write for socket."));
+	}
 }
 
 bool NetworkStream::IsEof(void) const
diff --git a/lib/base/socket.cpp b/lib/base/socket.cpp
index 667cbfe99..32facc858 100644
--- a/lib/base/socket.cpp
+++ b/lib/base/socket.cpp
@@ -216,7 +216,7 @@ void Socket::Listen(void)
 }
 
 /**
- * Processes data that is available for this socket.
+ * Sends data for the socket.
  */
 size_t Socket::Write(const void *buffer, size_t count)
 {