logreader: add exit-on-eof flag()

With the exit-on-eof flag, you can request syslog-ng to exit whenever the first EOF is hit on the specific source, similarly to how the stdin() source behaves. Signed-off-by: Balazs Scheidler <[email protected]>
axoflow · Oct 27, 2024 · 8949ea5 · 8949ea5
1 parent a136792
commit 8949ea5
Show file tree

Hide file tree

Showing 9 changed files with 54 additions and 21 deletions.
diff --git a/lib/logreader.c b/lib/logreader.c
@@ -389,6 +389,11 @@ log_reader_work_finished(void *s, gpointer arg)
 
       self->notify_code = 0;
       log_pipe_notify(self->control, notify_code, self);
+
+      if (notify_code == NC_CLOSE && (self->options->flags & LR_EXIT_ON_EOF))
+        {
+          cfg_shutdown(log_pipe_get_config(s));
+        }
     }
   if ((self->super.super.flags & PIF_INITIALIZED) && self->proto)
     {
@@ -872,6 +877,7 @@ CfgFlagHandler log_reader_flag_handlers[] =
   { "empty-lines",                CFH_SET, offsetof(LogReaderOptions, flags),               LR_EMPTY_LINES },
   { "threaded",                   CFH_SET, offsetof(LogReaderOptions, flags),               LR_THREADED },
   { "ignore-aux-data",            CFH_SET, offsetof(LogReaderOptions, flags),               LR_IGNORE_AUX_DATA },
+  { "exit-on-eof",                CFH_SET, offsetof(LogReaderOptions, flags),               LR_EXIT_ON_EOF },
   { NULL },
 };
 

diff --git a/lib/logreader.h b/lib/logreader.h
@@ -39,6 +39,7 @@
 #define LR_EMPTY_LINES     0x0004
 #define LR_IGNORE_AUX_DATA 0x0008
 #define LR_THREADED        0x0040
+#define LR_EXIT_ON_EOF     0x0080
 
 /* options */
 

diff --git a/modules/affile/affile-source.c b/modules/affile/affile-source.c
@@ -91,7 +91,7 @@ affile_sd_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_options)
   log_src_driver_queue_method(s, msg, path_options);
 }
 
-static gboolean
+gboolean
 affile_sd_init(LogPipe *s)
 {
   AFFileSourceDriver *self = (AFFileSourceDriver *) s;

diff --git a/modules/affile/affile-source.h b/modules/affile/affile-source.h
@@ -41,6 +41,7 @@ typedef struct _AFFileSourceDriver
   gsize transport_name_len;
 } AFFileSourceDriver;
 
+gboolean affile_sd_init(LogPipe *s);
 void affile_sd_set_transport_name(AFFileSourceDriver *s, const gchar *transport_name);
 AFFileSourceDriver *affile_sd_new_instance(gchar *filename, GlobalConfig *cfg);
 LogDriver *affile_sd_new(gchar *filename, GlobalConfig *cfg);

diff --git a/modules/affile/file-reader.c b/modules/affile/file-reader.c
@@ -267,8 +267,6 @@ file_reader_notify_method(LogPipe *s, gint notify_code, gpointer user_data)
   switch (notify_code)
     {
     case NC_CLOSE:
-      if (self->options->exit_on_eof)
-        cfg_shutdown(log_pipe_get_config(s));
       break;
 
     case NC_FILE_MOVED:

diff --git a/modules/affile/file-reader.h b/modules/affile/file-reader.h
@@ -32,7 +32,6 @@ typedef struct _FileReaderOptions
   gint multi_line_timeout;
   gboolean restore_state;
   LogReaderOptions reader_options;
-  gboolean exit_on_eof;
 } FileReaderOptions;
 
 typedef struct _FileReader

diff --git a/modules/affile/named-pipe.c b/modules/affile/named-pipe.c
@@ -35,6 +35,12 @@
 #include <fcntl.h>
 #include <errno.h>
 
+typedef struct _FileOpenerNamedPipe
+{
+  FileOpener super;
+  gboolean suppress_eof;
+} FileOpenerNamedPipe;
+
 static gboolean
 _prepare_open(FileOpener *self, const gchar *name)
 {
@@ -64,11 +70,19 @@ _prepare_open(FileOpener *self, const gchar *name)
 }
 
 static gint
-_get_open_flags(FileOpener *self, FileDirection dir)
+_get_open_flags(FileOpener *s, FileDirection dir)
 {
+  FileOpenerNamedPipe *self = (FileOpenerNamedPipe *) s;
   switch (dir)
     {
     case AFFILE_DIR_READ:
+      /* if a named pipe is opened for read write, we won't get an EOF, as
+       * there's always a writer (us, having opened in RW mode). EOF is only
+       * indicated if no writers remain */
+
+      if (self->suppress_eof)
+        return (O_RDWR | O_NOCTTY | O_NONBLOCK | O_LARGEFILE);
+      return (O_RDONLY | O_NOCTTY | O_NONBLOCK | O_LARGEFILE);
     case AFFILE_DIR_WRITE:
       return (O_RDWR | O_NOCTTY | O_NONBLOCK | O_LARGEFILE);
     default:
@@ -77,11 +91,13 @@ _get_open_flags(FileOpener *self, FileDirection dir)
 }
 
 static LogTransport *
-_construct_transport(FileOpener *self, gint fd)
+_construct_transport(FileOpener *s, gint fd)
 {
+  FileOpenerNamedPipe *self = (FileOpenerNamedPipe *) s;
   LogTransport *transport = log_transport_pipe_new(fd);
 
-  transport->read = log_transport_file_read_and_ignore_eof_method;
+  if (self->suppress_eof)
+    transport->read = log_transport_file_read_and_ignore_eof_method;
   return transport;
 }
 
@@ -106,23 +122,37 @@ pipe_sd_set_create_dirs(LogDriver *s, gboolean create_dirs)
 }
 
 FileOpener *
-file_opener_for_named_pipes_new(void)
+file_opener_for_named_pipes_new(gboolean suppress_eof)
 {
-  FileOpener *self = file_opener_new();
-
-  self->prepare_open = _prepare_open;
-  self->get_open_flags = _get_open_flags;
-  self->construct_transport = _construct_transport;
-  self->construct_src_proto = _construct_src_proto;
-  self->construct_dst_proto = _construct_dst_proto;
-  return self;
+  FileOpenerNamedPipe *self = g_new0(FileOpenerNamedPipe, 1);
+
+  file_opener_init_instance(&self->super);
+
+  self->super.prepare_open = _prepare_open;
+  self->super.get_open_flags = _get_open_flags;
+  self->super.construct_transport = _construct_transport;
+  self->super.construct_src_proto = _construct_src_proto;
+  self->super.construct_dst_proto = _construct_dst_proto;
+
+  self->suppress_eof = suppress_eof;
+  return &self->super;
+}
+
+static gboolean
+_init(LogPipe *s)
+{
+  AFFileSourceDriver *self = (AFFileSourceDriver *) s;
+  if (!self->file_opener)
+    self->file_opener = file_opener_for_named_pipes_new((self->file_reader_options.reader_options.flags & LR_EXIT_ON_EOF) == 0);
+  return affile_sd_init(s);
 }
 
 LogDriver *
 pipe_sd_new(gchar *filename, GlobalConfig *cfg)
 {
   AFFileSourceDriver *self = affile_sd_new_instance(filename, cfg);
 
+  self->super.super.super.init = _init;
   self->file_reader_options.reader_options.super.stats_source = stats_register_type("pipe");
 
   if (cfg_is_config_version_older(cfg, VERSION_VALUE_3_2))
@@ -137,8 +167,6 @@ pipe_sd_new(gchar *filename, GlobalConfig *cfg)
       self->file_reader_options.reader_options.parse_options.flags &= ~LP_EXPECT_HOSTNAME;
     }
 
-  self->file_opener = file_opener_for_named_pipes_new();
-
   affile_sd_set_transport_name(self, "local+pipe");
   return &self->super.super;
 }
@@ -149,6 +177,6 @@ pipe_dd_new(LogTemplate *filename_template, GlobalConfig *cfg)
   AFFileDestDriver *self = affile_dd_new_instance(filename_template, cfg);
 
   self->writer_options.stats_source = stats_register_type("pipe");
-  self->file_opener = file_opener_for_named_pipes_new();
+  self->file_opener = file_opener_for_named_pipes_new(FALSE);
   return &self->super.super;
 }
diff --git a/modules/affile/named-pipe.h b/modules/affile/named-pipe.h
@@ -30,7 +30,7 @@
 
 void pipe_sd_set_create_dirs(LogDriver *s, gboolean create_dirs);
 
-FileOpener *file_opener_for_named_pipes_new(void);
+FileOpener *file_opener_for_named_pipes_new(gboolean open_for_readonly);
 LogDriver *pipe_sd_new(gchar *filename, GlobalConfig *cfg);
 LogDriver *pipe_dd_new(LogTemplate *filename_template, GlobalConfig *cfg);
 

diff --git a/modules/affile/stdin.c b/modules/affile/stdin.c
@@ -60,7 +60,7 @@ stdin_sd_new(GlobalConfig *cfg)
 {
   AFFileSourceDriver *self = affile_sd_new_instance("-", cfg);
 
-  self->file_reader_options.exit_on_eof = TRUE;
+  self->file_reader_options.reader_options.flags |= LR_EXIT_ON_EOF;
   self->file_reader_options.reader_options.super.stats_source = stats_register_type("stdin");
   self->file_opener = file_opener_for_stdin_new();
   affile_sd_set_transport_name(self, "local+stdin");