Merge pull request #2395 from ZhouBox/v2.10

[v2.10]otel:remove duplicate ctx

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.10.2
+  VERSION: 2.10.3
   UI_VERSION: 2.6.1
   BRANCH: v2.10
 

CMakeLists.txt

@@ -173,6 +173,7 @@ 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)
 

default_plugins.json

@@ -5,6 +5,7 @@
 		"libplugin-azure-iot.so",
 		"libplugin-ekuiper.so",
 		"libplugin-modbus-tcp.so",
-		"libplugin-modbus-rtu.so"
+		"libplugin-modbus-rtu.so",
+		"libplugin-file.so"
 	]
 }

include/neuron/define.h

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 10
-#define NEU_VERSION_FIX 2
+#define NEU_VERSION_FIX 3
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)

include/neuron/type.h

@@ -146,7 +146,7 @@ typedef struct {
     uint8_t *  ptr;
 } neu_value_ptr_t;
 
-#define NEU_VALUE_SIZE 128
+#define NEU_VALUE_SIZE 1024
 
 typedef struct {
     uint8_t bytes[NEU_VALUE_SIZE];

plugins/ekuiper/plugin_ekuiper.c

@@ -316,7 +316,7 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
 
     neu_otel_trace_ctx trace = NULL;
     neu_otel_scope_ctx scope = NULL;
-    if (neu_otel_control_is_started()) {
+    if (neu_otel_control_is_started() && header->ctx) {
         trace = neu_otel_find_trace(header->ctx);
         if (trace) {
             scope = neu_otel_add_span(trace);

plugins/file/CMakeLists.txt

@@ -0,0 +1,11 @@
+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)

plugins/file/file.json

@@ -0,0 +1,9 @@
+{
+	"tag_regex": [
+		{
+			"type": 13,
+			"regex": ""
+		}
+	],
+	"group_interval": 1000
+}

plugins/file/file_plugin.c

@@ -0,0 +1,247 @@
+/**
+ * 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, 1, 1024, fp);
+            if (ret < 0) {
+                dvalue.type      = NEU_TYPE_ERROR;
+                dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
+            } else {
+                dvalue.type = NEU_TYPE_STRING;
+                strncpy(dvalue.value.str, buf, NEU_VALUE_SIZE - 1);
+            }
+
+            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;
+}

plugins/mqtt/mqtt_plugin_intf.c

@@ -463,7 +463,7 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     neu_otel_trace_ctx trace           = NULL;
     neu_otel_scope_ctx scope           = NULL;
     char               new_span_id[36] = { 0 };
-    if (neu_otel_control_is_started()) {
+    if (neu_otel_control_is_started() && head->ctx) {
         trace = neu_otel_find_trace(head->ctx);
         if (trace) {
             scope = neu_otel_add_span(trace);

plugins/restful/normal_handle.c

@@ -307,7 +307,7 @@ void handle_get_plugin_schema(nng_aio *aio)
         return;
     }
 
-    if (0 > neu_asprintf(&schema_path, "%s/schema/%s.json", g_plugin_dir,
+    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);
@@ -332,8 +332,8 @@ void handle_get_plugin_schema(nng_aio *aio)
 
     if (NULL == buf) {
         free(schema_path);
-        if (0 > neu_asprintf(&schema_path, "%s/system/schema/%s.json",
-                             g_plugin_dir, schema_name)) {
+        if (0 > neu_asprintf(&schema_path, "%s/schema/%s.json", g_plugin_dir,
+                             schema_name)) {
             NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
                 neu_http_response(aio, error_code.error, result_error);
             });

plugins/restful/rest.c

@@ -176,7 +176,7 @@ static int dashb_plugin_request(neu_plugin_t *      plugin,
 
     neu_otel_trace_ctx trace = NULL;
     neu_otel_scope_ctx scope = NULL;
-    if (neu_otel_control_is_started()) {
+    if (neu_otel_control_is_started() && header->ctx) {
         trace = neu_otel_find_trace(header->ctx);
         if (trace) {
             scope = neu_otel_add_span(trace);

plugins/restful/rw_handle.c

@@ -284,7 +284,7 @@ void handle_write(nng_aio *aio)
             }
 
             if (req->t == NEU_JSON_ARRAY_INT64 &&
-                req->value.val_array_int64.length > NEU_VALUE_SIZE) {
+                req->value.val_array_int64.length > 128) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, {
                     neu_http_response(aio, NEU_ERR_STRING_TOO_LONG,
                                       result_error);

src/core/manager.c

@@ -1538,6 +1538,47 @@ static char *file_save_tmp(const char *data, const char *suffix)
     return tmp_path;
 }
 
+static int copy_file(const char *src, const char *dst)
+{
+    FILE * source, *destination;
+    char   buffer[1024];
+    size_t bytes_read;
+
+    source = fopen(src, "rb");
+    if (source == NULL) {
+        nlog_error("failed to open source file!");
+        return -1;
+    }
+
+    destination = fopen(dst, "wb");
+    if (destination == NULL) {
+        nlog_error("failed to open destination file!");
+        fclose(source);
+        return -1;
+    }
+
+    while ((bytes_read = fread(buffer, 1, sizeof(buffer), source)) > 0) {
+        if (fwrite(buffer, 1, bytes_read, destination) != bytes_read) {
+            nlog_error("failed to write to destination file!");
+            fclose(source);
+            fclose(destination);
+            return -1;
+        }
+    }
+
+    if (ferror(source)) {
+        nlog_error("error reading source file");
+        fclose(source);
+        fclose(destination);
+        return -1;
+    }
+
+    fclose(source);
+    fclose(destination);
+
+    return 0;
+}
+
 static bool mv_tmp_library_file(neu_plugin_kind_e kind, const char *tmp_path,
                                 const char *library)
 {
@@ -1564,9 +1605,8 @@ static bool mv_tmp_library_file(neu_plugin_kind_e kind, const char *tmp_path,
         }
     }
 
-    if (rename(tmp_path, file_name) != 0) {
-        nlog_error("%s rename %s fail, err:%s", tmp_path, file_name,
-                   strerror(errno));
+    if (copy_file(tmp_path, file_name) != 0) {
+        nlog_error("%s copy_file %s fail!", tmp_path, file_name);
         return false;
     } else {
         return true;
@@ -1604,9 +1644,8 @@ static bool mv_tmp_schema_file(neu_plugin_kind_e kind, const char *tmp_path,
         }
     }
 
-    if (rename(tmp_path, file_name) != 0) {
-        nlog_error("%s rename %s fail, err:%s", tmp_path, file_name,
-                   strerror(errno));
+    if (copy_file(tmp_path, file_name) != 0) {
+        nlog_error("%s copy_file %s fail!", tmp_path, file_name);
         return false;
     } else {
         return true;

src/otel/otel_manager.c

@@ -186,6 +186,20 @@ void neu_otel_free_span(Opentelemetry__Proto__Trace__V1__Span *span)
 neu_otel_trace_ctx neu_otel_create_trace(const char *trace_id, void *req_ctx,
                                          uint32_t flags, const char *tracestate)
 {
+    trace_ctx_table_ele_t *find = NULL;
+
+    pthread_mutex_lock(&table_mutex);
+
+    HASH_FIND(hh, traces_table, &req_ctx, sizeof(req_ctx), find);
+
+    if (find) {
+        HASH_DEL(traces_table, find);
+        neu_otel_free_trace(find->ctx);
+        free(find);
+    }
+
+    pthread_mutex_unlock(&table_mutex);
+
     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);
@@ -458,7 +472,16 @@ neu_otel_trace_ctx neu_otel_find_trace(void *req_ctx)
     pthread_mutex_unlock(&table_mutex);
 
     if (find) {
-        return (void *) find->ctx;
+        if (find->ctx->final &&
+            find->ctx->span_num ==
+                find->ctx->trace_data.resource_spans[0]
+                    ->scope_spans[0]
+                    ->n_spans &&
+            find->ctx->expected_span_num <= 0) {
+            return NULL;
+        } else {
+            return (void *) find->ctx;
+        }
     } else {
         return NULL;
     }
@@ -880,7 +903,8 @@ void neu_otel_scope_set_status_code(neu_otel_scope_ctx     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->code =
+        (Opentelemetry__Proto__Trace__V1__Status__StatusCode) code;
     scope->span->status->message = strdup(desc);
 }
 
@@ -893,7 +917,8 @@ void neu_otel_scope_set_status_code2(neu_otel_scope_ctx     ctx,
     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->code =
+        (Opentelemetry__Proto__Trace__V1__Status__StatusCode) code;
     scope->span->status->message = error_buf;
 }
 
@@ -930,8 +955,10 @@ int neu_otel_trace_pack(neu_otel_trace_ctx ctx, uint8_t *out)
 
 void neu_otel_new_span_id(char *id)
 {
+    int64_t p_id = (int64_t) pthread_self();
     for (int i = SPAN_ID_LENGTH - 1; i >= 0; i--) {
-        id[i] = ID_CHARSET[rand() % 16];
+        int64_t rand_id = (int64_t) rand() + neu_time_ns() + p_id;
+        id[i]           = ID_CHARSET[rand_id % 16];
     }
     id[SPAN_ID_LENGTH] = '\0';
 }
@@ -1007,6 +1034,8 @@ static int otel_timer_cb(void *data)
                 HASH_DEL(traces_table, el);
                 neu_otel_free_trace(el->ctx);
                 free(el);
+            } else {
+                break;
             }
         } else if (neu_time_ms() - el->ctx->ts >= TRACE_TIME_OUT) {
             nlog_debug("trace:%s time out", (char *) el->ctx->trace_id);
@@ -1033,7 +1062,7 @@ void neu_otel_start()
         param.usr_data = NULL;
 
         param.second      = 0;
-        param.millisecond = 100;
+        param.millisecond = 80;
         param.cb          = otel_timer_cb;
         param.type        = NEU_EVENT_TIMER_BLOCK;
 

version

@@ -1 +1 @@
-2.10.2
+2.10.3