peter/infix
1
0
Fork 0
mirror of https://github.com/kernelkit/infix.git synced 2026-07-22 01:13:00 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
infix/package/rousette/0010-http-send-keep-alive-pings-from-EventStream.patch
T
Joachim Wiberg 4dee5e4f1f package/rousette: silence rousette warnings in log 
Add --log-level command line option to filter out log messages from
lower log levels.  Then fix the annoying CzechLight warning message
and useless "NACM config validation" log.  Then add audit trail as
we have in netopeer2-server, and finish off by stripping redundant
fields from log message: timestamp, identity, and log level.

Fixes #892

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
2026-03-20 16:18:16 +01:00

430 lines
19 KiB
Diff
Raw Permalink Blame History

From fc240be374f70b0f006bb83152364678ffdbfa86 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tom=C3=A1=C5=A1=20Pecka?= <tomas.pecka@cesnet.cz>
Date: Mon, 12 May 2025 14:56:54 +0200
Subject: [PATCH 10/42] http: send keep-alive pings from EventStream
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Organization: Wires
The nghttp2-asio server disconnects client after 60 seconds of no
activity in the stream [1]. In our use-case I can imagine it can be
quite common that no data will flow for a long period of time. If no
data are updated (i.e., no notifications, or no YANG data changes in
sysrepo when we have YANG push), then nothing flows forth and back.
The 60 second value is configurable [2], but I am not sure about the
value that should be used. 5 minutes? hours? days?
Let's approach this differently. We can instead send a comment into the
stream every few seconds which will make sure that the connection will
not be dropped by nghttp2-asio. This approach is described in the
HTML Living Standard [3].
For now, let's resort to sending a ping-alive every 55 seconds and
hardcode 60 seconds no activity timeout.
I have verified that using this method a curl client keeps connected
to the server when the server sends these "keep-alive comments".
[1] https://github.com/nghttp2/nghttp2/issues/555
[2] https://github.com/nghttp2/nghttp2-asio/blob/e877868abe06a83ed0a6ac6e245c07f6f20866b5/lib/asio_server_http2_impl.h#L51
[3] https://html.spec.whatwg.org/multipage/server-sent-events.html#authoring-notes
Change-Id: I57e510d0b61ac7ed032c582779780c64768b7d53
Signed-off-by: Mattias Walström <lazzer@gmail.com>
Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
---
src/clock.cpp | 4 +-
src/http/EventStream.cpp | 44 +++++++++++++--
src/http/EventStream.h | 13 ++++-
src/restconf/NotificationStream.cpp | 3 +-
src/restconf/NotificationStream.h | 1 +
src/restconf/Server.cpp | 21 +++++--
src/restconf/Server.h | 6 +-
tests/restconf-eventstream.cpp | 85 +++++++++++++++++++++--------
8 files changed, 139 insertions(+), 38 deletions(-)
diff --git a/src/clock.cpp b/src/clock.cpp
index c85be19..24e9efb 100644
--- a/src/clock.cpp
+++ b/src/clock.cpp
@@ -14,6 +14,8 @@
using namespace std::literals;
+constexpr auto keepAlivePingInterval = std::chrono::seconds{55};
+
int main(int argc [[maybe_unused]], char** argv [[maybe_unused]])
{
spdlog::set_level(spdlog::level::trace);
@@ -32,7 +34,7 @@ int main(int argc [[maybe_unused]], char** argv [[maybe_unused]])
server.num_threads(4);
server.handle("/events", [&shutdown, &sig](const auto& req, const auto& res) {
- auto client = std::make_shared<rousette::http::EventStream>(req, res, shutdown, sig);
+ auto client = std::make_shared<rousette::http::EventStream>(req, res, shutdown, sig, keepAlivePingInterval);
client->activate();
});
diff --git a/src/http/EventStream.cpp b/src/http/EventStream.cpp
index 77c282d..82f8b38 100644
--- a/src/http/EventStream.cpp
+++ b/src/http/EventStream.cpp
@@ -17,17 +17,26 @@ using namespace nghttp2::asio_http2;
namespace rousette::http {
+constexpr auto FIELD_DATA = "data";
+
/** @short After constructing, make sure to call activate() immediately. */
-EventStream::EventStream(const server::request& req, const server::response& res, Termination& termination, EventSignal& signal, const std::optional<std::string>& initialEvent)
+EventStream::EventStream(const server::request& req,
+ const server::response& res,
+ Termination& termination,
+ EventSignal& signal,
+ const std::chrono::seconds keepAlivePingInterval,
+ const std::optional<std::string>& initialEvent)
: res{res}
+ , ping{res.io_service()}
, peer{peer_from_request(req)}
+ , m_keepAlivePingInterval(keepAlivePingInterval)
{
if (initialEvent) {
- enqueue(*initialEvent);
+ enqueue(FIELD_DATA, *initialEvent);
}
eventSub = signal.connect([this](const auto& msg) {
- enqueue(msg);
+ enqueue(FIELD_DATA, msg);
});
terminateSub = termination.connect([this]() {
@@ -56,6 +65,8 @@ shared_from_this() throws bad_weak_ptr, so we need a two-phase construction.
*/
void EventStream::activate()
{
+ start_ping();
+
auto client = shared_from_this();
res.write_head(200, {
{"content-type", {"text/event-stream", false}},
@@ -65,6 +76,7 @@ void EventStream::activate()
res.on_close([client](const auto ec) {
spdlog::debug("{}: closed ({})", client->peer, nghttp2_http2_strerror(ec));
std::lock_guard lock{client->mtx};
+ client->ping.cancel();
client->eventSub.disconnect();
client->terminateSub.disconnect();
client->state = Closed;
@@ -113,14 +125,15 @@ ssize_t EventStream::process(uint8_t* destination, std::size_t len, uint32_t* da
__builtin_unreachable();
}
-void EventStream::enqueue(const std::string& what)
+void EventStream::enqueue(const std::string& fieldName, const std::string& what)
{
std::string buf;
buf.reserve(what.size());
const std::regex newline{"\n"};
for (auto it = std::sregex_token_iterator{what.begin(), what.end(), newline, -1};
it != std::sregex_token_iterator{}; ++it) {
- buf += "data: ";
+ buf += fieldName;
+ buf += ": ";
buf += *it;
buf += '\n';
}
@@ -139,4 +152,25 @@ void EventStream::enqueue(const std::string& what)
state = HasEvents;
boost::asio::post(res.io_service(), [&res = this->res]() { res.resume(); });
}
+
+void EventStream::start_ping()
+{
+ ping.expires_from_now(boost::posix_time::seconds(m_keepAlivePingInterval.count()));
+ ping.async_wait([maybeClient = weak_from_this()](const boost::system::error_code& ec) {
+ auto client = maybeClient.lock();
+ if (!client) {
+ spdlog::trace("ping: client already gone");
+ return;
+ }
+
+ if (ec == boost::asio::error::operation_aborted) {
+ spdlog::trace("{}: ping scheduler cancelled", client->peer);
+ return;
+ }
+
+ client->enqueue("", "\n");
+ spdlog::trace("{}: keep-alive ping enqueued", client->peer);
+ client->start_ping();
+ });
+}
}
diff --git a/src/http/EventStream.h b/src/http/EventStream.h
index b427fb3..4b3578a 100644
--- a/src/http/EventStream.h
+++ b/src/http/EventStream.h
@@ -7,6 +7,7 @@
#pragma once
+#include <boost/asio/deadline_timer.hpp>
#include <boost/signals2.hpp>
#include <list>
#include <memory>
@@ -30,7 +31,12 @@ public:
using EventSignal = boost::signals2::signal<void(const std::string& message)>;
using Termination = boost::signals2::signal<void()>;
- EventStream(const nghttp2::asio_http2::server::request& req, const nghttp2::asio_http2::server::response& res, Termination& terminate, EventSignal& signal, const std::optional<std::string>& initialEvent = std::nullopt);
+ EventStream(const nghttp2::asio_http2::server::request& req,
+ const nghttp2::asio_http2::server::response& res,
+ Termination& terminate,
+ EventSignal& signal,
+ const std::chrono::seconds keepAlivePingInterval,
+ const std::optional<std::string>& initialEvent = std::nullopt);
void activate();
private:
@@ -43,13 +49,16 @@ private:
};
State state = WaitingForEvents;
+ boost::asio::deadline_timer ping;
std::list<std::string> queue;
mutable std::mutex mtx; // for `state` and `queue`
boost::signals2::scoped_connection eventSub, terminateSub;
const std::string peer;
+ const std::chrono::seconds m_keepAlivePingInterval;
size_t send_chunk(uint8_t* destination, std::size_t len, uint32_t* data_flags);
ssize_t process(uint8_t* destination, std::size_t len, uint32_t* data_flags);
- void enqueue(const std::string& what);
+ void enqueue(const std::string& fieldName, const std::string& what);
+ void start_ping();
};
}
diff --git a/src/restconf/NotificationStream.cpp b/src/restconf/NotificationStream.cpp
index d0ba938..5017f3e 100644
--- a/src/restconf/NotificationStream.cpp
+++ b/src/restconf/NotificationStream.cpp
@@ -89,12 +89,13 @@ NotificationStream::NotificationStream(
const nghttp2::asio_http2::server::response& res,
rousette::http::EventStream::Termination& termination,
std::shared_ptr<rousette::http::EventStream::EventSignal> signal,
+ const std::chrono::seconds keepAlivePingInterval,
sysrepo::Session session,
libyang::DataFormat dataFormat,
const std::optional<std::string>& filter,
const std::optional<sysrepo::NotificationTimeStamp>& startTime,
const std::optional<sysrepo::NotificationTimeStamp>& stopTime)
- : EventStream(req, res, termination, *signal)
+ : EventStream(req, res, termination, *signal, keepAlivePingInterval)
, m_notificationSignal(signal)
, m_session(std::move(session))
, m_dataFormat(dataFormat)
diff --git a/src/restconf/NotificationStream.h b/src/restconf/NotificationStream.h
index 46f8416..daa79d6 100644
--- a/src/restconf/NotificationStream.h
+++ b/src/restconf/NotificationStream.h
@@ -44,6 +44,7 @@ public:
const nghttp2::asio_http2::server::response& res,
rousette::http::EventStream::Termination& termination,
std::shared_ptr<rousette::http::EventStream::EventSignal> signal,
+ const std::chrono::seconds keepAlivePingInterval,
sysrepo::Session sess,
libyang::DataFormat dataFormat,
const std::optional<std::string>& filter,
diff --git a/src/restconf/Server.cpp b/src/restconf/Server.cpp
index bd2ce0d..31b3d0f 100644
--- a/src/restconf/Server.cpp
+++ b/src/restconf/Server.cpp
@@ -864,13 +864,14 @@ std::vector<std::shared_ptr<boost::asio::io_context>> Server::io_services() cons
return server->io_services();
}
-Server::Server(sysrepo::Connection conn, const std::string& address, const std::string& port, const std::chrono::milliseconds timeout)
+Server::Server(sysrepo::Connection conn, const std::string& address, const std::string& port, const std::chrono::milliseconds timeout, const std::chrono::seconds keepAlivePingInterval)
: m_monitoringSession(conn.sessionStart(sysrepo::Datastore::Operational))
, nacm(conn)
, server{std::make_unique<nghttp2::asio_http2::server::http2>()}
, dwdmEvents{std::make_unique<sr::OpticalEvents>(conn.sessionStart())}
{
server->num_threads(1); // we only use one thread for the server, so we can call join() right away
+ server->read_timeout(boost::posix_time::seconds{60}); // terminate connection after 60 seconds of inactivity (this is explicitly setting the default value)
for (const auto& [module, version, features] : {
std::tuple<std::string, std::string, std::vector<std::string>>{"ietf-restconf", "2017-01-26", {}},
@@ -930,14 +931,14 @@ Server::Server(sysrepo::Connection conn, const std::string& address, const std::
res.end("<XRD xmlns='http://docs.oasis-open.org/ns/xri/xrd-1.0'><Link rel='restconf' href='"s + restconfRoot + "'></XRD>"s);
});
- server->handle("/telemetry/optics", [this](const auto& req, const auto& res) {
+ server->handle("/telemetry/optics", [this, keepAlivePingInterval](const auto& req, const auto& res) {
logRequest(req);
- auto client = std::make_shared<http::EventStream>(req, res, shutdownRequested, opticsChange, as_restconf_push_update(dwdmEvents->currentData(), std::chrono::system_clock::now()));
+ auto client = std::make_shared<http::EventStream>(req, res, shutdownRequested, opticsChange, keepAlivePingInterval, as_restconf_push_update(dwdmEvents->currentData(), std::chrono::system_clock::now()));
client->activate();
});
- server->handle(netconfStreamRoot, [this, conn](const auto& req, const auto& res) mutable {
+ server->handle(netconfStreamRoot, [this, conn, keepAlivePingInterval](const auto& req, const auto& res) mutable {
logRequest(req);
std::optional<std::string> xpathFilter;
@@ -970,7 +971,17 @@ Server::Server(sysrepo::Connection conn, const std::string& address, const std::
// The signal is constructed outside NotificationStream class because it is required to be passed to
// NotificationStream's parent (EventStream) constructor where it already must be constructed
// Yes, this is a hack.
- auto client = std::make_shared<NotificationStream>(req, res, shutdownRequested, std::make_shared<rousette::http::EventStream::EventSignal>(), sess, streamRequest.type.encoding, xpathFilter, startTime, stopTime);
+ auto client = std::make_shared<NotificationStream>(
+ req,
+ res,
+ shutdownRequested,
+ std::make_shared<rousette::http::EventStream::EventSignal>(),
+ keepAlivePingInterval,
+ sess,
+ streamRequest.type.encoding,
+ xpathFilter,
+ startTime,
+ stopTime);
client->activate();
} catch (const auth::Error& e) {
processAuthError(req, res, e, [&res]() {
diff --git a/src/restconf/Server.h b/src/restconf/Server.h
index 112c943..0720c8d 100644
--- a/src/restconf/Server.h
+++ b/src/restconf/Server.h
@@ -28,7 +28,11 @@ std::optional<std::string> as_subtree_path(const std::string& path);
/** @short A RESTCONF-ish server */
class Server {
public:
- explicit Server(sysrepo::Connection conn, const std::string& address, const std::string& port, const std::chrono::milliseconds timeout = std::chrono::milliseconds{0});
+ explicit Server(sysrepo::Connection conn,
+ const std::string& address,
+ const std::string& port,
+ const std::chrono::milliseconds timeout = std::chrono::milliseconds{0},
+ const std::chrono::seconds keepAlivePingInterval = std::chrono::seconds{55});
~Server();
void join();
void stop();
diff --git a/tests/restconf-eventstream.cpp b/tests/restconf-eventstream.cpp
index 3d7fc15..ad1040d 100644
--- a/tests/restconf-eventstream.cpp
+++ b/tests/restconf-eventstream.cpp
@@ -21,9 +21,9 @@ static const auto SERVER_PORT = "10091";
using namespace std::chrono_literals;
-TEST_CASE("Termination on server shutdown")
+TEST_CASE("Event stream tests")
{
- trompeloeil::sequence seqMod1;
+ trompeloeil::sequence seq1, seq2;
sysrepo::setLogLevelStderr(sysrepo::LogLevel::Information);
spdlog::set_level(spdlog::level::trace);
@@ -34,37 +34,76 @@ TEST_CASE("Termination on server shutdown")
srSess.sendRPC(srSess.getContext().newPath("/ietf-factory-default:factory-reset"));
auto nacmGuard = manageNacm(srSess);
- auto server = std::make_unique<rousette::restconf::Server>(srConn, SERVER_ADDRESS, SERVER_PORT);
setupRealNacm(srSess);
+ const std::string notification(R"({"example:eventA":{"message":"blabla","progress":11}})");
RestconfNotificationWatcher netconfWatcher(srConn.sessionStart().getContext());
- const std::string notification(R"({"example:eventA":{"message":"blabla","progress":11}})");
- EXPECT_NOTIFICATION(notification, seqMod1);
- EXPECT_NOTIFICATION(notification, seqMod1);
- EXPECT_NOTIFICATION(notification, seqMod1);
+ SECTION("Termination on server shutdown")
+ {
+ auto server = std::make_unique<rousette::restconf::Server>(srConn, SERVER_ADDRESS, SERVER_PORT);
+
+ EXPECT_NOTIFICATION(notification, seq1);
+ EXPECT_NOTIFICATION(notification, seq1);
+ EXPECT_NOTIFICATION(notification, seq1);
+
+ auto notifSession = sysrepo::Connection{}.sessionStart();
+ auto ctx = notifSession.getContext();
+
+ PREPARE_LOOP_WITH_EXCEPTIONS;
+ auto notificationThread = std::jthread(wrap_exceptions_and_asio(bg, io, [&]() {
+ auto notifSession = sysrepo::Connection{}.sessionStart();
+ auto ctx = notifSession.getContext();
+
+ WAIT_UNTIL_SSE_CLIENT_REQUESTS;
+ SEND_NOTIFICATION(notification);
+ SEND_NOTIFICATION(notification);
+ SEND_NOTIFICATION(notification);
+ waitForCompletionAndBitMore(seq1);
+
+ auto beforeShutdown = std::chrono::system_clock::now();
+ server.reset();
+ auto shutdownDuration = std::chrono::system_clock::now() - beforeShutdown;
+ REQUIRE(shutdownDuration < 5s);
+ }));
+
+ SSEClient client(io, SERVER_ADDRESS, SERVER_PORT, requestSent, netconfWatcher, "/streams/NETCONF/JSON", std::map<std::string, std::string>{AUTH_ROOT});
+
+ RUN_LOOP_WITH_EXCEPTIONS;
+ }
+
+ SECTION("Keep-alive pings")
+ {
+ constexpr auto pingInterval = 1s;
- auto notifSession = sysrepo::Connection{}.sessionStart();
- auto ctx = notifSession.getContext();
+ RestconfNotificationWatcher netconfWatcher(srConn.sessionStart().getContext());
+ auto server = std::make_unique<rousette::restconf::Server>(srConn, SERVER_ADDRESS, SERVER_PORT, std::chrono::milliseconds{0}, pingInterval);
- PREPARE_LOOP_WITH_EXCEPTIONS;
- auto notificationThread = std::jthread(wrap_exceptions_and_asio(bg, io, [&]() {
auto notifSession = sysrepo::Connection{}.sessionStart();
auto ctx = notifSession.getContext();
- WAIT_UNTIL_SSE_CLIENT_REQUESTS;
- SEND_NOTIFICATION(notification);
- SEND_NOTIFICATION(notification);
- SEND_NOTIFICATION(notification);
- waitForCompletionAndBitMore(seqMod1);
+ EXPECT_NOTIFICATION(notification, seq1);
+ expectations.emplace_back(NAMED_REQUIRE_CALL(netconfWatcher, comment(": ")).IN_SEQUENCE(seq2).TIMES(AT_LEAST(1)));
- auto beforeShutdown = std::chrono::system_clock::now();
- server.reset();
- auto shutdownDuration = std::chrono::system_clock::now() - beforeShutdown;
- REQUIRE(shutdownDuration < 5s);
- }));
+ PREPARE_LOOP_WITH_EXCEPTIONS;
+ auto notificationThread = std::jthread(wrap_exceptions_and_asio(bg, io, [&]() {
+ WAIT_UNTIL_SSE_CLIENT_REQUESTS;
+ SEND_NOTIFICATION(notification);
+ std::this_thread::sleep_for(3s); // Wait for the server to send at least one keep-alive ping
+ server.reset();
+ }));
- SSEClient client(io, SERVER_ADDRESS, SERVER_PORT, requestSent, netconfWatcher, "/streams/NETCONF/JSON", std::map<std::string, std::string>{AUTH_ROOT});
+ SSEClient client(
+ io,
+ SERVER_ADDRESS,
+ SERVER_PORT,
+ requestSent,
+ netconfWatcher,
+ "/streams/NETCONF/JSON",
+ std::map<std::string, std::string>{AUTH_ROOT},
+ boost::posix_time::seconds{5},
+ SSEClient::ReportIgnoredLines::Yes);
- RUN_LOOP_WITH_EXCEPTIONS;
+ RUN_LOOP_WITH_EXCEPTIONS;
+ }
}
--
2.43.0
Reference in New Issue View Git Blame Copy Permalink