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

Merge pull request #2257 from ZhouBox/main

Browse Source 
otel:add new api and support ns
This commit is contained in:
fengzeroz
2024-10-16 10:31:22 +08:00
committed by GitHub
parent caec4d1585 51333961d6
commit 5f5a376859
15 changed files with 326 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/otel/otel_manager.h
+18 -2
View File
@@ -23,6 +23,12 @@
#include <stdbool.h>
#include <stdint.h>
typedef enum {
NEU_OTEL_STATUS_UNSET = 0,
NEU_OTEL_STATUS_OK = 1,
NEU_OTEL_STATUS_ERROR = 2,
} neu_otel_status_code_e;
typedef void *neu_otel_trace_ctx;
typedef void *neu_otel_scope_ctx;
@@ -42,6 +48,9 @@ neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id);
void neu_otel_free_trace(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,
const char * span_id);
void neu_otel_scope_set_parent_span_id(neu_otel_scope_ctx ctx,
const char * parent_span_id);
@@ -66,11 +75,18 @@ void neu_otel_scope_add_span_attr_string(neu_otel_scope_ctx ctx,
void neu_otel_scope_add_span_attr_bool(neu_otel_scope_ctx ctx, const char *key,
bool val);
void neu_otel_scope_set_span_start_time(neu_otel_scope_ctx ctx, int64_t ms);
void neu_otel_scope_set_span_end_time(neu_otel_scope_ctx ctx, int64_t ms);
void neu_otel_scope_set_span_start_time(neu_otel_scope_ctx ctx, int64_t ns);
void neu_otel_scope_set_span_end_time(neu_otel_scope_ctx ctx, int64_t ns);
uint8_t *neu_otel_scope_get_pre_span_id(neu_otel_scope_ctx ctx);
void neu_otel_scope_set_status_code(neu_otel_scope_ctx ctx,
neu_otel_status_code_e code,
const char * desc);
void neu_otel_scope_set_status_code2(neu_otel_scope_ctx ctx,
neu_otel_status_code_e code, int errorno);
int neu_otel_trace_pack_size(neu_otel_trace_ctx ctx);
int neu_otel_trace_pack(neu_otel_trace_ctx ctx, uint8_t *out);
include/neuron/utils/time.h
+7
View File
@@ -35,6 +35,13 @@ static inline int64_t neu_time_ms()
return (int64_t) tv.tv_sec * 1000 + (int64_t) tv.tv_usec / 1000;
}
static inline int64_t neu_time_ns()
{
struct timespec ts = { 0 };
clock_gettime(CLOCK_REALTIME, &ts);
return (int64_t) ts.tv_sec * 1000000000 + (int64_t) ts.tv_nsec;
}
static inline void neu_msleep(unsigned msec)
{
struct timespec tv = {
plugins/ekuiper/plugin_ekuiper.c
+16 -16
View File
@@ -330,7 +330,7 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin,
}
neu_otel_scope_add_span_attr_int(scope, "thread id",
(int64_t)(pthread_self()));
neu_otel_scope_set_span_start_time(scope, neu_time_ms());
neu_otel_scope_set_span_start_time(scope, neu_time_ns());
}
}
@@ -339,8 +339,14 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin,
neu_resp_error_t *error = (neu_resp_error_t *) data;
plog_debug(plugin, "receive resp errcode: %d", error->error);
if (trace) {
neu_otel_scope_add_span_attr_int(scope, "error", error->error);
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
if (error->error != NEU_ERR_SUCCESS) {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
error->error);
} else {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK,
error->error);
}
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
neu_otel_trace_set_final(trace);
}
if (header->ctx) {
@@ -367,25 +373,19 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin,
if (neu_otel_data_is_started() && trans_data->trace_ctx) {
trans_trace = neu_otel_find_trace(trans_data->trace_ctx);
if (trans_trace) {
trans_scope = neu_otel_add_span(trans_trace);
neu_otel_scope_set_span_name(trans_scope, "ekuiper send");
char new_span_id[36] = { 0 };
neu_otel_new_span_id(new_span_id);
neu_otel_scope_set_span_id(trans_scope, new_span_id);
uint8_t *p_sp_id =
neu_otel_scope_get_pre_span_id(trans_scope);
if (p_sp_id) {
neu_otel_scope_set_parent_span_id2(trans_scope, p_sp_id,
8);
}
trans_scope = neu_otel_add_span2(
trans_trace, "ekuiper send", new_span_id);
neu_otel_scope_add_span_attr_int(trans_scope, "thread id",
(int64_t)(pthread_self()));
neu_otel_scope_add_span_attr_int(
trans_scope, "error", NEU_ERR_PLUGIN_DISCONNECTED);
neu_otel_scope_set_status_code2(
trans_scope, NEU_OTEL_STATUS_ERROR,
NEU_ERR_PLUGIN_DISCONNECTED);
neu_otel_scope_set_span_start_time(trans_scope,
neu_time_ms());
neu_time_ns());
neu_otel_scope_set_span_end_time(trans_scope,
neu_time_ms());
neu_time_ns());
neu_otel_trace_set_final(trans_trace);
}
}
plugins/ekuiper/read_write.c
+10 -12
View File
@@ -100,17 +100,12 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
if (neu_otel_data_is_started() && trans_data->trace_ctx) {
trans_trace = neu_otel_find_trace(trans_data->trace_ctx);
if (trans_trace) {
trans_scope = neu_otel_add_span(trans_trace);
neu_otel_scope_set_span_name(trans_scope, "ekuiper send");
neu_otel_new_span_id(new_span_id);
neu_otel_scope_set_span_id(trans_scope, new_span_id);
uint8_t *p_sp_id = neu_otel_scope_get_pre_span_id(trans_scope);
if (p_sp_id) {
neu_otel_scope_set_parent_span_id2(trans_scope, p_sp_id, 8);
}
trans_scope =
neu_otel_add_span2(trans_trace, "ekuiper send", 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_ms());
neu_otel_scope_set_span_start_time(trans_scope, neu_time_ns());
}
}
@@ -137,8 +132,6 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
}
if (trans_trace) {
neu_otel_scope_add_span_attr_string(trans_scope, "playload",
json_str);
trace_id = neu_otel_get_trace_id(trans_trace);
uint8_t span_id[8] = { 0 };
hex_string_to_binary(new_span_id, span_id, 8);
@@ -172,8 +165,13 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
} while (0);
if (trans_trace) {
neu_otel_scope_add_span_attr_int(trans_scope, "error", rv);
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
if (rv == 0) {
neu_otel_scope_set_status_code2(trans_scope, NEU_OTEL_STATUS_OK, 0);
} else {
neu_otel_scope_set_status_code(trans_scope, NEU_OTEL_STATUS_ERROR,
nng_strerror(rv));
}
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ns());
neu_otel_trace_set_final(trans_trace);
}
}
plugins/mqtt/azure_iot_plugin.c
+2 -1
View File
@@ -330,7 +330,8 @@ static int azure_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
switch (head->type) {
case NEU_RESP_ERROR:
error = handle_write_response(plugin, head->ctx, data, NULL);
error =
handle_write_response(plugin, head->ctx, data, NULL, NULL, NULL);
break;
case NEU_RESP_READ_GROUP:
// error = handle_read_response(plugin, head->ctx, data);
plugins/mqtt/mqtt_handle.c
+25 -18
View File
@@ -460,7 +460,8 @@ void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
}
int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_resp_error_t *data, void *trace_scope)
neu_resp_error_t *data, void *trace_scope,
void *trace_ctx, char *span_id)
{
int rv = 0;
char *json_str = NULL;
@@ -489,9 +490,21 @@ int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_otel_scope_add_span_attr_string(trace_scope, "playload", json_str);
}
char trace_parent[128] = { 0 };
if (trace_ctx && span_id) {
uint8_t *trace_id = neu_otel_get_trace_id(trace_ctx);
to_traceparent(trace_id, span_id, trace_parent);
}
char * topic = plugin->config.write_resp_topic;
neu_mqtt_qos_e qos = plugin->config.qos;
rv = publish(plugin, qos, topic, json_str, strlen(json_str));
if (trace_ctx != 0 && strlen(trace_parent) != 0) {
rv = publish_with_trace(plugin, qos, topic, json_str, strlen(json_str),
trace_parent);
} else {
rv = publish(plugin, qos, topic, json_str, strlen(json_str));
}
json_str = NULL;
end:
@@ -607,18 +620,13 @@ int handle_trans_data(neu_plugin_t * plugin,
if (neu_otel_data_is_started() && trans_data->trace_ctx) {
trans_trace = neu_otel_find_trace(trans_data->trace_ctx);
if (trans_trace) {
trans_scope = neu_otel_add_span(trans_trace);
neu_otel_scope_set_span_name(trans_scope, "mqtt publish");
char new_span_id[36] = { 0 };
neu_otel_new_span_id(new_span_id);
neu_otel_scope_set_span_id(trans_scope, new_span_id);
uint8_t *p_sp_id = neu_otel_scope_get_pre_span_id(trans_scope);
if (p_sp_id) {
neu_otel_scope_set_parent_span_id2(trans_scope, p_sp_id, 8);
}
trans_scope =
neu_otel_add_span2(trans_trace, "mqtt publish", 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_ms());
neu_otel_scope_set_span_start_time(trans_scope, neu_time_ns());
trace_id = neu_otel_get_trace_id(trans_trace);
@@ -661,15 +669,9 @@ int handle_trans_data(neu_plugin_t * plugin,
neu_mqtt_qos_e qos = plugin->config.qos;
if (plugin->config.version == NEU_MQTT_VERSION_V5 && trans_trace) {
neu_otel_scope_add_span_attr_string(trans_scope, "playload",
json_str);
rv = publish_with_trace(plugin, qos, topic, json_str,
strlen(json_str), trace_parent);
} else {
if (trans_trace) {
neu_otel_scope_add_span_attr_string(trans_scope, "playload",
json_str);
}
rv = publish(plugin, qos, topic, json_str, strlen(json_str));
}
@@ -677,8 +679,13 @@ int handle_trans_data(neu_plugin_t * plugin,
} while (0);
if (trans_trace) {
neu_otel_scope_add_span_attr_int(trans_scope, "error", rv);
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
if (rv == NEU_ERR_SUCCESS) {
neu_otel_scope_set_status_code2(trans_scope, NEU_OTEL_STATUS_OK, 0);
} else {
neu_otel_scope_set_status_code2(trans_scope, NEU_OTEL_STATUS_ERROR,
rv);
}
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ns());
neu_otel_trace_set_final(trans_trace);
}
plugins/mqtt/mqtt_handle.h
+2 -1
View File
@@ -40,7 +40,8 @@ void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
trace_w3c_t *trace_w3c);
int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_resp_error_t *data, void *trace_scope);
neu_resp_error_t *data, void *trace_scope,
void *trace_ctx, char *span_id);
void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
const uint8_t *payload, uint32_t len, void *data,
plugins/mqtt/mqtt_plugin_intf.c
+13 -7
View File
@@ -460,14 +460,14 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
plugin->cache_metric_update_ts = global_timestamp;
}
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
char new_span_id[36] = { 0 };
if (neu_otel_control_is_started()) {
trace = neu_otel_find_trace(head->ctx);
if (trace) {
scope = neu_otel_add_span(trace);
neu_otel_scope_set_span_name(scope, "mqtt response");
char new_span_id[36] = { 0 };
neu_otel_new_span_id(new_span_id);
neu_otel_scope_set_span_id(scope, new_span_id);
uint8_t *p_sp_id = neu_otel_scope_get_pre_span_id(scope);
@@ -476,13 +476,14 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
}
neu_otel_scope_add_span_attr_int(scope, "thread id",
(int64_t)(pthread_self()));
neu_otel_scope_set_span_start_time(scope, neu_time_ms());
neu_otel_scope_set_span_start_time(scope, neu_time_ns());
}
}
switch (head->type) {
case NEU_RESP_ERROR:
error = handle_write_response(plugin, head->ctx, data, scope);
error = handle_write_response(plugin, head->ctx, data, scope, trace,
new_span_id);
break;
case NEU_RESP_GET_NODES_STATE: {
handle_nodes_state(head->ctx, (neu_resp_get_nodes_state_t *) data);
@@ -537,8 +538,13 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
}
if (trace) {
neu_otel_scope_add_span_attr_int(scope, "handle error", error);
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
if (error == NEU_ERR_SUCCESS) {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
} else {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
error);
}
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
neu_otel_trace_set_final(trace);
}
plugins/restful/rest.c
+30 -17
View File
@@ -190,7 +190,7 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
}
neu_otel_scope_add_span_attr_int(scope, "thread id",
(int64_t)(pthread_self()));
neu_otel_scope_set_span_start_time(scope, neu_time_ms());
neu_otel_scope_set_span_start_time(scope, neu_time_ns());
}
}
@@ -201,33 +201,39 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
neu_http_response(header->ctx, error->error, result_error);
});
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", error->error);
if (error->error != NEU_ERR_SUCCESS) {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
error->error);
} else {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK,
error->error);
}
}
break;
}
case NEU_RESP_GET_PLUGIN:
handle_get_plugin_resp(header->ctx, (neu_resp_get_plugin_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_NODE:
handle_get_adapter_resp(header->ctx, (neu_resp_get_node_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_GROUP:
handle_get_group_resp(header->ctx, (neu_resp_get_group_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_DRIVER_GROUP:
handle_get_driver_group_resp(header->ctx,
(neu_resp_get_driver_group_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_ADD_TAG:
@@ -254,62 +260,69 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
case NEU_RESP_GET_TAG:
handle_get_tags_resp(header->ctx, (neu_resp_get_tag_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_SUBSCRIBE_GROUP:
handle_grp_get_subscribe_resp(header->ctx,
(neu_resp_get_subscribe_group_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_NODE_SETTING:
handle_get_node_setting_resp(header->ctx,
(neu_resp_get_node_setting_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_NODES_STATE:
handle_get_nodes_state_resp(header->ctx,
(neu_resp_get_nodes_state_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_NODE_STATE:
handle_get_node_state_resp(header->ctx,
(neu_resp_get_node_state_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_READ_GROUP:
handle_read_resp(header->ctx, (neu_resp_read_group_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_READ_GROUP_PAGINATE:
handle_read_paginate_resp(header->ctx,
(neu_resp_read_group_paginate_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_TEST_READ_TAG:
handle_test_read_tag_resp(header->ctx,
(neu_resp_test_read_tag_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", 0);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_SCAN_TAGS:
handle_scan_tags_resp(header->ctx, (neu_resp_scan_tags_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(
scope, "error", ((neu_resp_scan_tags_t *) data)->error);
if (((neu_resp_scan_tags_t *) data)->error != NEU_ERR_SUCCESS) {
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_ERROR,
((neu_resp_scan_tags_t *) data)->error);
} else {
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_OK,
((neu_resp_scan_tags_t *) data)->error);
}
}
break;
default:
@@ -320,7 +333,7 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
}
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
neu_otel_trace_set_final(trace);
}
plugins/restful/rw_handle.c
+23 -21
View File
@@ -245,7 +245,7 @@ void handle_write(nng_aio *aio)
}
neu_otel_scope_set_span_flags(scope, flags);
neu_otel_scope_set_span_start_time(scope,
neu_time_ms());
neu_time_ns());
neu_otel_scope_add_span_attr_int(
scope, "thread id", (int64_t) pthread_self());
@@ -275,9 +275,9 @@ void handle_write(nng_aio *aio)
result_error);
});
if (neu_otel_control_is_started() && trace_flag) {
neu_otel_scope_add_span_attr_int(scope, "error type",
NEU_ERR_STRING_TOO_LONG);
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_ERROR, NEU_ERR_STRING_TOO_LONG);
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
neu_otel_trace_set_final(trace);
}
return;
@@ -290,9 +290,9 @@ void handle_write(nng_aio *aio)
result_error);
});
if (neu_otel_control_is_started() && trace_flag) {
neu_otel_scope_add_span_attr_int(scope, "error type",
NEU_ERR_STRING_TOO_LONG);
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_ERROR, NEU_ERR_STRING_TOO_LONG);
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
neu_otel_trace_set_final(trace);
}
return;
@@ -355,7 +355,7 @@ void handle_write(nng_aio *aio)
if (neu_otel_control_is_started() && trace_flag) {
ret = neu_plugin_op(plugin, header, &cmd);
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
} else {
ret = neu_plugin_op(plugin, header, &cmd);
}
@@ -366,8 +366,8 @@ void handle_write(nng_aio *aio)
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
if (neu_otel_control_is_started() && trace_flag) {
neu_otel_scope_add_span_attr_int(scope, "error type",
NEU_ERR_IS_BUSY);
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_ERROR, NEU_ERR_IS_BUSY);
neu_otel_trace_set_final(trace);
}
}
@@ -377,7 +377,8 @@ void handle_write(nng_aio *aio)
NEU_JSON_RESPONSE_ERROR(
err_type, { neu_http_response(aio, err_type, result_error); });
if (neu_otel_control_is_started() && trace_flag) {
neu_otel_scope_add_span_attr_int(scope, "error type", err_type);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
err_type);
neu_otel_trace_set_final(trace);
}
@@ -432,7 +433,7 @@ void handle_write_tags(nng_aio *aio)
}
neu_otel_scope_set_span_flags(scope, flags);
neu_otel_scope_set_span_start_time(scope,
neu_time_ms());
neu_time_ns());
neu_otel_scope_add_span_attr_int(
scope, "thread id", (int64_t) pthread_self());
@@ -466,7 +467,7 @@ void handle_write_tags(nng_aio *aio)
neu_otel_scope_add_span_attr_int(
scope, "error type", NEU_ERR_STRING_TOO_LONG);
neu_otel_scope_set_span_end_time(scope,
neu_time_ms());
neu_time_ns());
neu_otel_trace_set_final(trace);
}
return;
@@ -530,7 +531,7 @@ void handle_write_tags(nng_aio *aio)
if (neu_otel_control_is_started() && trace_flag) {
ret = neu_plugin_op(plugin, header, &cmd);
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
} else {
ret = neu_plugin_op(plugin, header, &cmd);
}
@@ -541,8 +542,8 @@ void handle_write_tags(nng_aio *aio)
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
if (neu_otel_control_is_started() && trace_flag) {
neu_otel_scope_add_span_attr_int(scope, "error type",
NEU_ERR_IS_BUSY);
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_ERROR, NEU_ERR_IS_BUSY);
neu_otel_trace_set_final(trace);
}
}
@@ -552,7 +553,8 @@ void handle_write_tags(nng_aio *aio)
NEU_JSON_RESPONSE_ERROR(
err_type, { neu_http_response(aio, err_type, result_error); });
if (neu_otel_control_is_started() && trace_flag) {
neu_otel_scope_add_span_attr_int(scope, "error type", err_type);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
err_type);
neu_otel_trace_set_final(trace);
}
@@ -659,7 +661,7 @@ void handle_write_gtags(nng_aio *aio)
}
neu_otel_scope_set_span_flags(scope, flags);
neu_otel_scope_set_span_start_time(scope,
neu_time_ms());
neu_time_ns());
neu_otel_scope_add_span_attr_int(
scope, "thread id", (int64_t) pthread_self());
@@ -688,7 +690,7 @@ void handle_write_gtags(nng_aio *aio)
if (neu_otel_control_is_started() && trace_flag) {
ret = neu_plugin_op(plugin, header, &cmd);
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
} else {
ret = neu_plugin_op(plugin, header, &cmd);
}
@@ -699,8 +701,8 @@ void handle_write_gtags(nng_aio *aio)
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
if (neu_otel_control_is_started() && trace_flag) {
neu_otel_scope_add_span_attr_int(scope, "error type",
NEU_ERR_IS_BUSY);
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_ERROR, NEU_ERR_IS_BUSY);
neu_otel_trace_set_final(trace);
}
}
src/adapter/adapter.c
+37 -15
View File
@@ -790,7 +790,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
neu_otel_scope_add_span_attr_int(scope, "thread id",
(int64_t) pthread_self());
neu_otel_scope_set_span_start_time(scope, neu_time_ms());
neu_otel_scope_set_span_start_time(scope, neu_time_ns());
}
}
@@ -803,7 +803,13 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
re_flag = true;
} else {
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", w_error);
if (w_error != NEU_ERR_SUCCESS) {
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_ERROR, w_error);
} else {
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_OK, w_error);
}
}
}
} else {
@@ -813,13 +819,13 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_msg_exchange(header);
reply(adapter, header, &error);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error",
NEU_ERR_GROUP_NOT_ALLOW);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
NEU_ERR_GROUP_NOT_ALLOW);
}
}
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
if (!re_flag) {
neu_otel_trace_set_final(trace);
}
@@ -846,7 +852,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
neu_otel_scope_add_span_attr_int(scope, "thread id",
(int64_t) pthread_self());
neu_otel_scope_set_span_start_time(scope, neu_time_ms());
neu_otel_scope_set_span_start_time(scope, neu_time_ns());
}
}
@@ -859,8 +865,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_msg_exchange(header);
reply(adapter, header, &error);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error",
NEU_ERR_GROUP_NOT_ALLOW);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
NEU_ERR_GROUP_NOT_ALLOW);
}
} else {
int w_error = neu_adapter_driver_write_tags(
@@ -870,13 +876,21 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
re_flag = true;
} else {
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", w_error);
if (neu_otel_control_is_started() && trace) {
if (w_error != NEU_ERR_SUCCESS) {
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_ERROR, w_error);
} else {
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_OK, w_error);
}
}
}
}
}
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
if (!re_flag) {
neu_otel_trace_set_final(trace);
}
@@ -903,7 +917,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
neu_otel_scope_add_span_attr_int(scope, "thread id",
(int64_t) pthread_self());
neu_otel_scope_set_span_start_time(scope, neu_time_ms());
neu_otel_scope_set_span_start_time(scope, neu_time_ns());
}
}
@@ -917,8 +931,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(adapter, header, &error);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error",
NEU_ERR_GROUP_NOT_ALLOW);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
NEU_ERR_GROUP_NOT_ALLOW);
}
} else {
int w_error = neu_adapter_driver_write_gtags(
@@ -928,13 +942,21 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
re_flag = true;
} else {
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", w_error);
if (neu_otel_control_is_started() && trace) {
if (w_error != NEU_ERR_SUCCESS) {
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_ERROR, w_error);
} else {
neu_otel_scope_set_status_code2(
scope, NEU_OTEL_STATUS_OK, w_error);
}
}
}
}
}
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
if (!re_flag) {
neu_otel_trace_set_final(trace);
}
src/adapter/driver/driver.c
+17 -16
View File
@@ -180,7 +180,7 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
}
neu_otel_scope_add_span_attr_int(scope, "thread id",
(int64_t) pthread_self());
neu_otel_scope_set_span_start_time(scope, neu_time_ms());
neu_otel_scope_set_span_start_time(scope, neu_time_ns());
}
}
@@ -208,8 +208,14 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
adapter->cb_funs.response(adapter, req, &nerror);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error", error);
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
if (error == NEU_ERR_SUCCESS) {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, error);
} else {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
error);
}
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
}
}
@@ -1987,18 +1993,13 @@ static int report_callback(void *usr_data)
if (neu_otel_data_is_started() && data->trace_ctx) {
trans_trace = neu_otel_find_trace(data->trace_ctx);
if (trans_trace) {
trans_scope = neu_otel_add_span(trans_trace);
neu_otel_scope_set_span_name(trans_scope, "report cb");
char new_span_id[36] = { 0 };
neu_otel_new_span_id(new_span_id);
neu_otel_scope_set_span_id(trans_scope, new_span_id);
uint8_t *p_sp_id = neu_otel_scope_get_pre_span_id(trans_scope);
if (p_sp_id) {
neu_otel_scope_set_parent_span_id2(trans_scope, p_sp_id, 8);
}
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_ms());
neu_otel_scope_set_span_start_time(trans_scope, neu_time_ns());
}
}
}
@@ -2038,7 +2039,7 @@ static int report_callback(void *usr_data)
}
if (trans_trace) {
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ns());
neu_otel_trace_set_expected_span_num(trans_trace, app_num);
if (app_num == 0) {
neu_otel_trace_set_final(trans_trace);
@@ -2060,7 +2061,7 @@ static int report_callback(void *usr_data)
if (trans_trace) {
neu_otel_scope_add_span_attr_int(trans_scope, "no sub app", 1);
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ns());
neu_otel_trace_set_final(trans_trace);
}
}
@@ -2072,7 +2073,7 @@ static int report_callback(void *usr_data)
free(data->driver);
if (trans_trace) {
neu_otel_scope_add_span_attr_int(trans_scope, "no tags", 1);
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
neu_otel_scope_set_span_end_time(trans_scope, neu_time_ns());
neu_otel_trace_set_final(trans_trace);
}
}
@@ -2207,13 +2208,13 @@ static int write_callback(void *usr_data)
}
}
s_time = neu_time_ms();
s_time = neu_time_ns();
if (wtag->single) {
group->driver->adapter.module->intf_funs->driver.write_tag(
group->driver->adapter.plugin, (void *) wtag->req, wtag->tag,
wtag->value);
e_time = neu_time_ms();
e_time = neu_time_ns();
neu_tag_free(wtag->tag);
} else {
group->driver->adapter.module->intf_funs->driver.write_tags(
src/base/neu_plugin_common.c
+6 -6
View File
@@ -82,7 +82,7 @@ int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data)
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_ms());
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());
@@ -128,7 +128,7 @@ int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data)
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_ms());
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());
@@ -167,7 +167,7 @@ int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data)
}
neu_otel_scope_set_span_flags(scope, flags);
neu_otel_scope_set_span_start_time(scope,
neu_time_ms());
neu_time_ns());
neu_otel_scope_add_span_attr_int(
scope, "thread id", (int64_t) pthread_self());
@@ -185,11 +185,11 @@ int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data)
head, data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
if (ret != 0) {
neu_otel_scope_add_span_attr_int(scope, "error type",
NEU_ERR_IS_BUSY);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
NEU_ERR_IS_BUSY);
neu_otel_trace_set_final(trace);
}
}
src/core/manager.c
+4 -4
View File
@@ -1077,7 +1077,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
neu_otel_scope_add_span_attr_int(scope, "thread id",
(int64_t) pthread_self());
neu_otel_scope_set_span_start_time(scope, neu_time_ms());
neu_otel_scope_set_span_start_time(scope, neu_time_ns());
}
}
@@ -1102,8 +1102,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_msg_exchange(header);
reply(manager, header, &e);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_add_span_attr_int(scope, "error",
NEU_ERR_NODE_NOT_EXIST);
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_ERROR,
NEU_ERR_NODE_NOT_EXIST);
}
} else {
forward_msg(manager, header, header->receiver);
@@ -1111,7 +1111,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
neu_otel_scope_set_span_end_time(scope, neu_time_ns());
if (!re_flag) {
neu_otel_trace_set_final(trace);
}
src/otel/otel_manager.c
+116 -5
View File
@@ -96,6 +96,9 @@ static int hex_char_to_int(char c)
static int hex_string_to_binary(const char * hex_string,
unsigned char *binary_array, int max_length)
{
if (hex_string == NULL) {
return -1;
}
int length = strlen(hex_string);
if (length % 2 != 0 || length <= 0)
return -1;
@@ -166,7 +169,12 @@ void neu_otel_free_span(Opentelemetry__Proto__Trace__V1__Span *span)
free(span->attributes[i]->value);
free(span->attributes[i]);
}
if (span->status) {
if (span->status->message) {
free(span->status->message);
}
free(span->status);
}
free(span->attributes);
free(span->name);
free(span->trace_id.data);
@@ -601,6 +609,84 @@ neu_otel_scope_ctx neu_otel_add_span(neu_otel_trace_ctx ctx)
return scope;
}
neu_otel_scope_ctx neu_otel_add_span2(neu_otel_trace_ctx ctx,
const char * span_name,
const char * span_id)
{
trace_scope_t *scope = calloc(1, sizeof(trace_scope_t));
trace_ctx_t * trace_ctx = (trace_ctx_t *) ctx;
pthread_mutex_lock(&trace_ctx->mutex);
scope->trace_ctx = ctx;
if (trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->n_spans == 0) {
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->spans =
calloc(1, sizeof(Opentelemetry__Proto__Trace__V1__Span *));
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->spans[0] =
calloc(1, sizeof(Opentelemetry__Proto__Trace__V1__Span));
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->n_spans = 1;
scope->span =
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->spans[0];
scope->span_index = 0;
} else {
Opentelemetry__Proto__Trace__V1__Span **t_spans =
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->spans;
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->spans =
calloc(1 +
trace_ctx->trace_data.resource_spans[0]
->scope_spans[0]
->n_spans,
sizeof(Opentelemetry__Proto__Trace__V1__Span *));
for (size_t i = 0; i <
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->n_spans;
i++) {
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->spans[i] =
t_spans[i];
}
free(t_spans);
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->spans
[trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->n_spans] =
calloc(1, sizeof(Opentelemetry__Proto__Trace__V1__Span));
scope->span = trace_ctx->trace_data.resource_spans[0]
->scope_spans[0]
->spans[trace_ctx->trace_data.resource_spans[0]
->scope_spans[0]
->n_spans];
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->n_spans += 1;
scope->span_index =
trace_ctx->trace_data.resource_spans[0]->scope_spans[0]->n_spans -
1;
}
opentelemetry__proto__trace__v1__span__init(scope->span);
scope->span->kind =
OPENTELEMETRY__PROTO__TRACE__V1__SPAN__SPAN_KIND__SPAN_KIND_SERVER;
uint8_t *t_id = calloc(1, 16);
hex_string_to_binary((char *) trace_ctx->trace_id, t_id, 16);
scope->span->trace_id.data = t_id;
scope->span->trace_id.len = 16;
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) {
neu_otel_scope_set_parent_span_id2(
scope,
trace_ctx->trace_data.resource_spans[0]
->scope_spans[0]
->spans[scope->span_index - 1]
->span_id.data,
trace_ctx->trace_data.resource_spans[0]
->scope_spans[0]
->spans[scope->span_index - 1]
->span_id.len);
}
utarray_push_back(trace_ctx->scopes, &scope);
pthread_mutex_unlock(&trace_ctx->mutex);
return scope;
}
void neu_otel_scope_set_parent_span_id(neu_otel_scope_ctx ctx,
const char * parent_span_id)
{
@@ -771,21 +857,46 @@ void neu_otel_scope_add_span_attr_bool(neu_otel_scope_ctx ctx, const char *key,
free(t_kvs);
}
void neu_otel_scope_set_span_start_time(neu_otel_scope_ctx ctx, int64_t ms)
void neu_otel_scope_set_span_start_time(neu_otel_scope_ctx ctx, int64_t ns)
{
trace_scope_t *scope = (trace_scope_t *) ctx;
scope->span->start_time_unix_nano = ((uint64_t) ms) * 1000000;
scope->span->start_time_unix_nano = ((uint64_t) ns);
}
void neu_otel_scope_set_span_end_time(neu_otel_scope_ctx ctx, int64_t ms)
void neu_otel_scope_set_span_end_time(neu_otel_scope_ctx ctx, int64_t ns)
{
trace_scope_t *scope = (trace_scope_t *) ctx;
scope->span->end_time_unix_nano = ((uint64_t) ms) * 1000000;
scope->span->end_time_unix_nano = ((uint64_t) ns);
trace_ctx_t *trace_ctx = (trace_ctx_t *) scope->trace_ctx;
trace_ctx->span_num += 1;
trace_ctx->expected_span_num -= 1;
}
void neu_otel_scope_set_status_code(neu_otel_scope_ctx ctx,
neu_otel_status_code_e code,
const char * desc)
{
trace_scope_t *scope = (trace_scope_t *) ctx;
scope->span->status =
calloc(1, sizeof(Opentelemetry__Proto__Trace__V1__Status));
opentelemetry__proto__trace__v1__status__init(scope->span->status);
scope->span->status->code = code;
scope->span->status->message = strdup(desc);
}
void neu_otel_scope_set_status_code2(neu_otel_scope_ctx ctx,
neu_otel_status_code_e code, int errorno)
{
trace_scope_t *scope = (trace_scope_t *) ctx;
scope->span->status =
calloc(1, sizeof(Opentelemetry__Proto__Trace__V1__Status));
opentelemetry__proto__trace__v1__status__init(scope->span->status);
char *error_buf = calloc(1, 16);
sprintf(error_buf, "%d", errorno);
scope->span->status->code = code;
scope->span->status->message = error_buf;
}
uint8_t *neu_otel_scope_get_pre_span_id(neu_otel_scope_ctx ctx)
{
trace_scope_t *scope = (trace_scope_t *) ctx;