diff --git a/plugins/in_forward/fw_conn.c b/plugins/in_forward/fw_conn.c index edece47d87f..d0e74d5aa43 100644 --- a/plugins/in_forward/fw_conn.c +++ b/plugins/in_forward/fw_conn.c @@ -189,6 +189,9 @@ struct fw_conn *fw_conn_add(struct flb_connection *connection, struct flb_in_fw_ conn->buf_size = ctx->buffer_chunk_size; conn->in = ctx->ins; + conn->compression_type = FLB_COMPRESSION_ALGORITHM_NONE; + conn->d_ctx = NULL; + /* Register instance into the event loop */ ret = mk_event_add(flb_engine_evl_get(), connection->fd, @@ -219,6 +222,11 @@ int fw_conn_del(struct fw_conn *conn) /* Release resources */ mk_list_del(&conn->_head); + /* Release decompression context if it exists */ + if (conn->d_ctx) { + flb_decompression_context_destroy(conn->d_ctx); + } + if (conn->helo != NULL) { if (conn->helo->nonce != NULL) { flb_sds_destroy(conn->helo->nonce); diff --git a/plugins/in_forward/fw_conn.h b/plugins/in_forward/fw_conn.h index 6e24c022767..380a7899192 100644 --- a/plugins/in_forward/fw_conn.h +++ b/plugins/in_forward/fw_conn.h @@ -20,6 +20,8 @@ #ifndef FLB_IN_FW_CONN_H #define FLB_IN_FW_CONN_H +#include + #define FLB_IN_FW_CHUNK_SIZE "1024000" /* 1MB */ #define FLB_IN_FW_CHUNK_MAX_SIZE "6144000" /* =FLB_IN_FW_CHUNK_SIZE * 6. 6MB */ #define FLB_IN_FW_NONCE_SIZE 16 @@ -48,6 +50,10 @@ struct fw_conn { int buf_size; /* Buffer size */ size_t rest; /* Unpacking offset */ + /* Decompression context */ + int compression_type; /* e.g., FLB_COMPRESSION_ALGORITHM_GZIP */ + struct flb_decompression_context *d_ctx; /* Stateful decompressor context */ + struct flb_in_fw_helo *helo; /* secure forward HELO phase */ struct flb_input_instance *in; /* Parent plugin instance */ diff --git a/plugins/in_forward/fw_prot.c b/plugins/in_forward/fw_prot.c index 4ae7636aa0d..099a908ad46 100644 --- a/plugins/in_forward/fw_prot.c +++ b/plugins/in_forward/fw_prot.c @@ -90,50 +90,33 @@ static int get_chunk_event_type(struct flb_input_instance *ins, msgpack_object o return type; } -static int is_gzip_compressed(msgpack_object options) +static int get_compression_type(msgpack_object options) { int i; - msgpack_object k; - msgpack_object v; + msgpack_object k, v; if (options.type != MSGPACK_OBJECT_MAP) { return -1; } - for (i = 0; i < options.via.map.size; i++) { k = options.via.map.ptr[i].key; v = options.via.map.ptr[i].val; - if (k.type != MSGPACK_OBJECT_STR) { - return -1; - } - - if (k.via.str.size != 10) { - continue; - } - - if (strncmp(k.via.str.ptr, "compressed", 10) == 0) { - if (v.type != MSGPACK_OBJECT_STR) { - return -1; - } - - if (v.via.str.size != 4) { - return -1; - } - - if (strncmp(v.via.str.ptr, "gzip", 4) == 0) { - return FLB_TRUE; - } - else if (strncmp(v.via.str.ptr, "text", 4) == 0) { - return FLB_FALSE; + if (k.type == MSGPACK_OBJECT_STR && k.via.str.size == 10 && + strncmp(k.via.str.ptr, "compressed", 10) == 0) { + if (v.type == MSGPACK_OBJECT_STR) { + if (v.via.str.size == 4 && strncmp(v.via.str.ptr, "gzip", 4) == 0) { + return FLB_COMPRESSION_ALGORITHM_GZIP; + } + if (v.via.str.size == 4 && strncmp(v.via.str.ptr, "zstd", 4) == 0) { + return FLB_COMPRESSION_ALGORITHM_ZSTD; + } } - - return -1; } } - return FLB_FALSE; + return FLB_COMPRESSION_ALGORITHM_NONE; } static inline void print_msgpack_error_code(struct flb_input_instance *in, @@ -1246,6 +1229,18 @@ int fw_prot_secure_forward_handshake(struct flb_input_instance *ins, return -1; } +static int sniff_magic(const uint8_t *p, size_t n) { + if (n >= 2 && p[0]==0x1f && p[1]==0x8b) { + return FLB_COMPRESSION_ALGORITHM_GZIP; + } + if (n >= 4 && p[0]==0x28 && p[1]==0xb5 && + p[2]==0x2f && p[3]==0xfd) { + return FLB_COMPRESSION_ALGORITHM_ZSTD; + } + + return FLB_COMPRESSION_ALGORITHM_NONE; +} + int fw_prot_process(struct flb_input_instance *ins, struct fw_conn *conn) { int ret; @@ -1259,8 +1254,6 @@ int fw_prot_process(struct flb_input_instance *ins, struct fw_conn *conn) flb_sds_t out_tag = NULL; size_t bytes; size_t recv_len; - size_t gz_size; - void *gz_data; msgpack_object tag; msgpack_object entry; msgpack_object map; @@ -1524,93 +1517,124 @@ int fw_prot_process(struct flb_input_instance *ins, struct fw_conn *conn) } if (data) { - ret = is_gzip_compressed(root.via.array.ptr[2]); - if (ret == -1) { - flb_plg_error(ctx->ins, "invalid 'compressed' option"); - msgpack_unpacked_destroy(&result); - msgpack_unpacker_free(unp); - flb_sds_destroy(out_tag); - return -1; + /* Get event type early for use in both compressed/uncompressed paths */ + event_type = FLB_EVENT_TYPE_LOGS; + if (contain_options) { + ret = get_chunk_event_type(ins, root.via.array.ptr[2]); + if (ret == -1) { + flb_plg_error(ctx->ins, "invalid chunk event type"); + msgpack_unpacked_destroy(&result); + flb_sds_destroy(out_tag); + msgpack_unpacker_free(unp); + return -1; + } + event_type = ret; } - if (ret == FLB_TRUE) { - size_t remaining = len; - - while (remaining > 0) { - ret = flb_gzip_uncompress_multi((void *) (data + (len - remaining)), remaining, - &gz_data, &gz_size, &remaining); + /* Initialize decompressor on first compressed chunk */ + if (conn->d_ctx == NULL && contain_options) { + int opt_type = contain_options ? get_compression_type(root.via.array.ptr[2]) : FLB_COMPRESSION_ALGORITHM_NONE; + int sniff = sniff_magic((const uint8_t *)data, len); + int type = opt_type; + + if (sniff != FLB_COMPRESSION_ALGORITHM_NONE && + opt_type != FLB_COMPRESSION_ALGORITHM_NONE && + sniff != opt_type) { + flb_plg_warn(ctx->ins, "compressed=%s but magic says %s; using magic", + opt_type == FLB_COMPRESSION_ALGORITHM_ZSTD ? "zstd" : "gzip", + sniff == FLB_COMPRESSION_ALGORITHM_ZSTD ? "zstd" : "gzip"); + type = sniff; + } + else if (opt_type == FLB_COMPRESSION_ALGORITHM_NONE) { + type = sniff; + } + if (type > 0) { + conn->compression_type = type; + conn->d_ctx = flb_decompression_context_create( + conn->compression_type, + FLB_DECOMPRESSION_BUFFER_SIZE); + if (!conn->d_ctx) { + flb_plg_error(ctx->ins, "failed to create decompression context"); + + goto cleanup_msgpack; + } + } + } - if (ret == -1) { - flb_plg_error(ctx->ins, "gzip uncompress failure"); - msgpack_unpacked_destroy(&result); - msgpack_unpacker_free(unp); - flb_sds_destroy(out_tag); - return -1; + if (conn->compression_type != FLB_COMPRESSION_ALGORITHM_NONE) { + char *decomp_buf = NULL; + uint8_t *append_ptr; + size_t available_space; + size_t decomp_len; + int decomp_ret; + size_t required_size; + + available_space = flb_decompression_context_get_available_space(conn->d_ctx); + if (len > available_space) { + required_size = conn->d_ctx->input_buffer_length + len; + if (flb_decompression_context_resize_buffer(conn->d_ctx, required_size) != 0) { + flb_plg_error(ctx->ins, "cannot resize decompression buffer"); + + goto cleanup_decompress; } + } + append_ptr = flb_decompression_context_get_append_buffer(conn->d_ctx); + memcpy(append_ptr, data, len); + conn->d_ctx->input_buffer_length += len; + + decomp_buf = flb_malloc(ctx->buffer_chunk_size); + if (!decomp_buf) { + flb_errno(); + + goto cleanup_decompress; + } + + do { + decomp_len = ctx->buffer_chunk_size; + decomp_ret = flb_decompress(conn->d_ctx, decomp_buf, &decomp_len); + + if (decomp_ret == FLB_DECOMPRESSOR_FAILURE) { + if (decomp_len > 0) { + flb_plg_error(ctx->ins, "decompression failed, data may be corrupt"); + flb_free(decomp_buf); - event_type = FLB_EVENT_TYPE_LOGS; - if (contain_options) { - ret = get_chunk_event_type(ins, root.via.array.ptr[2]); - if (ret == -1) { - msgpack_unpacked_destroy(&result); - msgpack_unpacker_free(unp); - flb_sds_destroy(out_tag); - flb_free(gz_data); - return -1; + goto cleanup_decompress; } - event_type = ret; + break; } - ret = append_log(ins, conn, - event_type, - out_tag, gz_data, gz_size); - if (ret == -1) { - msgpack_unpacked_destroy(&result); - msgpack_unpacker_free(unp); - flb_sds_destroy(out_tag); - flb_free(gz_data); - return -1; + if (decomp_len > 0) { + if (append_log(ins, conn, event_type, out_tag, decomp_buf, decomp_len) == -1) { + flb_free(decomp_buf); + + goto cleanup_decompress; + } } - flb_free(gz_data); - } + } while (decomp_len > 0); + + flb_free(decomp_buf); + + flb_decompression_context_destroy(conn->d_ctx); + conn->d_ctx = NULL; + conn->compression_type = FLB_COMPRESSION_ALGORITHM_NONE; } else { - event_type = FLB_EVENT_TYPE_LOGS; - if (contain_options) { - ret = get_chunk_event_type(ins, root.via.array.ptr[2]); - if (ret == -1) { - msgpack_unpacked_destroy(&result); - msgpack_unpacker_free(unp); - flb_sds_destroy(out_tag); - return -1; - } - event_type = ret; - } - - ret = append_log(ins, conn, - event_type, - out_tag, data, len); - if (ret == -1) { - msgpack_unpacked_destroy(&result); - msgpack_unpacker_free(unp); - flb_sds_destroy(out_tag); - return -1; + if (append_log(ins, conn, event_type, out_tag, data, len) == -1) { + goto cleanup_msgpack; } } + } - /* Handle ACK response */ - if (chunk_id != -1) { - chunk = root.via.array.ptr[2].via.map.ptr[chunk_id].val; - send_ack(ctx->ins, conn, chunk); - } + /* Handle ACK response (common to all paths) */ + if (chunk_id != -1) { + chunk = root.via.array.ptr[2].via.map.ptr[chunk_id].val; + send_ack(ctx->ins, conn, chunk); } } else { flb_plg_warn(ctx->ins, "invalid data format, type=%i", entry.type); - msgpack_unpacked_destroy(&result); - msgpack_unpacker_free(unp); - return -1; + goto cleanup_msgpack; } ret = msgpack_unpacker_next(unp, &result); @@ -1637,4 +1661,16 @@ int fw_prot_process(struct flb_input_instance *ins, struct fw_conn *conn) }; return 0; + +cleanup_decompress: + flb_decompression_context_destroy(conn->d_ctx); + conn->d_ctx = NULL; + conn->compression_type = FLB_COMPRESSION_ALGORITHM_NONE; + /* FALLTHRU */ +cleanup_msgpack: + msgpack_unpacked_destroy(&result); + msgpack_unpacker_free(unp); + flb_sds_destroy(out_tag); + + return -1; } diff --git a/tests/runtime/in_forward.c b/tests/runtime/in_forward.c index 6cabfa94a95..7577bae01ac 100644 --- a/tests/runtime/in_forward.c +++ b/tests/runtime/in_forward.c @@ -23,6 +23,9 @@ #include #include #include +#include +#include +#include #include #include #ifdef FLB_HAVE_UNIX_SOCKET @@ -565,6 +568,228 @@ void flb_test_unix_perm() } #endif /* FLB_HAVE_UNIX_SOCKET */ +/* + * Creates a forward-protocol-compliant, Gzip-compressed MessagePack payload. + * The final structure is: [tag, compressed_events, {options}] + */ +static int create_simple_json_gzip(msgpack_sbuffer *sbuf) +{ + int ret; + char *event_buf; + size_t event_size; + char *compressed_buf; + size_t compressed_size; + int root_type; + msgpack_packer pck; + + char *tag = "test"; + char event_json[] = "[1234567890,{\"test\":\"msg\"}]"; + + ret = flb_pack_json(event_json, strlen(event_json), + &event_buf, &event_size, &root_type, NULL); + if (!TEST_CHECK(ret == 0)) { + return -1; + } + + ret = flb_gzip_compress(event_buf, event_size, + (void **)&compressed_buf, &compressed_size); + if (!TEST_CHECK(ret == 0)) { + flb_free(event_buf); + return -1; + } + flb_free(event_buf); + + /* Create temporary msgpack buffer */ + msgpack_packer_init(&pck, sbuf, msgpack_sbuffer_write); + + msgpack_pack_array(&pck, 3); + msgpack_pack_str_with_body(&pck, tag, strlen(tag)); + msgpack_pack_bin_with_body(&pck, compressed_buf, compressed_size); + msgpack_pack_map(&pck, 2); + msgpack_pack_str_with_body(&pck, "compressed", 10); + msgpack_pack_str_with_body(&pck, "gzip", 4); + msgpack_pack_str_with_body(&pck, "size", 4); + msgpack_pack_uint64(&pck, event_size); + + flb_free(compressed_buf); + + return 0; +} + +void flb_test_forward_gzip() +{ + struct flb_lib_out_cb cb_data; + struct test_ctx *ctx; + flb_sockfd_t fd; + int ret; + int num; + ssize_t w_size; + + char *buf; + size_t size; + + msgpack_sbuffer sbuf; + + clear_output_num(); + + cb_data.cb = cb_check_result_json; + cb_data.data = "\"test\":\"msg\""; + + ctx = test_ctx_create(&cb_data); + if (!TEST_CHECK(ctx != NULL)) { + TEST_MSG("test_ctx_create failed"); + exit(EXIT_FAILURE); + } + + ret = flb_output_set(ctx->flb, ctx->o_ffd, + "match", "test", + "format", "json", + NULL); + TEST_CHECK(ret == 0); + + ret = flb_start(ctx->flb); + TEST_CHECK(ret == 0); + + fd = connect_tcp(NULL, -1); + if (!TEST_CHECK(fd >= 0)) { + exit(EXIT_FAILURE); + } + + msgpack_sbuffer_init(&sbuf); + create_simple_json_gzip(&sbuf); + + w_size = send(fd, sbuf.data, sbuf.size, 0); + if (!TEST_CHECK(w_size == sbuf.size)) { + TEST_MSG("failed to send, errno=%d", errno); + flb_socket_close(fd); + msgpack_sbuffer_destroy(&sbuf); + exit(EXIT_FAILURE); + } + + msgpack_sbuffer_destroy(&sbuf); + + flb_time_msleep(1500); + + num = get_output_num(); + if (!TEST_CHECK(num > 0)) { + TEST_MSG("no outputs"); + } + + flb_socket_close(fd); + test_ctx_destroy(ctx); +} + +/* + * Creates a forward-protocol-compliant, Zstd-compressed MessagePack payload. + * The final structure is: [tag, compressed_events, {options}] + */ +static int create_simple_json_zstd(msgpack_sbuffer *sbuf) +{ + int ret; + char *event_buf; + size_t event_size; + char *compressed_buf; + size_t compressed_size; + int root_type; + msgpack_packer pck; + + char *tag = "test"; + char event_json[] = "[1234567890,{\"test\":\"msg\"}]"; + + ret = flb_pack_json(event_json, strlen(event_json), + &event_buf, &event_size, &root_type, NULL); + if (!TEST_CHECK(ret == 0)) { + return -1; + } + + ret = flb_zstd_compress(event_buf, event_size, + (void **)&compressed_buf, &compressed_size); + if (!TEST_CHECK(ret == 0)) { + flb_free(event_buf); + return -1; + } + flb_free(event_buf); + + /* Create temporary msgpack buffer */ + msgpack_packer_init(&pck, sbuf, msgpack_sbuffer_write); + + msgpack_pack_array(&pck, 3); + msgpack_pack_str_with_body(&pck, tag, strlen(tag)); + msgpack_pack_bin_with_body(&pck, compressed_buf, compressed_size); + msgpack_pack_map(&pck, 2); + msgpack_pack_str_with_body(&pck, "compressed", 10); + msgpack_pack_str_with_body(&pck, "zstd", 4); + msgpack_pack_str_with_body(&pck, "size", 4); + msgpack_pack_uint64(&pck, event_size); + + flb_free(compressed_buf); + + return 0; +} + +void flb_test_forward_zstd() +{ + struct flb_lib_out_cb cb_data; + struct test_ctx *ctx; + flb_sockfd_t fd; + int ret; + int num; + ssize_t w_size; + + char *buf; + size_t size; + + msgpack_sbuffer sbuf; + + clear_output_num(); + + cb_data.cb = cb_check_result_json; + cb_data.data = "\"test\":\"msg\""; + + ctx = test_ctx_create(&cb_data); + if (!TEST_CHECK(ctx != NULL)) { + TEST_MSG("test_ctx_create failed"); + exit(EXIT_FAILURE); + } + + ret = flb_output_set(ctx->flb, ctx->o_ffd, + "match", "test", + "format", "json", + NULL); + TEST_CHECK(ret == 0); + + ret = flb_start(ctx->flb); + TEST_CHECK(ret == 0); + + fd = connect_tcp(NULL, -1); + if (!TEST_CHECK(fd >= 0)) { + exit(EXIT_FAILURE); + } + + msgpack_sbuffer_init(&sbuf); + create_simple_json_zstd(&sbuf); + + w_size = send(fd, sbuf.data, sbuf.size, 0); + if (!TEST_CHECK(w_size == sbuf.size)) { + TEST_MSG("failed to send, errno=%d", errno); + flb_socket_close(fd); + msgpack_sbuffer_destroy(&sbuf); + exit(EXIT_FAILURE); + } + + msgpack_sbuffer_destroy(&sbuf); + + flb_time_msleep(1500); + + num = get_output_num(); + if (!TEST_CHECK(num > 0)) { + TEST_MSG("no outputs"); + } + + flb_socket_close(fd); + test_ctx_destroy(ctx); +} + TEST_LIST = { {"forward", flb_test_forward}, @@ -574,6 +799,7 @@ TEST_LIST = { {"unix_path", flb_test_unix_path}, {"unix_perm", flb_test_unix_perm}, #endif + {"forward_gzip", flb_test_forward_gzip}, + {"forward_zstd", flb_test_forward_zstd}, {NULL, NULL} }; -