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

Merge branch 'emqx:main' into main

Browse Source
This commit is contained in:
hxy7yx
2023-05-04 09:34:11 +08:00
committed by GitHub
parent c6aadcbc4f af89e3d21d
commit 209eebdd90
19 changed files with 1235 additions and 338 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+37 -14
View File
@@ -94,6 +94,10 @@ typedef enum neu_reqresp_type {
NEU_RESP_GET_TEMPLATE,
NEU_REQ_GET_TEMPLATES,
NEU_RESP_GET_TEMPLATES,
NEU_REQ_ADD_TEMPLATE_GROUP,
NEU_REQ_DEL_TEMPLATE_GROUP,
NEU_REQ_UPDATE_TEMPLATE_GROUP,
NEU_REQ_GET_TEMPLATE_GROUP,
NEU_REQ_ADD_TEMPLATE_TAG,
NEU_RESP_ADD_TEMPLATE_TAG,
NEU_REQ_DEL_TEMPLATE_TAG,
@@ -173,20 +177,24 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_GET_PLUGIN] = "NEU_REQ_GET_PLUGIN",
[NEU_RESP_GET_PLUGIN] = "NEU_RESP_GET_PLUGIN",
[NEU_REQ_ADD_TEMPLATE] = "NEU_REQ_ADD_TEMPLATE",
[NEU_REQ_DEL_TEMPLATE] = "NEU_REQ_DEL_TEMPLATE",
[NEU_REQ_GET_TEMPLATE] = "NEU_REQ_GET_TEMPLATE",
[NEU_RESP_GET_TEMPLATE] = "NEU_RESP_GET_TEMPLATE",
[NEU_REQ_GET_TEMPLATES] = "NEU_REQ_GET_TEMPLATES",
[NEU_RESP_GET_TEMPLATES] = "NEU_RESP_GET_TEMPLATES",
[NEU_REQ_ADD_TEMPLATE_TAG] = "REQ_REQ_ADD_TEMPLATE_TAG",
[NEU_RESP_ADD_TEMPLATE_TAG] = "REQ_RESP_ADD_TEMPLATE_TAG",
[NEU_REQ_DEL_TEMPLATE_TAG] = "REQ_REQ_DEL_TEMPLATE_TAG",
[NEU_REQ_UPDATE_TEMPLATE_TAG] = "REQ_REQ_UPDATE_TEMPLATE_TAG",
[NEU_RESP_UPDATE_TEMPLATE_TAG] = "REQ_RESP_UPDATE_TEMPLATE_TAG",
[NEU_REQ_GET_TEMPLATE_TAG] = "REQ_REQ_GET_TEMPLATE_TAG",
[NEU_RESP_GET_TEMPLATE_TAG] = "REQ_RESP_GET_TEMPLATE_TAG",
[NEU_REQ_INST_TEMPLATE] = "NEU_REQ_INST_TEMPLATE",
[NEU_REQ_ADD_TEMPLATE] = "NEU_REQ_ADD_TEMPLATE",
[NEU_REQ_DEL_TEMPLATE] = "NEU_REQ_DEL_TEMPLATE",
[NEU_REQ_GET_TEMPLATE] = "NEU_REQ_GET_TEMPLATE",
[NEU_RESP_GET_TEMPLATE] = "NEU_RESP_GET_TEMPLATE",
[NEU_REQ_GET_TEMPLATES] = "NEU_REQ_GET_TEMPLATES",
[NEU_RESP_GET_TEMPLATES] = "NEU_RESP_GET_TEMPLATES",
[NEU_REQ_ADD_TEMPLATE_GROUP] = "NEU_REQ_ADD_TEMPLATE_GROUP",
[NEU_REQ_DEL_TEMPLATE_GROUP] = "NEU_REQ_DEL_TEMPLATE_GROUP",
[NEU_REQ_UPDATE_TEMPLATE_GROUP] = "NEU_REQ_UPDATE_TEMPLATE_GROUP",
[NEU_REQ_GET_TEMPLATE_GROUP] = "NEU_REQ_GET_TEMPLATE_GROUP",
[NEU_REQ_ADD_TEMPLATE_TAG] = "REQ_REQ_ADD_TEMPLATE_TAG",
[NEU_RESP_ADD_TEMPLATE_TAG] = "REQ_RESP_ADD_TEMPLATE_TAG",
[NEU_REQ_DEL_TEMPLATE_TAG] = "REQ_REQ_DEL_TEMPLATE_TAG",
[NEU_REQ_UPDATE_TEMPLATE_TAG] = "REQ_REQ_UPDATE_TEMPLATE_TAG",
[NEU_RESP_UPDATE_TEMPLATE_TAG] = "REQ_RESP_UPDATE_TEMPLATE_TAG",
[NEU_REQ_GET_TEMPLATE_TAG] = "REQ_REQ_GET_TEMPLATE_TAG",
[NEU_RESP_GET_TEMPLATE_TAG] = "REQ_RESP_GET_TEMPLATE_TAG",
[NEU_REQ_INST_TEMPLATE] = "NEU_REQ_INST_TEMPLATE",
[NEU_REQRESP_TRANS_DATA] = "NEU_REQRESP_TRANS_DATA",
[NEU_REQRESP_NODES_STATE] = "NEU_REQRESP_NODES_STATE",
@@ -506,6 +514,21 @@ static inline void neu_resp_get_templates_fini(neu_resp_get_templates_t *resp)
free(resp->templates);
}
typedef struct {
char tmpl[NEU_TEMPLATE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
uint32_t interval;
} neu_req_add_template_group_t, neu_req_update_template_group_t;
typedef struct {
char tmpl[NEU_TEMPLATE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
} neu_req_del_template_group_t;
typedef struct {
char tmpl[NEU_TEMPLATE_NAME_LEN];
} neu_req_get_template_group_t;
typedef struct {
char tmpl[NEU_TEMPLATE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
include/neuron/connection/neu_connection.h
+11
View File
@@ -30,6 +30,7 @@ typedef enum neu_conn_type {
NEU_CONN_TCP_CLIENT,
NEU_CONN_UDP,
NEU_CONN_TTY_CLIENT,
NEU_CONN_UDP_TO,
} neu_conn_type_e;
typedef enum neu_conn_tty_baud {
@@ -103,6 +104,12 @@ typedef struct neu_conn_param {
uint16_t timeout; // millisecond
} udp;
struct {
char * src_ip;
uint16_t src_port;
uint16_t timeout; // millisecond
} udpto;
struct {
char * device;
neu_conn_tty_data_e data;
@@ -231,6 +238,8 @@ int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd);
* less than or equal to 0 fails.
*/
ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len);
ssize_t neu_conn_udp_sendto(neu_conn_t *conn, uint8_t *buf, ssize_t len,
void *dst);
/**
* @brief Receive data via connection read.
@@ -242,6 +251,8 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len);
* received, less than or equal to 0 fails.
*/
ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len);
ssize_t neu_conn_udp_recvfrom(neu_conn_t *conn, uint8_t *buf, ssize_t len,
void *src);
/**
* @brief Specify the client to send data.
include/neuron/connection/neu_connection_eth.h
+2 -6
View File
@@ -32,19 +32,15 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx);
int neu_conn_eth_uninit(neu_conn_eth_t *conn);
int neu_conn_eth_check_interface(const char *interface);
void neu_conn_eth_get_mac(neu_conn_eth_t *conn, uint8_t *mac);
int neu_conn_eth_size();
typedef void (*neu_conn_eth_msg_callback)(neu_conn_eth_t *conn, void *ctx,
uint16_t protocol, uint16_t n_byte,
uint8_t *bytes, uint8_t src_mac[6]);
int neu_conn_eth_register_lldp(neu_conn_eth_t *conn, const char *device,
neu_conn_eth_msg_callback callback);
int neu_conn_eth_unregister_lldp(neu_conn_eth_t *conn, const char *device);
typedef struct neu_conn_eth_sub neu_conn_eth_sub_t;
neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn,
uint16_t protocol, uint8_t mac[6],
neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn, uint8_t mac[6],
neu_conn_eth_msg_callback callback);
int neu_conn_eth_unregister(neu_conn_eth_t *conn, neu_conn_eth_sub_t *sub);
include/neuron/errcodes.h
+1
View File
@@ -75,6 +75,7 @@ typedef enum {
NEU_ERR_TAG_ADDRESS_TOO_LONG = 2207,
NEU_ERR_TAG_DESCRIPTION_TOO_LONG = 2208,
NEU_ERR_TAG_PRECISION_INVALID = 2209,
NEU_ERR_TAG_EXIST = 2210,
NEU_ERR_LIBRARY_NOT_FOUND = 2301,
NEU_ERR_LIBRARY_INFO_INVALID = 2302,
include/neuron/persist/persist.h
+84
View File
@@ -355,6 +355,90 @@ int neu_persister_delete_template(const char *name);
*/
int neu_persister_clear_templates();
/**
* Persist template group.
* @param tmpl_name template name.
* @param group_info group info to persist.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_store_template_group(const char * tmpl_name,
neu_persist_group_info_t *group_info);
/**
* Update template group.
* @param tmpl_name template name.
* @param group_info group info to persist.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_update_template_group(const char * tmpl_name,
neu_persist_group_info_t *group_info);
/**
* Load all group infos of a template.
* @param tmpl_name template name.
* @param[out] group_infos used to return pointer to heap allocated
* vector of neu_persist_group_info_t.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_load_template_groups(const char *tmpl_name,
UT_array ** group_infos);
/**
* Delete template group.
* @param tmpl_name template name.
* @param group_name name of the group.
* @return 0 on success, none-zero on failure
*/
int neu_persister_delete_template_group(const char *tmpl_name,
const char *group_name);
/**
* Persist template tags.
* @param tmpl_name name of the template
* @param group_name name of the group
* @param tags the tags to store
* @param n the number of tags
* @return 0 on success, non-zero otherwise
*/
int neu_persister_store_template_tags(const char * tmpl_name,
const char * group_name,
const neu_datatag_t *tags, size_t n);
/**
* Load template tag infos.
* @param tmpl_name name of the template
* @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_template_tags(const char *tmpl_name,
const char *group_name,
UT_array ** tag_infos);
/**
* Update template tags.
* @param tmpl_name name of the template
* @param group_name name of the group
* @param tags the tags to store
* @param n the number of tags
* @return 0 on success, non-zero otherwise
*/
int neu_persister_update_template_tags(const char * tmpl_name,
const char * group_name,
const neu_datatag_t *tags, size_t n);
/**
* Delete template tags.
* @param tmpl_name name of the template
* @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_template_tags(const char * tmpl_name,
const char * group_name,
const char *const *tags, size_t n_tag);
#ifdef __cplusplus
}
#endif
plugins/restful/handle.c
+15
View File
@@ -112,6 +112,9 @@ static struct neu_http_handler cors_handler[] = {
{
.url = "/api/v2/template",
},
{
.url = "/api/v2/template/group",
},
{
.url = "/api/v2/template/tag",
},
@@ -393,6 +396,18 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/template/inst",
.value.handler = handle_instantiate_template,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/template/group",
.value.handler = handle_add_template_group,
},
{
.method = NEU_HTTP_METHOD_PUT,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/template/group",
.value.handler = handle_update_template_group,
},
};
void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size)
plugins/restful/template_handle.c
+59
View File
@@ -321,6 +321,65 @@ void handle_get_templates_resp(nng_aio *aio, neu_resp_get_templates_t *resp)
return;
}
static inline int
send_template_mod_group_req(nng_aio *aio, neu_reqresp_type_e type,
neu_json_template_mod_group_req_t *req)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
neu_reqresp_head_t header = {
.ctx = aio,
.type = type,
};
neu_req_add_template_group_t cmd = { 0 };
strcpy(cmd.tmpl, req->tmpl);
strcpy(cmd.group, req->group);
cmd.interval = req->interval;
if (0 != neu_plugin_op(plugin, header, &cmd)) {
return NEU_ERR_IS_BUSY;
}
return 0;
}
static inline void process_mod_template_group(nng_aio * aio,
neu_reqresp_type_e type)
{
NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_template_mod_group_req_t,
neu_json_decode_template_mod_group_req, {
if (strlen(req->group) >= NEU_GROUP_NAME_LEN) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_GROUP_NAME_TOO_LONG, {
neu_http_response(aio, NEU_ERR_GROUP_NAME_TOO_LONG,
result_error);
});
} else if (req->interval < NEU_GROUP_INTERVAL_LIMIT) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_GROUP_PARAMETER_INVALID, {
neu_http_response(aio, NEU_ERR_GROUP_PARAMETER_INVALID,
result_error);
});
} else {
int ret = send_template_mod_group_req(aio, type, req);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(
ret, { neu_http_response(aio, ret, result_error); });
}
}
})
}
void handle_add_template_group(nng_aio *aio)
{
process_mod_template_group(aio, NEU_REQ_ADD_TEMPLATE_GROUP);
}
void handle_update_template_group(nng_aio *aio)
{
process_mod_template_group(aio, NEU_REQ_UPDATE_TEMPLATE_GROUP);
}
static int send_template_mod_tags_req(nng_aio *aio, neu_reqresp_type_e type,
neu_json_template_mod_tags_req_t *req)
{
plugins/restful/template_handle.h
+3
View File
@@ -30,6 +30,9 @@ void handle_get_template(nng_aio *aio);
void handle_get_template_resp(nng_aio *aio, neu_resp_get_template_t *resp);
void handle_get_templates_resp(nng_aio *aio, neu_resp_get_templates_t *resp);
void handle_add_template_group(nng_aio *aio);
void handle_update_template_group(nng_aio *aio);
void handle_add_template_tags(nng_aio *aio);
void handle_update_template_tags(nng_aio *aio);
void handle_del_template_tags(nng_aio *aio);
src/adapter/adapter.c
+12
View File
@@ -1150,6 +1150,18 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_RESP_GET_TEMPLATES:
data_size = sizeof(neu_resp_get_templates_t);
break;
case NEU_REQ_ADD_TEMPLATE_GROUP:
data_size = sizeof(neu_req_add_template_group_t);
break;
case NEU_REQ_DEL_TEMPLATE_GROUP:
data_size = sizeof(neu_req_del_template_group_t);
break;
case NEU_REQ_UPDATE_TEMPLATE_GROUP:
data_size = sizeof(neu_req_update_template_group_t);
break;
case NEU_REQ_GET_TEMPLATE_GROUP:
data_size = sizeof(neu_req_get_template_group_t);
break;
case NEU_REQ_ADD_TEMPLATE_TAG:
data_size = sizeof(neu_req_add_template_tag_t);
break;
src/connection/connection.c
+159
View File
@@ -312,6 +312,7 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
if (conn->is_connected) {
switch (conn->param.type) {
case NEU_CONN_UDP_TO:
case NEU_CONN_TCP_SERVER:
assert(false);
break;
@@ -364,6 +365,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
}
switch (conn->param.type) {
case NEU_CONN_UDP_TO:
case NEU_CONN_TCP_SERVER:
zlog_fatal(conn->param.log, "neu_conn_recv cann't recv tcp server msg");
assert(1 == 0);
@@ -414,6 +416,115 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
return ret;
}
ssize_t neu_conn_udp_sendto(neu_conn_t *conn, uint8_t *buf, ssize_t len,
void *dst)
{
ssize_t ret = 0;
pthread_mutex_lock(&conn->mtx);
if (conn->stop) {
pthread_mutex_unlock(&conn->mtx);
return ret;
}
if (!conn->is_connected) {
conn_connect(conn);
}
if (conn->is_connected) {
switch (conn->param.type) {
case NEU_CONN_TCP_CLIENT:
case NEU_CONN_TCP_SERVER:
case NEU_CONN_TTY_CLIENT:
case NEU_CONN_UDP:
assert(false);
break;
case NEU_CONN_UDP_TO:
if (conn->block) {
ret = sendto(conn->fd, buf, len, MSG_NOSIGNAL,
(struct sockaddr *) dst, sizeof(struct sockaddr));
} else {
ret = sendto(conn->fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT,
(struct sockaddr *) dst, sizeof(struct sockaddr));
}
break;
}
if (ret != len) {
zlog_error(
conn->param.log,
"conn udp fd: %d, sendto (%s:%d) buf len: %zd, ret: %zd, "
"errno: %s(%d)",
conn->fd, inet_ntoa(((struct sockaddr_in *) dst)->sin_addr),
htons(((struct sockaddr_in *) dst)->sin_port), len, ret,
strerror(errno), errno);
}
if (ret == -1 && errno != EAGAIN) {
conn_disconnect(conn);
}
if (ret > 0 && conn->callback_trigger == false) {
conn->connected(conn->data, conn->fd);
conn->callback_trigger = true;
}
}
if (ret > 0) {
conn->state.send_bytes += ret;
}
pthread_mutex_unlock(&conn->mtx);
return ret;
}
ssize_t neu_conn_udp_recvfrom(neu_conn_t *conn, uint8_t *buf, ssize_t len,
void *src)
{
ssize_t ret = 0;
socklen_t addr_len = sizeof(struct sockaddr_in);
pthread_mutex_lock(&conn->mtx);
if (conn->stop) {
pthread_mutex_unlock(&conn->mtx);
return ret;
}
switch (conn->param.type) {
case NEU_CONN_TCP_SERVER:
case NEU_CONN_TCP_CLIENT:
case NEU_CONN_TTY_CLIENT:
case NEU_CONN_UDP:
assert(1 == 0);
break;
case NEU_CONN_UDP_TO:
ret =
recvfrom(conn->fd, buf, len, 0, (struct sockaddr *) src, &addr_len);
break;
}
if (ret <= 0) {
zlog_error(conn->param.log,
"conn udp fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
conn->fd, len, ret, strerror(errno), errno);
}
if (errno == EPIPE || ret == -1 || ret == 0) {
conn_disconnect(conn);
}
if (ret > 0 && conn->callback_trigger == false) {
conn->connected(conn->data, conn->fd);
conn->callback_trigger = true;
}
pthread_mutex_unlock(&conn->mtx);
if (ret > 0) {
conn->state.recv_bytes += ret;
}
return ret;
}
void neu_conn_connect(neu_conn_t *conn)
{
@@ -453,6 +564,9 @@ static void conn_free_param(neu_conn_t *conn)
free(conn->param.params.udp.src_ip);
free(conn->param.params.udp.dst_ip);
break;
case NEU_CONN_UDP_TO:
free(conn->param.params.udpto.src_ip);
break;
case NEU_CONN_TTY_CLIENT:
free(conn->param.params.tty_client.device);
break;
@@ -499,6 +613,12 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
conn->param.params.udp.timeout = param->params.udp.timeout;
conn->block = conn->param.params.udp.timeout > 0;
break;
case NEU_CONN_UDP_TO:
conn->param.params.udpto.src_ip = strdup(param->params.udpto.src_ip);
conn->param.params.udpto.src_port = param->params.udpto.src_port;
conn->param.params.udpto.timeout = param->params.udpto.timeout;
conn->block = conn->param.params.udpto.timeout > 0;
break;
case NEU_CONN_TTY_CLIENT:
conn->param.params.tty_client.device =
strdup(param->params.tty_client.device);
@@ -683,6 +803,44 @@ static void conn_connect(neu_conn_t *conn)
}
break;
}
case NEU_CONN_UDP_TO: {
if (conn->block) {
struct timeval tv = {
.tv_sec = conn->param.params.udpto.timeout / 1000,
.tv_usec = (conn->param.params.udpto.timeout % 1000) * 1000,
};
fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
} else {
fd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP);
}
int so_broadcast = 1;
setsockopt(fd, SOL_SOCKET, SO_BROADCAST, &so_broadcast,
sizeof(so_broadcast));
struct sockaddr_in local = {
.sin_family = AF_INET,
.sin_port = htons(conn->param.params.udpto.src_port),
.sin_addr.s_addr = inet_addr(conn->param.params.udpto.src_ip),
};
ret = bind(fd, (struct sockaddr *) &local, sizeof(struct sockaddr_in));
if (ret != 0) {
close(fd);
zlog_error(conn->param.log, "bind %s:%d error: %s(%d)",
conn->param.params.udpto.src_ip,
conn->param.params.udpto.src_port, strerror(errno),
errno);
conn->is_connected = false;
return;
}
conn->is_connected = true;
conn->fd = fd;
break;
}
case NEU_CONN_TTY_CLIENT: {
struct termios tty_opt = { 0 };
@@ -854,6 +1012,7 @@ static void conn_disconnect(neu_conn_t *conn)
break;
case NEU_CONN_TCP_CLIENT:
case NEU_CONN_UDP:
case NEU_CONN_UDP_TO:
case NEU_CONN_TTY_CLIENT:
if (conn->fd > 0) {
close(conn->fd);
src/connection/connection_eth.c
+101 -198
View File
@@ -33,6 +33,7 @@
#include "connection/neu_connection_eth.h"
#include "event/event.h"
#include "type.h"
#include "utils/log.h"
#include "utils/utextend.h"
@@ -40,13 +41,12 @@
#define ETH_P_LLDP 0x88CC
#endif
typedef struct neu_conn_eth_sub {
uint16_t protocol;
char mac[ETH_ALEN];
} callback_elem_key_t;
struct neu_conn_eth_sub {
uint8_t mac[6];
};
typedef struct {
callback_elem_key_t key;
char mac[20];
neu_conn_eth_msg_callback callback;
@@ -54,8 +54,6 @@ typedef struct {
} callback_elem_t;
typedef struct {
char device[NEU_NODE_NAME_LEN];
neu_conn_eth_msg_callback callback;
UT_hash_handle hh;
@@ -65,20 +63,16 @@ typedef struct interface_conn {
char * interface;
neu_events_t *events;
uint8_t count;
uint8_t buf[1500];
int lldp_fd;
int profinet_fd;
int vlan_fd;
uint8_t mac[ETH_ALEN];
neu_event_io_t *profinet_event;
neu_event_io_t *lldp_event;
neu_event_io_t *vlan_event;
lldp_callback_elem_t *lldp_callbacks;
callback_elem_t * callbacks;
pthread_mutex_t mtx;
callback_elem_t *callbacks;
pthread_mutex_t mtx;
} interface_conn_t;
static pthread_mutex_t mtx = { 0 };
@@ -92,13 +86,25 @@ struct neu_conn_eth {
};
static int get_mac(neu_conn_eth_t *conn, const char *interface);
static int init_socket(const char *interface, uint16_t protocol);
static int init_socket(const char *interface, uint16_t protocol,
uint8_t mac[6]);
static int uninit_socket(int fd);
static int lldp_msg_cb(enum neu_event_io_type type, int fd, void *usr_data);
static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data);
static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data);
static uint8_t pf_dcp_broadcast[ETH_ALEN] = {
0x01, 0x0e, 0xcf, 0x00, 0x00, 0x01
};
static char *lldp_parse_port_id(uint16_t n_byte, uint8_t *bytes);
int neu_conn_eth_size()
{
int ret = 0;
for (int i = 0; i < interface_max; i++) {
ret += in_conns[i].count;
}
return ret;
}
neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
{
@@ -127,18 +133,20 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
if (in_conns[i].interface == NULL) {
neu_event_io_param_t param = { 0 };
pthread_mutex_init(&in_conns[i].mtx, NULL);
conn_eth->ic = &in_conns[i];
in_conns[i].interface = strdup(interface);
in_conns[i].events = neu_event_new();
in_conns[i].count += 1;
in_conns[i].profinet_fd = init_socket(interface, 0x8892);
in_conns[i].lldp_fd = init_socket(interface, ETH_P_LLDP);
in_conns[i].vlan_fd = init_socket(interface, 0x8100);
pthread_mutex_init(&in_conns[i].mtx, NULL);
conn_eth->ic = &in_conns[i];
get_mac(conn_eth, interface);
in_conns[i].profinet_fd =
init_socket(interface, 0x8892, in_conns[i].mac);
// in_conns[i].vlan_fd =
// init_socket(interface, 0x8100, in_conns[i].mac);
param.fd = in_conns[i].profinet_fd;
param.usr_data = (void *) conn_eth;
param.cb = eth_msg_cb;
@@ -146,17 +154,11 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
in_conns[i].profinet_event =
neu_event_add_io(in_conns[i].events, param);
param.fd = in_conns[i].lldp_fd;
param.cb = lldp_msg_cb;
// param.fd = in_conns[i].vlan_fd;
// param.cb = eth_msg_cb;
in_conns[i].lldp_event =
neu_event_add_io(in_conns[i].events, param);
param.fd = in_conns[i].vlan_fd;
param.cb = eth_msg_cb;
in_conns[i].vlan_event =
neu_event_add_io(in_conns[i].events, param);
// in_conns[i].vlan_event =
// neu_event_add_io(in_conns[i].events, param);
break;
}
@@ -175,16 +177,15 @@ int neu_conn_eth_uninit(neu_conn_eth_t *conn)
conn->ic->count -= 1;
if (conn->ic->count == 0) {
neu_event_del_io(conn->ic->events, conn->ic->profinet_event);
neu_event_del_io(conn->ic->events, conn->ic->lldp_event);
neu_event_del_io(conn->ic->events, conn->ic->vlan_event);
// neu_event_del_io(conn->ic->events, conn->ic->vlan_event);
neu_event_close(conn->ic->events);
uninit_socket(conn->ic->profinet_fd);
uninit_socket(conn->ic->lldp_fd);
uninit_socket(conn->ic->vlan_fd);
// uninit_socket(conn->ic->vlan_fd);
free(conn->ic->interface);
conn->ic->interface = NULL;
pthread_mutex_destroy(&conn->ic->mtx);
}
pthread_mutex_unlock(&mtx);
@@ -219,33 +220,29 @@ int neu_conn_eth_check_interface(const char *interface)
return ret;
}
neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn,
uint16_t protocol, uint8_t mac[6],
neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn, uint8_t xmac[6],
neu_conn_eth_msg_callback callback)
{
callback_elem_key_t key = { 0 };
neu_conn_eth_sub_t *sub = NULL;
callback_elem_t * elem = NULL;
char tmp_mac[20] = { 0 };
neu_conn_eth_sub_t *sub = NULL;
callback_elem_t * elem = NULL;
key.protocol = protocol;
memcpy(key.mac, mac, ETH_ALEN);
snprintf(tmp_mac, sizeof(tmp_mac), "%X:%X:%X-%X:%X:%X", xmac[0], xmac[1],
xmac[2], xmac[3], xmac[4], xmac[5]);
pthread_mutex_lock(&conn->ic->mtx);
HASH_FIND(hh, conn->ic->callbacks, &key, sizeof(callback_elem_key_t), elem);
HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
if (elem == NULL) {
elem = calloc(1, sizeof(callback_elem_t));
sub = calloc(1, sizeof(neu_conn_eth_sub_t));
elem->key.protocol = protocol;
memcpy(key.mac, mac, ETH_ALEN);
elem->callback = callback;
memcpy(elem->mac, tmp_mac, sizeof(elem->mac));
memcpy(sub->mac, mac, ETH_ALEN);
sub->protocol = protocol;
memcpy(sub->mac, xmac, ETH_ALEN);
HASH_ADD(hh, conn->ic->callbacks, key, sizeof(callback_elem_key_t),
elem);
HASH_ADD_STR(conn->ic->callbacks, mac, elem);
}
pthread_mutex_unlock(&conn->ic->mtx);
@@ -255,15 +252,15 @@ neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn,
int neu_conn_eth_unregister(neu_conn_eth_t *conn, neu_conn_eth_sub_t *sub)
{
callback_elem_key_t key = { 0 };
callback_elem_t * elem = NULL;
char tmp_mac[20] = { 0 };
callback_elem_t *elem = NULL;
key.protocol = sub->protocol;
memcpy(key.mac, sub->mac, ETH_ALEN);
snprintf(tmp_mac, sizeof(tmp_mac), "%X:%X:%X-%X:%X:%X", sub->mac[0],
sub->mac[1], sub->mac[2], sub->mac[3], sub->mac[4], sub->mac[5]);
pthread_mutex_lock(&conn->ic->mtx);
HASH_FIND(hh, conn->ic->callbacks, &key, sizeof(callback_elem_key_t), elem);
HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
if (elem != NULL) {
HASH_DEL(conn->ic->callbacks, elem);
free(elem);
@@ -276,46 +273,6 @@ int neu_conn_eth_unregister(neu_conn_eth_t *conn, neu_conn_eth_sub_t *sub)
return 0;
}
int neu_conn_eth_register_lldp(neu_conn_eth_t *conn, const char *device,
neu_conn_eth_msg_callback callback)
{
lldp_callback_elem_t *elem = NULL;
int ret = 0;
pthread_mutex_lock(&conn->ic->mtx);
HASH_FIND_STR(conn->ic->lldp_callbacks, device, elem);
if (elem != NULL) {
ret = -1;
} else {
elem = calloc(1, sizeof(lldp_callback_elem_t));
elem->callback = callback;
strcpy(elem->device, device);
HASH_ADD_STR(conn->ic->lldp_callbacks, device, elem);
}
pthread_mutex_unlock(&conn->ic->mtx);
return ret;
}
int neu_conn_eth_unregister_lldp(neu_conn_eth_t *conn, const char *device)
{
lldp_callback_elem_t *elem = NULL;
pthread_mutex_lock(&conn->ic->mtx);
HASH_FIND_STR(conn->ic->lldp_callbacks, device, elem);
if (elem != NULL) {
HASH_DEL(conn->ic->lldp_callbacks, elem);
free(elem);
}
pthread_mutex_unlock(&conn->ic->mtx);
return 0;
}
int neu_conn_eth_send(neu_conn_eth_t *conn, uint16_t protocol, uint16_t n_byte,
uint8_t *bytes)
{
@@ -323,12 +280,10 @@ int neu_conn_eth_send(neu_conn_eth_t *conn, uint16_t protocol, uint16_t n_byte,
int ret = 0;
switch (protocol) {
case ETH_P_LLDP:
case 0x8892:
fd = conn->ic->profinet_fd;
break;
case ETH_P_LLDP:
fd = conn->ic->lldp_fd;
break;
case 0x8100:
fd = conn->ic->vlan_fd;
break;
@@ -357,20 +312,22 @@ static int get_mac(neu_conn_eth_t *conn, const char *interface)
return ret;
}
static int init_socket(const char *interface, uint16_t protocol)
static int init_socket(const char *interface, uint16_t protocol, uint8_t mac[6])
{
struct ifreq ifr = { 0 };
struct sockaddr_ll sll = { 0 };
struct packet_mreq mreq = { 0 };
const uint8_t pn_mcast_addr[ETH_ALEN] = {
0x01, 0x0e, 0xcf, 0x00, 0x00, 0x00
};
int fd = socket(PF_PACKET, SOCK_RAW, htons(protocol));
int ret = 0;
int ret = 1;
assert(fd > 0);
setsockopt(fd, SOL_SOCKET, SO_DONTROUTE, &ret, sizeof(ret));
(void) mac;
(void) pn_mcast_addr;
strcpy(ifr.ifr_name, interface);
if (ioctl(fd, SIOCGIFINDEX, &ifr) != 0) {
close(fd);
@@ -380,25 +337,12 @@ static int init_socket(const char *interface, uint16_t protocol)
sll.sll_family = AF_PACKET;
sll.sll_ifindex = ifr.ifr_ifindex;
sll.sll_protocol = htons(protocol);
mreq.mr_ifindex = ifr.ifr_ifindex;
mreq.mr_type = PACKET_HOST | PACKET_MR_MULTICAST;
mreq.mr_alen = ETH_ALEN;
memcpy(mreq.mr_address, pn_mcast_addr, ETH_ALEN);
ifr.ifr_flags = 0;
ioctl(fd, SIOCGIFFLAGS, &ifr);
bind(fd, (struct sockaddr *) &sll, sizeof(sll));
if (protocol == 0x8892) {
ret = setsockopt(fd, SOL_PACKET, PACKET_ADD_MEMBERSHIP, &mreq,
sizeof(mreq));
assert(ret == 0);
}
ret = bind(fd, (struct sockaddr *) &sll, sizeof(sll));
assert(ret == 0);
return fd;
}
static int uninit_socket(int fd)
{
close(fd);
@@ -412,27 +356,53 @@ static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data)
switch (type) {
case NEU_EVENT_IO_READ: {
callback_elem_key_t key = { 0 };
callback_elem_t * elem = NULL;
memset(conn->ic->buf, 0, sizeof(conn->ic->buf));
int ret = recv(fd, conn->ic->buf, sizeof(conn->ic->buf), 0);
callback_elem_t *elem = NULL;
uint8_t buf[1500] = { 0 };
int ret = recv(fd, buf, 1500, 0);
if (ret < ETH_HLEN) {
nlog_warn("eth recv length < %d msg", ETH_HLEN);
return 0;
}
struct ethhdr *ehdr = (struct ethhdr *) conn->ic->buf;
struct ethhdr *ehdr = (struct ethhdr *) buf;
ehdr->h_proto = ntohs(ehdr->h_proto);
key.protocol = ntohs(ehdr->h_proto);
memcpy(key.mac, ehdr->h_source, ETH_ALEN);
if (ehdr->h_proto != 0x8100 && ehdr->h_proto != 0x8892) {
return 0;
}
HASH_FIND(hh, conn->ic->callbacks, &key, sizeof(key), elem);
if (memcmp(ehdr->h_dest, pf_dcp_broadcast, ETH_ALEN) == 0) {
callback_elem_t *tmp = NULL;
if (elem != NULL) {
elem->callback(conn, conn->ctx, key.protocol, ret - ETH_HLEN,
conn->ic->buf + ETH_HLEN, ehdr->h_source);
HASH_ITER(hh, conn->ic->callbacks, elem, tmp)
{
elem->callback(conn, conn->ctx, ehdr->h_proto, ret - ETH_HLEN,
buf + ETH_HLEN, ehdr->h_source);
}
} else {
if (memcmp(ehdr->h_dest, conn->ic->mac, ETH_ALEN) != 0) {
return 0;
}
char tmp_mac[20] = { 0 };
snprintf(tmp_mac, sizeof(tmp_mac), "%X:%X:%X-%X:%X:%X",
ehdr->h_source[0], ehdr->h_source[1], ehdr->h_source[2],
ehdr->h_source[3], ehdr->h_source[4], ehdr->h_source[5]);
HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
if (elem != NULL) {
elem->callback(conn, conn->ctx, ehdr->h_proto, ret - ETH_HLEN,
buf + ETH_HLEN, ehdr->h_source);
} else {
memset(tmp_mac, 0, sizeof(tmp_mac));
snprintf(tmp_mac, sizeof(tmp_mac), "0:0:0-0:0:0");
HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
if (elem != NULL) {
elem->callback(conn, conn->ctx, ehdr->h_proto,
ret - ETH_HLEN, buf + ETH_HLEN,
ehdr->h_source);
}
}
}
break;
@@ -446,70 +416,3 @@ static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data)
return 0;
}
static int lldp_msg_cb(enum neu_event_io_type type, int fd, void *usr_data)
{
neu_conn_eth_t *conn = (neu_conn_eth_t *) usr_data;
int ret = 0;
char * port_id = NULL;
switch (type) {
case NEU_EVENT_IO_READ:
memset(conn->ic->buf, 0, sizeof(conn->ic->buf));
ret = recv(fd, conn->ic->buf, sizeof(conn->ic->buf), 0);
port_id = lldp_parse_port_id(ret - sizeof(struct ethhdr),
conn->ic->buf + sizeof(struct ethhdr));
if (port_id != NULL) {
lldp_callback_elem_t *elem = NULL;
struct ethhdr * ehdr = (struct ethhdr *) conn->ic->buf;
HASH_FIND_STR(conn->ic->callbacks, port_id, elem);
if (elem != NULL) {
elem->callback(
conn, conn->ctx, ETH_P_LLDP, ret - sizeof(struct ethhdr),
conn->ic->buf + sizeof(struct ethhdr), ehdr->h_source);
}
}
break;
case NEU_EVENT_IO_CLOSED:
case NEU_EVENT_IO_HUP:
nlog_warn("eth conn lldp error type: %d, fd: %d(%s)", type, fd,
conn->ic->interface);
break;
}
return 0;
}
struct lldp_tvl {
uint16_t type : 7;
uint16_t length : 9;
} __attribute__((packed));
#define END_OF_LLDPDU_TLV_TYPE 0x00
#define CHASSIS_ID_TLV_TYPE 0x01
#define PORT_ID_TLV_TYPE 0x02
#define TIME_TO_LIVE_TLV_TYPE 0x03
static char *lldp_parse_port_id(uint16_t n_byte, uint8_t *bytes)
{
struct lldp_tvl *tvl = (struct lldp_tvl *) bytes;
if (n_byte < sizeof(struct lldp_tvl) + tvl->length) {
return NULL;
}
if (tvl->type == PORT_ID_TLV_TYPE) {
static __thread char port_id[NEU_NODE_NAME_LEN] = { 0 };
uint8_t * data = (uint8_t *) &tvl[1];
strncpy(port_id, (char *) data + 1, tvl->length - 1);
return port_id;
} else {
return lldp_parse_port_id(
n_byte - sizeof(struct lldp_tvl) - tvl->length,
bytes + sizeof(struct lldp_tvl) + tvl->length);
}
}
src/core/manager.c
+49 -3
View File
@@ -380,6 +380,48 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &resp);
break;
}
case NEU_REQ_ADD_TEMPLATE_GROUP: {
neu_req_add_template_group_t *cmd =
(neu_req_add_template_group_t *) &header[1];
neu_resp_error_t e = { 0 };
if (cmd->interval < NEU_GROUP_INTERVAL_LIMIT) {
e.error = NEU_ERR_GROUP_PARAMETER_INVALID;
} else {
e.error = neu_manager_add_template_group(manager, cmd);
}
if (e.error == NEU_ERR_SUCCESS) {
manager_storage_add_template_group(cmd->tmpl, cmd->group,
cmd->interval);
}
neu_msg_exchange(header);
header->type = NEU_RESP_ERROR;
reply(manager, header, &e);
break;
}
case NEU_REQ_UPDATE_TEMPLATE_GROUP: {
neu_req_update_template_group_t *cmd =
(neu_req_update_template_group_t *) &header[1];
neu_resp_error_t e = { 0 };
if (cmd->interval < NEU_GROUP_INTERVAL_LIMIT) {
e.error = NEU_ERR_GROUP_PARAMETER_INVALID;
} else {
e.error = neu_manager_update_template_group(manager, cmd);
}
if (e.error == NEU_ERR_SUCCESS) {
manager_storage_update_template_group(cmd->tmpl, cmd->group,
cmd->interval);
}
neu_msg_exchange(header);
header->type = NEU_RESP_ERROR;
reply(manager, header, &e);
break;
}
case NEU_REQ_ADD_TEMPLATE_TAG: {
neu_req_add_template_tag_t *cmd =
(neu_req_add_template_tag_t *) &header[1];
@@ -387,7 +429,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
resp.error = neu_manager_add_template_tags(manager, cmd, &resp.index);
if (resp.index > 0) {
// TODO: storage
manager_storage_add_template_tags(cmd->tmpl, cmd->group, cmd->tags,
cmd->n_tag);
}
neu_req_add_template_tag_fini(cmd);
@@ -405,7 +448,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
resp.error =
neu_manager_update_template_tags(manager, cmd, &resp.index);
if (resp.index > 0) {
// TODO: storage
manager_storage_update_template_tags(cmd->tmpl, cmd->group,
cmd->tags, cmd->n_tag);
}
neu_req_update_template_tag_fini(cmd);
@@ -422,7 +466,9 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
resp.error = neu_manager_del_template_tags(manager, cmd);
if (0 == resp.error) {
// TODO: storage
manager_storage_del_template_tags(cmd->tmpl, cmd->group,
(const char *const *) cmd->tags,
cmd->n_tag);
}
neu_req_del_template_tag_fini(cmd);
src/core/manager_internal.c
+24
View File
@@ -229,6 +229,30 @@ int neu_manager_get_templates(neu_manager_t * manager,
return 0;
}
int neu_manager_add_template_group(neu_manager_t * manager,
neu_req_add_template_group_t *req)
{
neu_template_t *tmpl =
neu_template_manager_find(manager->template_manager, req->tmpl);
if (NULL == tmpl) {
return NEU_ERR_TEMPLATE_NOT_FOUND;
}
return neu_template_add_group(tmpl, req->group, req->interval);
}
int neu_manager_update_template_group(neu_manager_t * manager,
neu_req_update_template_group_t *req)
{
neu_template_t *tmpl =
neu_template_manager_find(manager->template_manager, req->tmpl);
if (NULL == tmpl) {
return NEU_ERR_TEMPLATE_NOT_FOUND;
}
return neu_template_update_group(tmpl, req->group, req->interval);
}
int neu_manager_add_template_tags(neu_manager_t * manager,
neu_req_add_template_tag_t *req,
uint16_t * index_p)
src/core/manager_internal.h
+4
View File
@@ -65,6 +65,10 @@ int neu_manager_get_template(neu_manager_t *manager, const char *name,
neu_resp_get_template_t *resp);
int neu_manager_get_templates(neu_manager_t * manager,
neu_resp_get_templates_t *resp);
int neu_manager_add_template_group(neu_manager_t * manager,
neu_req_add_template_group_t *req);
int neu_manager_update_template_group(neu_manager_t * manager,
neu_req_update_template_group_t *req);
int neu_manager_add_template_tags(neu_manager_t * manager,
neu_req_add_template_tag_t *req,
uint16_t * index_p);
src/core/storage.c
+150 -3
View File
@@ -85,7 +85,7 @@ void manager_storage_unsubscribe(neu_manager_t *manager, const char *app,
}
}
static int save_group_and_tags(neu_group_t *grp, void *data)
static int save_node_group_and_tags(neu_group_t *grp, void *data)
{
const char *node = data;
const char *name = neu_group_get_name(grp);
@@ -117,13 +117,32 @@ void manager_storage_inst_node(neu_manager_t *manager, const char *tmpl_name,
manager_storage_add_node(manager, node);
int rv =
neu_template_for_each_group(tmpl, save_group_and_tags, (void *) node);
int rv = neu_template_for_each_group(tmpl, save_node_group_and_tags,
(void *) node);
if (0 != rv) {
nlog_error("save instantiated node `%s` groups and tags fail", node);
}
}
static int save_tmpl_group_and_tags(neu_group_t *grp, void *data)
{
const char *tmpl_name = data;
const char *name = neu_group_get_name(grp);
uint32_t interval = neu_group_get_interval(grp);
UT_array *tags = neu_group_get_tag(grp);
if (NULL == tags) {
return NEU_ERR_EINTERNAL;
}
manager_storage_add_template_group(tmpl_name, name, interval);
manager_storage_add_template_tags(tmpl_name, name, utarray_front(tags),
utarray_len(tags));
utarray_free(tags);
return 0;
}
void manager_storage_add_template(neu_manager_t *manager, const char *tmpl_name)
{
neu_template_t *tmpl =
@@ -136,6 +155,13 @@ void manager_storage_add_template(neu_manager_t *manager, const char *tmpl_name)
int rv = neu_persister_store_template(tmpl_name, neu_template_plugin(tmpl));
if (0 != rv) {
nlog_error("failed to store template info");
return;
}
rv = neu_template_for_each_group(tmpl, save_tmpl_group_and_tags,
(void *) tmpl_name);
if (0 != rv) {
nlog_error("save template `%s` groups and tags fail", tmpl_name);
}
}
@@ -159,6 +185,127 @@ void manager_storage_clear_templates(neu_manager_t *manager)
}
}
void manager_storage_add_template_group(const char *tmpl, const char *group,
uint32_t interval)
{
neu_persist_group_info_t info = {
.name = (char *) group,
.interval = interval,
};
int rv = neu_persister_store_template_group(tmpl, &info);
if (0 != rv) {
nlog_error("store template:%s grp:%s, interval:%" PRIu32 " fail", tmpl,
group, interval);
}
}
void manager_storage_update_template_group(const char *tmpl, const char *group,
uint32_t interval)
{
neu_persist_group_info_t info = {
.name = (char *) group,
.interval = interval,
};
int rv = neu_persister_update_template_group(tmpl, &info);
if (0 != rv) {
nlog_error("update template:%s grp:%s, interval:%" PRIu32 " fail", tmpl,
group, interval);
}
}
void manager_storage_del_template_group(const char *tmpl, const char *group)
{
int rv = neu_persister_delete_template_group(tmpl, group);
if (0 != rv) {
nlog_error("delete template:%s grp:%s fail" PRIu32, tmpl, group);
}
}
void manager_storage_add_template_tag(const char *tmpl, const char *group,
const neu_datatag_t *tag)
{
int rv = neu_persister_store_template_tags(tmpl, group, tag, 1);
if (0 != rv) {
nlog_error("store tag:%s template:%s grp:%s fail", tag->name, tmpl,
group);
}
}
void manager_storage_add_template_tags(const char *tmpl, const char *group,
const neu_datatag_t *tags, size_t n)
{
if (0 == n) {
return;
}
if (1 == n) {
return manager_storage_add_template_tag(tmpl, group, &tags[0]);
}
int rv = neu_persister_store_template_tags(tmpl, group, tags, n);
if (0 != rv) {
nlog_error("store %zu tags:[%s ... %s] template:%s grp:%s fail", n,
tags->name, tags[n - 1].name, tmpl, group);
}
}
void manager_storage_update_template_tag(const char *tmpl, const char *group,
const neu_datatag_t *tag)
{
int rv = neu_persister_update_template_tags(tmpl, group, tag, 1);
if (0 != rv) {
nlog_error("update tag:%s template:%s grp:%s fail", tag->name, tmpl,
group);
}
}
void manager_storage_update_template_tags(const char *tmpl, const char *group,
const neu_datatag_t *tags, size_t n)
{
if (0 == n) {
return;
}
if (1 == n) {
return manager_storage_update_template_tag(tmpl, group, &tags[0]);
}
int rv = neu_persister_update_template_tags(tmpl, group, tags, n);
if (0 != rv) {
nlog_error("update %zu tags:[%s ... %s] template:%s grp:%s fail", n,
tags->name, tags[n - 1].name, tmpl, group);
}
}
void manager_storage_del_template_tag(const char *tmpl, const char *group,
const char *tag)
{
int rv = neu_persister_delete_template_tags(tmpl, group, &tag, 1);
if (0 != rv) {
nlog_error("delete tag:%s template:%s grp:%s fail", tag, tmpl, group);
}
}
void manager_storage_del_template_tags(const char *tmpl, const char *group,
const char *const *tags, size_t n)
{
if (0 == n) {
return;
}
if (1 == n) {
return manager_storage_del_template_tag(tmpl, group, tags[0]);
}
int rv = neu_persister_delete_template_tags(tmpl, group, tags, n);
if (0 != rv) {
nlog_error("del %zu tags:[%s ... %s] template:%s grp:%s fail", n,
tags[0], tags[n - 1], tmpl, group);
}
}
int manager_load_plugin(neu_manager_t *manager)
{
UT_array *plugin_infos = NULL;
src/core/storage.h
+17
View File
@@ -41,6 +41,23 @@ void manager_storage_add_template(neu_manager_t *manager,
void manager_storage_del_template(neu_manager_t *manager,
const char * tmpl_name);
void manager_storage_clear_templates(neu_manager_t *manager);
void manager_storage_add_template_group(const char *tmpl, const char *group,
uint32_t interval);
void manager_storage_update_template_group(const char *tmpl, const char *group,
uint32_t interval);
void manager_storage_del_template_group(const char *tmpl, const char *group);
void manager_storage_add_template_tag(const char *tmpl, const char *group,
const neu_datatag_t *tag);
void manager_storage_add_template_tags(const char *tmpl, const char *group,
const neu_datatag_t *tags, size_t n);
void manager_storage_update_template_tag(const char *tmpl, const char *group,
const neu_datatag_t *tag);
void manager_storage_update_template_tags(const char *tmpl, const char *group,
const neu_datatag_t *tags, size_t n);
void manager_storage_del_template_tag(const char *tmpl, const char *group,
const char *tag);
void manager_storage_del_template_tags(const char *tmpl, const char *group,
const char *const *tags, size_t n);
int manager_load_plugin(neu_manager_t *manager);
int manager_load_node(neu_manager_t *manager);
src/parser/neu_json_template.c
+86
View File
@@ -377,6 +377,92 @@ int neu_json_encode_get_templates_resp(void *json_obj, void *param)
NEU_JSON_ELEM_SIZE(req_elems));
}
int neu_json_encode_template_mod_group_req(void *json_obj, void *param)
{
int ret = 0;
neu_json_template_mod_group_req_t *req = param;
neu_json_elem_t req_elems[] = { {
.name = "template",
.t = NEU_JSON_STR,
.v.val_str = req->tmpl,
},
{
.name = "group",
.t = NEU_JSON_STR,
.v.val_str = req->group,
},
{
.name = "interval",
.t = NEU_JSON_INT,
.v.val_int = req->interval,
} };
ret = neu_json_encode_field(json_obj, req_elems,
NEU_JSON_ELEM_SIZE(req_elems));
return ret;
}
int neu_json_decode_template_mod_group_req(
char *buf, neu_json_template_mod_group_req_t **result)
{
int ret = 0;
void * json_obj = NULL;
neu_json_template_mod_group_req_t *req = NULL;
json_obj = neu_json_decode_new(buf);
if (NULL == json_obj) {
return -1;
}
req = calloc(1, sizeof(*req));
if (req == NULL) {
goto decode_fail;
}
neu_json_elem_t req_elems[] = { {
.name = "template",
.t = NEU_JSON_STR,
},
{
.name = "group",
.t = NEU_JSON_STR,
},
{
.name = "interval",
.t = NEU_JSON_INT,
} };
ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
req_elems);
req->tmpl = req_elems[0].v.val_str;
req->group = req_elems[1].v.val_str;
req->interval = req_elems[2].v.val_int;
if (ret != 0) {
goto decode_fail;
}
*result = req;
neu_json_decode_free(json_obj);
return 0;
decode_fail:
neu_json_decode_template_mod_group_req_free(req);
neu_json_decode_free(json_obj);
return -1;
}
void neu_json_decode_template_mod_group_req_free(
neu_json_template_mod_group_req_t *req)
{
if (req) {
free(req->tmpl);
free(req->group);
free(req);
}
}
int neu_json_encode_template_mod_tags_req(void *json_obj, void *param)
{
int ret = 0;
src/parser/neu_json_template.h
+12
View File
@@ -77,6 +77,18 @@ int neu_json_encode_template_info_array(void *json_obj, void *param);
// param should be a neu_json_template_info_array_t
int neu_json_encode_get_templates_resp(void *json_obj, void *param);
typedef struct {
char * tmpl;
char * group;
int64_t interval;
} neu_json_template_mod_group_req_t;
int neu_json_encode_template_mod_group_req(void *json_obj, void *param);
int neu_json_decode_template_mod_group_req(
char *buf, neu_json_template_mod_group_req_t **result);
void neu_json_decode_template_mod_group_req_free(
neu_json_template_mod_group_req_t *req);
typedef struct {
char * tmpl;
char * group;
src/persist/persist.c
+409 -114
View File
@@ -720,15 +720,85 @@ int neu_persister_store_tag(const char *driver_name, const char *group_name,
return rv;
}
static int put_tags(const char *query, sqlite3_stmt *stmt,
const neu_datatag_t *tags, size_t n)
{
for (size_t i = 0; i < n; ++i) {
const neu_datatag_t *tag = &tags[i];
sqlite3_reset(stmt);
if (SQLITE_OK != sqlite3_bind_text(stmt, 3, tag->name, -1, NULL)) {
nlog_error("bind `%s` with name=`%s` fail: %s", query, tag->name,
sqlite3_errmsg(global_db));
return -1;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 4, tag->address, -1, NULL)) {
nlog_error("bind `%s` with address=`%s` fail: %s", query,
tag->address, sqlite3_errmsg(global_db));
return -1;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 5, tag->attribute)) {
nlog_error("bind `%s` with attribute=`%i` fail: %s", query,
tag->attribute, sqlite3_errmsg(global_db));
return -1;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 6, tag->precision)) {
nlog_error("bind `%s` with precision=`%i` fail: %s", query,
tag->precision, sqlite3_errmsg(global_db));
return -1;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 7, tag->type)) {
nlog_error("bind `%s` with type=`%i` fail: %s", query, tag->type,
sqlite3_errmsg(global_db));
return -1;
}
if (SQLITE_OK != sqlite3_bind_double(stmt, 8, tag->decimal)) {
nlog_error("bind `%s` with decimal=`%f` fail: %s", query,
tag->decimal, sqlite3_errmsg(global_db));
return -1;
}
if (SQLITE_OK !=
sqlite3_bind_text(stmt, 9, tag->description, -1, NULL)) {
nlog_error("bind `%s` with description=`%s` fail: %s", query,
tag->description, sqlite3_errmsg(global_db));
return -1;
}
char *val_str = neu_tag_dump_static_value(tag);
if (SQLITE_OK != sqlite3_bind_text(stmt, 10, val_str, -1, NULL)) {
nlog_error("bind `%s` with value=`%s` fail: %s", query, val_str,
sqlite3_errmsg(global_db));
free(val_str);
return -1;
}
if (SQLITE_DONE != sqlite3_step(stmt)) {
nlog_error("sqlite3_step fail: %s", sqlite3_errmsg(global_db));
free(val_str);
return -1;
}
free(val_str);
}
return 0;
}
int neu_persister_store_tags(const char *driver_name, const char *group_name,
const neu_datatag_t *tags, size_t n)
{
char * val_str = NULL;
sqlite3_stmt *stmt = NULL;
const char * query = "INSERT INTO tags ("
sqlite3_stmt *stmt = NULL;
const char * query = "INSERT INTO tags ("
" driver_name, group_name, name, address, attribute,"
" precision, type, decimal, description, value"
") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
") VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)";
if (SQLITE_OK != sqlite3_exec(global_db, "BEGIN", NULL, NULL, NULL)) {
nlog_error("begin transaction fail: %s", sqlite3_errmsg(global_db));
@@ -740,80 +810,20 @@ int neu_persister_store_tags(const char *driver_name, const char *group_name,
goto error;
}
for (size_t i = 0; i < n; ++i) {
const neu_datatag_t *tag = &tags[i];
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
nlog_error("bind `%s` with driver_name=`%s` fail: %s", query,
driver_name, sqlite3_errmsg(global_db));
goto error;
}
sqlite3_reset(stmt);
val_str = neu_tag_dump_static_value(tag);
if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
nlog_error("bind `%s` with group_name=`%s` fail: %s", query, group_name,
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
nlog_error("bind `%s` with driver_name=`%s` fail: %s", query,
driver_name, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
nlog_error("bind `%s` with group_name=`%s` fail: %s", query,
group_name, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 3, tag->name, -1, NULL)) {
nlog_error("bind `%s` with name=`%s` fail: %s", query, tag->name,
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 4, tag->address, -1, NULL)) {
nlog_error("bind `%s` with address=`%s` fail: %s", query,
tag->address, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 5, tag->attribute)) {
nlog_error("bind `%s` with attribute=`%i` fail: %s", query,
tag->attribute, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 6, tag->precision)) {
nlog_error("bind `%s` with precision=`%i` fail: %s", query,
tag->precision, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 7, tag->type)) {
nlog_error("bind `%s` with type=`%i` fail: %s", query, tag->type,
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_double(stmt, 8, tag->decimal)) {
nlog_error("bind `%s` with decimal=`%f` fail: %s", query,
tag->decimal, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK !=
sqlite3_bind_text(stmt, 9, tag->description, -1, NULL)) {
nlog_error("bind `%s` with description=`%s` fail: %s", query,
tag->description, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 10, val_str, -1, NULL)) {
nlog_error("bind `%s` with value=`%s` fail: %s", query, val_str,
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_DONE != sqlite3_step(stmt)) {
nlog_error("sqlite3_step fail: %s", sqlite3_errmsg(global_db));
goto error;
}
free(val_str);
val_str = NULL;
if (0 != put_tags(query, stmt, tags, n)) {
goto error;
}
if (SQLITE_OK != sqlite3_exec(global_db, "COMMIT", NULL, NULL, NULL)) {
@@ -828,10 +838,38 @@ error:
nlog_info("rollback transaction");
sqlite3_exec(global_db, "ROLLBACK", NULL, NULL, NULL);
sqlite3_finalize(stmt);
free(val_str);
return NEU_ERR_EINTERNAL;
}
static int collect_tag_info(sqlite3_stmt *stmt, UT_array **tags)
{
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),
.precision = sqlite3_column_int64(stmt, 3),
.type = sqlite3_column_int(stmt, 4),
.decimal = sqlite3_column_double(stmt, 5),
.description = (char *) sqlite3_column_text(stmt, 6),
};
utarray_push_back(*tags, &tag);
if (neu_tag_attribute_test(&tag, NEU_ATTRIBUTE_STATIC)) {
neu_tag_load_static_value(utarray_back(*tags),
(char *) sqlite3_column_text(stmt, 7));
}
step = sqlite3_step(stmt);
}
if (SQLITE_DONE != step) {
return -1;
}
return 0;
}
int neu_persister_load_tags(const char *driver_name, const char *group_name,
UT_array **tags)
{
@@ -860,27 +898,7 @@ int neu_persister_load_tags(const char *driver_name, const char *group_name,
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),
.precision = sqlite3_column_int64(stmt, 3),
.type = sqlite3_column_int(stmt, 4),
.decimal = sqlite3_column_double(stmt, 5),
.description = (char *) sqlite3_column_text(stmt, 6),
};
utarray_push_back(*tags, &tag);
if (neu_tag_attribute_test(&tag, NEU_ATTRIBUTE_STATIC)) {
neu_tag_load_static_value(utarray_back(*tags),
(char *) sqlite3_column_text(stmt, 7));
}
step = sqlite3_step(stmt);
}
if (SQLITE_DONE != step) {
if (0 != collect_tag_info(stmt, tags)) {
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
// do not set return code, return partial or empty result
}
@@ -1051,6 +1069,30 @@ static UT_icd group_info_icd = {
(dtor_f *) neu_persist_group_info_fini,
};
static int collect_group_info(sqlite3_stmt *stmt, UT_array **group_infos)
{
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step) {
neu_persist_group_info_t info = {};
char *name = strdup((char *) sqlite3_column_text(stmt, 0));
if (NULL == name) {
break;
}
info.name = name;
info.interval = sqlite3_column_int(stmt, 1);
utarray_push_back(*group_infos, &info);
step = sqlite3_step(stmt);
}
if (SQLITE_DONE != step) {
return -1;
}
return 0;
}
int neu_persister_load_groups(const char *driver_name, UT_array **group_infos)
{
sqlite3_stmt *stmt = NULL;
@@ -1069,22 +1111,7 @@ int neu_persister_load_groups(const char *driver_name, UT_array **group_infos)
goto error;
}
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step) {
neu_persist_group_info_t info = {};
char *name = strdup((char *) sqlite3_column_text(stmt, 0));
if (NULL == name) {
break;
}
info.name = name;
info.interval = sqlite3_column_int(stmt, 1);
utarray_push_back(*group_infos, &info);
step = sqlite3_step(stmt);
}
if (SQLITE_DONE != step) {
if (0 != collect_group_info(stmt, group_infos)) {
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
// do not set return code, return partial or empty result
}
@@ -1310,3 +1337,271 @@ int neu_persister_clear_templates()
// rely on foreign key constraints to remove groups and tags
return execute_sql(global_db, "DELETE FROM templates");
}
int neu_persister_store_template_group(const char * tmpl_name,
neu_persist_group_info_t *group_info)
{
return execute_sql(global_db,
"INSERT INTO template_groups ("
" tmpl_name, name, interval) VALUES (%Q, %Q, %u)",
tmpl_name, group_info->name,
(unsigned) group_info->interval);
}
int neu_persister_update_template_group(const char * tmpl_name,
neu_persist_group_info_t *group_info)
{
return execute_sql(
global_db,
"UPDATE template_groups SET interval=%i WHERE tmpl_name=%Q AND name=%Q",
group_info->interval, tmpl_name, group_info->name);
}
int neu_persister_load_template_groups(const char *tmpl_name,
UT_array ** group_infos)
{
sqlite3_stmt *stmt = NULL;
const char * query =
"SELECT name, interval FROM template_groups WHERE tmpl_name=?";
utarray_new(*group_infos, &group_info_icd);
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, tmpl_name, -1, NULL)) {
nlog_error("bind `%s` with `%s` fail: %s", query, tmpl_name,
sqlite3_errmsg(global_db));
goto error;
}
if (0 != collect_group_info(stmt, group_infos)) {
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
// do not set return code, return partial or empty result
}
sqlite3_finalize(stmt);
return 0;
error:
utarray_free(*group_infos);
*group_infos = NULL;
return NEU_ERR_EINTERNAL;
}
int neu_persister_delete_template_group(const char *tmpl_name,
const char *group_name)
{
// rely on foreign key constraints to delete tags
return execute_sql(
global_db, "DELETE FROM template_groups WHERE tmpl_name=%Q AND name=%Q",
tmpl_name, group_name);
}
int neu_persister_store_template_tags(const char * tmpl_name,
const char * group_name,
const neu_datatag_t *tags, size_t n)
{
char * val_str = NULL;
sqlite3_stmt *stmt = NULL;
const char * query = "INSERT INTO template_tags ("
" tmpl_name, group_name, name, address, attribute,"
" precision, type, decimal, description, value"
") VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)";
if (SQLITE_OK != sqlite3_exec(global_db, "BEGIN", NULL, NULL, NULL)) {
nlog_error("begin transaction fail: %s", sqlite3_errmsg(global_db));
return NEU_ERR_EINTERNAL;
}
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, tmpl_name, -1, NULL)) {
nlog_error("bind `%s` with tmpl_name=`%s` fail: %s", query, tmpl_name,
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
nlog_error("bind `%s` with group_name=`%s` fail: %s", query, group_name,
sqlite3_errmsg(global_db));
goto error;
}
if (0 != put_tags(query, stmt, tags, n)) {
goto error;
}
if (SQLITE_OK != sqlite3_exec(global_db, "COMMIT", NULL, NULL, NULL)) {
nlog_error("commit transaction fail: %s", sqlite3_errmsg(global_db));
goto error;
}
sqlite3_finalize(stmt);
return 0;
error:
nlog_info("rollback transaction");
sqlite3_exec(global_db, "ROLLBACK", NULL, NULL, NULL);
sqlite3_finalize(stmt);
free(val_str);
return NEU_ERR_EINTERNAL;
}
int neu_persister_load_template_tags(const char *tmpl_name,
const char *group_name, UT_array **tags)
{
sqlite3_stmt *stmt = NULL;
const char * query = "SELECT name, address, attribute, precision, type, "
"decimal, description, value "
"FROM tags WHERE tmpl_name=? AND group_name=? "
"ORDER BY rowid ASC";
utarray_new(*tags, neu_tag_get_icd());
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, tmpl_name, -1, NULL)) {
nlog_error("bind `%s` with `%s` fail: %s", query, tmpl_name,
sqlite3_errmsg(global_db));
goto error;
}
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(global_db));
goto error;
}
if (0 != collect_tag_info(stmt, tags)) {
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_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_template_tags(const char * tmpl_name,
const char * group_name,
const neu_datatag_t *tags, size_t n)
{
sqlite3_stmt *stmt = NULL;
const char * query = "UPDATE template_tags SET"
" address=?4, attribute=?5, precision=?6, type=?7,"
" decimal=?8, description=?9, value=?10 "
"WHERE tmpl_name=?1 AND group_name=?2 AND name=?3";
if (SQLITE_OK != sqlite3_exec(global_db, "BEGIN", NULL, NULL, NULL)) {
nlog_error("begin transaction fail: %s", sqlite3_errmsg(global_db));
return NEU_ERR_EINTERNAL;
}
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, tmpl_name, -1, NULL)) {
nlog_error("bind `%s` with tmpl_name=`%s` fail: %s", query, tmpl_name,
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
nlog_error("bind `%s` with group_name=`%s` fail: %s", query, group_name,
sqlite3_errmsg(global_db));
goto error;
}
if (0 != put_tags(query, stmt, tags, n)) {
goto error;
}
if (SQLITE_OK != sqlite3_exec(global_db, "COMMIT", NULL, NULL, NULL)) {
nlog_error("commit transaction fail: %s", sqlite3_errmsg(global_db));
goto error;
}
sqlite3_finalize(stmt);
return 0;
error:
nlog_info("rollback transaction");
sqlite3_exec(global_db, "ROLLBACK", NULL, NULL, NULL);
sqlite3_finalize(stmt);
return NEU_ERR_EINTERNAL;
}
int neu_persister_delete_template_tags(const char * tmpl_name,
const char * group_name,
const char *const *tags, size_t n_tag)
{
sqlite3_stmt *stmt = NULL;
const char * query = "DELETE FROM template_tags WHERE tmpl_name=? AND "
"group_name=? AND name=?";
if (SQLITE_OK != sqlite3_exec(global_db, "BEGIN", NULL, NULL, NULL)) {
nlog_error("begin transaction fail: %s", sqlite3_errmsg(global_db));
return NEU_ERR_EINTERNAL;
}
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, tmpl_name, -1, NULL)) {
nlog_error("bind `%s` with tmpl_name=`%s` fail: %s", query, tmpl_name,
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
nlog_error("bind `%s` with group_name=`%s` fail: %s", query, group_name,
sqlite3_errmsg(global_db));
goto error;
}
for (size_t i = 0; i < n_tag; ++i) {
sqlite3_reset(stmt);
if (SQLITE_OK != sqlite3_bind_text(stmt, 3, tags[i], -1, NULL)) {
nlog_error("bind `%s` with tmpl_name=`%s` fail: %s", query,
tmpl_name, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_DONE != sqlite3_step(stmt)) {
nlog_error("sqlite3_step fail: %s", sqlite3_errmsg(global_db));
goto error;
}
}
if (SQLITE_OK != sqlite3_exec(global_db, "COMMIT", NULL, NULL, NULL)) {
nlog_error("commit transaction fail: %s", sqlite3_errmsg(global_db));
goto error;
}
sqlite3_finalize(stmt);
return 0;
error:
nlog_info("rollback transaction");
sqlite3_exec(global_db, "ROLLBACK", NULL, NULL, NULL);
sqlite3_finalize(stmt);
return NEU_ERR_EINTERNAL;
}