Adding upstream version 1.44.3.upstream/1.44.3upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

11 files changed, 1503 insertions, 0 deletions

diff --git a/fluent-bit/plugins/in_syslog/CMakeLists.txt b/fluent-bit/plugins/in_syslog/CMakeLists.txt
new file mode 100644
index 00000000..88f698b1
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/CMakeLists.txt
@@ -0,0 +1,8 @@
+set(src
+  syslog_conf.c
+  syslog_server.c
+  syslog_conn.c
+  syslog_prot.c
+  syslog.c)
+
+FLB_PLUGIN(in_syslog "${src}" "")
diff --git a/fluent-bit/plugins/in_syslog/syslog.c b/fluent-bit/plugins/in_syslog/syslog.c
new file mode 100644
index 00000000..d478dfc3
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog.c
@@ -0,0 +1,263 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
+#include <msgpack.h>
+#include <fluent-bit/flb_input_plugin.h>
+#include <fluent-bit/flb_downstream.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_error.h>
+#include <fluent-bit/flb_utils.h>
+
+#include "syslog.h"
+#include "syslog_conf.h"
+#include "syslog_server.h"
+#include "syslog_conn.h"
+#include "syslog_prot.h"
+
+/* cb_collect callback */
+static int in_syslog_collect_tcp(struct flb_input_instance *i_ins,
+                                 struct flb_config *config, void *in_context)
+{
+    struct flb_connection *connection;
+    struct syslog_conn    *conn;
+    struct flb_syslog     *ctx;
+
+    (void) i_ins;
+
+    ctx = in_context;
+
+    connection = flb_downstream_conn_get(ctx->downstream);
+
+    if (connection == NULL) {
+        flb_plg_error(ctx->ins, "could not accept new connection");
+
+        return -1;
+    }
+
+    if (ctx->dgram_mode_flag) {
+        return syslog_dgram_conn_event(connection);
+    }
+    else {
+        flb_plg_trace(ctx->ins, "new Unix connection arrived FD=%i", connection->fd);
+
+        conn = syslog_conn_add(connection, ctx);
+
+        if (conn == NULL) {
+            flb_plg_error(ctx->ins, "could not accept new connection");
+
+            flb_downstream_conn_release(connection);
+
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
+/*
+ * Collect a datagram, per Syslog specification a datagram contains only
+ * one syslog message and it should not exceed 1KB.
+ */
+static int in_syslog_collect_udp(struct flb_input_instance *i_ins,
+                                 struct flb_config *config,
+                                 void *in_context)
+{
+    struct flb_syslog *ctx;
+
+    (void) i_ins;
+
+    ctx = in_context;
+
+    return syslog_dgram_conn_event(ctx->dummy_conn->connection);
+}
+
+/* Initialize plugin */
+static int in_syslog_init(struct flb_input_instance *in,
+                          struct flb_config *config, void *data)
+{
+    int ret;
+    struct flb_syslog *ctx;
+    struct flb_connection *connection;
+
+    /* Allocate space for the configuration */
+    ctx = syslog_conf_create(in, config);
+    if (!ctx) {
+        flb_plg_error(in, "could not initialize plugin");
+        return -1;
+    }
+    ctx->collector_id = -1;
+
+    if ((ctx->mode == FLB_SYSLOG_UNIX_TCP || ctx->mode == FLB_SYSLOG_UNIX_UDP)
+        && !ctx->unix_path) {
+        flb_plg_error(ctx->ins, "Unix path not defined");
+        syslog_conf_destroy(ctx);
+        return -1;
+    }
+
+    /* Create Unix Socket */
+    ret = syslog_server_create(ctx);
+    if (ret == -1) {
+        syslog_conf_destroy(ctx);
+        return -1;
+    }
+
+    flb_input_downstream_set(ctx->downstream, ctx->ins);
+
+    if (ctx->dgram_mode_flag) {
+        connection = flb_downstream_conn_get(ctx->downstream);
+
+        if (connection == NULL) {
+            flb_plg_error(ctx->ins, "could not get DGRAM server dummy "
+                                    "connection");
+
+            syslog_conf_destroy(ctx);
+
+            return -1;
+        }
+
+        ctx->dummy_conn = syslog_conn_add(connection, ctx);
+
+        if (ctx->dummy_conn == NULL) {
+            flb_plg_error(ctx->ins, "could not track DGRAM server dummy "
+                                    "connection");
+
+            syslog_conf_destroy(ctx);
+
+            return -1;
+        }
+    }
+
+    /* Set context */
+    flb_input_set_context(in, ctx);
+
+    /* Collect events for every opened connection to our socket */
+    if (ctx->mode == FLB_SYSLOG_UNIX_TCP ||
+        ctx->mode == FLB_SYSLOG_TCP) {
+        ret = flb_input_set_collector_socket(in,
+                                             in_syslog_collect_tcp,
+                                             ctx->downstream->server_fd,
+                                             config);
+    }
+    else {
+        ret = flb_input_set_collector_socket(in,
+                                             in_syslog_collect_udp,
+                                             ctx->downstream->server_fd,
+                                             config);
+    }
+
+    if (ret == -1) {
+        flb_plg_error(ctx->ins, "Could not set collector");
+        syslog_conf_destroy(ctx);
+
+        return -1;
+    }
+
+    ctx->collector_id = ret;
+    ctx->collector_event = flb_input_collector_get_event(ret, in);
+
+    if (ret == -1) {
+        flb_plg_error(ctx->ins, "Could not get collector event");
+        syslog_conf_destroy(ctx);
+
+        return -1;
+    }
+
+    return 0;
+}
+
+static int in_syslog_exit(void *data, struct flb_config *config)
+{
+    struct flb_syslog *ctx = data;
+    (void) config;
+
+    syslog_conn_exit(ctx);
+    syslog_conf_destroy(ctx);
+
+    return 0;
+}
+
+static struct flb_config_map config_map[] = {
+    {
+     FLB_CONFIG_MAP_STR, "mode", (char *)NULL,
+     0, FLB_TRUE, offsetof(struct flb_syslog, mode_str),
+     "Set the socket mode: unix_tcp, unix_udp, tcp or udp"
+    },
+    {
+     FLB_CONFIG_MAP_STR, "path", (char *)NULL,
+     0, FLB_TRUE, offsetof(struct flb_syslog, unix_path),
+     "Set the path for the UNIX socket"
+    },
+    {
+     FLB_CONFIG_MAP_STR, "unix_perm", (char *)NULL,
+     0, FLB_TRUE, offsetof(struct flb_syslog, unix_perm_str),
+     "Set the permissions for the UNIX socket"
+    },
+    {
+      FLB_CONFIG_MAP_SIZE, "buffer_chunk_size", FLB_SYSLOG_CHUNK,
+      0, FLB_TRUE, offsetof(struct flb_syslog, buffer_chunk_size),
+      "Set the buffer chunk size"
+    },
+    {
+      FLB_CONFIG_MAP_SIZE, "buffer_max_size", (char *)NULL,
+      0, FLB_TRUE, offsetof(struct flb_syslog, buffer_max_size),
+      "Set the buffer chunk size"
+    },
+    {
+     FLB_CONFIG_MAP_STR, "parser", (char *)NULL,
+     0, FLB_TRUE, offsetof(struct flb_syslog, parser_name),
+     "Set the parser"
+    },
+    {
+      FLB_CONFIG_MAP_SIZE, "receive_buffer_size", (char *)NULL,
+      0, FLB_TRUE, offsetof(struct flb_syslog, receive_buffer_size),
+      "Set the socket receiving buffer size"
+    },
+    {
+     FLB_CONFIG_MAP_STR, "raw_message_key", (char *) NULL,
+     0, FLB_TRUE, offsetof(struct flb_syslog, raw_message_key),
+     "Key where the raw message will be preserved"
+    },
+    {
+     FLB_CONFIG_MAP_STR, "source_address_key", (char *) NULL,
+     0, FLB_TRUE, offsetof(struct flb_syslog, source_address_key),
+     "Key where the source address will be injected"
+    },
+
+
+    /* EOF */
+    {0}
+};
+
+struct flb_input_plugin in_syslog_plugin = {
+    .name         = "syslog",
+    .description  = "Syslog",
+    .cb_init      = in_syslog_init,
+    .cb_pre_run   = NULL,
+    .cb_collect   = NULL,
+    .cb_flush_buf = NULL,
+    .cb_exit      = in_syslog_exit,
+    .config_map   = config_map,
+    .flags        = FLB_INPUT_NET_SERVER | FLB_IO_OPT_TLS
+};
diff --git a/fluent-bit/plugins/in_syslog/syslog.h b/fluent-bit/plugins/in_syslog/syslog.h
new file mode 100644
index 00000000..6da2fbd8
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog.h
@@ -0,0 +1,82 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_IN_SYSLOG_H
+#define FLB_IN_SYSLOG_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_input.h>
+#include <fluent-bit/flb_log_event_encoder.h>
+
+/* Syslog modes */
+#define FLB_SYSLOG_UNIX_TCP  1
+#define FLB_SYSLOG_UNIX_UDP  2
+#define FLB_SYSLOG_TCP       3
+#define FLB_SYSLOG_UDP       4
+
+/* 32KB chunk size */
+#define FLB_SYSLOG_CHUNK   "32768"
+
+struct syslog_conn;
+
+/* Context / Config*/
+struct flb_syslog {
+    /* Listening mode: unix udp, unix tcp or normal tcp */
+    flb_sds_t mode_str;
+    int mode;
+
+    /* Network mode */
+    char *listen;
+    char *port;
+
+    /* Unix socket (UDP/TCP)*/
+    int server_fd;
+    flb_sds_t unix_path;
+    flb_sds_t unix_perm_str;
+    unsigned int unix_perm;
+    size_t receive_buffer_size;
+
+    /* UDP buffer, data length and buffer size */
+        // char *buffer_data;
+        // size_t buffer_len;
+        // size_t buffer_size;
+
+    /* Buffers setup */
+    size_t buffer_max_size;
+    size_t buffer_chunk_size;
+
+    /* Configuration */
+    flb_sds_t parser_name;
+    struct flb_parser *parser;
+    flb_sds_t raw_message_key;
+    flb_sds_t source_address_key;
+
+    int dgram_mode_flag;
+    int collector_id;
+    struct mk_event *collector_event;
+    struct flb_downstream *downstream;
+    struct syslog_conn *dummy_conn;
+
+    /* List for connections and event loop */
+    struct mk_list connections;
+    struct flb_input_instance *ins;
+    struct flb_log_event_encoder *log_encoder;
+};
+
+#endif
diff --git a/fluent-bit/plugins/in_syslog/syslog_conf.c b/fluent-bit/plugins/in_syslog/syslog_conf.c
new file mode 100644
index 00000000..4db3f162
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog_conf.c
@@ -0,0 +1,193 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_input.h>
+#include <fluent-bit/flb_input_plugin.h>
+#include <fluent-bit/flb_str.h>
+#include <fluent-bit/flb_log.h>
+#include <fluent-bit/flb_parser.h>
+#include <fluent-bit/flb_utils.h>
+
+#include "syslog.h"
+#include "syslog_server.h"
+#include "syslog_conf.h"
+
+struct flb_syslog *syslog_conf_create(struct flb_input_instance *ins,
+                                      struct flb_config *config)
+{
+    int ret;
+    char port[16];
+    struct flb_syslog *ctx;
+
+    ctx = flb_calloc(1, sizeof(struct flb_syslog));
+
+    if (ctx == NULL) {
+        flb_errno();
+
+        return NULL;
+    }
+
+    ctx->ins = ins;
+
+    ctx->log_encoder = flb_log_event_encoder_create(FLB_LOG_EVENT_FORMAT_DEFAULT);
+
+    if (ctx->log_encoder == NULL) {
+        flb_plg_error(ins, "could not initialize event encoder");
+        syslog_conf_destroy(ctx);
+
+        return NULL;
+    }
+
+    mk_list_init(&ctx->connections);
+
+    ret = flb_input_config_map_set(ins, (void *)ctx);
+    if (ret == -1) {
+        flb_log_event_encoder_destroy(ctx->log_encoder);
+
+        flb_plg_error(ins, "unable to load configuration");
+        flb_free(ctx);
+
+        return NULL;
+    }
+
+    /* Syslog mode: unix_udp, unix_tcp, tcp or udp */
+    if (ctx->mode_str) {
+#ifdef FLB_SYSTEM_WINDOWS
+        if (strcasestr(ctx->mode_str, "unix") != NULL) {
+            flb_log_event_encoder_destroy(ctx->log_encoder);
+
+            flb_plg_error(ins, "unix sockets are note available in windows");
+            flb_free(ctx);
+
+            return NULL;
+        }
+
+#undef FLB_SYSLOG_UNIX_UDP
+#define FLB_SYSLOG_UNIX_UDP FLB_SYSLOG_UDP
+#endif
+        if (strcasecmp(ctx->mode_str, "unix_tcp") == 0) {
+            ctx->mode = FLB_SYSLOG_UNIX_TCP;
+        }
+        else if (strcasecmp(ctx->mode_str, "unix_udp") == 0) {
+            ctx->mode = FLB_SYSLOG_UNIX_UDP;
+        }
+        else if (strcasecmp(ctx->mode_str, "tcp") == 0) {
+            ctx->mode = FLB_SYSLOG_TCP;
+        }
+        else if (strcasecmp(ctx->mode_str, "udp") == 0) {
+            ctx->mode = FLB_SYSLOG_UDP;
+        }
+        else {
+            flb_log_event_encoder_destroy(ctx->log_encoder);
+
+            flb_error("[in_syslog] Unknown syslog mode %s", ctx->mode_str);
+            flb_free(ctx);
+            return NULL;
+        }
+    }
+    else {
+        ctx->mode = FLB_SYSLOG_UNIX_UDP;
+    }
+
+    /* Check if TCP mode was requested */
+    if (ctx->mode == FLB_SYSLOG_TCP || ctx->mode == FLB_SYSLOG_UDP) {
+        /* Listen interface (if not set, defaults to 0.0.0.0:5140) */
+        flb_input_net_default_listener("0.0.0.0", 5140, ins);
+        ctx->listen = ins->host.listen;
+        snprintf(port, sizeof(port) - 1, "%d", ins->host.port);
+        ctx->port = flb_strdup(port);
+    }
+
+    /* Unix socket path and permission */
+    if (ctx->mode == FLB_SYSLOG_UNIX_UDP || ctx->mode == FLB_SYSLOG_UNIX_TCP) {
+        if (ctx->unix_perm_str) {
+            ctx->unix_perm = strtol(ctx->unix_perm_str, NULL, 8) & 07777;
+        } else {
+            ctx->unix_perm = 0644;
+        }
+    }
+
+    /* Buffer Chunk Size */
+    if (ctx->buffer_chunk_size == -1) {
+        flb_log_event_encoder_destroy(ctx->log_encoder);
+
+        flb_plg_error(ins, "invalid buffer_chunk_size");
+        flb_free(ctx);
+        return NULL; 
+    }
+
+    /* Buffer Max Size */
+    if (ctx->buffer_max_size == -1) {
+        flb_log_event_encoder_destroy(ctx->log_encoder);
+
+        flb_plg_error(ins, "invalid buffer_max_size");
+        flb_free(ctx);
+        return NULL;
+    }
+    else if (ctx->buffer_max_size == 0) {
+        ctx->buffer_max_size = ctx->buffer_chunk_size;
+    }
+
+    /* Socket rcv buffer size */
+    if (ctx->receive_buffer_size == -1 || ctx->receive_buffer_size>INT_MAX) {
+        flb_log_event_encoder_destroy(ctx->log_encoder);
+
+        flb_plg_error(ins, "invalid receive_buffer_size");
+        flb_free(ctx);
+        return NULL;
+    }
+
+    /* Parser */
+    if (ctx->parser_name) {
+        ctx->parser = flb_parser_get(ctx->parser_name, config);
+    }
+    else {
+        if (ctx->mode == FLB_SYSLOG_TCP || ctx->mode == FLB_SYSLOG_UDP) {
+            ctx->parser = flb_parser_get("syslog-rfc5424", config);
+        }
+        else {
+            ctx->parser = flb_parser_get("syslog-rfc3164-local", config);
+        }
+    }
+
+    if (!ctx->parser) {
+        flb_log_event_encoder_destroy(ctx->log_encoder);
+
+        flb_error("[in_syslog] parser not set");
+        syslog_conf_destroy(ctx);
+        return NULL;
+    }
+
+    return ctx;
+}
+
+int syslog_conf_destroy(struct flb_syslog *ctx)
+{
+    if (ctx->log_encoder != NULL) {
+        flb_log_event_encoder_destroy(ctx->log_encoder);
+    }
+
+    syslog_server_destroy(ctx);
+
+    flb_free(ctx);
+
+    return 0;
+}
diff --git a/fluent-bit/plugins/in_syslog/syslog_conf.h b/fluent-bit/plugins/in_syslog/syslog_conf.h
new file mode 100644
index 00000000..ac230403
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog_conf.h
@@ -0,0 +1,32 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_IN_SYSLOG_CONF_H
+#define FLB_IN_SYSLOG_CONF_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_input.h>
+
+#include "syslog.h"
+
+struct flb_syslog *syslog_conf_create(struct flb_input_instance *i_ins,
+                                      struct flb_config *config);
+int syslog_conf_destroy(struct flb_syslog *ctx);
+
+#endif
diff --git a/fluent-bit/plugins/in_syslog/syslog_conn.c b/fluent-bit/plugins/in_syslog/syslog_conn.c
new file mode 100644
index 00000000..8785c1e8
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog_conn.c
@@ -0,0 +1,247 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_input_plugin.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_engine.h>
+#include <fluent-bit/flb_network.h>
+#include <fluent-bit/flb_downstream.h>
+
+#include "syslog.h"
+#include "syslog_conf.h"
+#include "syslog_conn.h"
+#include "syslog_prot.h"
+
+/* Callback invoked every time an event is triggered for a connection */
+int syslog_conn_event(void *data)
+{
+    struct flb_connection *connection;
+    struct syslog_conn    *conn;
+    struct flb_syslog     *ctx;
+
+    connection = (struct flb_connection *) data;
+
+    conn = connection->user_data;
+
+    ctx = conn->ctx;
+
+    if (ctx->dgram_mode_flag) {
+        return syslog_dgram_conn_event(data);
+    }
+
+    return syslog_stream_conn_event(data);
+}
+
+int syslog_stream_conn_event(void *data)
+{
+    int ret;
+    int bytes;
+    int available;
+    size_t size;
+    char *tmp;
+    struct mk_event *event;
+    struct syslog_conn *conn;
+    struct flb_syslog *ctx;
+    struct flb_connection *connection;
+
+    connection = (struct flb_connection *) data;
+
+    conn = connection->user_data;
+
+    ctx = conn->ctx;
+
+    event = &connection->event;
+
+    if (event->mask & MK_EVENT_READ) {
+        available = (conn->buf_size - conn->buf_len) - 1;
+        if (available < 1) {
+            if (conn->buf_size + ctx->buffer_chunk_size > ctx->buffer_max_size) {
+                flb_plg_debug(ctx->ins,
+                              "fd=%i incoming data exceed limit (%zd bytes)",
+                              event->fd, (ctx->buffer_max_size));
+                syslog_conn_del(conn);
+                return -1;
+            }
+
+            size = conn->buf_size + ctx->buffer_chunk_size;
+            tmp = flb_realloc(conn->buf_data, size);
+            if (!tmp) {
+                flb_errno();
+                return -1;
+            }
+            flb_plg_trace(ctx->ins, "fd=%i buffer realloc %zd -> %zd",
+                          event->fd, conn->buf_size, size);
+
+            conn->buf_data = tmp;
+            conn->buf_size = size;
+            available = (conn->buf_size - conn->buf_len) - 1;
+        }
+
+        bytes = flb_io_net_read(connection,
+                                (void *) &conn->buf_data[conn->buf_len],
+                                available);
+
+        if (bytes > 0) {
+            flb_plg_trace(ctx->ins, "read()=%i pre_len=%zu now_len=%zu",
+                          bytes, conn->buf_len, conn->buf_len + bytes);
+            conn->buf_len += bytes;
+            conn->buf_data[conn->buf_len] = '\0';
+            ret = syslog_prot_process(conn);
+            if (ret == -1) {
+                return -1;
+            }
+            return bytes;
+        }
+        else {
+            flb_plg_trace(ctx->ins, "fd=%i closed connection", event->fd);
+            syslog_conn_del(conn);
+            return -1;
+        }
+    }
+
+    if (event->mask & MK_EVENT_CLOSE) {
+        flb_plg_trace(ctx->ins, "fd=%i hangup", event->fd);
+        syslog_conn_del(conn);
+        return -1;
+    }
+    return 0;
+}
+
+int syslog_dgram_conn_event(void *data)
+{
+    struct flb_connection *connection;
+    int                    bytes;
+    struct syslog_conn    *conn;
+
+    connection = (struct flb_connection *) data;
+
+    conn = connection->user_data;
+
+    bytes = flb_io_net_read(connection,
+                            (void *) &conn->buf_data[conn->buf_len],
+                            conn->buf_size - 1);
+
+    if (bytes > 0) {
+        conn->buf_data[bytes] = '\0';
+        conn->buf_len = bytes;
+
+        syslog_prot_process_udp(conn);
+    }
+    else {
+        flb_errno();
+    }
+
+    conn->buf_len = 0;
+
+    return 0;
+}
+
+/* Create a new mqtt request instance */
+struct syslog_conn *syslog_conn_add(struct flb_connection *connection,
+                                    struct flb_syslog *ctx)
+{
+    int ret;
+    struct syslog_conn *conn;
+
+    conn = flb_malloc(sizeof(struct syslog_conn));
+    if (!conn) {
+        return NULL;
+    }
+
+    conn->connection = connection;
+
+    /* Set data for the event-loop */
+    MK_EVENT_NEW(&connection->event);
+
+    connection->user_data     = conn;
+    connection->event.type    = FLB_ENGINE_EV_CUSTOM;
+    connection->event.handler = syslog_conn_event;
+
+    /* Connection info */
+    conn->ctx     = ctx;
+    conn->ins     = ctx->ins;
+    conn->buf_len = 0;
+    conn->buf_parsed = 0;
+
+    /* Allocate read buffer */
+    conn->buf_data = flb_malloc(ctx->buffer_chunk_size);
+    if (!conn->buf_data) {
+        flb_errno();
+
+        flb_free(conn);
+
+        return NULL;
+    }
+    conn->buf_size = ctx->buffer_chunk_size;
+
+    /* Register instance into the event loop if we're in
+     * stream mode (UDP events are received through the collector)
+     */
+    if (!ctx->dgram_mode_flag) {
+        ret = mk_event_add(flb_engine_evl_get(),
+                           connection->fd,
+                           FLB_ENGINE_EV_CUSTOM,
+                           MK_EVENT_READ,
+                           &connection->event);
+        if (ret == -1) {
+            flb_plg_error(ctx->ins, "could not register new connection");
+
+            flb_free(conn->buf_data);
+            flb_free(conn);
+
+            return NULL;
+        }
+    }
+
+    mk_list_add(&conn->_head, &ctx->connections);
+
+    return conn;
+}
+
+int syslog_conn_del(struct syslog_conn *conn)
+{
+    /* The downstream unregisters the file descriptor from the event-loop
+     * so there's nothing to be done by the plugin
+     */
+    if (!conn->ctx->dgram_mode_flag) {
+        flb_downstream_conn_release(conn->connection);
+    }
+
+    /* Release resources */
+    mk_list_del(&conn->_head);
+
+    flb_free(conn->buf_data);
+    flb_free(conn);
+
+    return 0;
+}
+
+int syslog_conn_exit(struct flb_syslog *ctx)
+{
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct syslog_conn *conn;
+
+    mk_list_foreach_safe(head, tmp, &ctx->connections) {
+        conn = mk_list_entry(head, struct syslog_conn, _head);
+        syslog_conn_del(conn);
+    }
+
+    return 0;
+}
diff --git a/fluent-bit/plugins/in_syslog/syslog_conn.h b/fluent-bit/plugins/in_syslog/syslog_conn.h
new file mode 100644
index 00000000..684d3f9a
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog_conn.h
@@ -0,0 +1,53 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_IN_SYSLOG_CONN_H
+#define FLB_IN_SYSLOG_CONN_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_connection.h>
+
+#include "syslog.h"
+
+/* Respresents a connection */
+struct syslog_conn {
+    int status;                      /* Connection status                 */
+
+    /* Buffer */
+    char *buf_data;                  /* Buffer data                       */
+    size_t buf_size;                 /* Buffer size                       */
+    size_t buf_len;                  /* Buffer length                     */
+    size_t buf_parsed;               /* Parsed buffer (offset)            */
+    struct flb_input_instance *ins;  /* Parent plugin instance            */
+    struct flb_syslog *ctx;          /* Plugin configuration context      */
+    struct flb_connection *connection;
+
+    struct mk_list _head;
+};
+
+int syslog_conn_event(void *data);
+int syslog_stream_conn_event(void *data);
+int syslog_dgram_conn_event(void *data);
+struct syslog_conn *syslog_conn_add(struct flb_connection *connection,
+                                    struct flb_syslog *ctx);
+int syslog_conn_del(struct syslog_conn *conn);
+int syslog_conn_exit(struct flb_syslog *ctx);
+
+#endif
diff --git a/fluent-bit/plugins/in_syslog/syslog_prot.c b/fluent-bit/plugins/in_syslog/syslog_prot.c
new file mode 100644
index 00000000..1ec2c97c
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog_prot.c
@@ -0,0 +1,324 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_input_plugin.h>
+#include <fluent-bit/flb_parser.h>
+#include <fluent-bit/flb_time.h>
+#include <fluent-bit/flb_pack.h>
+
+#include "syslog.h"
+#include "syslog_conn.h"
+#include "syslog_prot.h"
+
+#include <string.h>
+
+static inline void consume_bytes(char *buf, int bytes, int length)
+{
+    memmove(buf, buf + bytes, length - bytes);
+}
+
+static int append_message_to_record_data(char **result_buffer,
+                                         size_t *result_size,
+                                         flb_sds_t message_key_name,
+                                         char *base_object_buffer,
+                                         size_t base_object_size,
+                                         char *message_buffer,
+                                         size_t message_size,
+                                         int message_type)
+{
+    int                result = FLB_MAP_NOT_MODIFIED;
+    char              *modified_data_buffer;
+    int                modified_data_size;
+    msgpack_object_kv *new_map_entries[1];
+    msgpack_object_kv  message_entry;
+    *result_buffer = NULL;
+    *result_size = 0;
+    modified_data_buffer = NULL;
+
+    if (message_key_name != NULL) {
+        new_map_entries[0] = &message_entry;
+
+        message_entry.key.type = MSGPACK_OBJECT_STR;
+        message_entry.key.via.str.size = flb_sds_len(message_key_name);
+        message_entry.key.via.str.ptr  = message_key_name;
+
+        if (message_type == MSGPACK_OBJECT_BIN) {
+            message_entry.val.type = MSGPACK_OBJECT_BIN;
+            message_entry.val.via.bin.size = message_size;
+            message_entry.val.via.bin.ptr  = message_buffer;
+        }
+        else if (message_type == MSGPACK_OBJECT_STR) {
+            message_entry.val.type = MSGPACK_OBJECT_STR;
+            message_entry.val.via.str.size = message_size;
+            message_entry.val.via.str.ptr  = message_buffer;
+        }
+        else {
+            result = FLB_MAP_EXPANSION_INVALID_VALUE_TYPE;
+        }
+
+        if (result == FLB_MAP_NOT_MODIFIED) {
+            result = flb_msgpack_expand_map(base_object_buffer,
+                                            base_object_size,
+                                            new_map_entries, 1,
+                                            &modified_data_buffer,
+                                            &modified_data_size);
+            if (result == 0) {
+                result = FLB_MAP_EXPAND_SUCCESS;
+            }
+            else {
+                result = FLB_MAP_EXPANSION_ERROR;
+            }
+        }
+    }
+
+    if (result == FLB_MAP_EXPAND_SUCCESS) {
+        *result_buffer = modified_data_buffer;
+        *result_size = modified_data_size;
+    }
+
+    return result;
+}
+
+static inline int pack_line(struct flb_syslog *ctx,
+                            struct flb_time *time,
+                            struct flb_connection *connection,
+                            char *data, size_t data_size,
+                            char *raw_data, size_t raw_data_size)
+{
+    char   *modified_data_buffer;
+    size_t  modified_data_size;
+    char   *appended_address_buffer;
+    size_t  appended_address_size;
+    int     result;
+    char   *source_address;
+
+    source_address = NULL;
+    modified_data_buffer = NULL;
+    appended_address_buffer = NULL;
+
+    if (ctx->raw_message_key != NULL) {
+        result = append_message_to_record_data(&modified_data_buffer,
+                                               &modified_data_size,
+                                               ctx->raw_message_key,
+                                               data,
+                                               data_size,
+                                               raw_data,
+                                               raw_data_size,
+                                               MSGPACK_OBJECT_BIN);
+
+        if (result == FLB_MAP_EXPANSION_ERROR) {
+            flb_plg_debug(ctx->ins, "error expanding raw message : %d", result);
+        }
+    }
+
+    if (ctx->source_address_key != NULL) {
+        source_address = flb_connection_get_remote_address(connection);
+        if (source_address != NULL) {
+            if (modified_data_buffer != NULL) {
+                result = append_message_to_record_data(&appended_address_buffer,
+                                                       &appended_address_size,
+                                                       ctx->source_address_key,
+                                                       modified_data_buffer,
+                                                       modified_data_size,
+                                                       source_address,
+                                                       strlen(source_address),
+                                                       MSGPACK_OBJECT_STR);
+            }
+            else {
+                result = append_message_to_record_data(&appended_address_buffer,
+                                                       &appended_address_size,
+                                                       ctx->source_address_key,
+                                                       data,
+                                                       data_size,
+                                                       source_address,
+                                                       strlen(source_address),
+                                                       MSGPACK_OBJECT_STR);
+            }
+
+            if (result == FLB_MAP_EXPANSION_ERROR) {
+                flb_plg_debug(ctx->ins, "error expanding source_address : %d", result);
+            }
+        }
+    }
+
+    result = flb_log_event_encoder_begin_record(ctx->log_encoder);
+
+    if (result == FLB_EVENT_ENCODER_SUCCESS) {
+        result = flb_log_event_encoder_set_timestamp(ctx->log_encoder, time);
+    }
+
+    if (result == FLB_EVENT_ENCODER_SUCCESS) {
+        if (appended_address_buffer != NULL) {
+            result = flb_log_event_encoder_set_body_from_raw_msgpack(
+                    ctx->log_encoder, appended_address_buffer, appended_address_size);
+        }
+        else if (modified_data_buffer != NULL) {
+            result = flb_log_event_encoder_set_body_from_raw_msgpack(
+                    ctx->log_encoder, modified_data_buffer, modified_data_size);
+        }
+        else {
+            result = flb_log_event_encoder_set_body_from_raw_msgpack(
+                        ctx->log_encoder, data, data_size);
+        }
+    }
+
+    if (result == FLB_EVENT_ENCODER_SUCCESS) {
+        result = flb_log_event_encoder_commit_record(ctx->log_encoder);
+    }
+
+    if (result == FLB_EVENT_ENCODER_SUCCESS) {
+        flb_input_log_append(ctx->ins, NULL, 0,
+                             ctx->log_encoder->output_buffer,
+                             ctx->log_encoder->output_length);
+        result = 0;
+    }
+    else {
+        flb_plg_error(ctx->ins, "log event encoding error : %d", result);
+
+        result = -1;
+    }
+
+    flb_log_event_encoder_reset(ctx->log_encoder);
+
+    if (modified_data_buffer != NULL) {
+        flb_free(modified_data_buffer);
+    }
+    if (appended_address_buffer != NULL) {
+        flb_free(appended_address_buffer);
+    }
+
+    return result;
+}
+
+int syslog_prot_process(struct syslog_conn *conn)
+{
+    int len;
+    int ret;
+    char *p;
+    char *eof;
+    char *end;
+    void *out_buf;
+    size_t out_size;
+    struct flb_time out_time;
+    struct flb_syslog *ctx = conn->ctx;
+
+    eof = conn->buf_data;
+    end = conn->buf_data + conn->buf_len;
+
+    /* Always parse while some remaining bytes exists */
+    while (eof < end) {
+        /* Lookup the ending byte */
+        eof = p = conn->buf_data + conn->buf_parsed;
+        while (*eof != '\n' && *eof != '\0' && eof < end) {
+            eof++;
+        }
+
+        /* Incomplete message */
+        if (eof == end || (*eof != '\n' && *eof != '\0')) {
+            break;
+        }
+
+        /* No data ? */
+        len = (eof - p);
+        if (len == 0) {
+            consume_bytes(conn->buf_data, 1, conn->buf_len);
+            conn->buf_len--;
+            conn->buf_parsed = 0;
+            conn->buf_data[conn->buf_len] = '\0';
+            end = conn->buf_data + conn->buf_len;
+
+            if (conn->buf_len == 0) {
+                break;
+            }
+
+            continue;
+        }
+
+        /* Process the string */
+        ret = flb_parser_do(ctx->parser, p, len,
+                            &out_buf, &out_size, &out_time);
+        if (ret >= 0) {
+            if (flb_time_to_nanosec(&out_time) == 0L) {
+                flb_time_get(&out_time);
+            }
+            pack_line(ctx, &out_time,
+                      conn->connection,
+                      out_buf, out_size,
+                      p, len);
+            flb_free(out_buf);
+        }
+        else {
+            flb_plg_warn(ctx->ins, "error parsing log message with parser '%s'",
+                         ctx->parser->name);
+            flb_plg_debug(ctx->ins, "unparsed log message: %.*s", len, p);
+        }
+
+        conn->buf_parsed += len + 1;
+        end = conn->buf_data + conn->buf_len;
+        eof = conn->buf_data + conn->buf_parsed;
+    }
+
+    if (conn->buf_parsed > 0) {
+        consume_bytes(conn->buf_data, conn->buf_parsed, conn->buf_len);
+        conn->buf_len -= conn->buf_parsed;
+        conn->buf_parsed = 0;
+        conn->buf_data[conn->buf_len] = '\0';
+    }
+
+    return 0;
+}
+
+int syslog_prot_process_udp(struct syslog_conn *conn)
+{
+    int ret;
+    void *out_buf;
+    size_t out_size;
+    struct flb_time out_time = {0};
+    char *buf;
+    size_t size;
+    struct flb_syslog *ctx;
+    struct flb_connection *connection;
+
+    buf = conn->buf_data;
+    size = conn->buf_len;
+    ctx = conn->ctx;
+    connection = conn->connection;
+
+    ret = flb_parser_do(ctx->parser, buf, size,
+                        &out_buf, &out_size, &out_time);
+    if (ret >= 0) {
+        if (flb_time_to_double(&out_time) == 0) {
+            flb_time_get(&out_time);
+        }
+        pack_line(ctx, &out_time,
+                  connection,
+                  out_buf, out_size,
+                  buf, size);
+        flb_free(out_buf);
+    }
+    else {
+        flb_plg_warn(ctx->ins, "error parsing log message with parser '%s'",
+                     ctx->parser->name);
+        flb_plg_debug(ctx->ins, "unparsed log message: %.*s",
+                      (int) size, buf);
+        return -1;
+    }
+
+    return 0;
+}
diff --git a/fluent-bit/plugins/in_syslog/syslog_prot.h b/fluent-bit/plugins/in_syslog/syslog_prot.h
new file mode 100644
index 00000000..cb5976b7
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog_prot.h
@@ -0,0 +1,35 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_IN_SYSLOG_PROT_H
+#define FLB_IN_SYSLOG_PROT_H
+
+#include <fluent-bit/flb_info.h>
+
+#include "syslog.h"
+
+#define FLB_MAP_EXPAND_SUCCESS   0
+#define FLB_MAP_NOT_MODIFIED    -1
+#define FLB_MAP_EXPANSION_ERROR -2
+#define FLB_MAP_EXPANSION_INVALID_VALUE_TYPE -3
+
+int syslog_prot_process(struct syslog_conn *conn);
+int syslog_prot_process_udp(struct syslog_conn *conn);
+
+#endif
diff --git a/fluent-bit/plugins/in_syslog/syslog_server.c b/fluent-bit/plugins/in_syslog/syslog_server.c
new file mode 100644
index 00000000..5317851d
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog_server.c
@@ -0,0 +1,235 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_macros.h>
+#include <fluent-bit/flb_log.h>
+#include <fluent-bit/flb_socket.h>
+#include <fluent-bit/flb_network.h>
+#include <fluent-bit/tls/flb_tls.h>
+#include <fluent-bit/flb_downstream.h>
+#include <fluent-bit/flb_input_plugin.h>
+
+#if !defined(FLB_SYSTEM_WINDOWS)
+#include <unistd.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#endif
+#include <sys/stat.h>
+#include <sys/types.h>
+
+#include "syslog.h"
+
+static int remove_existing_socket_file(char *socket_path)
+{
+    struct stat file_data;
+    int         result;
+
+    result = stat(socket_path, &file_data);
+
+    if (result == -1) {
+        if (errno == ENOENT) {
+            return 0;
+        }
+
+        flb_errno();
+
+        return -1;
+    }
+
+    if (S_ISSOCK(file_data.st_mode) == 0) {
+        return -2;
+    }
+
+    result = unlink(socket_path);
+
+    if (result != 0) {
+        return -3;
+    }
+
+    return 0;
+}
+
+#if !defined(FLB_SYSTEM_WINDOWS)
+static int syslog_server_unix_create(struct flb_syslog *ctx)
+{
+    int             result;
+    int             mode;
+    struct flb_tls *tls;
+
+    if (ctx->mode == FLB_SYSLOG_UNIX_TCP) {
+        mode = FLB_TRANSPORT_UNIX_STREAM;
+        tls = ctx->ins->tls;
+    }
+    else if (ctx->mode == FLB_SYSLOG_UNIX_UDP) {
+        ctx->dgram_mode_flag = FLB_TRUE;
+
+        mode = FLB_TRANSPORT_UNIX_DGRAM;
+        tls = NULL;
+    }
+    else {
+        return -1;
+    }
+
+    result = remove_existing_socket_file(ctx->unix_path);
+
+    if (result != 0) {
+        if (result == -2) {
+            flb_plg_error(ctx->ins,
+                          "%s exists and it is not a unix socket. Aborting",
+                          ctx->unix_path);
+        }
+        else {
+            flb_plg_error(ctx->ins,
+                          "could not remove existing unix socket %s. Aborting",
+                          ctx->unix_path);
+        }
+
+        return -1;
+    }
+
+    ctx->downstream = flb_downstream_create(mode,
+                                            ctx->ins->flags,
+                                            ctx->unix_path,
+                                            0,
+                                            tls,
+                                            ctx->ins->config,
+                                            &ctx->ins->net_setup);
+
+    if (ctx->downstream == NULL) {
+        return -1;
+    }
+
+    if (chmod(ctx->unix_path, ctx->unix_perm)) {
+        flb_errno();
+        flb_error("[in_syslog] cannot set permission on '%s' to %04o",
+                  ctx->unix_path, ctx->unix_perm);
+
+        return -1;
+    }
+
+    return 0;
+}
+#else
+static int syslog_server_unix_create(struct flb_syslog *ctx)
+{
+    return -1;
+}
+#endif
+
+static int syslog_server_net_create(struct flb_syslog *ctx)
+{
+    unsigned short int port;
+    int                mode;
+    struct flb_tls    *tls;
+
+    port = (unsigned short int) strtoul(ctx->port, NULL, 10);
+
+    if (ctx->mode == FLB_SYSLOG_TCP) {
+        mode = FLB_TRANSPORT_TCP;
+        tls = ctx->ins->tls;
+    }
+    else if (ctx->mode == FLB_SYSLOG_UDP) {
+        ctx->dgram_mode_flag = FLB_TRUE;
+
+        mode = FLB_TRANSPORT_UDP;
+        tls = NULL;
+    }
+    else {
+        return -1;
+    }
+
+    ctx->downstream = flb_downstream_create(mode,
+                                            ctx->ins->flags,
+                                            ctx->listen,
+                                            port,
+                                            tls,
+                                            ctx->ins->config,
+                                            &ctx->ins->net_setup);
+
+    if (ctx->downstream != NULL) {
+        flb_info("[in_syslog] %s server binding %s:%s",
+                 ((ctx->mode == FLB_SYSLOG_TCP) ? "TCP" : "UDP"),
+                 ctx->listen, ctx->port);
+    }
+    else {
+        flb_error("[in_syslog] could not bind address %s:%s. Aborting",
+                  ctx->listen, ctx->port);
+
+        return -1;
+    }
+
+    if (ctx->receive_buffer_size) {
+        if (flb_net_socket_rcv_buffer(ctx->downstream->server_fd,
+                                      ctx->receive_buffer_size)) {
+            flb_error("[in_syslog] could not set rcv buffer to %ld. Aborting",
+                      ctx->receive_buffer_size);
+            return -1;
+        }
+    }
+
+    flb_net_socket_nonblocking(ctx->downstream->server_fd);
+
+    return 0;
+}
+
+int syslog_server_create(struct flb_syslog *ctx)
+{
+    int ret;
+
+    if (ctx->mode == FLB_SYSLOG_TCP || ctx->mode == FLB_SYSLOG_UDP) {
+        ret = syslog_server_net_create(ctx);
+    }
+    else {
+        /* Create unix socket end-point */
+        ret = syslog_server_unix_create(ctx);
+    }
+
+    if (ret != 0) {
+        return -1;
+    }
+
+    return 0;
+}
+
+int syslog_server_destroy(struct flb_syslog *ctx)
+{
+    if (ctx->collector_id != -1) {
+        flb_input_collector_delete(ctx->collector_id, ctx->ins);
+
+        ctx->collector_id = -1;
+    }
+
+    if (ctx->downstream != NULL) {
+        flb_downstream_destroy(ctx->downstream);
+
+        ctx->downstream = NULL;
+    }
+
+    if (ctx->mode == FLB_SYSLOG_UNIX_TCP || ctx->mode == FLB_SYSLOG_UNIX_UDP) {
+        if (ctx->unix_path) {
+            unlink(ctx->unix_path);
+        }
+    }
+    else {
+        flb_free(ctx->port);
+    }
+
+    return 0;
+}
diff --git a/fluent-bit/plugins/in_syslog/syslog_server.h b/fluent-bit/plugins/in_syslog/syslog_server.h
new file mode 100644
index 00000000..d14feba7
--- /dev/null
+++ b/fluent-bit/plugins/in_syslog/syslog_server.h
@@ -0,0 +1,31 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_IN_SYSLOG_SERVER_H
+#define FLB_IN_SYSLOG_SERVER_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_config.h>
+
+#include "syslog.h"
+
+int syslog_server_create(struct flb_syslog *ctx);
+int syslog_server_destroy(struct flb_syslog *ctx);
+
+#endif