check license when add node/tag

include/neuron/plugin.h

@@ -72,9 +72,7 @@ typedef struct {
 typedef struct neu_plugin_intf_funs {
     neu_plugin_t *(*open)(void);
     int (*close)(neu_plugin_t *plugin);
-    int (*init)(neu_plugin_t *plugin); // create node by API
-    int (*load)(
-        neu_plugin_t *plugin); // crate node using data from the database
+    int (*init)(neu_plugin_t *plugin, bool load);
     int (*uninit)(neu_plugin_t *plugin);
     int (*start)(neu_plugin_t *plugin);
     int (*stop)(neu_plugin_t *plugin);
@@ -99,8 +97,7 @@ typedef struct neu_plugin_intf_funs {
             int (*add_tags)(neu_plugin_t *plugin, const char *group,
                             neu_datatag_t *tags,
                             int            n_tag); // create tags by API
-            int (*del_tags)(neu_plugin_t *plugin, const char *group,
-                            neu_datatag_t *tags, int n_tag);
+            int (*del_tags)(neu_plugin_t *plugin, int n_tag);
         } driver;
     };
 

plugins/ekuiper/plugin_ekuiper.c

@@ -73,8 +73,9 @@ static void pipe_rm_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     nng_mtx_unlock(plugin->mtx);
 }
 
-static int ekuiper_plugin_init(neu_plugin_t *plugin)
+static int ekuiper_plugin_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     int      rv       = 0;
     nng_aio *recv_aio = NULL;
 

plugins/modbus/modbus_rtu.c

@@ -29,7 +29,7 @@
 static neu_plugin_t *driver_open(void);
 
 static int driver_close(neu_plugin_t *plugin);
-static int driver_init(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);
@@ -58,6 +58,9 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
     .driver.write_tag     = driver_write,
     .driver.tag_validator = driver_tag_validator,
     .driver.write_tags    = NULL,
+    .driver.add_tags      = NULL,
+    .driver.load_tags     = NULL,
+    .driver.del_tags      = NULL,
 };
 
 const neu_plugin_module_t neu_plugin_module = {
@@ -94,8 +97,9 @@ static int driver_close(neu_plugin_t *plugin)
     return 0;
 }
 
-static int driver_init(neu_plugin_t *plugin)
+static int driver_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     plugin->protocol = MODBUS_PROTOCOL_RTU;
     plugin->events   = neu_event_new();
     plugin->stack    = modbus_stack_create((void *) plugin, MODBUS_PROTOCOL_RTU,

plugins/modbus/modbus_tcp.c

@@ -29,7 +29,7 @@
 static neu_plugin_t *driver_open(void);
 
 static int driver_close(neu_plugin_t *plugin);
-static int driver_init(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);
@@ -58,6 +58,9 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
     .driver.write_tag     = driver_write,
     .driver.tag_validator = driver_tag_validator,
     .driver.write_tags    = NULL,
+    .driver.add_tags      = NULL,
+    .driver.load_tags     = NULL,
+    .driver.del_tags      = NULL,
 };
 
 const neu_plugin_module_t neu_plugin_module = {
@@ -94,8 +97,9 @@ static int driver_close(neu_plugin_t *plugin)
     return 0;
 }
 
-static int driver_init(neu_plugin_t *plugin)
+static int driver_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     plugin->protocol = MODBUS_PROTOCOL_TCP;
     plugin->events   = neu_event_new();
     plugin->stack    = modbus_stack_create((void *) plugin, MODBUS_PROTOCOL_TCP,

plugins/monitor/monitor.c

@@ -292,10 +292,11 @@ static int monitor_plugin_close(neu_plugin_t *plugin)
     return rv;
 }
 
-static int monitor_plugin_init(neu_plugin_t *plugin)
+static int monitor_plugin_init(neu_plugin_t *plugin, bool load)
 {
     int rv = 0;
     (void) plugin;
+    (void) load;
 
     plog_notice(plugin, "Initialize plugin: %s", neu_plugin_module.module_name);
     return rv;

plugins/mqtt/mqtt_plugin.c

@@ -57,8 +57,9 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
     return NEU_ERR_SUCCESS;
 }
 
-static int mqtt_plugin_init(neu_plugin_t *plugin)
+static int mqtt_plugin_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     plog_notice(plugin, "initialize plugin `%s` success",
                 neu_plugin_module.module_name);
     neu_adapter_register_metric_cb_t register_metric =

plugins/restful/rest.c

@@ -130,8 +130,9 @@ static int dashb_plugin_close(neu_plugin_t *plugin)
     return rv;
 }
 
-static int dashb_plugin_init(neu_plugin_t *plugin)
+static int dashb_plugin_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     int rv = 0;
 
     (void) plugin;

src/adapter/adapter.c

@@ -65,6 +65,8 @@ static const adapter_callbacks_t callback_funs = {
     .update_metric   = adapter_update_metric,
 };
 
+static __thread int create_adapter_error = 0;
+
 #define REGISTER_METRIC(adapter, name, init) \
     adapter_register_metric(adapter, name, name##_HELP, name##_TYPE, init);
 
@@ -108,9 +110,20 @@ static inline void stop_log_level_timer(neu_adapter_t *adapter)
     adapter->timer_lev = NULL;
 }
 
-neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
+int neu_adapter_error()
+{
+    return create_adapter_error;
+}
+
+void neu_adapter_set_error(int error)
+{
+    create_adapter_error = error;
+}
+
+neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
 {
     int                  rv      = 0;
+    int                  init_rv = 0;
     neu_adapter_t *      adapter = NULL;
     neu_event_io_param_t param   = { 0 };
 
@@ -170,7 +183,8 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
 
     zlog_level_switch(common->log, default_log_level);
 
-    adapter->module->intf_funs->init(adapter->plugin);
+    init_rv = adapter->module->intf_funs->init(adapter->plugin, load);
+
     if (adapter_load_setting(adapter->name, &adapter->setting) == 0) {
         if (adapter->module->intf_funs->setting(adapter->plugin,
                                                 adapter->setting) == 0) {
@@ -198,7 +212,23 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     nlog_notice("Success to create adapter: %s", adapter->name);
 
     adapter_storage_state(adapter->name, adapter->state);
-    return adapter;
+
+    if (init_rv != 0) {
+        nlog_warn("Failed to init adapter: %s", adapter->name);
+        neu_adapter_set_error(init_rv);
+
+        if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
+            neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter);
+        }
+        stop_log_level_timer(adapter);
+        neu_event_del_io(adapter->events, adapter->nng_io);
+
+        neu_adapter_destroy(adapter);
+        return NULL;
+    } else {
+        neu_adapter_set_error(0);
+        return adapter;
+    }
 }
 
 int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name)
@@ -782,9 +812,9 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
             for (int i = 0; i < cmd->n_tag; i++) {
-                int ret =
-                    neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter,
-                                               cmd->group, &cmd->tags[i]);
+                int ret = neu_adapter_driver_validate_tag(
+                    (neu_adapter_driver_t *) adapter, cmd->group,
+                    &cmd->tags[i]);
                 if (ret == 0) {
                     resp.index += 1;
                 } else {
@@ -796,6 +826,28 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
             resp.error = NEU_ERR_GROUP_NOT_ALLOW;
         }
 
+        if (resp.index > 0) {
+            int ret = neu_adapter_driver_try_add_tag(
+                (neu_adapter_driver_t *) adapter, cmd->group, cmd->tags,
+                resp.index);
+            if (ret != 0) {
+                resp.index = 0;
+                resp.error = ret;
+            }
+        }
+
+        for (int i = 0; i < resp.index; i++) {
+            int ret = neu_adapter_driver_add_tag(
+                (neu_adapter_driver_t *) adapter, cmd->group, &cmd->tags[i]);
+            if (ret != 0) {
+                neu_adapter_driver_try_del_tag((neu_adapter_driver_t *) adapter,
+                                               resp.index - i);
+                resp.index = i;
+                resp.error = ret;
+                break;
+            }
+        }
+
         for (uint16_t i = resp.index; i < cmd->n_tag; i++) {
             neu_tag_fini(&cmd->tags[i]);
         }

src/adapter/adapter_internal.h

@@ -63,7 +63,10 @@ typedef struct adapter_msg_handler {
     adapter_handler    handler;
 } adapter_msg_handler_t;
 
-neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info);
+int  neu_adapter_error();
+void neu_adapter_set_error(int error);
+
+neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load);
 void neu_adapter_init(neu_adapter_t *adapter, neu_node_running_state_e state);
 
 int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name);

src/adapter/driver/driver.c

@@ -242,6 +242,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
     {
         HASH_DEL(driver->groups, el);
 
+        neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(el->group));
         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);
@@ -733,6 +734,8 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
     if (find != NULL) {
         HASH_DEL(driver->groups, find);
 
+        neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(find->group));
+
         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);
@@ -823,11 +826,43 @@ UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver)
     return groups;
 }
 
-int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
-                               neu_datatag_t *tag)
+int neu_adapter_driver_try_add_tag(neu_adapter_driver_t *driver,
+                                   const char *group, neu_datatag_t *tags,
+                                   int n_tag)
 {
-    int      ret  = NEU_ERR_SUCCESS;
-    group_t *find = NULL;
+    int ret = 0;
+    if (driver->adapter.module->intf_funs->driver.add_tags != NULL) {
+        ret = driver->adapter.module->intf_funs->driver.add_tags(
+            driver->adapter.plugin, group, tags, n_tag);
+    }
+    return ret;
+}
+
+int neu_adapter_driver_load_tag(neu_adapter_driver_t *driver, const char *group,
+                                neu_datatag_t *tags, int n_tag)
+{
+    int ret = 0;
+    if (driver->adapter.module->intf_funs->driver.load_tags != NULL) {
+        ret = driver->adapter.module->intf_funs->driver.load_tags(
+            driver->adapter.plugin, group, tags, n_tag);
+    }
+    return ret;
+}
+
+int neu_adapter_driver_try_del_tag(neu_adapter_driver_t *driver, int n_tag)
+{
+    int ret = 0;
+    if (driver->adapter.module->intf_funs->driver.del_tags != NULL) {
+        ret = driver->adapter.module->intf_funs->driver.del_tags(
+            driver->adapter.plugin, n_tag);
+    }
+    return ret;
+}
+
+int neu_adapter_driver_validate_tag(neu_adapter_driver_t *driver,
+                                    const char *group, neu_datatag_t *tag)
+{
+    (void) group;
 
     if (strlen(tag->name) >= NEU_TAG_NAME_LEN) {
         return NEU_ERR_TAG_NAME_TOO_LONG;
@@ -851,7 +886,7 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
     }
 
     if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC)) {
-        ret = driver->adapter.module->intf_funs->driver.validate_tag(
+        int ret = driver->adapter.module->intf_funs->driver.validate_tag(
             driver->adapter.plugin, tag);
         if (ret != NEU_ERR_SUCCESS) {
             return ret;
@@ -860,6 +895,17 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
 
     neu_datatag_parse_addr_option(tag, &tag->option);
 
+    return NEU_ERR_SUCCESS;
+}
+
+int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
+                               neu_datatag_t *tag)
+{
+    int      ret  = NEU_ERR_SUCCESS;
+    group_t *find = NULL;
+
+    neu_datatag_parse_addr_option(tag, &tag->option);
+
     HASH_FIND_STR(driver->groups, group, find);
     if (find == NULL) {
         neu_adapter_driver_add_group(driver, group, 3000);
@@ -892,6 +938,7 @@ int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group,
     }
 
     if (ret == NEU_ERR_SUCCESS) {
+        neu_adapter_driver_try_del_tag(driver, 1);
         neu_adapter_update_group_metric(&driver->adapter, group,
                                         NEU_METRIC_GROUP_TAGS_TOTAL,
                                         neu_group_tag_size(find->group));

src/adapter/driver/driver_internal.h

@@ -51,6 +51,16 @@ int neu_adapter_driver_group_exist(neu_adapter_driver_t *driver,
                                    const char *          name);
 UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver);
 
+int neu_adapter_driver_try_del_tag(neu_adapter_driver_t *driver, int n_tag);
+int neu_adapter_driver_try_add_tag(neu_adapter_driver_t *driver,
+                                   const char *group, neu_datatag_t *tags,
+                                   int n_tag);
+int neu_adapter_driver_load_tag(neu_adapter_driver_t *driver, const char *group,
+                                neu_datatag_t *tags, int n_tag);
+
+int neu_adapter_driver_validate_tag(neu_adapter_driver_t *driver,
+                                    const char *group, neu_datatag_t *tag);
+
 int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
                                neu_datatag_t *tag);
 int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group,

src/adapter/storage.c

@@ -157,6 +157,7 @@ int adapter_load_group_and_tag(neu_adapter_driver_t *driver)
         utarray_foreach(tags, neu_datatag_t *, tag)
         {
             neu_adapter_driver_add_tag(driver, p->name, tag);
+            neu_adapter_driver_load_tag(driver, p->name, tag, 1);
         }
 
         utarray_free(tags);

src/core/manager.c

@@ -564,7 +564,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_ADD_NODE: {
         neu_req_add_node_t *cmd = (neu_req_add_node_t *) &header[1];
         int                 error =
-            neu_manager_add_node(manager, cmd->node, cmd->plugin, false);
+            neu_manager_add_node(manager, cmd->node, cmd->plugin, false, false);
         neu_resp_error_t e = { .error = error };
 
         if (error == NEU_ERR_SUCCESS) {
@@ -1096,7 +1096,7 @@ static void start_static_adapter(neu_manager_t *manager, const char *name)
     adapter_info.handle = instance.handle;
     adapter_info.module = instance.module;
 
-    adapter = neu_adapter_create(&adapter_info);
+    adapter = neu_adapter_create(&adapter_info, true);
     neu_node_manager_add_static(manager->node_manager, adapter);
     neu_adapter_init(adapter, false);
     neu_adapter_start(adapter);
@@ -1119,7 +1119,7 @@ static void start_single_adapter(neu_manager_t *manager, const char *name,
 
     adapter_info.handle = instance.handle;
     adapter_info.module = instance.module;
-    adapter             = neu_adapter_create(&adapter_info);
+    adapter             = neu_adapter_create(&adapter_info, true);
 
     neu_node_manager_add_single(manager->node_manager, adapter, display);
     if (display) {

src/core/manager_internal.c

@@ -46,7 +46,7 @@ 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,
-                         neu_node_running_state_e state)
+                         neu_node_running_state_e state, bool load)
 {
     neu_adapter_t *       adapter      = NULL;
     neu_plugin_instance_t instance     = { 0 };
@@ -78,7 +78,10 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
     adapter_info.handle = instance.handle;
     adapter_info.module = instance.module;
 
-    adapter = neu_adapter_create(&adapter_info);
+    adapter = neu_adapter_create(&adapter_info, load);
+    if (adapter == NULL) {
+        return neu_adapter_error();
+    }
     neu_node_manager_add(manager->node_manager, adapter);
     neu_adapter_init(adapter, state);
 
@@ -517,7 +520,7 @@ int neu_manager_instantiate_template(neu_manager_t *          manager,
     }
 
     int ret = neu_manager_add_node(manager, req->node,
-                                   neu_template_plugin(tmpl), false);
+                                   neu_template_plugin(tmpl), false, false);
     if (0 != ret) {
         return ret;
     }

src/core/manager_internal.h

@@ -53,7 +53,7 @@ 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,
-                               neu_node_running_state_e state);
+                               neu_node_running_state_e state, bool load);
 int       neu_manager_del_node(neu_manager_t *manager, const char *node_name);
 UT_array *neu_manager_get_nodes(neu_manager_t *manager, int type,
                                 const char *plugin, const char *node);

src/core/storage.c

@@ -363,7 +363,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);
+                                  node_info->plugin_name, node_info->state,
+                                  true);
         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,