GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-23 10:22:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(persistence): migrate tags persistence to sqlite

Browse Source
This commit is contained in:
eeff
2022-07-14 11:31:23 +08:00
parent dc8e69fea0
commit 384f231da5
5 changed files with 158 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+12 -17
View File
@@ -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);
src/adapter/storage.c
+25 -17
View File
@@ -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);
src/adapter/storage.h
+6 -2
View File
@@ -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);
src/persist/persist.c
+77 -129
View File
@@ -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,
src/persist/persist.h
+38 -25
View File
@@ -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.