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:span optimization

Browse Source
This commit is contained in:
xiang.zhou
2025-05-08 10:41:15 +08:00
parent fdfd008858
commit 193b4b96e1
8 changed files with 87 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/define.h
+4
View File
@@ -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
include/neuron/otel/otel_manager.h
+2
View File
@@ -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,
plugins/ekuiper/read_write.c
+2 -1
View File
@@ -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;
}
src/adapter/driver/driver.c
+29 -13
View File
@@ -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) {
src/core/manager_internal.h
+2 -2
View File
@@ -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);
src/main.c
+9 -9
View File
@@ -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;
src/otel/otel_manager.c
+35 -13
View File
@@ -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);
}
}
tests/ft/app/test_ekuiper.py
+4 -3
View File
@@ -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):