Skip to content

Commit 72f327c

Browse files
sy-csy-c
authored
Merge pull request #59 from sy-c/master
v1.3.20
2 parents b83ac3a + 6b877b6 commit 72f327c

File tree

5 files changed

+338
-82
lines changed

5 files changed

+338
-82
lines changed

‎CMakeLists.txt‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -241,6 +241,7 @@ add_executable(
241241
$<TARGET_OBJECTS:objCommonSimpleLog>
242242
$<TARGET_OBJECTS:objCommonDaemon>
243243
$<TARGET_OBJECTS:objCommonThread>
244+
$<TARGET_OBJECTS:objCommonTimer>
244245
src/infoLoggerServer.cxx
245246
src/InfoLoggerDispatch.cxx
246247
src/InfoLoggerDispatchBrowser.cxx
@@ -375,6 +376,7 @@ set(TEST_SRCS
375376
test/testInfoLogger.c
376377
test/testInfoLogger.cxx
377378
test/testInfoLoggerPerf.cxx
379+
test/testInfoLoggerDB.cxx
378380
)
379381
foreach (f ${TEST_SRCS})
380382
get_filename_component(test_name ${f} NAME)
@@ -387,6 +389,19 @@ foreach (f ${TEST_SRCS})
387389
add_test(NAME ${test_name} COMMAND ${exe})
388390
endforeach()
389391

392+
target_include_directories(
393+
testInfoLoggerDB.exe
394+
PRIVATE
395+
${INFOLOGGER_INCLUDE_DIRS_PUBLIC}
396+
${COMMON_STANDALONE_INCLUDE_DIRS}
397+
${MYSQL_INCLUDE_DIRS}
398+
)
399+
target_link_libraries(
400+
testInfoLoggerDB.exe
401+
pthread
402+
${MYSQL_LIBRARIES}
403+
)
404+
390405

391406
# define what should be built
392407
# exclude modules requiring optional dependencies not found

‎doc/releaseNotes.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,3 +72,8 @@ This file describes the main feature changes for each InfoLogger released versio
7272
## v1.3.19 - 22/01/2021
7373
- added infoLoggerServer debug options
7474
- added option to reset local message queue on infoLoggerD startup
75+
76+
## v1.3.20 - 01/02/2021
77+
- infoLoggerServer:
78+
- bulk insert mode (transactions).
79+
- drop messages too long for DB.

‎src/InfoLogger.cxx‎

Lines changed: 70 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -301,6 +301,7 @@ class InfoLogger::Impl
301301

302302
// message flood prevention
303303
// constants
304+
const bool flood_protection = 1; // if set, flood protection mechanism enabled
304305
const unsigned int flood_maxmsg_sec=500; // maximum messages in one second to trigger flood
305306
const unsigned int flood_maxmsg_min=1000; // maximum messages in one minute to trigger flood
306307
const unsigned int flood_maxmsg_file=1000; // maximum number of messages in overflow log file
@@ -475,76 +476,79 @@ int InfoLogger::Impl::pushMessage(const InfoLoggerMessageOption& options, const
475476
InfoLoggerMessageHelperSetValue(msg, msgHelper.ix_username, String, context.userName.c_str());
476477
}
477478

478-
// update message statistics */
479-
if (now-floodStat_time_lastsec>1) {
480-
floodStat_time_lastsec=now;
481-
floodStat_msgs_sec=1;
482-
} else {
483-
floodStat_msgs_sec++;
484-
}
485-
if (now-floodStat_time_lastmin>60) {
486-
floodStat_time_lastmin=now;
487-
floodStat_msgs_lastmin=floodStat_msgs_min;
488-
floodStat_msgs_min=1;
489-
} else {
490-
floodStat_msgs_min++;
491-
}
492-
493-
// message flood prevention
494-
if (!noFlood) {
495-
switch (floodMode) {
496-
case 0:
497-
if ((floodStat_msgs_sec>flood_maxmsg_sec)||(floodStat_msgs_min>flood_maxmsg_min)) {
498-
floodFile_msg=0;
499-
floodFile_msg_drop=0;
500-
std::string floodFile_path = floodFile_dir + "/infoLogger.flood-" + std::to_string(context.processId) + "@" + context.hostName + "-" + std::to_string((int)now);
501-
floodFile_fp=fopen(floodFile_path.c_str(),"w");
502-
if (floodFile_fp==NULL) {
503-
floodMode=2;
504-
std::string msg = "Message flood detected - further messages will be dropped (failed to create flood file " + floodFile_path + ")";
505-
pushMessage(LogWarningSupport_(1101), currentContext, msg.c_str(),1);
506-
goto case2;
479+
if (flood_protection) {
480+
481+
// update message statistics */
482+
if (now - floodStat_time_lastsec > 1) {
483+
floodStat_time_lastsec = now;
484+
floodStat_msgs_sec = 1;
485+
} else {
486+
floodStat_msgs_sec++;
487+
}
488+
if (now - floodStat_time_lastmin > 60) {
489+
floodStat_time_lastmin = now;
490+
floodStat_msgs_lastmin = floodStat_msgs_min;
491+
floodStat_msgs_min = 1;
492+
} else {
493+
floodStat_msgs_min++;
494+
}
495+
496+
// message flood prevention
497+
if (!noFlood) {
498+
switch (floodMode) {
499+
case 0:
500+
if ((floodStat_msgs_sec > flood_maxmsg_sec) || (floodStat_msgs_min > flood_maxmsg_min)) {
501+
floodFile_msg = 0;
502+
floodFile_msg_drop = 0;
503+
std::string floodFile_path = floodFile_dir + "/infoLogger.flood-" + std::to_string(context.processId) + "@" + context.hostName + "-" + std::to_string((int)now);
504+
floodFile_fp = fopen(floodFile_path.c_str(), "w");
505+
if (floodFile_fp == NULL) {
506+
floodMode = 2;
507+
std::string msg = "Message flood detected - further messages will be dropped (failed to create flood file " + floodFile_path + ")";
508+
pushMessage(LogWarningSupport_(1101), currentContext, msg.c_str(), 1);
509+
goto case2;
510+
} else {
511+
floodMode = 1;
512+
std::string msg = "Message flood detected - further messages will be stored locally in " + floodFile_path;
513+
pushMessage(LogWarningSupport_(1101), currentContext, msg.c_str(), 1);
514+
}
507515
} else {
508-
floodMode=1;
509-
std::string msg = "Message flood detected - further messages will be stored locally in " + floodFile_path;
510-
pushMessage(LogWarningSupport_(1101), currentContext, msg.c_str(),1);
516+
break;
511517
}
512-
} else {
513-
break;
514-
}
515-
case 1:
516-
if (floodStat_msgs_lastmin<flood_maxmsg_reset) {
517-
// reset flood mode
518-
break;
519-
}
520-
if (floodFile_msg>=flood_maxmsg_file) {
521-
if (floodFile_fp!=nullptr) {
522-
fclose(floodFile_fp);
523-
floodFile_fp=nullptr;
518+
case 1:
519+
if (floodStat_msgs_lastmin < flood_maxmsg_reset) {
520+
// reset flood mode
521+
break;
522+
}
523+
if (floodFile_msg >= flood_maxmsg_file) {
524+
if (floodFile_fp != nullptr) {
525+
fclose(floodFile_fp);
526+
floodFile_fp = nullptr;
527+
}
528+
pushMessage(LogWarningSupport_(1101), currentContext, "Message flood - maximum entries in local file exceeded, further messages will be dropped", 1);
529+
floodMode = 2;
530+
} else {
531+
// log to flood file
532+
fprintf(floodFile_fp, "%f\t%c\t%s\t%s\n", now, (char)options.severity, context.facility.c_str(), messageBody);
533+
fflush(floodFile_fp);
534+
floodFile_msg++;
535+
return 0;
524536
}
525-
pushMessage(LogWarningSupport_(1101), currentContext, "Message flood - maximum entries in local file exceeded, further messages will be dropped",1);
526-
floodMode=2;
527-
} else {
528-
// log to flood file
529-
fprintf(floodFile_fp,"%f\t%c\t%s\t%s\n",now,(char)options.severity,context.facility.c_str(),messageBody);
530-
fflush(floodFile_fp);
531-
floodFile_msg++;
537+
case 2:
538+
case2:
539+
if (floodStat_msgs_lastmin < flood_maxmsg_reset) {
540+
// reset flood mode
541+
break;
542+
}
543+
// drop message
544+
floodFile_msg_drop++;
532545
return 0;
533-
}
534-
case 2:
535-
case2:
536-
if (floodStat_msgs_lastmin<flood_maxmsg_reset) {
537-
// reset flood mode
538-
break;
539-
}
540-
// drop message
541-
floodFile_msg_drop++;
542-
return 0;
543-
}
544-
if (floodMode) {
545-
std::string msg = "Message flood - resuming normal operation: " + std::to_string(floodFile_msg) + " messages stored locally, " + std::to_string(floodFile_msg_drop) + " messages dropped";
546-
pushMessage(LogWarningSupport_(1102), currentContext, msg.c_str(),1);
547-
floodReset();
546+
}
547+
if (floodMode) {
548+
std::string msg = "Message flood - resuming normal operation: " + std::to_string(floodFile_msg) + " messages stored locally, " + std::to_string(floodFile_msg_drop) + " messages dropped";
549+
pushMessage(LogWarningSupport_(1102), currentContext, msg.c_str(), 1);
550+
floodReset();
551+
}
548552
}
549553
}
550554

‎src/InfoLoggerDispatchSQL.cxx‎

Lines changed: 78 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -10,10 +10,12 @@
1010

1111
#include "InfoLoggerDispatch.h"
1212
#include <mysql.h>
13+
#include <mysqld_error.h>
1314
#include "utility.h"
1415
#include "infoLoggerMessage.h"
1516
#include <unistd.h>
1617
#include <string.h>
18+
#include <Common/Timer.h>
1719

1820
#if LIBMYSQL_VERSION_ID >= 80000
1921
typedef bool my_bool;
@@ -47,10 +49,17 @@ class InfoLoggerDispatchSQLImpl
4749
std::string sql_insert;
4850

4951
unsigned long long insertCount = 0; // counter for number of queries executed
50-
unsigned long long msgDroppedCount = 0; // counter for number of messages dropped (DB unavailable, etc)
52+
unsigned long long msgDelayedCount = 0; // counter for number of messages delayed (insert failed, retry)
53+
unsigned long long msgDroppedCount = 0; // counter for number of messages dropped (insert failed, dropped)
5154

5255
int connectDB(); // function to connect to database
5356
int disconnectDB(); // disconnect/cleanup DB connection
57+
58+
int commitEnabled = 1; // flag to enable transactions
59+
int commitDebug = 0; // log transactions
60+
int commitTimeout = 1000000; // time between commits
61+
Timer commitTimer; // timer for transaction
62+
int commitNumberOfMsg; // number of messages since last commit
5463
};
5564

5665
void InfoLoggerDispatchSQLImpl::start()
@@ -114,7 +123,7 @@ InfoLoggerDispatchSQL::InfoLoggerDispatchSQL(ConfigInfoLoggerServer* config, Sim
114123
void InfoLoggerDispatchSQLImpl::stop()
115124
{
116125
disconnectDB();
117-
theLog->info("DB thread insert count = %llu, dropped msg count = %llu", insertCount, msgDroppedCount);
126+
theLog->info("DB thread insert count = %llu, delayed msg count = %llu, dropped msg count = %llu", insertCount, msgDelayedCount, msgDroppedCount);
118127
}
119128

120129
InfoLoggerDispatchSQL::~InfoLoggerDispatchSQL()
@@ -201,6 +210,9 @@ int InfoLoggerDispatchSQLImpl::connectDB()
201210
disconnectDB();
202211
return -1;
203212
}
213+
214+
// reset transactions
215+
commitNumberOfMsg = 0;
204216
}
205217

206218
return 0;
@@ -229,17 +241,46 @@ int InfoLoggerDispatchSQLImpl::customLoop()
229241
if (err) {
230242
// temporization to avoid immediate retry
231243
sleep(SQL_RETRY_CONNECT);
244+
} else if (commitEnabled) {
245+
// complete pending transactions
246+
if (commitNumberOfMsg) {
247+
if (commitTimer.isTimeout()) {
248+
if (mysql_query(db, "COMMIT")) {
249+
theLog->error("DB transaction commit failed: %s", mysql_error(db));
250+
commitEnabled = 0;
251+
} else {
252+
if (commitDebug) {
253+
theLog->info("DB commit - %d msgs", commitNumberOfMsg);
254+
}
255+
}
256+
commitNumberOfMsg = 0;
257+
}
258+
}
232259
}
260+
233261
return err;
234262
}
235263

236264
int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMessageList> lmsg)
237265
{
238-
// todo: keep message in queue on error!
266+
// procedure for dropped messages and keep count of them
267+
auto returnDroppedMessage = [&](const char* message) {
268+
// log bad message content (truncated)
269+
const int maxLen = 200;
270+
int msgLen = (int)strlen(message);
271+
theLog->error("Dropping message (%d bytes): %.*s%s", msgLen, maxLen, message, (msgLen > maxLen) ? "..." : "");
272+
msgDroppedCount++;
273+
return 0; // remove message from queue
274+
};
275+
276+
// procedure for delayed messages and keep count of them
277+
auto returnDelayedMessage = [&]() {
278+
msgDelayedCount++;
279+
return -1; // keep message in queue
280+
};
239281

240282
if (!dbIsConnected) {
241-
msgDroppedCount++;
242-
return -1;
283+
return returnDelayedMessage();
243284
}
244285

245286
infoLog_msg_t* m;
@@ -274,6 +315,21 @@ int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMe
274315
}
275316
}
276317

318+
if (commitEnabled) {
319+
if (commitNumberOfMsg == 0) {
320+
if (mysql_query(db, "START TRANSACTION")) {
321+
theLog->error("DB start transaction failed: %s", mysql_error(db));
322+
commitEnabled = 0;
323+
return returnDelayedMessage();
324+
} else {
325+
if (commitDebug) {
326+
theLog->info("DB transaction started");
327+
}
328+
}
329+
commitTimer.reset(commitTimeout);
330+
}
331+
}
332+
277333
// re-format message with multiple line - assumes it is the LAST field in the protocol
278334
for (msg = (char*)m->values[nFields - 1].value.vString; msg != NULL; msg = nl) {
279335
nl = strchr(msg, '\f');
@@ -288,27 +344,33 @@ int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMe
288344
// update bind variables
289345
if (mysql_stmt_bind_param(stmt, bind)) {
290346
theLog->error("mysql_stmt_bind() failed: %s", mysql_error(db));
291-
disconnectDB();
292-
msgDroppedCount++;
293-
return -1;
347+
theLog->error("message: %s", msg);
348+
// if can not bind, message malformed, drop it
349+
return returnDroppedMessage(msg);
294350
}
295351

296352
// Do the insertion
297353
if (mysql_stmt_execute(stmt)) {
298-
theLog->error("mysql_stmt_exec() failed: %s", mysql_error(db));
354+
theLog->error("mysql_stmt_exec() failed: (%d) %s", mysql_errno(db), mysql_error(db));
355+
// column too long
356+
if (mysql_errno(db) == ER_DATA_TOO_LONG) {
357+
return returnDroppedMessage(msg);
358+
}
299359
// retry with new connection - usually it means server was down
300360
disconnectDB();
301-
msgDroppedCount++;
302-
return -1;
361+
return returnDelayedMessage();
303362
}
304363

305364
insertCount++;
306-
/*
307-
if (insertCount%1000==0) {
308-
theLog->info("insert count = %llu",insertCount);
309-
}
310-
*/
365+
commitNumberOfMsg++;
366+
367+
if (commitDebug) {
368+
if (insertCount % 1000 == 0) {
369+
theLog->info("insert count = %llu", insertCount);
370+
}
371+
}
311372
}
312373
}
374+
// report message success, it will be removed from queue
313375
return 0;
314376
}

0 commit comments

Comments
 (0)