20058059008

Committed 09 Dec 2025 09:11AM UTC coverage: 73.334% (-0.007%) from 73.341%

Build # 20058059008

Build Type

Pull #16622

github

Committed by

web-flow

Commit Message

Merge a34b21f18 into b39353591

Pull Request Pull Request #16622: LuaWrapper: Add support for `std::variant`

Run Details

38772 of 63654 branches covered (60.91%)

Branch coverage included in aggregate %.

48 of 48 new or added lines in 1 file covered. (100.0%)

62 existing lines in 11 files now uncovered.

128842 of 164907 relevant lines covered (78.13%)

5485937.19 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

58.62

/pdns/remote_logger.cc

#include <unistd.h>
#include "threadname.hh"
#include "remote_logger.hh"
#include <sys/uio.h>
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#ifdef RECURSOR
#include "logger.hh"
#else
#include "dolog.hh"
#endif
#include "logging.hh"

bool CircularWriteBuffer::hasRoomFor(const std::string& str) const
{
  if (d_buffer.size() + 2 + str.size() > d_buffer.capacity()) {
    return false;
  }

  return true;
}

bool CircularWriteBuffer::write(const std::string& str)
{
  if (str.size() > std::numeric_limits<uint16_t>::max() || !hasRoomFor(str)) {
    return false;
  }

  uint16_t len = htons(str.size());
  const char* ptr = reinterpret_cast<const char*>(&len);
  d_buffer.insert(d_buffer.end(), ptr, ptr + 2);
  d_buffer.insert(d_buffer.end(), str.begin(), str.end());

  return true;
}

bool CircularWriteBuffer::flush(int fd)
{
  if (d_buffer.empty()) {
    // not optional, we report EOF otherwise
    return false;
  }

  auto arr1 = d_buffer.array_one();
  auto arr2 = d_buffer.array_two();

  struct iovec iov[2];
  int pos = 0;
  for(const auto& arr : {arr1, arr2}) {
    if(arr.second) {
      iov[pos].iov_base = arr.first;
      iov[pos].iov_len = arr.second;
      ++pos;
    }
  }

  ssize_t res = 0;
  do {
    res = writev(fd, iov, pos);

    if (res < 0) {
      if (errno == EINTR) {
        continue;
      }

      if (errno == EAGAIN || errno == EWOULDBLOCK) {
        return false;
      }

      /* we can't be sure we haven't sent a partial message,
         and we don't want to send the remaining part after reconnecting */
      d_buffer.clear();
      throw std::runtime_error("Couldn't flush a thing: " + stringerror());
    }
    else if (!res) {
      /* we can't be sure we haven't sent a partial message,
         and we don't want to send the remaining part after reconnecting */
      d_buffer.clear();
      throw std::runtime_error("EOF");
    }
  }
  while (res < 0);

  if (static_cast<size_t>(res) == d_buffer.size()) {
    d_buffer.clear();
  }
  else {
    while (res--) {
      d_buffer.pop_front();
    }
  }

  return true;
}

const std::string& RemoteLoggerInterface::toErrorString(Result r)
{
  static const std::array<std::string,5> str = {
    "Queued",
    "Queue full, dropping",
    "Not sending too large protobuf message",
    "Submiting to queue failed",
    "?"
  };
  auto i = static_cast<unsigned int>(r);
  return str[std::min(i, 4U)];
}

RemoteLogger::RemoteLogger(const ComboAddress& remote, uint16_t timeout, uint64_t maxQueuedBytes, uint8_t reconnectWaitTime, bool asyncConnect): d_remote(remote), d_timeout(timeout), d_reconnectWaitTime(reconnectWaitTime), d_asyncConnect(asyncConnect), d_runtime({CircularWriteBuffer(maxQueuedBytes), nullptr})
{
  if (!d_asyncConnect) {
    reconnect();
  }

  d_thread = std::thread(&RemoteLogger::maintenanceThread, this);
}

bool RemoteLogger::reconnect()
{
  try {
    auto newSock = make_unique<Socket>(d_remote.sin4.sin_family, SOCK_STREAM, 0);
    newSock->setNonBlocking();
    newSock->connect(d_remote, d_timeout);

    {
      /* we are now successfully connected, time to take the lock and update the
         socket */
      auto runtime = d_runtime.lock();
      runtime->d_socket = std::move(newSock);
    }
  }
  catch (const std::exception& e) {
#ifdef RECURSOR
    SLOG(g_log<<Logger::Warning<<"Error connecting to remote logger "<<d_remote.toStringWithPort()<<": "<<e.what()<<std::endl,
         g_slog->withName("protobuf")->error(Logr::Error, e.what(), "Exception while connecting to remote logger", "address", Logging::Loggable(d_remote)));
#else
    warnlog("Error connecting to remote logger %s: %s", d_remote.toStringWithPort(), e.what());
#endif

    return false;
  }
  return true;
}

RemoteLoggerInterface::Result RemoteLogger::queueData(const std::string& data)
{
  auto runtime = d_runtime.lock();

  if (data.size() > std::numeric_limits<uint16_t>::max()) {
    ++runtime->d_stats.d_tooLarge;
    return Result::TooLarge;
  }

  if (!runtime->d_writer.hasRoomFor(data)) {
    /* not connected, queue is full, just drop */
    if (!runtime->d_socket) {
      ++runtime->d_stats.d_pipeFull;
      return Result::PipeFull;
    }
    try {
      /* we try to flush some data */
      if (!runtime->d_writer.flush(runtime->d_socket->getHandle())) {
        /* but failed, let's just drop */
        ++runtime->d_stats.d_pipeFull;
        return Result::PipeFull;
      }

      /* see if we freed enough data */
      if (!runtime->d_writer.hasRoomFor(data)) {
        /* we didn't */
        ++runtime->d_stats.d_pipeFull;
        return Result::PipeFull;
      }
    }
    catch(const std::exception& e) {
      //      cout << "Got exception writing: "<<e.what()<<endl;
      runtime->d_socket.reset();
      ++runtime->d_stats.d_otherError;
      return Result::OtherError;
    }
  }

  runtime->d_writer.write(data);
#ifdef RECURSOR
  extern bool g_regressionTestMode;
  if (g_regressionTestMode) {
    runtime->d_writer.flush(runtime->d_socket->getHandle());
  }
#endif
  ++runtime->d_stats.d_queued;
  return Result::Queued;
}

void RemoteLogger::maintenanceThread()
{
  try {
#ifdef RECURSOR
    string threadName = "rec/remlog";
#else
    string threadName = "dnsdist/remLog";
#endif
    setThreadName(threadName);

    for (;;) {
      if (d_exiting) {
        break;
      }

      bool connected = true;
      if (d_runtime.lock()->d_socket == nullptr) {
        // if it was unset, it will remain so, we are the only ones setting it!
        connected = reconnect();
      }

      /* we will just go to sleep if the reconnection just failed */
      if (connected) {
        try {
          /* we don't want to take the lock while trying to reconnect */
          auto runtime = d_runtime.lock();
          if (runtime->d_socket) { // check if it is set
            /* if flush() returns false, it means that we couldn't flush anything yet
               either because there is nothing to flush, or because the outgoing TCP
               buffer is full. That's fine by us */
            runtime->d_writer.flush(runtime->d_socket->getHandle());
          }
          else {
            connected = false;
          }
        }
        catch (const std::exception& e) {
          d_runtime.lock()->d_socket.reset();
          connected = false;
        }

        if (!connected) {
          /* let's try to reconnect right away, we are about to sleep anyway */
          reconnect();
        }
      }

      sleep(d_reconnectWaitTime);
    }
  }
  catch (const std::exception& e)
  {
    SLOG(cerr << "Remote Logger's maintenance thread died on: " << e.what() << endl,
         g_slog->withName("protobuf")->error(Logr::Error, e.what(), "Remote Logger's maintenance thread died"));
  }
  catch (...) {
    SLOG(cerr << "Remote Logger's maintenance thread died on unknown exception" << endl,
         g_slog->withName("protobuf")->info(Logr::Error, "Remote Logger's maintenance thread died"));
  }
}

RemoteLogger::~RemoteLogger()
{
  d_exiting = true;

  d_thread.join();
}


1	#include <unistd.h>
2	#include "threadname.hh"
3	#include "remote_logger.hh"
4	#include <sys/uio.h>
5	#ifdef HAVE_CONFIG_H
6	#include "config.h"
7	#endif
8	#ifdef RECURSOR
9	#include "logger.hh"
10	#else
11	#include "dolog.hh"
12	#endif
13	#include "logging.hh"
14
15	bool CircularWriteBuffer::hasRoomFor(const std::string& str) const
16	{	864✔
17	if (d_buffer.size() + 2 + str.size() > d_buffer.capacity()) {	864!
18	return false;	×
19	}	×
20
21	return true;	864✔
22	}	864✔
23
24	bool CircularWriteBuffer::write(const std::string& str)
25	{	432✔
26	if (str.size() > std::numeric_limits<uint16_t>::max() \|\| !hasRoomFor(str)) {	432!
27	return false;	×
28	}	×
29
30	uint16_t len = htons(str.size());	432✔
31	const char* ptr = reinterpret_cast<const char*>(&len);	432✔
32	d_buffer.insert(d_buffer.end(), ptr, ptr + 2);	432✔
33	d_buffer.insert(d_buffer.end(), str.begin(), str.end());	432✔
34
35	return true;	432✔
36	}	432✔
37
38	bool CircularWriteBuffer::flush(int fd)
39	{	984✔
40	if (d_buffer.empty()) {	984✔
41	// not optional, we report EOF otherwise
42	return false;	576✔
43	}	576✔
44
45	auto arr1 = d_buffer.array_one();	408✔
46	auto arr2 = d_buffer.array_two();	408✔
47
48	struct iovec iov[2];	408✔
49	int pos = 0;	408✔
50	for(const auto& arr : {arr1, arr2}) {	816✔
51	if(arr.second) {	816✔
52	iov[pos].iov_base = arr.first;	412✔
53	iov[pos].iov_len = arr.second;	412✔
54	++pos;	412✔
55	}	412✔
56	}	816✔
57
58	ssize_t res = 0;	408✔
59	do {	408✔
60	res = writev(fd, iov, pos);	408✔
61
62	if (res < 0) {	408!
63	if (errno == EINTR) {	×
64	continue;	×
65	}	×
66
67	if (errno == EAGAIN \|\| errno == EWOULDBLOCK) {	×
68	return false;	×
69	}	×
70
71	/* we can't be sure we haven't sent a partial message,
72	and we don't want to send the remaining part after reconnecting */
73	d_buffer.clear();	×
74	throw std::runtime_error("Couldn't flush a thing: " + stringerror());	×
75	}	×
76	else if (!res) {	408!
77	/* we can't be sure we haven't sent a partial message,
78	and we don't want to send the remaining part after reconnecting */
79	d_buffer.clear();	×
80	throw std::runtime_error("EOF");	×
81	}	×
82	}	408✔
83	while (res < 0);	408!
84
85	if (static_cast<size_t>(res) == d_buffer.size()) {	408!
86	d_buffer.clear();	408✔
87	}	408✔
UNCOV 88	else {	×
UNCOV 89	while (res--) {	×
90	d_buffer.pop_front();	×
91	}	×
UNCOV 92	}	×
93
94	return true;	408✔
95	}	408✔
96
97	const std::string& RemoteLoggerInterface::toErrorString(Result r)
98	{	×
99	static const std::array<std::string,5> str = {	×
100	"Queued",	×
101	"Queue full, dropping",	×
102	"Not sending too large protobuf message",	×
103	"Submiting to queue failed",	×
104	"?"	×
105	};	×
106	auto i = static_cast<unsigned int>(r);	×
107	return str[std::min(i, 4U)];	×
108	}	×
109
110	RemoteLogger::RemoteLogger(const ComboAddress& remote, uint16_t timeout, uint64_t maxQueuedBytes, uint8_t reconnectWaitTime, bool asyncConnect): d_remote(remote), d_timeout(timeout), d_reconnectWaitTime(reconnectWaitTime), d_asyncConnect(asyncConnect), d_runtime({CircularWriteBuffer(maxQueuedBytes), nullptr})	31✔
111	{	251✔
112	if (!d_asyncConnect) {	251!
113	reconnect();	251✔
114	}	251✔
115
116	d_thread = std::thread(&RemoteLogger::maintenanceThread, this);	251✔
117	}	251✔
118
119	bool RemoteLogger::reconnect()
120	{	251✔
121	try {	251✔
122	auto newSock = make_unique<Socket>(d_remote.sin4.sin_family, SOCK_STREAM, 0);	251✔
123	newSock->setNonBlocking();	251✔
124	newSock->connect(d_remote, d_timeout);	251✔
125
126	{	251✔
127	/* we are now successfully connected, time to take the lock and update the
128	socket */
129	auto runtime = d_runtime.lock();	251✔
130	runtime->d_socket = std::move(newSock);	251✔
131	}	251✔
132	}	251✔
133	catch (const std::exception& e) {	251✔
134	#ifdef RECURSOR
135	SLOG(g_log<<Logger::Warning<<"Error connecting to remote logger "<<d_remote.toStringWithPort()<<": "<<e.what()<<std::endl,
136	g_slog->withName("protobuf")->error(Logr::Error, e.what(), "Exception while connecting to remote logger", "address", Logging::Loggable(d_remote)));
137	#else
138	warnlog("Error connecting to remote logger %s: %s", d_remote.toStringWithPort(), e.what());
139	#endif
140
141	return false;	×
142	}	×
143	return true;	251✔
144	}	251✔
145
146	RemoteLoggerInterface::Result RemoteLogger::queueData(const std::string& data)
147	{	432✔
148	auto runtime = d_runtime.lock();	432✔
149
150	if (data.size() > std::numeric_limits<uint16_t>::max()) {	432!
151	++runtime->d_stats.d_tooLarge;	×
152	return Result::TooLarge;	×
153	}	×
154
155	if (!runtime->d_writer.hasRoomFor(data)) {	432!
156	/* not connected, queue is full, just drop */
157	if (!runtime->d_socket) {	×
158	++runtime->d_stats.d_pipeFull;	×
159	return Result::PipeFull;	×
160	}	×
161	try {	×
162	/* we try to flush some data */
163	if (!runtime->d_writer.flush(runtime->d_socket->getHandle())) {	×
164	/* but failed, let's just drop */
165	++runtime->d_stats.d_pipeFull;	×
166	return Result::PipeFull;	×
167	}	×
168
169	/* see if we freed enough data */
170	if (!runtime->d_writer.hasRoomFor(data)) {	×
171	/* we didn't */
172	++runtime->d_stats.d_pipeFull;	×
173	return Result::PipeFull;	×
174	}	×
175	}	×
176	catch(const std::exception& e) {	×
177	// cout << "Got exception writing: "<<e.what()<<endl;
178	runtime->d_socket.reset();	×
179	++runtime->d_stats.d_otherError;	×
180	return Result::OtherError;	×
181	}	×
182	}	×
183
184	runtime->d_writer.write(data);	432✔
185	#ifdef RECURSOR	354✔
186	extern bool g_regressionTestMode;	354✔
187	if (g_regressionTestMode) {	354✔
188	runtime->d_writer.flush(runtime->d_socket->getHandle());	340✔
189	}	340✔
190	#endif	354✔
191	++runtime->d_stats.d_queued;	432✔
192	return Result::Queued;	432✔
193	}	432✔
194
195	void RemoteLogger::maintenanceThread()
196	{	251✔
197	try {	251✔
198	#ifdef RECURSOR	220✔
199	string threadName = "rec/remlog";	220✔
200	#else
201	string threadName = "dnsdist/remLog";	31✔
202	#endif	31✔
203	setThreadName(threadName);	251✔
204
205	for (;;) {	865✔
206	if (d_exiting) {	865✔
207	break;	220✔
208	}	220✔
209
210	bool connected = true;	645✔
211	if (d_runtime.lock()->d_socket == nullptr) {	645!
212	// if it was unset, it will remain so, we are the only ones setting it!
213	connected = reconnect();	×
214	}	×
215
216	/* we will just go to sleep if the reconnection just failed */
217	if (connected) {	645✔
218	try {	642✔
219	/* we don't want to take the lock while trying to reconnect */
220	auto runtime = d_runtime.lock();	642✔
221	if (runtime->d_socket) { // check if it is set	645✔
222	/* if flush() returns false, it means that we couldn't flush anything yet
223	either because there is nothing to flush, or because the outgoing TCP
224	buffer is full. That's fine by us */
225	runtime->d_writer.flush(runtime->d_socket->getHandle());	645✔
226	}	645✔
227	else {	2,147,483,647✔
228	connected = false;	2,147,483,647✔
229	}	2,147,483,647✔
230	}	642✔
231	catch (const std::exception& e) {	642✔
232	d_runtime.lock()->d_socket.reset();	×
233	connected = false;	×
234	}	×
235
236	if (!connected) {	645!
237	/* let's try to reconnect right away, we are about to sleep anyway */
238	reconnect();	×
239	}	×
240	}	645✔
241
242	sleep(d_reconnectWaitTime);	648✔
243	}	648✔
244	}	251✔
245	catch (const std::exception& e)	251✔
246	{	251✔
247	SLOG(cerr << "Remote Logger's maintenance thread died on: " << e.what() << endl,	×
248	g_slog->withName("protobuf")->error(Logr::Error, e.what(), "Remote Logger's maintenance thread died"));	×
249	}	×
250	catch (...) {	251✔
251	SLOG(cerr << "Remote Logger's maintenance thread died on unknown exception" << endl,	×
252	g_slog->withName("protobuf")->info(Logr::Error, "Remote Logger's maintenance thread died"));	×
253	}	×
254	}	251✔
255
256	RemoteLogger::~RemoteLogger()
257	{	220✔
258	d_exiting = true;	220✔
259
260	d_thread.join();	220✔
261	}	220✔
262

PowerDNS / pdns / 20058059008

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous