diff --git a/include/neuron/otel/otel_manager.h b/include/neuron/otel/otel_manager.h index 253fe374..e7170638 100644 --- a/include/neuron/otel/otel_manager.h +++ b/include/neuron/otel/otel_manager.h @@ -84,6 +84,7 @@ bool neu_otel_data_is_started(); double neu_otel_data_sample_rate(); const char *neu_otel_collector_url(); void neu_otel_set_config(void *config); +void * neu_otel_get_config(); const char *neu_otel_service_name(); #endif // NEU_OTEL_MANAGER_H \ No newline at end of file diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index 59add016..3672f570 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -423,6 +423,12 @@ static struct neu_http_handler rest_handlers[] = { .url = "/api/v2/otel", .value.handler = handle_otel, }, + { + .method = NEU_HTTP_METHOD_GET, + .type = NEU_HTTP_HANDLER_FUNCTION, + .url = "/api/v2/otel", + .value.handler = handle_otel_get, + }, }; void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size) diff --git a/plugins/restful/otel_handle.c b/plugins/restful/otel_handle.c index e2ef3f5c..ffaefaa1 100644 --- a/plugins/restful/otel_handle.c +++ b/plugins/restful/otel_handle.c @@ -63,4 +63,15 @@ void handle_otel(nng_aio *aio) }); } }) +} + +void handle_otel_get(nng_aio *aio) +{ + char *result = NULL; + NEU_VALIDATE_JWT(aio); + neu_json_otel_conf_req_t *req = neu_otel_get_config(); + neu_json_encode_by_fn(req, neu_json_encode_otel_conf_req, &result); + neu_http_ok(aio, result); + free(result); + neu_json_decode_otel_conf_req_free(req); } \ No newline at end of file diff --git a/plugins/restful/otel_handle.h b/plugins/restful/otel_handle.h index 324acc5a..098d73ab 100644 --- a/plugins/restful/otel_handle.h +++ b/plugins/restful/otel_handle.h @@ -24,4 +24,6 @@ void handle_otel(nng_aio *aio); +void handle_otel_get(nng_aio *aio); + #endif \ No newline at end of file diff --git a/src/otel/otel_manager.c b/src/otel/otel_manager.c index da8dc950..d9705df8 100644 --- a/src/otel/otel_manager.c +++ b/src/otel/otel_manager.c @@ -27,6 +27,7 @@ #include "parser/neu_json_otel.h" #include "plugin.h" #include "utils/http.h" +#include "utils/log.h" #include "utils/utarray.h" #include "utils/uthash.h" @@ -860,6 +861,8 @@ static int otel_timer_cb(void *data) neu_otel_trace_pack(el->ctx, data_buf); int status = neu_http_post_otel_trace(data_buf, data_size); free(data_buf); + nlog_debug("send trace:%s status:%d", (char *) el->ctx->trace_id, + status); if (status == 200 || status == 400) { HASH_DEL(traces_table, el); neu_otel_free_trace(el->ctx); @@ -891,6 +894,8 @@ void neu_otel_start() otel_timer = neu_event_add_timer(otel_event, param); } + + nlog_debug("otel_start"); } void neu_otel_stop() @@ -917,6 +922,8 @@ void neu_otel_stop() } pthread_mutex_unlock(&table_mutex); + + nlog_debug("otel_stop"); } bool neu_otel_control_is_started() @@ -940,6 +947,27 @@ void neu_otel_set_config(void *config) strcpy(otel_service_name, req->service_name); } otel_data_sample_rate = req->data_sample_rate; + + nlog_debug("otel config: %s %s %s %.2f %d %d", req->action, + req->collector_url, req->service_name, req->data_sample_rate, + req->data_flag, req->control_flag); +} + +void *neu_otel_get_config() +{ + neu_json_otel_conf_req_t *req = calloc(1, sizeof(neu_json_otel_conf_req_t)); + if (otel_flag) { + req->action = strdup("start"); + } else { + req->action = strdup("stop"); + } + + req->collector_url = strdup(otel_collector_url); + req->control_flag = otel_control_flag; + req->data_flag = otel_data_flag; + req->service_name = strdup(otel_service_name); + req->data_sample_rate = otel_data_sample_rate; + return req; } double neu_otel_data_sample_rate() diff --git a/src/parser/neu_json_otel.c b/src/parser/neu_json_otel.c index 869b3fff..ad6b5edc 100644 --- a/src/parser/neu_json_otel.c +++ b/src/parser/neu_json_otel.c @@ -110,4 +110,44 @@ int neu_json_decode_otel_conf_req(char *buf, neu_json_otel_conf_req_t **result) neu_json_decode_free(json_obj); } return ret; +} + +int neu_json_encode_otel_conf_req(void *json_object, void *param) +{ + int ret = 0; + neu_json_otel_conf_req_t *req = (neu_json_otel_conf_req_t *) param; + neu_json_elem_t resp_elems[] = { { + .name = "action", + .t = NEU_JSON_STR, + .v.val_str = req->action, + }, + { + .name = "collector_url", + .t = NEU_JSON_STR, + .v.val_str = req->collector_url, + }, + { + .name = "control", + .t = NEU_JSON_BOOL, + .v.val_bool = req->control_flag, + }, + { + .name = "data", + .t = NEU_JSON_BOOL, + .v.val_bool = req->data_flag, + }, + { + .name = "data_sample_rate", + .t = NEU_JSON_DOUBLE, + .v.val_double = req->data_sample_rate, + }, + { + .name = "service_name", + .t = NEU_JSON_STR, + .v.val_str = req->service_name, + } }; + ret = neu_json_encode_field(json_object, resp_elems, + NEU_JSON_ELEM_SIZE(resp_elems)); + + return ret; } \ No newline at end of file diff --git a/src/parser/neu_json_otel.h b/src/parser/neu_json_otel.h index eda2c43f..4ed28da4 100644 --- a/src/parser/neu_json_otel.h +++ b/src/parser/neu_json_otel.h @@ -41,7 +41,7 @@ typedef struct { void neu_json_decode_otel_conf_req_free(neu_json_otel_conf_req_t *req); int neu_json_decode_otel_conf_req(char *buf, neu_json_otel_conf_req_t **result); - +int neu_json_encode_otel_conf_req(void *json_object, void *param); #ifdef __cplusplus } #endif diff --git a/src/utils/http.c b/src/utils/http.c index b016f74d..279911b7 100644 --- a/src/utils/http.c +++ b/src/utils/http.c @@ -426,7 +426,11 @@ int neu_http_post_otel_trace(uint8_t *data, int len) nng_aio_alloc(&aio, NULL, NULL); nng_aio_set_timeout(aio, 1000); - if (nng_url_parse(&url, neu_otel_collector_url()) != 0) { + char url_buf[128] = { 0 }; + + sprintf(url_buf, "http://%s/v1/traces", neu_otel_collector_url()); + + if (nng_url_parse(&url, url_buf) != 0) { nng_aio_free(aio); return -1; } @@ -438,8 +442,7 @@ int neu_http_post_otel_trace(uint8_t *data, int len) int rv = -1; if ((rv = nng_aio_result(aio)) != 0) { - nlog_error("(%s)nng error: %s", neu_otel_collector_url(), - nng_strerror(rv)); + nlog_error("(%s)nng error: %s", url_buf, nng_strerror(rv)); nng_url_free(url); nng_http_client_free(client); nng_aio_free(aio); diff --git a/tests/ft/app/test_ekuiper.py b/tests/ft/app/test_ekuiper.py index 7b6fadcd..98679b9b 100644 --- a/tests/ft/app/test_ekuiper.py +++ b/tests/ft/app/test_ekuiper.py @@ -65,7 +65,7 @@ def ekuiper_node(): otel_p = subprocess.Popen( ["python3", "otel_server.py", f'{otel_port}'], stderr=subprocess.PIPE, cwd="simulator") api.add_node_check(NODE, PLUGIN) - api.otel_start("127.0.0.1", otel_port, "/v1/traces") + api.otel_start("127.0.0.1", otel_port) yield NODE api.del_node(NODE) api.otel_stop() diff --git a/tests/ft/app/test_mqtt.py b/tests/ft/app/test_mqtt.py index 6f8ebbcd..90863828 100644 --- a/tests/ft/app/test_mqtt.py +++ b/tests/ft/app/test_mqtt.py @@ -68,7 +68,7 @@ def mqtt_node(): otel_p = subprocess.Popen( ["python3", "otel_server.py", f'{otel_port}'], stderr=subprocess.PIPE, cwd="simulator") api.add_node_check(NODE, "MQTT") - api.otel_start("127.0.0.1", otel_port, "/v1/traces") + api.otel_start("127.0.0.1", otel_port) yield api.del_node(NODE) api.otel_stop() diff --git a/tests/ft/driver/test_modbus.py b/tests/ft/driver/test_modbus.py index 2f143e5a..baeb1210 100644 --- a/tests/ft/driver/test_modbus.py +++ b/tests/ft/driver/test_modbus.py @@ -82,8 +82,14 @@ def simulator_setup_teardown(param): response = api.modbus_rtu_node_setting( node=param[0], interval=1, device=modbus_neuron_dev, link=0) - api.otel_start("127.0.0.1", otel_port, "/v1/traces") - + api.otel_start("127.0.0.1", otel_port) + resp = api.otel_get() + assert 200 == resp.status_code + assert resp.json()['action'] == 'start' + assert resp.json()['collector_url'] == f"127.0.0.1:{otel_port}" + assert resp.json()['control'] == True + assert resp.json()['data'] == True + compare_float(0.02, resp.json()['data_sample_rate']) yield api.otel_stop() process.stop_simulator(p) diff --git a/tests/ft/neuron/api.py b/tests/ft/neuron/api.py index 42c9332e..3e1cce61 100644 --- a/tests/ft/neuron/api.py +++ b/tests/ft/neuron/api.py @@ -390,9 +390,13 @@ def scan_tags(node, id, ctx=""): return requests.post(url=config.BASE_URL + "/api/v2/scan/tags", headers={"Authorization": config.default_jwt}, json={"node": node, "id": id, "ctx": ctx}) -def otel_start(host, port, traces_url): - return requests.post(url=config.BASE_URL + "/api/v2/otel", headers={"Authorization": config.default_jwt}, json={"action": "start", "collector_url": f"http://{host}:{port}{traces_url}", "control": True, "data": True, "data_sample_rate": 0.02}) +def otel_start(host, port): + return requests.post(url=config.BASE_URL + "/api/v2/otel", headers={"Authorization": config.default_jwt}, json={"action": "start", "collector_url": f"{host}:{port}", "control": True, "data": True, "data_sample_rate": 0.02}) def otel_stop(): return requests.post(url=config.BASE_URL + "/api/v2/otel", headers={"Authorization": config.default_jwt}, json={"action": "stop"}) + + +def otel_get(): + return requests.get(url=config.BASE_URL + "/api/v2/otel", headers={"Authorization": config.default_jwt})