path: root/app/src/dp_trace.c

#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>
#include <dp_trace.h>

extern __thread struct mr_thread_info thread_info;

int marsio_dp_trace_init(struct mr_instance * mr_instance)
{
    mr_instance->trace = dp_trace_process_create(DP_TRACE_PROCESS_APP);
    if (mr_instance->trace == NULL)
    {
        return RT_ERR;
    }

    return RT_SUCCESS;
}

int marsio_dp_trace_measurements_can_emit(__rte_unused struct mr_instance * instance, const marsio_buff_t * mbuf,
                                          uint8_t measurement_type)
{
    struct mrb_metadata * mrb_meta = rte_mbuf_to_priv((struct rte_mbuf *)mbuf);
    if (unlikely(mrb_meta->measurement_type == DP_TRACE_MEASUREMENT_TYPE_UNKNOWN))
    {
        dp_trace_filter_exec(instance->trace, (struct rte_mbuf *)mbuf, 0, marsio_thread_id_get());
    }
    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;
}

void marsio_dp_trace_record_write(struct mr_instance * mr_instance, marsio_buff_t * mbuf)
{
    return dp_trace_record_write(mr_instance->trace, (struct rte_mbuf *)mbuf, marsio_thread_id_get());
}

void marsio_rte_pktmbuf_free(struct rte_mbuf * m)
{
    if (unlikely(m == NULL))
        return;
    struct mr_instance * mr_instance = marsio_current();
    struct mrb_metadata * mrb_meta = rte_mbuf_to_priv(m);
    mrb_meta->egress_action = MR_EGRESS_ACTION_DROP;
    marsio_dp_trace_record_write(mr_instance, m);
    rte_pktmbuf_free(m);
}

thread_id_t marsio_thread_id_get()
{
    return thread_info.thread_id % RTE_MAX_LCORE;
}

cJSON * marsio_dp_trace_monit_loop(struct mr_instance * instance)
{
    cJSON * json_root = cJSON_CreateObject();
    struct dp_trace_process * trace = instance->trace;

    if (trace == NULL)
    {
        return json_root;
    }

    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];
        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;
        monit_statistics.record_buf_alloc_success += statistics_i->record_buf_alloc_success;
        monit_statistics.ring_enqueue_failed += statistics_i->ring_enqueue_failed;
        monit_statistics.ring_enqueue_success += statistics_i->ring_enqueue_success;
        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", 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);

    cJSON_AddNumberToObject(json_root, "record_buf_alloc_failed_no_mem",
                            monit_statistics.record_buf_alloc_failed_no_mem);
    cJSON_AddNumberToObject(json_root, "record_buf_alloc_success", monit_statistics.record_buf_alloc_success);

    cJSON_AddNumberToObject(json_root, "ring_enqueue_failed", monit_statistics.ring_enqueue_failed);
    cJSON_AddNumberToObject(json_root, "ring_enqueue_success", monit_statistics.ring_enqueue_success);

    cJSON_AddNumberToObject(json_root, "record_emit_failed_trace_oversize",
                            monit_statistics.record_emit_failed_trace_oversize);
    cJSON_AddNumberToObject(json_root, "record_emit_failed_no_space_in_buf",
                            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)
{
    memset(info, 0, sizeof(struct dp_trace_buffer_telemetry));
    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;

    info->traffic_link_id = dp_trace_buffer->traffic_link_id;
    snprintf(info->inner_src_addr_str, sizeof(info->inner_src_addr_str), "%s", dp_trace_buffer->inner_src_addr_str);
    snprintf(info->inner_dst_addr_str, sizeof(info->inner_dst_addr_str), "%s", dp_trace_buffer->inner_dst_addr_str);
    info->inner_src_port = dp_trace_buffer->inner_src_port;
    info->inner_dst_port = dp_trace_buffer->inner_dst_port;

    info->egress_action = mrb_meta->egress_action;

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

#if 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;

    if (pkt_parser_result->nr_layers == 0)
    {
        struct pkt_parser handler;
        pkt_parser_init(&handler, pkt_parser_result, LAYER_TYPE_ALL, MR_PKT_PARSER_LAYERS_MAX);
        pkt_parser_exec(&handler, mbuf);
    }

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

    return ret;
}
#endif