GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

otel:ensure the uniqueness of ctx

Browse Source
This commit is contained in:
xiang.zhou
2025-05-21 11:26:09 +08:00
parent fb0bcbf116
commit 0d507761fa
6 changed files with 206 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/otel/otel_manager.h
+1
View File
@@ -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);
plugins/modbus/modbus_stack.c
+15 -13
View File
@@ -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());
}
}
}
}
plugins/restful/rw_handle.c
+87 -78
View File
@@ -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);
}
}
}
}
src/adapter/driver/driver.c
+1 -1
View File
@@ -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 };
src/base/neu_plugin_common.c
+71 -61
View File
@@ -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);
}
}
}
src/otel/otel_manager.c
+31 -3
View File
@@ -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;