Merge pull request #57 from sy-c/master

v1.3.19

Author: sy-c

CMakeLists.txt

@@ -57,7 +57,7 @@ endif()
 include(ExternalProject)
 externalproject_add (Common-standalone
   GIT_REPOSITORY "https://github.com/AliceO2Group/Common.git"
-  GIT_TAG "v1.4.9"
+  GIT_TAG "v1.5.2"
   LOG_DOWNLOAD 1
   UPDATE_COMMAND ""
   PATCH_COMMAND ""

doc/README.md

@@ -276,3 +276,13 @@ achieved on CentOS 7 with e.g. (as root):
   This can be done by specifying in infoLoggerD configuration the rxSocketPath parameter, and name it with a path starting with '/', e.g. '/tmp/infoLoggerD.socket'.
   The same value has to be set for the client configuration, in the txSocketPath key.
   User is responsible to ensure that file access permissions are configured properly.
+  
+* infoLoggerD local cache
+
+  infoLoggerD stores messages in a persistent local file until messages are successfully transmitted and acknowledged by infoLoggerServer.
+  When infoLoggerServer is unavailable for a long time, messages may accumulate locally. They will all be transmitted to infoLoggerServer when available again.
+  In some cases, one may want to delete this local cache.
+  This can be configured on startup of infoLoggerD by one of the following ways:
+  - in the infoLoggerD configuration section, set: `msgQueueReset=1` (this is permanent, done on each startup of infoLoggerD, which might not be what you want)
+  - when starting infoLoggerD process from the command line (not with the systemctl service), add option: `-o msgQueueReset=1`
+  - create a file named [msgQueuePath].reset (by default, msgQueuePath=/tmp/infoLoggerD/infoLoggerD.queue), e.g. `touch /tmp/infoLoggerD/infoLoggerD.queue.reset`. This will reset the queue on next startup (by hand or with e.g. service infoLoggerD restart), and the reset file will also be deleted (which ensures cleanup is done once only).

doc/releaseNotes.md

@@ -68,3 +68,7 @@ This file describes the main feature changes for each InfoLogger released versio
 
 ## v1.3.18 - 08/01/2021
 - improved DB connection failure recovery
+
+## v1.3.19 - 22/01/2021
+- added infoLoggerServer debug options
+- added option to reset local message queue on infoLoggerD startup

src/ConfigInfoLoggerServer.cxx

@@ -23,7 +23,8 @@ void ConfigInfoLoggerServer::readFromConfigFile(ConfigFile& config)
   config.getOptionalValue<int>(INFOLOGGER_CONFIG_SECTION_NAME_SERVER ".serverPortRx", serverPortRx);
   config.getOptionalValue<int>(INFOLOGGER_CONFIG_SECTION_NAME_SERVER ".maxClientsRx", maxClientsRx);
   config.getOptionalValue<int>(INFOLOGGER_CONFIG_SECTION_NAME_SERVER ".msgQueueLengthRx", msgQueueLengthRx);
-
+  config.getOptionalValue<std::string>(INFOLOGGER_CONFIG_SECTION_NAME_SERVER ".msgDumpFile", msgDumpFile);
+      
   config.getOptionalValue<std::string>(INFOLOGGER_CONFIG_SECTION_NAME_SERVER ".dbHost", dbHost);
   config.getOptionalValue<std::string>(INFOLOGGER_CONFIG_SECTION_NAME_SERVER ".dbUser", dbUser);
   config.getOptionalValue<std::string>(INFOLOGGER_CONFIG_SECTION_NAME_SERVER ".dbPassword", dbPassword);

src/ConfigInfoLoggerServer.h

@@ -33,7 +33,8 @@ class ConfigInfoLoggerServer
   int serverPortRx = INFOLOGGER_DEFAULT_SERVER_RX_PORT; // IP port number to receive incoming messages, where infoLoggerD clients connect
   int maxClientsRx = 3000;                              // maximum number of connected infoLoggerD clients
   int msgQueueLengthRx = 10000;                         // reception queue size
-
+  std::string msgDumpFile = "";                         // a file to dump copy of all incoming messages
+  
   // settings for database connection
   std::string dbHost = "localhost";  // database host name
   std::string dbUser = "o2";         // database user name

src/InfoLoggerClient.cxx

@@ -242,9 +242,9 @@ int InfoLoggerClient::send(const char* message, unsigned int messageSize)
     // launch a thread for automatic reconnect, and buffer (reasonably) messages until then
     reconnectThreadStart();
   }
-  if (messageBuffer.size()<cfg.maxMessagesBuffered) {
+  if ((int)messageBuffer.size()<cfg.maxMessagesBuffered) {
     messageBuffer.push(message);
-    if (messageBuffer.size()==cfg.maxMessagesBuffered) {
+    if ((int)messageBuffer.size()==cfg.maxMessagesBuffered) {
       log.warning("Max buffer size reached, next messages will be lost until reconnect");
     }
   }

src/infoLoggerAdminDB.cxx

@@ -221,7 +221,7 @@ int main(int argc, char* argv[])
     unsigned, username varchar(32), `system` varchar(32), facility varchar(32), detector varchar(32), `partition` varchar(32), run int unsigned, errcode int unsigned, \
     errline smallint unsigned, errsource varchar(32), message text, index ix_severity(severity), index ix_level(level), index ix_timestamp(timestamp), index \
     ix_hostname(hostname(14)), index ix_rolename(rolename(20)), index ix_system(`system`(3)), index ix_facility(facility(20)), index ix_detector(detector(8)), index \
-    ix_partition(`partition`(10)), index ix_run(run), index ix_errcode(errcode), index ix_errline(errline), index ix_errsource(errsource(20))) ENGINE=MyISAM";
+    ix_partition(`partition`(10)), index ix_run(run), index ix_errcode(errcode), index ix_errline(errline), index ix_errsource(errsource(20)))";
 
   if (optPartitioning) {
     log.info("Using partitioning");

src/infoLoggerD.cxx

@@ -71,8 +71,9 @@ class ConfigInfoLoggerD
   // settings for remote infoLoggerServer access
   std::string serverHost = "localhost";                // IP name to connect infoLoggerServer
   int serverPort = INFOLOGGER_DEFAULT_SERVER_RX_PORT;  // IP port number to connect infoLoggerServer
-  int msgQueueLength = 1000;                           // transmission queue size
-  std::string msgQueuePath = "/tmp/infoLoggerD.queue"; // path to temp file storing messages
+  int msgQueueLength = 10000;                           // transmission queue size
+  std::string msgQueuePath = localLogDirectory + "/infoLoggerD.queue"; // path to temp file storing messages
+  int msgQueueReset = 0;                                               // when set, existing temp file is cleared (and pending messages lost)
   std::string clientName = "infoLoggerD";              // name identifying client to infoLoggerServer
   int isProxy = 0;                                     // flag set to allow infoLoggerD to be a transport proxy to infoLoggerServer
 
@@ -101,6 +102,7 @@ void ConfigInfoLoggerD::readFromConfigFile(ConfigFile& config)
   config.getOptionalValue<int>(INFOLOGGER_CONFIG_SECTION_NAME_INFOLOGGERD ".serverPort", serverPort);
   config.getOptionalValue<int>(INFOLOGGER_CONFIG_SECTION_NAME_INFOLOGGERD ".msgQueueLength", msgQueueLength);
   config.getOptionalValue<std::string>(INFOLOGGER_CONFIG_SECTION_NAME_INFOLOGGERD ".msgQueuePath", msgQueuePath);
+  config.getOptionalValue<int>(INFOLOGGER_CONFIG_SECTION_NAME_INFOLOGGERD ".msgQueueReset", msgQueueReset);
   config.getOptionalValue<std::string>(INFOLOGGER_CONFIG_SECTION_NAME_INFOLOGGERD ".clientName", clientName);
   config.getOptionalValue<int>(INFOLOGGER_CONFIG_SECTION_NAME_INFOLOGGERD ".isProxy", isProxy);
 
@@ -213,7 +215,13 @@ class InfoLoggerD : public Daemon
   FILE* logOutput = nullptr; // handle to local log file where to copy incoming messages, if configured to do so
 };
 
-InfoLoggerD::InfoLoggerD(int argc, char* argv[]) : Daemon(argc, argv)
+// list of extra keys accepted on the command line (-o key=value entries)
+#define EXTRA_OPTIONS \
+  {                   \
+    "msgQueueReset"   \
+  }
+
+InfoLoggerD::InfoLoggerD(int argc, char* argv[]) : Daemon(argc, argv, nullptr, EXTRA_OPTIONS)
 {
   if (isOk()) { // proceed only if base daemon init was a success
 
@@ -227,6 +235,13 @@ InfoLoggerD::InfoLoggerD(int argc, char* argv[]) : Daemon(argc, argv)
       // redirect legacy simplelog interface to SimpleLog
       setSimpleLog(&log);
 
+      // retrieve configuration parameters from command line
+      for (auto const& o : execOptions) {
+        if (o.key == "msgQueueReset") {
+          configInfoLoggerD.msgQueueReset = atoi(o.value.c_str());
+        }
+      }
+
       // retrieve configuration parameters from config file
       configInfoLoggerD.readFromConfigFile(config);
 
@@ -309,6 +324,38 @@ InfoLoggerD::InfoLoggerD(int argc, char* argv[]) : Daemon(argc, argv)
           cfgCx.proxy_state = TR_PROXY_CAN_BE_PROXY;
         }
 
+        // check message queue
+        struct stat queueInfo;
+        std::string msgQueuePathFifo = configInfoLoggerD.msgQueuePath + ".fifo";
+        log.info("Checking message queue file %s", msgQueuePathFifo.c_str());
+        if (lstat(msgQueuePathFifo.c_str(), &queueInfo) == 0) {
+          if (S_ISREG(queueInfo.st_mode)) {
+            log.info("Pending messages = %ld bytes", (long)queueInfo.st_size);
+          } else {
+            log.error("This is not a regular file");
+          }
+        } else {
+          log.info("No pending messages");
+        }
+
+        // if the queue.reset file exists, remove it and reset queue
+        // (so it's done once only, on this startup)
+        std::string msgQueuePathReset = configInfoLoggerD.msgQueuePath + ".reset";
+        if (lstat(msgQueuePathReset.c_str(), &queueInfo) == 0) {
+          if (S_ISREG(queueInfo.st_mode)) {
+            log.info("Detected file %s, forcing queue reset", msgQueuePathReset.c_str());
+            remove(msgQueuePathReset.c_str());
+            configInfoLoggerD.msgQueueReset = 1;
+          }
+        }
+
+        if (configInfoLoggerD.msgQueueReset) {
+          log.info("Clearing queue");
+          if (remove(msgQueuePathFifo.c_str())) {
+            log.info("Failed to delete %s", msgQueuePathFifo.c_str());
+          }
+        }
+
         hCx = TR_client_start(&cfgCx);
         if (hCx == nullptr) {
           throw __LINE__;

src/infoLoggerServer.cxx

@@ -87,6 +87,8 @@ class InfoLoggerServer : public Daemon
   unsigned int dbRoundRobinIx = 0;
 
   unsigned long long msgCount = 0;
+  
+  FILE *msgDump = nullptr;
 };
 
 InfoLoggerServer::InfoLoggerServer(int argc, char* argv[]) : Daemon(argc, argv)
@@ -111,6 +113,19 @@ InfoLoggerServer::InfoLoggerServer(int argc, char* argv[]) : Daemon(argc, argv)
         throw __LINE__;
       }
 
+      // create message dump (copy of all incoming messages)
+      if (configInfoLoggerServer.msgDumpFile.size()) {
+        time_t t = time(NULL);
+	std::string f = configInfoLoggerServer.msgDumpFile.c_str();
+	f += std::to_string(t) + ".log";
+        msgDump = fopen(f.c_str(),"wb");
+	if (msgDump != nullptr) {
+          log.info("Dumping copy of incoming messages to %s", f.c_str());
+	} else {
+	  log.error("Failed to create message dump file %s", f.c_str());
+	}
+      }
+
       // create dispatch engines
       try {
         //dispatchEngines.push_back(std::make_unique<InfoLoggerDispatchPrint>(&log));
@@ -147,6 +162,9 @@ InfoLoggerServer::~InfoLoggerServer()
     }
 
     log.info("Received %llu messages", msgCount);
+    if (msgDump != nullptr) {
+      fclose(msgDump);
+    }
   }
 }
 
@@ -164,6 +182,17 @@ Daemon::LoopStatus InfoLoggerServer::doLoop()
     //fflush(stdout);
     //TR_file_dump(newFile);
 
+    // dump new message
+    if (msgDump != nullptr) {
+      TR_blob* b;
+      for (b = newFile->first; b != NULL; b = b->next) {
+        fprintf(msgDump, "*** begin: %d bytes\n", (int)b->size);
+	fwrite(b->value, b->size, 1, msgDump);
+	fprintf(msgDump, "\n*** end\n");
+	fflush(msgDump);
+      }
+    }
+
     // decode raw message
     std::shared_ptr<InfoLoggerMessageList> msgList = nullptr;
     try {