diff options
Diffstat (limited to 'src/fluent-bit/plugins/in_syslog')
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/CMakeLists.txt | 8 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog.c | 263 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog.h | 82 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog_conf.c | 193 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog_conf.h | 32 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog_conn.c | 247 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog_conn.h | 53 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog_prot.c | 324 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog_prot.h | 35 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog_server.c | 235 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_syslog/syslog_server.h | 31 |
11 files changed, 1503 insertions, 0 deletions
diff --git a/src/fluent-bit/plugins/in_syslog/CMakeLists.txt b/src/fluent-bit/plugins/in_syslog/CMakeLists.txt new file mode 100644 index 000000000..88f698b12 --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog.c b/src/fluent-bit/plugins/in_syslog/syslog.c new file mode 100644 index 000000000..d478dfc3f --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog.h b/src/fluent-bit/plugins/in_syslog/syslog.h new file mode 100644 index 000000000..6da2fbd83 --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog_conf.c b/src/fluent-bit/plugins/in_syslog/syslog_conf.c new file mode 100644 index 000000000..4db3f1626 --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog_conf.h b/src/fluent-bit/plugins/in_syslog/syslog_conf.h new file mode 100644 index 000000000..ac2304031 --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog_conn.c b/src/fluent-bit/plugins/in_syslog/syslog_conn.c new file mode 100644 index 000000000..8785c1e86 --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog_conn.h b/src/fluent-bit/plugins/in_syslog/syslog_conn.h new file mode 100644 index 000000000..684d3f9a4 --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog_prot.c b/src/fluent-bit/plugins/in_syslog/syslog_prot.c new file mode 100644 index 000000000..1ec2c97cd --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog_prot.h b/src/fluent-bit/plugins/in_syslog/syslog_prot.h new file mode 100644 index 000000000..cb5976b7b --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog_server.c b/src/fluent-bit/plugins/in_syslog/syslog_server.c new file mode 100644 index 000000000..5317851db --- /dev/null +++ b/src/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/src/fluent-bit/plugins/in_syslog/syslog_server.h b/src/fluent-bit/plugins/in_syslog/syslog_server.h new file mode 100644 index 000000000..d14feba7c --- /dev/null +++ b/src/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 |