Compare commits

...
48 Commits
Author SHA1 Message Date
fengzeroz 8a6a8b6cdb Merge pull request #2258 from fengzeroz/v2.10
version: 2.10.0
2024-10-17 11:18:15 +08:00
fengzero 138079c2bc version: 2.10.0 2024-10-17 03:17:14 +00:00
fengzeroz 21f02e50eb Merge pull request #2255 from hxy7yx/v2.10-1
[v2.10]feat(modbus):support device degradation
2024-10-15 10:23:59 +08:00
fengzeroz 56f018497d Merge pull request #2256 from ZhouBox/v2.10
[v2.10]otel:support ns and add span status
2024-10-15 10:10:27 +08:00
xiang.zhou b37d8ab2dc otel:support ns and add span status 2024-10-15 09:10:23 +08:00
hxy7yx 9db92459d0 feat(modbus):support start address base 0 or 1 2024-10-14 17:13:25 +08:00
hxy7yx cfff223907 feat(modbus):supoort configuring endianness in node configuration 2024-10-14 14:07:40 +08:00
hxy7yx 9a63c81c02 feat(modbus):support device degradation 2024-10-14 10:57:41 +08:00
fengzeroz 7698962cc3 Merge pull request #2254 from fengzeroz/v2.10
version: 2.10.0-beta5
2024-10-12 14:51:07 +08:00
fengzero 1e0cf54999 version: 2.10.0-beta5 2024-10-12 06:50:29 +00:00
fengzeroz b24d3506d4 Merge pull request #2253 from ZhouBox/v2.10
[v2.10]otel:add api and use new api
2024-10-12 13:45:39 +08:00
xiang.zhou 32b7424333 otel:add api and use new api 2024-10-12 12:59:13 +08:00
fengzeroz a02390be0c Merge pull request #2249 from ZhouBox/v2.10
[v2.10]plugin:rm so and schema file when remove custom plugin &&  mqtt trace:add resp playload
2024-09-29 15:41:59 +08:00
xiang.zhou 18684d7e93 mqtt trace:add resp playload 2024-09-29 15:34:03 +08:00
fengzeroz 309610a685 Merge pull request #2248 from fengzeroz/v2.10
version: 2.10.0-beta4
2024-09-29 15:14:40 +08:00
xiang.zhou 6554e901b5 plugin:rm so and schema file when remove custom plugin 2024-09-29 15:14:23 +08:00
fengzero 0603eff1f2 version: 2.10.0-beta4 2024-09-29 07:11:39 +00:00
fengzeroz 54ecb33fcd Merge pull request #2246 from ZhouBox/v2.10
[v2.10]plugin:modify schema file path when update plugins
2024-09-29 14:00:48 +08:00
xiang.zhou 963acef468 plugin:modify schema file path when update plugins 2024-09-29 13:30:37 +08:00
fengzeroz 9ac6028408 Merge pull request #2244 from fengzeroz/v2.10
version: 2.10.0-beta3
2024-09-26 14:24:48 +08:00
fengzero 9859b2805f version: 2.10.0-beta3 2024-09-26 06:24:08 +00:00
fengzeroz 3442d4a814 Merge pull request #2243 from hxy7yx/v2.10-1
[v2.10]update mqtt schema
2024-09-26 14:14:24 +08:00
hxy7yx 2acdfcc7ef update mqtt schema 2024-09-26 13:34:56 +08:00
fengzeroz ac9d0011af Merge pull request #2241 from ZhouBox/v2.10
[v2.10]otel:mqtt trace and update uuid imple
2024-09-25 14:57:40 +08:00
xiang.zhou af75cf2dcc otel:mqtt trace and update uuid imple 2024-09-25 13:20:58 +08:00
fengzeroz af6bbf86ed Merge pull request #2240 from fengzeroz/v2.10
version: 2.10.0-beta2
2024-09-24 16:32:39 +08:00
fengzero 661ff5a323 version: 2.10.0-beta2 2024-09-24 08:32:05 +00:00
fengzeroz 5b220cae7d Merge pull request #2239 from ZhouBox/v2.10
[v2.10]otel:rm log
2024-09-24 15:31:09 +08:00
xiang.zhou 7a6dd51747 otel:rm log 2024-09-24 15:22:32 +08:00
fengzeroz 2fcdb66f6f Merge pull request #2237 from ZhouBox/v2.10
[v2.10]otel:del cache
2024-09-23 16:59:58 +08:00
xiang.zhou c11c51938c otel:del cache 2024-09-23 16:51:08 +08:00
fengzeroz e09099eedb Merge pull request #2236 from ZhouBox/v2.10
[v2.10]rm debug log
2024-09-23 16:33:05 +08:00
xiang.zhou 5cf1211437 rm debug log 2024-09-23 16:25:16 +08:00
fengzeroz 2d873cacc7 Merge pull request #2235 from fengzeroz/v2.10
version: 2.10.0-beta1
2024-09-23 15:29:53 +08:00
fengzero d8aa593b01 version: 2.10.0-beta1 2024-09-23 07:29:01 +00:00
fengzeroz 052f5c5f98 Merge pull request #2234 from ZhouBox/v2.10
[v2.10]feat(otel):add sample support
2024-09-23 15:26:10 +08:00
xiang.zhou 127c1a19c1 feat(otel):add sample support 2024-09-23 15:17:39 +08:00
fengzeroz c6ba1b114a Merge pull request #2231 from fengzeroz/v2.10
version: 2.10.0-alpha3
2024-09-20 11:22:42 +08:00
fengzero f06fd84ba0 version: 2.10.0-alpha3 2024-09-20 03:20:36 +00:00
fengzeroz 4a5d32604f Merge pull request #2230 from fengzeroz/v2.10
rm file plugin
2024-09-20 10:52:09 +08:00
fengzero 0e89e61d8c rm file plugin 2024-09-20 02:31:56 +00:00
fengzero d3957d7e32 fix str cpy 2024-09-20 02:01:18 +00:00
fengzeroz 2f2a1eb941 Merge pull request #2227 from fengzeroz/v2.10
meta size 8
2024-09-19 14:39:38 +08:00
fengzero abc9df7fc0 meta size 8 2024-09-19 06:38:29 +00:00
fengzeroz 1f67ee6a8e Merge pull request #2226 from fengzeroz/v2.10
version: 2.10.0-alpha2
2024-09-19 10:22:05 +08:00
fengzero bceb610b6a version: 2.10.0-alpha2 2024-09-19 02:21:26 +00:00
fengzeroz 664cd09024 Merge pull request #2221 from hxy7yx/v2.10-1
[2.10]fix(read paginate):handle only_write tags
2024-09-19 10:16:13 +08:00
hxy7yx f9f86e1f22 fix(read paginate):handle only_write tags 2024-09-18 15:59:50 +08:00
44 changed files with 1528 additions and 551 deletions
-1
View File
@@ -175,7 +175,6 @@ file(COPY ${CMAKE_SOURCE_DIR}/default_plugins.json DESTINATION ${CMAKE_BINARY_DI
add_subdirectory(plugins/modbus)
add_subdirectory(plugins/mqtt)
add_subdirectory(plugins/ekuiper)
add_subdirectory(plugins/file)
add_subdirectory(simulator)
+1 -2
View File
@@ -5,7 +5,6 @@
"libplugin-azure-iot.so",
"libplugin-ekuiper.so",
"libplugin-modbus-tcp.so",
"libplugin-modbus-rtu.so",
"libplugin-file.so"
"libplugin-modbus-rtu.so"
]
}
+4
View File
@@ -57,6 +57,10 @@ typedef struct adapter_callbacks {
struct {
void (*update)(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value);
void (*update_with_trace)(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value,
neu_tag_meta_t *metas, int n_meta,
void *trace_ctx);
void (*update_with_meta)(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value,
neu_tag_meta_t *metas, int n_meta);
+7
View File
@@ -134,6 +134,13 @@ int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
char *topic, uint8_t *payload, uint32_t len,
void *data, neu_mqtt_client_publish_cb_t cb);
int neu_mqtt_client_publish_with_trace(neu_mqtt_client_t *client,
neu_mqtt_qos_e qos, char *topic,
uint8_t *payload, uint32_t len,
void * data,
neu_mqtt_client_publish_cb_t cb,
const char *traceparent);
/** Subscribe to `topic` with service quality `qos`.
*
* This function tries to send a `SUBSCRIBE` packet with the given `qos` and
+1 -1
View File
@@ -53,7 +53,7 @@
#define NEU_MSG_MAX_SIZE 2048
#define NEU_TAG_META_LENGTH 20
#define NEU_TAG_META_SIZE 3
#define NEU_TAG_META_SIZE 8
#define NEU_LOG_LEVEL_DEBUG "debug"
#define NEU_LOG_LEVEL_INFO "info"
+3
View File
@@ -675,6 +675,8 @@ typedef struct neu_resp_tag_value {
neu_dvalue_t value;
} neu_resp_tag_value_t;
typedef neu_resp_tag_value_t neu_tag_value_t;
typedef struct neu_resp_tag_value_meta {
char tag[NEU_TAG_NAME_LEN];
neu_dvalue_t value;
@@ -799,6 +801,7 @@ typedef struct {
typedef struct {
char *driver;
char *group;
void *trace_ctx;
neu_reqresp_trans_data_ctx_t *ctx;
UT_array * tags; // neu_resp_tag_value_meta_t
+22 -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;
@@ -34,11 +40,17 @@ neu_otel_trace_ctx neu_otel_find_trace(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);
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);
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);
@@ -63,16 +75,24 @@ 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);
void neu_otel_new_span_id(char *id);
void neu_otel_new_trace_id(char *id);
void neu_otel_split_traceparent(const char *in, char *trace_id, char *span_id,
uint32_t *flags);
+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 = {
+34 -5
View File
@@ -314,8 +314,8 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin,
NEU_NODE_LINK_STATE_DISCONNECTED == plugin->common.link_state;
nng_mtx_unlock(plugin->mtx);
neu_otel_trace_ctx *trace = NULL;
neu_otel_scope_ctx scope = NULL;
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
if (neu_otel_control_is_started()) {
trace = neu_otel_find_trace(header->ctx);
if (trace) {
@@ -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) {
@@ -361,6 +367,29 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin,
if (disconnected) {
plog_debug(plugin, "not connected");
neu_otel_trace_ctx trans_trace = NULL;
neu_otel_scope_ctx trans_scope = NULL;
if (neu_otel_data_is_started() && trans_data->trace_ctx) {
trans_trace = neu_otel_find_trace(trans_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, "ekuiper send", new_span_id);
neu_otel_scope_add_span_attr_int(trans_scope, "thread id",
(int64_t)(pthread_self()));
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_ns());
neu_otel_scope_set_span_end_time(trans_scope,
neu_time_ns());
neu_otel_trace_set_final(trans_trace);
}
}
} else {
send_data(plugin, trans_data);
}
+112 -29
View File
@@ -20,6 +20,7 @@
#include <nng/nng.h>
#include "neuron.h"
#include "otel/otel_manager.h"
#include "json/neu_json_fn.h"
#include "json/neu_json_rw.h"
@@ -34,6 +35,40 @@ static int send_write_tags_req(neu_plugin_t * plugin,
bool trace_flag, uint8_t *trace_id,
uint8_t *span_id);
static int hex_char_to_int(char c)
{
if (c >= '0' && c <= '9')
return c - '0';
if (c >= 'a' && c <= 'f')
return c - 'a' + 10;
if (c >= 'A' && c <= 'F')
return c - 'A' + 10;
return -1;
}
static int hex_string_to_binary(const char * hex_string,
unsigned char *binary_array, int max_length)
{
int length = strlen(hex_string);
if (length % 2 != 0 || length <= 0)
return -1;
int binary_length = length / 2;
if (binary_length > max_length)
return -1;
for (int i = 0; i < binary_length; i++) {
int high_nibble = hex_char_to_int(hex_string[2 * i]);
int low_nibble = hex_char_to_int(hex_string[2 * i + 1]);
if (high_nibble == -1 || low_nibble == -1)
return -1;
binary_array[i] = (high_nibble << 4) | low_nibble;
}
return binary_length;
}
static void byte_to_hex(unsigned char byte, char *hex)
{
static const char hex_chars[] = "0123456789ABCDEF";
@@ -57,39 +92,87 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
.plugin = plugin,
.trans_data = trans_data,
};
rv = neu_json_encode_by_fn(&resp, json_encode_read_resp, &json_str);
if (0 != rv) {
plog_error(plugin, "fail encode trans data to json");
return;
neu_otel_trace_ctx trans_trace = NULL;
neu_otel_scope_ctx trans_scope = NULL;
uint8_t * trace_id = NULL;
char new_span_id[36] = { 0 };
if (neu_otel_data_is_started() && trans_data->trace_ctx) {
trans_trace = neu_otel_find_trace(trans_data->trace_ctx);
if (trans_trace) {
neu_otel_new_span_id(new_span_id);
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_ns());
}
}
nng_msg *msg = NULL;
size_t json_len = strlen(json_str);
rv = nng_msg_alloc(&msg, json_len);
if (0 != rv) {
plog_error(plugin, "nng cannot allocate msg");
do {
rv = neu_json_encode_by_fn(&resp, json_encode_read_resp, &json_str);
if (0 != rv) {
plog_error(plugin, "fail encode trans data to json");
break;
}
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) {
trace_header_len = 26;
}
rv = nng_msg_alloc(&msg, json_len + trace_header_len);
if (0 != rv) {
plog_error(plugin, "nng cannot allocate msg");
free(json_str);
break;
}
if (trans_trace) {
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);
uint16_t tarce_header_magic = 0xCE0A;
memcpy(nng_msg_body(msg), &tarce_header_magic, 2);
memcpy(nng_msg_body(msg) + 2, trace_id, 16);
memcpy(nng_msg_body(msg) + 2 + 16, span_id, 8);
}
memcpy(nng_msg_body(msg) + trace_header_len, json_str,
json_len); // no null byte
plog_debug(plugin, ">> %s", json_str);
free(json_str);
return;
}
rv = nng_sendmsg(plugin->sock, msg,
NNG_FLAG_NONBLOCK); // TODO: use aio to send message
if (0 == rv) {
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1,
NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, json_len,
NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S,
json_len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S,
json_len, NULL);
} else {
plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv));
nng_msg_free(msg);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL,
1, NULL);
}
} while (0);
memcpy(nng_msg_body(msg), json_str, json_len); // no null byte
plog_debug(plugin, ">> %s", json_str);
free(json_str);
rv = nng_sendmsg(plugin->sock, msg,
NNG_FLAG_NONBLOCK); // TODO: use aio to send message
if (0 == rv) {
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, json_len,
NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, json_len,
NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, json_len,
NULL);
} else {
plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv));
nng_msg_free(msg);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
NULL);
if (trans_trace) {
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);
}
}
-11
View File
@@ -1,11 +0,0 @@
set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins")
set(CMAKE_BUILD_RPATH ./)
file(COPY ${CMAKE_SOURCE_DIR}/plugins/file/file.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
set(PLUGIN_NAME plugin-file)
set(PLUGIN_SOURCES file_plugin.c)
add_library(${PLUGIN_NAME} SHARED)
target_include_directories(${PLUGIN_NAME} PRIVATE ${CMAKE_SOURCE_DIR}/include/neuron)
target_sources(${PLUGIN_NAME} PRIVATE ${PLUGIN_SOURCES})
target_link_libraries(${PLUGIN_NAME} neuron-base)
-9
View File
@@ -1,9 +0,0 @@
{
"tag_regex": [
{
"type": 13,
"regex": ""
}
],
"group_interval": 1000
}
-252
View File
@@ -1,252 +0,0 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include <errno.h>
#include <stdlib.h>
#include <neuron.h>
#include "errcodes.h"
struct neu_plugin {
neu_plugin_common_t common;
// neu_events_t *events;
// int file_length;
};
static neu_plugin_t *driver_open(void);
static int driver_close(neu_plugin_t *plugin);
static int driver_init(neu_plugin_t *plugin, bool load);
static int driver_uninit(neu_plugin_t *plugin);
static int driver_start(neu_plugin_t *plugin);
static int driver_stop(neu_plugin_t *plugin);
static int driver_config(neu_plugin_t *plugin, const char *config);
static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
void *data);
static int driver_tag_validator(const neu_datatag_t *tag);
static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag);
static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
neu_value_u value);
static const neu_plugin_intf_funs_t plugin_intf_funs = {
.open = driver_open,
.close = driver_close,
.init = driver_init,
.uninit = driver_uninit,
.start = driver_start,
.stop = driver_stop,
.setting = driver_config,
.request = driver_request,
.driver.validate_tag = driver_validate_tag,
.driver.group_timer = driver_group_timer,
.driver.group_sync = NULL,
.driver.write_tag = driver_write,
.driver.tag_validator = driver_tag_validator,
.driver.write_tags = NULL,
.driver.test_read_tag = NULL,
.driver.add_tags = NULL,
.driver.load_tags = NULL,
.driver.del_tags = NULL,
};
const neu_plugin_module_t neu_plugin_module = {
.version = NEURON_PLUGIN_VER_1_0,
.schema = "file",
.module_name = "File",
.module_descr = "The plugin is used to read and write files.",
.module_descr_zh = "该插件用于读写文件。",
.intf_funs = &plugin_intf_funs,
.kind = NEU_PLUGIN_KIND_SYSTEM,
.type = NEU_NA_TYPE_DRIVER,
.display = true,
.single = false,
};
static neu_plugin_t *driver_open(void)
{
neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t));
neu_plugin_common_init(&plugin->common);
return plugin;
}
static int driver_close(neu_plugin_t *plugin)
{
free(plugin);
return 0;
}
static int driver_init(neu_plugin_t *plugin, bool load)
{
(void) load;
plog_notice(plugin, "%s init success", plugin->common.name);
return 0;
}
static int driver_uninit(neu_plugin_t *plugin)
{
plog_notice(plugin, "%s uninit success", plugin->common.name);
return 0;
}
static int driver_start(neu_plugin_t *plugin)
{
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, 1, NULL);
plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
plog_notice(plugin, "start `%s` plugin success", plugin->common.name);
return 0;
}
static int driver_stop(neu_plugin_t *plugin)
{
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, 9999, NULL);
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
plog_notice(plugin, "stop `%s` plugin success", plugin->common.name);
return 0;
}
static int driver_config(neu_plugin_t *plugin, const char *config)
{
(void) plugin;
(void) config;
return 0;
}
static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
void *data)
{
(void) plugin;
(void) head;
(void) data;
return 0;
}
static int driver_tag_validator(const neu_datatag_t *tag)
{
(void) tag;
return 0;
}
static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag)
{
if (tag->type != NEU_TYPE_STRING) {
return NEU_ERR_TAG_TYPE_NOT_SUPPORT;
}
plog_notice(plugin, "validate tag success, name:%s, address:%s, type:%d ",
tag->name, tag->address, tag->type);
return 0;
}
static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
{
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
utarray_foreach(group->tags, neu_datatag_t *, tag)
{
neu_dvalue_t dvalue = { 0 };
FILE *fp = fopen(tag->address, "r");
if (fp == NULL) {
dvalue.type = NEU_TYPE_ERROR;
if (errno == ENOENT) {
dvalue.value.i32 = NEU_ERR_FILE_NOT_EXIST;
} else {
dvalue.value.i32 = NEU_ERR_FILE_OPEN_FAILURE;
}
} else {
char *buf = calloc(1, 1024);
int ret = fread(buf, 1024, 1, fp);
if (ret < 0) {
dvalue.type = NEU_TYPE_ERROR;
dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
} else if (ret < NEU_VALUE_SIZE) {
dvalue.type = NEU_TYPE_STRING;
strcpy(dvalue.value.str, buf);
} else {
dvalue.type = NEU_TYPE_PTR;
dvalue.value.ptr.type = NEU_TYPE_STRING;
dvalue.value.ptr.length = ret;
dvalue.value.ptr.ptr = (uint8_t *) strdup(buf);
}
fclose(fp);
free(buf);
}
plugin->common.adapter_callbacks->driver.update(
plugin->common.adapter, group->group_name, tag->name, dvalue);
}
update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, 0, NULL);
update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, 0, NULL);
update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, 1, NULL);
update_metric(plugin->common.adapter, NEU_METRIC_GROUP_LAST_SEND_MSGS, 1,
group->group_name);
return 0;
}
static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
neu_value_u value)
{
FILE *fp = fopen(tag->address, "w");
if (fp == NULL) {
if (errno == ENOENT) {
plugin->common.adapter_callbacks->driver.write_response(
plugin->common.adapter, req, NEU_ERR_FILE_NOT_EXIST);
} else {
plugin->common.adapter_callbacks->driver.write_response(
plugin->common.adapter, req, NEU_ERR_FILE_OPEN_FAILURE);
}
} else {
if (tag->type == NEU_TYPE_STRING) {
size_t ret = fwrite(value.str, 1, strlen(value.str), fp);
if (ret != strlen(value.str)) {
plugin->common.adapter_callbacks->driver.write_response(
plugin->common.adapter, req, NEU_ERR_FILE_WRITE_FAILURE);
}
} else {
plugin->common.adapter_callbacks->driver.write_response(
plugin->common.adapter, req, NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH);
}
fclose(fp);
plugin->common.adapter_callbacks->driver.write_response(
plugin->common.adapter, req, 0);
}
return 0;
}
+105
View File
@@ -78,6 +78,61 @@
"max": 30000
}
},
"device_degrade": {
"name": "Device Degradation",
"name_zh": "设备降级",
"description": "Enable or disable device degradation mechanism",
"description_zh": "启用或禁用设备降级机制",
"attribute": "required",
"type": "map",
"default": 0,
"valid": {
"map": [
{
"key": "False",
"value": 0
},
{
"key": "True",
"value": 1
}
]
}
},
"degrade_cycle": {
"name": "Failure Threshold for Degradation",
"name_zh": "降级失败阈值",
"description": "The number of consecutive failure cycles required to trigger device degradation",
"description_zh": "触发设备降级所需的连续失败周期数",
"attribute": "required",
"type": "int",
"default": 2,
"valid": {
"min": 1,
"max": 65535
},
"condition": {
"field": "device_degrade",
"value": 1
}
},
"degrade_time": {
"name": "Recovery Time After Degradation",
"name_zh": "降级恢复时间",
"description": "The time in seconds after which the device recovers from degradation",
"description_zh": "设备从降级中恢复所需的时间(单位:秒)",
"attribute": "required",
"type": "int",
"default": 600,
"valid": {
"min": 1,
"max": 65535
},
"condition": {
"field": "device_degrade",
"value": 1
}
},
"max_retries": {
"name": "Maximum Retry Times",
"name_zh": "最大重试次数",
@@ -102,6 +157,56 @@
"max": 10000
}
},
"endianess": {
"name": "Endianess",
"name_zh": "字节序",
"description": "Tag byte order, ABCD corresponds to 1234",
"description_zh": "点位字节序,ABCD 对应 1234",
"attribute": "required",
"type": "map",
"default": 1,
"valid": {
"map": [
{
"key": "ABCD",
"value": 1
},
{
"key": "BADC",
"value": 2
},
{
"key": "DCBA",
"value": 3
},
{
"key": "CDAB",
"value": 4
}
]
}
},
"address_base": {
"name": "Start Address",
"name_zh": "开始地址",
"description": "Address starts from 1 or 0",
"description_zh": "地址从 1 开始或从 0 开始",
"attribute": "required",
"type": "map",
"default": 1,
"valid": {
"map": [
{
"key": "Protocol Addresses (Base 0)",
"value": 0
},
{
"key": "PLC Addresses (Base 1)",
"value": 1
}
]
}
},
"interval": {
"name": "Send Interval (ms)",
"name_zh": "指令发送间隔 (ms)",
+105
View File
@@ -86,6 +86,61 @@
]
}
},
"device_degrade": {
"name": "Device Degradation",
"name_zh": "设备降级",
"description": "Enable or disable device degradation mechanism",
"description_zh": "启用或禁用设备降级机制",
"attribute": "required",
"type": "map",
"default": 0,
"valid": {
"map": [
{
"key": "False",
"value": 0
},
{
"key": "True",
"value": 1
}
]
}
},
"degrade_cycle": {
"name": "Failure Threshold for Degradation",
"name_zh": "降级失败阈值",
"description": "The number of consecutive failure cycles required to trigger device degradation",
"description_zh": "触发设备降级所需的连续失败周期数",
"attribute": "required",
"type": "int",
"default": 2,
"valid": {
"min": 1,
"max": 65535
},
"condition": {
"field": "device_degrade",
"value": 1
}
},
"degrade_time": {
"name": "Recovery Time After Degradation",
"name_zh": "降级恢复时间",
"description": "The time in seconds after which the device recovers from degradation",
"description_zh": "设备从降级中恢复所需的时间(单位:秒)",
"attribute": "required",
"type": "int",
"default": 600,
"valid": {
"min": 1,
"max": 65535
},
"condition": {
"field": "device_degrade",
"value": 1
}
},
"max_retries": {
"name": "Maximum Retry Times",
"name_zh": "最大重试次数",
@@ -110,6 +165,56 @@
"max": 10000
}
},
"endianess": {
"name": "Endianess",
"name_zh": "字节序",
"description": "Tag byte order, ABCD corresponds to 1234",
"description_zh": "点位字节序,ABCD 对应 1234",
"attribute": "required",
"type": "map",
"default": 1,
"valid": {
"map": [
{
"key": "ABCD",
"value": 1
},
{
"key": "BADC",
"value": 2
},
{
"key": "DCBA",
"value": 3
},
{
"key": "CDAB",
"value": 4
}
]
}
},
"address_base": {
"name": "Start Address",
"name_zh": "开始地址",
"description": "Address starts from 1 or 0",
"description_zh": "地址从 1 开始或从 0 开始",
"attribute": "required",
"type": "map",
"default": 1,
"valid": {
"map": [
{
"key": "Protocol Addresses (Base 0)",
"value": 0
},
{
"key": "PLC Addresses (Base 1)",
"value": 1
}
]
}
},
"interval": {
"name": "Send Interval (ms)",
"name_zh": "指令发送间隔 (ms)",
+12
View File
@@ -66,6 +66,18 @@ typedef enum modbus_area {
MODBUS_AREA_HOLD_REGISTER = 4,
} modbus_area_e;
typedef enum modbus_endianess {
MODBUS_ABCD = 1,
MODBUS_BADC = 2,
MODBUS_DCBA = 3,
MODBUS_CDAB = 4,
} modbus_endianess;
typedef enum modbus_address_base {
base_0 = 0,
base_1 = 1,
} modbus_address_base;
struct modbus_header {
uint16_t seq;
uint16_t protocol;
+49 -12
View File
@@ -35,7 +35,8 @@ static int tag_cmp_write(neu_tag_sort_elem_t *tag1, neu_tag_sort_elem_t *tag2);
static bool tag_sort_write(neu_tag_sort_t *sort, void *tag,
void *tag_to_be_sorted);
int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point)
int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point,
modbus_address_base address_base)
{
int ret = NEU_ERR_SUCCESS;
uint32_t start_address = 0;
@@ -47,13 +48,18 @@ int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point)
char area = 0;
int n = sscanf(tag->address, "%hhu!%c%u", &point->slave_id, &area,
&start_address);
if (n != 3 || start_address == 0) {
if (n != 3) {
return NEU_ERR_TAG_ADDRESS_FORMAT_INVALID;
}
point->start_address = (uint16_t) start_address;
if (start_address == 65536 && address_base == 0) {
point->start_address = 65535;
} else if (start_address == 0 && address_base == 1) {
point->start_address = 0;
} else {
point->start_address = (uint16_t) start_address - address_base;
}
point->start_address -= 1;
point->type = tag->type;
switch (area) {
@@ -194,10 +200,11 @@ int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point)
}
int modbus_write_tag_to_point(const neu_plugin_tag_value_t *tag,
modbus_point_write_t * point)
modbus_point_write_t * point,
modbus_address_base address_base)
{
int ret = NEU_ERR_SUCCESS;
ret = modbus_tag_to_point(tag->tag, &point->point);
ret = modbus_tag_to_point(tag->tag, &point->point, address_base);
point->value = tag->value;
return ret;
}
@@ -230,7 +237,8 @@ modbus_read_cmd_sort_t *modbus_tag_sort(UT_array *tags, uint16_t max_byte)
return sort_result;
}
int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option)
int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option,
modbus_endianess endianess, bool default_tag_endian)
{
int n = 0;
switch (type) {
@@ -242,6 +250,9 @@ int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option)
case NEU_TYPE_FLOAT:
case NEU_TYPE_UINT32:
case NEU_TYPE_INT32:
if (default_tag_endian) {
modbus_convert_endianess(value, endianess);
}
n = sizeof(uint32_t);
value->u32 = htonl(value->u32);
break;
@@ -282,7 +293,8 @@ int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option)
return n;
}
modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *tags)
modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array * tags,
modbus_endianess endianess)
{
neu_tag_sort_result_t *result =
neu_tag_sort(tags, tag_sort_write, tag_cmp_write);
@@ -304,14 +316,17 @@ modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *tags)
utarray_foreach(result->sorts[i].tags, modbus_point_write_t **, tag_s)
{
if ((*tag_s)->point.area == MODBUS_AREA_COIL) {
n_byte_tag = cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
(*tag_s)->point.option);
n_byte_tag =
cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
(*tag_s)->point.option, endianess, true);
data_bit[k / 8] += ((*tag_s)->value.i8) << k % 8;
n_byte += n_byte_tag;
k++;
} else {
n_byte_tag = cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
(*tag_s)->point.option);
n_byte_tag =
cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
(*tag_s)->point.option, endianess,
(*tag_s)->point.option.value32.is_default);
memcpy(sort_result->cmd[i].bytes +
2 *
((*tag_s)->point.start_address -
@@ -521,4 +536,26 @@ static bool tag_sort_write(neu_tag_sort_t *sort, void *tag,
}
return true;
}
void modbus_convert_endianess(neu_value_u *value, modbus_endianess endianess)
{
switch (endianess) {
case MODBUS_ABCD:
break;
case MODBUS_CDAB:
value->u32 = htonl(value->u32);
neu_ntohs_p((uint16_t *) value->bytes.bytes);
neu_ntohs_p((uint16_t *) (value->bytes.bytes + 2));
break;
case MODBUS_BADC:
neu_ntohs_p((uint16_t *) value->bytes.bytes);
neu_ntohs_p((uint16_t *) (value->bytes.bytes + 2));
break;
case MODBUS_DCBA:
value->u32 = htonl(value->u32);
break;
default:
break;
}
}
+8 -3
View File
@@ -45,9 +45,11 @@ typedef struct modbus_point_write {
neu_value_u value;
} modbus_point_write_t;
int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point);
int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point,
modbus_address_base address_base);
int modbus_write_tag_to_point(const neu_plugin_tag_value_t *tag,
modbus_point_write_t * point);
modbus_point_write_t * point,
modbus_address_base address_base);
typedef struct modbus_read_cmd {
uint8_t slave_id;
@@ -80,9 +82,12 @@ typedef struct modbus_write_cmd_sort {
} modbus_write_cmd_sort_t;
modbus_read_cmd_sort_t * modbus_tag_sort(UT_array *tags, uint16_t max_byte);
modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *tags);
modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array * tags,
modbus_endianess endianess);
void modbus_tag_sort_free(modbus_read_cmd_sort_t *cs);
void modbus_convert_endianess(neu_value_u *value, modbus_endianess endianess);
#ifdef __cplusplus
}
#endif
+98 -17
View File
@@ -23,10 +23,16 @@
#include "modbus_req.h"
#define MAX_SLAVES 256
uint8_t failed_cycles[MAX_SLAVES];
bool skip[MAX_SLAVES];
struct modbus_group_data {
UT_array * tags;
char * group;
modbus_read_cmd_sort_t *cmd_sort;
modbus_address_base address_base;
};
struct modbus_write_tags_data {
@@ -152,7 +158,8 @@ void handle_modbus_error(neu_plugin_t *plugin, struct modbus_group_data *gd,
void finalize_modbus_read_result(neu_plugin_t * plugin,
struct modbus_group_data *gd,
uint16_t cmd_index, int ret_r, int ret_buf,
uint64_t read_tms, int64_t *rtt)
uint64_t read_tms, int64_t *rtt,
bool *slave_err)
{
if (ret_r <= 0) {
handle_modbus_error(plugin, gd, cmd_index, NEU_ERR_PLUGIN_DISCONNECTED,
@@ -166,6 +173,7 @@ void finalize_modbus_read_result(neu_plugin_t * plugin,
NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE,
"no modbus response received");
*rtt = neu_time_ms() - read_tms;
slave_err[gd->cmd_sort->cmd[cmd_index].slave_id] = true;
break;
case -1:
handle_modbus_error(plugin, gd, cmd_index,
@@ -185,12 +193,13 @@ void finalize_modbus_read_result(neu_plugin_t * plugin,
}
} else {
*rtt = neu_time_ms() - read_tms;
failed_cycles[gd->cmd_sort->cmd[cmd_index].slave_id] = 0;
}
}
void check_modbus_read_result(neu_plugin_t * plugin,
struct modbus_group_data *gd, uint16_t cmd_index,
int64_t *rtt)
int64_t *rtt, bool *slave_err)
{
uint16_t response_size = 0;
uint64_t read_tms = neu_time_ms();
@@ -224,7 +233,7 @@ void check_modbus_read_result(neu_plugin_t * plugin,
}
finalize_modbus_read_result(plugin, gd, cmd_index, ret_r, ret_buf, read_tms,
rtt);
rtt, slave_err);
}
void update_metrics_after_read(neu_plugin_t *plugin, int64_t rtt,
@@ -246,14 +255,53 @@ void update_metrics_after_read(neu_plugin_t *plugin, int64_t rtt,
gd->cmd_sort->n_cmd, group->group_name);
}
typedef struct {
uint8_t slave_id;
uint16_t degrade_time;
} degrade_timer_data_t;
void *degrade_timer(void *arg)
{
degrade_timer_data_t *data = (degrade_timer_data_t *) arg;
struct timespec t1 = { .tv_sec = data->degrade_time, .tv_nsec = 0 };
struct timespec t2 = { 0 };
nanosleep(&t1, &t2);
skip[data->slave_id] = false;
free(data);
return NULL;
}
void set_skip_timer(uint8_t slave_id, uint32_t degrade_time)
{
degrade_timer_data_t *data =
(degrade_timer_data_t *) malloc(sizeof(degrade_timer_data_t));
data->slave_id = slave_id;
data->degrade_time = degrade_time;
failed_cycles[slave_id] = 0;
pthread_t timer_thread;
pthread_create(&timer_thread, NULL, degrade_timer, data);
pthread_detach(timer_thread);
}
int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
uint16_t max_byte)
{
neu_conn_state_t state = { 0 };
struct modbus_group_data *gd = NULL;
int64_t rtt = NEU_METRIC_LAST_RTT_MS_MAX;
struct modbus_group_data *gdt =
(struct modbus_group_data *) group->user_data;
if (group->user_data == NULL) {
if (group->user_data == NULL || gdt->address_base != plugin->address_base) {
if (group->user_data != NULL) {
plugin_group_free(group);
}
gd = calloc(1, sizeof(struct modbus_group_data));
group->user_data = gd;
@@ -262,8 +310,8 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
utarray_foreach(group->tags, neu_datatag_t *, tag)
{
modbus_point_t *p = calloc(1, sizeof(modbus_point_t));
int ret = modbus_tag_to_point(tag, p);
modbus_point_t *p = calloc(1, sizeof(modbus_point_t));
int ret = modbus_tag_to_point(tag, p, plugin->address_base);
if (ret != NEU_ERR_SUCCESS) {
plog_error(plugin, "invalid tag: %s, address: %s", tag->name,
tag->address);
@@ -272,16 +320,43 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
utarray_push_back(gd->tags, &p);
}
gd->group = strdup(group->group_name);
gd->cmd_sort = modbus_tag_sort(gd->tags, max_byte);
gd->group = strdup(group->group_name);
gd->cmd_sort = modbus_tag_sort(gd->tags, max_byte);
gd->address_base = plugin->address_base;
}
gd = (struct modbus_group_data *) group->user_data;
plugin->plugin_group_data = gd;
bool slave_err_record[MAX_SLAVES] = { false };
for (uint16_t i = 0; i < gd->cmd_sort->n_cmd; i++) {
plugin->cmd_idx = i;
check_modbus_read_result(plugin, gd, i, &rtt);
bool slave_err[MAX_SLAVES] = { false };
uint8_t slave_id = gd->cmd_sort->cmd[i].slave_id;
plugin->cmd_idx = i;
if (slave_err_record[slave_id] == true) {
continue;
}
if (plugin->degradation == false || skip[slave_id] == false) {
check_modbus_read_result(plugin, gd, i, &rtt, slave_err);
} else {
continue;
}
if (plugin->degradation) {
if (slave_err[slave_id]) {
failed_cycles[slave_id]++;
slave_err_record[slave_id] = true;
}
if (failed_cycles[slave_id] >= plugin->degrade_cycle) {
skip[slave_id] = true;
plog_warn(plugin, "Skip slave %hhu", slave_id);
set_skip_timer(slave_id, plugin->degrade_time);
}
}
if (plugin->interval > 0) {
struct timespec t1 = { .tv_sec = plugin->interval / 1000,
@@ -370,6 +445,9 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
case NEU_TYPE_FLOAT:
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32:
if ((*p_tag)->option.value32.is_default) {
modbus_convert_endianess(&dvalue.value, plugin->endianess);
}
dvalue.value.u32 = ntohl(dvalue.value.u32);
break;
case NEU_TYPE_DOUBLE:
@@ -575,7 +653,7 @@ int modbus_test_read_tag(neu_plugin_t *plugin, void *req, neu_datatag_t tag)
neu_json_value_u error_value;
error_value.val_int = 0;
int err = modbus_tag_to_point(&tag, &point);
int err = modbus_tag_to_point(&tag, &point, plugin->address_base);
if (err != NEU_ERR_SUCCESS) {
plugin->common.adapter_callbacks->driver.test_read_tag_response(
plugin->common.adapter, req, NEU_JSON_INT, NEU_TYPE_ERROR,
@@ -615,8 +693,8 @@ int modbus_test_read_tag(neu_plugin_t *plugin, void *req, neu_datatag_t tag)
return 0;
}
static uint8_t convert_value(neu_value_u *value, neu_datatag_t *tag,
modbus_point_t *point)
static uint8_t convert_value(neu_plugin_t *plugin, neu_value_u *value,
neu_datatag_t *tag, modbus_point_t *point)
{
uint8_t n_byte = 0;
switch (tag->type) {
@@ -628,6 +706,9 @@ static uint8_t convert_value(neu_value_u *value, neu_datatag_t *tag,
case NEU_TYPE_FLOAT:
case NEU_TYPE_UINT32:
case NEU_TYPE_INT32:
if (point->option.value32.is_default) {
modbus_convert_endianess(value, plugin->endianess);
}
value->u32 = htonl(value->u32);
n_byte = sizeof(uint32_t);
break;
@@ -702,10 +783,10 @@ int modbus_write_tag(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
neu_value_u value)
{
modbus_point_t point = { 0 };
int ret = modbus_tag_to_point(tag, &point);
int ret = modbus_tag_to_point(tag, &point, plugin->address_base);
assert(ret == 0);
uint8_t n_byte = convert_value(&value, tag, &point);
uint8_t n_byte = convert_value(plugin, &value, tag, &point);
return write_modbus_point(plugin, req, &point, value, n_byte);
}
@@ -721,12 +802,12 @@ int modbus_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags)
utarray_foreach(tags, neu_plugin_tag_value_t *, tag)
{
modbus_point_write_t *p = calloc(1, sizeof(modbus_point_write_t));
ret = modbus_write_tag_to_point(tag, p);
ret = modbus_write_tag_to_point(tag, p, plugin->address_base);
assert(ret == 0);
utarray_push_back(gtags->tags, &p);
}
gtags->cmd_sort = modbus_write_tags_sort(gtags->tags);
gtags->cmd_sort = modbus_write_tags_sort(gtags->tags, plugin->endianess);
for (uint16_t i = 0; i < gtags->cmd_sort->n_cmd; i++) {
ret = write_modbus_points(plugin, &gtags->cmd_sort->cmd[i], req);
if (ret <= 0) {
+6 -1
View File
@@ -38,12 +38,17 @@ struct neu_plugin {
int client_fd;
neu_events_t * events;
modbus_protocol_e protocol;
modbus_protocol_e protocol;
modbus_endianess endianess;
modbus_address_base address_base;
uint16_t interval;
uint16_t retry_interval;
uint16_t max_retries;
uint16_t check_header;
bool degradation;
uint16_t degrade_cycle;
uint16_t degrade_time;
};
void modbus_conn_connected(void *data, int fd);
+39 -9
View File
@@ -37,7 +37,6 @@ static int driver_config(neu_plugin_t *plugin, const char *config);
static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
void *data);
static int driver_tag_validator(const neu_datatag_t *tag);
static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag);
static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
@@ -58,7 +57,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
.driver.group_timer = driver_group_timer,
.driver.group_sync = driver_group_timer,
.driver.write_tag = driver_write,
.driver.tag_validator = driver_tag_validator,
.driver.tag_validator = NULL,
.driver.write_tags = driver_write_tags,
.driver.add_tags = NULL,
.driver.load_tags = NULL,
@@ -168,6 +167,17 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
neu_json_elem_t retry_interval = { .name = "retry_interval",
.t = NEU_JSON_INT };
neu_json_elem_t degradation = { .name = "device_degrade",
.t = NEU_JSON_INT };
neu_json_elem_t degrade_cycle = { .name = "degrade_cycle",
.t = NEU_JSON_INT };
neu_json_elem_t degrade_time = { .name = "degrade_time",
.t = NEU_JSON_INT };
neu_json_elem_t endianess = { .name = "endianess", .t = NEU_JSON_INT };
neu_json_elem_t address_base = { .name = "address_base",
.t = NEU_JSON_INT };
ret = neu_parse_param((char *) config, &err_param, 3, &link, &timeout,
&interval);
if (ret != 0) {
@@ -208,9 +218,35 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
retry_interval.v.val_int = 0;
}
ret = neu_parse_param((char *) config, &err_param, 3, &degradation,
&degrade_cycle, &degrade_time);
if (ret != 0) {
free(err_param);
degradation.v.val_int = 0;
degrade_cycle.v.val_int = 2;
degrade_time.v.val_int = 600;
}
ret = neu_parse_param((char *) config, &err_param, 1, &endianess);
if (ret != 0) {
free(err_param);
endianess.v.val_int = MODBUS_ABCD;
}
ret = neu_parse_param((char *) config, &err_param, 1, &address_base);
if (ret != 0) {
free(err_param);
address_base.v.val_int = base_1;
}
param.log = plugin->common.log;
plugin->max_retries = max_retries.v.val_int;
plugin->retry_interval = retry_interval.v.val_int;
plugin->degradation = degradation.v.val_int;
plugin->degrade_cycle = degrade_cycle.v.val_int;
plugin->degrade_time = degrade_time.v.val_int;
plugin->endianess = endianess.v.val_int;
plugin->address_base = address_base.v.val_int;
if (link.v.val_int == 0) {
param.type = NEU_CONN_TTY_CLIENT;
@@ -279,17 +315,11 @@ static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
return 0;
}
static int driver_tag_validator(const neu_datatag_t *tag)
{
modbus_point_t point = { 0 };
return modbus_tag_to_point(tag, &point);
}
static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag)
{
modbus_point_t point = { 0 };
int ret = modbus_tag_to_point(tag, &point);
int ret = modbus_tag_to_point(tag, &point, plugin->address_base);
if (ret == 0) {
plog_notice(plugin,
"validate tag success, name: %s, address: %s, type: %d, "
+39 -9
View File
@@ -37,7 +37,6 @@ static int driver_config(neu_plugin_t *plugin, const char *config);
static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
void *data);
static int driver_tag_validator(const neu_datatag_t *tag);
static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag);
static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
@@ -60,7 +59,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
.driver.group_timer = driver_group_timer,
.driver.group_sync = driver_group_timer,
.driver.write_tag = driver_write,
.driver.tag_validator = driver_tag_validator,
.driver.tag_validator = NULL,
.driver.write_tags = driver_write_tags,
.driver.test_read_tag = driver_test_read_tag,
.driver.add_tags = NULL,
@@ -165,6 +164,17 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
neu_json_elem_t check_header = { .name = "check_header",
.t = NEU_JSON_INT };
neu_json_elem_t degradation = { .name = "device_degrade",
.t = NEU_JSON_INT };
neu_json_elem_t degrade_cycle = { .name = "degrade_cycle",
.t = NEU_JSON_INT };
neu_json_elem_t degrade_time = { .name = "degrade_time",
.t = NEU_JSON_INT };
neu_json_elem_t endianess = { .name = "endianess", .t = NEU_JSON_INT };
neu_json_elem_t address_base = { .name = "address_base",
.t = NEU_JSON_INT };
ret = neu_parse_param((char *) config, &err_param, 5, &port, &host, &mode,
&timeout, &interval);
@@ -198,11 +208,37 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
check_header.v.val_int = 0;
}
ret = neu_parse_param((char *) config, &err_param, 3, &degradation,
&degrade_cycle, &degrade_time);
if (ret != 0) {
free(err_param);
degradation.v.val_int = 0;
degrade_cycle.v.val_int = 2;
degrade_time.v.val_int = 600;
}
ret = neu_parse_param((char *) config, &err_param, 1, &endianess);
if (ret != 0) {
free(err_param);
endianess.v.val_int = MODBUS_ABCD;
}
ret = neu_parse_param((char *) config, &err_param, 1, &address_base);
if (ret != 0) {
free(err_param);
address_base.v.val_int = base_1;
}
param.log = plugin->common.log;
plugin->interval = interval.v.val_int;
plugin->max_retries = max_retries.v.val_int;
plugin->retry_interval = retry_interval.v.val_int;
plugin->check_header = check_header.v.val_int;
plugin->degradation = degradation.v.val_int;
plugin->degrade_cycle = degrade_cycle.v.val_int;
plugin->degrade_time = degrade_time.v.val_int;
plugin->endianess = endianess.v.val_int;
plugin->address_base = address_base.v.val_int;
if (mode.v.val_int == 1) {
param.type = NEU_CONN_TCP_SERVER;
@@ -248,17 +284,11 @@ static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
return 0;
}
static int driver_tag_validator(const neu_datatag_t *tag)
{
modbus_point_t point = { 0 };
return modbus_tag_to_point(tag, &point);
}
static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag)
{
modbus_point_t point = { 0 };
int ret = modbus_tag_to_point(tag, &point);
int ret = modbus_tag_to_point(tag, &point, plugin->address_base);
if (ret == 0) {
plog_notice(
plugin,
+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);
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);
+6 -4
View File
@@ -99,15 +99,17 @@
}
},
"upload_drv_state": {
"name": "Upload Driver State",
"name": "Driver Status Report",
"name_zh": "驱动状态上报",
"description": "Reports status of all the southbound nodes to the specified topic.",
"description_zh": "上报所有南向驱动状态到指定的 MQTT 主题。",
"attribute": "optional",
"type": "bool",
"default": false,
"valid": {}
},
"upload_drv_state_topic": {
"name": "Status Reporting Topic",
"name": "Status Report Topic",
"name_zh": "状态上报主题",
"attribute": "required",
"type": "string",
@@ -121,8 +123,8 @@
}
},
"upload_drv_state_interval": {
"name": "Status Reporting Interval(S)",
"name_zh": "状态上报间隔(S)",
"name": "Status Report Interval (Second)",
"name_zh": "状态上报间隔(秒)",
"attribute": "required",
"type": "int",
"condition": {
+119 -26
View File
@@ -19,6 +19,7 @@
#include "connection/mqtt_client.h"
#include "errcodes.h"
#include "otel/otel_manager.h"
#include "utils/asprintf.h"
#include "version.h"
#include "json/neu_json_mqtt.h"
@@ -27,6 +28,17 @@
#include "mqtt_handle.h"
#include "mqtt_plugin.h"
static void to_traceparent(uint8_t *trace_id, char *span_id, char *out)
{
int size = 0;
size = sprintf(out, "00-");
for (size_t i = 0; i < 16; i++) {
size += sprintf(out + size, "%02x", trace_id[i]);
}
sprintf(out + size, "-%s-01", span_id);
}
static int tag_values_to_json(UT_array *tags, neu_json_read_resp_t *json)
{
int index = 0;
@@ -364,6 +376,24 @@ int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
return rv;
}
int publish_with_trace(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
char *payload, size_t payload_len,
const char *traceparent)
{
int rv = neu_mqtt_client_publish_with_trace(
plugin->client, qos, topic, (uint8_t *) payload, (uint32_t) payload_len,
plugin, publish_cb, traceparent);
if (0 != rv) {
plog_error(plugin, "pub [%s, QoS%d] fail", topic, qos);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
NULL);
free(payload);
rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
}
return rv;
}
void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
const uint8_t *payload, uint32_t len, void *data,
trace_w3c_t *trace_w3c)
@@ -430,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)
neu_resp_error_t *data, void *trace_scope,
void *trace_ctx, char *span_id)
{
int rv = 0;
char *json_str = NULL;
@@ -455,9 +486,25 @@ int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
goto end;
}
if (trace_scope) {
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:
@@ -566,35 +613,81 @@ int handle_trans_data(neu_plugin_t * plugin,
{
int rv = 0;
if (NULL == plugin->client) {
return NEU_ERR_MQTT_IS_NULL;
neu_otel_trace_ctx trans_trace = NULL;
neu_otel_scope_ctx trans_scope = NULL;
uint8_t * trace_id = NULL;
char trace_parent[128] = { 0 };
if (neu_otel_data_is_started() && trans_data->trace_ctx) {
trans_trace = neu_otel_find_trace(trans_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, "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_ns());
trace_id = neu_otel_get_trace_id(trans_trace);
to_traceparent(trace_id, new_span_id, trace_parent);
}
}
if (0 == plugin->config.cache &&
!neu_mqtt_client_is_connected(plugin->client)) {
// cache disable and we are disconnected
return NEU_ERR_MQTT_FAILURE;
}
do {
const route_entry_t *route = route_tbl_get(
&plugin->route_tbl, trans_data->driver, trans_data->group);
if (NULL == route) {
plog_error(plugin, "no route for driver:%s group:%s",
trans_data->driver, trans_data->group);
return NEU_ERR_GROUP_NOT_SUBSCRIBE;
}
if (NULL == plugin->client) {
rv = NEU_ERR_MQTT_IS_NULL;
break;
}
char *json_str =
generate_upload_json(plugin, trans_data, plugin->config.format);
if (NULL == json_str) {
plog_error(plugin, "generate upload json fail");
return NEU_ERR_EINTERNAL;
}
if (0 == plugin->config.cache &&
!neu_mqtt_client_is_connected(plugin->client)) {
// cache disable and we are disconnected
rv = NEU_ERR_MQTT_FAILURE;
break;
}
char * topic = route->topic;
neu_mqtt_qos_e qos = plugin->config.qos;
rv = publish(plugin, qos, topic, json_str, strlen(json_str));
json_str = NULL;
const route_entry_t *route = route_tbl_get(
&plugin->route_tbl, trans_data->driver, trans_data->group);
if (NULL == route) {
plog_error(plugin, "no route for driver:%s group:%s",
trans_data->driver, trans_data->group);
rv = NEU_ERR_GROUP_NOT_SUBSCRIBE;
break;
}
char *json_str =
generate_upload_json(plugin, trans_data, plugin->config.format);
if (NULL == json_str) {
plog_error(plugin, "generate upload json fail");
rv = NEU_ERR_EINTERNAL;
break;
}
char * topic = route->topic;
neu_mqtt_qos_e qos = plugin->config.qos;
if (plugin->config.version == NEU_MQTT_VERSION_V5 && trans_trace) {
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;
} while (0);
if (trans_trace) {
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);
}
return rv;
}
+6 -1
View File
@@ -31,12 +31,17 @@ extern "C" {
int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
char *payload, size_t payload_len);
int publish_with_trace(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
char *payload, size_t payload_len,
const char *traceparent);
void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
const uint8_t *payload, uint32_t len, void *data,
trace_w3c_t *trace_w3c);
int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_resp_error_t *data);
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,
+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);
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, "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);
}
+26 -1
View File
@@ -316,7 +316,32 @@ void handle_get_plugin_schema(nng_aio *aio)
}
char *buf = NULL;
buf = file_string_read(&len, schema_path);
buf = file_string_read(&len, schema_path);
if (NULL == buf) {
free(schema_path);
if (0 > neu_asprintf(&schema_path, "%s/custom/schema/%s.json",
g_plugin_dir, schema_name)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, error_code.error, result_error);
});
return;
}
buf = file_string_read(&len, schema_path);
}
if (NULL == buf) {
free(schema_path);
if (0 > neu_asprintf(&schema_path, "%s/system/schema/%s.json",
g_plugin_dir, schema_name)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, error_code.error, result_error);
});
return;
}
buf = file_string_read(&len, schema_path);
}
if (NULL == buf) {
nlog_info("open %s error: %d", schema_path, errno);
neu_http_not_found(aio, "{\"status\": \"error\"}");
+32 -19
View File
@@ -174,8 +174,8 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
return 0;
}
neu_otel_trace_ctx *trace = NULL;
neu_otel_scope_ctx scope = NULL;
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
if (neu_otel_control_is_started()) {
trace = neu_otel_find_trace(header->ctx);
if (trace) {
@@ -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);
}
+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);
}
}
+1 -1
View File
@@ -91,7 +91,7 @@ ssize_t modbus_s_rtu_req(uint8_t *req, uint16_t req_len, uint8_t *res,
}
if (code->slave_id != 1 && code->slave_id != 2 && code->slave_id != 3) {
return -1;
return -2;
}
switch (code->function) {
+43 -45
View File
@@ -593,26 +593,6 @@ static int adapter_responseto(neu_adapter_t * adapter,
neu_reqresp_head_t *pheader = neu_msg_get_header(msg);
strcpy(pheader->sender, adapter->name);
neu_otel_trace_ctx *trace = NULL;
neu_otel_scope_ctx scope = NULL;
if (neu_otel_control_is_started()) {
trace = neu_otel_find_trace(header->ctx);
if (trace) {
scope = neu_otel_add_span(trace);
neu_otel_scope_set_span_name(scope, "adapter write tag 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);
if (p_sp_id) {
neu_otel_scope_set_parent_span_id2(scope, p_sp_id, 8);
}
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());
}
}
int ret = neu_send_msg_to(adapter->control_fd, &dst, msg);
if (0 != ret) {
nlog_error("adapter: %s send responseto %s failed, ret: %d, errno: %d",
@@ -621,10 +601,6 @@ static int adapter_responseto(neu_adapter_t * adapter,
neu_msg_free(msg);
}
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_span_end_time(scope, neu_time_ms());
}
return ret;
}
@@ -798,8 +774,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_WRITE_TAG: {
neu_resp_error_t error = { 0 };
neu_otel_trace_ctx *trace = NULL;
neu_otel_scope_ctx scope = NULL;
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
if (neu_otel_control_is_started()) {
trace = neu_otel_find_trace(header->ctx);
if (trace) {
@@ -814,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());
}
}
@@ -827,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 {
@@ -837,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);
}
@@ -854,8 +836,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_WRITE_TAGS: {
neu_resp_error_t error = { 0 };
neu_otel_trace_ctx *trace = NULL;
neu_otel_scope_ctx scope = NULL;
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
if (neu_otel_control_is_started()) {
trace = neu_otel_find_trace(header->ctx);
if (trace) {
@@ -870,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());
}
}
@@ -883,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(
@@ -894,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);
}
@@ -911,8 +901,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_WRITE_GTAGS: {
neu_resp_error_t error = { 0 };
neu_otel_trace_ctx *trace = NULL;
neu_otel_scope_ctx scope = NULL;
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
if (neu_otel_control_is_started()) {
trace = neu_otel_find_trace(header->ctx);
if (trace) {
@@ -927,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());
}
}
@@ -941,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(
@@ -952,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);
}
+64 -2
View File
@@ -49,10 +49,16 @@ struct elem {
UT_hash_handle hh;
};
typedef struct {
char key[NEU_GROUP_NAME_LEN];
void * trace_ctx;
UT_hash_handle hh;
} group_trace_t;
struct neu_driver_cache {
pthread_mutex_t mtx;
struct elem *table;
group_trace_t * trace_table;
struct elem * table;
};
// static void update_tag_error(neu_driver_cache_t *cache, const char *group,
@@ -94,6 +100,16 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
}
free(elem);
}
group_trace_t *elem1 = NULL;
group_trace_t *tmp1 = NULL;
HASH_ITER(hh, cache->trace_table, elem1, tmp1)
{
HASH_DEL(cache->trace_table, elem1);
free(elem1);
}
pthread_mutex_unlock(&cache->mtx);
pthread_mutex_destroy(&cache->mtx);
@@ -132,6 +148,52 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
pthread_mutex_unlock(&cache->mtx);
}
void neu_driver_cache_update_trace(neu_driver_cache_t *cache, const char *group,
void *trace_ctx)
{
group_trace_t *elem = NULL;
char key[NEU_GROUP_NAME_LEN] = { 0 };
strcpy(key, group);
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->trace_table, &key, sizeof(key), elem);
if (elem == NULL) {
elem = calloc(1, sizeof(group_trace_t));
strcpy(elem->key, group);
elem->trace_ctx = trace_ctx;
HASH_ADD(hh, cache->trace_table, key, sizeof(key), elem);
}
elem->trace_ctx = trace_ctx;
pthread_mutex_unlock(&cache->mtx);
}
void *neu_driver_cache_get_trace(neu_driver_cache_t *cache, const char *group)
{
group_trace_t *elem = NULL;
char key[NEU_GROUP_NAME_LEN] = { 0 };
void *trace = NULL;
strcpy(key, group);
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->trace_table, &key, sizeof(key), elem);
if (elem != NULL) {
trace = elem->trace_ctx;
}
pthread_mutex_unlock(&cache->mtx);
return trace;
}
void neu_driver_cache_update_change(neu_driver_cache_t *cache,
const char *group, const char *tag,
int64_t timestamp, neu_dvalue_t value,
+5
View File
@@ -44,6 +44,11 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
const char *tag);
void neu_driver_cache_update_trace(neu_driver_cache_t *cache, const char *group,
void *trace_ctx);
void *neu_driver_cache_get_trace(neu_driver_cache_t *cache, const char *group);
typedef struct {
neu_dvalue_t value;
int64_t timestamp;
+92 -10
View File
@@ -104,6 +104,10 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
neu_tag_cache_type_e cache_type,
neu_driver_cache_t *cache, const char *group,
UT_array *tags, UT_array *tag_values);
static void update_with_trace(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value,
neu_tag_meta_t *metas, int n_meta,
void *trace_ctx);
static void update(neu_adapter_t *adapter, const char *group, const char *tag,
neu_dvalue_t value);
static void update_im(neu_adapter_t *adapter, const char *group,
@@ -161,8 +165,8 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
neu_reqresp_head_t *req = (neu_reqresp_head_t *) r;
neu_resp_error_t nerror = { .error = error };
neu_otel_trace_ctx *trace = NULL;
neu_otel_scope_ctx scope = NULL;
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
if (neu_otel_control_is_started()) {
trace = neu_otel_find_trace(req->ctx);
if (trace) {
@@ -176,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());
}
}
@@ -199,13 +203,19 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
req->type = NEU_RESP_ERROR;
nlog_notice("write tag response <%p>", req->ctx);
nlog_notice("write tag response start <%p>", req->ctx);
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());
}
}
@@ -259,6 +269,18 @@ static void update_with_meta(neu_adapter_t *adapter, const char *group,
global_timestamp, n_meta);
}
static void update_with_trace(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value,
neu_tag_meta_t *metas, int n_meta,
void *trace_ctx)
{
update_with_meta(adapter, group, tag, value, metas, n_meta);
if (trace_ctx) {
neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter;
neu_driver_cache_update_trace(driver->cache, group, trace_ctx);
}
}
static void update_im(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value,
neu_tag_meta_t *metas, int n_meta)
@@ -409,6 +431,7 @@ neu_adapter_driver_t *neu_adapter_driver_create()
driver->adapter.cb_funs.driver.update = update;
driver->adapter.cb_funs.driver.write_response = write_response;
driver->adapter.cb_funs.driver.update_im = update_im;
driver->adapter.cb_funs.driver.update_with_trace = update_with_trace;
driver->adapter.cb_funs.driver.update_with_meta = update_with_meta;
driver->adapter.cb_funs.driver.scan_tags_response = scan_tags_response;
driver->adapter.cb_funs.driver.test_read_tag_response =
@@ -1960,6 +1983,27 @@ static int report_callback(void *usr_data)
data->group = strdup(group->name);
utarray_new(data->tags, neu_resp_tag_value_meta_icd());
void *trace_ctx =
neu_driver_cache_get_trace(group->driver->cache, group->name);
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());
}
}
}
read_report_group(global_timestamp,
neu_group_get_interval(group->group) *
NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
@@ -1970,6 +2014,7 @@ static int report_callback(void *usr_data)
pthread_mutex_lock(&group->apps_mtx);
if (utarray_len(group->apps) > 0) {
int app_num = 0;
data->ctx = calloc(1, sizeof(neu_reqresp_trans_data_ctx_t));
data->ctx->index = utarray_len(group->apps);
pthread_mutex_init(&data->ctx->mtx, NULL);
@@ -1980,8 +2025,27 @@ static int report_callback(void *usr_data)
&group->driver->adapter, &header, data, app->addr) !=
0) {
neu_trans_data_free(data);
if (trans_trace) {
neu_otel_scope_add_span_attr_int(trans_scope, app->app,
0);
}
} else {
app_num += 1;
if (trans_trace) {
neu_otel_scope_add_span_attr_int(trans_scope, app->app,
1);
}
}
}
if (trans_trace) {
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);
}
}
} else {
utarray_foreach(data->tags, neu_resp_tag_value_meta_t *, tag_value)
{
@@ -1992,6 +2056,12 @@ static int report_callback(void *usr_data)
utarray_free(data->tags);
free(data->group);
free(data->driver);
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_ns());
neu_otel_trace_set_final(trans_trace);
}
}
pthread_mutex_unlock(&group->apps_mtx);
@@ -1999,6 +2069,11 @@ static int report_callback(void *usr_data)
utarray_free(data->tags);
free(data->group);
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_ns());
neu_otel_trace_set_final(trans_trace);
}
}
utarray_free(tags);
free(data);
@@ -2090,8 +2165,8 @@ static int write_callback(void *usr_data)
int64_t s_time = 0;
int64_t e_time = 0;
neu_otel_trace_ctx *trace = NULL;
neu_otel_scope_ctx scope = NULL;
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
if (neu_otel_control_is_started()) {
trace =
neu_otel_find_trace(((neu_reqresp_head_t *) wtag->req)->ctx);
@@ -2131,13 +2206,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(
@@ -2559,6 +2634,13 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
tag_value.datatag.option = tag->option;
memcpy(tag_value.datatag.meta, tag->meta, NEU_TAG_META_LENGTH);
if (tag->attribute == NEU_ATTRIBUTE_WRITE) {
tag_value.value.type = NEU_TYPE_STRING;
tag_value.value.value.str[0] = 0;
utarray_push_back(tag_values, &tag_value);
continue;
}
if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
tag_value.metas,
NEU_TAG_META_SIZE) != 0) {
+2 -2
View File
@@ -315,8 +315,8 @@ UT_array *neu_group_query_read_tag_paginate(neu_group_t *group,
};
pthread_mutex_lock(&group->mtx);
array = filter_tags_by_page(group->tags, is_readable_and_match_query, &q,
current_page, page_size, total_count);
array = filter_tags_by_page(group->tags, match_query, &q, current_page,
page_size, total_count);
pthread_mutex_unlock(&group->mtx);
return array;
+10 -11
View File
@@ -55,12 +55,11 @@ int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data)
neu_otel_scope_ctx scope = NULL;
if (head.otel_trace_type == NEU_OTEL_TRACE_TYPE_REST_COMM) {
nng_aio * aio = (nng_aio *) head.ctx;
nng_http_req *nng_req = nng_aio_get_input(aio, 0);
const char * http_method = nng_http_req_get_method(nng_req);
if (neu_otel_control_is_started()) {
const char *trace_parent =
nng_aio * aio = (nng_aio *) head.ctx;
nng_http_req *nng_req = nng_aio_get_input(aio, 0);
const char * http_method = nng_http_req_get_method(nng_req);
const char * trace_parent =
nng_http_req_get_header(nng_req, "traceparent");
const char *trace_state =
nng_http_req_get_header(nng_req, "tracestate");
@@ -83,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());
@@ -129,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());
@@ -168,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());
@@ -186,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);
}
}
+58
View File
@@ -1601,6 +1601,64 @@ int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
return 0;
}
int neu_mqtt_client_publish_with_trace(neu_mqtt_client_t *client,
neu_mqtt_qos_e qos, char *topic,
uint8_t *payload, uint32_t len,
void * data,
neu_mqtt_client_publish_cb_t cb,
const char * traceparent)
{
int rv = 0;
nng_msg *pub_msg = NULL;
task_t * task = NULL;
if (0 != (rv = nng_mqtt_msg_alloc(&pub_msg, 0))) {
log(error, "nng_mqtt_msg_alloc fail: %s", nng_strerror(rv));
return -1;
}
if (0 != (rv = nng_mqtt_msg_set_publish_topic(pub_msg, topic))) {
nng_msg_free(pub_msg);
log(error, "nng_mqtt_msg_set_publish_topic fail: %s", nng_strerror(rv));
return -1;
}
nng_mqtt_msg_set_packet_type(pub_msg, NNG_MQTT_PUBLISH);
nng_mqtt_msg_set_publish_payload(pub_msg, (uint8_t *) payload, len);
nng_mqtt_msg_set_publish_qos(pub_msg, qos);
if (client->version == MQTT_PROTOCOL_VERSION_v5) {
property *plist = mqtt_property_alloc();
property *p = mqtt_property_set_value_strpair(
USER_PROPERTY, "traceparent", strlen("traceparent"), traceparent,
strlen(traceparent), true);
mqtt_property_append(plist, p);
nng_mqtt_msg_set_publish_property(pub_msg, plist);
}
nng_mtx_lock(client->mtx);
task = client_alloc_task(client);
nng_mtx_unlock(client->mtx);
if (NULL == task) {
nng_msg_free(pub_msg);
log(error, "client_alloc_task fail");
return -1;
}
task->kind = TASK_PUB;
task->pub.cb = cb;
task->pub.qos = qos;
task->pub.topic = topic;
task->pub.payload = payload;
task->pub.len = len;
task->pub.data = data;
nng_aio_set_msg(task->aio, pub_msg);
nng_send_aio(client->sock, task->aio);
return 0;
}
int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
const char *topic, void *data,
neu_mqtt_client_subscribe_cb_t cb)
+35 -12
View File
@@ -71,7 +71,8 @@ static void start_single_adapter(neu_manager_t *manager, const char *name,
static char *file_save_tmp(const char *data, const char *suffix);
static bool mv_tmp_library_file(neu_plugin_kind_e kind, const char *tmp_path,
const char *library);
static bool mv_tmp_schema_file(const char *tmp_path, const char *schema);
static bool mv_tmp_schema_file(neu_plugin_kind_e kind, const char *tmp_path,
const char *schema);
uint16_t neu_manager_get_port()
{
@@ -400,7 +401,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
if (!mv_tmp_schema_file(schema_tmp_path, schema)) {
if (!mv_tmp_schema_file(kind, schema_tmp_path, schema)) {
nlog_warn("library %s schema file save schema fail", cmd->library);
free(cmd->so_file);
free(cmd->schema_file);
@@ -661,7 +662,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
if (!mv_tmp_schema_file(schema_tmp_path, schema)) {
if (!mv_tmp_schema_file(kind, schema_tmp_path, schema)) {
nlog_warn("library %s schema file save schema fail", cmd->library);
free(cmd->so_file);
free(cmd->schema_file);
@@ -1051,8 +1052,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_GET_TAG:
case NEU_REQ_NODE_CTL:
case NEU_REQ_ADD_GROUP: {
neu_otel_trace_ctx *trace = NULL;
neu_otel_scope_ctx scope = NULL;
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
if (neu_otel_control_is_started() &&
(NEU_REQ_WRITE_TAG == header->type ||
NEU_REQ_WRITE_TAGS == header->type ||
@@ -1076,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());
}
}
@@ -1101,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);
@@ -1110,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);
}
@@ -1574,12 +1575,34 @@ static bool mv_tmp_library_file(neu_plugin_kind_e kind, const char *tmp_path,
return false;
}
static bool mv_tmp_schema_file(const char *tmp_path, const char *schema)
static bool mv_tmp_schema_file(neu_plugin_kind_e kind, const char *tmp_path,
const char *schema)
{
char file_name[128] = { 0 };
snprintf(file_name, sizeof(file_name), "%s/schema/%s.json", g_plugin_dir,
schema);
char path_name[128] = { 0 };
if (kind == NEU_PLUGIN_KIND_CUSTOM) {
snprintf(file_name, sizeof(file_name), "%s/custom/schema/%s.json",
g_plugin_dir, schema);
snprintf(path_name, sizeof(path_name), "%s/custom/schema",
g_plugin_dir);
} else if (kind == NEU_PLUGIN_KIND_SYSTEM) {
snprintf(file_name, sizeof(file_name), "%s/system/schema/%s.json",
g_plugin_dir, schema);
snprintf(path_name, sizeof(path_name), "%s/system/schema",
g_plugin_dir);
} else {
return false;
}
struct stat st;
if (stat(path_name, &st) == -1) {
if (mkdir(path_name, 0700) == -1) {
nlog_error("%s mkdir fail", path_name);
return false;
}
}
if (rename(tmp_path, file_name) != 0) {
nlog_error("%s rename %s fail, err:%s", tmp_path, file_name,
+14
View File
@@ -252,6 +252,20 @@ int neu_plugin_manager_del(neu_plugin_manager_t *mgr, const char *plugin_name)
HASH_FIND_STR(mgr->plugins, plugin_name, plugin);
if (plugin != NULL) {
if (plugin->kind != NEU_PLUGIN_KIND_SYSTEM) {
char so_file_path[128] = { 0 };
char schema_file_path[128] = { 0 };
snprintf(so_file_path, sizeof(so_file_path), "%s/custom/%s",
g_plugin_dir, plugin->lib_name);
snprintf(schema_file_path, sizeof(schema_file_path),
"%s/custom/schema/%s.json", g_plugin_dir, plugin->schema);
if (remove(so_file_path) != 0) {
nlog_error("rm %s file fail!", so_file_path);
}
if (remove(schema_file_path) != 0) {
nlog_error("rm %s file fail!", schema_file_path);
}
HASH_DEL(mgr->plugins, plugin);
entity_free(plugin);
ret = NEU_ERR_SUCCESS;
+165 -20
View File
@@ -28,6 +28,7 @@
#include "plugin.h"
#include "utils/http.h"
#include "utils/log.h"
#include "utils/time.h"
#include "utils/utarray.h"
#include "utils/uthash.h"
@@ -35,7 +36,8 @@
#include "trace.pb-c.h"
#define SPAN_ID_LENGTH 16
#define SPAN_ID_CHARSET "0123456789abcdef"
#define ID_CHARSET "0123456789abcdef"
#define TRACE_TIME_OUT (3 * 60 * 1000)
bool otel_flag = false;
char otel_collector_url[128] = { 0 };
@@ -50,6 +52,8 @@ typedef struct {
uint32_t flags;
bool final;
size_t span_num;
int32_t expected_span_num;
int64_t ts;
UT_array * scopes;
pthread_mutex_t mutex;
} trace_ctx_t;
@@ -92,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;
@@ -162,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);
@@ -177,6 +189,7 @@ neu_otel_trace_ctx neu_otel_create_trace(const char *trace_id, void *req_ctx,
trace_ctx_t *ctx = calloc(1, sizeof(trace_ctx_t));
opentelemetry__proto__trace__v1__traces_data__init(&ctx->trace_data);
strncpy((char *) ctx->trace_id, trace_id, 64);
ctx->expected_span_num = 0;
ctx->trace_data.n_resource_spans = 1;
ctx->trace_data.resource_spans =
@@ -417,6 +430,8 @@ neu_otel_trace_ctx neu_otel_create_trace(const char *trace_id, void *req_ctx,
ctx->flags = flags;
ctx->ts = neu_time_ms();
trace_ctx_table_ele_t *ele = calloc(1, sizeof(trace_ctx_table_ele_t));
ele->key = req_ctx;
ele->ctx = ctx;
@@ -455,6 +470,24 @@ void neu_otel_trace_set_final(neu_otel_trace_ctx ctx)
trace_ctx->final = true;
}
void neu_otel_trace_set_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;
}
uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx)
{
trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;
pthread_mutex_lock(&trace_ctx->mutex);
uint8_t *id = trace_ctx->trace_data.resource_spans[0]
->scope_spans[0]
->spans[0]
->trace_id.data;
pthread_mutex_unlock(&trace_ctx->mutex);
return id;
}
neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id)
{
trace_ctx_table_ele_t *el = NULL, *tmp = NULL;
@@ -576,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)
{
@@ -746,31 +857,61 @@ 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;
trace_ctx_t * trace_ctx = (trace_ctx_t *) scope->trace_ctx;
pthread_mutex_lock(&trace_ctx->mutex);
if (scope->span_index <= 0) {
pthread_mutex_unlock(&trace_ctx->mutex);
return NULL;
}
return trace_ctx->trace_data.resource_spans[0]
->scope_spans[0]
->spans[scope->span_index - 1]
->span_id.data;
uint8_t *id = trace_ctx->trace_data.resource_spans[0]
->scope_spans[0]
->spans[scope->span_index - 1]
->span_id.data;
pthread_mutex_unlock(&trace_ctx->mutex);
return id;
}
int neu_otel_trace_pack_size(neu_otel_trace_ctx ctx)
@@ -789,20 +930,18 @@ int neu_otel_trace_pack(neu_otel_trace_ctx ctx, uint8_t *out)
void neu_otel_new_span_id(char *id)
{
static uint64_t counter = 0;
uint64_t timestamp = time(NULL);
uint64_t random = rand();
uint64_t combined = (timestamp << 24) | (counter << 12) | random;
counter++;
for (int i = SPAN_ID_LENGTH - 1; i >= 0; i--) {
id[i] = SPAN_ID_CHARSET[combined % 16];
combined /= 16;
id[i] = ID_CHARSET[rand() % 16];
}
id[SPAN_ID_LENGTH] = '\0';
}
void neu_otel_new_trace_id(char *id)
{
neu_otel_new_span_id(id);
neu_otel_new_span_id(id + SPAN_ID_LENGTH);
}
void neu_otel_split_traceparent(const char *in, char *trace_id, char *span_id,
uint32_t *flags)
{
@@ -854,7 +993,8 @@ static int otel_timer_cb(void *data)
el->ctx->span_num ==
el->ctx->trace_data.resource_spans[0]
->scope_spans[0]
->n_spans) {
->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);
@@ -868,6 +1008,11 @@ static int otel_timer_cb(void *data)
neu_otel_free_trace(el->ctx);
free(el);
}
} 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);
}
}
+151
View File
@@ -387,6 +387,14 @@ test_input_bit = {"driver": "modbus-tcp", "group": "group", "tag": "tag", "addre
test_input_register_bit = {"driver": "modbus-tcp", "group": "group", "tag": "tag", "address": "1!30010.0",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "precision": 0, "decimal": 0, "bias": 0.0}
hold_int16_slave255_1 = [{"name": "hold_int16_255_1", "address": "255!400001",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_INT16}]
hold_int16_slave255_2 = [{"name": "hold_int16_255_2", "address": "255!400003",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_INT16}]
hold_int16_slave100_1 = [{"name": "hold_int16_100_1", "address": "100!400001",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_INT16}]
hold_int16_slave100_2 = [{"name": "hold_int16_100_2", "address": "100!400003",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_INT16}]
class TestModbus:
@@ -1608,3 +1616,146 @@ class TestModbus:
assert 200 == response.status_code
assert error.NEU_ERR_SUCCESS == response.json()['error']
@description(given="created modbus node with degradation", when="trigger degradation, read tags", then="success")
def test_modbus_degradation(self, param):
response = api.add_node(node=param[0]+"_degradation", plugin=param[1])
assert 200 == response.status_code
if param[0] == 'modbus-tcp':
api.modbus_tcp_node_setting(
node=param[0]+"_degradation", port=tcp_port, device_degrade=1)
elif param[0] == 'modbus-rtu':
api.modbus_rtu_node_setting(
node=param[0]+"_degradation", port=rtu_port, device_degrade=1)
else:
pytest.skip("modbus rtu tty pass")
response = api.add_group(node=param[0]+"_degradation", group='group')
assert 200 == response.status_code
api.add_tags_check(node=param[0]+"_degradation",
group='group', tags=hold_int16)
api.add_tags_check(node=param[0]+"_degradation",
group='group', tags=hold_int16_slave100_1)
api.add_tags_check(node=param[0]+"_degradation",
group='group', tags=hold_int16_slave100_2)
api.add_tags_check(node=param[0]+"_degradation",
group='group', tags=hold_int16_slave255_1)
api.add_tags_check(node=param[0]+"_degradation",
group='group', tags=hold_int16_slave255_2)
time.sleep(3)
assert 1 == api.read_tag(
node=param[0]+"_degradation", group='group', tag=hold_int16[0]['name'])
@description(given="endianness in node configuration, without endianness in tag", when="write and read tag", then="write/read success")
def test_write_read_endianness_node(self, param):
if param[0] != 'modbus-tcp':
pytest.skip("test modbus tcp only")
response = api.add_node(node=param[0]+"_ABCD", plugin=param[1])
assert 200 == response.status_code
api.modbus_tcp_node_setting_endian(
node=param[0]+"_ABCD", port=tcp_port, endianness=4)
api.add_tags_check(node=param[0]+"_ABCD", group='group', tags=hold_uint32)
api.write_tag_check(
node=param[0]+"_ABCD", group='group', tag=hold_uint32[0]['name'], value=0x31323334)
time.sleep(0.3)
assert 0x31323334 == api.read_tag(
node=param[0]+"_ABCD", group='group', tag=hold_uint32[0]['name'])
response = api.modbus_tcp_node_setting_endian(
node=param[0]+"_ABCD", port=tcp_port, endianness=1)
time.sleep(0.3)
assert 0x33343132 == api.read_tag(
node=param[0]+"_ABCD", group='group', tag=hold_uint32[0]['name'])
response = api.modbus_tcp_node_setting_endian(
node=param[0]+"_ABCD", port=tcp_port, endianness=2)
time.sleep(0.3)
assert 0x34333231 == api.read_tag(
node=param[0]+"_ABCD", group='group', tag=hold_uint32[0]['name'])
response = api.modbus_tcp_node_setting_endian(
node=param[0]+"_ABCD", port=tcp_port, endianness=3)
time.sleep(0.3)
assert 0x32313433 == api.read_tag(
node=param[0]+"_ABCD", group='group', tag=hold_uint32[0]['name'])
@description(given="endianness in tag, without endianness in node configuration", when="write and read tag", then="write/read success")
def test_write_read_endianness_tag(self, param):
if param[0] != 'modbus-tcp':
pytest.skip("test modbus tcp only")
api.modbus_tcp_node_setting_endian(
node=param[0]+"_ABCD", port=tcp_port, endianness=1)
time.sleep(0.3)
api.add_tags_check(node=param[0]+"_ABCD", group='group', tags=hold_uint32_LL)
api.add_tags_check(node=param[0]+"_ABCD", group='group', tags=hold_uint32_LB)
api.add_tags_check(node=param[0]+"_ABCD", group='group', tags=hold_uint32_BL)
api.add_tags_check(node=param[0]+"_ABCD", group='group', tags=hold_uint32_BB)
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_LL[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_LB[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_BL[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_BB[0]['name'])
api.modbus_tcp_node_setting_endian(
node=param[0]+"_ABCD", port=tcp_port, endianness=4)
time.sleep(0.3)
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_LL[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_LB[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_BL[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_BB[0]['name'])
api.modbus_tcp_node_setting_endian(
node=param[0]+"_ABCD", port=tcp_port, endianness=2)
time.sleep(0.3)
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_LL[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_LB[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_BL[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_BB[0]['name'])
api.modbus_tcp_node_setting_endian(
node=param[0]+"_ABCD", port=tcp_port, endianness=3)
time.sleep(0.3)
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_LL[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_LB[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_BL[0]['name'])
assert 12345678 == api.read_tag(
node=param[0], group='group', tag=hold_uint32_BB[0]['name'])
@description(given="created modbus node", when="set modbus start base to 0", then="set success and read/write success")
def test_set_modbus_start_base(self, param):
assert 1 == api.read_tag(
node=param[0], group='group', tag=coil_bit_1[0]['name'])
assert 0 == api.read_tag(
node=param[0], group='group', tag=coil_bit_2[0]['name'])
assert 1 == api.read_tag(
node=param[0], group='group', tag=coil_bit_3[0]['name'])
if param[0] == 'modbus-rtu':
api.modbus_rtu_node_setting_base(
node=param[0], port=rtu_port, base=0)
elif param[0] == 'modbus-tcp':
api.modbus_tcp_node_setting_base(
node=param[0], port=tcp_port, base=0)
else:
pytest.skip()
time.sleep(0.3)
assert 0 == api.read_tag(
node=param[0], group='group', tag=coil_bit_1[0]['name'])
assert 1 == api.read_tag(
node=param[0], group='group', tag=coil_bit_2[0]['name'])
assert 0 == api.read_tag(
node=param[0], group='group', tag=coil_bit_3[0]['name'])
+8 -4
View File
@@ -339,16 +339,20 @@ def prgfile_process(driver):
# plugin setting
def modbus_tcp_node_setting(node, port, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, retry_interval=1):
def modbus_tcp_node_setting(node, port, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, retry_interval=1,
device_degrade=0, degrade_cycle=1, degrade_time=600):
return node_setting(node, json={"connection_mode": connection_mode, "transport_mode": transport_mode, "interval": interval,
"host": host, "port": port, "timeout": timeout, "max_retries": max_retries, "retry_interval": retry_interval})
"host": host, "port": port, "timeout": timeout, "max_retries": max_retries, "retry_interval": retry_interval,
"device_degrade": device_degrade, "degrade_cycle": degrade_cycle, "degrade_time": degrade_time})
def modbus_rtu_node_setting(node, port=502, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, retry_interval=1, link=1, device="", stop=0, parity=0, baud=4, data=3):
def modbus_rtu_node_setting(node, port=502, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, retry_interval=1, link=1, device="", stop=0, parity=0, baud=4, data=3,
device_degrade=0, degrade_cycle=1, degrade_time=600):
return node_setting(node, json={"connection_mode": connection_mode, "transport_mode": transport_mode,
"interval": interval, "host": host, "port": port, "timeout": timeout,
"max_retries": max_retries, "retry_interval": retry_interval, "link": link, "device": device, "stop": stop,
"parity": parity, "baud": baud, "data": data})
"parity": parity, "baud": baud, "data": data,
"device_degrade": device_degrade, "degrade_cycle": degrade_cycle, "degrade_time": degrade_time})
def modbus_tcp_node_setting_endian(node, port, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, retry_interval=1, endianness=1):
+1 -1
View File
@@ -1 +1 @@
2.10.0-alpha
2.10.0