(WIP) Abstract out connection-handling for Perfdata-Writers

Author: jschmidt-icinga

lib/base/i2-base.hpp

@@ -75,5 +75,10 @@
 #define BOOST_BIND_NO_PLACEHOLDERS
 
 #include <functional>
+#include <chrono>
+
+namespace icinga {
+using namespace std::chrono_literals;
+} // namespace icinga
 
 #endif /* I2BASE_H */

lib/perfdata/CMakeLists.txt

@@ -18,6 +18,7 @@ set(perfdata_SOURCES
   influxdb2writer.cpp influxdb2writer.hpp influxdb2writer-ti.hpp
   opentsdbwriter.cpp opentsdbwriter.hpp opentsdbwriter-ti.hpp
   perfdatawriter.cpp perfdatawriter.hpp perfdatawriter-ti.hpp
+  perfdatabackendconnection.cpp
 )
 
 if(ICINGA2_UNITY_BUILD)

lib/perfdata/elasticsearchwriter.cpp

@@ -91,10 +91,25 @@ void ElasticsearchWriter::Resume()
 	/* Setup timer for periodically flushing m_DataBuffer */
 	m_FlushTimer = Timer::Create();
 	m_FlushTimer->SetInterval(GetFlushInterval());
-	m_FlushTimer->OnTimerExpired.connect([this](const Timer * const&) { FlushTimeout(); });
+	m_FlushTimer->OnTimerExpired.connect([this](const Timer* const&) {
+		m_WorkQueue.Enqueue([&]() { FlushTimeout(); });
+	});
 	m_FlushTimer->Start();
 	m_FlushTimer->Reschedule(0);
 
+	Shared<boost::asio::ssl::context>::Ptr sslContext;
+	if (GetEnableTls()) {
+		try {
+			sslContext = MakeAsioSslContext(GetCertPath(), GetKeyPath(), GetCaPath());
+		} catch (const std::exception& ex) {
+			Log(LogWarning, GetReflectionType()->GetName())
+				<< "Unable to create SSL context.";
+			throw;
+		}
+	}
+	
+	m_Connection = new PerfdataWriterConnection{GetHost(), GetPort(), sslContext};
+
 	/* Register for new metrics. */
 	m_HandleCheckResults = Checkable::OnNewCheckResult.connect([this](const Checkable::Ptr& checkable,
 		const CheckResult::Ptr& cr, const MessageOrigin::Ptr&) {
@@ -118,6 +133,10 @@ void ElasticsearchWriter::Pause()
 	m_HandleStateChanges.disconnect();
 	m_HandleNotifications.disconnect();
 
+	m_Connection->StartDisconnectTimeout(
+		std::chrono::milliseconds{static_cast<unsigned>(GetDisconnectTimeout() * 1000)}
+	);
+
 	m_FlushTimer->Stop(true);
 	m_WorkQueue.Join();
 
@@ -465,22 +484,6 @@ void ElasticsearchWriter::SendRequest(const String& body)
 
 	url->SetPath(path);
 
-	OptionalTlsStream stream;
-
-	try {
-		stream = Connect();
-	} catch (const std::exception& ex) {
-		Log(LogWarning, "ElasticsearchWriter")
-			<< "Flush failed, cannot connect to Elasticsearch: " << DiagnosticInformation(ex, false);
-		return;
-	}
-
-	Defer s ([&stream]() {
-		if (stream.first) {
-			stream.first->next_layer().shutdown();
-		}
-	});
-
 	http::request<http::string_body> request (http::verb::post, std::string(url->Format(true)), 10);
 
 	request.set(http::field::user_agent, "Icinga/" + Application::GetAppVersion());
@@ -510,37 +513,19 @@ void ElasticsearchWriter::SendRequest(const String& body)
 		<< "Sending " << request.method_string() << " request" << ((!username.IsEmpty() && !password.IsEmpty()) ? " with basic auth" : "" )
 		<< " to '" << url->Format() << "'.";
 
+	decltype(m_Connection->Send(request)) response;
 	try {
-		if (stream.first) {
-			http::write(*stream.first, request);
-			stream.first->flush();
-		} else {
-			http::write(*stream.second, request);
-			stream.second->flush();
+		response = m_Connection->Send(request);
+	} catch (const std::exception& ex) {
+		if (const auto* se = dynamic_cast<const boost::system::system_error*>(&ex);
+			se->code() == boost::system::errc::operation_canceled) {
+			Log(LogDebug, "ElasticsearchWriter") << "Operation Cancelled.";
+			return;
 		}
-	} catch (const std::exception&) {
-		Log(LogWarning, "ElasticsearchWriter")
-			<< "Cannot write to HTTP API on host '" << GetHost() << "' port '" << GetPort() << "'.";
-		throw;
-	}
-
-	http::parser<false, http::string_body> parser;
-	beast::flat_buffer buf;
 
-	try {
-		if (stream.first) {
-			http::read(*stream.first, buf, parser);
-		} else {
-			http::read(*stream.second, buf, parser);
-		}
-	} catch (const std::exception& ex) {
-		Log(LogWarning, "ElasticsearchWriter")
-			<< "Failed to parse HTTP response from host '" << GetHost() << "' port '" << GetPort() << "': " << DiagnosticInformation(ex, false);
-		throw;
+		Log(LogCritical, "ElasticsearchWriter") << "Error sending Request: " << ex.what();
 	}
 
-	auto& response (parser.get());
-
 	if (response.result_int() > 299) {
 		if (response.result() == http::status::unauthorized) {
 			/* More verbose error logging with Elasticsearch is hidden behind a proxy. */
@@ -588,66 +573,6 @@ void ElasticsearchWriter::SendRequest(const String& body)
 	}
 }
 
-OptionalTlsStream ElasticsearchWriter::Connect()
-{
-	Log(LogNotice, "ElasticsearchWriter")
-		<< "Connecting to Elasticsearch on host '" << GetHost() << "' port '" << GetPort() << "'.";
-
-	OptionalTlsStream stream;
-	bool tls = GetEnableTls();
-
-	if (tls) {
-		Shared<boost::asio::ssl::context>::Ptr sslContext;
-
-		try {
-			sslContext = MakeAsioSslContext(GetCertPath(), GetKeyPath(), GetCaPath());
-		} catch (const std::exception&) {
-			Log(LogWarning, "ElasticsearchWriter")
-				<< "Unable to create SSL context.";
-			throw;
-		}
-
-		stream.first = Shared<AsioTlsStream>::Make(IoEngine::Get().GetIoContext(), *sslContext, GetHost());
-
-	} else {
-		stream.second = Shared<AsioTcpStream>::Make(IoEngine::Get().GetIoContext());
-	}
-
-	try {
-		icinga::Connect(tls ? stream.first->lowest_layer() : stream.second->lowest_layer(), GetHost(), GetPort());
-	} catch (const std::exception&) {
-		Log(LogWarning, "ElasticsearchWriter")
-			<< "Can't connect to Elasticsearch on host '" << GetHost() << "' port '" << GetPort() << "'.";
-		throw;
-	}
-
-	if (tls) {
-		auto& tlsStream (stream.first->next_layer());
-
-		try {
-			tlsStream.handshake(tlsStream.client);
-		} catch (const std::exception&) {
-			Log(LogWarning, "ElasticsearchWriter")
-				<< "TLS handshake with host '" << GetHost() << "' on port " << GetPort() << " failed.";
-			throw;
-		}
-
-		if (!GetInsecureNoverify()) {
-			if (!tlsStream.GetPeerCertificate()) {
-				BOOST_THROW_EXCEPTION(std::runtime_error("Elasticsearch didn't present any TLS certificate."));
-			}
-
-			if (!tlsStream.IsVerifyOK()) {
-				BOOST_THROW_EXCEPTION(std::runtime_error(
-					"TLS certificate validation failed: " + std::string(tlsStream.GetVerifyError())
-				));
-			}
-		}
-	}
-
-	return stream;
-}
-
 void ElasticsearchWriter::AssertOnWorkQueue()
 {
 	ASSERT(m_WorkQueue.IsWorkerThread());

lib/perfdata/elasticsearchwriter.hpp

@@ -9,6 +9,7 @@
 #include "base/workqueue.hpp"
 #include "base/timer.hpp"
 #include "base/tlsstream.hpp"
+#include "perfdata/perfdatabackendconnection.hpp"
 
 namespace icinga
 {
@@ -39,6 +40,8 @@ class ElasticsearchWriter final : public ObjectImpl<ElasticsearchWriter>
 	std::vector<String> m_DataBuffer;
 	std::mutex m_DataBufferMutex;
 
+	PerfdataWriterConnection::Ptr m_Connection;
+
 	void AddCheckResult(const Dictionary::Ptr& fields, const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
 	void AddTemplateTags(const Dictionary::Ptr& fields, const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
 
@@ -50,7 +53,6 @@ class ElasticsearchWriter final : public ObjectImpl<ElasticsearchWriter>
 	void Enqueue(const Checkable::Ptr& checkable, const String& type,
 		const Dictionary::Ptr& fields, double ts);
 
-	OptionalTlsStream Connect();
 	void AssertOnWorkQueue();
 	void ExceptionHandler(boost::exception_ptr exp);
 	void FlushTimeout();

lib/perfdata/elasticsearchwriter.ti

@@ -39,7 +39,10 @@ class ElasticsearchWriter : ConfigObject
 	[config] String cert_path;
 	[config] String key_path;
 
-	[config] int flush_interval {
+	[config] double disconnect_timeout {
+		default {{{ return 0.5; }}}
+	};
+	[config] double flush_interval {
 		default {{{ return 10; }}}
 	};
 	[config] int flush_threshold {