Merge pull request #1292 from fengzeroz/v2.3

version: 2.3.13

.github/workflows/ci.yml

@@ -8,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.3.7
-  UI_VERSION: 2.3.6
+  VERSION: 2.3.13
+  UI_VERSION: 2.3.9
   BRANCH: v2.3
 
 jobs:
@@ -96,7 +96,7 @@ jobs:
 
     steps:
       - name: delete old daily-pre-release
-        uses: dev-drprasad/delete-tag-and-release@v0.2.0
+        uses: dev-drprasad/delete-tag-and-release@v0.2.1
         with:
           delete_release: true
           tag_name: ${{ env.BRANCH }}-daily
@@ -258,4 +258,4 @@ jobs:
           upload_url: ${{ github.event.release.upload_url }}
           asset_path: ./${{ steps.package.outputs.name}}
           asset_name: ${{ matrix.type }}-${{ env.VERSION }}-linux-${{ matrix.arch }}.${{ matrix.package }}
-          asset_content_type: application/octet-stream
+          asset_content_type: application/octet-stream

CMakeLists.txt

@@ -59,6 +59,9 @@ else()
   link_directories(/usr/local/lib)
 endif()
 
+set(PERSIST_SOURCES
+    src/persist/persist.c
+    src/persist/json/persist_json_plugin.c)
 aux_source_directory(src/parser NEURON_SRC_PARSE)
 set(NEURON_BASE_SOURCES
     src/base/tag.c
@@ -74,7 +77,8 @@ set(NEURON_BASE_SOURCES
     src/utils/neu_jwt.c
     src/utils/base64.c
     src/utils/async_queue.c
-    src/utils/mem_cache.c)
+    src/utils/mem_cache.c
+    ${PERSIST_SOURCES})
 
 add_library(neuron-base SHARED)
 target_sources(neuron-base PRIVATE ${NEURON_BASE_SOURCES} ${NEURON_SRC_PARSE}) 
@@ -89,9 +93,6 @@ add_dependencies(neuron-base neuron-version)
 #find_package(MbedTLS)
 target_link_libraries(neuron-base mbedtls mbedx509 mbedcrypto)
 
-set(PERSIST_SOURCES
-    src/persist/persist.c
-    src/persist/json/persist_json_plugin.c)
 set(NEURON_SOURCES
     src/main.c
     src/argparse.c
@@ -123,8 +124,7 @@ set(NEURON_SOURCES
     plugins/restful/proxy.c
     plugins/restful/websocket.c
     plugins/monitor/monitor.c
-    plugins/monitor/metric_handle.c
-    ${PERSIST_SOURCES})
+    plugins/monitor/metric_handle.c)
 
 set(CMAKE_BUILD_RPATH ./)
 add_executable(neuron)

include/neuron/persist/persist.h

@@ -24,7 +24,9 @@
 extern "C" {
 #endif
 
-#include "adapter/adapter_info.h"
+#include <sqlite3.h>
+
+#include "adapter_info.h"
 #include "persist/json/persist_json_plugin.h"
 
 typedef neu_json_plugin_req_plugin_t neu_persist_plugin_info_t;
@@ -84,12 +86,14 @@ static inline void neu_persist_user_info_fini(neu_persist_user_info_t *info)
  * Create persister.
  * @return 0 on success, -1 otherwise.
  */
-int neu_persister_create();
+int neu_persister_create(const char *schema_dir);
 /**
  * Destroy perister.
  */
 void neu_persister_destroy();
 
+sqlite3 *neu_persister_get_db();
+
 /**
  * Persist nodes.
  * @param node_info                 neu_persist_node_info_t.
@@ -306,4 +310,4 @@ int neu_persister_delete_user(const char *user_name);
 }
 #endif
 
-#endif
+#endif

plugins/file/file_req.c

@@ -32,6 +32,10 @@ static void plugin_group_free(neu_plugin_group_t *pgp);
 
 int file_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;
+    int64_t rtt = NEU_METRIC_LAST_RTT_MS_MAX;
+
     struct file_group_data *gd     = NULL;
     int                     length = plugin->file_length;
 
@@ -83,6 +87,8 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         strncpy(dvalue.value.str, buf, strlen(buf));
         fclose(f);
 
+        rtt = 1;
+
     dvalue_result:
         plugin->common.adapter_callbacks->driver.update(
             plugin->common.adapter, group->group_name, tag->name, dvalue);
@@ -90,6 +96,8 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         free(buf);
     }
 
+    update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, rtt, NULL);
+
     return 0;
 }
 

plugins/restful/log_handle.c

@@ -50,25 +50,45 @@ static int http_resp_files(nng_aio *aio, void *data, size_t len,
 
 void handle_logs_files(nng_aio *aio)
 {
-    char   path[128] = { 0 };
-    char   disp[128] = { 0 };
-    void * data      = NULL;
-    size_t len       = 0;
-    int    rv        = 0;
+    void * data = NULL;
+    size_t len  = 0;
+    int    rv   = 0;
 
     VALIDATE_JWT(aio);
 
-    rv = mkdir("neuron_logs", S_IRWXU);
-    if (0 != rv && EEXIST == errno) {
-        rv = 0;
+    /* check whether the neuron directory exists */
+    rv = access("../neuron", F_OK);
+    if (0 != rv) {
+        nlog_error("The neuron directory does not exists");
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_FILE_NOT_EXIST, {
+            http_response(aio, error_code.error, result_error);
+        });
+        return;
     }
 
-    // check if coredump exists
-    rv = access("./core", F_OK);
-    if (0 == rv) {
-        rv = system("cp -r ./core neuron_logs");
-        if (0 != rv) {
-            nlog_error("failed to copy coredump file");
+    rv = system("rm -rf /tmp/neuron_debug.tar.gz");
+    nlog_warn("remove old neuron_debug.tar.gz, rv: %d", rv);
+    /* tar the neuron directory */
+    rv = system("tar -zcvf /tmp/neuron_debug.tar.gz ../neuron");
+    if (rv == -1) {
+        nlog_error("failed to create neuron_debug.tar.gz, rv: %d", rv);
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
+            http_response(aio, error_code.error, result_error);
+        });
+        return;
+    } else {
+        if (WIFEXITED(rv)) {
+            if (WEXITSTATUS(rv) < 0) {
+                nlog_error("failed to create neuron_debug.tar.gz, rv: %d",
+                           WEXITSTATUS(rv));
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
+                    http_response(aio, error_code.error, result_error);
+                });
+                return;
+            }
+        } else {
+            nlog_error("failed to create neuron_debug.tar.gz, rv: %d",
+                       WEXITSTATUS(rv));
             NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
                 http_response(aio, error_code.error, result_error);
             });
@@ -76,48 +96,18 @@ void handle_logs_files(nng_aio *aio)
         }
     }
 
-    rv = system("cp -r ./logs neuron_logs");
-    if (0 != rv) {
-        nlog_error("failed to copy logs file");
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
-            http_response(aio, error_code.error, result_error);
-        });
-        return;
-    }
-
-    rv = system("tar -zcvf neuron_logs.tar.gz neuron_logs");
-    if (0 != rv) {
-        nlog_error("failed to create neuron_logs.tar.gz");
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
-            http_response(aio, error_code.error, result_error);
-        });
-        return;
-    }
-
-    strcpy(path, "neuron_logs.tar.gz");
-    rv = read_file(path, &data, &len);
+    rv = read_file("/tmp/neuron_debug.tar.gz", &data, &len);
     if (0 != rv) {
         NEU_JSON_RESPONSE_ERROR(
             rv, { http_response(aio, error_code.error, result_error); });
         return;
     }
 
-    // delete neuron_logs folder from build
-    rv = system("rm -rf neuron_logs.tar.gz neuron_logs");
-    if (0 != rv) {
-        nlog_error("failed to delete neuron_logs folder");
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
-            http_response(aio, error_code.error, result_error);
-        });
-        return;
-    }
+    /* handle http response */
+    rv = http_resp_files(aio, data, len,
+                         "attachment; filename=neuron_debug.tar.gz");
 
-    // handle http response
-    strcpy(disp, "attachment; filename=neuron_logs.tar.gz");
-    rv = http_resp_files(aio, data, len, disp);
-    if (0 != rv) {
-        return;
-    }
+    nlog_notice("download neuron_debug.tar.gz, ret: %d", rv);
 }
 
 void handle_log_level(nng_aio *aio)

src/adapter/driver/driver.c

@@ -299,6 +299,8 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
                         (double) (cmd->value.value.d64 / tag->decimal);
                 } else if (cmd->value.type == NEU_TYPE_DOUBLE) {
                     cmd->value.value.d64 = cmd->value.value.d64 / tag->decimal;
+                } else if (cmd->value.type == NEU_TYPE_FLOAT) {
+                    cmd->value.value.f32 = cmd->value.value.f32 / tag->decimal;
                 }
                 break;
             default:
@@ -357,8 +359,11 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             }
             break;
         case NEU_TYPE_FLOAT:
-            cmd->value.type      = NEU_TYPE_FLOAT;
-            cmd->value.value.f32 = (float) cmd->value.value.d64;
+            if (NEU_TYPE_FLOAT != cmd->value.type) {
+                cmd->value.value.f32 = (float) cmd->value.value.d64;
+                cmd->value.type      = NEU_TYPE_FLOAT;
+            }
+
             switch (tag->option.value32.endian) {
             case NEU_DATATAG_ENDIAN_LB32:
                 neu_ntohs_p((uint16_t *) cmd->value.value.bytes);

src/argparse.c

@@ -94,7 +94,7 @@ static inline int reset_password()
             "$5$PwFeXpBBIBZuZdZl$fP8fFPWCLoaWcnVXVSR.3Xi8TEqCvX92gjhowNNn6S4",
     };
 
-    int rv = neu_persister_create();
+    int rv = neu_persister_create(g_config_dir);
     if (0 != rv) {
         return rv;
     }

src/connection/connection.c

@@ -389,7 +389,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                    conn->fd, len, ret, strerror(errno), errno);
     }
 
-    if (errno == EPIPE || ret == -1) {
+    if (errno == EPIPE || ret == -1 || ret == 0) {
         conn_disconnect(conn);
     }
 

src/connection/mqtt_client.c

@@ -150,6 +150,7 @@ static void connect_cb(nng_pipe p, nng_pipe_ev ev, void *arg);
 
 static inline task_t *client_alloc_task(neu_mqtt_client_t *client);
 static inline void    client_free_task(neu_mqtt_client_t *client, task_t *task);
+static inline size_t  client_task_free_list_len(neu_mqtt_client_t *client);
 static inline void    client_add_subscription(neu_mqtt_client_t *client,
                                               subscription_t *   sub);
 static int            client_send_sub_msg(neu_mqtt_client_t *client,
@@ -545,6 +546,16 @@ static inline void client_free_task(neu_mqtt_client_t *client, task_t *task)
     DL_PREPEND(client->task_free_list, task);
 }
 
+static inline size_t client_task_free_list_len(neu_mqtt_client_t *client)
+{
+    size_t  count = 0;
+    task_t *elt   = NULL;
+    // count by iterating the list,
+    // this won't be a problem as a short free list is expected
+    DL_COUNT(client->task_free_list, elt, count);
+    return count;
+}
+
 static inline void client_add_subscription(neu_mqtt_client_t *client,
                                            subscription_t *   sub)
 {
@@ -1149,6 +1160,12 @@ int neu_mqtt_client_close(neu_mqtt_client_t *client)
     }
 
     nng_mtx_lock(client->mtx);
+    // wait for all tasks
+    while (client_task_free_list_len(client) != client->task_count) {
+        nng_mtx_unlock(client->mtx);
+        nng_msleep(100);
+        nng_mtx_lock(client->mtx);
+    }
     client->open = false;
     nng_mtx_unlock(client->mtx);
 

src/event/event_linux.c

@@ -88,6 +88,11 @@ static void *event_loop(void *arg)
         if (ret == 0) {
             continue;
         }
+
+        if (ret == -1 && errno == EINTR) {
+            continue;
+        }
+
         if (ret == -1 || events->stop) {
             zlog_warn(neuron, "event loop exit, errno: %s(%d), stop: %d",
                       strerror(errno), errno, events->stop);

src/main.c

@@ -42,7 +42,7 @@ static void sig_handler(int sig)
 {
     nlog_warn("recv sig: %d", sig);
 
-    if (sig != SIGABRT) {
+    if (sig == SIGINT || sig == SIGTERM) {
         neu_manager_destroy(g_manager);
         neu_persister_destroy();
     }
@@ -58,6 +58,7 @@ static int neuron_run(const neu_cli_args_t *args)
     signal(SIGINT, sig_handler);
     signal(SIGTERM, sig_handler);
     signal(SIGABRT, sig_handler);
+    signal(SIGSEGV, sig_handler);
 
     // try to enable core dump
     rl.rlim_cur = rl.rlim_max = RLIM_INFINITY;
@@ -65,7 +66,7 @@ static int neuron_run(const neu_cli_args_t *args)
         nlog_warn("neuron process failed enable core dump, ignore");
     }
 
-    rv = neu_persister_create();
+    rv = neu_persister_create(args->config_dir);
     assert(rv == 0);
 
     zlog_notice(neuron, "neuron process, daemon: %d, version: %s (%s %s)",

src/persist/json/persist_json_plugin.c

@@ -27,7 +27,7 @@
 
 #include "json/json.h"
 
-#include "persist_json_plugin.h"
+#include "persist/json/persist_json_plugin.h"
 
 int neu_json_decode_plugin_req(char *buf, neu_json_plugin_req_t **result)
 {

src/persist/persist.c

@@ -430,7 +430,7 @@ static int apply_schemas(sqlite3 *db, const char *dir)
     return rv;
 }
 
-int neu_persister_create()
+int neu_persister_create(const char *schema_dir)
 {
     int rv = sqlite3_open(db_file, &global_db);
 
@@ -455,7 +455,7 @@ int neu_persister_create()
         return -1;
     }
 
-    rv = apply_schemas(global_db, g_config_dir);
+    rv = apply_schemas(global_db, schema_dir);
     if (rv != 0) {
         nlog_fatal("db apply schemas fail");
         sqlite3_close(global_db);
@@ -465,6 +465,11 @@ int neu_persister_create()
     return 0;
 }
 
+sqlite3 *neu_persister_get_db()
+{
+    return global_db;
+}
+
 void neu_persister_destroy()
 {
     sqlite3_close(global_db);

version

@@ -1 +1 @@
-2.3.7
+2.3.13