diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index 8e31aa21..aa0e1afa 100644 --- a/include/neuron/plugin.h +++ b/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; }; diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 562d6ed2..3d854500 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/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; diff --git a/plugins/modbus/modbus_rtu.c b/plugins/modbus/modbus_rtu.c index 7321fdb6..dc826d53 100644 --- a/plugins/modbus/modbus_rtu.c +++ b/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, diff --git a/plugins/modbus/modbus_tcp.c b/plugins/modbus/modbus_tcp.c index ab8546a5..aef882c2 100644 --- a/plugins/modbus/modbus_tcp.c +++ b/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, diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index 271e8c69..6dee9334 100644 --- a/plugins/monitor/monitor.c +++ b/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; diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 5ad1ca5c..9a6f52e9 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/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 = diff --git a/plugins/restful/rest.c b/plugins/restful/rest.c index 4cff6156..30889151 100644 --- a/plugins/restful/rest.c +++ b/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; diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 6b30a988..a2550be6 100644 --- a/src/adapter/adapter.c +++ b/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]); } diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index 00154bce..bb54b265 100644 --- a/src/adapter/adapter_internal.h +++ b/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); diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index eb2b6bba..2faa5238 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -260,6 +260,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); @@ -785,6 +786,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); @@ -875,11 +878,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; @@ -903,7 +938,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; @@ -912,6 +947,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); @@ -944,6 +990,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)); diff --git a/src/adapter/driver/driver_internal.h b/src/adapter/driver/driver_internal.h index ad949153..0365e079 100644 --- a/src/adapter/driver/driver_internal.h +++ b/src/adapter/driver/driver_internal.h @@ -52,6 +52,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, diff --git a/src/adapter/storage.c b/src/adapter/storage.c index 6b529977..ed69c2b0 100644 --- a/src/adapter/storage.c +++ b/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); diff --git a/src/core/manager.c b/src/core/manager.c index 1be9e8e6..ec88a3d6 100644 --- a/src/core/manager.c +++ b/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) { @@ -1159,7 +1159,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); @@ -1182,7 +1182,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) { diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 813ff014..8477fbac 100644 --- a/src/core/manager_internal.c +++ b/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); @@ -524,7 +527,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; } diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 4f9b59bf..8648185c 100644 --- a/src/core/manager_internal.h +++ b/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); diff --git a/src/core/storage.c b/src/core/storage.c index 06e608af..46a1311c 100644 --- a/src/core/storage.c +++ b/src/core/storage.c @@ -375,7 +375,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,