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

Merge pull request #2282 from fengzeroz/main

Browse Source 
check plugin schema exist
This commit is contained in:
fengzeroz
2024-11-04 15:58:34 +08:00
committed by GitHub
parent 654a409538 c9ce39747e
commit 1e0f51058e
10 changed files with 140 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/msg.h
+15
View File
@@ -122,6 +122,9 @@ typedef enum neu_reqresp_type {
NEU_REQ_SCAN_TAGS,
NEU_RESP_SCAN_TAGS,
NEU_REQ_CHECK_SCHEMA,
NEU_RESP_CHECK_SCHEMA,
} neu_reqresp_type_e;
static const char *neu_reqresp_type_string_t[] = {
@@ -204,6 +207,9 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_SCAN_TAGS] = "NEU_REQ_SCAN_TAGS",
[NEU_RESP_SCAN_TAGS] = "NEU_RESP_SCAN_TAGS",
[NEU_REQ_CHECK_SCHEMA] = "NEU_REQ_CHECK_SCHEMA",
[NEU_RESP_CHECK_SCHEMA] = "NEU_RESP_CHECK_SCHEMA",
};
inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type)
@@ -232,6 +238,15 @@ typedef struct neu_resp_error {
int error;
} neu_resp_error_t;
typedef struct neu_req_check_schema {
char schema[NEU_PLUGIN_NAME_LEN];
} neu_req_check_schema_t;
typedef struct {
bool exist;
char schema[NEU_PLUGIN_NAME_LEN];
} neu_resp_check_schema_t;
typedef struct {
char node[NEU_NODE_NAME_LEN];
neu_node_running_state_e state;
plugins/restful/normal_handle.c
+74 -47
View File
@@ -289,10 +289,9 @@ static inline const char *plugin_name_to_schema_name(const char *name)
void handle_get_plugin_schema(nng_aio *aio)
{
size_t len = 0;
char * schema_path = NULL;
char param[NEU_PLUGIN_NAME_LEN] = { 0 };
const char *schema_name = param;
char param[NEU_PLUGIN_NAME_LEN] = { 0 };
const char * schema_name = param;
neu_plugin_t *plugin = neu_rest_get_plugin();
NEU_VALIDATE_JWT(aio);
@@ -302,56 +301,84 @@ void handle_get_plugin_schema(nng_aio *aio)
rv = neu_http_get_param_str(aio, "plugin_name", param, sizeof(param));
schema_name = plugin_name_to_schema_name(param);
}
if (rv < 0) {
if (rv < 0 || strlen(schema_name) >= NEU_PLUGIN_NAME_LEN) {
neu_http_bad_request(aio, "{\"error\": 1002}");
return;
}
if (0 > neu_asprintf(&schema_path, "%s/schema/%s.json", g_plugin_dir,
schema_name)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
int ret = 0;
neu_reqresp_head_t header = { 0 };
neu_req_check_schema_t cmd = { 0 };
header.otel_trace_type = NEU_OTEL_TRACE_TYPE_REST_COMM;
header.ctx = aio;
header.type = NEU_REQ_CHECK_SCHEMA;
strcpy(cmd.schema, schema_name);
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_get_plugin_schema_resp(nng_aio *aio, neu_resp_check_schema_t *resp)
{
if (resp->exist) {
char * schema_path = NULL;
size_t len = 0;
char * buf = NULL;
if (0 > neu_asprintf(&schema_path, "%s/schema/%s.json", g_plugin_dir,
resp->schema)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, error_code.error, result_error);
});
return;
}
buf = file_string_read(&len, schema_path);
if (NULL == buf) {
free(schema_path);
if (0 > neu_asprintf(&schema_path, "%s/custom/schema/%s.json",
g_plugin_dir, resp->schema)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, error_code.error, result_error);
});
return;
}
buf = file_string_read(&len, schema_path);
}
if (NULL == buf) {
free(schema_path);
if (0 > neu_asprintf(&schema_path, "%s/system/schema/%s.json",
g_plugin_dir, resp->schema)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, error_code.error, result_error);
});
return;
}
buf = file_string_read(&len, schema_path);
}
if (NULL == buf) {
nlog_info("open %s error: %d", schema_path, errno);
neu_http_not_found(aio, "{\"status\": \"error\"}");
free(schema_path);
return;
}
neu_http_ok(aio, buf);
free(buf);
free(schema_path);
} else {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PLUGIN_NOT_FOUND, {
neu_http_response(aio, error_code.error, result_error);
});
return;
}
char *buf = NULL;
buf = file_string_read(&len, schema_path);
if (NULL == buf) {
free(schema_path);
if (0 > neu_asprintf(&schema_path, "%s/custom/schema/%s.json",
g_plugin_dir, schema_name)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, error_code.error, result_error);
});
return;
}
buf = file_string_read(&len, schema_path);
}
if (NULL == buf) {
free(schema_path);
if (0 > neu_asprintf(&schema_path, "%s/system/schema/%s.json",
g_plugin_dir, schema_name)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, error_code.error, result_error);
});
return;
}
buf = file_string_read(&len, schema_path);
}
if (NULL == buf) {
nlog_info("open %s error: %d", schema_path, errno);
neu_http_not_found(aio, "{\"status\": \"error\"}");
free(schema_path);
return;
}
neu_http_ok(aio, buf);
free(buf);
free(schema_path);
}
void handle_status(nng_aio *aio)
plugins/restful/normal_handle.h
+3
View File
@@ -22,10 +22,13 @@
#include <nng/nng.h>
#include "msg.h"
void handle_ping(nng_aio *aio);
void handle_login(nng_aio *aio);
void handle_password(nng_aio *aio);
void handle_get_plugin_schema(nng_aio *aio);
void handle_get_plugin_schema_resp(nng_aio *aio, neu_resp_check_schema_t *resp);
void handle_status(nng_aio *aio);
#endif
plugins/restful/rest.c
+9
View File
@@ -31,6 +31,7 @@
#include "global_config_handle.h"
#include "group_config_handle.h"
#include "handle.h"
#include "normal_handle.h"
#include "otel/otel_manager.h"
#include "plugin_handle.h"
#include "rest.h"
@@ -211,6 +212,14 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
}
break;
}
case NEU_RESP_CHECK_SCHEMA: {
handle_get_plugin_schema_resp(header->ctx,
(neu_resp_check_schema_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
}
case NEU_RESP_GET_PLUGIN:
handle_get_plugin_resp(header->ctx, (neu_resp_get_plugin_t *) data);
if (neu_otel_control_is_started() && trace) {
src/adapter/adapter.c
+2
View File
@@ -509,6 +509,7 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
strcpy(pheader->receiver, cmd->node);
break;
}
case NEU_REQ_CHECK_SCHEMA:
case NEU_REQ_GET_DRIVER_GROUP:
case NEU_REQ_GET_SUB_DRIVER_TAGS: {
strcpy(pheader->receiver, "manager");
@@ -721,6 +722,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQRESP_NODES_STATE:
case NEU_REQ_PRGFILE_PROCESS:
case NEU_RESP_PRGFILE_PROCESS:
case NEU_RESP_CHECK_SCHEMA:
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
neu_msg_free(msg);
src/base/msg_internal.h
+6 -1
View File
@@ -99,7 +99,9 @@ extern "C" {
XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t) \
XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t) \
XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t) \
XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t)
XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t) \
XX(NEU_REQ_CHECK_SCHEMA, neu_req_check_schema_t) \
XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t)
static inline size_t neu_reqresp_size(neu_reqresp_type_e t)
{
@@ -140,6 +142,9 @@ static inline neu_msg_t *neu_msg_new(neu_reqresp_type_e t, void *ctx,
// NOTE: ensure enough space to reuse message
size_t body_size = 0;
switch (t) {
case NEU_REQ_CHECK_SCHEMA:
body_size = neu_reqresp_size(NEU_RESP_CHECK_SCHEMA);
break;
case NEU_REQ_GET_PLUGIN:
body_size = neu_reqresp_size(NEU_RESP_GET_PLUGIN);
break;
src/core/manager.c
+11
View File
@@ -709,6 +709,17 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &resp);
break;
}
case NEU_REQ_CHECK_SCHEMA: {
neu_req_check_schema_t *cmd = (neu_req_check_schema_t *) &header[1];
neu_resp_check_schema_t resp = { 0 };
strcpy(resp.schema, cmd->schema);
resp.exist = neu_plugin_manager_schema_exist(manager->plugin_manager,
cmd->schema);
header->type = NEU_RESP_CHECK_SCHEMA;
strcpy(header->receiver, header->sender);
reply(manager, header, &resp);
break;
}
case NEU_REQ_ADD_NODE: {
neu_req_add_node_t *cmd = (neu_req_add_node_t *) &header[1];
nlog_notice("add node name:%s plugin:%s", cmd->node, cmd->plugin);
src/core/plugin_manager.c
+17
View File
@@ -556,3 +556,20 @@ bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
return ret;
}
bool neu_plugin_manager_schema_exist(neu_plugin_manager_t *mgr,
const char * schema)
{
bool exist = false;
plugin_entity_t *el = NULL, *tmp = NULL;
HASH_ITER(hh, mgr->plugins, el, tmp)
{
if (strcmp(el->schema, schema) == 0) {
exist = true;
break;
}
}
return exist;
}
src/core/plugin_manager.h
+3
View File
@@ -78,4 +78,7 @@ bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
int neu_plugin_manager_update(neu_plugin_manager_t *mgr,
const char * plugin_lib_name);
bool neu_plugin_manager_schema_exist(neu_plugin_manager_t *mgr,
const char * schema);
#endif
tests/ft/http_api/test_http_api.py
-1
View File
@@ -594,7 +594,6 @@ class TestHttp:
response = api.get_plugin_schema(plugin='invalid')
assert 404 == response.status_code
assert response.json()['status'] == 'error'
@description(given="running neuron", when="get plugin", then="success")
def test_get_plugin(self):