diff --git a/plugins/in_elasticsearch/in_elasticsearch.h b/plugins/in_elasticsearch/in_elasticsearch.h index 10deb62aa4b..7911cb72570 100644 --- a/plugins/in_elasticsearch/in_elasticsearch.h +++ b/plugins/in_elasticsearch/in_elasticsearch.h @@ -25,6 +25,7 @@ #include #include #include +#include #include #include @@ -35,14 +36,15 @@ struct flb_in_elasticsearch { flb_sds_t listen; flb_sds_t tcp_port; - const char *tag_key; - const char *meta_key; + flb_sds_t tag_key; + flb_sds_t meta_key; flb_sds_t hostname; flb_sds_t es_version; char cluster_name[16]; char node_name[12]; struct flb_log_event_encoder *log_encoder; + struct flb_record_accessor *ra_tag_key; struct flb_input_instance *ins; diff --git a/plugins/in_elasticsearch/in_elasticsearch_bulk_prot.c b/plugins/in_elasticsearch/in_elasticsearch_bulk_prot.c index e5818c9ddca..ea8c53d1282 100644 --- a/plugins/in_elasticsearch/in_elasticsearch_bulk_prot.c +++ b/plugins/in_elasticsearch/in_elasticsearch_bulk_prot.c @@ -22,6 +22,8 @@ #include #include #include +#include +#include #include #include @@ -245,67 +247,32 @@ static int send_response(struct in_elasticsearch_bulk_conn *conn, int http_statu /* implements functionality to get tag from key in record */ static flb_sds_t tag_key(struct flb_in_elasticsearch *ctx, msgpack_object *map) { - size_t map_size = map->via.map.size; - msgpack_object_kv *kv; - msgpack_object key; - msgpack_object val; - char *key_str = NULL; - char *val_str = NULL; - size_t key_str_size = 0; - size_t val_str_size = 0; - int j; - int check = FLB_FALSE; - int found = FLB_FALSE; - flb_sds_t tag; - - kv = map->via.map.ptr; + flb_sds_t tag = NULL; + struct flb_ra_value *ra_val; - for(j=0; j < map_size; j++) { - check = FLB_FALSE; - found = FLB_FALSE; - key = (kv+j)->key; - if (key.type == MSGPACK_OBJECT_BIN) { - key_str = (char *) key.via.bin.ptr; - key_str_size = key.via.bin.size; - check = FLB_TRUE; - } - if (key.type == MSGPACK_OBJECT_STR) { - key_str = (char *) key.via.str.ptr; - key_str_size = key.via.str.size; - check = FLB_TRUE; - } - - if (check == FLB_TRUE) { - if (strncmp(ctx->tag_key, key_str, key_str_size) == 0) { - val = (kv+j)->val; - if (val.type == MSGPACK_OBJECT_BIN) { - val_str = (char *) val.via.bin.ptr; - val_str_size = val.via.str.size; - found = FLB_TRUE; - break; - } - if (val.type == MSGPACK_OBJECT_STR) { - val_str = (char *) val.via.str.ptr; - val_str_size = val.via.str.size; - found = FLB_TRUE; - break; - } - } - } + /* If no record accessor is configured, return NULL */ + if (!ctx->ra_tag_key) { + return NULL; } - if (found == FLB_TRUE) { - tag = flb_sds_create_len(val_str, val_str_size); - if (!tag) { - flb_errno(); - return NULL; - } - return tag; + /* Use record accessor to get the value */ + ra_val = flb_ra_get_value_object(ctx->ra_tag_key, *map); + if (!ra_val) { + flb_plg_warn(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key); + return NULL; } + /* Convert the value to string */ + if (ra_val->type == FLB_RA_STRING) { + tag = flb_sds_create_len(ra_val->o.via.str.ptr, ra_val->o.via.str.size); + } + else { + flb_plg_error(ctx->ins, "tag_key %s value is not a string or binary", ctx->tag_key); + } - flb_plg_error(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key); - return NULL; + /* Clean up the record accessor value */ + flb_ra_key_value_destroy(ra_val); + return tag; } static int get_write_op(struct flb_in_elasticsearch *ctx, msgpack_object *map, flb_sds_t *out_write_op, size_t *out_key_size) diff --git a/plugins/in_elasticsearch/in_elasticsearch_config.c b/plugins/in_elasticsearch/in_elasticsearch_config.c index 56b5a4e099f..18da268db60 100644 --- a/plugins/in_elasticsearch/in_elasticsearch_config.c +++ b/plugins/in_elasticsearch/in_elasticsearch_config.c @@ -67,12 +67,25 @@ struct flb_in_elasticsearch *in_elasticsearch_config_create(struct flb_input_ins return NULL; } + /* Create record accessor for tag_key if specified */ + if (ctx->tag_key) { + ctx->ra_tag_key = flb_ra_create(ctx->tag_key, FLB_TRUE); + if (!ctx->ra_tag_key) { + flb_plg_error(ctx->ins, "invalid record accessor pattern for tag_key: %s", ctx->tag_key); + in_elasticsearch_config_destroy(ctx); + return NULL; + } + } return ctx; } int in_elasticsearch_config_destroy(struct flb_in_elasticsearch *ctx) { + if (ctx->ra_tag_key) { + flb_ra_destroy(ctx->ra_tag_key); + } + flb_log_event_encoder_destroy(ctx->log_encoder); /* release all connections */ diff --git a/plugins/in_http/http.h b/plugins/in_http/http.h index 4298a370c9c..2e3796798c9 100644 --- a/plugins/in_http/http.h +++ b/plugins/in_http/http.h @@ -25,6 +25,7 @@ #include #include #include +#include #include #include @@ -36,7 +37,8 @@ struct flb_http { int successful_response_code; flb_sds_t listen; flb_sds_t tcp_port; - const char *tag_key; + flb_sds_t tag_key; + struct flb_record_accessor *ra_tag_key; /* Success HTTP headers */ struct mk_list *success_headers; diff --git a/plugins/in_http/http_config.c b/plugins/in_http/http_config.c index d1927d91fa2..2459f004907 100644 --- a/plugins/in_http/http_config.c +++ b/plugins/in_http/http_config.c @@ -75,9 +75,7 @@ struct flb_http *http_config_create(struct flb_input_instance *ins) if (ret != FLB_EVENT_ENCODER_SUCCESS) { flb_plg_error(ctx->ins, "error initializing event encoder : %d", ret); - http_config_destroy(ctx); - return NULL; } @@ -85,10 +83,19 @@ struct flb_http *http_config_create(struct flb_input_instance *ins) if (ctx->success_headers_str == NULL) { http_config_destroy(ctx); - return NULL; } + /* Create record accessor for tag_key if specified */ + if (ctx->tag_key) { + ctx->ra_tag_key = flb_ra_create(ctx->tag_key, FLB_TRUE); + if (!ctx->ra_tag_key) { + flb_plg_error(ctx->ins, "invalid record accessor pattern for tag_key: %s", ctx->tag_key); + http_config_destroy(ctx); + return NULL; + } + } + flb_config_map_foreach(header_iterator, header_pair, ctx->success_headers) { header_name = mk_list_entry_first(header_pair->val.list, struct flb_slist_entry, @@ -132,6 +139,10 @@ struct flb_http *http_config_create(struct flb_input_instance *ins) int http_config_destroy(struct flb_http *ctx) { + if (ctx->ra_tag_key) { + flb_ra_destroy(ctx->ra_tag_key); + } + /* release all connections */ http_conn_release_all(ctx); diff --git a/plugins/in_http/http_prot.c b/plugins/in_http/http_prot.c index 001209bdb8f..e92220a51d9 100644 --- a/plugins/in_http/http_prot.c +++ b/plugins/in_http/http_prot.c @@ -27,6 +27,8 @@ #include #include #include +#include +#include #include #include @@ -42,8 +44,8 @@ static inline char hex2nibble(char c) if ((c >= 0x30) && (c <= '9')) { return c - 0x30; } - // 0x30-0x39 are digits, 0x41-0x46 A-F, - // so there is a gap at 0x40 + + /* 0x30-0x39 are digits, 0x41-0x46 A-F, so there is a gap at 0x40 */ if ((c >= 'A') && (c <= 'F')) { return (c - 'A') + 10; } @@ -163,70 +165,54 @@ static int send_response(struct http_conn *conn, int http_status, char *message) return 0; } -/* implements functionality to get tag from key in record */ -static flb_sds_t tag_key(struct flb_http *ctx, msgpack_object *map) +static void sanitize_tag(flb_sds_t tag) { - size_t map_size = map->via.map.size; - msgpack_object_kv *kv; - msgpack_object key; - msgpack_object val; - char *key_str = NULL; - char *val_str = NULL; - size_t key_str_size = 0; - size_t val_str_size = 0; - int j; - int check = FLB_FALSE; - int found = FLB_FALSE; - flb_sds_t tag; + size_t i; - kv = map->via.map.ptr; + if (!tag) { + return; + } - for(j=0; j < map_size; j++) { - check = FLB_FALSE; - found = FLB_FALSE; - key = (kv+j)->key; - if (key.type == MSGPACK_OBJECT_BIN) { - key_str = (char *) key.via.bin.ptr; - key_str_size = key.via.bin.size; - check = FLB_TRUE; - } - if (key.type == MSGPACK_OBJECT_STR) { - key_str = (char *) key.via.str.ptr; - key_str_size = key.via.str.size; - check = FLB_TRUE; + for (i = 0; i < flb_sds_len(tag); i++) { + if (!isalnum(tag[i]) && tag[i] != '_' && tag[i] != '.') { + tag[i] = '_'; } + } +} - if (check == FLB_TRUE) { - if (strncmp(ctx->tag_key, key_str, key_str_size) == 0) { - val = (kv+j)->val; - if (val.type == MSGPACK_OBJECT_BIN) { - val_str = (char *) val.via.bin.ptr; - val_str_size = val.via.bin.size; - found = FLB_TRUE; - break; - } - if (val.type == MSGPACK_OBJECT_STR) { - val_str = (char *) val.via.str.ptr; - val_str_size = val.via.str.size; - found = FLB_TRUE; - break; - } - } - } +/* implements functionality to get tag from key in record */ +static flb_sds_t tag_key(struct flb_http *ctx, msgpack_object *map) +{ + struct flb_ra_value *ra_val; + flb_sds_t tag = NULL; + + /* If no record accessor is configured, return NULL */ + if (!ctx->ra_tag_key) { + return NULL; } - if (found == FLB_TRUE) { - tag = flb_sds_create_len(val_str, val_str_size); - if (!tag) { - flb_errno(); - return NULL; + /* Use record accessor to get the value */ + ra_val = flb_ra_get_value_object(ctx->ra_tag_key, *map); + if (!ra_val) { + flb_plg_debug(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key); + return NULL; + } + + /* Convert the value to string */ + if (ra_val->type == FLB_RA_STRING) { + tag = flb_sds_create_len(ra_val->o.via.str.ptr, ra_val->o.via.str.size); + if (tag) { + sanitize_tag(tag); } - return tag; + } + else { + flb_plg_debug(ctx->ins, "tag_key %s value is not a string", ctx->tag_key); } + /* Clean up the record accessor value */ + flb_ra_key_value_destroy(ra_val); - flb_plg_warn(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key); - return NULL; + return tag; } static int process_pack_record(struct flb_http *ctx, struct flb_time *tm, @@ -878,7 +864,6 @@ int http_prot_handle(struct flb_http *ctx, struct http_conn *conn, struct mk_http_session *session, struct mk_http_request *request) { - int i; int ret; int len; char *uri; @@ -927,12 +912,7 @@ int http_prot_handle(struct flb_http *ctx, struct http_conn *conn, /* New tag skipping the URI '/' */ flb_sds_cat_safe(&tag, uri + 1, len - 1); - /* Sanitize, only allow alphanum chars */ - for (i = 0; i < flb_sds_len(tag); i++) { - if (!isalnum(tag[i]) && tag[i] != '_' && tag[i] != '.') { - tag[i] = '_'; - } - } + sanitize_tag(tag); } mk_mem_free(uri); diff --git a/plugins/in_splunk/splunk.c b/plugins/in_splunk/splunk.c index ee78e6afd96..523048becb6 100644 --- a/plugins/in_splunk/splunk.c +++ b/plugins/in_splunk/splunk.c @@ -135,6 +135,9 @@ static int in_splunk_init(struct flb_input_instance *ins, ctx->http_server.request_callback = splunk_prot_handle_ng; flb_input_downstream_set(ctx->http_server.downstream, ctx->ins); + + flb_plg_info(ctx->ins, "listening on %s:%u", + ins->host.listen, ins->host.port); } else { ctx->downstream = flb_downstream_create(FLB_TRANSPORT_TCP, @@ -157,6 +160,8 @@ static int in_splunk_init(struct flb_input_instance *ins, flb_input_downstream_set(ctx->downstream, ctx->ins); + flb_plg_info(ctx->ins, "listening on %s:%s", ctx->listen, ctx->tcp_port); + /* Collect upon data available on the standard input */ ret = flb_input_set_collector_socket(ins, in_splunk_collect, @@ -172,6 +177,7 @@ static int in_splunk_init(struct flb_input_instance *ins, ctx->collector_id = ret; } + return 0; } diff --git a/plugins/in_splunk/splunk.h b/plugins/in_splunk/splunk.h index 5dc4645088c..ef04edf767a 100644 --- a/plugins/in_splunk/splunk.h +++ b/plugins/in_splunk/splunk.h @@ -25,6 +25,7 @@ #include #include #include +#include #include #include @@ -41,7 +42,8 @@ struct flb_splunk_tokens { struct flb_splunk { flb_sds_t listen; flb_sds_t tcp_port; - const char *tag_key; + flb_sds_t tag_key; + struct flb_record_accessor *ra_tag_key; /* Success HTTP headers */ struct mk_list *success_headers; diff --git a/plugins/in_splunk/splunk_config.c b/plugins/in_splunk/splunk_config.c index a7c588658ee..3907d5a54f3 100644 --- a/plugins/in_splunk/splunk_config.c +++ b/plugins/in_splunk/splunk_config.c @@ -229,11 +229,25 @@ struct flb_splunk *splunk_config_create(struct flb_input_instance *ins) } } + /* Create record accessor for tag_key if specified */ + if (ctx->tag_key) { + ctx->ra_tag_key = flb_ra_create(ctx->tag_key, FLB_TRUE); + if (!ctx->ra_tag_key) { + flb_plg_error(ctx->ins, "invalid record accessor pattern for tag_key: %s", ctx->tag_key); + splunk_config_destroy(ctx); + return NULL; + } + } + return ctx; } int splunk_config_destroy(struct flb_splunk *ctx) { + if (ctx->ra_tag_key) { + flb_ra_destroy(ctx->ra_tag_key); + } + /* release all connections */ splunk_conn_release_all(ctx); diff --git a/plugins/in_splunk/splunk_prot.c b/plugins/in_splunk/splunk_prot.c index af0ebc0c02b..4fd2cf8bb3d 100644 --- a/plugins/in_splunk/splunk_prot.c +++ b/plugins/in_splunk/splunk_prot.c @@ -22,6 +22,8 @@ #include #include #include +#include +#include #include #include @@ -156,67 +158,33 @@ static int send_json_message_response(struct splunk_conn *conn, int http_status, /* implements functionality to get tag from key in record */ static flb_sds_t tag_key(struct flb_splunk *ctx, msgpack_object *map) { - size_t map_size = map->via.map.size; - msgpack_object_kv *kv; - msgpack_object key; - msgpack_object val; - char *key_str = NULL; - char *val_str = NULL; - size_t key_str_size = 0; - size_t val_str_size = 0; - int j; - int check = FLB_FALSE; - int found = FLB_FALSE; - flb_sds_t tag; + flb_sds_t tag = NULL; + struct flb_ra_value *ra_val; - kv = map->via.map.ptr; - - for(j=0; j < map_size; j++) { - check = FLB_FALSE; - found = FLB_FALSE; - key = (kv+j)->key; - if (key.type == MSGPACK_OBJECT_BIN) { - key_str = (char *) key.via.bin.ptr; - key_str_size = key.via.bin.size; - check = FLB_TRUE; - } - if (key.type == MSGPACK_OBJECT_STR) { - key_str = (char *) key.via.str.ptr; - key_str_size = key.via.str.size; - check = FLB_TRUE; - } + /* If no record accessor is configured, return NULL */ + if (!ctx->ra_tag_key) { + return NULL; + } - if (check == FLB_TRUE) { - if (strncmp(ctx->tag_key, key_str, key_str_size) == 0) { - val = (kv+j)->val; - if (val.type == MSGPACK_OBJECT_BIN) { - val_str = (char *) val.via.bin.ptr; - val_str_size = val.via.str.size; - found = FLB_TRUE; - break; - } - if (val.type == MSGPACK_OBJECT_STR) { - val_str = (char *) val.via.str.ptr; - val_str_size = val.via.str.size; - found = FLB_TRUE; - break; - } - } - } + /* Use record accessor to get the value */ + ra_val = flb_ra_get_value_object(ctx->ra_tag_key, *map); + if (!ra_val) { + flb_plg_debug(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key); + return NULL; } - if (found == FLB_TRUE) { - tag = flb_sds_create_len(val_str, val_str_size); - if (!tag) { - flb_errno(); - return NULL; - } - return tag; + /* Convert the value to string */ + if (ra_val->type == FLB_RA_STRING) { + tag = flb_sds_create_len(ra_val->o.via.str.ptr, ra_val->o.via.str.size); + } + else { + flb_plg_debug(ctx->ins, "tag_key %s value is not a string", ctx->tag_key); } + /* Clean up the record accessor value */ + flb_ra_key_value_destroy(ra_val); - flb_plg_error(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key); - return NULL; + return tag; } /*