MENT-237 Audit to show INSERT DELAYED for the executing user.

Add notifications about the user and connection that actually
did the DELAYED insert.

1 files changed, 75 insertions, 4 deletions

diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 550851bd215..9dca28ce48f 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2020,11 +2020,15 @@ public:
   ulong auto_increment_offset;
   LEX_STRING query;
   Time_zone *time_zone;
+  char *user, *host, *ip;
+  query_id_t query_id;
+  my_thread_id thread_id;
 
   delayed_row(LEX_STRING const query_arg, enum_duplicates dup_arg,
               bool ignore_arg, bool log_query_arg)
     : record(0), dup(dup_arg), ignore(ignore_arg), log_query(log_query_arg),
-      forced_insert_id(0), query(query_arg), time_zone(0)
+      forced_insert_id(0), query(query_arg), time_zone(0),
+      user(0), host(0), ip(0)
     {}
   ~delayed_row()
   {
@@ -2072,6 +2076,26 @@ public:
     passed from connection thread to the handler thread.
   */
   MDL_request grl_protection;
+  void set_default_user()
+  {
+    thd.security_ctx->user=(char*) delayed_user;
+    thd.security_ctx->host=(char*) my_localhost;
+    thd.security_ctx->ip= NULL;
+    thd.query_id= 0;
+    thd.thread_id= 0;
+  }
+
+  void set_user_from_row(const delayed_row *r)
+  {
+    if (r)
+    {
+      thd.security_ctx->user= r->user;
+      thd.security_ctx->host= r->host;
+      thd.security_ctx->ip= r->ip;
+      thd.query_id= r->query_id;
+      thd.thread_id= r->thread_id;
+    }
+  }
 
   Delayed_insert(SELECT_LEX *current_select)
     :locks_in_memory(0), thd(next_thread_id()),
@@ -2079,8 +2103,7 @@ public:
      status(0), retry(0), handler_thread_initialized(FALSE), group_count(0)
   {
     DBUG_ENTER("Delayed_insert constructor");
-    thd.security_ctx->user=(char*) delayed_user;
-    thd.security_ctx->host=(char*) my_localhost;
+    set_default_user();
     strmake_buf(thd.security_ctx->priv_user, thd.security_ctx->user);
     thd.current_tablenr=0;
     thd.set_command(COM_DELAYED_INSERT);
@@ -2584,6 +2607,7 @@ int write_delayed(THD *thd, TABLE *table, enum_duplicates duplic,
   delayed_row *row= 0;
   Delayed_insert *di=thd->di;
   const Discrete_interval *forced_auto_inc;
+  size_t user_len, host_len, ip_len;
   DBUG_ENTER("write_delayed");
   DBUG_PRINT("enter", ("query = '%s' length %lu", query.str,
                        (ulong) query.length));
@@ -2617,11 +2641,45 @@ int write_delayed(THD *thd, TABLE *table, enum_duplicates duplic,
     goto err;
   }
 
+  user_len= host_len= ip_len= 0;
+  row->user= row->host= row->ip= NULL;
+  if (thd->security_ctx)
+  {
+    if (thd->security_ctx->user)
+      user_len= strlen(thd->security_ctx->user) + 1;
+    if (thd->security_ctx->host)
+      host_len= strlen(thd->security_ctx->host) + 1;
+    if (thd->security_ctx->ip)
+      ip_len= strlen(thd->security_ctx->ip) + 1;
+  }
   /* This can't be THREAD_SPECIFIC as it's freed in delayed thread */
-  if (!(row->record= (char*) my_malloc(table->s->reclength,
+  if (!(row->record= (char*) my_malloc(table->s->reclength +
+                                       user_len + host_len + ip_len,
                                        MYF(MY_WME))))
     goto err;
   memcpy(row->record, table->record[0], table->s->reclength);
+
+  if (thd->security_ctx)
+  {
+    if (thd->security_ctx->user)
+    {
+      row->user= row->record + table->s->reclength;
+      memcpy(row->user, thd->security_ctx->user, user_len);
+    }
+    if (thd->security_ctx->host)
+    {
+      row->host= row->record + table->s->reclength + user_len;
+      memcpy(row->host, thd->security_ctx->host, host_len);
+    }
+    if (thd->security_ctx->ip)
+    {
+      row->ip= row->record + table->s->reclength + user_len + host_len;
+      memcpy(row->ip, thd->security_ctx->ip, ip_len);
+    }
+  }
+  row->query_id= thd->query_id;
+  row->thread_id= thd->thread_id;
+
   row->start_time=                thd->start_time;
   row->query_start_used=          thd->query_start_used;
   row->start_time_sec_part=       thd->start_time_sec_part;
@@ -3021,6 +3079,7 @@ pthread_handler_t handle_delayed_insert(void *arg)
 
       if (di->tables_in_use && ! thd->lock && !thd->killed)
       {
+        di->set_user_from_row(di->rows.head());
         /*
           Request for new delayed insert.
           Lock the table, but avoid to be blocked by a global read lock.
@@ -3040,6 +3099,18 @@ pthread_handler_t handle_delayed_insert(void *arg)
       }
       if (di->stacked_inserts)
       {
+        delayed_row *row;
+        I_List_iterator<delayed_row> it(di->rows);
+        while ((row= it++))
+        {
+          if (di->thd.thread_id != row->thread_id)
+          {
+            di->set_user_from_row(row);
+            mysql_audit_external_lock(&di->thd, di->table->s, F_WRLCK);
+          }
+        }
+        di->set_default_user();
+
         if (di->handle_inserts())
         {
           /* Some fatal error */