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

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

mysql-test/suite/plugins/r/server_audit.result

@@ -212,7 +212,21 @@ select 2;
 2
 2
 drop table t1;
+set global server_audit_events='table';
+set global server_audit_incl_users='user1';
+create user user1@localhost;
+grant all on sa_db.* to user1@localhost;
+connect  cn1,localhost,user1,,sa_db;
+connection cn1;
+create table t1(id int) engine=myisam;
+insert delayed into t1 values (1), (2);
+connection default;
+# Waiting until ALTER TABLE is blocked.
+drop table t1;
+disconnect cn1;
+drop user user1@localhost;
 set global server_audit_logging= off;
+set global server_audit_incl_users='';
 set global server_audit_logging= on;
 set global server_audit_events='';
 set global server_audit_query_log_limit= 15;
@@ -250,7 +264,7 @@ server_audit_file_path
 server_audit_file_rotate_now	OFF
 server_audit_file_rotate_size	1000000
 server_audit_file_rotations	9
-server_audit_incl_users	odin, root, dva, tri
+server_audit_incl_users	
 server_audit_logging	ON
 server_audit_mode	1
 server_audit_output_type	file
@@ -381,8 +395,9 @@ TIME,HOSTNAME,root,localhost,ID,ID,QUERY,sa_db,'SET PASSWORD FOR u1=<secret>',ID
 TIME,HOSTNAME,root,localhost,ID,ID,QUERY,sa_db,'CREATE USER u3 IDENTIFIED BY *****',0
 TIME,HOSTNAME,root,localhost,ID,ID,QUERY,sa_db,'drop user u1, u2, u3',0
 TIME,HOSTNAME,root,localhost,ID,ID,QUERY,sa_db,'insert into t1 values (1), (2)',0
+TIME,HOSTNAME,user1,localhost,ID,ID,CREATE,sa_db,t1,
+TIME,HOSTNAME,user1,localhost,ID,ID,WRITE,sa_db,t1,
 TIME,HOSTNAME,root,localhost,ID,ID,QUERY,sa_db,'set global server_audit_logging= off',0
-TIME,HOSTNAME,root,localhost,ID,ID,QUERY,sa_db,'set global server_audit_logging= on',0
 TIME,HOSTNAME,root,localhost,ID,ID,QUERY,sa_db,'set global server_audit_events=\'\'',0
 TIME,HOSTNAME,root,localhost,ID,ID,QUERY,sa_db,'set global serv',0
 TIME,HOSTNAME,root,localhost,ID,ID,QUERY,sa_db,'select (1), (2)',0

mysql-test/suite/plugins/t/server_audit.test

@@ -136,7 +136,27 @@ select * from t1;
 select 2;
 drop table t1;
 
+set global server_audit_events='table';
+set global server_audit_incl_users='user1';
+
+create user user1@localhost;
+grant all on sa_db.* to user1@localhost;
+
+connect (cn1,localhost,user1,,sa_db);
+connection cn1;
+
+create table t1(id int) engine=myisam;
+insert delayed into t1 values (1), (2);
+connection default;
+--echo # Waiting until ALTER TABLE is blocked.
+let $wait_condition= SELECT COUNT(*) = 2 FROM t1;
+--source include/wait_condition.inc
+drop table t1;
+disconnect cn1;
+drop user user1@localhost;
+
 set global server_audit_logging= off;
+set global server_audit_incl_users='';
 set global server_audit_logging= on;
 
 set global server_audit_events='';

sql/sql_insert.cc

@@ -2020,11 +2020,15 @@ class delayed_row :public ilink {
   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 @@ class Delayed_insert :public ilink {
     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 @@ class Delayed_insert :public ilink {
      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 */