drop data too long messages

Author: sy-c

src/InfoLoggerDispatchSQL.cxx

@@ -10,6 +10,7 @@
 
 #include "InfoLoggerDispatch.h"
 #include <mysql.h>
+#include <mysqld_error.h>
 #include "utility.h"
 #include "infoLoggerMessage.h"
 #include <unistd.h>
@@ -48,7 +49,8 @@ class InfoLoggerDispatchSQLImpl
   std::string sql_insert;
 
   unsigned long long insertCount = 0;     // counter for number of queries executed
-  unsigned long long msgDroppedCount = 0; // counter for number of messages dropped (DB unavailable, etc)
+  unsigned long long msgDelayedCount = 0; // counter for number of messages delayed (insert failed, retry)
+  unsigned long long msgDroppedCount = 0; // counter for number of messages dropped (insert failed, dropped)
 
   int connectDB(); // function to connect to database
   int disconnectDB(); // disconnect/cleanup DB connection
@@ -121,7 +123,7 @@ InfoLoggerDispatchSQL::InfoLoggerDispatchSQL(ConfigInfoLoggerServer* config, Sim
 void InfoLoggerDispatchSQLImpl::stop()
 {
   disconnectDB();
-  theLog->info("DB thread insert count = %llu, dropped msg count = %llu", insertCount, msgDroppedCount);
+  theLog->info("DB thread insert count = %llu, delayed msg count = %llu, dropped msg count = %llu", insertCount, msgDelayedCount, msgDroppedCount);
 }
 
 InfoLoggerDispatchSQL::~InfoLoggerDispatchSQL()
@@ -261,11 +263,24 @@ int InfoLoggerDispatchSQLImpl::customLoop()
 
 int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMessageList> lmsg)
 {
-  // todo: keep message in queue on error!
-
-  if (!dbIsConnected) {
+  // procedure for dropped messages and keep count of them
+  auto returnDroppedMessage = [&](const char* message) {
+    // log bad message content (truncated)
+    const int maxLen = 200;
+    int msgLen = (int)strlen(message);
+    theLog->error("Dropping message (%d bytes): %.*s%s", msgLen, maxLen, message, (msgLen>maxLen) ? "..." : "");
     msgDroppedCount++;
-    return -1;
+    return 0; // remove message from queue
+  };
+  
+  // procedure for delayed messages and keep count of them
+  auto returnDelayedMessage = [&]() {
+    msgDelayedCount++;
+    return -1; // keep message in queue
+  };
+    
+  if (!dbIsConnected) {
+    return returnDelayedMessage();
   }
 
   infoLog_msg_t* m;
@@ -305,7 +320,7 @@ int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMe
         if (mysql_query(db, "START TRANSACTION")) {
           theLog->error("DB start transaction failed: %s", mysql_error(db));
           commitEnabled = 0;
-          return -1;
+	  return returnDelayedMessage();
         } else {
           if (commitDebug) {
             theLog->info("DB transaction started");
@@ -329,18 +344,21 @@ 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));
-        disconnectDB();
-        msgDroppedCount++;
-        return -1;
+	theLog->error("message: %s", msg);
+        // if can not bind, message malformed, drop it
+        return returnDroppedMessage(msg);
       }
 
       // Do the insertion
       if (mysql_stmt_execute(stmt)) {
-        theLog->error("mysql_stmt_exec() failed: %s", mysql_error(db));
+        theLog->error("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);
+	}
         // retry with new connection - usually it means server was down
         disconnectDB();
-        msgDroppedCount++;
-        return -1;
+        return returnDelayedMessage();
       }
 
       insertCount++;
@@ -353,5 +371,6 @@ int InfoLoggerDispatchSQLImpl::customMessageProcess(std::shared_ptr<InfoLoggerMe
       }
     }
   }
+  // report message success, it will be removed from queue
   return 0;
 }