refactor data path trace for telemetryv4.8.4-20240427

1 files changed, 258 insertions, 6 deletions

diff --git a/app/src/dp_trace.c b/app/src/dp_trace.c
index 3c220e3..77d162b 100644
--- a/app/src/dp_trace.c
+++ b/app/src/dp_trace.c
@@ -1,6 +1,18 @@
 #include "mrapp.h"
 #include "mrdp_trace.h"
 
+#include <libgen.h>
+#include <mpack.h>
+#include <rte_ip.h>
+#include <rte_pcapng.h>
+#include <rte_tcp.h>
+#include <rte_udp.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/stat.h>
+#include <sys/utsname.h>
+#include <unistd.h>
+
 extern __thread struct mr_thread_info thread_info;
 
 int marsio_dp_trace_init(struct mr_instance * mr_instance)
@@ -13,13 +25,14 @@ int marsio_dp_trace_init(struct mr_instance * mr_instance)
 
 int marsio_dp_trace_record_can_emit(const marsio_buff_t * mbuf)
 {
-    return dp_trace_record_can_emit((struct rte_mbuf *)mbuf);
+    return dp_trace_record_can_emit((struct rte_mbuf *)mbuf, DP_TRACE_MEASUREMENT_TYPE_TRACE);
 }
 
 int marsio_dp_trace_record_emit_str(struct mr_instance * mr_instance, marsio_buff_t * mbuf, const char * module,
                                     const char * str)
 {
-    struct dp_trace_record_meta meta = {.module = module, .appsym = mr_instance->appsym};
+    struct dp_trace_record_meta meta = {
+        .measurement_type = DP_TRACE_MEASUREMENT_TYPE_TRACE, .module = module, .appsym = mr_instance->appsym};
     return dp_trace_record_emit_str(mr_instance->trace, (struct rte_mbuf *)mbuf, marsio_thread_id_get(), &meta, str);
 }
 
@@ -41,7 +54,47 @@ int marsio_dp_trace_record_emit_fmt(struct mr_instance * mr_instance, marsio_buf
         return -1;
     }
 
-    struct dp_trace_record_meta meta = {.module = module, .appsym = mr_instance->appsym};
+    struct dp_trace_record_meta meta = {
+        .measurement_type = DP_TRACE_MEASUREMENT_TYPE_TRACE, .module = module, .appsym = mr_instance->appsym};
+
+    ret = dp_trace_record_emit_str(trace, (struct rte_mbuf *)mbuf, marsio_thread_id_get(), &meta, buffer);
+    return ret;
+}
+
+int marsio_dp_trace_measurements_can_emit(__rte_unused struct mr_instance * instance, const marsio_buff_t * mbuf,
+                                          uint8_t measurement_type)
+{
+    return dp_trace_record_can_emit((struct rte_mbuf *)mbuf, measurement_type);
+}
+
+int marsio_dp_trace_measurement_emit_str(struct mr_instance * mr_instance, marsio_buff_t * mbuf,
+                                         uint8_t measurement_type, const char * module, const char * str)
+{
+    struct dp_trace_record_meta meta = {
+        .measurement_type = measurement_type, .module = module, .appsym = mr_instance->appsym};
+    return dp_trace_record_emit_str(mr_instance->trace, (struct rte_mbuf *)mbuf, marsio_thread_id_get(), &meta, str);
+}
+
+int marsio_dp_trace_measurement_emit_fmt(struct mr_instance * mr_instance, marsio_buff_t * mbuf,
+                                         uint8_t measurement_type, const char * module, const char * format, ...)
+{
+    struct dp_trace_process * trace = mr_instance->trace;
+
+    char buffer[512];
+    va_list args;
+    va_start(args, format);
+    int ret = vsnprintf(buffer, sizeof(buffer), format, args);
+    va_end(args);
+
+    if (unlikely(ret < 0 || ret >= sizeof(buffer)))
+    {
+        thread_id_t thread_id = marsio_thread_id_get();
+        trace->statistics[thread_id].record_emit_failed_trace_oversize++;
+        return -1;
+    }
+
+    struct dp_trace_record_meta meta = {
+        .measurement_type = measurement_type, .module = module, .appsym = mr_instance->appsym};
 
     ret = dp_trace_record_emit_str(trace, (struct rte_mbuf *)mbuf, marsio_thread_id_get(), &meta, buffer);
     return ret;
@@ -79,7 +132,7 @@ void marsio_rte_mempool_generic_put(struct rte_mempool * mp, void * const * obj_
 
 thread_id_t marsio_thread_id_get()
 {
-    return thread_info.thread_id;
+    return thread_info.thread_id % RTE_MAX_LCORE;
 }
 
 cJSON * marsio_dp_trace_monit_loop(struct mr_instance * instance)
@@ -88,11 +141,15 @@ cJSON * marsio_dp_trace_monit_loop(struct mr_instance * instance)
     struct dp_trace_process * trace = instance->trace;
 
     struct dp_trace_stat monit_statistics = {};
+    uint64_t filter_exec_hit = 0;
 
     for (unsigned int i = 0; i < RTE_MAX_LCORE; i++)
     {
         struct dp_trace_stat * statistics_i = &trace->statistics[i];
-        monit_statistics.filter_exec_hit += statistics_i->filter_exec_hit;
+        for (unsigned int j = 0; j < DP_TRACE_JOB_NUM_MAX; j++)
+        {
+            filter_exec_hit += statistics_i->filter_exec_hit[j];
+        }
         monit_statistics.filter_exec_miss += statistics_i->filter_exec_miss;
         monit_statistics.reach_pkt_cnt_limit += statistics_i->reach_pkt_cnt_limit;
         monit_statistics.record_buf_alloc_failed_no_mem += statistics_i->record_buf_alloc_failed_no_mem;
@@ -102,11 +159,12 @@ cJSON * marsio_dp_trace_monit_loop(struct mr_instance * instance)
         monit_statistics.record_emit_failed_trace_oversize += statistics_i->record_emit_failed_trace_oversize;
         monit_statistics.record_emit_failed_no_space_in_buf += statistics_i->record_emit_failed_no_space_in_buf;
         monit_statistics.record_emit_success += statistics_i->record_emit_success;
+        monit_statistics.record_buf_free += statistics_i->record_buf_free;
     }
 
     cJSON_AddBoolToObject(json_root, "trace_enable", trace->inst->enable);
 
-    cJSON_AddNumberToObject(json_root, "filter_exec_hit", monit_statistics.filter_exec_hit);
+    cJSON_AddNumberToObject(json_root, "filter_exec_hit", filter_exec_hit);
     cJSON_AddNumberToObject(json_root, "filter_exec_miss", monit_statistics.filter_exec_miss);
 
     cJSON_AddNumberToObject(json_root, "reach_pkt_cnt_limit", monit_statistics.reach_pkt_cnt_limit);
@@ -124,5 +182,199 @@ cJSON * marsio_dp_trace_monit_loop(struct mr_instance * instance)
                             monit_statistics.record_emit_failed_no_space_in_buf);
     cJSON_AddNumberToObject(json_root, "record_emit_success", monit_statistics.record_emit_success);
 
+    cJSON_AddNumberToObject(json_root, "record_buf_free", monit_statistics.record_buf_free);
+
     return json_root;
+}
+
+int marsio_dp_trace_job_id_uesd_get(__rte_unused struct mr_instance * instance, job_bitmap_t * jobs_id)
+{
+    // Initiate a request and add a job. After the addition is successful, the job id is returned.
+    struct rte_mp_msg req_msg = {};
+    int ret = -1;
+
+    snprintf(req_msg.name, sizeof(req_msg.name), "%s", DP_TRACE_MP_MSG_NAME);
+    struct dp_trace_req * dp_trace_req = (struct dp_trace_req *)req_msg.param;
+    req_msg.len_param = sizeof(struct dp_trace_req);
+
+    dp_trace_req->action = DP_TRACE_JOB_ID_USED_GET;
+
+    struct rte_mp_reply mp_reply;
+    const struct timespec wait_timespec = {
+        .tv_nsec = 0,
+        .tv_sec = 30,
+    };
+
+    int tmp = rte_mp_request_sync(&req_msg, &mp_reply, &wait_timespec);
+    if (tmp < 0)
+    {
+        MR_WARNING("Failed to execute rte_mp_request_sync in marsio_dp_trace_job_get:%s", rte_strerror(rte_errno));
+        goto end;
+    }
+
+    struct dp_trace_resp * dp_trace_resp = (struct dp_trace_resp *)mp_reply.msgs->param;
+
+    if (dp_trace_resp->errcode == DP_TRACE_SUCCESS)
+    {
+        *jobs_id = dp_trace_resp->jobs_id_used_get;
+        ret = 1;
+    }
+    else
+    {
+        MR_ERROR("%s", dp_trace_strerror(dp_trace_resp->errcode));
+    }
+
+end:
+    free(mp_reply.msgs);
+    return ret;
+}
+
+job_bitmap_t marsio_dp_trace_job_add(struct mr_instance * instance, const struct dp_trace_job_desc * desc)
+{
+    // Initiate a request and add a job. After the addition is successful, the job id is returned.
+    struct rte_mp_msg req_msg = {};
+    job_bitmap_t job_id = 0;
+
+    snprintf(req_msg.name, sizeof(req_msg.name), "%s", DP_TRACE_MP_MSG_NAME);
+    struct dp_trace_req * dp_trace_req = (struct dp_trace_req *)req_msg.param;
+    req_msg.len_param = sizeof(struct dp_trace_req);
+
+    snprintf(dp_trace_req->appsym, sizeof(dp_trace_req->appsym), "%s", instance->appsym);
+    dp_trace_req->action = DP_TRACE_JOB_ADD;
+    struct dp_trace_job_desc * desc_copy = ZMALLOC(sizeof(struct dp_trace_job_desc));
+    MR_VERIFY_MALLOC(desc_copy);
+    memcpy(desc_copy, desc, sizeof(struct dp_trace_job_desc));
+    dp_trace_req->desc = desc_copy;
+
+    struct rte_mp_reply mp_reply;
+    const struct timespec wait_timespec = {
+        .tv_nsec = 0,
+        .tv_sec = 30,
+    };
+
+    int ret = rte_mp_request_sync(&req_msg, &mp_reply, &wait_timespec);
+    if (ret < 0)
+    {
+        MR_WARNING("Failed to execute rte_mp_request_sync in marsio_dp_trace_job_add:%s", rte_strerror(rte_errno));
+        goto end;
+    }
+
+    struct dp_trace_resp * dp_trace_resp = (struct dp_trace_resp *)mp_reply.msgs->param;
+
+    if (dp_trace_resp->errcode == DP_TRACE_SUCCESS)
+    {
+        job_id = dp_trace_resp->job_id_add;
+    }
+    else
+    {
+        MR_ERROR("%s", dp_trace_strerror(dp_trace_resp->errcode));
+    }
+
+end:
+    FREE(desc_copy);
+    free(mp_reply.msgs);
+    return job_id;
+}
+
+int marsio_dp_trace_job_del(__rte_unused struct mr_instance * instance, job_bitmap_t jobs_id)
+{
+    int ret = 0;
+
+    struct rte_mp_msg req_msg = {};
+    snprintf(req_msg.name, sizeof(req_msg.name), "%s", DP_TRACE_MP_MSG_NAME);
+    struct dp_trace_req * dp_trace_req = (struct dp_trace_req *)req_msg.param;
+    req_msg.len_param = sizeof(struct dp_trace_req);
+
+    dp_trace_req->action = DP_TRACE_JOB_DESTROY;
+    dp_trace_req->jobs_id_destroy = jobs_id;
+
+    struct rte_mp_reply mp_reply;
+    const struct timespec wait_timespec = {
+        .tv_nsec = 0,
+        .tv_sec = 30,
+    };
+
+    if (rte_mp_request_sync(&req_msg, &mp_reply, &wait_timespec) < 0)
+    {
+        MR_WARNING("Failed to execute rte_mp_request_sync in marsio_dp_trace_job_del:%s", rte_strerror(rte_errno));
+        goto end;
+    }
+
+end:
+    free(mp_reply.msgs);
+    return ret;
+}
+
+int marsio_dp_trace_mbuf_recv_burst(struct mr_instance * instance, queue_id_t qid, marsio_buff_t * mbufs[],
+                                    int nr_mbufs)
+{
+    struct dp_trace_process * dp_trace_process = instance->trace;
+    assert(qid < dp_trace_process->nr_ring);
+
+    int n = rte_ring_dequeue_burst(dp_trace_process->ring[qid], mbufs, nr_mbufs, NULL);
+    return n;
+}
+
+void marsio_dp_trace_mbuf_free(struct mr_instance * instance, marsio_buff_t * mbufs[], int nr_mbufs)
+{
+    struct dp_trace_process * dp_trace_process = instance->trace;
+
+    for (unsigned int i = 0; i < nr_mbufs; i++)
+    {
+        struct rte_mbuf * mbuf = mbufs[i];
+        struct mrb_metadata * mrb_meta = rte_mbuf_to_priv(mbuf);
+        struct dp_trace_buffer * dp_trace_buffer = mrb_meta->dp_trace_buffer;
+        dp_trace_buffer->buffer_refcnt--;
+        if (dp_trace_buffer->buffer_refcnt == 0)
+        {
+            rte_mempool_put(dp_trace_process->inst->pool, (void *)dp_trace_buffer);
+            thread_id_t thread_id = marsio_thread_id_get();
+            dp_trace_process->statistics[thread_id].record_buf_free++;
+        }
+    }
+
+    rte_pktmbuf_free_bulk((struct rte_mbuf **)mbufs, nr_mbufs);
+}
+
+int marsio_dp_trace_buffer_info_get(const marsio_buff_t * mbuf, struct dp_trace_buffer_telemetry * info)
+{
+    struct mrb_metadata * mrb_meta = rte_mbuf_to_priv((struct rte_mbuf *)mbuf);
+    struct dp_trace_buffer * dp_trace_buffer = mrb_meta->dp_trace_buffer;
+    assert(dp_trace_buffer != NULL);
+
+    info->jobs_id = dp_trace_buffer->jobs;
+    info->snaplen = dp_trace_buffer->snaplen;
+    info->buffer = dp_trace_buffer->buffer;
+    info->buffer_len = dp_trace_buffer->buffer_len;
+    info->buffer_used = dp_trace_buffer->buffer_used;
+
+    return 0;
+}
+
+int marsio_dp_trace_mbuf_refcnt_update(const marsio_buff_t * mbuf, int16_t value)
+{
+    struct mrb_metadata * mrb_meta = rte_mbuf_to_priv((struct rte_mbuf *)mbuf);
+    struct dp_trace_buffer * dp_trace_buffer = mrb_meta->dp_trace_buffer;
+    assert(dp_trace_buffer != NULL);
+
+    dp_trace_buffer->buffer_refcnt += value;
+    rte_mbuf_refcnt_update((struct rte_mbuf *)mbuf, value);
+    return 0;
+}
+
+void * marsio_pkt_jump_to_innermost_layer(const marsio_buff_t * mbuf, enum complex_layer_type_id layer_id)
+{
+    struct mrb_metadata * mrb_meta = rte_mbuf_to_priv((struct rte_mbuf *)mbuf);
+    struct pkt_parser_result * pkt_parser_result = &mrb_meta->pkt_parser_result;
+
+    void * ret = NULL;
+    for (int i = pkt_parser_result->nr_layers - 1; i >= 0; i--)
+    {
+        if (pkt_parser_result->layers[i].type_id == layer_id)
+        {
+            ret = rte_pktmbuf_mtod((struct rte_mbuf *)mbuf, void *) + pkt_parser_result->layers[i].offset;
+        }
+    }
+
+    return ret;
 }
\ No newline at end of file