diff --git a/plugins/out_kafka/kafka.c b/plugins/out_kafka/kafka.c index 8dc8cb86ea8..7f72427a2e4 100644 --- a/plugins/out_kafka/kafka.c +++ b/plugins/out_kafka/kafka.c @@ -100,6 +100,8 @@ int produce_message(struct flb_time *tm, msgpack_object *map, char *dynamic_topic; char *message_key = NULL; size_t message_key_len = 0; + char *raw_key = NULL; + size_t raw_key_len = 0; struct flb_kafka_topic *topic = NULL; msgpack_sbuffer mp_sbuf; msgpack_packer mp_pck; @@ -211,6 +213,15 @@ int produce_message(struct flb_time *tm, msgpack_object *map, } } + /* Lookup raw_log_key */ + if (ctx->raw_log_key && !raw_key && val.type == MSGPACK_OBJECT_STR) { + if (key.via.str.size == ctx->raw_log_key_len && + strncmp(key.via.str.ptr, ctx->raw_log_key, ctx->raw_log_key_len) == 0) { + raw_key = (char *) val.via.str.ptr; + raw_key_len = val.via.str.size; + } + } + /* Lookup key/topic */ if (ctx->topic_key && !topic && val.type == MSGPACK_OBJECT_STR) { if (key.via.str.size == ctx->topic_key_len && @@ -346,6 +357,15 @@ int produce_message(struct flb_time *tm, msgpack_object *map, } #endif + else if (ctx->format == FLB_KAFKA_FMT_RAW) { + if (raw_key == NULL) { + flb_plg_error(ctx->ins, "missing raw_log_key"); + msgpack_sbuffer_destroy(&mp_sbuf); + return FLB_ERROR; + } + out_buf = raw_key; + out_size = raw_key_len; + } if (!message_key) { message_key = ctx->message_key; @@ -643,6 +663,13 @@ static struct flb_config_map config_map[] = { 0, FLB_FALSE, 0, "Set the kafka group_id." }, + { + FLB_CONFIG_MAP_STR, "raw_log_key", NULL, + 0, FLB_TRUE, offsetof(struct flb_out_kafka, raw_log_key), + "By default, the whole log record will be sent to Kafka. " + "If you specify a key name with this option, then only the value of " + "that key will be sent to Kafka." + }, /* EOF */ {0} }; diff --git a/plugins/out_kafka/kafka_config.c b/plugins/out_kafka/kafka_config.c index 04e91255635..fb2f5ca146c 100644 --- a/plugins/out_kafka/kafka_config.c +++ b/plugins/out_kafka/kafka_config.c @@ -93,6 +93,9 @@ struct flb_out_kafka *flb_out_kafka_create(struct flb_output_instance *ins, ctx->format = FLB_KAFKA_FMT_AVRO; } #endif + else if (strcasecmp(ctx->format_str, "raw") == 0) { + ctx->format = FLB_KAFKA_FMT_RAW; + } } else { ctx->format = FLB_KAFKA_FMT_JSON; @@ -114,6 +117,14 @@ struct flb_out_kafka *flb_out_kafka_create(struct flb_output_instance *ins, ctx->message_key_field_len = 0; } + /* Config: Log_Key */ + if (ctx->raw_log_key) { + ctx->raw_log_key_len = strlen(ctx->raw_log_key); + } + else { + ctx->raw_log_key_len = 0; + } + /* Config: Timestamp_Key */ if (ctx->timestamp_key) { ctx->timestamp_key_len = strlen(ctx->timestamp_key); diff --git a/plugins/out_kafka/kafka_config.h b/plugins/out_kafka/kafka_config.h index 42af378e161..14e036f8184 100644 --- a/plugins/out_kafka/kafka_config.h +++ b/plugins/out_kafka/kafka_config.h @@ -34,6 +34,7 @@ #ifdef FLB_HAVE_AVRO_ENCODER #define FLB_KAFKA_FMT_AVRO 3 #endif +#define FLB_KAFKA_FMT_RAW 4 #define FLB_KAFKA_TS_KEY "@timestamp" #define FLB_KAFKA_QUEUE_FULL_RETRIES "10" @@ -80,6 +81,9 @@ struct flb_out_kafka { int message_key_field_len; char *message_key_field; + int raw_log_key_len; + char *raw_log_key; + /* Gelf Keys */ struct flb_gelf_fields gelf_fields;