Merge pull request #2915 from hxy7yx/main-1

feat:preserve tag order after renaming

src/argparse.c

@@ -17,6 +17,7 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
+#include <ctype.h>
 #include <errno.h>
 #include <fcntl.h>
 #include <getopt.h>
@@ -31,12 +32,15 @@
 #include <zlog.h>
 
 #include "argparse.h"
+#include "define.h"
 #include "persist/persist.h"
 #include "utils/log.h"
 #include "version.h"
 #include "json/json.h"
 #include "json/neu_json_param.h"
 
+#include <sqlite3.h>
+
 #define OPTIONAL_ARGUMENT_IS_PRESENT                             \
     ((optarg == NULL && optind < argc && argv[optind][0] != '-') \
          ? (bool) (optarg = argv[optind++])                      \
@@ -65,6 +69,9 @@ const char *usage_text =
 "    -d, --daemon         run as daemon process\n"
 "    -h, --help           show this help message\n"
 "    stop                 stop running neuron\n"
+"    node stop --node <NAME>\n"
+"                         set a south/north node state to STOPPED in the DB\n"
+"                         (offline; restart neuron manually afterwards)\n"
 "    --log                log to the stdout\n"
 "    --log_level <LEVEL>  default log level(DEBUG,NOTICE)\n"
 "    --reset-password     reset dashboard to use default password\n"
@@ -112,6 +119,120 @@ static inline int reset_password()
     return rv;
 }
 
+static inline bool file_exists(const char *const path)
+{
+    struct stat buf = { 0 };
+    return -1 != stat(path, &buf);
+}
+
+#define NEURON_DB_FILE "persistence/sqlite.db"
+
+int neu_cli_stop_node_offline(const char *node_name)
+{
+    if (NULL == node_name || '\0' == node_name[0] ||
+        strlen(node_name) >= NEU_NODE_NAME_LEN) {
+        fprintf(stderr, "node stop requires --node with 1-%d bytes\n",
+                NEU_NODE_NAME_LEN - 1);
+        return 1;
+    }
+
+    for (const unsigned char *p = (const unsigned char *) node_name; *p; ++p) {
+        if (iscntrl(*p)) {
+            fprintf(stderr, "node name must not contain control characters\n");
+            return 1;
+        }
+    }
+
+    if (!file_exists(NEURON_DB_FILE)) {
+        fprintf(stderr, "database `%s` not exists\n", NEURON_DB_FILE);
+        return 1;
+    }
+
+    if (0 != access(NEURON_DB_FILE, R_OK | W_OK)) {
+        fprintf(stderr,
+                "no read/write permission on `%s` (errno=%d: %s).\n"
+                "If Neuron was started with sudo, run this command with the "
+                "same user, e.g. `sudo ./neuron node stop --node ...`\n",
+                NEURON_DB_FILE, errno, strerror(errno));
+        return 1;
+    }
+
+    sqlite3 *db = NULL;
+    int      rc =
+        sqlite3_open_v2(NEURON_DB_FILE, &db,
+                        SQLITE_OPEN_READWRITE | SQLITE_OPEN_FULLMUTEX, NULL);
+    if (SQLITE_OK != rc) {
+        fprintf(stderr, "failed to open `%s`: %s\n", NEURON_DB_FILE,
+                db ? sqlite3_errmsg(db) : sqlite3_errstr(rc));
+        if (db) {
+            sqlite3_close(db);
+        }
+        return 1;
+    }
+
+    sqlite3_busy_timeout(db, 3000);
+
+    sqlite3_stmt *stmt       = NULL;
+    const char *  select_sql = "SELECT type FROM nodes WHERE name=? LIMIT 1";
+    rc = sqlite3_prepare_v2(db, select_sql, -1, &stmt, NULL);
+    if (SQLITE_OK != rc) {
+        fprintf(stderr, "prepare failed: %s\n", sqlite3_errmsg(db));
+        sqlite3_close(db);
+        return 1;
+    }
+    sqlite3_bind_text(stmt, 1, node_name, -1, SQLITE_TRANSIENT);
+    rc = sqlite3_step(stmt);
+    if (SQLITE_ROW != rc) {
+        fprintf(stderr, "node `%s` not found\n", node_name);
+        sqlite3_finalize(stmt);
+        sqlite3_close(db);
+        return 1;
+    }
+    int type = sqlite3_column_int(stmt, 0);
+    sqlite3_finalize(stmt);
+    if (NEU_NA_TYPE_DRIVER != type && NEU_NA_TYPE_APP != type) {
+        fprintf(stderr, "node `%s` is not a south/north node\n", node_name);
+        sqlite3_close(db);
+        return 1;
+    }
+
+    const char *update_sql =
+        "UPDATE nodes SET state=? WHERE name=? AND type IN (?,?)";
+    rc = sqlite3_prepare_v2(db, update_sql, -1, &stmt, NULL);
+    if (SQLITE_OK != rc) {
+        fprintf(stderr, "prepare update failed: %s\n", sqlite3_errmsg(db));
+        sqlite3_close(db);
+        return 1;
+    }
+    sqlite3_bind_int(stmt, 1, NEU_NODE_RUNNING_STATE_STOPPED);
+    sqlite3_bind_text(stmt, 2, node_name, -1, SQLITE_TRANSIENT);
+    sqlite3_bind_int(stmt, 3, NEU_NA_TYPE_DRIVER);
+    sqlite3_bind_int(stmt, 4, NEU_NA_TYPE_APP);
+    rc = sqlite3_step(stmt);
+    if (SQLITE_DONE != rc) {
+        fprintf(stderr,
+                "failed to update node `%s`: %s\n"
+                "If Neuron is still running, stop it first (`sudo ./neuron "
+                "stop`), then retry.\n",
+                node_name, sqlite3_errmsg(db));
+        sqlite3_finalize(stmt);
+        sqlite3_close(db);
+        return 1;
+    }
+    if (0 == sqlite3_changes(db)) {
+        fprintf(stderr, "node `%s` was not updated\n", node_name);
+        sqlite3_finalize(stmt);
+        sqlite3_close(db);
+        return 1;
+    }
+    sqlite3_finalize(stmt);
+    sqlite3_close(db);
+
+    printf("node `%s` state set to STOPPED (uid=%u); restart neuron manually\n",
+           node_name, (unsigned) getuid());
+    return 0;
+}
+
 static inline size_t parse_restart_policy(const char *s, size_t *out)
 {
     if (0 == strcmp(s, "always")) {
@@ -135,17 +256,14 @@ static inline size_t parse_restart_policy(const char *s, size_t *out)
     return 0;
 }
 
-static inline bool file_exists(const char *const path)
-{
-    struct stat buf = { 0 };
-    return -1 != stat(path, &buf);
-}
-
 static inline int load_spec_arg(int argc, char *argv[], neu_cli_args_t *args)
 {
     int ret = 0;
     if (argc > 1 && strcmp(argv[1], "stop") == 0) {
         args->stop = true;
+    } else if (argc > 2 && strcmp(argv[1], "node") == 0 &&
+               strcmp(argv[2], "stop") == 0) {
+        args->node_stop = true;
     }
     return ret;
 }
@@ -450,6 +568,7 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
         { "syslog_host", required_argument, NULL, 'S' },
         { "syslog_port", required_argument, NULL, 'P' },
         { "sub_filter_error", no_argument, NULL, 'f' },
+        { "node", required_argument, NULL, 'n' },
         { NULL, 0, NULL, 0 },
     };
 
@@ -546,6 +665,10 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
         case 'f':
             args->sub_filter_err = true;
             break;
+        case 'n':
+            free(args->node_name);
+            args->node_name = strdup(optarg);
+            break;
         case '?':
         default:
             usage();
@@ -596,6 +719,18 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
         goto quit;
     }
 
+    if (args->node_stop) {
+        if (args->node_name == NULL || args->node_name[0] == '\0') {
+            fprintf(stderr, "node stop requires --node <NAME>\n");
+            ret = 1;
+            goto quit;
+        }
+    } else if (args->node_name != NULL) {
+        fprintf(stderr, "--node requires `node stop`\n");
+        ret = 1;
+        goto quit;
+    }
+
     if (log_level != NULL) {
         if (strcmp(log_level, "DEBUG") == 0) {
             default_log_level = ZLOG_LEVEL_DEBUG;
@@ -619,6 +754,7 @@ void neu_cli_args_fini(neu_cli_args_t *args)
         free(args->log_init_file);
         free(args->config_dir);
         free(args->plugin_dir);
+        free(args->node_name);
         free(args->ip);
         free(args->syslog_host);
     }

src/argparse.h

@@ -61,6 +61,8 @@ typedef struct {
     char *   config_dir;
     char *   plugin_dir;
     bool     stop;
+    bool     node_stop;
+    char *   node_name;
     char *   ip;
     int      port;
     char *   syslog_host;
@@ -72,6 +74,9 @@ typedef struct {
  */
 void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[]);
 
+/** Offline: set node state to STOPPED in sqlite.db. */
+int neu_cli_stop_node_offline(const char *node_name);
+
 /** Clean up the command line arguments.
  */
 void neu_cli_args_fini(neu_cli_args_t *args);

src/base/group.c

@@ -29,6 +29,7 @@ typedef struct tag_elem {
     char *name;
 
     neu_datatag_t *tag;
+    uint64_t       order;
 
     UT_hash_handle hh;
 } tag_elem_t;
@@ -37,15 +38,22 @@ struct neu_group {
     char *name;
 
     tag_elem_t *tags;
+    uint64_t    next_tag_order;
     uint32_t    interval;
 
     int64_t         timestamp;
     pthread_mutex_t mtx;
 };
 
+static int       tag_order_cmp(tag_elem_t *left, tag_elem_t *right);
 static UT_array *to_array(tag_elem_t *tags);
 static void      update_timestamp(neu_group_t *group);
 
+static int tag_order_cmp(tag_elem_t *left, tag_elem_t *right)
+{
+    return (left->order > right->order) - (left->order < right->order);
+}
+
 neu_group_t *neu_group_new(const char *name, uint32_t interval)
 {
     neu_group_t *group = calloc(1, sizeof(neu_group_t));
@@ -166,6 +174,7 @@ int neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag)
         return NEU_ERR_EINTERNAL;
     }
 
+    el->order = group->next_tag_order++;
     HASH_ADD_STR(group->tags, name, el);
     update_timestamp(group);
     pthread_mutex_unlock(&group->mtx);
@@ -232,7 +241,8 @@ int neu_group_rename_tag(neu_group_t *group, const char *old_name,
     el->name = new_el_name;
     free(el->tag->name);
     el->tag->name = new_tag_name;
-    HASH_ADD_STR(group->tags, name, el);
+    HASH_ADD_KEYPTR_INORDER(hh, group->tags, el->name, strlen(el->name), el,
+                            tag_order_cmp);
 
     update_timestamp(group);
     ret = NEU_ERR_SUCCESS;

src/main.c

@@ -173,6 +173,12 @@ int main(int argc, char *argv[])
     global_timestamp = neu_time_ms();
     neu_cli_args_init(&args, argc, argv);
 
+    if (args.node_stop) {
+        rv = neu_cli_stop_node_offline(args.node_name);
+        neu_cli_args_fini(&args);
+        return rv;
+    }
+
     disable_jwt    = args.disable_auth;
     sub_filter_err = args.sub_filter_err;
     snprintf(host_port, sizeof(host_port), "http://%s:%d", args.ip, args.port);
@@ -191,7 +197,9 @@ int main(int argc, char *argv[])
     }
 
     neuron = zlog_get_category("neuron");
-    zlog_level_switch(neuron, default_log_level);
+    if (neuron) {
+        zlog_level_switch(neuron, default_log_level);
+    }
 
     if (neu_modbus_simulator_init(args.config_dir) != 0) {
         nlog_warn("modbus simulator init failed");

tests/ft/persist/test_persist.py

@@ -5,6 +5,7 @@ import neuron.process as process
 from neuron.error import *
 import shutil
 import os
+import subprocess
 import time
 
 hold_bit = [{"name": "hold_bit", "address": "1!400001.15",
@@ -507,4 +508,44 @@ class TestPersist:
         p.stop()
 
         os.remove(dst_file)
-        os.rename(dst_backup, dst_file)
+        os.rename(dst_backup, dst_file)
+
+
+class TestOfflineNodeStopCli:
+
+    def test_offline_node_stop_persists_stopped(self):
+        process.remove_persistence()
+        p = process.NeuronProcess()
+        p.start()
+        api.add_node_check(node='cli-offline-stop', plugin=PLUGIN_MODBUS_TCP)
+        response = api.modbus_tcp_node_setting(node='cli-offline-stop',
+                                               port=60502)
+        assert 200 == response.status_code
+        p.stop()
+
+        result = subprocess.run(
+            ['./neuron', 'node', 'stop', '--node', 'cli-offline-stop'],
+            cwd='build', text=True, capture_output=True, check=False)
+        assert result.returncode == 0, result.stderr
+        assert 'STOPPED' in result.stdout
+
+        p.start()
+        try:
+            time.sleep(1)
+            response = api.get_nodes_state(node='cli-offline-stop')
+            assert 200 == response.status_code
+            assert NEU_NODE_STATE_STOPPED == response.json()['running']
+        finally:
+            p.stop()
+
+    def test_offline_node_stop_rejects_missing_node(self):
+        process.remove_persistence()
+        p = process.NeuronProcess()
+        p.start()
+        p.stop()
+
+        result = subprocess.run(
+            ['./neuron', 'node', 'stop', '--node', 'missing-node'],
+            cwd='build', text=True, capture_output=True, check=False)
+        assert result.returncode != 0
+        assert 'not found' in result.stderr

tests/ft/tag/test_tag.py

@@ -826,24 +826,39 @@ class TestTag:
         assert NEU_ERR_TAG_DECIMAL_INVALID == response.json()['error']
         assert 1 == response.json()['index']
 
-    @description(given="an existing tag", when="renaming to a new name", then="successfully renamed")
+    @description(given="three existing tags", when="renaming the middle tag", then="successfully renamed without changing tag order")
     def test_rename_tag_success(self):
-        tag = [{"name": "rename_test_tag", "address": "1!400001",
-                "attribute": NEU_TAG_ATTRIBUTE_RW, "type": NEU_TYPE_INT16}]
-        api.add_tags(node="modbus-tcp", group="group1", tags=tag)
+        tags = [
+            {"name": "rename_first_tag", "address": "1!400001",
+             "attribute": NEU_TAG_ATTRIBUTE_RW, "type": NEU_TYPE_INT16},
+            {"name": "rename_middle_tag", "address": "1!400002",
+             "attribute": NEU_TAG_ATTRIBUTE_RW, "type": NEU_TYPE_INT16},
+            {"name": "rename_last_tag", "address": "1!400003",
+             "attribute": NEU_TAG_ATTRIBUTE_RW, "type": NEU_TYPE_INT16},
+        ]
+        api.add_tags(node="modbus-tcp", group="group1", tags=tags)
 
         response = api.rename_tag(node="modbus-tcp", group="group1",
-                                  old_name="rename_test_tag", new_name="renamed_tag")
+                                  old_name="rename_middle_tag",
+                                  new_name="renamed_middle_tag")
         assert 200 == response.status_code
         assert NEU_ERR_SUCCESS == response.json()['error']
 
         response = api.get_tags(node="modbus-tcp", group="group1")
         assert 200 == response.status_code
         tag_names = [t['name'] for t in response.json()['tags']]
-        assert "renamed_tag" in tag_names
-        assert "rename_test_tag" not in tag_names
+        renamed_tags = [
+            name for name in tag_names
+            if name in {"rename_first_tag", "renamed_middle_tag",
+                        "rename_last_tag"}
+        ]
+        assert ["rename_first_tag", "renamed_middle_tag",
+                "rename_last_tag"] == renamed_tags
+        assert "rename_middle_tag" not in tag_names
 
-        api.del_tags(node="modbus-tcp", group="group1", tags=["renamed_tag"])
+        api.del_tags(node="modbus-tcp", group="group1",
+                     tags=["rename_first_tag", "renamed_middle_tag",
+                           "rename_last_tag"])
 
     @description(given="an existing tag", when="renaming to a name that already exists", then="fail with conflict error")
     def test_rename_tag_conflict(self):