diff options
Diffstat (limited to 'src/fluent-bit/plugins/in_udp')
| -rw-r--r-- | src/fluent-bit/plugins/in_udp/CMakeLists.txt | 6 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_udp/udp.c | 197 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_udp/udp.h | 54 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_udp/udp_config.c | 155 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_udp/udp_config.h | 28 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_udp/udp_conn.c | 500 | ||||
| -rw-r--r-- | src/fluent-bit/plugins/in_udp/udp_conn.h | 57 |
7 files changed, 997 insertions, 0 deletions
diff --git a/src/fluent-bit/plugins/in_udp/CMakeLists.txt b/src/fluent-bit/plugins/in_udp/CMakeLists.txt new file mode 100644 index 000000000..0b623f169 --- /dev/null +++ b/src/fluent-bit/plugins/in_udp/CMakeLists.txt @@ -0,0 +1,6 @@ +set(src + udp.c + udp_conn.c + udp_config.c) + +FLB_PLUGIN(in_udp "${src}" "") diff --git a/src/fluent-bit/plugins/in_udp/udp.c b/src/fluent-bit/plugins/in_udp/udp.c new file mode 100644 index 000000000..ad5a28497 --- /dev/null +++ b/src/fluent-bit/plugins/in_udp/udp.c @@ -0,0 +1,197 @@ +/* -*- 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_network.h> +#include <msgpack.h> + +#include "udp.h" +#include "udp_conn.h" +#include "udp_config.h" + +static int in_udp_collect(struct flb_input_instance *in, + struct flb_config *config, + void *in_context) +{ + struct flb_connection *connection; + struct flb_in_udp_config *ctx; + + ctx = in_context; + + connection = flb_downstream_conn_get(ctx->downstream); + + if (connection == NULL) { + flb_plg_error(ctx->ins, "could get UDP server dummy connection"); + + return -1; + } + + return udp_conn_event(connection); +} + +/* Initialize plugin */ +static int in_udp_init(struct flb_input_instance *in, + struct flb_config *config, void *data) +{ + struct flb_connection *connection; + unsigned short int port; + int ret; + struct flb_in_udp_config *ctx; + + (void) data; + + /* Allocate space for the configuration */ + ctx = udp_config_init(in); + + if (ctx == NULL) { + return -1; + } + + ctx->collector_id = -1; + ctx->ins = in; + + /* Set the context */ + flb_input_set_context(in, ctx); + + port = (unsigned short int) strtoul(ctx->port, NULL, 10); + + ctx->downstream = flb_downstream_create(FLB_TRANSPORT_UDP, + in->flags, + ctx->listen, + port, + in->tls, + config, + &in->net_setup); + + if (ctx->downstream == NULL) { + flb_plg_error(ctx->ins, + "could not initialize downstream on %s:%s. Aborting", + ctx->listen, ctx->port); + + udp_config_destroy(ctx); + + return -1; + } + + flb_input_downstream_set(ctx->downstream, ctx->ins); + + connection = flb_downstream_conn_get(ctx->downstream); + + if (connection == NULL) { + flb_plg_error(ctx->ins, "could not get UDP server dummy connection"); + + udp_config_destroy(ctx); + + return -1; + } + + ctx->dummy_conn = udp_conn_add(connection, ctx); + + if (ctx->dummy_conn == NULL) { + flb_plg_error(ctx->ins, "could not track UDP server dummy connection"); + + udp_config_destroy(ctx); + + return -1; + } + + /* Collect upon data available on the standard input */ + ret = flb_input_set_collector_socket(in, + in_udp_collect, + ctx->downstream->server_fd, + config); + if (ret == -1) { + flb_plg_error(ctx->ins, "Could not set collector for IN_UDP input plugin"); + udp_config_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"); + udp_config_destroy(ctx); + + return -1; + } + + return 0; +} + +static int in_udp_exit(void *data, struct flb_config *config) +{ + struct flb_in_udp_config *ctx; + + (void) *config; + + ctx = data; + + if (ctx->dummy_conn != NULL) { + udp_conn_del(ctx->dummy_conn); + } + + udp_config_destroy(ctx); + + return 0; +} + +static struct flb_config_map config_map[] = { + { + FLB_CONFIG_MAP_STR, "format", (char *)NULL, + 0, FLB_TRUE, offsetof(struct flb_in_udp_config, format_name), + "Set the format: json or none" + }, + { + FLB_CONFIG_MAP_STR, "separator", (char *)NULL, + 0, FLB_TRUE, offsetof(struct flb_in_udp_config, raw_separator), + "Set separator" + }, + { + FLB_CONFIG_MAP_STR, "chunk_size", (char *)NULL, + 0, FLB_TRUE, offsetof(struct flb_in_udp_config, chunk_size_str), + "Set the chunk size" + }, + { + FLB_CONFIG_MAP_STR, "buffer_size", (char *)NULL, + 0, FLB_TRUE, offsetof(struct flb_in_udp_config, buffer_size_str), + "Set the buffer size" + }, + { + FLB_CONFIG_MAP_STR, "source_address_key", (char *) NULL, + 0, FLB_TRUE, offsetof(struct flb_in_udp_config, source_address_key), + "Key where the source address will be injected" + }, + /* EOF */ + {0} +}; + +/* Plugin reference */ +struct flb_input_plugin in_udp_plugin = { + .name = "udp", + .description = "UDP", + .cb_init = in_udp_init, + .cb_pre_run = NULL, + .cb_collect = in_udp_collect, + .cb_flush_buf = NULL, + .cb_exit = in_udp_exit, + .config_map = config_map, + .flags = FLB_INPUT_NET_SERVER, +}; diff --git a/src/fluent-bit/plugins/in_udp/udp.h b/src/fluent-bit/plugins/in_udp/udp.h new file mode 100644 index 000000000..1a7bfce30 --- /dev/null +++ b/src/fluent-bit/plugins/in_udp/udp.h @@ -0,0 +1,54 @@ +/* -*- 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_UDP_H +#define FLB_IN_UDP_H + +#define FLB_UDP_FMT_JSON 0 /* default */ +#define FLB_UDP_FMT_NONE 1 /* no format, use delimiters */ + +#include <fluent-bit/flb_downstream.h> +#include <fluent-bit/flb_input.h> +#include <fluent-bit/flb_sds.h> +#include <fluent-bit/flb_log_event_encoder.h> +#include <msgpack.h> + +struct udp_conn; + +struct flb_in_udp_config { + struct mk_event *collector_event; + flb_sds_t format_name; /* Data format name */ + int format; /* Data format */ + size_t buffer_size; /* Buffer size for each reader */ + flb_sds_t buffer_size_str; /* Buffer size in string form */ + size_t chunk_size; /* Chunk allocation size */ + flb_sds_t chunk_size_str; /* Chunk size in string form */ + char *listen; /* Listen interface */ + char *port; /* Port */ + flb_sds_t raw_separator; /* Unescaped string delimiterr */ + flb_sds_t separator; /* String delimiter */ + flb_sds_t source_address_key; /* Source IP address */ + int collector_id; /* Listener collector id */ + struct flb_downstream *downstream; /* Client manager */ + struct udp_conn *dummy_conn; /* Datagram dummy connection */ + struct flb_input_instance *ins; /* Input plugin instace */ + struct flb_log_event_encoder *log_encoder; +}; + +#endif diff --git a/src/fluent-bit/plugins/in_udp/udp_config.c b/src/fluent-bit/plugins/in_udp/udp_config.c new file mode 100644 index 000000000..ad2995490 --- /dev/null +++ b/src/fluent-bit/plugins/in_udp/udp_config.c @@ -0,0 +1,155 @@ +/* -*- 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_unescape.h> + +#include "udp.h" +#include "udp_conn.h" +#include "udp_config.h" + +#include <stdlib.h> + +struct flb_in_udp_config *udp_config_init(struct flb_input_instance *ins) +{ + int ret; + int len; + char port[16]; + char *out; + struct flb_in_udp_config *ctx; + + /* Allocate plugin context */ + ctx = flb_calloc(1, sizeof(struct flb_in_udp_config)); + if (!ctx) { + flb_errno(); + return NULL; + } + ctx->ins = ins; + ctx->format = FLB_UDP_FMT_JSON; + + /* Load the config map */ + ret = flb_input_config_map_set(ins, (void *)ctx); + if (ret == -1) { + flb_plg_error(ins, "unable to load configuration"); + flb_free(ctx); + return NULL; + } + + /* Data format (expected payload) */ + if (ctx->format_name) { + if (strcasecmp(ctx->format_name, "json") == 0) { + ctx->format = FLB_UDP_FMT_JSON; + } + else if (strcasecmp(ctx->format_name, "none") == 0) { + ctx->format = FLB_UDP_FMT_NONE; + } + else { + flb_plg_error(ctx->ins, "unrecognized format value '%s'", ctx->format_name); + flb_free(ctx); + return NULL; + } + } + + /* String separator used to split records when using 'format none' */ + if (ctx->raw_separator) { + len = strlen(ctx->raw_separator); + out = flb_malloc(len + 1); + if (!out) { + flb_errno(); + flb_free(ctx); + return NULL; + } + ret = flb_unescape_string(ctx->raw_separator, len, &out); + if (ret <= 0) { + flb_plg_error(ctx->ins, "invalid separator"); + flb_free(out); + flb_free(ctx); + return NULL; + } + + ctx->separator = flb_sds_create_len(out, ret); + if (!ctx->separator) { + flb_free(out); + flb_free(ctx); + return NULL; + } + flb_free(out); + } + if (!ctx->separator) { + ctx->separator = flb_sds_create_len("\n", 1); + } + + /* Listen interface (if not set, defaults to 0.0.0.0:5170) */ + flb_input_net_default_listener("0.0.0.0", 5170, ins); + ctx->listen = ins->host.listen; + snprintf(port, sizeof(port) - 1, "%d", ins->host.port); + ctx->port = flb_strdup(port); + + /* Chunk size */ + if (ctx->chunk_size_str) { + /* Convert KB unit to Bytes */ + ctx->chunk_size = (atoi(ctx->chunk_size_str) * 1024); + } else { + ctx->chunk_size = atoi(FLB_IN_UDP_CHUNK); + } + + /* Buffer size */ + if (!ctx->buffer_size_str) { + ctx->buffer_size = ctx->chunk_size; + } + else { + /* Convert KB unit to Bytes */ + ctx->buffer_size = (atoi(ctx->buffer_size_str) * 1024); + } + + ctx->log_encoder = flb_log_event_encoder_create(FLB_LOG_EVENT_FORMAT_DEFAULT); + + if (ctx->log_encoder == NULL) { + flb_plg_error(ctx->ins, "could not initialize event encoder"); + udp_config_destroy(ctx); + + ctx = NULL; + } + + return ctx; +} + +int udp_config_destroy(struct flb_in_udp_config *ctx) +{ + if (ctx->log_encoder != NULL) { + flb_log_event_encoder_destroy(ctx->log_encoder); + } + + 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); + } + + flb_sds_destroy(ctx->separator); + flb_free(ctx->port); + flb_free(ctx); + + return 0; +} diff --git a/src/fluent-bit/plugins/in_udp/udp_config.h b/src/fluent-bit/plugins/in_udp/udp_config.h new file mode 100644 index 000000000..dcddb74a7 --- /dev/null +++ b/src/fluent-bit/plugins/in_udp/udp_config.h @@ -0,0 +1,28 @@ +/* -*- 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_UDP_CONFIG_H +#define FLB_IN_UDP_CONFIG_H + +#include "udp.h" + +struct flb_in_udp_config *udp_config_init(struct flb_input_instance *i_ins); +int udp_config_destroy(struct flb_in_udp_config *config); + +#endif diff --git a/src/fluent-bit/plugins/in_udp/udp_conn.c b/src/fluent-bit/plugins/in_udp/udp_conn.c new file mode 100644 index 000000000..d8cc4d5e6 --- /dev/null +++ b/src/fluent-bit/plugins/in_udp/udp_conn.c @@ -0,0 +1,500 @@ +/* -*- 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_pack.h> +#include <fluent-bit/flb_error.h> + +#include "udp.h" +#include "udp_conn.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 process_pack(struct udp_conn *conn, + char *pack, size_t size) +{ + int ret; + size_t off = 0; + msgpack_unpacked result; + msgpack_object entry; + msgpack_sbuffer sbuf; + msgpack_packer pck; + struct flb_in_udp_config *ctx; + char *appended_address_buffer; + size_t appended_address_size; + char *source_address; + int i; + int len; + + ctx = conn->ctx; + + flb_log_event_encoder_reset(ctx->log_encoder); + + /* First pack the results, iterate concatenated messages */ + msgpack_unpacked_init(&result); + while (msgpack_unpack_next(&result, pack, size, &off) == MSGPACK_UNPACK_SUCCESS) { + entry = result.data; + + appended_address_buffer = NULL; + source_address = NULL; + + ret = flb_log_event_encoder_begin_record(ctx->log_encoder); + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_set_current_timestamp(ctx->log_encoder); + } + + if (ctx->source_address_key != NULL) { + source_address = flb_connection_get_remote_address(conn->connection); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + if (entry.type == MSGPACK_OBJECT_MAP) { + if (source_address != NULL) { + msgpack_sbuffer_init(&sbuf); + msgpack_packer_init(&pck, &sbuf, msgpack_sbuffer_write); + + len = entry.via.map.size; + msgpack_pack_map(&pck, len); + + for (i=0; i<len; i++) { + msgpack_pack_object(&pck, entry.via.map.ptr[i].key); + msgpack_pack_object(&pck, entry.via.map.ptr[i].val); + } + + ret = append_message_to_record_data(&appended_address_buffer, + &appended_address_size, + ctx->source_address_key, + sbuf.data, + sbuf.size, + source_address, + strlen(source_address), + MSGPACK_OBJECT_STR); + msgpack_sbuffer_destroy(&sbuf); + } + + if (ret == FLB_MAP_EXPANSION_ERROR) { + flb_plg_debug(ctx->ins, "error expanding source_address : %d", ret); + } + + if (appended_address_buffer != NULL) { + ret = flb_log_event_encoder_set_body_from_raw_msgpack( + ctx->log_encoder, appended_address_buffer, appended_address_size); + } + else { + ret = flb_log_event_encoder_set_body_from_msgpack_object( + ctx->log_encoder, &entry); + } + } + else if (entry.type == MSGPACK_OBJECT_ARRAY) { + if (source_address != NULL) { + ret = flb_log_event_encoder_append_body_values( + ctx->log_encoder, + FLB_LOG_EVENT_CSTRING_VALUE("msg"), + FLB_LOG_EVENT_MSGPACK_OBJECT_VALUE(&entry), + FLB_LOG_EVENT_CSTRING_VALUE(ctx->source_address_key), + FLB_LOG_EVENT_CSTRING_VALUE(source_address)); + } + else { + ret = flb_log_event_encoder_append_body_values( + ctx->log_encoder, + FLB_LOG_EVENT_CSTRING_VALUE("msg"), + FLB_LOG_EVENT_MSGPACK_OBJECT_VALUE(&entry)); + } + } + else { + ret = FLB_EVENT_ENCODER_ERROR_INVALID_VALUE_TYPE; + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_commit_record(ctx->log_encoder); + } + + if (appended_address_buffer != NULL) { + flb_free(appended_address_buffer); + } + + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + break; + } + } + } + + msgpack_unpacked_destroy(&result); + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + flb_input_log_append(conn->ins, NULL, 0, + ctx->log_encoder->output_buffer, + ctx->log_encoder->output_length); + ret = 0; + } + else { + flb_plg_error(ctx->ins, "log event encoding error : %d", ret); + + ret = -1; + } + + return ret; +} + +/* Process a JSON payload, return the number of processed bytes */ +static ssize_t parse_payload_json(struct udp_conn *conn) +{ + int ret; + int out_size; + char *pack; + + ret = flb_pack_json_state(conn->buf_data, conn->buf_len, + &pack, &out_size, &conn->pack_state); + if (ret == FLB_ERR_JSON_PART) { + flb_plg_debug(conn->ins, "JSON incomplete, waiting for more data..."); + return 0; + } + else if (ret == FLB_ERR_JSON_INVAL) { + flb_plg_warn(conn->ins, "invalid JSON message, skipping"); + conn->buf_len = 0; + conn->pack_state.multiple = FLB_TRUE; + return -1; + } + else if (ret == -1) { + return -1; + } + + /* Process the packaged JSON and return the last byte used */ + process_pack(conn, pack, out_size); + flb_free(pack); + + return conn->pack_state.last_byte; +} + +/* + * Process a raw text payload, uses the delimited character to split records, + * return the number of processed bytes + */ +static ssize_t parse_payload_none(struct udp_conn *conn) +{ + int ret; + int len; + int sep_len; + size_t consumed = 0; + char *buf; + char *s; + char *separator; + struct flb_in_udp_config *ctx; + + ctx = conn->ctx; + + separator = conn->ctx->separator; + sep_len = flb_sds_len(conn->ctx->separator); + + buf = conn->buf_data; + ret = FLB_EVENT_ENCODER_SUCCESS; + + flb_log_event_encoder_reset(ctx->log_encoder); + + while ((s = strstr(buf, separator))) { + len = (s - buf); + if (len == 0) { + break; + } + else if (len > 0) { + ret = flb_log_event_encoder_begin_record(ctx->log_encoder); + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_set_current_timestamp(ctx->log_encoder); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_append_body_values( + ctx->log_encoder, + FLB_LOG_EVENT_CSTRING_VALUE("log"), + FLB_LOG_EVENT_STRING_VALUE(buf, len)); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_commit_record(ctx->log_encoder); + } + + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + break; + } + + consumed += len + 1; + buf += len + sep_len; + } + else { + break; + } + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + flb_input_log_append(conn->ins, NULL, 0, + ctx->log_encoder->output_buffer, + ctx->log_encoder->output_length); + } + else { + flb_plg_error(ctx->ins, "log event encoding error : %d", ret); + } + + return consumed; +} + +/* Callback invoked every time an event is triggered for a connection */ +int udp_conn_event(void *data) +{ + int bytes; + int available; + int size; + ssize_t ret_payload = -1; + char *tmp; + struct udp_conn *conn; + struct flb_connection *connection; + struct flb_in_udp_config *ctx; + + connection = (struct flb_connection *) data; + + conn = connection->user_data; + + ctx = conn->ctx; + + if (ctx->format == FLB_UDP_FMT_JSON && + conn->buf_len > 0) { + flb_pack_state_reset(&conn->pack_state); + flb_pack_state_init(&conn->pack_state); + + conn->pack_state.multiple = FLB_TRUE; + } + + conn->buf_len = 0; + + available = (conn->buf_size - conn->buf_len) - 1; + if (available < 1) { + if (conn->buf_size + ctx->chunk_size > ctx->buffer_size) { + flb_plg_trace(ctx->ins, + "fd=%i incoming data exceed limit (%zu KB)", + connection->fd, (ctx->buffer_size / 1024)); + return -1; + } + + size = conn->buf_size + ctx->chunk_size; + tmp = flb_realloc(conn->buf_data, size); + if (!tmp) { + flb_errno(); + return -1; + } + flb_plg_trace(ctx->ins, "fd=%i buffer realloc %i -> %i", + connection->fd, conn->buf_size, size); + + conn->buf_data = tmp; + conn->buf_size = size; + available = (conn->buf_size - conn->buf_len) - 1; + } + + /* Read data */ + bytes = flb_io_net_read(connection, + (void *) &conn->buf_data[conn->buf_len], + available); + + if (bytes <= 0) { + return -1; + } + + flb_plg_trace(ctx->ins, "read()=%i pre_len=%i now_len=%i", + bytes, conn->buf_len, conn->buf_len + bytes); + conn->buf_len += bytes; + conn->buf_data[conn->buf_len] = '\0'; + + /* Strip CR or LF if found at first byte */ + if (conn->buf_data[0] == '\r' || conn->buf_data[0] == '\n') { + /* Skip message with one byte with CR or LF */ + flb_plg_trace(ctx->ins, "skip one byte message with ASCII code=%i", + conn->buf_data[0]); + consume_bytes(conn->buf_data, 1, conn->buf_len); + conn->buf_len--; + conn->buf_data[conn->buf_len] = '\0'; + } + + /* JSON Format handler */ + if (ctx->format == FLB_UDP_FMT_JSON) { + ret_payload = parse_payload_json(conn); + if (ret_payload == 0) { + /* Incomplete JSON message, we need more data */ + return -1; + } + else if (ret_payload == -1) { + flb_pack_state_reset(&conn->pack_state); + flb_pack_state_init(&conn->pack_state); + conn->pack_state.multiple = FLB_TRUE; + return -1; + } + } + else if (ctx->format == FLB_UDP_FMT_NONE) { + ret_payload = parse_payload_none(conn); + if (ret_payload == 0) { + return -1; + } + else if (ret_payload == -1) { + conn->buf_len = 0; + return -1; + } + } + + consume_bytes(conn->buf_data, ret_payload, conn->buf_len); + conn->buf_len -= ret_payload; + conn->buf_data[conn->buf_len] = '\0'; + + if (ctx->format == FLB_UDP_FMT_JSON) { + jsmn_init(&conn->pack_state.parser); + conn->pack_state.tokens_count = 0; + conn->pack_state.last_byte = 0; + conn->pack_state.buf_len = 0; + } + + return bytes; +} + +struct udp_conn *udp_conn_add(struct flb_connection *connection, + struct flb_in_udp_config *ctx) +{ + struct udp_conn *conn; + + conn = flb_malloc(sizeof(struct udp_conn)); + if (!conn) { + flb_errno(); + 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 = udp_conn_event; + + /* Connection info */ + conn->ctx = ctx; + conn->buf_len = 0; + + conn->buf_data = flb_malloc(ctx->chunk_size); + if (!conn->buf_data) { + flb_errno(); + + flb_plg_error(ctx->ins, "could not allocate new connection"); + flb_free(conn); + + return NULL; + } + conn->buf_size = ctx->chunk_size; + conn->ins = ctx->ins; + + /* Initialize JSON parser */ + if (ctx->format == FLB_UDP_FMT_JSON) { + flb_pack_state_init(&conn->pack_state); + conn->pack_state.multiple = FLB_TRUE; + } + + return conn; +} + +int udp_conn_del(struct udp_conn *conn) +{ + struct flb_in_udp_config *ctx; + + ctx = conn->ctx; + + if (ctx->format == FLB_UDP_FMT_JSON) { + flb_pack_state_reset(&conn->pack_state); + } + + flb_free(conn->buf_data); + flb_free(conn); + + return 0; +} diff --git a/src/fluent-bit/plugins/in_udp/udp_conn.h b/src/fluent-bit/plugins/in_udp/udp_conn.h new file mode 100644 index 000000000..25b8ef3ef --- /dev/null +++ b/src/fluent-bit/plugins/in_udp/udp_conn.h @@ -0,0 +1,57 @@ +/* -*- 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_UDP_CONN_H +#define FLB_IN_UDP_CONN_H + +#include <fluent-bit/flb_pack.h> +#include <fluent-bit/flb_connection.h> + +#define FLB_IN_UDP_CHUNK "32768" + +#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 + +struct udp_conn_stream { + char *tag; + size_t tag_len; +}; + +/* Respresents a connection */ +struct udp_conn { + /* Buffer */ + char *buf_data; /* Buffer data */ + int buf_len; /* Data length */ + int buf_size; /* Buffer size */ + + struct flb_input_instance *ins; /* Parent plugin instance */ + struct flb_in_udp_config *ctx; /* Plugin configuration context */ + struct flb_pack_state pack_state; /* Internal JSON parser */ + struct flb_connection *connection; + + struct mk_list _head; +}; + +struct udp_conn *udp_conn_add(struct flb_connection *connection, struct flb_in_udp_config *ctx); +int udp_conn_del(struct udp_conn *conn); +int udp_conn_event(void *data); + +#endif |