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

Merge pull request #1552 from ZhouBox/main

Browse Source 
feat:dynamic plugin management
This commit is contained in:
fengzeroz
2023-10-08 14:32:12 +08:00
committed by GitHub
parent 0d2992257b bc67be3386
commit a77403712b
15 changed files with 744 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+11 -5
View File
@@ -93,6 +93,7 @@ typedef enum neu_reqresp_type {
NEU_REQ_ADD_PLUGIN,
NEU_REQ_DEL_PLUGIN,
NEU_REQ_UPDATE_PLUGIN,
NEU_REQ_GET_PLUGIN,
NEU_RESP_GET_PLUGIN,
@@ -198,10 +199,11 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_GET_TAG] = "NEU_REQ_GET_TAG",
[NEU_RESP_GET_TAG] = "NEU_RESP_GET_TAG",
[NEU_REQ_ADD_PLUGIN] = "NEU_REQ_ADD_PLUGIN",
[NEU_REQ_DEL_PLUGIN] = "NEU_REQ_DEL_PLUGIN",
[NEU_REQ_GET_PLUGIN] = "NEU_REQ_GET_PLUGIN",
[NEU_RESP_GET_PLUGIN] = "NEU_RESP_GET_PLUGIN",
[NEU_REQ_ADD_PLUGIN] = "NEU_REQ_ADD_PLUGIN",
[NEU_REQ_DEL_PLUGIN] = "NEU_REQ_DEL_PLUGIN",
[NEU_REQ_UPDATE_PLUGIN] = "NEU_REQ_UPDATE_PLUGIN",
[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",
@@ -272,9 +274,13 @@ typedef struct {
} neu_req_node_init_t, neu_req_node_uninit_t, neu_resp_node_uninit_t;
typedef struct neu_req_add_plugin {
char library[NEU_PLUGIN_LIBRARY_LEN];
char library[NEU_PLUGIN_LIBRARY_LEN];
char *schema_file;
char *so_file;
} neu_req_add_plugin_t;
typedef neu_req_add_plugin_t neu_req_update_plugin_t;
typedef struct neu_req_del_plugin {
char plugin[NEU_PLUGIN_NAME_LEN];
} neu_req_del_plugin_t;
include/neuron/errcodes.h
+5
View File
@@ -92,6 +92,11 @@ typedef enum {
NEU_ERR_LIBRARY_SYSTEM_NOT_ALLOW_DEL = 2306,
NEU_ERR_LIBRARY_NOT_ALLOW_CREATE_INSTANCE = 2307,
NEU_ERR_LIBRARY_ARCH_NOT_SUPPORT = 2308,
NEU_ERR_LIBRARY_IN_USE = 2309,
NEU_ERR_LIBRARY_ADD_FAIL = 2310,
NEU_ERR_LIBRARY_MODULE_ALREADY_EXIST = 2311,
NEU_ERR_LIBRARY_MODULE_NOT_EXISTS = 2312,
NEU_ERR_LIBRARY_MODULE_KIND_NOT_SUPPORT = 2313,
NEU_ERR_LICENSE_NOT_FOUND = 2400,
NEU_ERR_LICENSE_INVALID = 2401,
include/neuron/persist/persist.h
+5
View File
@@ -457,6 +457,11 @@ int neu_persister_delete_template_tags(const char * tmpl_name,
const char * group_name,
const char *const *tags, size_t n_tag);
char *neu_persister_save_file_tmp(const char *file_data, uint32_t len,
const char *suffix);
bool neu_persister_library_exists(const char *library);
#ifdef __cplusplus
}
#endif
plugins/restful/handle.c
+6
View File
@@ -272,6 +272,12 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/plugin",
.value.handler = handle_del_plugin,
},
{
.method = NEU_HTTP_METHOD_PUT,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/plugin",
.value.handler = handle_update_plugin,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
plugins/restful/plugin_handle.c
+27 -1
View File
@@ -41,7 +41,33 @@ void handle_add_plugin(nng_aio *aio)
header.ctx = aio;
header.type = NEU_REQ_ADD_PLUGIN;
strcpy(cmd.library, req->library);
ret = neu_plugin_op(plugin, header, &cmd);
cmd.schema_file = req->schema_file;
cmd.so_file = req->so_file;
ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
}
})
}
void handle_update_plugin(nng_aio *aio)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_update_plugin_req_t, neu_json_decode_add_plugin_req, {
int ret = 0;
neu_reqresp_head_t header = { 0 };
neu_req_update_plugin_t cmd = { 0 };
header.ctx = aio;
header.type = NEU_REQ_UPDATE_PLUGIN;
strcpy(cmd.library, req->library);
cmd.schema_file = req->schema_file;
cmd.so_file = req->so_file;
ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
plugins/restful/plugin_handle.h
+1
View File
@@ -23,6 +23,7 @@
void handle_add_plugin(nng_aio *aio);
void handle_del_plugin(nng_aio *aio);
void handle_update_plugin(nng_aio *aio);
void handle_get_plugin(nng_aio *aio);
void handle_get_plugin_resp(nng_aio *aio, neu_resp_get_plugin_t *plugins);
src/adapter/adapter.c
+21
View File
@@ -18,6 +18,7 @@
**/
#include <assert.h>
#include <dlfcn.h>
#include <errno.h>
#include <inttypes.h>
#include <pthread.h>
@@ -207,6 +208,16 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
}
}
bool neu_adapter_reset(neu_adapter_t *adapter, neu_adapter_info_t *info)
{
if (adapter->state == NEU_NODE_RUNNING_STATE_RUNNING) {
return false;
}
adapter->module = info->module;
adapter->handle = info->handle;
return true;
}
int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name)
{
char *name = strdup(new_name);
@@ -1035,6 +1046,13 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
free(adapter);
}
void neu_adapter_handle_close(neu_adapter_t *adapter)
{
dlclose(adapter->handle);
adapter->handle = NULL;
adapter->module = NULL;
}
int neu_adapter_uninit(neu_adapter_t *adapter)
{
if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
@@ -1355,6 +1373,9 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_DEL_PLUGIN:
data_size = sizeof(neu_req_del_plugin_t);
break;
case NEU_REQ_UPDATE_PLUGIN:
data_size = sizeof(neu_req_update_plugin_t);
break;
case NEU_REQ_GET_PLUGIN:
data_size = sizeof(neu_req_get_plugin_t);
break;
src/adapter/adapter_internal.h
+2
View File
@@ -67,6 +67,7 @@ int neu_adapter_error();
void neu_adapter_set_error(int error);
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load);
bool neu_adapter_reset(neu_adapter_t *adapter, neu_adapter_info_t *info);
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);
@@ -80,6 +81,7 @@ neu_tag_cache_type_e neu_adapter_get_tag_cache_type(neu_adapter_t *adapter);
int neu_adapter_uninit(neu_adapter_t *adapter);
void neu_adapter_destroy(neu_adapter_t *adapter);
void neu_adapter_handle_close(neu_adapter_t *adapter);
neu_event_timer_t *neu_adapter_add_timer(neu_adapter_t * adapter,
neu_event_timer_param_t param);
src/core/manager.c
+476 -9
View File
@@ -16,7 +16,9 @@
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include <errno.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <unistd.h>
@@ -26,18 +28,21 @@
#include "event/event.h"
#include "persist/persist.h"
#include "utils/base64.h"
#include "utils/log.h"
#include "utils/time.h"
#include "adapter.h"
#include "adapter/adapter_internal.h"
#include "adapter/driver/driver_internal.h"
#include "argparse.h"
#include "errcodes.h"
#include "node_manager.h"
#include "plugin_manager.h"
#include "storage.h"
#include "subscribe.h"
#include "template_manager.h"
#include "manager.h"
#include "manager_internal.h"
@@ -64,6 +69,18 @@ static int update_timestamp(void *usr_data);
static void start_single_adapter(neu_manager_t *manager, const char *name,
const char *plugin_name, bool display);
static int template_each_cb(neu_template_t *tmpl, void *data);
static char *file_save_tmp(const char *data, const char *suffix);
static bool mv_tmp_library_file(neu_plugin_kind_e kind, const char *tmp_path,
const char *library);
static bool mv_tmp_schema_file(const char *tmp_path, const char *schema);
typedef struct {
char *plugin;
bool ret;
} template_each_cb_data_t;
neu_manager_t *neu_manager_create()
{
int rv = 0;
@@ -258,9 +275,175 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case NEU_REQ_ADD_PLUGIN: {
neu_req_add_plugin_t *cmd = (neu_req_add_plugin_t *) &header[1];
int error = neu_manager_add_plugin(manager, cmd->library);
neu_resp_error_t e = { .error = error };
neu_req_add_plugin_t *cmd = (neu_req_add_plugin_t *) &header[1];
neu_resp_error_t e = { 0 };
neu_plugin_kind_e kind = -1;
char schema[64] = { 0 };
if (neu_persister_library_exists(cmd->library)) {
nlog_warn("library %s had exited", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_NAME_CONFLICT;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
break;
}
char *so_tmp_path = file_save_tmp(cmd->so_file, "so");
if (so_tmp_path == NULL) {
nlog_warn("library %s so file save tmp fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_BODY_IS_WRONG;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
break;
}
neu_plugin_instance_t ins = { 0 };
if (neu_plugin_manager_create_instance_by_path(manager->plugin_manager,
so_tmp_path, &ins)) {
char module_name[64] = { 0 };
kind = ins.module->kind;
strncpy(module_name, ins.module->module_name, sizeof(module_name));
strncpy(schema, ins.module->schema, sizeof(schema));
nlog_debug("library %s, module_name %s, schema:%s", cmd->library,
module_name, schema);
neu_plugin_manager_destroy_instance(manager->plugin_manager, &ins);
if (neu_plugin_manager_exists(manager->plugin_manager,
module_name)) {
nlog_warn("%s module name had existed", module_name);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_MODULE_ALREADY_EXIST;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
break;
}
} else {
nlog_warn("library %s so file is not a vaild file", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_MODULE_INVALID;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
break;
}
if (kind != NEU_PLUGIN_KIND_CUSTOM && kind != NEU_PLUGIN_KIND_SYSTEM) {
nlog_warn("library %s kind no support", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_MODULE_KIND_NOT_SUPPORT;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
break;
}
char *schema_tmp_path = file_save_tmp(cmd->schema_file, "json");
if (schema_tmp_path == NULL) {
nlog_warn("library %s schema file save tmp fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_BODY_IS_WRONG;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
break;
}
if (!mv_tmp_library_file(kind, so_tmp_path, cmd->library)) {
nlog_warn("library %s mv library tmp file fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_ADD_FAIL;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
free(schema_tmp_path);
break;
}
if (!mv_tmp_schema_file(schema_tmp_path, schema)) {
nlog_warn("library %s schema file save schema fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_ADD_FAIL;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
free(schema_tmp_path);
break;
}
int error = neu_manager_add_plugin(manager, cmd->library);
if (error == NEU_ERR_SUCCESS) {
manager_strorage_plugin(manager);
}
header->type = NEU_RESP_ERROR;
e.error = error;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(so_tmp_path);
free(cmd->so_file);
free(cmd->schema_file);
free(schema_tmp_path);
break;
}
case NEU_REQ_DEL_PLUGIN: {
neu_req_del_plugin_t *cmd = (neu_req_del_plugin_t *) &header[1];
neu_resp_error_t e = { 0 };
UT_array *nodes = neu_manager_get_nodes(
manager, NEU_NA_TYPE_DRIVER | NEU_NA_TYPE_APP, cmd->plugin, "");
if (nodes != NULL) {
if (utarray_len(nodes) > 0) {
utarray_free(nodes);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_IN_USE;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
break;
}
}
template_each_cb_data_t data = { .plugin = cmd->plugin, .ret = false };
neu_template_manager_for_each(manager->template_manager,
template_each_cb, (void *) &data);
if (data.ret) {
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_IN_USE;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
break;
}
int error = neu_manager_del_plugin(manager, cmd->plugin);
e.error = error;
if (error == NEU_ERR_SUCCESS) {
manager_strorage_plugin(manager);
@@ -271,18 +454,209 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &e);
break;
}
case NEU_REQ_DEL_PLUGIN: {
neu_req_del_plugin_t *cmd = (neu_req_del_plugin_t *) &header[1];
int error = neu_manager_del_plugin(manager, cmd->plugin);
neu_resp_error_t e = { .error = error };
case NEU_REQ_UPDATE_PLUGIN: {
neu_req_update_plugin_t *cmd = (neu_req_update_plugin_t *) &header[1];
neu_resp_error_t e = { 0 };
neu_plugin_kind_e kind = -1;
char module_name[64] = { 0 };
char schema[64] = { 0 };
if (error == NEU_ERR_SUCCESS) {
manager_strorage_plugin(manager);
if (!neu_persister_library_exists(cmd->library)) {
nlog_warn("library %s no exited", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_NOT_FOUND;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
break;
}
char *so_tmp_path = file_save_tmp(cmd->so_file, "so");
if (so_tmp_path == NULL) {
nlog_warn("library %s so file save tmp fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_BODY_IS_WRONG;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
break;
}
neu_plugin_instance_t ins = { 0 };
if (neu_plugin_manager_create_instance_by_path(manager->plugin_manager,
so_tmp_path, &ins)) {
kind = ins.module->kind;
strncpy(module_name, ins.module->module_name, sizeof(module_name));
strncpy(schema, ins.module->schema, sizeof(schema));
nlog_debug("library %s, module_name %s, schema:%s", cmd->library,
module_name, schema);
neu_plugin_manager_destroy_instance(manager->plugin_manager, &ins);
if (!neu_plugin_manager_exists(manager->plugin_manager,
module_name)) {
nlog_warn("library %s plugin name no existed", module_name);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_MODULE_NOT_EXISTS;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
break;
}
} else {
nlog_warn("library %s so file is not a vaild file", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_MODULE_INVALID;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
break;
}
if (kind != NEU_PLUGIN_KIND_CUSTOM && kind != NEU_PLUGIN_KIND_SYSTEM) {
nlog_warn("library %s kind no support", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_MODULE_KIND_NOT_SUPPORT;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
break;
}
char *schema_tmp_path = file_save_tmp(cmd->schema_file, "json");
if (schema_tmp_path == NULL) {
nlog_warn("library %s schema file save tmp fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_BODY_IS_WRONG;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
break;
}
UT_array *nodes = neu_manager_get_nodes(
manager, NEU_NA_TYPE_DRIVER | NEU_NA_TYPE_APP, module_name, "");
UT_array *running_adapters = NULL;
UT_icd icd = { sizeof(neu_adapter_t *), NULL, NULL, NULL };
utarray_new(running_adapters, &icd);
if (nodes != NULL) {
if (utarray_len(nodes) > 0) {
utarray_foreach(nodes, neu_resp_node_info_t *, node)
{
neu_adapter_t *adapter = neu_node_manager_find(
manager->node_manager, node->node);
if (neu_adapter_get_state(adapter).running ==
NEU_NODE_RUNNING_STATE_RUNNING) {
neu_adapter_stop(adapter);
utarray_push_back(running_adapters, &adapter);
}
neu_adapter_handle_close(adapter);
}
}
}
if (!neu_plugin_manager_remove_library(manager->plugin_manager,
cmd->library)) {
nlog_warn("library %s src file remove fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_ADD_FAIL;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
break;
}
if (!mv_tmp_library_file(kind, so_tmp_path, cmd->library)) {
nlog_warn("library %s mv library tmp file fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_ADD_FAIL;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
free(schema_tmp_path);
break;
}
if (!mv_tmp_schema_file(schema_tmp_path, schema)) {
nlog_warn("library %s schema file save schema fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_ADD_FAIL;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
free(schema_tmp_path);
break;
}
if (nodes != NULL) {
if (utarray_len(nodes) > 0) {
neu_adapter_info_t adapter_info = { 0 };
int ret = neu_plugin_manager_create_instance(
manager->plugin_manager, module_name, &ins);
if (ret != 0) {
nlog_warn("library %s open new library fail", cmd->library);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_FAILED_TO_OPEN;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
free(schema_tmp_path);
break;
}
utarray_foreach(nodes, neu_resp_node_info_t *, node)
{
neu_adapter_t *adapter = neu_node_manager_find(
manager->node_manager, node->node);
adapter_info.handle = ins.handle;
adapter_info.module = ins.module;
neu_adapter_reset(adapter, &adapter_info);
}
}
utarray_free(nodes);
}
if (running_adapters != NULL) {
if (utarray_len(running_adapters) > 0) {
utarray_foreach(running_adapters, neu_adapter_t **, adapter)
{
neu_adapter_start(*adapter);
}
}
utarray_free(running_adapters);
}
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_SUCCESS;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
break;
}
case NEU_REQ_GET_PLUGIN: {
@@ -1341,4 +1715,97 @@ static int update_timestamp(void *usr_data)
(void) usr_data;
global_timestamp = neu_time_ms();
return 0;
}
static int template_each_cb(neu_template_t *tmpl, void *data)
{
template_each_cb_data_t *cb_data = (template_each_cb_data_t *) data;
if (strcmp(neu_template_plugin(tmpl), cb_data->plugin) == 0) {
cb_data->ret = true;
return 1;
}
return 0;
}
static char *file_save_tmp(const char *data, const char *suffix)
{
int d_len = 0;
char *file = (char *) neu_decode64(&d_len, data);
if (file == NULL) {
nlog_warn("library so file decode64 ret is NULL");
return NULL;
}
if (d_len == 0) {
nlog_warn("library so file decode64 len is 0");
free(file);
return NULL;
}
char *tmp_path = neu_persister_save_file_tmp(file, d_len, suffix);
if (tmp_path == NULL) {
free(file);
return NULL;
}
free(file);
return tmp_path;
}
static bool mv_tmp_library_file(neu_plugin_kind_e kind, const char *tmp_path,
const char *library)
{
char file_name[128] = { 0 };
char path_name[128] = { 0 };
if (kind == NEU_PLUGIN_KIND_CUSTOM) {
snprintf(file_name, sizeof(file_name), "%s/custom/%s", g_plugin_dir,
library);
snprintf(path_name, sizeof(path_name), "%s/custom", g_plugin_dir);
} else if (kind == NEU_PLUGIN_KIND_SYSTEM) {
snprintf(file_name, sizeof(file_name), "%s/system/%s", g_plugin_dir,
library);
snprintf(path_name, sizeof(path_name), "%s/system", g_plugin_dir);
} else {
return false;
}
struct stat st;
if (stat(path_name, &st) == -1) {
if (mkdir(path_name, 0700) == -1) {
nlog_error("%s mkdir fail", path_name);
return false;
}
}
if (rename(tmp_path, file_name) != 0) {
nlog_error("%s rename %s fail, err:%s", tmp_path, file_name,
strerror(errno));
return false;
} else {
return true;
}
return false;
}
static bool mv_tmp_schema_file(const char *tmp_path, const char *schema)
{
char file_name[128] = { 0 };
snprintf(file_name, sizeof(file_name), "%s/schema/%s.json", g_plugin_dir,
schema);
if (rename(tmp_path, file_name) != 0) {
nlog_error("%s rename %s fail, err:%s", tmp_path, file_name,
strerror(errno));
return false;
} else {
return true;
}
return false;
}
src/core/plugin_manager.c
+93 -18
View File
@@ -19,6 +19,7 @@
#include <assert.h>
#include <dlfcn.h>
#include <stdlib.h>
#include <unistd.h>
#include "errcodes.h"
#include "utils/log.h"
@@ -79,19 +80,34 @@ void neu_plugin_manager_destroy(neu_plugin_manager_t *mgr)
int neu_plugin_manager_add(neu_plugin_manager_t *mgr,
const char * plugin_lib_name)
{
char lib_path[256] = { 0 };
void * handle = NULL;
void * module = NULL;
neu_plugin_module_t *pm = NULL;
plugin_entity_t * plugin = NULL;
char lib_path[256] = { 0 };
void * handle = NULL;
void * module = NULL;
neu_plugin_module_t *pm = NULL;
plugin_entity_t * plugin = NULL;
char lib_paths[3][64] = { 0 };
snprintf(lib_paths[0], sizeof(lib_paths[0]), "%s", g_plugin_dir);
snprintf(lib_paths[1], sizeof(lib_paths[1]), "%s/system", g_plugin_dir);
snprintf(lib_paths[2], sizeof(lib_paths[2]), "%s/custom", g_plugin_dir);
assert(strlen(plugin_lib_name) <= NEU_PLUGIN_LIBRARY_LEN);
snprintf(lib_path, sizeof(lib_path) - 1, "./plugins/%s", plugin_lib_name);
handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE);
for (size_t i = 0; i < sizeof(lib_paths) / sizeof(lib_paths[0]); i++) {
snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
plugin_lib_name);
handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE);
if (handle == NULL) {
nlog_warn("failed to open library %s, error: %s", lib_path,
dlerror());
} else {
break;
}
}
if (handle == NULL) {
nlog_warn("failed to open library %s, error: %s", lib_path, dlerror());
return NEU_ERR_LIBRARY_FAILED_TO_OPEN;
}
@@ -280,23 +296,35 @@ int neu_plugin_manager_create_instance(neu_plugin_manager_t * mgr,
const char * plugin_name,
neu_plugin_instance_t *instance)
{
plugin_entity_t *plugin = NULL;
plugin_entity_t *plugin = NULL;
char lib_paths[3][64] = { 0 };
snprintf(lib_paths[0], sizeof(lib_paths[0]), "%s", g_plugin_dir);
snprintf(lib_paths[1], sizeof(lib_paths[1]), "%s/system", g_plugin_dir);
snprintf(lib_paths[2], sizeof(lib_paths[2]), "%s/custom", g_plugin_dir);
HASH_FIND_STR(mgr->plugins, plugin_name, plugin);
if (plugin != NULL) {
char lib_path[256] = { 0 };
snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", g_plugin_dir,
plugin->lib_name);
instance->handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE);
assert(instance->handle != NULL);
for (size_t i = 0; i < sizeof(lib_paths) / sizeof(lib_paths[0]); i++) {
char lib_path[256] = { 0 };
instance->module = (neu_plugin_module_t *) dlsym(instance->handle,
"neu_plugin_module");
assert(instance->module != NULL);
return 0;
snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
plugin->lib_name);
instance->handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE);
if (instance->handle == NULL) {
continue;
}
instance->module = (neu_plugin_module_t *) dlsym(
instance->handle, "neu_plugin_module");
assert(instance->module != NULL);
return 0;
}
}
assert(instance->handle != NULL);
return -1;
}
@@ -338,3 +366,50 @@ static void entity_free(plugin_entity_t *entity)
free(entity->description_zh);
free(entity);
}
bool neu_plugin_manager_create_instance_by_path(neu_plugin_manager_t *mgr,
const char *plugin_path,
neu_plugin_instance_t *instance)
{
(void) mgr;
instance->handle = dlopen(plugin_path, RTLD_NOW | RTLD_NODELETE);
if (instance->handle == NULL) {
return false;
}
instance->module =
(neu_plugin_module_t *) dlsym(instance->handle, "neu_plugin_module");
if (instance->module == NULL) {
dlclose(instance->handle);
instance->handle = NULL;
return false;
}
return true;
}
bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
const char * library)
{
bool ret = true;
(void) mgr;
char lib_paths[3][64] = { 0 };
snprintf(lib_paths[0], sizeof(lib_paths[0]), "%s", g_plugin_dir);
snprintf(lib_paths[1], sizeof(lib_paths[1]), "%s/system", g_plugin_dir);
snprintf(lib_paths[2], sizeof(lib_paths[2]), "%s/custom", g_plugin_dir);
char lib_path[256] = { 0 };
for (size_t i = 0; i < sizeof(lib_paths) / sizeof(lib_paths[0]); i++) {
snprintf(lib_path, sizeof(lib_path), "%s/%s", lib_paths[i], library);
if (access(lib_path, F_OK) != -1 && remove(lib_path) != 0) {
nlog_debug("library %s remove fail", lib_path);
ret = false;
break;
}
}
return ret;
}
src/core/plugin_manager.h
+7
View File
@@ -63,4 +63,11 @@ void neu_plugin_manager_load_static(neu_plugin_manager_t * mgr,
const char * plugin_name,
neu_plugin_instance_t *instance);
bool neu_plugin_manager_create_instance_by_path(
neu_plugin_manager_t *mgr, const char *plugin_path,
neu_plugin_instance_t *instance);
bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
const char * library);
#endif
src/parser/neu_json_plugin.c
+14 -4
View File
@@ -43,16 +43,26 @@ int neu_json_decode_add_plugin_req(char * buf,
json_obj = neu_json_decode_new(buf);
neu_json_elem_t req_elems[] = { {
.name = "library",
.t = NEU_JSON_STR,
} };
.name = "library",
.t = NEU_JSON_STR,
},
{
.name = "schema_file",
.t = NEU_JSON_STR,
},
{
.name = "so_file",
.t = NEU_JSON_STR,
} };
ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
req_elems);
if (ret != 0) {
goto decode_fail;
}
req->library = req_elems[0].v.val_str;
req->library = req_elems[0].v.val_str;
req->schema_file = req_elems[1].v.val_str;
req->so_file = req_elems[2].v.val_str;
*result = req;
goto decode_exit;
src/parser/neu_json_plugin.h
+4
View File
@@ -33,8 +33,12 @@ extern "C" {
typedef struct {
char *library;
char *schema_file;
char *so_file;
} neu_json_add_plugin_req_t;
typedef neu_json_add_plugin_req_t neu_json_update_plugin_req_t;
int neu_json_decode_add_plugin_req(char * buf,
neu_json_add_plugin_req_t **result);
void neu_json_decode_add_plugin_req_free(neu_json_add_plugin_req_t *req);
src/persist/persist.c
+67 -4
View File
@@ -35,6 +35,7 @@
#include "errcodes.h"
#include "utils/asprintf.h"
#include "utils/log.h"
#include "utils/time.h"
#include "argparse.h"
#include "persist/json/persist_json_plugin.h"
@@ -56,6 +57,7 @@ static sqlite3 * global_db = NULL;
pthread_rwlock_t global_rwlock = PTHREAD_RWLOCK_INITIALIZER;
static int global_node_count = 0;
static int global_tag_count = 0;
static const char *tmp_path = "tmp";
static inline bool ends_with(const char *str, const char *suffix)
{
@@ -700,10 +702,10 @@ int neu_persister_store_plugins(UT_array *plugin_infos)
utarray_foreach(plugin_infos, neu_resp_plugin_info_t *, plugin)
{
if (NEU_PLUGIN_KIND_SYSTEM == plugin->kind) {
// filter out system plugins
continue;
}
// if (NEU_PLUGIN_KIND_SYSTEM == plugin->kind) {
// // filter out system plugins
// continue;
// }
plugin_resp.plugins[index] = plugin->library;
index++;
}
@@ -1754,3 +1756,64 @@ error:
sqlite3_finalize(stmt);
return NEU_ERR_EINTERNAL;
}
char *neu_persister_save_file_tmp(const char *file_data, uint32_t len,
const char *suffix)
{
struct stat st;
if (stat(tmp_path, &st) == -1) {
if (mkdir(tmp_path, 0700) == -1) {
nlog_error("%s mkdir fail", tmp_path);
return NULL;
}
}
char *file_name = calloc(1, 128);
snprintf(file_name, 128, "%s/%" PRId64 ".%s", tmp_path, neu_time_ms(),
suffix);
FILE *fp = NULL;
fp = fopen(file_name, "wb+");
if (fp == NULL) {
nlog_error("not create tmp file: %s, err:%s", file_name,
strerror(errno));
free(file_name);
return NULL;
}
size_t size = fwrite(file_data, 1, len, fp);
if (size < len) {
fclose(fp);
free(file_name);
return NULL;
}
fclose(fp);
return file_name;
}
bool neu_persister_library_exists(const char *library)
{
bool ret = false;
UT_array *plugin_infos = NULL;
int rv = neu_persister_load_plugins(&plugin_infos);
if (rv != 0) {
return ret;
}
utarray_foreach(plugin_infos, char **, name)
{
if (strcmp(library, *name) == 0) {
ret = true;
break;
}
}
utarray_foreach(plugin_infos, char **, name) { free(*name); }
utarray_free(plugin_infos);
return ret;
}
src/utils/http.c
+5
View File
@@ -343,6 +343,11 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_TEMPLATE_NAME_TOO_LONG:
case NEU_ERR_PLUGIN_NAME_TOO_LONG:
case NEU_ERR_PLUGIN_NOT_SUPPORT_TEMPLATE:
case NEU_ERR_LIBRARY_IN_USE:
case NEU_ERR_LIBRARY_ADD_FAIL:
case NEU_ERR_LIBRARY_MODULE_ALREADY_EXIST:
case NEU_ERR_LIBRARY_MODULE_NOT_EXISTS:
case NEU_ERR_LIBRARY_MODULE_KIND_NOT_SUPPORT:
status = NNG_HTTP_STATUS_BAD_REQUEST;
break;
case NEU_ERR_FILE_NOT_EXIST: