summaryrefslogtreecommitdiffstats
path: root/fluent-bit/plugins/in_tcp
diff options
context:
space:
mode:
Diffstat (limited to 'fluent-bit/plugins/in_tcp')
-rw-r--r--fluent-bit/plugins/in_tcp/CMakeLists.txt6
-rw-r--r--fluent-bit/plugins/in_tcp/tcp.c184
-rw-r--r--fluent-bit/plugins/in_tcp/tcp.h50
-rw-r--r--fluent-bit/plugins/in_tcp/tcp_config.c155
-rw-r--r--fluent-bit/plugins/in_tcp/tcp_config.h28
-rw-r--r--fluent-bit/plugins/in_tcp/tcp_conn.c412
-rw-r--r--fluent-bit/plugins/in_tcp/tcp_conn.h59
7 files changed, 894 insertions, 0 deletions
diff --git a/fluent-bit/plugins/in_tcp/CMakeLists.txt b/fluent-bit/plugins/in_tcp/CMakeLists.txt
new file mode 100644
index 000000000..df6763cd6
--- /dev/null
+++ b/fluent-bit/plugins/in_tcp/CMakeLists.txt
@@ -0,0 +1,6 @@
+set(src
+ tcp.c
+ tcp_conn.c
+ tcp_config.c)
+
+FLB_PLUGIN(in_tcp "${src}" "")
diff --git a/fluent-bit/plugins/in_tcp/tcp.c b/fluent-bit/plugins/in_tcp/tcp.c
new file mode 100644
index 000000000..084ea6887
--- /dev/null
+++ b/fluent-bit/plugins/in_tcp/tcp.c
@@ -0,0 +1,184 @@
+/* -*- 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 "tcp.h"
+#include "tcp_conn.h"
+#include "tcp_config.h"
+
+/*
+ * For a server event, the collection event means a new client have arrived, we
+ * accept the connection and create a new TCP instance which will wait for
+ * JSON map messages.
+ */
+static int in_tcp_collect(struct flb_input_instance *in,
+ struct flb_config *config, void *in_context)
+{
+ struct flb_connection *connection;
+ struct tcp_conn *conn;
+ struct flb_in_tcp_config *ctx;
+
+ 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;
+ }
+
+ flb_plg_trace(ctx->ins, "new TCP connection arrived FD=%i", connection->fd);
+
+ conn = tcp_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;
+}
+
+/* Initialize plugin */
+static int in_tcp_init(struct flb_input_instance *in,
+ struct flb_config *config, void *data)
+{
+ unsigned short int port;
+ int ret;
+ struct flb_in_tcp_config *ctx;
+
+ (void) data;
+
+ /* Allocate space for the configuration */
+ ctx = tcp_config_init(in);
+ if (!ctx) {
+ return -1;
+ }
+ ctx->collector_id = -1;
+ ctx->ins = in;
+ mk_list_init(&ctx->connections);
+
+ /* Set the context */
+ flb_input_set_context(in, ctx);
+
+ port = (unsigned short int) strtoul(ctx->tcp_port, NULL, 10);
+
+ ctx->downstream = flb_downstream_create(FLB_TRANSPORT_TCP,
+ 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->tcp_port);
+
+ tcp_config_destroy(ctx);
+
+ return -1;
+ }
+
+ flb_input_downstream_set(ctx->downstream, ctx->ins);
+
+ /* Collect upon data available on the standard input */
+ ret = flb_input_set_collector_socket(in,
+ in_tcp_collect,
+ ctx->downstream->server_fd,
+ config);
+ if (ret == -1) {
+ flb_plg_error(ctx->ins, "Could not set collector for IN_TCP input plugin");
+ tcp_config_destroy(ctx);
+
+ return -1;
+ }
+
+ ctx->collector_id = ret;
+
+ return 0;
+}
+
+static int in_tcp_exit(void *data, struct flb_config *config)
+{
+ struct mk_list *tmp;
+ struct mk_list *head;
+ struct flb_in_tcp_config *ctx;
+ struct tcp_conn *conn;
+
+ (void) *config;
+
+ ctx = data;
+
+ mk_list_foreach_safe(head, tmp, &ctx->connections) {
+ conn = mk_list_entry(head, struct tcp_conn, _head);
+
+ tcp_conn_del(conn);
+ }
+
+ tcp_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_tcp_config, format_name),
+ "Set the format: json or none"
+ },
+ {
+ FLB_CONFIG_MAP_STR, "separator", (char *)NULL,
+ 0, FLB_TRUE, offsetof(struct flb_in_tcp_config, raw_separator),
+ "Set separator"
+ },
+ {
+ FLB_CONFIG_MAP_STR, "chunk_size", (char *)NULL,
+ 0, FLB_TRUE, offsetof(struct flb_in_tcp_config, chunk_size_str),
+ "Set the chunk size"
+ },
+ {
+ FLB_CONFIG_MAP_STR, "buffer_size", (char *)NULL,
+ 0, FLB_TRUE, offsetof(struct flb_in_tcp_config, buffer_size_str),
+ "Set the buffer size"
+ },
+ /* EOF */
+ {0}
+};
+
+/* Plugin reference */
+struct flb_input_plugin in_tcp_plugin = {
+ .name = "tcp",
+ .description = "TCP",
+ .cb_init = in_tcp_init,
+ .cb_pre_run = NULL,
+ .cb_collect = in_tcp_collect,
+ .cb_flush_buf = NULL,
+ .cb_exit = in_tcp_exit,
+ .config_map = config_map,
+ .flags = FLB_INPUT_NET_SERVER | FLB_IO_OPT_TLS
+};
diff --git a/fluent-bit/plugins/in_tcp/tcp.h b/fluent-bit/plugins/in_tcp/tcp.h
new file mode 100644
index 000000000..3ddcbed06
--- /dev/null
+++ b/fluent-bit/plugins/in_tcp/tcp.h
@@ -0,0 +1,50 @@
+/* -*- 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_TCP_H
+#define FLB_IN_TCP_H
+
+#define FLB_TCP_FMT_JSON 0 /* default */
+#define FLB_TCP_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 flb_in_tcp_config {
+ 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 *tcp_port; /* TCP Port */
+ flb_sds_t raw_separator; /* Unescaped string delimiterr */
+ flb_sds_t separator; /* String delimiter */
+ int collector_id; /* Listener collector id */
+ struct flb_downstream *downstream; /* Client manager */
+ struct mk_list connections; /* List of active connections */
+ struct flb_input_instance *ins; /* Input plugin instace */
+ struct flb_log_event_encoder *log_encoder;
+};
+
+#endif
diff --git a/fluent-bit/plugins/in_tcp/tcp_config.c b/fluent-bit/plugins/in_tcp/tcp_config.c
new file mode 100644
index 000000000..db9a36a01
--- /dev/null
+++ b/fluent-bit/plugins/in_tcp/tcp_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 "tcp.h"
+#include "tcp_conn.h"
+#include "tcp_config.h"
+
+#include <stdlib.h>
+
+struct flb_in_tcp_config *tcp_config_init(struct flb_input_instance *ins)
+{
+ int ret;
+ int len;
+ char port[16];
+ char *out;
+ struct flb_in_tcp_config *ctx;
+
+ /* Allocate plugin context */
+ ctx = flb_calloc(1, sizeof(struct flb_in_tcp_config));
+ if (!ctx) {
+ flb_errno();
+ return NULL;
+ }
+ ctx->ins = ins;
+ ctx->format = FLB_TCP_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_TCP_FMT_JSON;
+ }
+ else if (strcasecmp(ctx->format_name, "none") == 0) {
+ ctx->format = FLB_TCP_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->tcp_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_TCP_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");
+ tcp_config_destroy(ctx);
+
+ ctx = NULL;
+ }
+
+ return ctx;
+}
+
+int tcp_config_destroy(struct flb_in_tcp_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->tcp_port);
+ flb_free(ctx);
+
+ return 0;
+}
diff --git a/fluent-bit/plugins/in_tcp/tcp_config.h b/fluent-bit/plugins/in_tcp/tcp_config.h
new file mode 100644
index 000000000..36df27873
--- /dev/null
+++ b/fluent-bit/plugins/in_tcp/tcp_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_TCP_CONFIG_H
+#define FLB_IN_TCP_CONFIG_H
+
+#include "tcp.h"
+
+struct flb_in_tcp_config *tcp_config_init(struct flb_input_instance *i_ins);
+int tcp_config_destroy(struct flb_in_tcp_config *config);
+
+#endif
diff --git a/fluent-bit/plugins/in_tcp/tcp_conn.c b/fluent-bit/plugins/in_tcp/tcp_conn.c
new file mode 100644
index 000000000..28b4b3222
--- /dev/null
+++ b/fluent-bit/plugins/in_tcp/tcp_conn.c
@@ -0,0 +1,412 @@
+/* -*- 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 "tcp.h"
+#include "tcp_conn.h"
+
+static inline void consume_bytes(char *buf, int bytes, int length)
+{
+ memmove(buf, buf + bytes, length - bytes);
+}
+
+static inline int process_pack(struct tcp_conn *conn,
+ char *pack, size_t size)
+{
+ int ret;
+ size_t off = 0;
+ msgpack_unpacked result;
+ msgpack_object entry;
+ struct flb_in_tcp_config *ctx;
+
+ 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;
+
+ 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) {
+ if (entry.type == MSGPACK_OBJECT_MAP) {
+ ret = flb_log_event_encoder_set_body_from_msgpack_object(
+ ctx->log_encoder, &entry);
+ }
+ else if (entry.type == MSGPACK_OBJECT_ARRAY) {
+ 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 (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 tcp_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 tcp_conn *conn)
+{
+ int ret;
+ int len;
+ int sep_len;
+ size_t consumed = 0;
+ char *buf;
+ char *s;
+ char *separator;
+ struct flb_in_tcp_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 tcp_conn_event(void *data)
+{
+ int bytes;
+ int available;
+ int size;
+ ssize_t ret_payload = -1;
+ char *tmp;
+ struct mk_event *event;
+ struct tcp_conn *conn;
+ struct flb_connection *connection;
+ struct flb_in_tcp_config *ctx;
+
+ 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->chunk_size > ctx->buffer_size) {
+ flb_plg_warn(ctx->ins,
+ "fd=%i incoming data exceeds 'Buffer_Size' (%zu KB)",
+ event->fd, (ctx->buffer_size / 1024));
+ tcp_conn_del(conn);
+ 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",
+ event->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) {
+ flb_plg_trace(ctx->ins, "fd=%i closed connection", event->fd);
+ tcp_conn_del(conn);
+ 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_TCP_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_TCP_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_TCP_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;
+ }
+
+ if (event->mask & MK_EVENT_CLOSE) {
+ flb_plg_trace(ctx->ins, "fd=%i hangup", event->fd);
+ tcp_conn_del(conn);
+ return -1;
+ }
+
+ return 0;
+}
+
+/* Create a new mqtt request instance */
+struct tcp_conn *tcp_conn_add(struct flb_connection *connection,
+ struct flb_in_tcp_config *ctx)
+{
+ struct tcp_conn *conn;
+ int ret;
+
+ conn = flb_malloc(sizeof(struct tcp_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 = tcp_conn_event;
+
+ /* Connection info */
+ conn->ctx = ctx;
+ conn->buf_len = 0;
+ conn->rest = 0;
+ conn->status = TCP_NEW;
+
+ 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_TCP_FMT_JSON) {
+ flb_pack_state_init(&conn->pack_state);
+ conn->pack_state.multiple = FLB_TRUE;
+ }
+
+ /* Register instance into the event loop */
+ 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 tcp_conn_del(struct tcp_conn *conn)
+{
+ struct flb_in_tcp_config *ctx;
+
+ ctx = conn->ctx;
+
+ if (ctx->format == FLB_TCP_FMT_JSON) {
+ flb_pack_state_reset(&conn->pack_state);
+ }
+
+ /* The downstream unregisters the file descriptor from the event-loop
+ * so there's nothing to be done by the plugin
+ */
+ flb_downstream_conn_release(conn->connection);
+
+ /* Release resources */
+ mk_list_del(&conn->_head);
+
+ flb_free(conn->buf_data);
+ flb_free(conn);
+
+ return 0;
+}
diff --git a/fluent-bit/plugins/in_tcp/tcp_conn.h b/fluent-bit/plugins/in_tcp/tcp_conn.h
new file mode 100644
index 000000000..f9af869f2
--- /dev/null
+++ b/fluent-bit/plugins/in_tcp/tcp_conn.h
@@ -0,0 +1,59 @@
+/* -*- 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_TCP_CONN_H
+#define FLB_IN_TCP_CONN_H
+
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_connection.h>
+
+#define FLB_IN_TCP_CHUNK "32768"
+
+enum {
+ TCP_NEW = 1, /* it's a new connection */
+ TCP_CONNECTED = 2, /* MQTT connection per protocol spec OK */
+};
+
+struct tcp_conn_stream {
+ char *tag;
+ size_t tag_len;
+};
+
+/* Respresents a connection */
+struct tcp_conn {
+ int status; /* Connection status */
+
+ /* Buffer */
+ char *buf_data; /* Buffer data */
+ int buf_len; /* Data length */
+ int buf_size; /* Buffer size */
+ size_t rest; /* Unpacking offset */
+
+ struct flb_input_instance *ins; /* Parent plugin instance */
+ struct flb_in_tcp_config *ctx; /* Plugin configuration context */
+ struct flb_pack_state pack_state; /* Internal JSON parser */
+ struct flb_connection *connection;
+
+ struct mk_list _head;
+};
+
+struct tcp_conn *tcp_conn_add(struct flb_connection *connection, struct flb_in_tcp_config *ctx);
+int tcp_conn_del(struct tcp_conn *conn);
+
+#endif
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-02 17:10:51 +0000