refactor data path trace for telemetryv4.8.4-20240427

4 files changed, 284 insertions, 19 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
diff --git a/app/src/mrb.c b/app/src/mrb.c
index 31c1102..f7eb885 100644
--- a/app/src/mrb.c
+++ b/app/src/mrb.c
@@ -29,12 +29,12 @@ void marsio_buff_ctrlzone_reset(marsio_buff_t * mr_buff)
 {
     struct mrb_metadata * mrb_meta = mrbuf_cz_data(mr_buff, MR_NODE_CTRLZONE_ID);
     struct dp_trace_buffer * dp_trace_buffer_ptr = mrb_meta->dp_trace_buffer;
-    int8_t dp_trace_can_emit = mrb_meta->dp_trace_can_emit;
+    int8_t measurement_type = mrb_meta->measurement_type;
 
     /* reset the ctrlzone but not dp_trace_buffer */
     memset(mrbuf_cz_data(mr_buff, MR_NODE_CTRLZONE_ID), 0, sizeof(struct mrb_metadata));
     mrb_meta->dp_trace_buffer = dp_trace_buffer_ptr;
-    mrb_meta->dp_trace_can_emit = dp_trace_can_emit;
+    mrb_meta->measurement_type = measurement_type;
 }
 
 void * mr_buffer_ctrlzone(struct rte_mbuf * mr_buff, uint8_t id)
@@ -120,11 +120,12 @@ void marsio_buff_free_v2(struct mr_instance * instance, marsio_buff_t * buff[],
         struct rte_mbuf * m = (struct rte_mbuf *)buff[i];
         struct rte_mbuf * m_next = NULL;
 
-        if (marsio_dp_trace_record_can_emit(m))
+        if (marsio_dp_trace_measurements_can_emit(instance, m, DP_TRACE_MEASUREMENT_TYPE_TRACE))
         {
-            marsio_dp_trace_record_emit_str(instance, m, NULL, "packet dropped by application");
-            dp_trace_record_write(instance->trace, m, marsio_thread_id_get());
+            marsio_dp_trace_measurement_emit_str(instance, m, DP_TRACE_MEASUREMENT_TYPE_TRACE, NULL,
+                                                 "packet dropped by application");
         }
+        dp_trace_record_write(instance->trace, m, marsio_thread_id_get());
 
         __rte_mbuf_sanity_check(m, 1);
         while (m != NULL)
diff --git a/app/src/rawio.c b/app/src/rawio.c
index d322b49..4a17e4f 100644
--- a/app/src/rawio.c
+++ b/app/src/rawio.c
@@ -23,8 +23,8 @@ static inline unsigned int packet_total_len(struct rte_mbuf * mbufs[], unsigned
     return total_len;
 }
 
-int mrapp_packet_fast_send_burst(struct mr_instance * instance,
-    struct vdev_instance * vdi, queue_id_t qid, struct rte_mbuf * mbufs[], int nr_mbufs)
+int mrapp_packet_fast_send_burst(struct mr_instance * instance, struct vdev_instance * vdi, queue_id_t qid,
+                                 struct rte_mbuf * mbufs[], int nr_mbufs)
 {
     hash_t mbufs_hash[MR_BURST_MAX];
     for (int i = 0; i < nr_mbufs; i++)
@@ -103,10 +103,10 @@ int marsio_recv_burst(struct mr_vdev * vdev, queue_id_t qid, marsio_buff_t * mbu
     {
         mbufs[i] = rx_buffer->mbufs[rx_buffer->curser + i];
 
-        if (unlikely(marsio_dp_trace_record_can_emit(mbufs[i])))
+        if (unlikely(marsio_dp_trace_measurements_can_emit(vdev->instance, mbufs[i], DP_TRACE_MEASUREMENT_TYPE_TRACE)))
         {
-            marsio_dp_trace_record_emit_fmt(vdev->instance, mbufs[i], "marsio_recv", "packet rx, dev=%s, qid=%u",
-                                            vdev->devsym, qid);
+            marsio_dp_trace_measurement_emit_fmt(vdev->instance, mbufs[i], DP_TRACE_MEASUREMENT_TYPE_TRACE,
+                                                 "marsio_recv", "packet rx, dev=%s, qid=%u", vdev->devsym, qid);
         }
     }
 
@@ -142,10 +142,11 @@ int marsio_send_buffer_flush(struct mr_vdev * vdev, queue_id_t sid)
     for (int i = 0; i < tx_buffer->length; i++)
     {
         hash[i] = tx_buffer->mbufs[i]->hash.usr;
-        if (marsio_dp_trace_record_can_emit(tx_buffer->mbufs[i]))
+        if (marsio_dp_trace_measurements_can_emit(instance, tx_buffer->mbufs[i], DP_TRACE_MEASUREMENT_TYPE_TRACE))
         {
-            marsio_dp_trace_record_emit_fmt(vdev->instance, tx_buffer->mbufs[i], "marsio_send",
-                                            "packet tx, dev=%s , qid=%u, hash=%u", vdev->devsym, sid, hash[i]);
+            marsio_dp_trace_measurement_emit_fmt(vdev->instance, tx_buffer->mbufs[i], DP_TRACE_MEASUREMENT_TYPE_TRACE,
+                                                 "marsio_send", "packet tx, dev=%s , qid=%u, hash=%u", vdev->devsym,
+                                                 sid, hash[i]);
         }
     }
 
diff --git a/app/src/version.map b/app/src/version.map
index 49adbfe..07bc6eb 100644
--- a/app/src/version.map
+++ b/app/src/version.map
@@ -88,6 +88,17 @@ global:
 	marsio_dp_trace_record_can_emit;
 	marsio_dp_trace_record_emit_str;
 	marsio_dp_trace_record_emit_fmt;
+	marsio_dp_trace_measurements_can_emit;
+	marsio_dp_trace_measurement_emit_str;
+	marsio_dp_trace_measurement_emit_fmt;
+	marsio_dp_trace_job_id_uesd_get;
+	marsio_dp_trace_job_add;
+	marsio_dp_trace_job_del;
+	marsio_dp_trace_mbuf_recv_burst;
+	marsio_dp_trace_mbuf_free;
+	marsio_dp_trace_buffer_info_get;
+	marsio_pkt_jump_to_innermost_layer;
+	marsio_dp_trace_mbuf_refcnt_update;
 
 local:	*;
 };