Merge branch 'cmocka_add' into 'master'

Cmocka add

See merge request tsg/dp_telemetry_app!5

15 files changed, 320 insertions, 153 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index baa544c..1176ef1 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -15,5 +15,8 @@ endif()
 
 add_definitions(-D_GNU_SOURCE -D__STDC_LIMIT_MACROS)
 
+option(ENABLE_DEVELOP_MOCKING "enable develop mocking" FALSE)
+
 add_subdirectory(${CMAKE_SOURCE_DIR}/support)
-add_subdirectory(${CMAKE_SOURCE_DIR}/src)
\ No newline at end of file
+add_subdirectory(${CMAKE_SOURCE_DIR}/src)
+add_subdirectory(${CMAKE_SOURCE_DIR}/test)
\ No newline at end of file
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 829cbb6..3d9f6bb 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -7,7 +7,12 @@ set(DP_TELEMETRY_SRC
     ${CMAKE_CURRENT_SOURCE_DIR}/config.c 
     ${CMAKE_CURRENT_SOURCE_DIR}/job_ctx.c 
     ${CMAKE_CURRENT_SOURCE_DIR}/trace_output.c
+    ${CMAKE_CURRENT_SOURCE_DIR}/kafka.c
+    ${CMAKE_CURRENT_SOURCE_DIR}/mocking.c
     ${CMAKE_SOURCE_DIR}/support/mpack/mpack.c)
 
 add_executable(${PROJECT_NAME} ${DP_TELEMETRY_SRC})
-target_link_libraries(${PROJECT_NAME} libmarsio maatframe MESA_prof_load libevent-static rdkafka uuid pthread)
\ No newline at end of file
+target_link_libraries(${PROJECT_NAME} libmarsio maatframe MESA_prof_load libevent-static rdkafka libcmocka uuid pthread)
+if(ENABLE_DEVELOP_MOCKING)
+    target_link_options(${PROJECT_NAME} PRIVATE -Wl,--wrap=kafka_handle_create,--wrap=kafka_topic_new,--wrap=kafka_produce)
+endif()
\ No newline at end of file
diff --git a/src/config.c b/src/config.c
index 2de6c45..0e458e2 100644
--- a/src/config.c
+++ b/src/config.c
@@ -52,6 +52,15 @@ void config_load()
     }
     g_conf->log_level = loglevel;
 
+    MESA_load_profile_string_def(g_conf->device_group, "global", "device_group", g_conf->device_group,
+                                 sizeof(g_conf->device_group), "unknow");
+
+    g_conf->sled_ip = getenv("SLED_IP");
+    if (g_conf->sled_ip == NULL)
+    {
+        syslog(LOG_WARNING, "SLED_IP environment variable does not exist.");
+    }
+
     MESA_load_profile_string_def(g_conf->config_path, "kafka", "borker_list", g_conf->broker_list,
                                  sizeof(g_conf->broker_list), "");
     MESA_load_profile_string_def(g_conf->config_path, "kafka", "topic_name", g_conf->topic_name,
diff --git a/src/config.h b/src/config.h
index 22707d8..9757c42 100644
--- a/src/config.h
+++ b/src/config.h
@@ -20,6 +20,10 @@ struct config
     struct event_base * evbase;
     struct event * evsignal[8];
 
+    // device Information
+    char * sled_ip;
+    char device_group[MR_SYMBOL_MAX];
+
     // kafka
     char topic_name[MR_SYMBOL_MAX];
     char broker_list[1024];
diff --git a/src/job_ctx.c b/src/job_ctx.c
index 2c8022e..10e5a30 100644
--- a/src/job_ctx.c
+++ b/src/job_ctx.c
@@ -56,14 +56,6 @@ void job_rule_apply(struct dp_trace_job_desc desc[], unsigned int nr_desc)
         job_bitmap_t cur_job_id = 1 << i;
         unsigned int cur_job_uesd = cur_job_id & jobs_id_used;
 
-#if 0
-        // The job that was not in the system originally is not in the configuration now.
-        if (cur_job_uesd == 0 && (desc[i].enable == false || strlen(desc[i].bpf_expr) == 0))
-        {
-            continue;
-        }
-#endif
-
         // The job that was not originally in the system is now included in the configuration.
         if (cur_job_uesd == 0 && desc[i].enable && strlen(desc[i].bpf_expr) != 0)
         {
@@ -72,38 +64,6 @@ void job_rule_apply(struct dp_trace_job_desc desc[], unsigned int nr_desc)
             continue;
         }
 
-#if 0
-        // Modification functionality is no longer available
-        if (cur_job_uesd == 1 && desc[i].enable && strlen(desc[i].bpf_expr) != 0)
-        {
-            struct dp_trace_job_desc * job_ctx_i_desc = &desc[i];
-            if (strcmp(job_ctx_i_desc->bpf_expr, desc[i].bpf_expr) != 0)
-            {
-                syslog(LOG_INFO, " [mod rule %u] bpf_expr: %s;", i, desc[i].bpf_expr);
-                marsio_dp_trace_job_del(mr_instance, cur_job_id);
-                marsio_dp_trace_job_add(mr_instance, job_ctx_i_desc);
-            }
-            else
-            {
-                if (job_ctx_i_desc->pkt_cnt_max != desc[i].pkt_cnt_max)
-                {
-                    job_ctx_i_desc->pkt_cnt_max = desc[i].pkt_cnt_max;
-                }
-
-                if (job_ctx_i_desc->sampling != desc[i].sampling)
-                {
-                    job_ctx_i_desc->sampling = desc[i].sampling;
-                }
-
-                if (job_ctx_i_desc->snaplen != desc[i].snaplen)
-                {
-                    job_ctx_i_desc->snaplen = desc[i].snaplen;
-                }
-            }
-            continue;
-        }
-#endif
-
         // The job that originally existed in the system is not in the configuration now.
         if (cur_job_uesd == 1 && (desc[i].enable == false || strlen(desc[i].bpf_expr) == 0))
         {
diff --git a/src/kafka.c b/src/kafka.c
new file mode 100644
index 0000000..cf3a4d2
--- /dev/null
+++ b/src/kafka.c
@@ -0,0 +1,81 @@
+#include "kafka.h"
+#include "common.h"
+
+rd_kafka_t * kafka_handle_create(const char * brokerlist, const char * sasl_username, const char * sasl_passwd)
+{
+    int ret;
+    char kafka_errstr[1024] = {0};
+    rd_kafka_t * handle = NULL;
+    rd_kafka_conf_t * rconf = NULL;
+
+    rconf = rd_kafka_conf_new();
+
+    ret = rd_kafka_conf_set(rconf, "queue.buffering.max.messages", "1000000", kafka_errstr, sizeof(kafka_errstr));
+    if (ret != RD_KAFKA_CONF_OK)
+    {
+        syslog(LOG_ERR, "Error to set kafka \"queue.buffering.max.messages\", %s.", kafka_errstr);
+        goto error;
+    }
+    ret = rd_kafka_conf_set(rconf, "topic.metadata.refresh.interval.ms", "600000", kafka_errstr, sizeof(kafka_errstr));
+    if (ret != RD_KAFKA_CONF_OK)
+    {
+        syslog(LOG_ERR, "Error to set kafka \"topic.metadata.refresh.interval.ms\", %s.", kafka_errstr);
+        goto error;
+    }
+
+    assert(strlen(sasl_username) > 0);
+    assert(strlen(sasl_passwd) > 0);
+
+    rd_kafka_conf_set(rconf, "security.protocol", "sasl_plaintext", kafka_errstr, sizeof(kafka_errstr));
+    rd_kafka_conf_set(rconf, "sasl.mechanisms", "PLAIN", kafka_errstr, sizeof(kafka_errstr));
+    ret = rd_kafka_conf_set(rconf, "sasl.username", sasl_username, kafka_errstr, sizeof(kafka_errstr));
+    if (ret != RD_KAFKA_CONF_OK)
+    {
+        syslog(LOG_ERR, "Error to set kafka \"sasl.username\", %s.", kafka_errstr);
+        goto error;
+    }
+    ret = rd_kafka_conf_set(rconf, "sasl.password", sasl_passwd, kafka_errstr, sizeof(kafka_errstr));
+    if (ret != RD_KAFKA_CONF_OK)
+    {
+        syslog(LOG_ERR, "Error to set kafka \"sasl.password\", %s.", kafka_errstr);
+        goto error;
+    }
+
+    // The conf object is freed by this function and must not be used or destroyed by the application sub-sequently.
+    handle = rd_kafka_new(RD_KAFKA_PRODUCER, rconf, kafka_errstr, sizeof(kafka_errstr));
+    rconf = NULL;
+    if (handle == NULL)
+    {
+        syslog(LOG_ERR, "Error to new kafka, %s.", kafka_errstr);
+        goto error;
+    }
+
+    if (rd_kafka_brokers_add(handle, brokerlist) == 0)
+    {
+        syslog(LOG_ERR, "Error to add kakfa bokers.");
+        goto error;
+    }
+
+    return handle;
+
+error:
+    if (rconf != NULL)
+    {
+        rd_kafka_conf_destroy(rconf);
+    }
+    if (handle != NULL)
+    {
+        rd_kafka_destroy(handle);
+    }
+    return NULL;
+}
+
+rd_kafka_topic_t * kafka_topic_new(rd_kafka_t * rk, const char * topic, rd_kafka_topic_conf_t * conf)
+{
+    return rd_kafka_topic_new(rk, topic, conf);
+}
+
+int kafka_produce(rd_kafka_topic_t * rkt, void * payload, size_t len)
+{
+    return rd_kafka_produce(rkt, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_FREE, (void *)payload, len, NULL, 0, NULL);
+}
\ No newline at end of file
diff --git a/src/kafka.h b/src/kafka.h
new file mode 100644
index 0000000..197c4f8
--- /dev/null
+++ b/src/kafka.h
@@ -0,0 +1,6 @@
+#pragma once
+#include <librdkafka/rdkafka.h>
+
+rd_kafka_t * kafka_handle_create(const char * brokerlist, const char * sasl_username, const char * sasl_passwd);
+rd_kafka_topic_t * kafka_topic_new(rd_kafka_t * rk, const char * topic, rd_kafka_topic_conf_t * conf);
+int kafka_produce(rd_kafka_topic_t * rkt, void * payload, size_t len);
\ No newline at end of file
diff --git a/src/mocking.c b/src/mocking.c
new file mode 100644
index 0000000..63788a4
--- /dev/null
+++ b/src/mocking.c
@@ -0,0 +1,109 @@
+#include "mocking.h"
+#include "common.h"
+#include <mpack.h>
+
+rd_kafka_t * __wrap_kafka_handle_create(const char * brokerlist, const char * sasl_username, const char * sasl_passwd)
+{
+    return NULL;
+}
+
+rd_kafka_topic_t * __wrap_kafka_topic_new(rd_kafka_t * rk, const char * topic, rd_kafka_topic_conf_t * conf)
+{
+    return NULL;
+}
+
+int __wrap_kafka_produce(rd_kafka_topic_t * rkt, void * payload, size_t len)
+{
+    struct measurements
+    {
+        int32_t tv_sec;
+        int32_t tv_nsec;
+        const char * app;
+        uint32_t app_len;
+        const char * comments;
+        uint32_t comments_len;
+    };
+
+    struct dp_trace_message_pack
+    {
+        int64_t microseconds;
+        char job_id_str[MR_SYMBOL_MAX];
+        char sled_ip[INET6_ADDRSTRLEN];
+        char device_group[MR_SYMBOL_MAX];
+        char source_ip[INET6_ADDRSTRLEN];
+        int32_t source_port;
+        char server_ip[INET6_ADDRSTRLEN];
+        int32_t server_port;
+        const char * packet;
+        int32_t packet_length;
+        int32_t measurements_num;
+        struct measurements record[128];
+    };
+
+    struct dp_trace_message_pack packet = {};
+
+    mpack_tree_t tree;
+    mpack_tree_init_data(&tree, payload, len);
+    mpack_tree_parse(&tree);
+
+    mpack_node_t root = mpack_tree_root(&tree);
+    packet.microseconds = mpack_node_i64(mpack_node_map_cstr(root, "timestamp"));
+
+    mpack_node_copy_cstr(mpack_node_map_cstr(root, "job_id"), packet.job_id_str, sizeof(packet.job_id_str));
+
+    mpack_node_copy_cstr(mpack_node_map_cstr(root, "sled_ip"), packet.sled_ip, sizeof(packet.sled_ip));
+
+    mpack_node_copy_cstr(mpack_node_map_cstr(root, "device_group"), packet.device_group, sizeof(packet.device_group));
+
+    mpack_node_copy_cstr(mpack_node_map_cstr(root, "source_ip"), packet.source_ip, sizeof(packet.source_ip));
+
+    packet.source_port = mpack_node_i32(mpack_node_map_cstr(root, "source_port"));
+
+    mpack_node_copy_cstr(mpack_node_map_cstr(root, "server_ip"), packet.server_ip, sizeof(packet.server_ip));
+
+    packet.server_port = mpack_node_i32(mpack_node_map_cstr(root, "server_port"));
+
+    packet.packet = mpack_node_bin_data(mpack_node_map_cstr(root, "packet"));
+
+    packet.packet_length = mpack_node_i32(mpack_node_map_cstr(root, "packet_length"));
+
+    mpack_node_t measurements_val = mpack_node_map_cstr(root, "measurements");
+    packet.measurements_num = mpack_node_array_length(mpack_node_map_cstr(root, "measurements"));
+    for (unsigned int i = 0; i < packet.measurements_num; i++)
+    {
+        if (i >= 128)
+        {
+            printf("too many measurements...");
+            continue;
+        }
+        mpack_node_t measurement = mpack_node_array_at(measurements_val, i);
+        packet.record[i].tv_sec = mpack_node_i32(mpack_node_map_cstr(measurement, "tv_sec"));
+        packet.record[i].tv_nsec = mpack_node_i32(mpack_node_map_cstr(measurement, "tv_nsec"));
+        packet.record[i].app = mpack_node_str(mpack_node_map_cstr(measurement, "app"));
+        packet.record[i].app_len = mpack_node_strlen(mpack_node_map_cstr(measurement, "app"));
+        packet.record[i].comments = mpack_node_str(mpack_node_map_cstr(measurement, "comments"));
+        packet.record[i].comments_len = mpack_node_strlen(mpack_node_map_cstr(measurement, "comments"));
+    }
+
+    // print
+    printf("microseconds %ld\n", packet.microseconds);
+    printf("job_id %s\n", packet.job_id_str);
+    printf("sled_ip %s\n", packet.sled_ip);
+    printf("device_group %s\n", packet.device_group);
+    printf("source_ip %s\n", packet.source_ip);
+    printf("source_port %d\n", packet.source_port);
+    printf("server_ip %s\n", packet.server_ip);
+    printf("server_port %d\n", packet.server_port);
+    printf("packet_length %d\n", packet.packet_length);
+    for (unsigned int i = 0; i < packet.measurements_num; i++)
+    {
+        printf("record %u:\n", i);
+        printf("tv_sec %d\n", packet.record[i].tv_sec);
+        printf("tv_nsec %d\n", packet.record[i].tv_nsec);
+        printf("app %.*s\n", packet.record[i].app_len, packet.record[i].app);
+        printf("comments %.*s\n", packet.record[i].comments_len, packet.record[i].comments);
+    }
+
+    free(payload);
+    return 0;
+}
\ No newline at end of file
diff --git a/src/mocking.h b/src/mocking.h
new file mode 100644
index 0000000..d509081
--- /dev/null
+++ b/src/mocking.h
@@ -0,0 +1,6 @@
+#pragma once
+#include <librdkafka/rdkafka.h>
+
+rd_kafka_t * __wrap_kafka_handle_create(const char * brokerlist, const char * sasl_username, const char * sasl_passwd);
+rd_kafka_topic_t * __wrap_kafka_topic_new(rd_kafka_t * rk, const char * topic, rd_kafka_topic_conf_t * conf);
+int __wrap_kafka_produce(rd_kafka_topic_t * rkt, void * payload, size_t len);
\ No newline at end of file
diff --git a/src/trace_output.c b/src/trace_output.c
index 4f44070..f72d038 100644
--- a/src/trace_output.c
+++ b/src/trace_output.c
@@ -2,6 +2,8 @@
 #include "common.h"
 #include "config.h"
 #include "job_ctx.h"
+#include "kafka.h"
+#include "mocking.h"
 
 #include <mpack.h>
 
@@ -12,6 +14,8 @@
 
 #define BURST_MAX 64
 
+extern struct mr_instance * mr_instance;
+
 int dp_trace_file_mutex_lock(job_bitmap_t job_id);
 int dp_trace_file_mutex_unlock(job_bitmap_t job_id);
 bool dp_trace_file_reach_max_size(job_bitmap_t job_id);
@@ -20,7 +24,6 @@ void dp_trace_file_rollbak(job_bitmap_t job_id);
 int dp_trace_classification(struct mr_instance * instance, marsio_buff_t * mbufs[], int nr_mbufs,
                             marsio_buff_t * class_mbufs[], int nr_class_mbufs, int nr_jobs_mbufs[DP_TRACE_JOB_NUM_MAX]);
 static void dp_trace_decode_to_message_pack(marsio_buff_t * mr_mbuf, char ** data, size_t * size);
-static rd_kafka_t * kafka_handle_create(const char * brokerlist, const char * sasl_username, const char * sasl_passwd);
 
 struct dp_trace_output
 {
@@ -65,11 +68,8 @@ void dp_trace_output_init()
         }
     }
 
-    // kafka init
-#if 0
     g_conf->kafka_handle = kafka_handle_create(g_conf->broker_list, g_conf->sasl_password, g_conf->sasl_username);
-    g_conf->kafka_topic = rd_kafka_topic_new(g_conf->kafka_handle, g_conf->topic_name, NULL);
-#endif
+    g_conf->kafka_topic = kafka_topic_new(g_conf->kafka_handle, g_conf->topic_name, NULL);
 }
 
 void * dp_trace_process_thread(void * arg)
@@ -129,11 +129,7 @@ void * dp_trace_process_thread(void * arg)
                 char * data;
                 size_t size;
                 dp_trace_decode_to_message_pack(rx_buff[j], &data, &size);
-                printf("%.*s\n", data, size);
-#if 0
-                rd_kafka_produce(g_conf->kafka_topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_FREE, (void *)data, size,
-                                 NULL, 0, NULL);
-#endif
+                kafka_produce(g_conf->kafka_topic, (void *)data, size);
             }
         }
     }
@@ -403,17 +399,30 @@ static void dp_trace_decode_to_message_pack(marsio_buff_t * mr_mbuf, char ** dat
     mpack_build_map(&writer);
 
     mpack_write_cstr(&writer, "timestamp");
-    mpack_write_u64(&writer, microseconds);
+    mpack_write_i64(&writer, microseconds);
 
     char job_id_str[10];
     snprintf(job_id_str, sizeof(job_id_str), "%d", job_id);
     mpack_write_cstr(&writer, "job_id");
     mpack_write_cstr(&writer, job_id_str);
 
-    mpack_write_cstr(&writer, "src_ip");
+    mpack_write_cstr(&writer, "sled_ip");
+    if (g_conf->sled_ip != NULL)
+    {
+        mpack_write_cstr(&writer, g_conf->sled_ip);
+    }
+    else
+    {
+        mpack_write_cstr(&writer, "0.0.0.0");
+    }
+
+    mpack_write_cstr(&writer, "device_group");
+    mpack_write_cstr(&writer, g_conf->device_group);
+
+    mpack_write_cstr(&writer, "source_ip");
     mpack_write_cstr(&writer, pkt_info.src_addr_str);
 
-    mpack_write_cstr(&writer, "src_port");
+    mpack_write_cstr(&writer, "source_port");
     mpack_write_i32(&writer, pkt_info.src_port);
 
     mpack_write_cstr(&writer, "server_ip");
@@ -425,6 +434,9 @@ static void dp_trace_decode_to_message_pack(marsio_buff_t * mr_mbuf, char ** dat
     mpack_write_cstr(&writer, "packet");
     mpack_write_bin(&writer, marsio_buff_mtod(mr_mbuf), marsio_buff_datalen(mr_mbuf));
 
+    mpack_write_cstr(&writer, "packet_length");
+    mpack_write_i32(&writer, marsio_buff_datalen(mr_mbuf));
+
     mpack_write_cstr(&writer, "measurements");
     mpack_build_array(&writer);
     while (position < trace_buff_info.buffer_used)
@@ -465,75 +477,4 @@ static void dp_trace_decode_to_message_pack(marsio_buff_t * mr_mbuf, char ** dat
     {
         syslog(LOG_ERR, "An error occurred during the data path decode to message pack!\n");
     }
-}
-
-static rd_kafka_t * kafka_handle_create(const char * brokerlist, const char * sasl_username, const char * sasl_passwd)
-{
-    int ret;
-    char kafka_errstr[1024] = {0};
-    rd_kafka_t * handle = NULL;
-    rd_kafka_conf_t * rconf = NULL;
-
-    rconf = rd_kafka_conf_new();
-
-    ret = rd_kafka_conf_set(rconf, "queue.buffering.max.messages", "1000000", kafka_errstr, sizeof(kafka_errstr));
-    if (ret != RD_KAFKA_CONF_OK)
-    {
-        syslog(LOG_ERR, "Error to set kafka \"queue.buffering.max.messages\", %s.", kafka_errstr);
-        goto error;
-    }
-    ret = rd_kafka_conf_set(rconf, "topic.metadata.refresh.interval.ms", "600000", kafka_errstr, sizeof(kafka_errstr));
-    if (ret != RD_KAFKA_CONF_OK)
-    {
-        syslog(LOG_ERR, "Error to set kafka \"topic.metadata.refresh.interval.ms\", %s.", kafka_errstr);
-        goto error;
-    }
-
-#if 0
-    assert(strlen(sasl_username) > 0);
-    assert(strlen(sasl_passwd) > 0);
-#endif
-
-    rd_kafka_conf_set(rconf, "security.protocol", "sasl_plaintext", kafka_errstr, sizeof(kafka_errstr));
-    rd_kafka_conf_set(rconf, "sasl.mechanisms", "PLAIN", kafka_errstr, sizeof(kafka_errstr));
-    ret = rd_kafka_conf_set(rconf, "sasl.username", sasl_username, kafka_errstr, sizeof(kafka_errstr));
-    if (ret != RD_KAFKA_CONF_OK)
-    {
-        syslog(LOG_ERR, "Error to set kafka \"sasl.username\", %s.", kafka_errstr);
-        goto error;
-    }
-    ret = rd_kafka_conf_set(rconf, "sasl.password", sasl_passwd, kafka_errstr, sizeof(kafka_errstr));
-    if (ret != RD_KAFKA_CONF_OK)
-    {
-        syslog(LOG_ERR, "Error to set kafka \"sasl.password\", %s.", kafka_errstr);
-        goto error;
-    }
-
-    // The conf object is freed by this function and must not be used or destroyed by the application sub-sequently.
-    handle = rd_kafka_new(RD_KAFKA_PRODUCER, rconf, kafka_errstr, sizeof(kafka_errstr));
-    rconf = NULL;
-    if (handle == NULL)
-    {
-        syslog(LOG_ERR, "Error to new kafka, %s.", kafka_errstr);
-        goto error;
-    }
-
-    if (rd_kafka_brokers_add(handle, brokerlist) == 0)
-    {
-        syslog(LOG_ERR, "Error to add kakfa bokers.");
-        goto error;
-    }
-
-    return handle;
-
-error:
-    if (rconf != NULL)
-    {
-        rd_kafka_conf_destroy(rconf);
-    }
-    if (handle != NULL)
-    {
-        rd_kafka_destroy(handle);
-    }
-    return NULL;
 }
\ No newline at end of file
diff --git a/support/CMakeLists.txt b/support/CMakeLists.txt
index 7f45057..f42c82a 100644
--- a/support/CMakeLists.txt
+++ b/support/CMakeLists.txt
@@ -2,36 +2,22 @@
 
 include(ExternalProject)
 
-#### GoogleTest
-ExternalProject_Add(googletest PREFIX googletest
-    URL ${CMAKE_CURRENT_SOURCE_DIR}/googletest-release-1.8.0.tar.gz
-    URL_MD5 16877098823401d1bf2ed7891d7dce36
-    CMAKE_ARGS -DCMAKE_INSTALL_PREFIX=<INSTALL_DIR> -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE})
+#### cmocka
+ExternalProject_Add(cmocka PREFIX cmocka
+    URL ${CMAKE_CURRENT_SOURCE_DIR}/cmocka-1.1.7.tar.xz
+    URL_MD5 dc23f4870eb7dfa6da065f3056e815b0
+    CMAKE_ARGS -DCMAKE_INSTALL_PREFIX=<INSTALL_DIR> -DBUILD_SHARED_LIBS=OFF -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE})
 
-ExternalProject_Get_Property(googletest INSTALL_DIR)
+ExternalProject_Get_Property(cmocka INSTALL_DIR)
 file(MAKE_DIRECTORY ${INSTALL_DIR}/include)
 
-add_library(gtest STATIC IMPORTED GLOBAL)
-add_dependencies(gtest googletest)
-set_property(TARGET gtest PROPERTY IMPORTED_LOCATION ${INSTALL_DIR}/lib/libgtest.a)
-set_property(TARGET gtest PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${INSTALL_DIR}/include)
-set_property(TARGET gtest PROPERTY INTERFACE_LINK_LIBRARIES pthread)
-
-add_library(gtest_main STATIC IMPORTED GLOBAL)
-add_dependencies(gtest_main googletest)
-set_property(TARGET gtest_main PROPERTY IMPORTED_LOCATION ${INSTALL_DIR}/lib/libgtest_main.a)
-set_property(TARGET gtest_main PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${INSTALL_DIR}/include)
-set_property(TARGET gtest_main PROPERTY INTERFACE_LINK_LIBRARIES pthread gtest)
-
-add_library(gmock STATIC IMPORTED GLOBAL)
-add_dependencies(gmock googletest)
-set_property(TARGET gmock PROPERTY IMPORTED_LOCATION ${INSTALL_DIR}/lib/libgmock.a)
-set_property(TARGET gmock PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${INSTALL_DIR}/include)
-
-add_library(gmock_main STATIC IMPORTED GLOBAL)
-add_dependencies(gmock_main googletest)
-set_property(TARGET gmock_main PROPERTY IMPORTED_LOCATION ${INSTALL_DIR}/lib/libgmock_main.a)
-set_property(TARGET gmock_main PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${INSTALL_DIR}/include)
+add_library(libcmocka STATIC IMPORTED GLOBAL)
+add_dependencies(libcmocka cmocka)
+set_target_properties(libcmocka PROPERTIES
+    IMPORTED_LOCATION ${INSTALL_DIR}/lib64/libcmocka.a
+    INTERFACE_LINK_DIRECTORIES ${INSTALL_DIR}/lib64
+    INTERFACE_INCLUDE_DIRECTORIES ${INSTALL_DIR}/include
+)
 
 #### libevent
 ExternalProject_Add(libevent PREFIX libevent
diff --git a/support/cmocka-1.1.7.tar.xz b/support/cmocka-1.1.7.tar.xz
new file mode 100644
index 0000000..42aedb0
--- /dev/null
+++ b/support/cmocka-1.1.7.tar.xz
diff --git a/support/googletest-release-1.8.0.tar.gz b/support/googletest-release-1.8.0.tar.gz
deleted file mode 100644
index a40df33..0000000
--- a/support/googletest-release-1.8.0.tar.gz
+++ /dev/null
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
new file mode 100644
index 0000000..1ba8214
--- /dev/null
+++ b/test/CMakeLists.txt
@@ -0,0 +1,13 @@
+include_directories(${CMAKE_SOURCE_DIR}/support/mpack)
+include_directories(${CMAKE_SOURCE_DIR}/src)
+
+set(DP_TRACE_TELEMETRY_SOURCES 
+    ${CMAKE_SOURCE_DIR}/src/config.c 
+    ${CMAKE_SOURCE_DIR}/src/job_ctx.c 
+    ${CMAKE_SOURCE_DIR}/src/trace_output.c
+    ${CMAKE_SOURCE_DIR}/src/kafka.c
+    ${CMAKE_SOURCE_DIR}/src/mocking.c
+    ${CMAKE_SOURCE_DIR}/support/mpack/mpack.c)
+
+add_executable(cmocka_test cmocka_test.c  ${DP_TRACE_TELEMETRY_SOURCES})
+target_link_libraries(cmocka_test libmarsio maatframe MESA_prof_load libevent-static rdkafka libcmocka uuid pthread)
diff --git a/test/cmocka_test.c b/test/cmocka_test.c
new file mode 100644
index 0000000..0c59699
--- /dev/null
+++ b/test/cmocka_test.c
@@ -0,0 +1,44 @@
+#include "common.h"
+#include "config.h"
+
+#include <setjmp.h>
+#include <stdarg.h>
+#include <stddef.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <cmocka.h>
+
+struct mr_instance * mr_instance = NULL;
+
+int setup(void ** state)
+{
+    mr_instance = marsio_create();
+    return 0;
+}
+
+int teardown(void ** state)
+{
+    if (g_conf != NULL)
+    {
+        free(g_conf);
+        g_conf == NULL;
+        printf("free g_conf\n");
+    }
+    return 0;
+}
+
+void config_create_test(void ** state)
+{
+    config_create();
+    assert_non_null(g_conf);
+}
+
+int main(int argc, char * argv[])
+{
+    const struct CMUnitTest config_tests[] = {
+        cmocka_unit_test_setup_teardown(config_create_test, setup, teardown),
+    };
+    cmocka_run_group_tests(config_tests, NULL, NULL);
+    return 0;
+}
\ No newline at end of file