Merge pull request #326 from bazsi/syslog-ng-ctl-attach-command

reattach to the syslog-ng console from syslog-ng-ctl

lib/console.c

@@ -78,41 +78,31 @@ console_is_present(void)
   return result;
 }
 
-gboolean
-console_is_attached(void)
-{
-  gboolean result;
-  /* the lock only serves a memory barrier but is not a real synchronization */
-  g_mutex_lock(&console_lock);
-  if (using_initial_console)
-    result = FALSE;
-  else
-    result = console_present;
-  g_mutex_unlock(&console_lock);
-  return result;
-}
-
 /* re-acquire a console after startup using an array of fds */
-void
+gboolean
 console_acquire_from_fds(gint fds[3])
 {
   const gchar *takeover_message_on_old_console = "[Console taken over, no further output here]\n";
-  g_assert(!console_is_attached());
+  gboolean result = FALSE;
 
-  if (using_initial_console)
+  g_mutex_lock(&console_lock);
+  if (console_present)
     {
+      if (!using_initial_console)
+        goto exit;
       (void) write(1, takeover_message_on_old_console, strlen(takeover_message_on_old_console));
     }
 
-  g_mutex_lock(&console_lock);
-
   dup2(fds[0], STDIN_FILENO);
   dup2(fds[1], STDOUT_FILENO);
   dup2(fds[2], STDERR_FILENO);
 
   console_present = TRUE;
   using_initial_console = FALSE;
+  result = TRUE;
+exit:
   g_mutex_unlock(&console_lock);
+  return result;
 }
 
 /**

lib/console.h

@@ -30,9 +30,7 @@
 void console_printf(const gchar *fmt, ...) __attribute__ ((format (printf, 1, 2)));
 
 gboolean console_is_present(void);
-gboolean console_is_attached(void);
-void console_acquire_from_fds(gint fds[3]);
-void console_acquire_from_stdio(void);
+gboolean console_acquire_from_fds(gint fds[3]);
 void console_release(void);
 
 void console_global_init(const gchar *console_prefix);

lib/control/control-command-thread.c

@@ -27,6 +27,7 @@
 #include "messages.h"
 #include "secret-storage/secret-storage.h"
 #include "scratch-buffers.h"
+#include "apphook.h"
 #include <iv_event.h>
 
 struct _ControlCommandThread
@@ -44,6 +45,12 @@ struct _ControlCommandThread
   struct iv_event thread_finished;
 };
 
+gboolean
+control_command_thread_relates_to_connection(ControlCommandThread *self, ControlConnection *cc)
+{
+  return self->connection == cc;
+}
+
 static void
 _on_thread_finished(gpointer user_data)
 {
@@ -64,7 +71,7 @@ _thread(gpointer user_data)
   ControlCommandThread *self = (ControlCommandThread *) user_data;
 
   iv_init();
-  scratch_buffers_allocator_init();
+  app_thread_start();
 
   msg_debug("Control command thread has started",
             evt_tag_str("control_command", self->command->str));
@@ -82,8 +89,8 @@ _thread(gpointer user_data)
             evt_tag_str("control_command", self->command->str));
 
   scratch_buffers_explicit_gc();
-  scratch_buffers_allocator_deinit();
   control_command_thread_unref(self);
+  app_thread_stop();
   iv_deinit();
 }
 

lib/control/control-command-thread.h

@@ -27,6 +27,8 @@
 
 #include "control.h"
 
+gboolean control_command_thread_relates_to_connection(ControlCommandThread *self, ControlConnection *cc);
+
 void control_command_thread_run(ControlCommandThread *self);
 void control_command_thread_cancel(ControlCommandThread *self);
 const gchar *control_command_thread_get_command(ControlCommandThread *self);

lib/control/control-connection.c

@@ -36,6 +36,14 @@ _g_string_destroy(gpointer user_data)
   g_string_free(str, TRUE);
 }
 
+gboolean
+control_connection_get_attached_fds(ControlConnection *self, gint *fds, gsize *num_fds)
+{
+  if (self->get_attached_fds)
+    return self->get_attached_fds(self, fds, num_fds);
+  return FALSE;
+}
+
 static void
 _control_connection_free(ControlConnection *self)
 {
@@ -215,7 +223,7 @@ control_connection_io_input(void *s)
     }
   else if (rc == 0)
     {
-      msg_debug("EOF on control channel, closing connection");
+      msg_trace("EOF on control channel, closing connection");
       goto destroy_connection;
     }
   else

lib/control/control-connection.h

@@ -41,6 +41,7 @@ struct _ControlConnection
   GString *output_buffer;
   gsize pos;
   ControlServer *server;
+  gboolean (*get_attached_fds)(ControlConnection *self, gint *fds, gsize *num_fds);
   gboolean (*run_command)(ControlConnection *self, ControlCommand *command_desc, GString *command_string);
   int (*read)(ControlConnection *self, gpointer buffer, gsize size);
   int (*write)(ControlConnection *self, gpointer buffer, gsize size);
@@ -56,6 +57,7 @@ struct _ControlConnection
 
 };
 
+gboolean control_connection_get_attached_fds(ControlConnection *self, gint *fds, gsize *num_fds);
 gboolean control_connection_run_command(ControlConnection *self, GString *command_string);
 void control_connection_send_reply(ControlConnection *self, GString *reply);
 void control_connection_send_batched_reply(ControlConnection *self, GString *reply);

lib/control/control-server-unix.c

@@ -42,20 +42,80 @@ typedef struct _ControlConnectionUnix
   ControlConnection super;
   struct iv_fd control_io;
   gint fd;
+  /* stdin, stdout, stderr as passed by syslog-ng-ctl */
+  gint attached_fds[3];
 } ControlConnectionUnix;
 
+static gboolean
+control_connection_unix_get_attached_fds(ControlConnection *s, gint *fds, gsize *num_fds)
+{
+  ControlConnectionUnix *self = (ControlConnectionUnix *)s;
+
+  g_assert(*num_fds >= 3);
+  memcpy(fds, self->attached_fds, sizeof(self->attached_fds));
+  *num_fds = 3;
+  return TRUE;
+}
+
 gint
 control_connection_unix_write(ControlConnection *s, gpointer buffer, gsize size)
 {
   ControlConnectionUnix *self = (ControlConnectionUnix *)s;
   return write(self->control_io.fd, buffer, size);
 }
 
+static gint
+_extract_ancillary_data(ControlConnectionUnix *self, gint rc, struct msghdr *msg)
+{
+  if (G_UNLIKELY(msg->msg_flags & MSG_CTRUNC))
+    {
+      msg_warning_once("WARNING: recvmsg() on control socket returned truncated control data",
+                       evt_tag_int("control_len", msg->msg_controllen));
+      return -1;
+    }
+
+  for (struct cmsghdr *cmsg = CMSG_FIRSTHDR(msg); cmsg != NULL; cmsg = CMSG_NXTHDR(msg, cmsg))
+    {
+      if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS)
+        {
+          gint header_len = CMSG_DATA(cmsg) - (unsigned char *) cmsg;
+          gint fd_array_size = (cmsg->cmsg_len - header_len);
+
+          if (fd_array_size != sizeof(self->attached_fds))
+            {
+              msg_warning_once("WARNING: invalid number of fds received on control socket",
+                               evt_tag_int("fd_array_size", fd_array_size));
+              return -1;
+            }
+          memcpy(&self->attached_fds, CMSG_DATA(cmsg), sizeof(self->attached_fds));
+          break;
+        }
+    }
+
+  return rc;
+}
+
 gint
 control_connection_unix_read(ControlConnection *s, gpointer buffer, gsize size)
 {
   ControlConnectionUnix *self = (ControlConnectionUnix *)s;
-  return read(self->control_io.fd, buffer, size);
+  gchar cmsg_buf[256];
+  struct iovec iov[1] =
+  {
+    { .iov_base = buffer, .iov_len = size },
+  };
+  struct msghdr msg =
+  {
+    .msg_iov = iov,
+    .msg_iovlen = G_N_ELEMENTS(iov),
+    .msg_control = cmsg_buf,
+    .msg_controllen = sizeof(cmsg_buf),
+  };
+  gint rc = recvmsg(self->control_io.fd, &msg, 0);
+  if (rc < 0)
+    return rc;
+
+  return _extract_ancillary_data(self, rc, &msg);
 }
 
 static void
@@ -114,6 +174,11 @@ control_connection_unix_free(ControlConnection *s)
 {
   ControlConnectionUnix *self = (ControlConnectionUnix *)s;
   close(self->control_io.fd);
+  for (gint i = 0; i < G_N_ELEMENTS(self->attached_fds); i++)
+    {
+      if (self->attached_fds[i] >= 0)
+        close(self->attached_fds[i]);
+    }
 }
 
 ControlConnection *
@@ -129,6 +194,12 @@ control_connection_unix_new(ControlServer *server, gint sock)
   self->super.events.start_watches = control_connection_unix_start_watches;
   self->super.events.update_watches = control_connection_unix_update_watches;
   self->super.events.stop_watches = control_connection_unix_stop_watches;
+  self->super.get_attached_fds = control_connection_unix_get_attached_fds;
+
+  for (gint i = 0; i < G_N_ELEMENTS(self->attached_fds); i++)
+    {
+      self->attached_fds[i] = -1;
+    }
 
   return &self->super;
 }
@@ -152,7 +223,6 @@ _control_socket_accept(void *cookie)
       goto error;
     }
 
-
   cc = control_connection_unix_new(&self->super, conn_socket);
 
   /* NOTE: with the call below, the reference to the control connection (cc)

lib/control/control-server.c

@@ -31,6 +31,16 @@ void
 _cancel_worker(gpointer data, gpointer user_data)
 {
   ControlCommandThread *thread = (ControlCommandThread *) data;
+  ControlConnection *cc = (ControlConnection *) user_data;
+
+  if (cc && !control_command_thread_relates_to_connection(thread, cc))
+    {
+      /* check if we relate to a specific connection and cancel only those.
+       * This is only used when a connection closed while the thread is
+       * still running.
+       */
+      return;
+    }
 
   msg_warning("Requesting the cancellation of control command thread",
               evt_tag_str("control_command", control_command_thread_get_command(thread)));
@@ -52,17 +62,21 @@ _cancel_worker(gpointer data, gpointer user_data)
    */
 }
 
-void
-control_server_cancel_workers(ControlServer *self)
+static void
+control_server_cancel_workers(ControlServer *self, ControlConnection *cc)
 {
   if (self->worker_threads)
     {
-      msg_debug("Cancelling control server worker threads");
-      g_list_foreach(self->worker_threads, _cancel_worker, NULL);
-      msg_debug("Control server worker threads have been cancelled");
+      g_list_foreach(self->worker_threads, _cancel_worker, cc);
     }
 }
 
+void
+control_server_cancel_all_workers(ControlServer *self)
+{
+  control_server_cancel_workers(self, NULL);
+}
+
 void
 control_server_worker_started(ControlServer *self, ControlCommandThread *worker)
 {
@@ -80,6 +94,7 @@ control_server_worker_finished(ControlServer *self, ControlCommandThread *worker
 void
 control_server_connection_closed(ControlServer *self, ControlConnection *cc)
 {
+  control_server_cancel_workers(self, cc);
   control_connection_stop_watches(cc);
   control_connection_unref(cc);
 }

lib/control/control-server.h

@@ -38,7 +38,7 @@ struct _ControlServer
   void (*free_fn)(ControlServer *self);
 };
 
-void control_server_cancel_workers(ControlServer *self);
+void control_server_cancel_all_workers(ControlServer *self);
 void control_server_connection_closed(ControlServer *self, ControlConnection *cc);
 void control_server_worker_started(ControlServer *self, ControlCommandThread *worker);
 void control_server_worker_finished(ControlServer *self, ControlCommandThread *worker);