From 384f231da5d840ed8c6399180edd3fccdf56e542 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Thu, 14 Jul 2022 11:31:23 +0800 Subject: [PATCH] feat(persistence): migrate tags persistence to sqlite --- src/adapter/adapter.c | 29 +++--- src/adapter/storage.c | 42 +++++---- src/adapter/storage.h | 8 +- src/persist/persist.c | 206 ++++++++++++++++-------------------------- src/persist/persist.h | 63 ++++++++----- 5 files changed, 158 insertions(+), 190 deletions(-) diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index de0078b6..ef9dc621 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -466,8 +466,12 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) error.error = NEU_ERR_GROUP_NOT_ALLOW; } else { for (int i = 0; i < cmd->n_tag; i++) { - neu_adapter_driver_del_tag((neu_adapter_driver_t *) adapter, - cmd->group, cmd->tags[i]); + int ret = neu_adapter_driver_del_tag( + (neu_adapter_driver_t *) adapter, cmd->group, cmd->tags[i]); + if (0 == ret) { + adapter_storage_del_tag(adapter->persister, cmd->driver, + cmd->group, cmd->tags[i]); + } } } @@ -476,11 +480,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } free(cmd->tags); - if (error.error == NEU_ERR_SUCCESS) { - adapter_storage_tag(adapter->persister, adapter, cmd->driver, - cmd->group); - } - neu_msg_exchange(header); header->type = NEU_RESP_ERROR; reply(adapter, header, &error); @@ -498,7 +497,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter, cmd->group, &cmd->tags[i]); if (ret == 0) { + adapter_storage_add_tag(adapter->persister, cmd->driver, + cmd->group, &cmd->tags[i]); resp.index += 1; + } else { resp.error = ret; break; @@ -512,11 +514,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } free(cmd->tags); - if (resp.index > 0) { - adapter_storage_tag(adapter->persister, adapter, cmd->driver, - cmd->group); - } - neu_msg_exchange(header); header->type = NEU_RESP_ADD_TAG; reply(adapter, header, &resp); @@ -534,6 +531,9 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) (neu_adapter_driver_t *) adapter, cmd->group, &cmd->tags[i]); if (ret == 0) { + adapter_storage_update_tag(adapter->persister, cmd->driver, + cmd->group, &cmd->tags[i]); + resp.index += 1; } else { resp.error = ret; @@ -548,11 +548,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } free(cmd->tags); - if (resp.index > 0) { - adapter_storage_tag(adapter->persister, adapter, cmd->driver, - cmd->group); - } - neu_msg_exchange(header); header->type = NEU_RESP_UPDATE_TAG; reply(adapter, header, &resp); diff --git a/src/adapter/storage.c b/src/adapter/storage.c index b11e90a2..07c54674 100644 --- a/src/adapter/storage.c +++ b/src/adapter/storage.c @@ -46,24 +46,33 @@ void adapter_storage_del_group(neu_persister_t *persister, const char *node, neu_persister_delete_group(persister, node, group); } -void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter, - const char *node, const char *group) +void adapter_storage_add_tag(neu_persister_t *persister, const char *node, + const char *group, const neu_datatag_t *tag) { - UT_array *tags = NULL; - - int rv = neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, group, - &tags); - if (rv != 0) { - nlog_warn("fail get tag infos"); - return; - } - - rv = neu_persister_store_datatags(persister, node, group, tags); + int rv = neu_persister_store_tag(persister, node, group, tag); if (0 != rv) { - nlog_error("fail store adapter:%s grp:%s datatag infos", node, group); + nlog_error("fail store tag:%s adapter:%s grp:%s", tag->name, node, + group); } +} - neu_persist_datatag_infos_free(tags); +void adapter_storage_update_tag(neu_persister_t *persister, const char *node, + const char *group, const neu_datatag_t *tag) +{ + int rv = neu_persister_update_tag(persister, node, group, tag); + if (0 != rv) { + nlog_error("fail update tag:%s adapter:%s grp:%s", tag->name, node, + group); + } +} + +void adapter_storage_del_tag(neu_persister_t *persister, const char *node, + const char *group, const char *name) +{ + int rv = neu_persister_delete_tag(persister, node, group, name); + if (0 != rv) { + nlog_error("fail del tag:%s adapter:%s grp:%s", name, node, group); + } } int adapter_load_setting(neu_persister_t *persister, const char *node, @@ -96,8 +105,7 @@ int adapter_load_group_and_tag(neu_persister_t * persister, UT_array *tags = NULL; neu_adapter_driver_add_group(driver, p->name, p->interval); - rv = neu_persister_load_datatags(persister, adapter->name, p->name, - &tags); + rv = neu_persister_load_tags(persister, adapter->name, p->name, &tags); if (0 != rv) { nlog_warn("load %s:%s tags fail", adapter->name, p->name); continue; @@ -108,7 +116,7 @@ int adapter_load_group_and_tag(neu_persister_t * persister, neu_adapter_driver_add_tag(driver, p->name, tag); } - neu_persist_datatag_infos_free(tags); + utarray_free(tags); } utarray_free(group_infos); diff --git a/src/adapter/storage.h b/src/adapter/storage.h index e09b0ea1..24549e58 100644 --- a/src/adapter/storage.h +++ b/src/adapter/storage.h @@ -30,8 +30,12 @@ void adapter_storage_add_group(neu_persister_t *persister, const char *node, const char *group, uint32_t interval); void adapter_storage_del_group(neu_persister_t *persister, const char *node, const char *group); -void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter, - const char *node, const char *group); +void adapter_storage_add_tag(neu_persister_t *persister, const char *node, + const char *group, const neu_datatag_t *tag); +void adapter_storage_update_tag(neu_persister_t *persister, const char *node, + const char *group, const neu_datatag_t *tag); +void adapter_storage_del_tag(neu_persister_t *persister, const char *node, + const char *group, const char *name); int adapter_load_setting(neu_persister_t *persister, const char *node, char **setting); diff --git a/src/persist/persist.c b/src/persist/persist.c index 30366af9..46aa8474 100644 --- a/src/persist/persist.c +++ b/src/persist/persist.c @@ -194,42 +194,6 @@ static inline int create_dir(char *dir_name) return rv; } -/** - * Create a directory and any intermediate directory if necessary. - */ -static int create_dir_recursive(const char *path) -{ - int rv = 0; - char p[PATH_MAX_SIZE] = { 0 }; - - rv = snprintf(p, sizeof(p), "%s", path); - if (sizeof(p) == rv) { - return -1; - } - - int i = 0; - while (1) { - while (p[i] && PATH_SEP_CHAR == p[i]) { - ++i; - } - if ('\0' == p[i]) { - break; - } - while (p[i] && PATH_SEP_CHAR != p[i]) { - ++i; - } - - char c = p[i]; - p[i] = '\0'; - rv = create_dir(p); - if (0 != rv) { - break; - } - p[i] = c; - } - return rv; -} - /** * Ensure that a file with the given name exists. * @@ -657,109 +621,93 @@ int neu_persister_load_plugins(neu_persister_t *persister, return 0; } -int neu_persister_store_datatags(neu_persister_t *persister, - const char * adapter_name, - const char * group_config_name, - UT_array * datatag_infos) +int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name, + const char *group_name, const neu_datatag_t *tag) { - char path[PATH_MAX_SIZE] = { 0 }; - - int n = persister_group_config_dir(path, sizeof(path), persister, - adapter_name, group_config_name); - if (sizeof(path) == n) { - nlog_error("path too long: %s", path); - return -1; - } - - int rv = create_dir_recursive(path); - if (0 != rv) { - nlog_error("fail to create dir: %s", path); - return -1; - } - - n = path_cat(path, n, sizeof(path), "datatags.json"); - if (sizeof(path) == n) { - nlog_error("path too long: %s", path); - return -1; - } - - neu_json_datatag_req_t datatags_resp = { 0 }; - datatags_resp.n_tag = utarray_len(datatag_infos); - datatags_resp.tags = - calloc(datatags_resp.n_tag, sizeof(neu_json_datatag_req_tag_t)); - int index = 0; - utarray_foreach(datatag_infos, neu_datatag_t *, tag) - { - datatags_resp.tags[index].name = tag->name; - datatags_resp.tags[index].description = tag->description; - datatags_resp.tags[index].address = tag->address; - datatags_resp.tags[index].attribute = tag->attribute; - datatags_resp.tags[index].type = tag->type; - index += 1; - } - - char *result = NULL; - rv = neu_json_encode_by_fn(&datatags_resp, neu_json_encode_datatag_resp, - &result); - - if (rv == 0) { - write_file_string(path, result); - } - - free(result); - free(datatags_resp.tags); - - return rv; + return neu_persister_sql_exec( + persister, + "INSERT INTO tags (driver_name, group_name, name, address, attribute, " + "type, description) VALUES(%Q, %Q, %Q, %Q, %i, %i, %Q)", + driver_name, group_name, tag->name, tag->address, tag->attribute, + tag->type, tag->description); } -int neu_persister_load_datatags(neu_persister_t *persister, - const char * adapter_name, - const char *group_config_name, UT_array **tags) +int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name, + const char *group_name, UT_array **tags) { - char path[PATH_MAX_SIZE] = { 0 }; - - int n = persister_group_config_dir(path, sizeof(path), persister, - adapter_name, group_config_name); - if (sizeof(path) == n) { - nlog_error("path too long: %s", path); - return -1; - } - - n = path_cat(path, n, sizeof(path), "datatags.json"); - if (sizeof(path) == n) { - nlog_error("path too long: %s", path); - return -1; - } - - char *json_str = NULL; - int rv = read_file_string(path, &json_str); - if (rv != 0) { - return rv; - } - - neu_json_datatag_req_t *datatag_req = NULL; - rv = neu_json_decode_datatag_req(json_str, &datatag_req); - free(json_str); - if (rv != 0) { - return rv; - } + sqlite3_stmt *stmt = NULL; + const char * query = "SELECT name, address, attribute, type, description " + "FROM tags WHERE driver_name=? AND group_name=?"; utarray_new(*tags, neu_tag_get_icd()); - for (int i = 0; i < datatag_req->n_tag; i++) { - neu_datatag_t tag = { - .name = datatag_req->tags[i].name, - .address = datatag_req->tags[i].address, - .type = datatag_req->tags[i].type, - .attribute = datatag_req->tags[i].attribute, - .description = datatag_req->tags[i].description, - }; - utarray_push_back(*tags, &tag); + if (SQLITE_OK != + sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) { + nlog_error("prepare `%s` fail: %s", query, + sqlite3_errmsg(persister->db)); + goto error; } - neu_json_decode_datatag_req_free(datatag_req); + if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) { + nlog_error("bind `%s` with `%s` fail: %s", query, driver_name, + sqlite3_errmsg(persister->db)); + goto error; + } - return rv; + if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) { + nlog_error("bind `%s` with `%s` fail: %s", query, group_name, + sqlite3_errmsg(persister->db)); + goto error; + } + + int step = sqlite3_step(stmt); + while (SQLITE_ROW == step) { + neu_datatag_t tag = { + .name = (char *) sqlite3_column_text(stmt, 0), + .address = (char *) sqlite3_column_text(stmt, 1), + .attribute = sqlite3_column_int(stmt, 2), + .type = sqlite3_column_int(stmt, 3), + .description = (char *) sqlite3_column_text(stmt, 4), + }; + utarray_push_back(*tags, &tag); + + step = sqlite3_step(stmt); + } + + if (SQLITE_DONE != step) { + nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db)); + // do not set return code, return partial or empty result + } + + sqlite3_finalize(stmt); + return 0; + +error: + utarray_free(*tags); + *tags = NULL; + return NEU_ERR_EINTERNAL; +} + +int neu_persister_update_tag(neu_persister_t *persister, + const char *driver_name, const char *group_name, + const neu_datatag_t *tag) +{ + return neu_persister_sql_exec( + persister, + "UPDATE tags SET address=%Q, attribute=%i, type=%i, description=%Q " + "WHERE driver_name=%Q AND group_name=%Q AND name=%Q", + tag->address, tag->attribute, tag->type, tag->description, driver_name, + group_name, tag->name); +} + +int neu_persister_delete_tag(neu_persister_t *persister, + const char *driver_name, const char *group_name, + const char *tag_name) +{ + return neu_persister_sql_exec( + persister, + "DELETE FROM tags WHERE driver_name=%Q AND group_name=%Q AND name=%Q", + driver_name, group_name, tag_name); } int neu_persister_store_subscriptions(neu_persister_t *persister, diff --git a/src/persist/persist.h b/src/persist/persist.h index 764710a5..a033033b 100644 --- a/src/persist/persist.h +++ b/src/persist/persist.h @@ -25,7 +25,6 @@ extern "C" { #endif #include "adapter/adapter_info.h" -#include "persist/json/persist_json_datatags.h" #include "persist/json/persist_json_plugin.h" #include "persist/json/persist_json_subs.h" @@ -38,8 +37,6 @@ typedef struct { int64_t state; } neu_persist_node_info_t; -typedef neu_json_datatag_req_tag_t neu_persist_datatag_info_t; - typedef struct { uint32_t interval; char * name; @@ -64,11 +61,6 @@ static inline void neu_persist_group_info_fini(neu_persist_group_info_t *info) free(info->name); } -static inline void neu_persist_datatag_infos_free(UT_array *datatag_infos) -{ - utarray_free(datatag_infos); -} - static inline void neu_persist_subscription_infos_free(UT_array *subscription_infos) { @@ -153,30 +145,51 @@ int neu_persister_load_plugins(neu_persister_t *persister, UT_array ** plugin_infos); /** - * Persist adapter datatags. + * Persist node tags. * @param persister persiter object. - * @param adapter_name name of the adapter who owns the datatags. - * @param group_config_name name of the group config. - * @param datatag_infos vector of neu_persist_datatag_info_t. + * @param driver_name name of the driver who owns the tags + * @param group_name name of the group + * @param tag the tag to store * @return 0 on success, non-zero otherwise */ -int neu_persister_store_datatags(neu_persister_t *persister, - const char * adapter_name, - const char * group_config_name, - UT_array * datatag_infos); +int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name, + const char *group_name, const neu_datatag_t *tag); + /** - * Load adapter datatag infos. + * Load node tag infos. * @param persister persiter object. - * @param adapter_name name of the adapter who owns the datatags. - * @param group_config_name name of the group config. - * @param[out] datatag_infos used to return pointer to heap allocated - * vector of neu_persist_datatag_info_t. + * @param node_name name of the node who owns the tags + * @param group_name name of the group + * @param[out] tag_infos used to return pointer to heap allocated + * vector of neu_datatag_t * @return 0 on success, non-zero otherwise */ -int neu_persister_load_datatags(neu_persister_t *persister, - const char * adapter_name, - const char * group_config_name, - UT_array ** datatag_infos); +int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name, + const char *group_name, UT_array **tag_infos); + +/** + * Update node tags. + * @param persister persiter object. + * @param driver_name name of the driver who owns the tags + * @param group_name name of the group + * @param tag the tag to update + * @return 0 on success, non-zero otherwise + */ +int neu_persister_update_tag(neu_persister_t *persister, + const char *driver_name, const char *group_name, + const neu_datatag_t *tag); + +/** + * Delete node tags. + * @param persister persiter object. + * @param driver_name name of the driver who owns the tags + * @param group_name name of the group + * @param tag_name name of the tag + * @return 0 on success, non-zero otherwise + */ +int neu_persister_delete_tag(neu_persister_t *persister, + const char *driver_name, const char *group_name, + const char *tag_name); /** * Persist adapter subscriptions.