Merge pull request #92 from sy-c/master

v2.7.2

Author: sy-c

CMakeLists.txt

@@ -22,7 +22,7 @@ endif()
 # global compilation options
 enable_language(C)
 enable_language(CXX)
-set(CMAKE_CXX_STANDARD 14)
+set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 set(CMAKE_CXX_EXTENSIONS OFF)
 set(CMAKE_POSITION_INDEPENDENT_CODE ON)
@@ -65,7 +65,7 @@ endif()
 include(ExternalProject)
 externalproject_add (Common-standalone
   GIT_REPOSITORY "https://github.com/AliceO2Group/Common.git"
-  GIT_TAG "v1.6.1"
+  GIT_TAG "v1.6.3"
   LOG_DOWNLOAD 1
   UPDATE_COMMAND ""
   PATCH_COMMAND ""

doc/releaseNotes.md

@@ -150,3 +150,12 @@ This file describes the main feature changes for each InfoLogger released versio
 - o2-infologger-adminDb:
   - added interactive confirmation prompt when clearing/deleting non-empty tables, unless option '-o noWarning' is set.
   - added command '-c status' to display status information about infoLogger database content (number of tables, number of rows, data size).
+
+## v2.7.2 - 26/09/2024
+- o2-infologger-daemon: improved handling of large number of connections.
+  - fixed message flood of "accept() failed" errors when reaching maximum number allowed file descriptors. A single message is now reported when condition occurs, and then when it is cleared. Reaching system limits of max number of descriptors is now handled gracefuly and recovers fine. NB: loop of accept() failures was still occuring even after remote process dead, because fds content are buffered to avoid message loss.
+  - upgraded to o2-Common v1.6.3, which fixes unclosed file descriptor when rotating log files (which was contributing to trigger issue above, increasing number of fds in use).
+  - added startup checks to verify compatibility of rxMaxConnections with max number of file descriptors allowed by system. If needed, tries to increase the limit. If it does not work, rxMaxConnections is automatically reduced to a lower value.
+  - increased default value of rxMaxConnections from 1024 to 2048. This is the number of concurrent clients allowed to connect to o2-infologger-daemon. Can be changed in configuration file.
+- o2-infologger-server: improved logging per thread/database.
+- o2-infologger-newdb: added option to skip user creation

newMysql.sh

@@ -27,6 +27,9 @@ FORCE_REDO=0
 # if set, SQL queries & output printed
 SQL_DEBUG=0
 
+# if set, recreate user accounts unless access to db already ok (existing ones will be deleted, including their privileges)
+SQL_RECREATE_USER=1
+
 # where are we running now
 HERE=`hostname -f`
 
@@ -240,9 +243,11 @@ for CONFIG in "${EXTRA_CONFIG[@]}"; do
   fi
 
   for QHOST in "%" "localhost" "${HERE}"; do
-    mysqlExecute "drop user \"${EXTRA_USER[$CONFIG]}\"@\"${QHOST}\";"
-    mysqlExecute "create user \"${EXTRA_USER[$CONFIG]}\"@\"${QHOST}\";"
-    mysqlExecute "set password for  \"${EXTRA_USER[$CONFIG]}\"@\"${QHOST}\" = PASSWORD(\"${EXTRA_PWD[$CONFIG]}\");"
+    if [ "$SQL_RECREATE_USER" == "1" ]; then
+      mysqlExecute "drop user \"${EXTRA_USER[$CONFIG]}\"@\"${QHOST}\";"
+      mysqlExecute "create user \"${EXTRA_USER[$CONFIG]}\"@\"${QHOST}\";"
+      mysqlExecute "set password for  \"${EXTRA_USER[$CONFIG]}\"@\"${QHOST}\" = PASSWORD(\"${EXTRA_PWD[$CONFIG]}\");"
+    fi
     mysqlExecute "grant ${EXTRA_PRIVILEGE[$CONFIG]} on $INFOLOGGER_DB_NAME.* to \"${EXTRA_USER[$CONFIG]}\"@\"${QHOST}\";"
   done
 done

src/InfoLoggerDispatch.cxx

@@ -15,8 +15,9 @@
 // class InfoLoggerDispatch implementation
 ///////////////////////////////////////////
 
-InfoLoggerDispatch::InfoLoggerDispatch(ConfigInfoLoggerServer* vConfig, SimpleLog* vLog)
+InfoLoggerDispatch::InfoLoggerDispatch(ConfigInfoLoggerServer* vConfig, SimpleLog* vLog, std::string vLogPrefix)
 {
+  logPrefix = vLogPrefix;
   input = std::make_unique<AliceO2::Common::Fifo<std::shared_ptr<InfoLoggerMessageList>>>(vConfig->dbDispatchQueueSize);
   if (vLog != NULL) {
     theLog = vLog;
@@ -87,6 +88,33 @@ int InfoLoggerDispatchPrint::customMessageProcess(std::shared_ptr<InfoLoggerMess
   return 0;
 }
 
+void InfoLoggerDispatch::logInfo(const char* message, ...) {
+  char buffer[1024] = "";
+  va_list ap;
+  va_start(ap, message);
+  vsnprintf(buffer,sizeof(buffer), message, ap);
+  va_end(ap);
+  theLog->info("%s%s", logPrefix.c_str(), buffer);
+}
+
+void InfoLoggerDispatch::logError(const char* message, ...) {
+  char buffer[1024] = "";
+  va_list ap;
+  va_start(ap, message);
+  vsnprintf(buffer,sizeof(buffer), message, ap);
+  va_end(ap);
+  theLog->error("%s%s", logPrefix.c_str(), buffer);
+}
+
+void InfoLoggerDispatch::logWarning(const char* message, ...) {
+  char buffer[1024] = "";
+  va_list ap;
+  va_start(ap, message);
+  vsnprintf(buffer,sizeof(buffer), message, ap);
+  va_end(ap);
+  theLog->warning("%s%s", logPrefix.c_str(), buffer);
+}
+
 /////////////////////////////////////////////////
 // end of class InfoLoggerDispatch implementation
 /////////////////////////////////////////////////

src/InfoLoggerDispatch.h

@@ -31,7 +31,7 @@ class InfoLoggerDispatch
 {
 
  public:
-  InfoLoggerDispatch(ConfigInfoLoggerServer* theConfig = NULL, SimpleLog* theLog = NULL);
+  InfoLoggerDispatch(ConfigInfoLoggerServer* theConfig = NULL, SimpleLog* theLog = NULL, std::string theLogPrefix = "");
   virtual ~InfoLoggerDispatch();
 
   // todo: define settings: non-blocking, etc
@@ -46,11 +46,16 @@ class InfoLoggerDispatch
 
   //    virtual customDispatch...
 
+  void logInfo(const char* message, ...) __attribute__((format(printf, 2, 3)));
+  void logError(const char* message, ...) __attribute__((format(printf, 2, 3)));
+  void logWarning(const char* message, ...) __attribute__((format(printf, 2, 3)));
+
  protected:
   std::unique_ptr<AliceO2::Common::Fifo<std::shared_ptr<InfoLoggerMessageList>>> input;
   std::unique_ptr<AliceO2::Common::Thread> dispatchThread;
   SimpleLog* theLog;
   SimpleLog defaultLog;
+  std::string logPrefix; // a string appended to each log
 
   ConfigInfoLoggerServer* theConfig;
 
@@ -82,7 +87,7 @@ class InfoLoggerDispatchSQLImpl;
 class InfoLoggerDispatchSQL : public InfoLoggerDispatch
 {
  public:
-  InfoLoggerDispatchSQL(ConfigInfoLoggerServer* theConfig, SimpleLog* theLog);
+  InfoLoggerDispatchSQL(ConfigInfoLoggerServer* theConfig, SimpleLog* theLog, std::string prefix);
   ~InfoLoggerDispatchSQL();
   int customMessageProcess(std::shared_ptr<InfoLoggerMessageList> msg);
   int customLoop();

src/InfoLoggerDispatchSQL.cxx

@@ -28,8 +28,8 @@ typedef bool my_bool;
 class InfoLoggerDispatchSQLImpl
 {
  public:
-  SimpleLog* theLog;                 // handle for logging
   ConfigInfoLoggerServer* theConfig; // struct with config parameters
+  InfoLoggerDispatchSQL* parent;
 
   void start();
   void stop();
@@ -67,7 +67,7 @@ void InfoLoggerDispatchSQLImpl::start()
 {
 
   // log DB params
-  theLog->info("Using DB %s@%s:%s", theConfig->dbUser.c_str(), theConfig->dbHost.c_str(), theConfig->dbName.c_str());
+  parent->logInfo("Using DB %s@%s:%s", theConfig->dbUser.c_str(), theConfig->dbHost.c_str(), theConfig->dbName.c_str());
 
   // prepare insert query from 1st protocol definition
   // e.g. INSERT INTO messages(severity,level,timestamp,hostname,rolename,pid,username,system,facility,detector,partition,run,errcode,errLine,errsource) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) */
@@ -101,20 +101,20 @@ void InfoLoggerDispatchSQLImpl::start()
     }
   }
   sql_insert += ")";
-  theLog->info("insert query = %s", sql_insert.c_str());
+  parent->logInfo("insert query = %s", sql_insert.c_str());
   if (errLine) {
-    theLog->error("Failed to initialize db query: error %d", errLine);
+    parent->logError("Failed to initialize db query: error %d", errLine);
   }
 
   // try to connect DB
   // done automatically in customloop
 }
 
-InfoLoggerDispatchSQL::InfoLoggerDispatchSQL(ConfigInfoLoggerServer* config, SimpleLog* log) : InfoLoggerDispatch(config, log)
+InfoLoggerDispatchSQL::InfoLoggerDispatchSQL(ConfigInfoLoggerServer* config, SimpleLog* log, std::string prefix) : InfoLoggerDispatch(config, log, prefix)
 {
   dPtr = std::make_unique<InfoLoggerDispatchSQLImpl>();
-  dPtr->theLog = log;
   dPtr->theConfig = config;
+  dPtr->parent = this;
   dPtr->start();
 
   // enable customloop callback
@@ -124,7 +124,7 @@ InfoLoggerDispatchSQL::InfoLoggerDispatchSQL(ConfigInfoLoggerServer* config, Sim
 void InfoLoggerDispatchSQLImpl::stop()
 {
   disconnectDB();
-  theLog->info("DB thread insert count = %llu, delayed msg count = %llu, dropped msg count = %llu", insertCount, msgDelayedCount, msgDroppedCount);
+  parent->logInfo("DB thread insert count = %llu, delayed msg count = %llu, dropped msg count = %llu", insertCount, msgDelayedCount, msgDroppedCount);
 }
 
 InfoLoggerDispatchSQL::~InfoLoggerDispatchSQL()
@@ -156,18 +156,18 @@ int InfoLoggerDispatchSQLImpl::connectDB()
       // init mysql handle
       db = mysql_init(db);
       if (db == NULL) {
-        theLog->error("mysql_init() failed");
+        parent->logError("mysql_init() failed");
         return 1;
       }
     }
-
+    parent->logInfo("DB connecting: %s@%s:%s", theConfig->dbUser.c_str(), theConfig->dbHost.c_str(), theConfig->dbName.c_str());
     if (mysql_real_connect(db, theConfig->dbHost.c_str(), theConfig->dbUser.c_str(), theConfig->dbPassword.c_str(), theConfig->dbName.c_str(), 0, NULL, 0)) {
-      theLog->info("DB connected");
+      parent->logInfo("DB connected");
       dbIsConnected = 1;
       dbConnectTrials = 0;
     } else {
       if (dbConnectTrials == 0) { // log only first attempt
-        theLog->error("DB connection failed: %s", mysql_error(db));
+        parent->logError("DB connection failed: %s", mysql_error(db));
       }
       dbConnectTrials++;
       return 1;
@@ -176,13 +176,13 @@ int InfoLoggerDispatchSQLImpl::connectDB()
     // create prepared insert statement
     stmt = mysql_stmt_init(db);
     if (stmt == NULL) {
-      theLog->error("mysql_stmt_init() failed: %s", mysql_error(db));
+      parent->logError("mysql_stmt_init() failed: %s", mysql_error(db));
       disconnectDB();
       return -1;
     }
 
     if (mysql_stmt_prepare(stmt, sql_insert.c_str(), sql_insert.length())) {
-      theLog->error("mysql_stmt_prepare() failed: %s", mysql_error(db));
+      parent->logError("mysql_stmt_prepare() failed: %s", mysql_error(db));
       disconnectDB();
       return -1;
     }
@@ -202,7 +202,7 @@ int InfoLoggerDispatchSQLImpl::connectDB()
           bind[i].buffer_type = MYSQL_TYPE_DOUBLE;
           break;
         default:
-          theLog->error("undefined field type %d", protocols[0].fields[i].type);
+          parent->logError("undefined field type %d", protocols[0].fields[i].type);
           errline = __LINE__;
           break;
       }
@@ -230,7 +230,7 @@ int InfoLoggerDispatchSQLImpl::disconnectDB()
     db = NULL;
   }
   if (dbIsConnected) {    
-    theLog->info("DB disconnected");
+    parent->logInfo("DB disconnected");
   }
   dbIsConnected = 0;
   return 0;
@@ -247,11 +247,11 @@ int InfoLoggerDispatchSQLImpl::customLoop()
     if (commitNumberOfMsg) {
       if (commitTimer.isTimeout()) {
         if (mysql_query(db, "COMMIT")) {
-          theLog->error("DB transaction commit failed: %s", mysql_error(db));
+          parent->logError("DB transaction commit failed: %s", mysql_error(db));
           commitEnabled = 0;
         } else {
           if (commitDebug) {
-            theLog->info("DB commit - %d msgs", commitNumberOfMsg);
+            parent->logInfo("DB commit - %d msgs", commitNumberOfMsg);
           }
         }
         commitNumberOfMsg = 0;
@@ -299,8 +299,8 @@ int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMe
     }
 
     int msgLen = (int)strlen(message);
-    theLog->error("Dropping message (%d bytes): %.*s%s", msgLen, maxLen, message, (msgLen > maxLen) ? "..." : "");
-    theLog->error("                 %s", logDetails.c_str());
+    parent->logError("Dropping message (%d bytes): %.*s%s", msgLen, maxLen, message, (msgLen > maxLen) ? "..." : "");
+    parent->logError("                 %s", logDetails.c_str());
     msgDroppedCount++;
     return 0; // remove message from queue
   };
@@ -350,12 +350,12 @@ int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMe
     if (commitEnabled) {
       if (commitNumberOfMsg == 0) {
         if (mysql_query(db, "START TRANSACTION")) {
-          theLog->error("DB start transaction failed: %s", mysql_error(db));
+          parent->logError("DB start transaction failed: %s", mysql_error(db));
           commitEnabled = 0;
           return returnDelayedMessage();
         } else {
           if (commitDebug) {
-            theLog->info("DB transaction started");
+            parent->logInfo("DB transaction started");
           }
         }
         commitTimer.reset(commitTimeout);
@@ -375,15 +375,15 @@ int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMe
 
       // update bind variables
       if (mysql_stmt_bind_param(stmt, bind)) {
-        theLog->error("mysql_stmt_bind() failed: %s", mysql_error(db));
-        theLog->error("message: %s", msg);
+        parent->logError("mysql_stmt_bind() failed: %s", mysql_error(db));
+        parent->logError("message: %s", msg);
         // if can not bind, message malformed, drop it
         return returnDroppedMessage(msg, m);
       }
 
       // Do the insertion
       if (mysql_stmt_execute(stmt)) {
-        theLog->error("mysql_stmt_exec() failed: (%d) %s", mysql_errno(db), mysql_error(db));
+        parent->logError("mysql_stmt_exec() failed: (%d) %s", mysql_errno(db), mysql_error(db));
         // column too long
         if (mysql_errno(db) == ER_DATA_TOO_LONG) {
           return returnDroppedMessage(msg, m);
@@ -398,7 +398,7 @@ int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMe
 
       if (commitDebug) {
         if (insertCount % 1000 == 0) {
-          theLog->info("insert count = %llu", insertCount);
+          parent->logInfo("insert count = %llu", insertCount);
         }
       }
     }

src/infoLoggerAdminDB.cxx

@@ -144,7 +144,7 @@ int main(int argc, char* argv[])
 
   // connect database
   if (mysql_real_connect(&db, dbHost.c_str(), dbUser.c_str(), dbPwd.c_str(), dbName.c_str(), 0, NULL, 0) != 0) {
-    log.info("Database connected");
+    log.info("Database %s @ %s connected", dbName.c_str(), dbHost.c_str());
   } else {
     log.error("Failed to connect database : %s", mysql_error(&db));
     return -1;
@@ -171,7 +171,7 @@ int main(int argc, char* argv[])
 
   // execute command(s)
   if (optStatus) {
-    std::string sqlQuery = "select TABLE_NAME,TABLE_ROWS,DATA_LENGTH from information_schema.TABLES where table_name like '" INFOLOGGER_TABLE_MESSAGES "%' order by table_name";
+    std::string sqlQuery = "select TABLE_NAME,TABLE_ROWS,DATA_LENGTH from information_schema.TABLES where table_name like '" INFOLOGGER_TABLE_MESSAGES "%' and table_schema like '" + dbName + "' order by table_name";
     if (mysql_query(&db, sqlQuery.c_str())) {
       log.error("Failed to execute %s\n%s", sqlQuery.c_str(), mysql_error(&db));
       return -1;