diff --git a/include/neuron/otel/otel_manager.h b/include/neuron/otel/otel_manager.h index 7de4f82a..25aad8a0 100644 --- a/include/neuron/otel/otel_manager.h +++ b/include/neuron/otel/otel_manager.h @@ -37,6 +37,7 @@ neu_otel_trace_ctx neu_otel_create_trace(const char *trace_id, void *req_ctx, const char *tracestate); neu_otel_trace_ctx neu_otel_find_trace(void *req_ctx); +neu_otel_trace_ctx neu_otel_find_trace2(void *req_ctx); void neu_otel_trace_set_final(neu_otel_trace_ctx ctx); diff --git a/plugins/modbus/modbus_stack.c b/plugins/modbus/modbus_stack.c index 67f96375..34facfa4 100644 --- a/plugins/modbus/modbus_stack.c +++ b/plugins/modbus/modbus_stack.c @@ -376,21 +376,23 @@ int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id, trace = neu_otel_create_trace( new_trace_id, (void *) (intptr_t) stack->read_seq, 0, trace_state); - scope = neu_otel_add_span(trace); - neu_otel_scope_set_span_name(scope, "driver cmd send"); - char new_span_id[36] = { 0 }; - neu_otel_new_span_id(new_span_id); - neu_otel_scope_set_span_id(scope, new_span_id); - neu_otel_scope_set_span_flags(scope, 0); - neu_otel_scope_set_span_start_time(scope, ts_start); + if (trace) { + scope = neu_otel_add_span(trace); + neu_otel_scope_set_span_name(scope, "driver cmd send"); + char new_span_id[36] = { 0 }; + neu_otel_new_span_id(new_span_id); + neu_otel_scope_set_span_id(scope, new_span_id); + neu_otel_scope_set_span_flags(scope, 0); + neu_otel_scope_set_span_start_time(scope, ts_start); - neu_otel_scope_add_span_attr_int(scope, "thread id", - (int64_t) pthread_self()); + neu_otel_scope_add_span_attr_int( + scope, "thread id", (int64_t) pthread_self()); - neu_otel_scope_add_span_attr_string( - scope, "node", - ((neu_plugin_t *) stack->ctx)->common.name); - neu_otel_scope_set_span_end_time(scope, neu_time_ns()); + neu_otel_scope_add_span_attr_string( + scope, "node", + ((neu_plugin_t *) stack->ctx)->common.name); + neu_otel_scope_set_span_end_time(scope, neu_time_ns()); + } } } } diff --git a/plugins/restful/rw_handle.c b/plugins/restful/rw_handle.c index a750f53c..c92dbe60 100644 --- a/plugins/restful/rw_handle.c +++ b/plugins/restful/rw_handle.c @@ -234,36 +234,39 @@ void handle_write(nng_aio *aio) trace_flag = true; trace = neu_otel_create_trace(trace_id, header.ctx, flags, trace_state); - scope = neu_otel_add_span(trace); - neu_otel_scope_set_span_name( - scope, nng_http_req_get_uri(nng_req)); - char new_span_id[36] = { 0 }; - neu_otel_new_span_id(new_span_id); - neu_otel_scope_set_span_id(scope, new_span_id); - if (strlen(span_id) == 16) { - neu_otel_scope_set_parent_span_id(scope, span_id); - } - neu_otel_scope_set_span_flags(scope, flags); - neu_otel_scope_set_span_start_time(scope, - neu_time_ns()); + if (trace) { + scope = neu_otel_add_span(trace); + neu_otel_scope_set_span_name( + scope, nng_http_req_get_uri(nng_req)); + char new_span_id[36] = { 0 }; + neu_otel_new_span_id(new_span_id); + neu_otel_scope_set_span_id(scope, new_span_id); + if (strlen(span_id) == 16) { + neu_otel_scope_set_parent_span_id(scope, + span_id); + } + neu_otel_scope_set_span_flags(scope, flags); + neu_otel_scope_set_span_start_time(scope, + neu_time_ns()); - neu_otel_scope_add_span_attr_int( - scope, "thread id", (int64_t) pthread_self()); + neu_otel_scope_add_span_attr_int( + scope, "thread id", (int64_t) pthread_self()); - neu_otel_scope_add_span_attr_string( - scope, "HTTP method", - nng_http_req_get_method(nng_req)); - - char * req_data = NULL; - size_t req_data_size = 0; - - if (neu_http_get_body((aio), (void **) &req_data, - &req_data_size) == 0) { neu_otel_scope_add_span_attr_string( - scope, "HTTP body", req_data); - } + scope, "HTTP method", + nng_http_req_get_method(nng_req)); - free(req_data); + char * req_data = NULL; + size_t req_data_size = 0; + + if (neu_http_get_body((aio), (void **) &req_data, + &req_data_size) == 0) { + neu_otel_scope_add_span_attr_string( + scope, "HTTP body", req_data); + } + + free(req_data); + } } } } @@ -445,36 +448,39 @@ void handle_write_tags(nng_aio *aio) trace_flag = true; trace = neu_otel_create_trace(trace_id, header.ctx, flags, trace_state); - scope = neu_otel_add_span(trace); - neu_otel_scope_set_span_name( - scope, nng_http_req_get_uri(nng_req)); - char new_span_id[36] = { 0 }; - neu_otel_new_span_id(new_span_id); - neu_otel_scope_set_span_id(scope, new_span_id); - if (strlen(span_id) == 16) { - neu_otel_scope_set_parent_span_id(scope, span_id); - } - neu_otel_scope_set_span_flags(scope, flags); - neu_otel_scope_set_span_start_time(scope, - neu_time_ns()); + if (trace) { + scope = neu_otel_add_span(trace); + neu_otel_scope_set_span_name( + scope, nng_http_req_get_uri(nng_req)); + char new_span_id[36] = { 0 }; + neu_otel_new_span_id(new_span_id); + neu_otel_scope_set_span_id(scope, new_span_id); + if (strlen(span_id) == 16) { + neu_otel_scope_set_parent_span_id(scope, + span_id); + } + neu_otel_scope_set_span_flags(scope, flags); + neu_otel_scope_set_span_start_time(scope, + neu_time_ns()); - neu_otel_scope_add_span_attr_int( - scope, "thread id", (int64_t) pthread_self()); + neu_otel_scope_add_span_attr_int( + scope, "thread id", (int64_t) pthread_self()); - neu_otel_scope_add_span_attr_string( - scope, "HTTP method", - nng_http_req_get_method(nng_req)); - - char * req_data = NULL; - size_t req_data_size = 0; - - if (neu_http_get_body((aio), (void **) &req_data, - &req_data_size) == 0) { neu_otel_scope_add_span_attr_string( - scope, "HTTP body", req_data); - } + scope, "HTTP method", + nng_http_req_get_method(nng_req)); - free(req_data); + char * req_data = NULL; + size_t req_data_size = 0; + + if (neu_http_get_body((aio), (void **) &req_data, + &req_data_size) == 0) { + neu_otel_scope_add_span_attr_string( + scope, "HTTP body", req_data); + } + + free(req_data); + } } } } @@ -775,36 +781,39 @@ void handle_write_gtags(nng_aio *aio) trace_flag = true; trace = neu_otel_create_trace(trace_id, header.ctx, flags, trace_state); - scope = neu_otel_add_span(trace); - neu_otel_scope_set_span_name( - scope, nng_http_req_get_uri(nng_req)); - char new_span_id[36] = { 0 }; - neu_otel_new_span_id(new_span_id); - neu_otel_scope_set_span_id(scope, new_span_id); - if (strlen(span_id) == 16) { - neu_otel_scope_set_parent_span_id(scope, span_id); - } - neu_otel_scope_set_span_flags(scope, flags); - neu_otel_scope_set_span_start_time(scope, - neu_time_ns()); + if (trace) { + scope = neu_otel_add_span(trace); + neu_otel_scope_set_span_name( + scope, nng_http_req_get_uri(nng_req)); + char new_span_id[36] = { 0 }; + neu_otel_new_span_id(new_span_id); + neu_otel_scope_set_span_id(scope, new_span_id); + if (strlen(span_id) == 16) { + neu_otel_scope_set_parent_span_id(scope, + span_id); + } + neu_otel_scope_set_span_flags(scope, flags); + neu_otel_scope_set_span_start_time(scope, + neu_time_ns()); - neu_otel_scope_add_span_attr_int( - scope, "thread id", (int64_t) pthread_self()); + neu_otel_scope_add_span_attr_int( + scope, "thread id", (int64_t) pthread_self()); - neu_otel_scope_add_span_attr_string( - scope, "HTTP method", - nng_http_req_get_method(nng_req)); - - char * req_data = NULL; - size_t req_data_size = 0; - - if (neu_http_get_body((aio), (void **) &req_data, - &req_data_size) == 0) { neu_otel_scope_add_span_attr_string( - scope, "HTTP body", req_data); - } + scope, "HTTP method", + nng_http_req_get_method(nng_req)); - free(req_data); + char * req_data = NULL; + size_t req_data_size = 0; + + if (neu_http_get_body((aio), (void **) &req_data, + &req_data_size) == 0) { + neu_otel_scope_add_span_attr_string( + scope, "HTTP body", req_data); + } + + free(req_data); + } } } } diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 3a47527c..d95a04ff 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -2245,7 +2245,7 @@ static int report_callback(void *usr_data) neu_otel_trace_ctx trans_trace = NULL; neu_otel_scope_ctx trans_scope = NULL; if (neu_otel_data_is_started() && trace_ctx) { - trans_trace = neu_otel_find_trace(trace_ctx); + trans_trace = neu_otel_find_trace2(trace_ctx); if (trans_trace) { data->trace_ctx = trace_ctx; char new_span_id[36] = { 0 }; diff --git a/src/base/neu_plugin_common.c b/src/base/neu_plugin_common.c index 13cd381d..5c051c18 100644 --- a/src/base/neu_plugin_common.c +++ b/src/base/neu_plugin_common.c @@ -72,35 +72,38 @@ int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data) if (strlen(trace_id) == 32) { trace = neu_otel_create_trace(trace_id, head.ctx, flags, trace_state); - scope = neu_otel_add_span(trace); - neu_otel_scope_set_span_name(scope, - nng_http_req_get_uri(nng_req)); - char new_span_id[36] = { 0 }; - neu_otel_new_span_id(new_span_id); - neu_otel_scope_set_span_id(scope, new_span_id); - if (strlen(span_id) == 16) { - neu_otel_scope_set_parent_span_id(scope, span_id); - } - neu_otel_scope_set_span_flags(scope, flags); - neu_otel_scope_set_span_start_time(scope, neu_time_ns()); - - neu_otel_scope_add_span_attr_int(scope, "thread id", - (int64_t) pthread_self()); - - neu_otel_scope_add_span_attr_string(scope, "HTTP method", - http_method); - - if (strcmp(http_method, "POST") == 0) { - char * req_data = NULL; - size_t req_data_size = 0; - - if (neu_http_get_body((aio), (void **) &req_data, - &req_data_size) == 0) { - neu_otel_scope_add_span_attr_string( - scope, "HTTP body", req_data); + if (trace) { + scope = neu_otel_add_span(trace); + neu_otel_scope_set_span_name( + scope, nng_http_req_get_uri(nng_req)); + char new_span_id[36] = { 0 }; + neu_otel_new_span_id(new_span_id); + neu_otel_scope_set_span_id(scope, new_span_id); + if (strlen(span_id) == 16) { + neu_otel_scope_set_parent_span_id(scope, span_id); } + neu_otel_scope_set_span_flags(scope, flags); + neu_otel_scope_set_span_start_time(scope, + neu_time_ns()); - free(req_data); + neu_otel_scope_add_span_attr_int( + scope, "thread id", (int64_t) pthread_self()); + + neu_otel_scope_add_span_attr_string( + scope, "HTTP method", http_method); + + if (strcmp(http_method, "POST") == 0) { + char * req_data = NULL; + size_t req_data_size = 0; + + if (neu_http_get_body((aio), (void **) &req_data, + &req_data_size) == 0) { + neu_otel_scope_add_span_attr_string( + scope, "HTTP body", req_data); + } + + free(req_data); + } } } } @@ -119,26 +122,29 @@ int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data) if (strlen(trace_id) == 32) { trace = neu_otel_create_trace(trace_id, head.ctx, flags, trace_state); - scope = neu_otel_add_span(trace); - neu_otel_scope_set_span_name(scope, "MQTT command"); - char new_span_id[36] = { 0 }; - neu_otel_new_span_id(new_span_id); - neu_otel_scope_set_span_id(scope, new_span_id); - if (strlen(span_id) == 16) { - neu_otel_scope_set_parent_span_id(scope, span_id); - } - neu_otel_scope_set_span_flags(scope, flags); - neu_otel_scope_set_span_start_time(scope, neu_time_ns()); + if (trace) { + scope = neu_otel_add_span(trace); + neu_otel_scope_set_span_name(scope, "MQTT command"); + char new_span_id[36] = { 0 }; + neu_otel_new_span_id(new_span_id); + neu_otel_scope_set_span_id(scope, new_span_id); + if (strlen(span_id) == 16) { + neu_otel_scope_set_parent_span_id(scope, span_id); + } + neu_otel_scope_set_span_flags(scope, flags); + neu_otel_scope_set_span_start_time(scope, + neu_time_ns()); - neu_otel_scope_add_span_attr_int(scope, "thread id", - (int64_t) pthread_self()); + neu_otel_scope_add_span_attr_int( + scope, "thread id", (int64_t) pthread_self()); - neu_otel_scope_add_span_attr_string( - scope, "MQTT command uuid", d_mqtt->uuid); - - if (d_mqtt->payload) { neu_otel_scope_add_span_attr_string( - scope, "MQTT command payload", d_mqtt->payload); + scope, "MQTT command uuid", d_mqtt->uuid); + + if (d_mqtt->payload) { + neu_otel_scope_add_span_attr_string( + scope, "MQTT command payload", d_mqtt->payload); + } } } } @@ -157,24 +163,28 @@ int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data) uint32_t flags = 0; trace = neu_otel_create_trace(trace_id, head.ctx, flags, trace_state); - scope = neu_otel_add_span(trace); - neu_otel_scope_set_span_name(scope, "ekuiper command"); - char new_span_id[36] = { 0 }; - neu_otel_new_span_id(new_span_id); - neu_otel_scope_set_span_id(scope, new_span_id); - if (strlen(span_id) == 16) { - neu_otel_scope_set_parent_span_id(scope, span_id); + if (trace) { + scope = neu_otel_add_span(trace); + neu_otel_scope_set_span_name(scope, + "ekuiper command"); + char new_span_id[36] = { 0 }; + neu_otel_new_span_id(new_span_id); + neu_otel_scope_set_span_id(scope, new_span_id); + if (strlen(span_id) == 16) { + neu_otel_scope_set_parent_span_id(scope, + span_id); + } + neu_otel_scope_set_span_flags(scope, flags); + neu_otel_scope_set_span_start_time(scope, + neu_time_ns()); + + neu_otel_scope_add_span_attr_int( + scope, "thread id", (int64_t) pthread_self()); + + neu_otel_scope_add_span_attr_string( + scope, "ekuiper command playload", + (char *) head.ctx + 48); } - neu_otel_scope_set_span_flags(scope, flags); - neu_otel_scope_set_span_start_time(scope, - neu_time_ns()); - - neu_otel_scope_add_span_attr_int( - scope, "thread id", (int64_t) pthread_self()); - - neu_otel_scope_add_span_attr_string( - scope, "ekuiper command playload", - (char *) head.ctx + 48); } } } diff --git a/src/otel/otel_manager.c b/src/otel/otel_manager.c index 9b7e0ce3..dfff5fd2 100644 --- a/src/otel/otel_manager.c +++ b/src/otel/otel_manager.c @@ -57,6 +57,7 @@ typedef struct { UT_array * scopes; pthread_mutex_t mutex; ProtobufCBinaryData internal_parent_span; + bool report_flag; } trace_ctx_t; typedef struct { @@ -206,9 +207,8 @@ neu_otel_trace_ctx neu_otel_create_trace(const char *trace_id, void *req_ctx, HASH_FIND(hh, traces_table, &req_ctx, sizeof(req_ctx), find); if (find) { - HASH_DEL(traces_table, find); - neu_otel_free_trace(find->ctx); - free(find); + pthread_mutex_unlock(&table_mutex); + return NULL; } pthread_mutex_unlock(&table_mutex); @@ -500,6 +500,34 @@ neu_otel_trace_ctx neu_otel_find_trace(void *req_ctx) } } +neu_otel_trace_ctx neu_otel_find_trace2(void *req_ctx) +{ + trace_ctx_table_ele_t *find = NULL; + + pthread_mutex_lock(&table_mutex); + + HASH_FIND(hh, traces_table, &req_ctx, sizeof(req_ctx), find); + + pthread_mutex_unlock(&table_mutex); + + if (find) { + if ((find->ctx->final && + find->ctx->span_num == + find->ctx->trace_data.resource_spans[0] + ->scope_spans[0] + ->n_spans && + find->ctx->expected_span_num <= 0) || + find->ctx->report_flag) { + return NULL; + } else { + find->ctx->report_flag = true; + return (void *) find->ctx; + } + } else { + return NULL; + } +} + void neu_otel_trace_set_final(neu_otel_trace_ctx ctx) { trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;