From 4a608906ad7eac40b880cf70b5dcdc9970bb3023 Mon Sep 17 00:00:00 2001 From: hxy7yx <1595670487@qq.com> Date: Mon, 21 Oct 2024 14:21:22 +0800 Subject: [PATCH] feat(modbus):support opentelemetry --- plugins/modbus/modbus_req.c | 14 +++++-- plugins/modbus/modbus_req.h | 2 +- plugins/modbus/modbus_stack.c | 75 +++++++++++++++++++++++++++++++---- plugins/modbus/modbus_stack.h | 2 +- tests/ft/app/test_ekuiper.py | 1 + 5 files changed, 80 insertions(+), 14 deletions(-) diff --git a/plugins/modbus/modbus_req.c b/plugins/modbus/modbus_req.c index 0169d8d4..c5a6e923 100644 --- a/plugins/modbus/modbus_req.c +++ b/plugins/modbus/modbus_req.c @@ -147,7 +147,7 @@ void handle_modbus_error(neu_plugin_t *plugin, struct modbus_group_data *gd, const char *error_message) { modbus_value_handle(plugin, gd->cmd_sort->cmd[cmd_index].slave_id, 0, NULL, - error_code); + error_code, NULL); if (error_message) { plog_error(plugin, "%s, skip, %hhu!%hu", error_message, gd->cmd_sort->cmd[cmd_index].slave_id, @@ -372,7 +372,7 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group, } int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte, - uint8_t *bytes, int error) + uint8_t *bytes, int error, void *trace) { neu_plugin_t * plugin = (neu_plugin_t *) ctx; struct modbus_group_data *gd = @@ -498,8 +498,14 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte, } } - plugin->common.adapter_callbacks->driver.update( - plugin->common.adapter, gd->group, (*p_tag)->name, dvalue); + if (trace) { + plugin->common.adapter_callbacks->driver.update_with_trace( + plugin->common.adapter, gd->group, (*p_tag)->name, dvalue, NULL, + 0, trace); + } else { + plugin->common.adapter_callbacks->driver.update( + plugin->common.adapter, gd->group, (*p_tag)->name, dvalue); + } } return 0; } diff --git a/plugins/modbus/modbus_req.h b/plugins/modbus/modbus_req.h index 1407084d..7f386aeb 100644 --- a/plugins/modbus/modbus_req.h +++ b/plugins/modbus/modbus_req.h @@ -62,7 +62,7 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group, uint16_t max_byte); int modbus_send_msg(void *ctx, uint16_t n_byte, uint8_t *bytes); int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte, - uint8_t *bytes, int error); + uint8_t *bytes, int error, void *trace); int modbus_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag, neu_value_u value, bool response); int modbus_write_tag(neu_plugin_t *plugin, void *req, neu_datatag_t *tag, diff --git a/plugins/modbus/modbus_stack.c b/plugins/modbus/modbus_stack.c index babcc6e6..67f96375 100644 --- a/plugins/modbus/modbus_stack.c +++ b/plugins/modbus/modbus_stack.c @@ -23,6 +23,8 @@ #include "modbus_req.h" #include "modbus_stack.h" +#include "otel/otel_manager.h" + struct modbus_stack { void * ctx; modbus_stack_send send_fn; @@ -35,6 +37,8 @@ struct modbus_stack { uint8_t *buf; uint16_t buf_size; + + int64_t sample_mod; }; modbus_stack_t *modbus_stack_create(void *ctx, modbus_protocol_e protocol, @@ -65,9 +69,10 @@ void modbus_stack_destroy(modbus_stack_t *stack) int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id, neu_protocol_unpack_buf_t *buf) { - struct modbus_header header = { 0 }; - struct modbus_code code = { 0 }; - int ret = 0; + struct modbus_header header = { 0 }; + struct modbus_code code = { 0 }; + int ret = 0; + int64_t ts_start = neu_time_ns(); if (stack->protocol == MODBUS_PROTOCOL_TCP) { ret = modbus_header_unwrap(buf, &header); @@ -99,6 +104,22 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id, case MODBUS_READ_INPUT: case MODBUS_READ_HOLD_REG: case MODBUS_READ_INPUT_REG: { + neu_otel_trace_ctx trace = NULL; + neu_otel_scope_ctx scope = NULL; + if (neu_otel_data_is_started()) { + trace = neu_otel_find_trace((void *) (intptr_t) stack->read_seq); + if (trace) { + char new_span_id[36] = { 0 }; + neu_otel_new_span_id(new_span_id); + scope = + neu_otel_add_span2(trace, "driver cmd recv", new_span_id); + neu_otel_scope_add_span_attr_int(scope, "thread id", + (int64_t)(pthread_self())); + neu_otel_scope_set_span_start_time(scope, ts_start); + neu_otel_scope_set_span_end_time(scope, neu_time_ns()); + } + } + struct modbus_data data = { 0 }; uint8_t * bytes = NULL; ret = modbus_data_unwrap(buf, &data); @@ -119,14 +140,14 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id, stack->value_fn(stack->ctx, code.slave_id, header.len - sizeof(struct modbus_code) - sizeof(struct modbus_data), - bytes, 0); + bytes, 0, (void *) (intptr_t) stack->read_seq); } else { bytes = neu_protocol_unpack_buf(buf, data.n_byte); if (bytes == NULL) { return -1; } stack->value_fn(stack->ctx, code.slave_id, data.n_byte, bytes, - 0); + 0, (void *) (intptr_t) stack->read_seq); } break; case MODBUS_PROTOCOL_RTU: @@ -134,7 +155,8 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id, if (bytes == NULL) { return -1; } - stack->value_fn(stack->ctx, code.slave_id, data.n_byte, bytes, 0); + stack->value_fn(stack->ctx, code.slave_id, data.n_byte, bytes, 0, + (void *) (intptr_t) stack->read_seq); break; } @@ -287,6 +309,7 @@ int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id, int ret = 0; *response_size = 0; modbus_action_e m_action = MODBUS_ACTION_DEFAULT; + int64_t ts_start = neu_time_ns(); neu_protocol_pack_buf_init(&pbuf, buf, sizeof(buf)); @@ -331,11 +354,47 @@ int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id, ret = stack->send_fn(stack->ctx, neu_protocol_pack_buf_used_size(&pbuf), neu_protocol_pack_buf_get(&pbuf)); if (ret <= 0 && !is_test) { - stack->value_fn(stack->ctx, 0, 0, NULL, NEU_ERR_PLUGIN_DISCONNECTED); + stack->value_fn(stack->ctx, 0, 0, NULL, NEU_ERR_PLUGIN_DISCONNECTED, + NULL); plog_warn((neu_plugin_t *) stack->ctx, "send read req fail, %hhu!%hu", slave_id, start_address); - } + } else { + if (neu_otel_data_is_started()) { + double rate = neu_otel_data_sample_rate(); + int sample_rate = 0; + if (rate > 0.0 && rate <= 1.0) { + sample_rate = (int) (1.0 / rate); + } + if (sample_rate != 0) { + stack->sample_mod += 1; + if (stack->sample_mod % sample_rate == 0) { + char new_trace_id[64] = { 0 }; + neu_otel_new_trace_id(new_trace_id); + const char *trace_state = "span.mytype=data-collection"; + neu_otel_trace_ctx trace = NULL; + neu_otel_scope_ctx scope = NULL; + 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); + 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()); + } + } + } + } return ret; } diff --git a/plugins/modbus/modbus_stack.h b/plugins/modbus/modbus_stack.h index 9d48e81d..83d70460 100644 --- a/plugins/modbus/modbus_stack.h +++ b/plugins/modbus/modbus_stack.h @@ -30,7 +30,7 @@ typedef struct modbus_stack modbus_stack_t; typedef int (*modbus_stack_send)(void *ctx, uint16_t n_byte, uint8_t *bytes); typedef int (*modbus_stack_value)(void *ctx, uint8_t slave_id, uint16_t n_byte, - uint8_t *bytes, int error); + uint8_t *bytes, int error, void *trace); typedef int (*modbus_stack_write_resp)(void *ctx, void *req, int error); typedef enum modbus_protocol { diff --git a/tests/ft/app/test_ekuiper.py b/tests/ft/app/test_ekuiper.py index 98679b9b..81dca973 100644 --- a/tests/ft/app/test_ekuiper.py +++ b/tests/ft/app/test_ekuiper.py @@ -159,6 +159,7 @@ class Mocker: def recv(self): msg = self.sock.recv() + msg = msg[26:] return json.loads(msg.decode()) def send(self, data):