diff --git a/include/neuron/define.h b/include/neuron/define.h index 65778333..9f43d49e 100644 --- a/include/neuron/define.h +++ b/include/neuron/define.h @@ -170,4 +170,8 @@ typedef enum { // forward declaration for neu_adapter_t typedef struct neu_adapter neu_adapter_t; +typedef struct neu_manager neu_manager_t; + +extern neu_manager_t *g_manager; + #endif diff --git a/include/neuron/otel/otel_manager.h b/include/neuron/otel/otel_manager.h index 3782f9af..7de4f82a 100644 --- a/include/neuron/otel/otel_manager.h +++ b/include/neuron/otel/otel_manager.h @@ -47,6 +47,8 @@ neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id); void neu_otel_free_trace(neu_otel_trace_ctx ctx); +void neu_otel_set_internal_parent_span(neu_otel_trace_ctx ctx); + neu_otel_scope_ctx neu_otel_add_span(neu_otel_trace_ctx ctx); neu_otel_scope_ctx neu_otel_add_span2(neu_otel_trace_ctx ctx, const char * span_name, diff --git a/plugins/ekuiper/read_write.c b/plugins/ekuiper/read_write.c index 11d13807..54348958 100644 --- a/plugins/ekuiper/read_write.c +++ b/plugins/ekuiper/read_write.c @@ -120,7 +120,8 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data) nng_msg *msg = NULL; size_t json_len = strlen(json_str); size_t trace_header_len = 0; - if (neu_otel_data_is_started() && trans_data->trace_ctx) { + if (neu_otel_data_is_started() && trans_data->trace_ctx && + trans_trace) { trace_header_len = 26; } diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 3d8e2dbc..603b143e 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -40,6 +40,7 @@ #include "errcodes.h" #include "tag.h" +#include "core/node_manager.h" #include "otel/otel_manager.h" typedef struct to_be_write_tag { @@ -2215,19 +2216,17 @@ static int report_callback(void *usr_data) neu_otel_trace_ctx trans_trace = NULL; neu_otel_scope_ctx trans_scope = NULL; - if (trace_ctx) { - data->trace_ctx = trace_ctx; - if (neu_otel_data_is_started() && data->trace_ctx) { - trans_trace = neu_otel_find_trace(data->trace_ctx); - if (trans_trace) { - char new_span_id[36] = { 0 }; - neu_otel_new_span_id(new_span_id); - trans_scope = - neu_otel_add_span2(trans_trace, "report cb", new_span_id); - neu_otel_scope_add_span_attr_int(trans_scope, "thread id", - (int64_t)(pthread_self())); - neu_otel_scope_set_span_start_time(trans_scope, neu_time_ns()); - } + if (neu_otel_data_is_started() && trace_ctx) { + trans_trace = neu_otel_find_trace(trace_ctx); + if (trans_trace) { + data->trace_ctx = trace_ctx; + char new_span_id[36] = { 0 }; + neu_otel_new_span_id(new_span_id); + trans_scope = + neu_otel_add_span2(trans_trace, "report cb", new_span_id); + neu_otel_scope_add_span_attr_int(trans_scope, "thread id", + (int64_t)(pthread_self())); + neu_otel_scope_set_span_start_time(trans_scope, neu_time_ns()); } } @@ -2256,9 +2255,26 @@ static int report_callback(void *usr_data) } } + if (trans_trace) { + neu_otel_set_internal_parent_span(trans_trace); + } + utarray_foreach(group->apps, sub_app_t *, app) { + neu_adapter_t *sub_app = + neu_node_manager_find(g_manager->node_manager, app->app); + if (sub_app != NULL) { + if (sub_app->state != NEU_NODE_RUNNING_STATE_RUNNING) { + neu_trans_data_free(data); + continue; + } + } else { + nlog_error("sub app %s not exist", app->app); + neu_trans_data_free(data); + continue; + } + if (group->driver->adapter.cb_funs.responseto( &group->driver->adapter, &header, data, app->addr) != 0) { diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index ab2991e3..05e1dd9c 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -30,7 +30,7 @@ #include "subscribe.h" #include "utils/log.h" -typedef struct neu_manager { +struct neu_manager { int server_fd; neu_events_t * events; @@ -45,7 +45,7 @@ typedef struct neu_manager { int64_t timestamp_lev_manager; int log_level; -} neu_manager_t; +}; int neu_manager_add_plugin(neu_manager_t *manager, const char *library); int neu_manager_del_plugin(neu_manager_t *manager, const char *plugin); diff --git a/src/main.c b/src/main.c index 0edef043..57c134e8 100644 --- a/src/main.c +++ b/src/main.c @@ -35,15 +35,15 @@ #include "daemon.h" #include "version.h" -static bool exit_flag = false; -static neu_manager_t *g_manager = NULL; -zlog_category_t * neuron = NULL; -bool disable_jwt = false; -bool sub_filter_err = false; -int default_log_level = ZLOG_LEVEL_NOTICE; -char host_port[32] = { 0 }; -char g_status[32] = { 0 }; -static bool sig_trigger = false; +static bool exit_flag = false; +neu_manager_t * g_manager = NULL; +zlog_category_t *neuron = NULL; +bool disable_jwt = false; +bool sub_filter_err = false; +int default_log_level = ZLOG_LEVEL_NOTICE; +char host_port[32] = { 0 }; +char g_status[32] = { 0 }; +static bool sig_trigger = false; int64_t global_timestamp = 0; diff --git a/src/otel/otel_manager.c b/src/otel/otel_manager.c index fefd92e2..9b7e0ce3 100644 --- a/src/otel/otel_manager.c +++ b/src/otel/otel_manager.c @@ -56,6 +56,7 @@ typedef struct { int64_t ts; UT_array * scopes; pthread_mutex_t mutex; + ProtobufCBinaryData internal_parent_span; } trace_ctx_t; typedef struct { @@ -158,6 +159,18 @@ static int parse_tracestate(const char *tracestate, trace_kv_t *kvs, return 1; } +void neu_otel_set_internal_parent_span(neu_otel_trace_ctx ctx) +{ + trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx; + pthread_mutex_lock(&trace_ctx->mutex); + if (utarray_len(trace_ctx->scopes) > 0) { + trace_scope_t *scope = + *(trace_scope_t **) utarray_back(trace_ctx->scopes); + trace_ctx->internal_parent_span = scope->span->span_id; + } + pthread_mutex_unlock(&trace_ctx->mutex); +} + void neu_otel_free_span(Opentelemetry__Proto__Trace__V1__Span *span) { for (size_t i = 0; i < span->n_attributes; i++) { @@ -693,7 +706,11 @@ neu_otel_scope_ctx neu_otel_add_span2(neu_otel_trace_ctx ctx, scope->span->flags = trace_ctx->flags; neu_otel_scope_set_span_name(scope, span_name); neu_otel_scope_set_span_id(scope, span_id); - if (scope->span_index != 0) { + if (trace_ctx->internal_parent_span.data != NULL) { + neu_otel_scope_set_parent_span_id2(scope, + trace_ctx->internal_parent_span.data, + trace_ctx->internal_parent_span.len); + } else if (scope->span_index != 0) { neu_otel_scope_set_parent_span_id2( scope, trace_ctx->trace_data.resource_spans[0] @@ -1009,19 +1026,25 @@ void neu_otel_split_traceparent(const char *in, char *trace_id, char *span_id, static int otel_timer_cb(void *data) { (void) data; - + int batch_size = 50; + int index = 0; trace_ctx_table_ele_t *el = NULL, *tmp = NULL; pthread_mutex_lock(&table_mutex); HASH_ITER(hh, traces_table, el, tmp) { - if (el->ctx->final && - el->ctx->span_num == - el->ctx->trace_data.resource_spans[0] - ->scope_spans[0] - ->n_spans && - el->ctx->expected_span_num <= 0) { + if (neu_time_ms() - el->ctx->ts >= TRACE_TIME_OUT) { + nlog_debug("trace:%s time out", (char *) el->ctx->trace_id); + HASH_DEL(traces_table, el); + neu_otel_free_trace(el->ctx); + free(el); + } else if (el->ctx->final && + el->ctx->span_num == + el->ctx->trace_data.resource_spans[0] + ->scope_spans[0] + ->n_spans && + el->ctx->expected_span_num <= 0) { int data_size = neu_otel_trace_pack_size(el->ctx); uint8_t *data_buf = calloc(1, data_size); @@ -1034,14 +1057,13 @@ static int otel_timer_cb(void *data) HASH_DEL(traces_table, el); neu_otel_free_trace(el->ctx); free(el); + index += 1; + if (index > batch_size) { + break; + } } else { break; } - } else if (neu_time_ms() - el->ctx->ts >= TRACE_TIME_OUT) { - nlog_debug("trace:%s time out", (char *) el->ctx->trace_id); - HASH_DEL(traces_table, el); - neu_otel_free_trace(el->ctx); - free(el); } } diff --git a/tests/ft/app/test_ekuiper.py b/tests/ft/app/test_ekuiper.py index 81dca973..2a1f5dbf 100644 --- a/tests/ft/app/test_ekuiper.py +++ b/tests/ft/app/test_ekuiper.py @@ -25,7 +25,7 @@ PLUGIN = "eKuiper" DRIVER = "modbus" DRIVER_PORT = random_port() GROUP = "group" -INTERVAL = 100 +INTERVAL = 200 TAGS = [ { "name": "tag0", @@ -126,7 +126,7 @@ def conf_bad_port(conf_base): @pytest.fixture(scope="function") def mocker(): - mock = Mocker(timeout=1000) + mock = Mocker(timeout=2000) yield mock mock.disconnect() @@ -159,7 +159,8 @@ class Mocker: def recv(self): msg = self.sock.recv() - msg = msg[26:] + if msg[0:2] == b'\x0A\xCE': + msg = msg[26:] return json.loads(msg.decode()) def send(self, data):