otel:add new api and modify old api

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

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)

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);
 }

plugins/restful/otel_handle.h

@@ -24,4 +24,6 @@
 
 void handle_otel(nng_aio *aio);
 
+void handle_otel_get(nng_aio *aio);
+
 #endif

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()

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;
 }

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

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);

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()

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()

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)

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})