[HAPROXY] use $SOURCEIP instead of $PROXIED_SRCIP

lib/apphook.c

@@ -49,6 +49,7 @@
 #include "timeutils/cache.h"
 #include "multi-line/multi-line-factory.h"
 #include "filterx/filterx-globals.h"
+#include "transport/transport-globals.h"
 
 #include <iv.h>
 #include <iv_work.h>
@@ -243,6 +244,7 @@ app_startup(void)
   timeutils_global_init();
   multi_line_global_init();
   filterx_global_init();
+  log_transport_global_init();
 }
 
 void
@@ -270,6 +272,7 @@ app_shutdown(void)
   msg_stats_deinit();
   run_application_hook(AH_SHUTDOWN);
 
+  log_transport_global_deinit();
   filterx_global_deinit();
   multi_line_global_deinit();
   main_loop_thread_resource_deinit();

lib/logproto/logproto-auto-server.c

@@ -33,35 +33,36 @@ typedef struct _LogProtoAutoServer
 static LogProtoServer *
 _construct_detected_proto(LogProtoAutoServer *self, const gchar *detect_buffer, gsize detect_buffer_len)
 {
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
+  gint fd = self->super.transport_stack.fd;
 
   if (g_ascii_isdigit(detect_buffer[0]))
     {
-      msg_debug("Auto-detected octet-counted-framing on RFC6587 connection, using framed protocol",
-                evt_tag_int("fd", transport->fd));
+      msg_debug("Auto-detected octet-counted-framing, using framed protocol",
+                evt_tag_int("fd", fd));
       return log_proto_framed_server_new(NULL, self->super.options);
     }
   if (detect_buffer[0] == '<')
     {
-      msg_debug("Auto-detected non-transparent-framing on RFC6587 connection, using simple text protocol",
-                evt_tag_int("fd", transport->fd));
+      msg_debug("Auto-detected non-transparent-framing, using simple text protocol",
+                evt_tag_int("fd", fd));
     }
   else
     {
-      msg_debug("Unable to detect framing on RFC6587 connection, falling back to simple text transport",
-                evt_tag_int("fd", transport->fd),
+      msg_debug("Unable to detect framing, falling back to simple text transport",
+                evt_tag_int("fd", fd),
                 evt_tag_mem("detect_buffer", detect_buffer, detect_buffer_len));
     }
   return log_proto_text_server_new(NULL, self->super.options);
 }
 
 static LogProtoPrepareAction
-log_proto_auto_server_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout G_GNUC_UNUSED)
+log_proto_auto_server_poll_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout G_GNUC_UNUSED)
 {
   LogProtoAutoServer *self = (LogProtoAutoServer *) s;
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
 
-  *cond = transport->cond;
+  if (log_transport_stack_poll_prepare(&self->super.transport_stack, cond))
+    return LPPA_FORCE_SCHEDULE_FETCH;
+
   if (*cond == 0)
     *cond = G_IO_IN;
 
@@ -72,12 +73,11 @@ static LogProtoStatus
 log_proto_auto_handshake(LogProtoServer *s, gboolean *handshake_finished, LogProtoServer **proto_replacement)
 {
   LogProtoAutoServer *self = (LogProtoAutoServer *) s;
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
   gchar detect_buffer[8];
   gboolean moved_forward;
   gint rc;
 
-  rc = log_transport_read_ahead(transport, detect_buffer, sizeof(detect_buffer), &moved_forward);
+  rc = log_transport_stack_read_ahead(&self->super.transport_stack, detect_buffer, sizeof(detect_buffer), &moved_forward);
   if (rc == 0)
     return LPS_EOF;
   else if (rc < 0)
@@ -101,6 +101,6 @@ log_proto_auto_server_new(LogTransport *transport, const LogProtoServerOptions *
 
   log_proto_server_init(&self->super, transport, options);
   self->super.handshake = log_proto_auto_handshake;
-  self->super.prepare = log_proto_auto_server_prepare;
+  self->super.poll_prepare = log_proto_auto_server_poll_prepare;
   return &self->super;
 }

lib/logproto/logproto-buffered-server.c

@@ -179,10 +179,9 @@ log_proto_buffered_server_apply_state(LogProtoBufferedServer *self, PersistEntry
   struct stat st;
   gint64 ofs = 0;
   LogProtoBufferedServerState *state;
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
   gint fd;
 
-  fd = transport->fd;
+  fd = self->super.transport_stack.fd;
   self->persist_handle = handle;
 
   if (fstat(fd, &st) < 0)
@@ -255,7 +254,7 @@ log_proto_buffered_server_apply_state(LogProtoBufferedServer *self, PersistEntry
           raw_buffer = g_alloca(state->raw_buffer_size);
         }
 
-      rc = log_transport_read(transport, raw_buffer, state->raw_buffer_size, NULL);
+      rc = log_transport_stack_read(&self->super.transport_stack, raw_buffer, state->raw_buffer_size, NULL);
       if (rc != state->raw_buffer_size)
         {
           msg_notice("Error re-reading buffer contents of the file to be continued, restarting from the beginning",
@@ -584,12 +583,12 @@ log_proto_buffered_server_restart_with_state(LogProtoServer *s, PersistState *pe
 }
 
 LogProtoPrepareAction
-log_proto_buffered_server_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout G_GNUC_UNUSED)
+log_proto_buffered_server_poll_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout G_GNUC_UNUSED)
 {
   LogProtoBufferedServer *self = (LogProtoBufferedServer *) s;
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
 
-  *cond = transport->cond;
+  if (log_transport_stack_poll_prepare(&self->super.transport_stack, cond))
+    return LPPA_FORCE_SCHEDULE_FETCH;
 
   /* if there's no pending I/O in the transport layer, then we want to do a read */
   if (*cond == 0)
@@ -602,9 +601,7 @@ static gint
 log_proto_buffered_server_read_data_method(LogProtoBufferedServer *self, guchar *buf, gsize len,
                                            LogTransportAuxData *aux)
 {
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
-
-  return log_transport_read(transport, buf, len, aux);
+  return log_transport_stack_read(&self->super.transport_stack, buf, len, aux);
 }
 
 static void
@@ -1081,7 +1078,7 @@ log_proto_buffered_server_init(LogProtoBufferedServer *self, LogTransport *trans
                                const LogProtoServerOptions *options)
 {
   log_proto_server_init(&self->super, transport, options);
-  self->super.prepare = log_proto_buffered_server_prepare;
+  self->super.poll_prepare = log_proto_buffered_server_poll_prepare;
   self->super.fetch = log_proto_buffered_server_fetch;
   self->super.free_fn = log_proto_buffered_server_free_method;
   self->super.restart_with_state = log_proto_buffered_server_restart_with_state;

lib/logproto/logproto-buffered-server.h

@@ -114,8 +114,8 @@ log_proto_buffered_server_cue_flush(LogProtoBufferedServer *self)
   self->flush_partial_message = TRUE;
 }
 
-LogProtoPrepareAction log_proto_buffered_server_prepare(LogProtoServer *s, GIOCondition *cond,
-                                                        gint *timeout G_GNUC_UNUSED);
+LogProtoPrepareAction log_proto_buffered_server_poll_prepare(LogProtoServer *s, GIOCondition *cond,
+    gint *timeout G_GNUC_UNUSED);
 LogProtoBufferedServerState *log_proto_buffered_server_get_state(LogProtoBufferedServer *self);
 void log_proto_buffered_server_put_state(LogProtoBufferedServer *self);
 gboolean log_proto_buffered_server_restart_with_state(LogProtoServer *s,

lib/logproto/logproto-client.h

@@ -67,8 +67,7 @@ struct _LogProtoClient
   LogProtoStatus status;
   const LogProtoClientOptions *options;
   LogTransportStack transport_stack;
-  /* FIXME: rename to something else */
-  gboolean (*prepare)(LogProtoClient *s, gint *fd, GIOCondition *cond, gint *timeout);
+  gboolean (*poll_prepare)(LogProtoClient *s, gint *fd, GIOCondition *cond, gint *timeout);
   LogProtoStatus (*post)(LogProtoClient *s, LogMessage *logmsg, guchar *msg, gsize msg_len, gboolean *consumed);
   LogProtoStatus (*process_in)(LogProtoClient *s);
   LogProtoStatus (*flush)(LogProtoClient *s);
@@ -124,9 +123,9 @@ log_proto_client_handshake(LogProtoClient *s, gboolean *handshake_finished)
 }
 
 static inline gboolean
-log_proto_client_prepare(LogProtoClient *s, gint *fd, GIOCondition *cond, gint *timeout)
+log_proto_client_poll_prepare(LogProtoClient *s, gint *fd, GIOCondition *cond, gint *timeout)
 {
-  gboolean result = s->prepare(s, fd, cond, timeout);
+  gboolean result = s->poll_prepare(s, fd, cond, timeout);
 
   if (!result && *timeout < 0)
     *timeout = s->options->idle_timeout;

lib/logproto/logproto-framed-server.c

@@ -66,12 +66,12 @@ typedef struct _LogProtoFramedServer
 } LogProtoFramedServer;
 
 static LogProtoPrepareAction
-log_proto_framed_server_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout G_GNUC_UNUSED)
+log_proto_framed_server_poll_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout G_GNUC_UNUSED)
 {
   LogProtoFramedServer *self = (LogProtoFramedServer *) s;
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
 
-  *cond = transport->cond;
+  if (log_transport_stack_poll_prepare(&self->super.transport_stack, cond))
+    return LPPA_FORCE_SCHEDULE_FETCH;
 
   /* there is a half message in our buffer so try to wait */
   if (!self->half_message_in_buffer)
@@ -97,7 +97,6 @@ static gboolean
 log_proto_framed_server_fetch_data(LogProtoFramedServer *self, gboolean *may_read,
                                    LogProtoStatus *status)
 {
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
   gint rc;
   *status = LPS_SUCCESS;
 
@@ -111,15 +110,16 @@ log_proto_framed_server_fetch_data(LogProtoFramedServer *self, gboolean *may_rea
     return FALSE;
 
   log_transport_aux_data_reinit(&self->buffer_aux);
-  rc = log_transport_read(transport, &self->buffer[self->buffer_end], self->buffer_size - self->buffer_end,
-                          &self->buffer_aux);
+  rc = log_transport_stack_read(&self->super.transport_stack,
+                                &self->buffer[self->buffer_end], self->buffer_size - self->buffer_end,
+                                &self->buffer_aux);
 
   if (rc < 0)
     {
       if (errno != EAGAIN)
         {
           msg_error("Error reading RFC6587 style framed data",
-                    evt_tag_int("fd", transport->fd),
+                    evt_tag_int("fd", self->super.transport_stack.fd),
                     evt_tag_error("error"));
           *status = LPS_ERROR;
         }
@@ -134,7 +134,7 @@ log_proto_framed_server_fetch_data(LogProtoFramedServer *self, gboolean *may_rea
   if (rc == 0)
     {
       msg_trace("EOF occurred while reading",
-                evt_tag_int(EVT_TAG_FD, transport->fd));
+                evt_tag_int(EVT_TAG_FD, self->super.transport_stack.fd));
       *status = LPS_EOF;
       return FALSE;
     }
@@ -430,7 +430,7 @@ log_proto_framed_server_new(LogTransport *transport, const LogProtoServerOptions
   LogProtoFramedServer *self = g_new0(LogProtoFramedServer, 1);
 
   log_proto_server_init(&self->super, transport, options);
-  self->super.prepare = log_proto_framed_server_prepare;
+  self->super.poll_prepare = log_proto_framed_server_poll_prepare;
   self->super.fetch = log_proto_framed_server_fetch;
   self->super.free_fn = log_proto_framed_server_free;
   self->half_message_in_buffer = FALSE;

lib/logproto/logproto-record-server.c

@@ -54,17 +54,16 @@ static gint
 log_proto_record_server_read_data(LogProtoBufferedServer *s, guchar *buf, gsize len, LogTransportAuxData *aux)
 {
   LogProtoRecordServer *self = (LogProtoRecordServer *) s;
-  LogTransport *transport = log_transport_stack_get_active(&self->super.super.transport_stack);
   gint rc;
 
   /* assert that we have enough space in the buffer to read record_size bytes */
   g_assert(len >= self->record_size);
   len = self->record_size;
-  rc = log_transport_read(transport, buf, len, aux);
+  rc = log_transport_stack_read(&s->super.transport_stack, buf, len, aux);
   if (rc > 0 && rc != self->record_size)
     {
       msg_error("Record size was set, and couldn't read enough bytes",
-                evt_tag_int(EVT_TAG_FD, transport->fd),
+                evt_tag_int(EVT_TAG_FD, s->super.transport_stack.fd),
                 evt_tag_int("record_size", self->record_size),
                 evt_tag_int("read", rc));
       errno = EIO;

lib/logproto/logproto-server.h

@@ -83,8 +83,7 @@ struct _LogProtoServer
   AckTracker *ack_tracker;
 
   LogProtoServerWakeupCallback wakeup_callback;
-  /* FIXME: rename to something else */
-  LogProtoPrepareAction (*prepare)(LogProtoServer *s, GIOCondition *cond, gint *timeout);
+  LogProtoPrepareAction (*poll_prepare)(LogProtoServer *s, GIOCondition *cond, gint *timeout);
   gboolean (*restart_with_state)(LogProtoServer *s, PersistState *state, const gchar *persist_name);
   LogProtoStatus (*fetch)(LogProtoServer *s, const guchar **msg, gsize *msg_len, gboolean *may_read,
                           LogTransportAuxData *aux, Bookmark *bookmark);
@@ -125,9 +124,9 @@ log_proto_server_set_options(LogProtoServer *self, const LogProtoServerOptions *
 }
 
 static inline LogProtoPrepareAction
-log_proto_server_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout)
+log_proto_server_poll_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout)
 {
-  LogProtoPrepareAction result = s->prepare(s, cond, timeout);
+  LogProtoPrepareAction result = s->poll_prepare(s, cond, timeout);
 
   if (result == LPPA_POLL_IO && *timeout < 0)
     *timeout = s->options->idle_timeout;

lib/logproto/logproto-text-client.c

@@ -27,13 +27,14 @@
 #include <errno.h>
 
 static gboolean
-log_proto_text_client_prepare(LogProtoClient *s, gint *fd, GIOCondition *cond, gint *timeout)
+log_proto_text_client_poll_prepare(LogProtoClient *s, gint *fd, GIOCondition *cond, gint *timeout)
 {
   LogProtoTextClient *self = (LogProtoTextClient *) s;
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
 
-  *fd = transport->fd;
-  *cond = transport->cond;
+  if (log_transport_stack_poll_prepare(&self->super.transport_stack, cond))
+    return TRUE;
+
+  *fd = self->super.transport_stack.fd;
 
   /* if there's no pending I/O in the transport layer, then we want to do a write */
   if (*cond == 0)
@@ -46,24 +47,26 @@ static LogProtoStatus
 log_proto_text_client_drop_input(LogProtoClient *s)
 {
   LogProtoTextClient *self = (LogProtoTextClient *) s;
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
   guchar buf[1024];
   gint rc = -1;
 
   do
     {
-      rc = log_transport_read(transport, buf, sizeof(buf), NULL);
+      rc = log_transport_stack_read(&self->super.transport_stack, buf, sizeof(buf), NULL);
     }
   while (rc > 0);
 
   if (rc == -1 && errno != EAGAIN)
     {
-      msg_error("Error reading data", evt_tag_int("fd", transport->fd), evt_tag_error("error"));
+      msg_error("Error reading data",
+                evt_tag_int("fd", self->super.transport_stack.fd),
+                evt_tag_error("error"));
       return LPS_ERROR;
     }
   else if (rc == 0)
     {
-      msg_error("EOF occurred while idle", evt_tag_int("fd", transport->fd));
+      msg_error("EOF occurred while idle",
+                evt_tag_int("fd", self->super.transport_stack.fd));
       return LPS_ERROR;
     }
 
@@ -74,7 +77,6 @@ static LogProtoStatus
 log_proto_text_client_flush(LogProtoClient *s)
 {
   LogProtoTextClient *self = (LogProtoTextClient *) s;
-  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
   gint rc;
 
   if (!self->partial)
@@ -85,13 +87,13 @@ log_proto_text_client_flush(LogProtoClient *s)
   /* attempt to flush previously buffered data */
   gint len = self->partial_len - self->partial_pos;
 
-  rc = log_transport_write(transport, &self->partial[self->partial_pos], len);
+  rc = log_transport_stack_write(&self->super.transport_stack, &self->partial[self->partial_pos], len);
   if (rc < 0)
     {
       if (errno != EAGAIN && errno != EINTR)
         {
           msg_error("I/O error occurred while writing",
-                    evt_tag_int("fd", transport->fd),
+                    evt_tag_int("fd", self->super.transport_stack.fd),
                     evt_tag_error(EVT_TAG_OSERROR));
           return LPS_ERROR;
         }
@@ -194,7 +196,7 @@ void
 log_proto_text_client_init(LogProtoTextClient *self, LogTransport *transport, const LogProtoClientOptions *options)
 {
   log_proto_client_init(&self->super, transport, options);
-  self->super.prepare = log_proto_text_client_prepare;
+  self->super.poll_prepare = log_proto_text_client_poll_prepare;
   self->super.flush = log_proto_text_client_flush;
   if (options->drop_input)
     self->super.process_in = log_proto_text_client_drop_input;

lib/logproto/logproto-text-server.c

@@ -27,12 +27,12 @@
 #include <string.h>
 
 LogProtoPrepareAction
-log_proto_text_server_prepare_method(LogProtoServer *s, GIOCondition *cond, gint *timeout)
+log_proto_text_server_poll_prepare_method(LogProtoServer *s, GIOCondition *cond, gint *timeout)
 {
   LogProtoTextServer *self = (LogProtoTextServer *) s;
   gboolean avail;
 
-  LogProtoPrepareAction action = log_proto_buffered_server_prepare(s, cond, timeout);
+  LogProtoPrepareAction action = log_proto_buffered_server_poll_prepare(s, cond, timeout);
   if (action != LPPA_POLL_IO)
     return action;
 
@@ -331,7 +331,7 @@ void
 log_proto_text_server_init(LogProtoTextServer *self, LogTransport *transport, const LogProtoServerOptions *options)
 {
   log_proto_buffered_server_init(&self->super, transport, options);
-  self->super.super.prepare = log_proto_text_server_prepare_method;
+  self->super.super.poll_prepare = log_proto_text_server_poll_prepare_method;
   self->super.super.free_fn = log_proto_text_server_free;
   self->super.super.restart_with_state = log_proto_text_server_restart_with_state;
   self->super.fetch_from_buffer = log_proto_text_server_fetch_from_buffer;