MXS-1734 Add statement dumping mechanism

With the configuration entry

   retain_last_statements=<unsigned>

or the debug flag '--debug=retain-last-statements=<unsigned>',
MaxScale will store the specified number of last statements
for each session. By calling

    session_dump_statements(session);

MaxScale will dump the last statements as NOTICE messages.
For debugging purposes.

include/maxscale/config.h

@@ -142,6 +142,7 @@ extern const char CN_QUERY_RETRY_TIMEOUT[];
 extern const char CN_RELATIONSHIPS[];
 extern const char CN_LINKS[];
 extern const char CN_REQUIRED[];
+extern const char CN_RETAIN_LAST_STATEMENTS[];
 extern const char CN_RETRY_ON_FAILURE[];
 extern const char CN_ROUTER[];
 extern const char CN_ROUTER_DIAGNOSTICS[];

include/maxscale/session.h

@@ -27,6 +27,14 @@
 #include <maxscale/spinlock.h>
 #include <maxscale/jansson.h>
 
+#if defined(__cplusplus)
+#include <deque>
+#include <vector>
+typedef std::deque<std::vector<uint8_t> > SessionStmtQueue;
+#else
+typedef void SessionStmtQueue;
+#endif
+
 MXS_BEGIN_DECLS
 
 struct dcb;
@@ -153,6 +161,7 @@ typedef struct session
         const struct server *target; /**< Where the statement was sent */
     } stmt;  /**< Current statement being executed */
     bool qualifies_for_pooling; /**< Whether this session qualifies for the connection pool */
+    SessionStmtQueue*      last_statements;  /*< The N last statements by the client */
     skygw_chk_t     ses_chk_tail;
 } MXS_SESSION;
 
@@ -481,4 +490,27 @@ MXS_SESSION* session_get_current();
  **/
 uint64_t session_get_current_id();
 
+/**
+ * @brief Specify how many statements each session should retain for
+ *        debugging purposes.
+ *
+ * @param n  The number of statements.
+ */
+void session_retain_last_statements(uint32_t n);
+
+/**
+ * @brief Retain provided statement, if configured to do so.
+ *
+ * @param session  The session.
+ * @param buffer   Buffer assumed to contain a full statement.
+ */
+void session_retain_statement(MXS_SESSION* session, GWBUF* buffer);
+
+/**
+ * @brief Dump the last statements, if statements have been retained.
+ *
+ * @param session  The session.
+ */
+void session_dump_statements(MXS_SESSION* pSession);
+
 MXS_END_DECLS

server/core/config.cc

@@ -121,6 +121,7 @@ const char CN_RELATIONSHIPS[]                 = "relationships";
 const char CN_LINKS[]                         = "links";
 const char CN_LOCAL_ADDRESS[]                 = "local_address";
 const char CN_REQUIRED[]                      = "required";
+const char CN_RETAIN_LAST_STATEMENTS[]        = "retain_last_statements";
 const char CN_RETRY_ON_FAILURE[]              = "retry_on_failure";
 const char CN_ROUTER[]                        = "router";
 const char CN_ROUTER_DIAGNOSTICS[]            = "router_diagnostics";
@@ -1733,6 +1734,19 @@ handle_global_item(const char *name, const char *value)
             gateway.users_refresh_time = USERS_REFRESH_TIME_DEFAULT;
         }
     }
+    else if (strcmp(name, CN_RETAIN_LAST_STATEMENTS) == 0)
+    {
+        char* endptr;
+        int intval = strtol(value, &endptr, 0);
+        if (*endptr == '\0' && intval >= 0)
+        {
+            session_retain_last_statements(intval);
+        }
+        else
+        {
+            MXS_ERROR("Invalid value for '%s': %s", CN_RETAIN_LAST_STATEMENTS, value);
+        }
+    }
     else
     {
         for (i = 0; lognames[i].name; i++)

server/core/gateway.cc

@@ -192,6 +192,7 @@ static void enable_module_unloading(const char* arg);
 static void enable_statement_logging(const char* arg);
 static void disable_statement_logging(const char* arg);
 static void redirect_output_to_file(const char* arg);
+static void retain_last_statements(const char* arg);
 static bool user_is_acceptable(const char* specified_user);
 static bool init_sqlite3();
 
@@ -228,6 +229,11 @@ const DEBUG_ARGUMENT debug_arguments[] =
         "disable-statement-logging", disable_statement_logging,
         "disable the logging of SQL statements sent by MaxScale to the servers"
     },
+    {
+        "retain-last-statements", retain_last_statements,
+        "retain the specified number of client statements\n"
+        SPACER "allows statements to be logged in problem situations"
+    },
     {NULL, NULL, NULL}
 };
 
@@ -3206,6 +3212,18 @@ static void disable_statement_logging(const char* arg)
     mxs_mysql_set_log_statements(false);
 }
 
+static void retain_last_statements(const char* arg)
+{
+    int n = atoi(arg);
+
+    if (n < 0)
+    {
+        n = 0;
+    }
+
+    session_retain_last_statements(n);
+}
+
 static void redirect_output_to_file(const char* arg)
 {
     if (arg)

server/core/session.cc

@@ -36,6 +36,7 @@
 #include <maxscale/spinlock.h>
 #include <maxscale/utils.h>
 #include <maxscale/json_api.h>
+#include <maxscale/protocol/mysql.h>
 
 #include "internal/dcb.h"
 #include "internal/session.h"
@@ -51,6 +52,8 @@ using std::stringstream;
  */
 static uint64_t next_session_id = 1;
 
+static uint32_t retain_last_statements = 0;
+
 static struct session session_dummy_struct;
 
 static void session_initialize(void *session);
@@ -86,6 +89,7 @@ session_initialize(MXS_SESSION *session)
 
     session->ses_chk_top = CHK_NUM_SESSION;
     session->state = SESSION_STATE_ALLOC;
+    session->last_statements = new SessionStmtQueue;
     session->ses_chk_tail = CHK_NUM_SESSION;
 }
 
@@ -389,6 +393,7 @@ static void
 session_final_free(MXS_SESSION *session)
 {
     gwbuf_free(session->stmt.buffer);
+    delete session->last_statements;
     MXS_FREE(session);
 }
 
@@ -1105,3 +1110,84 @@ uint64_t session_get_current_id()
 
     return session ? session->ses_id : 0;
 }
+
+void session_retain_last_statements(uint32_t n)
+{
+    retain_last_statements = n;
+}
+
+void session_retain_statement(MXS_SESSION* pSession, GWBUF* pBuffer)
+{
+    if (retain_last_statements)
+    {
+        size_t len = gwbuf_length(pBuffer);
+
+        if (len > MYSQL_HEADER_LEN)
+        {
+            uint8_t header[MYSQL_HEADER_LEN + 1];
+            uint8_t* pHeader = NULL;
+
+            if (GWBUF_LENGTH(pBuffer) > MYSQL_HEADER_LEN)
+            {
+                pHeader = GWBUF_DATA(pBuffer);
+            }
+            else
+            {
+                gwbuf_copy_data(pBuffer, 0, MYSQL_HEADER_LEN + 1, header);
+                pHeader = header;
+            }
+
+            if (MYSQL_GET_COMMAND(pHeader) == MXS_COM_QUERY)
+            {
+                ss_dassert(pSession->last_statements->size() <= retain_last_statements);
+
+                if (pSession->last_statements->size() == retain_last_statements)
+                {
+                    pSession->last_statements->pop_back();
+                }
+
+                std::vector<uint8_t> stmt(len - MYSQL_HEADER_LEN - 1);
+                gwbuf_copy_data(pBuffer, MYSQL_HEADER_LEN + 1, len - (MYSQL_HEADER_LEN + 1), &stmt.front());
+
+                pSession->last_statements->push_front(stmt);
+            }
+        }
+    }
+}
+
+void session_dump_statements(MXS_SESSION* pSession)
+{
+    if (retain_last_statements)
+    {
+        int n = pSession->last_statements->size();
+
+        uint64_t id = session_get_current_id();
+
+        if ((id != 0) && (id != pSession->ses_id))
+        {
+            MXS_WARNING("Current session is %" PRIu64 ", yet statements are dumped for %" PRIu64 ". "
+                        "The session id in the subsequent dumped statements is the wrong one.",
+                        id, pSession->ses_id);
+        }
+
+        for (auto i = pSession->last_statements->rbegin(); i != pSession->last_statements->rend(); ++i)
+        {
+            int len = i->size();
+            const char* pStmt = (char*) &i->front();
+
+            if (id != 0)
+            {
+                MXS_NOTICE("Stmt %d: %.*s", n, len, pStmt);
+            }
+            else
+            {
+                // We are in a context where we do not have a current session, so we need to
+                // log the session id ourselves.
+
+                MXS_NOTICE("(%" PRIu64 ") Stmt %d: %.*s", pSession->ses_id, n, len, pStmt);
+            }
+
+            --n;
+        }
+    }
+}

server/modules/protocol/MySQL/mariadbclient/mysql_client.cc

@@ -973,6 +973,8 @@ gw_read_normal_data(DCB *dcb, GWBUF *read_buffer, int nbytes_read)
         }
 
         set_qc_mode(session, &read_buffer);
+
+        session_retain_statement(session, read_buffer);
     }
     /** Update the current protocol command being executed */
     else if (!process_client_commands(dcb, nbytes_read, &read_buffer))
@@ -1408,6 +1410,7 @@ static int gw_client_hangup_event(DCB *dcb)
         CHK_SESSION(session);
         if (session->state != SESSION_STATE_DUMMY && !session_valid_for_pool(session))
         {
+            session_dump_statements(session);
             // The client did not send a COM_QUIT packet
             modutil_send_mysql_err_packet(dcb, 0, 0, 1927, "08S01", "Connection killed by MaxScale");
         }