Merge pull request #1587 from fengzeroz/v2.4

version: 2.4.11

.github/workflows/ci.yml

@@ -8,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.4.8
-  UI_VERSION: 2.4.6
+  VERSION: 2.4.11
+  UI_VERSION: 2.4.8
   BRANCH: v2.4
 
 jobs:

include/neuron/adapter.h

@@ -207,8 +207,8 @@ typedef struct neu_resp_error {
 } neu_resp_error_t;
 
 typedef struct {
-    char node[NEU_NODE_NAME_LEN];
-    bool auto_start;
+    char                     node[NEU_NODE_NAME_LEN];
+    neu_node_running_state_e state;
 } neu_req_node_init_t, neu_req_node_uninit_t, neu_resp_node_uninit_t;
 
 typedef struct neu_req_add_plugin {

include/neuron/errcodes.h

@@ -137,6 +137,7 @@ typedef enum {
     NEU_ERR_FILE_OPEN_FAILURE  = 4101,
     NEU_ERR_FILE_READ_FAILURE  = 4102,
     NEU_ERR_FILE_WRITE_FAILURE = 4103,
+    NEU_ERR_FILE_READ_NULL     = 4104,
 
     NEU_ERR_PLUGIN_ERROR_START = 10000,
     NEU_ERR_PLUGIN_ERROR_END   = 20000,

include/neuron/persist/persist.h

@@ -95,6 +95,8 @@ int neu_persister_create(const char *schema_dir);
 void neu_persister_destroy();
 
 sqlite3 *neu_persister_get_db();
+int      neu_persister_node_count();
+int      neu_persister_tag_count();
 
 /**
  * Persist nodes.

plugins/file/file_req.c

@@ -84,10 +84,16 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         fseek(f, 0, SEEK_SET);
         int ret = fread(buf, sizeof(char), length, f);
         if (ret == 0) {
-            nlog_warn("read file failed:%s", tag->address);
-            dvalue.type      = NEU_TYPE_ERROR;
-            dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
-
+            if (feof(f)) {
+                nlog_warn("file is empty:%s", tag->address);
+                dvalue.type      = NEU_TYPE_ERROR;
+                dvalue.value.i32 = NEU_ERR_FILE_READ_NULL;
+            } else if (ferror(f)) {
+                nlog_warn("read file failed: %s", tag->address);
+                clearerr(f);
+                dvalue.type      = NEU_TYPE_ERROR;
+                dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
+            }
             fclose(f);
             goto dvalue_result;
         } else if (ret >= NEU_VALUE_SIZE) {

plugins/restful/log_handle.c

@@ -63,7 +63,8 @@ void handle_logs_files(nng_aio *aio)
     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");
+    rv = system("tar -zcvf /tmp/neuron_debug.tar.gz "
+                "--exclude='persistence/neuron.lic' ../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, {

src/adapter/adapter.c

@@ -187,7 +187,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     return adapter;
 }
 
-void neu_adapter_init(neu_adapter_t *adapter, bool auto_start)
+void neu_adapter_init(neu_adapter_t *adapter, neu_node_running_state_e state)
 {
     neu_reqresp_head_t  header   = { .type = NEU_REQ_NODE_INIT };
     neu_req_node_init_t init     = { 0 };
@@ -197,8 +197,8 @@ void neu_adapter_init(neu_adapter_t *adapter, bool auto_start)
     strcpy(header.receiver, "manager");
 
     strcpy(init.node, adapter->name);
-    init.auto_start = auto_start;
-    init_msg        = neu_msg_gen(&header, &init);
+    init.state = state;
+    init_msg   = neu_msg_gen(&header, &init);
 
     nng_sendmsg(adapter->sock, init_msg, 0);
 }

src/adapter/adapter_internal.h

@@ -64,7 +64,7 @@ typedef struct adapter_msg_handler {
 } adapter_msg_handler_t;
 
 neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info);
-void           neu_adapter_init(neu_adapter_t *adapter, bool auto_start);
+void neu_adapter_init(neu_adapter_t *adapter, neu_node_running_state_e state);
 
 int neu_adapter_start(neu_adapter_t *adapter);
 int neu_adapter_start_single(neu_adapter_t *adapter);

src/adapter/driver/driver.c

@@ -55,6 +55,7 @@ typedef struct group {
 
     neu_event_timer_t *report;
     neu_event_timer_t *read;
+    neu_event_timer_t *write;
 
     neu_plugin_group_t    grp;
     neu_adapter_driver_t *driver;
@@ -73,6 +74,7 @@ struct neu_adapter_driver {
 
 static int  report_callback(void *usr_data);
 static int  read_callback(void *usr_data);
+static int  write_callback(void *usr_data);
 static void read_group(int64_t timestamp, int64_t timeout,
                        neu_driver_cache_t *cache, const char *group,
                        UT_array *tags, neu_resp_tag_value_t *datas);
@@ -185,6 +187,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
 
         neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
         neu_event_del_timer(driver->driver_events, el->read);
+        neu_event_del_timer(driver->driver_events, el->write);
         if (el->grp.group_free != NULL) {
             el->grp.group_free(&el->grp);
         }
@@ -481,6 +484,11 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
         param.cb     = read_callback;
         find->read   = neu_event_add_timer(driver->driver_events, param);
 
+        param.second      = 0;
+        param.millisecond = 100;
+        param.cb          = write_callback;
+        find->write       = neu_event_add_timer(driver->driver_events, param);
+
         REGISTER_GROUP_METRIC(&driver->adapter, find->name,
                               NEU_METRIC_GROUP_TAGS_TOTAL,
                               neu_group_tag_size(find->group));
@@ -539,6 +547,7 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
 
         neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
         neu_event_del_timer(driver->driver_events, find->read);
+        neu_event_del_timer(driver->driver_events, find->write);
         if (find->grp.group_free != NULL) {
             find->grp.group_free(&find->grp);
         }
@@ -917,7 +926,7 @@ static void group_change(void *arg, int64_t timestamp, UT_array *static_tags,
                 timestamp);
 }
 
-static int read_callback(void *usr_data)
+static int write_callback(void *usr_data)
 {
     group_t *                group = (group_t *) usr_data;
     neu_node_running_state_e state = group->driver->adapter.state;
@@ -936,6 +945,17 @@ static int read_callback(void *usr_data)
     utarray_clear(group->wt_tags);
     pthread_mutex_unlock(&group->wt_mtx);
 
+    return 0;
+}
+
+static int read_callback(void *usr_data)
+{
+    group_t *                group = (group_t *) usr_data;
+    neu_node_running_state_e state = group->driver->adapter.state;
+    if (state != NEU_NODE_RUNNING_STATE_RUNNING) {
+        return 0;
+    }
+
     if (neu_group_is_change(group->group, group->timestamp)) {
         neu_group_change_test(group->group, group->timestamp, (void *) group,
                               group_change);

src/base/group.c

@@ -286,7 +286,8 @@ static void split_static_array(tag_elem_t *tags, UT_array **static_tags,
     {
         if (neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_STATIC)) {
             utarray_push_back(*static_tags, el->tag);
-        } else {
+        } else if (neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_SUBSCRIBE) ||
+                   neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_READ)) {
             utarray_push_back(*other_tags, el->tag);
         }
     }

src/connection/connection.c

@@ -47,6 +47,7 @@ struct neu_conn {
     void *           data;
     bool             is_connected;
     bool             stop;
+    bool             connection_ok;
 
     neu_conn_callback connected;
     neu_conn_callback disconnected;
@@ -340,8 +341,14 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                        conn->fd, len, ret, strerror(errno), errno);
         }
 
-        if (ret == -1 && errno != EAGAIN) {
-            conn_disconnect(conn);
+        if (ret == -1) {
+            if (errno != EAGAIN) {
+                conn_disconnect(conn);
+            } else {
+                if (conn->connection_ok == true) {
+                    conn_disconnect(conn);
+                }
+            }
         }
 
         if (ret > 0 && conn->callback_trigger == false) {
@@ -352,6 +359,7 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
 
     if (ret > 0) {
         conn->state.send_bytes += ret;
+        conn->connection_ok = true;
     }
 
     pthread_mutex_unlock(&conn->mtx);
@@ -397,13 +405,23 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
         }
         break;
     }
-    if (ret <= 0) {
-        zlog_error(conn->param.log,
-                   "conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
-                   conn->fd, len, ret, strerror(errno), errno);
-        if (errno == EPIPE || (ret == 0 && errno != 0)) {
+    if (conn->param.type == NEU_CONN_TTY_CLIENT) {
+        if (ret == -1) {
+            zlog_error(
+                conn->param.log,
+                "tty conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
+                conn->fd, len, ret, strerror(errno), errno);
             conn_disconnect(conn);
         }
+    } else {
+        if (ret <= 0) {
+            zlog_error(conn->param.log,
+                       "conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
+                       conn->fd, len, ret, strerror(errno), errno);
+            if (ret == 0 || (ret == -1 && errno != EAGAIN)) {
+                conn_disconnect(conn);
+            }
+        }
     }
 
     pthread_mutex_unlock(&conn->mtx);
@@ -483,6 +501,10 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         } else {
             conn->block = false;
         }
+        zlog_notice(conn->param.log, "tcp server %s:%d, timeout: %d, block: %d",
+                    conn->param.params.tcp_server.ip,
+                    conn->param.params.tcp_server.port,
+                    conn->param.params.tcp_server.timeout, conn->block);
         break;
     case NEU_CONN_TCP_CLIENT:
         conn->param.params.tcp_client.ip = strdup(param->params.tcp_client.ip);
@@ -490,6 +512,10 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         conn->param.params.tcp_client.timeout =
             param->params.tcp_client.timeout;
         conn->block = conn->param.params.tcp_client.timeout > 0;
+        zlog_notice(conn->param.log, "tcp client %s:%d, timeout: %d, block: %d",
+                    conn->param.params.tcp_client.ip,
+                    conn->param.params.tcp_client.port,
+                    conn->param.params.tcp_client.timeout, conn->block);
         break;
     case NEU_CONN_UDP:
         conn->param.params.udp.src_ip   = strdup(param->params.udp.src_ip);
@@ -498,6 +524,11 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         conn->param.params.udp.dst_port = param->params.udp.dst_port;
         conn->param.params.udp.timeout  = param->params.udp.timeout;
         conn->block                     = conn->param.params.udp.timeout > 0;
+        zlog_notice(
+            conn->param.log, "udp %s:%d -> %s:%d, timeout: %d, block: %d",
+            conn->param.params.udp.src_ip, conn->param.params.udp.src_port,
+            conn->param.params.udp.dst_ip, conn->param.params.udp.dst_port,
+            conn->param.params.udp.timeout, conn->block);
         break;
     case NEU_CONN_TTY_CLIENT:
         conn->param.params.tty_client.device =
@@ -620,9 +651,9 @@ static void conn_connect(neu_conn_t *conn)
             conn->is_connected = false;
             return;
         } else {
-            zlog_notice(conn->param.log, "connect %s:%d success",
+            zlog_notice(conn->param.log, "connect %s:%d, block: %d success",
                         conn->param.params.tcp_client.ip,
-                        conn->param.params.tcp_client.port);
+                        conn->param.params.tcp_client.port, conn->block);
             conn->is_connected = true;
             conn->fd           = fd;
         }
@@ -842,7 +873,8 @@ static void conn_connect(neu_conn_t *conn)
 
 static void conn_disconnect(neu_conn_t *conn)
 {
-    conn->is_connected = false;
+    conn->is_connected  = false;
+    conn->connection_ok = false;
     if (conn->callback_trigger == true) {
         conn->disconnected(conn->data, conn->fd);
         conn->callback_trigger = false;

src/core/manager.c

@@ -267,10 +267,14 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         neu_node_manager_update(manager->node_manager, init->node, pipe);
 
-        if (init->auto_start) {
+        if (init->state == NEU_NODE_RUNNING_STATE_STOPPED ||
+            init->state == NEU_NODE_RUNNING_STATE_RUNNING) {
             neu_adapter_t *adapter =
                 neu_node_manager_find(manager->node_manager, init->node);
             neu_adapter_start(adapter);
+            if (init->state == NEU_NODE_RUNNING_STATE_STOPPED) {
+                neu_adapter_stop(adapter);
+            }
         }
 
         nlog_notice("bind node %s to pipe(%d)", init->node, pipe.id);

src/core/manager_internal.c

@@ -43,7 +43,8 @@ UT_array *neu_manager_get_plugins(neu_manager_t *manager)
 }
 
 int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
-                         const char *plugin_name, bool start)
+                         const char *             plugin_name,
+                         neu_node_running_state_e state)
 {
     neu_adapter_t *       adapter      = NULL;
     neu_plugin_instance_t instance     = { 0 };
@@ -77,7 +78,7 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
 
     adapter = neu_adapter_create(&adapter_info);
     neu_node_manager_add(manager->node_manager, adapter);
-    neu_adapter_init(adapter, start);
+    neu_adapter_init(adapter, state);
 
     return NEU_ERR_SUCCESS;
 }

src/core/manager_internal.h

@@ -52,7 +52,8 @@ int       neu_manager_del_plugin(neu_manager_t *manager, const char *plugin);
 UT_array *neu_manager_get_plugins(neu_manager_t *manager);
 
 int       neu_manager_add_node(neu_manager_t *manager, const char *node_name,
-                               const char *plugin_name, bool start);
+                               const char *             plugin_name,
+                               neu_node_running_state_e state);
 int       neu_manager_del_node(neu_manager_t *manager, const char *node_name);
 UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type,
                                 const char *plugin, const char *node);

src/core/storage.c

@@ -118,9 +118,8 @@ int manager_load_node(neu_manager_t *manager)
 
     utarray_foreach(node_infos, neu_persist_node_info_t *, node_info)
     {
-        rv = neu_manager_add_node(
-            manager, node_info->name, node_info->plugin_name,
-            node_info->state == NEU_NODE_RUNNING_STATE_RUNNING);
+        rv                    = neu_manager_add_node(manager, node_info->name,
+                                  node_info->plugin_name, node_info->state);
         const char *ok_or_err = (0 == rv) ? "success" : "fail";
         nlog_notice("load adapter %s type:%d, name:%s plugin:%s state:%d",
                     ok_or_err, node_info->type, node_info->name,

src/persist/persist.c

@@ -25,6 +25,7 @@
 #include <dirent.h>
 #include <errno.h>
 #include <fcntl.h>
+#include <pthread.h>
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <unistd.h>
@@ -49,9 +50,12 @@
 
 #define PATH_MAX_SIZE 128
 
-static const char *plugin_file = "persistence/plugins.json";
-static const char *db_file     = "persistence/sqlite.db";
-static sqlite3 *   global_db   = NULL;
+static const char *plugin_file       = "persistence/plugins.json";
+static const char *db_file           = "persistence/sqlite.db";
+static sqlite3 *   global_db         = NULL;
+pthread_rwlock_t   global_rwlock     = PTHREAD_RWLOCK_INITIALIZER;
+static int         global_node_count = 0;
+static int         global_tag_count  = 0;
 
 static inline bool ends_with(const char *str, const char *suffix)
 {
@@ -474,6 +478,44 @@ static int apply_schemas(sqlite3 *db, const char *dir)
     return rv;
 }
 
+static inline int node_count_load()
+{
+    int           rv    = 0;
+    sqlite3_stmt *stmt  = NULL;
+    char *        query = "select count(*) from nodes";
+
+    if (SQLITE_OK == sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL) &&
+        SQLITE_ROW == sqlite3_step(stmt)) {
+        pthread_rwlock_wrlock(&global_rwlock);
+        global_node_count = sqlite3_column_int(stmt, 0);
+        pthread_rwlock_unlock(&global_rwlock);
+    } else {
+        rv = -1;
+    }
+
+    sqlite3_finalize(stmt);
+    return rv;
+}
+
+static inline int tag_count_load()
+{
+    int           rv    = 0;
+    sqlite3_stmt *stmt  = NULL;
+    char *        query = "select count(*) from tags";
+
+    if (SQLITE_OK == sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL) &&
+        SQLITE_ROW == sqlite3_step(stmt)) {
+        pthread_rwlock_wrlock(&global_rwlock);
+        global_tag_count = sqlite3_column_int(stmt, 0);
+        pthread_rwlock_unlock(&global_rwlock);
+    } else {
+        rv = -1;
+    }
+
+    sqlite3_finalize(stmt);
+    return rv;
+}
+
 int neu_persister_create(const char *schema_dir)
 {
     int rv = sqlite3_open(db_file, &global_db);
@@ -506,6 +548,12 @@ int neu_persister_create(const char *schema_dir)
         return -1;
     }
 
+    if (0 != node_count_load() || 0 != tag_count_load()) {
+        nlog_fatal("load node/tag count fail");
+        sqlite3_close(global_db);
+        return -1;
+    }
+
     return 0;
 }
 
@@ -514,6 +562,36 @@ sqlite3 *neu_persister_get_db()
     return global_db;
 }
 
+static inline void node_count_add(int n)
+{
+    pthread_rwlock_wrlock(&global_rwlock);
+    global_node_count += n;
+    pthread_rwlock_unlock(&global_rwlock);
+}
+
+int neu_persister_node_count()
+{
+    pthread_rwlock_rdlock(&global_rwlock);
+    int cnt = global_node_count;
+    pthread_rwlock_unlock(&global_rwlock);
+    return cnt;
+}
+
+static inline void tag_count_add(int n)
+{
+    pthread_rwlock_wrlock(&global_rwlock);
+    global_tag_count += n;
+    pthread_rwlock_unlock(&global_rwlock);
+}
+
+int neu_persister_tag_count()
+{
+    pthread_rwlock_rdlock(&global_rwlock);
+    int cnt = global_tag_count;
+    pthread_rwlock_unlock(&global_rwlock);
+    return cnt;
+}
+
 void neu_persister_destroy()
 {
     sqlite3_close(global_db);
@@ -521,10 +599,15 @@ void neu_persister_destroy()
 
 int neu_persister_store_node(neu_persist_node_info_t *info)
 {
-    return execute_sql(global_db,
-                       "INSERT INTO nodes (name, type, state, plugin_name) "
-                       "VALUES (%Q, %i, %i, %Q)",
-                       info->name, info->type, info->state, info->plugin_name);
+    int rv =
+        execute_sql(global_db,
+                    "INSERT INTO nodes (name, type, state, plugin_name) "
+                    "VALUES (%Q, %i, %i, %Q)",
+                    info->name, info->type, info->state, info->plugin_name);
+    if (0 == rv) {
+        node_count_add(1);
+    }
+    return rv;
 }
 
 static UT_icd node_info_icd = {
@@ -585,8 +668,13 @@ int neu_persister_delete_node(const char *node_name)
 {
     // rely on foreign key constraints to remove settings, groups, tags and
     // subscriptions
-    return execute_sql(global_db, "DELETE FROM nodes WHERE name=%Q;",
-                       node_name);
+    int rv =
+        execute_sql(global_db, "DELETE FROM nodes WHERE name=%Q;", node_name);
+    if (0 == rv) {
+        node_count_add(-1);
+        tag_count_load();
+    }
+    return rv;
 }
 
 int neu_persister_update_node(const char *node_name, const char *new_name)
@@ -716,6 +804,9 @@ int neu_persister_store_tag(const char *driver_name, const char *group_name,
                          tag->attribute, tag->precision, tag->type,
                          tag->decimal, tag->description, val_str);
 
+    if (0 == rv) {
+        tag_count_add(1);
+    }
     free(val_str);
     return rv;
 }
@@ -821,6 +912,7 @@ int neu_persister_store_tags(const char *driver_name, const char *group_name,
         goto error;
     }
 
+    tag_count_add(n);
     sqlite3_finalize(stmt);
     return 0;
 
@@ -926,10 +1018,14 @@ int neu_persister_update_tag_value(const char *         driver_name,
 int neu_persister_delete_tag(const char *driver_name, const char *group_name,
                              const char *tag_name)
 {
-    return execute_sql(
+    int rv = execute_sql(
         global_db,
         "DELETE FROM tags WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
         driver_name, group_name, tag_name);
+    if (0 == rv) {
+        tag_count_add(-1);
+    }
+    return rv;
 }
 
 int neu_persister_store_subscription(const char *app_name,
@@ -1101,9 +1197,13 @@ error:
 int neu_persister_delete_group(const char *driver_name, const char *group_name)
 {
     // rely on foreign key constraints to delete tags and subscriptions
-    return execute_sql(global_db,
-                       "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
-                       driver_name, group_name);
+    int rv = execute_sql(global_db,
+                         "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
+                         driver_name, group_name);
+    if (0 == rv) {
+        tag_count_load();
+    }
+    return rv;
 }
 
 int neu_persister_store_node_setting(const char *node_name, const char *setting)

version

@@ -1 +1 @@
-2.4.8
+2.4.11