Merge pull request #7643 from Icinga/bugfix/icingadb-recover

IcingaDB: handle Redis restarts

lib/icingadb/redisconnection.cpp

@@ -9,9 +9,9 @@
 #include "base/objectlock.hpp"
 #include "base/string.hpp"
 #include "base/tcpsocket.hpp"
-#include <boost/asio/post.hpp>
+#include <boost/asio.hpp>
-#include <boost/asio/spawn.hpp>
 #include <boost/coroutine/exceptions.hpp>
+#include <boost/date_time/posix_time/posix_time_duration.hpp>
 #include <boost/utility/string_view.hpp>
 #include <boost/variant/get.hpp>
 #include <exception>
@@ -144,6 +144,9 @@ void RedisConnection::Connect(asio::yield_context& yc)
 {
 	Defer notConnecting ([this]() { m_Connecting.store(m_Connected.load()); });
 
+	boost::asio::deadline_timer timer (m_Strand.context());
+
+	for (;;) {
 		try {
 			if (m_Path.IsEmpty()) {
 				Log(LogInformation, "IcingaDB")
@@ -164,12 +167,19 @@ void RedisConnection::Connect(asio::yield_context& yc)
 			m_Connected.store(true);
 
 			Log(LogInformation, "IcingaDB", "Connected to Redis server");
+
+			break;
 		} catch (const boost::coroutines::detail::forced_unwind&) {
 			throw;
 		} catch (const std::exception& ex) {
 			Log(LogCritical, "IcingaDB")
 				<< "Cannot connect to " << m_Host << ":" << m_Port << ": " << ex.what();
 		}
+
+		timer.expires_from_now(boost::posix_time::seconds(5));
+		timer.async_wait(yc);
+	}
+
 }
 
 void RedisConnection::ReadLoop(asio::yield_context& yc)

lib/icingadb/redisconnection.hpp

@@ -207,6 +207,8 @@ private:
 template<class StreamPtr>
 RedisConnection::Reply RedisConnection::ReadOne(StreamPtr& stream, boost::asio::yield_context& yc)
 {
+	namespace asio = boost::asio;
+
 	if (!stream) {
 		throw RedisDisconnected();
 	}
@@ -221,6 +223,12 @@ RedisConnection::Reply RedisConnection::ReadOne(StreamPtr& stream, boost::asio::
 		if (m_Connecting.exchange(false)) {
 			m_Connected.store(false);
 			stream = nullptr;
+
+			if (!m_Connecting.exchange(true)) {
+				Ptr keepAlive (this);
+
+				asio::spawn(m_Strand, [this, keepAlive](asio::yield_context yc) { Connect(yc); });
+			}
 		}
 
 		throw;
@@ -230,6 +238,8 @@ RedisConnection::Reply RedisConnection::ReadOne(StreamPtr& stream, boost::asio::
 template<class StreamPtr>
 void RedisConnection::WriteOne(StreamPtr& stream, RedisConnection::Query& query, boost::asio::yield_context& yc)
 {
+	namespace asio = boost::asio;
+
 	if (!stream) {
 		throw RedisDisconnected();
 	}
@@ -245,6 +255,12 @@ void RedisConnection::WriteOne(StreamPtr& stream, RedisConnection::Query& query,
 		if (m_Connecting.exchange(false)) {
 			m_Connected.store(false);
 			stream = nullptr;
+
+			if (!m_Connecting.exchange(true)) {
+				Ptr keepAlive (this);
+
+				asio::spawn(m_Strand, [this, keepAlive](asio::yield_context yc) { Connect(yc); });
+			}
 		}
 
 		throw;