Add unit tests for JsonRpcConnection connection handling

.github/workflows/container-image.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches:
       - master
-  pull_request: {}
   release:
     types:
       - published

.github/workflows/linux.yml

@@ -5,7 +5,6 @@ on:
     branches:
       - master
       - 'support/*'
-  pull_request: {}
 
 concurrency:
   group: linux-${{ github.event_name == 'push' && github.sha || github.ref }}

lib/base/tlsstream.hpp

@@ -28,32 +28,31 @@ class SeenStream : public ARS
 {
 public:
 	template<class... Args>
-	SeenStream(Args&&... args) : ARS(std::forward<Args>(args)...)
+	explicit SeenStream(Args&&... args) : ARS(std::forward<Args>(args)...), m_Seen(nullptr)
 	{
-		m_Seen.store(nullptr);
 	}
 
 	template<class... Args>
 	auto async_read_some(Args&&... args) -> decltype(((ARS*)nullptr)->async_read_some(std::forward<Args>(args)...))
 	{
 		{
-			auto seen (m_Seen.load());
+			auto* seen (m_Seen.load());
 
 			if (seen) {
-				*seen = Utility::GetTime();
+				*seen = std::chrono::steady_clock::now();
 			}
 		}
 
 		return ((ARS*)this)->async_read_some(std::forward<Args>(args)...);
 	}
 
-	inline void SetSeen(double* seen)
+	void SetSeen(std::chrono::steady_clock::time_point* seen)
 	{
 		m_Seen.store(seen);
 	}
 
 private:
-	std::atomic<double*> m_Seen;
+	std::atomic<std::chrono::steady_clock::time_point*> m_Seen;
 };
 
 struct UnbufferedAsioTlsStreamParams

lib/remote/jsonrpcconnection-heartbeat.cpp

@@ -21,12 +21,17 @@ REGISTER_APIFUNCTION(Heartbeat, event, &JsonRpcConnection::HeartbeatAPIHandler);
  * cluster connection alive when there isn't much going on.
  */
 
+void JsonRpcConnection::SetHeartbeatInterval(std::chrono::milliseconds interval)
+{
+	m_HeartbeatInterval = interval;
+}
+
 void JsonRpcConnection::HandleAndWriteHeartbeats(boost::asio::yield_context yc)
 {
 	boost::system::error_code ec;
 
 	for (;;) {
-		m_HeartbeatTimer.expires_from_now(boost::posix_time::seconds(20));
+		m_HeartbeatTimer.expires_after(m_HeartbeatInterval);
 		m_HeartbeatTimer.async_wait(yc[ec]);
 
 		if (m_ShuttingDown) {

lib/remote/jsonrpcconnection.cpp

@@ -38,7 +38,7 @@ JsonRpcConnection::JsonRpcConnection(const WaitGroup::Ptr& waitGroup, const Stri
 JsonRpcConnection::JsonRpcConnection(const WaitGroup::Ptr& waitGroup, const String& identity, bool authenticated,
 	const Shared<AsioTlsStream>::Ptr& stream, ConnectionRole role, boost::asio::io_context& io)
 	: m_Identity(identity), m_Authenticated(authenticated), m_Stream(stream), m_Role(role),
-	m_Timestamp(Utility::GetTime()), m_Seen(Utility::GetTime()), m_IoStrand(io),
+	m_Timestamp(Utility::GetTime()), m_Seen(std::chrono::steady_clock::now()), m_IoStrand(io),
 	m_OutgoingMessagesQueued(io), m_WriterDone(io), m_ShuttingDown(false), m_WaitGroup(waitGroup),
 	m_CheckLivenessTimer(io), m_HeartbeatTimer(io)
 {
@@ -81,7 +81,7 @@ void JsonRpcConnection::HandleIncomingMessages(boost::asio::yield_context yc)
 			break;
 		}
 
-		m_Seen = Utility::GetTime();
+		m_Seen = std::chrono::steady_clock::now();
 		if (m_Endpoint) {
 			m_Endpoint->AddMessageReceived(jsonString.GetLength());
 		}
@@ -236,6 +236,11 @@ void JsonRpcConnection::SendRawMessage(const String& message)
 	});
 }
 
+void JsonRpcConnection::SetLivenessTimeout(std::chrono::milliseconds timeout)
+{
+	m_LivenessTimeout = timeout;
+}
+
 void JsonRpcConnection::SendMessageInternal(const Dictionary::Ptr& message)
 {
 	if (m_ShuttingDown) {
@@ -411,31 +416,53 @@ void JsonRpcConnection::CheckLiveness(boost::asio::yield_context yc)
 		 * leaking the connection. Therefore close it after a timeout.
 		 */
 
-		m_CheckLivenessTimer.expires_from_now(boost::posix_time::seconds(10));
+		auto anonymousTimeout = m_LivenessTimeout / 6;
+		m_CheckLivenessTimer.expires_after(anonymousTimeout);
 		m_CheckLivenessTimer.async_wait(yc[ec]);
+		if (ec) {
+			Log(LogCritical, "JsonRpcConnection") << "Error waiting for Liveness timer: " << ec.message();
+		}
 
 		if (m_ShuttingDown) {
 			return;
 		}
 
-		auto remote (m_Stream->lowest_layer().remote_endpoint());
+		{
+			auto remote(m_Stream->lowest_layer().remote_endpoint());
 
-		Log(LogInformation, "JsonRpcConnection")
-			<< "Closing anonymous connection [" << remote.address() << "]:" << remote.port() << " after 10 seconds.";
+			auto msg = Log(LogInformation, "JsonRpcConnection");
+			msg << "Closing anonymous connection [" << remote.address() << "]:" << remote.port() << " after ";
+			if (anonymousTimeout > 1s) {
+				msg << anonymousTimeout.count() / 1000 << " seconds.";
+			} else {
+				msg << anonymousTimeout.count() << " milliseconds";
+			}
+		}
 
 		Disconnect();
 	} else {
 		for (;;) {
-			m_CheckLivenessTimer.expires_from_now(boost::posix_time::seconds(30));
+			m_CheckLivenessTimer.expires_after(m_LivenessTimeout / 2);
 			m_CheckLivenessTimer.async_wait(yc[ec]);
+			if (ec) {
+				Log(LogCritical, "JsonRpcConnection") << "Error waiting for Liveness timer: " << ec.message();
+			}
 
 			if (m_ShuttingDown) {
 				break;
 			}
 
-			if (m_Seen < Utility::GetTime() - 60 && (!m_Endpoint || !m_Endpoint->GetSyncing())) {
-				Log(LogInformation, "JsonRpcConnection")
-					<<  "No messages for identity '" << m_Identity << "' have been received in the last 60 seconds.";
+			if (m_Seen + m_LivenessTimeout < std::chrono::steady_clock::now() &&
+				(!m_Endpoint || !m_Endpoint->GetSyncing())) {
+				{
+					auto msg = Log(LogInformation, "JsonRpcConnection");
+					msg << "No messages for identity '" << m_Identity << "' have been received in the last ";
+					if (m_LivenessTimeout > 1s) {
+						msg << m_LivenessTimeout.count() / 1000 << " seconds.";
+					} else {
+						msg << m_LivenessTimeout.count() << " milliseconds";
+					}
+				}
 
 				Disconnect();
 				break;

lib/remote/jsonrpcconnection.hpp

@@ -20,6 +20,8 @@
 namespace icinga
 {
 
+using namespace std::chrono_literals;
+
 enum ClientRole
 {
 	ClientInbound,
@@ -61,6 +63,9 @@ class JsonRpcConnection final : public Object
 	void SendMessage(const Dictionary::Ptr& request);
 	void SendRawMessage(const String& request);
 
+	void SetLivenessTimeout(std::chrono::milliseconds timeout);
+	void SetHeartbeatInterval(std::chrono::milliseconds interval);
+
 	static Value HeartbeatAPIHandler(const intrusive_ptr<MessageOrigin>& origin, const Dictionary::Ptr& params);
 
 	static double GetWorkQueueRate();
@@ -74,14 +79,16 @@ class JsonRpcConnection final : public Object
 	Shared<AsioTlsStream>::Ptr m_Stream;
 	ConnectionRole m_Role;
 	double m_Timestamp;
-	double m_Seen;
+	std::chrono::steady_clock::time_point m_Seen;
 	boost::asio::io_context::strand m_IoStrand;
 	std::vector<String> m_OutgoingMessagesQueue;
 	AsioEvent m_OutgoingMessagesQueued;
 	AsioEvent m_WriterDone;
 	Atomic<bool> m_ShuttingDown;
 	WaitGroup::Ptr m_WaitGroup;
-	boost::asio::deadline_timer m_CheckLivenessTimer, m_HeartbeatTimer;
+	std::chrono::milliseconds m_LivenessTimeout{60s};
+	std::chrono::milliseconds m_HeartbeatInterval{20s};
+	boost::asio::steady_timer m_CheckLivenessTimer, m_HeartbeatTimer;
 
 	JsonRpcConnection(const WaitGroup::Ptr& waitgroup, const String& identity, bool authenticated,
 		const Shared<AsioTlsStream>::Ptr& stream, ConnectionRole role, boost::asio::io_context& io);
@@ -95,8 +102,6 @@ class JsonRpcConnection final : public Object
 
 	void MessageHandler(const Dictionary::Ptr& message);
 
-	void CertificateRequestResponseHandler(const Dictionary::Ptr& message);
-
 	void SendMessageInternal(const Dictionary::Ptr& request);
 };
 

test/CMakeLists.txt

@@ -122,6 +122,7 @@ set(base_test_SOURCES
   remote-configpackageutility.cpp
   remote-httpserverconnection.cpp
   remote-httpmessage.cpp
+  remote-jsonrpcconnection.cpp
   remote-url.cpp
   ${base_OBJS}
   $<TARGET_OBJECTS:config>

test/base-testloggerfixture.hpp

@@ -10,6 +10,12 @@
 #include <boost/test/test_tools.hpp>
 #include <future>
 
+#define CHECK_LOG_MESSAGE(pattern, timeout) BOOST_CHECK(ExpectLogPattern(pattern, timeout))
+#define REQUIRE_LOG_MESSAGE(pattern, timeout) BOOST_REQUIRE(ExpectLogPattern(pattern, timeout))
+
+#define CHECK_NO_LOG_MESSAGE(pattern, timeout) BOOST_CHECK(!ExpectLogPattern(pattern, timeout))
+#define REQUIRE_NO_LOG_MESSAGE(pattern, timeout) BOOST_REQUIRE(!ExpectLogPattern(pattern, timeout))
+
 namespace icinga {
 
 class TestLogger : public Logger
@@ -52,6 +58,13 @@ class TestLogger : public Logger
 		return ret;
 	}
 
+	void Clear()
+	{
+		std::lock_guard lock(m_Mutex);
+		m_Expects.clear();
+		m_LogEntries.clear();
+	}
+
 private:
 	void ProcessLogEntry(const LogEntry& entry) override
 	{
@@ -88,8 +101,8 @@ struct TestLoggerFixture
 	TestLoggerFixture()
 	{
 		testLogger->SetSeverity(testLogger->SeverityToString(LogDebug));
-		testLogger->Activate(true);
 		testLogger->SetActive(true);
+		testLogger->Activate(true);
 	}
 
 	~TestLoggerFixture()