GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-18 05:36:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

adapter:check state when do with trans data

Browse Source
This commit is contained in:
xiang.zhou
2025-06-09 10:54:45 +08:00
parent b9516ba4d0
commit 5b05867c2b
4 changed files with 25 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/otel/otel_manager.h
+2
View File
@@ -42,6 +42,8 @@ neu_otel_trace_ctx neu_otel_find_trace2(void *req_ctx);
void neu_otel_trace_set_final(neu_otel_trace_ctx ctx);
void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num);
void neu_otel_trace_reduce_expected_span_num(neu_otel_trace_ctx ctx,
uint32_t num);
uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx);
neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id);
src/adapter/adapter.c
+13 -2
View File
@@ -123,8 +123,19 @@ static void *adapter_consumer(void *arg)
nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
adapter->name, header->sender, header->ctx,
neu_reqresp_type_string(header->type), n);
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
if (adapter->state == NEU_NODE_RUNNING_STATE_RUNNING) {
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
} else {
void *ctx = ((neu_reqresp_trans_data_t *) &header[1])->trace_ctx;
if (neu_otel_data_is_started() && ctx) {
neu_otel_trace_ctx trace = neu_otel_find_trace(ctx);
if (trace) {
neu_otel_trace_reduce_expected_span_num(trace, 1);
}
}
}
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
neu_msg_free(msg);
}
src/adapter/driver/driver.c
-14
View File
@@ -2289,20 +2289,6 @@ static int report_callback(void *usr_data)
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) {
src/otel/otel_manager.c
+10
View File
@@ -540,6 +540,16 @@ void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num)
trace_ctx->expected_span_num = num;
}
void neu_otel_trace_reduce_expected_span_num(neu_otel_trace_ctx ctx,
uint32_t num)
{
trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;
trace_ctx->expected_span_num -= num;
if (trace_ctx->expected_span_num <= 0) {
trace_ctx->final = true;
}
}
uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx)
{
trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;