fix mem leak

3 files changed, 53 insertions, 55 deletions

diff --git a/shaping/include/shaper_stat.h b/shaping/include/shaper_stat.h
index 124757c..22ca620 100644
--- a/shaping/include/shaper_stat.h
+++ b/shaping/include/shaper_stat.h
@@ -49,8 +49,7 @@ struct shaping_stat {
     rd_kafka_t *kafka_handle; 
     rd_kafka_topic_t *topic_rkt;
     int output_interval_s;
-    int in_latency_histogram_id;
-    int out_latency_histogram_id;
+    int latency_histogram_id;
     unsigned int column_ids[STAT_COLUNM_IDX_MAX];
 };
 
diff --git a/shaping/src/shaper_stat.cpp b/shaping/src/shaper_stat.cpp
index f629425..e37f5e0 100644
--- a/shaping/src/shaper_stat.cpp
+++ b/shaping/src/shaper_stat.cpp
@@ -15,8 +15,6 @@
 #define SHAPER_STAT_REFRESH_TIME_US 10000  //10 ms
 #define HINCRBY_RETRY_MAX 5
 
-#define OUTPUT_JSON_BUF_LEN 2048
-
 struct shaper_stat_conf {
     char device_group[32];
     char device_id[32];
@@ -47,10 +45,6 @@ void shaper_stat_destroy(struct shaping_stat *stat)
         free(stat);
     }
 
-    if (output_json_buf) {
-        free(output_json_buf);
-    }
-
     return;
 }
 
@@ -70,6 +64,8 @@ static void shaper_stat_kafka_init(struct shaping_stat *stat, struct shaper_stat
 
     rd_kafka_conf_t *rdkafka_conf = rd_kafka_conf_new();
 
+    rd_kafka_conf_set(rdkafka_conf, "queue.buffering.max.messages", "1000000", kafka_errstr, sizeof(kafka_errstr));
+    rd_kafka_conf_set(rdkafka_conf, "topic.metadata.refresh.interval.ms", "600000", kafka_errstr, sizeof(kafka_errstr));
     rd_kafka_conf_set(rdkafka_conf, "socket.keepalive.enable", "true", kafka_errstr, sizeof(kafka_errstr));	
 	rd_kafka_conf_set(rdkafka_conf, "bootstrap.servers", conf->kafka_brokers, kafka_errstr, sizeof(kafka_errstr));
     rd_kafka_conf_set(rdkafka_conf, "security.protocol", "sasl_plaintext", kafka_errstr, sizeof(kafka_errstr));
@@ -109,8 +105,6 @@ struct shaping_stat* shaper_stat_init(int thread_num)
     struct shaper_stat_conf conf;
     struct shaping_stat *stat = (struct shaping_stat *)calloc(1, sizeof(struct shaping_stat));
 
-    output_json_buf = (char *)calloc(1, OUTPUT_JSON_BUF_LEN);
-
     if (shaper_stat_conf_load(stat, &conf) != 0) {
         LOG_ERROR("%s: shaping init metric conf failed", LOG_TAG_STAT);
         goto ERROR;
@@ -144,9 +138,8 @@ struct shaping_stat* shaper_stat_init(int thread_num)
         goto ERROR;
     }
 
-    stat->in_latency_histogram_id = fieldstat_easy_register_histogram(stat->instance, "in_latency_us", 1, 1000000, 5);
-    stat->out_latency_histogram_id = fieldstat_easy_register_histogram(stat->instance, "out_latency_us", 1, 1000000, 5);
-    if (stat->in_latency_histogram_id < 0 || stat->out_latency_histogram_id < 0) {
+    stat->latency_histogram_id = fieldstat_easy_register_histogram(stat->instance, "latency_distribution_us", 1, 1000000, 5);
+    if (stat->latency_histogram_id < 0) {
         LOG_ERROR("%s: shaping fieldstat register histogram failed", LOG_TAG_STAT);
         goto ERROR;
     }
@@ -177,9 +170,7 @@ ERROR:
         }
         free(stat);
     }
-    if (output_json_buf) {
-        free(output_json_buf);
-    }
+
     return NULL;
 }
 
@@ -332,8 +323,7 @@ static void shaper_stat_profile_metirc_refresh(struct shaping_thread_ctx *ctx, s
     fieldstat_easy_counter_incrby(stat->instance, thread_id, stat->column_ids[OUT_PKTS_IDX], tags, TAG_IDX_MAX, profile_stat->out.pkts);
     fieldstat_easy_counter_incrby(stat->instance, thread_id, stat->column_ids[OUT_BYTES_IDX], tags, TAG_IDX_MAX, profile_stat->out.bytes);
 
-    fieldstat_easy_histogram_record(stat->instance, thread_id, stat->in_latency_histogram_id, tags, TAG_IDX_MAX, profile_stat->in.max_latency);
-    fieldstat_easy_histogram_record(stat->instance, thread_id, stat->out_latency_histogram_id, tags, TAG_IDX_MAX, profile_stat->out.max_latency);
+    fieldstat_easy_histogram_record(stat->instance, thread_id, stat->latency_histogram_id, tags, TAG_IDX_MAX, profile_stat->out.max_latency);
 
     if (need_update_guage) {
         if (profile_type == PROFILE_IN_RULE_TYPE_PRIMARY) {
@@ -497,7 +487,7 @@ void shaper_stat_max_latency_update(struct shaping_stat_for_profile *profile_sta
 
 void shaper_stat_output(struct shaping_stat *stat)
 {
-    size_t len = OUTPUT_JSON_BUF_LEN;
+    size_t len = 0;
     fieldstat_easy_output(stat->instance, &output_json_buf, &len);
 
     int status=rd_kafka_produce(stat->topic_rkt, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, output_json_buf, len, NULL, 0, NULL);
@@ -510,5 +500,10 @@ void shaper_stat_output(struct shaping_stat *stat)
 							status);
     }
 
+    if (output_json_buf) {
+        free(output_json_buf);
+        output_json_buf = NULL;
+    }
+
     return;
 }
\ No newline at end of file
diff --git a/shaping/test/gtest_shaper.cpp b/shaping/test/gtest_shaper.cpp
index fbe98eb..cde8aef 100644
--- a/shaping/test/gtest_shaper.cpp
+++ b/shaping/test/gtest_shaper.cpp
@@ -14,8 +14,6 @@
 #define SHAPING_SESSION_QUEUE_LEN 128
 #define FIELDSTAT_AUTO_TIME_MAX 999999000
 
-#define FIELDSTAT_OUTPUT_BUF_LEN 4096
-
 char profile_type_primary[] = "primary";
 char profile_type_borrow[] = "borrow";
 
@@ -108,6 +106,7 @@ static void shaping_stat_judge(char *file_line, int json_array_idx, int rule_id,
                             unsigned char direction, char profile_type[])
 {
     cJSON *json = NULL;
+    cJSON *json_array_element = NULL;
     cJSON *fields_json = NULL;
     cJSON *tags_json = NULL;
     cJSON *tmp_obj = NULL;
@@ -119,14 +118,14 @@ static void shaping_stat_judge(char *file_line, int json_array_idx, int rule_id,
     ASSERT_EQ(json->type, cJSON_Array);
     ASSERT_GT(cJSON_GetArraySize(json), json_array_idx);
 
-    json = cJSON_GetArrayItem(json, json_array_idx);
+    json_array_element = cJSON_GetArrayItem(json, json_array_idx);
 
-    tmp_obj = cJSON_GetObjectItem(json, "name");
+    tmp_obj = cJSON_GetObjectItem(json_array_element, "name");
     ASSERT_TRUE(tmp_obj != NULL);
     EXPECT_STREQ("shaping_stat", tmp_obj->valuestring);
 
     /******************parse tags***********************************/
-    tags_json = cJSON_GetObjectItem(json, "tags");
+    tags_json = cJSON_GetObjectItem(json_array_element, "tags");
     ASSERT_TRUE(tags_json != NULL);
 
     tmp_obj = cJSON_GetObjectItem(tags_json, "vsys_id");
@@ -150,7 +149,7 @@ static void shaping_stat_judge(char *file_line, int json_array_idx, int rule_id,
     EXPECT_STREQ(tmp_obj->valuestring, profile_type);
 
     /******************parse fields**********************************/
-    fields_json = cJSON_GetObjectItem(json, "fields");
+    fields_json = cJSON_GetObjectItem(json_array_element, "fields");
     ASSERT_TRUE(fields_json != NULL);
 
     snprintf(attr_name, sizeof(attr_name), "%s_pkts", direction == SHAPING_DIR_OUT ? "out" : "in");
@@ -202,11 +201,12 @@ static int shaping_global_stat_field_get(cJSON *metrics, const char *field_name)
 static void shaping_global_stat_judge(char *file_line, int tx_pkts, int tx_bytes, int drop_pkts, int drop_bytes, int queueing_pkts, int queueing_bytes)
 {
     cJSON *metrics = NULL;
+    cJSON *json_array_element = NULL;
     cJSON *json = cJSON_Parse(file_line);
 
-    json = cJSON_GetArrayItem(json, 0);
+    json_array_element = cJSON_GetArrayItem(json, 0);
 
-    metrics = cJSON_GetObjectItem(json, "fields");
+    metrics = cJSON_GetObjectItem(json_array_element, "fields");
 
     EXPECT_EQ(tx_pkts, shaping_global_stat_field_get(metrics, "all_tx_pkts"));
     EXPECT_EQ(tx_bytes, shaping_global_stat_field_get(metrics, "all_tx_bytes"));
@@ -272,10 +272,10 @@ TEST(single_session, udp_tx_in_order)
     shaping_flow_free(&ctx->thread_ctx[0], sf);
 
     /***********send stat data here********************/
-    char *global_stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t global_stat_str_len = sizeof(global_stat_str);
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *global_stat_str = NULL;
+    size_t global_stat_str_len = 0;
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     shaper_thread_global_stat_refresh(&ctx->thread_ctx[0]);
 
@@ -411,10 +411,10 @@ TEST(max_min_host_fairness_profile, udp_tx_in_order)
     shaping_flow_free(&ctx->thread_ctx[0], sf);
 
     /***********send stat data here********************/
-    char *global_stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t global_stat_str_len = sizeof(global_stat_str);
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *global_stat_str = NULL;
+    size_t global_stat_str_len = 0;
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     shaper_thread_global_stat_refresh(&ctx->thread_ctx[0]);
 
@@ -477,12 +477,13 @@ TEST(single_session, tcp_tx_in_order)
 
 
     /***********send stat data here********************/
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     shaper_stat_refresh(&ctx->thread_ctx[0], sf, 1);
     fieldstat_easy_output(ctx->thread_ctx[0].stat->instance, &stat_str, &stat_str_len);
     shaping_stat_judge(stat_str, 0, 0, 0, 1, 20, 2000, 0, 10, 0, SHAPING_DIR_OUT, profile_type_primary);//*test statistics
+    free(stat_str);
 
 
     stub_refresh_token_bucket(0);
@@ -568,8 +569,8 @@ TEST(single_session, udp_diff_direction)
     shaping_flow_free(&ctx->thread_ctx[0], sf);
 
     /***********send stat data here********************/
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     fieldstat_easy_output(ctx->thread_ctx[0].stat->instance, &stat_str, &stat_str_len);
 
@@ -640,8 +641,8 @@ TEST(single_session, udp_multi_rules)
     shaping_flow_free(&ctx->thread_ctx[0], sf);
 
     /***********send stat data here********************/
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     fieldstat_easy_output(ctx->thread_ctx[0].stat->instance, &stat_str, &stat_str_len);
 
@@ -710,8 +711,8 @@ TEST(single_session, udp_borrow)
     shaping_flow_free(&ctx->thread_ctx[0], sf);
 
     /***********send stat data here********************/
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     fieldstat_easy_output(ctx->thread_ctx[0].stat->instance, &stat_str, &stat_str_len);
 
@@ -780,8 +781,8 @@ TEST(single_session, udp_borrow_same_priority_9)
     shaping_flow_free(&ctx->thread_ctx[0], sf);
 
     /***********send stat data here********************/
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     fieldstat_easy_output(ctx->thread_ctx[0].stat->instance, &stat_str, &stat_str_len);
 
@@ -942,8 +943,8 @@ TEST(two_session_diff_priority_same_profile, udp_borrow_in_order)
     shaping_flow_free(&ctx->thread_ctx[0], sf2);
 
     /***********send stat data here********************/
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     fieldstat_easy_output(ctx->thread_ctx[0].stat->instance, &stat_str, &stat_str_len);
 
@@ -1511,10 +1512,10 @@ TEST(statistics, udp_drop_pkt)
     shaping_flow_free(&ctx->thread_ctx[0], sf);
 
     /***********send stat data here********************/
-    char *global_stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t global_stat_str_len = sizeof(global_stat_str);
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *global_stat_str = NULL;
+    size_t global_stat_str_len = 0;
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     shaper_thread_global_stat_refresh(&ctx->thread_ctx[0]);
 
@@ -1568,10 +1569,10 @@ TEST(statistics, udp_queueing_pkt)
 
 
     /***********send stat data here********************/
-    char *global_stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t global_stat_str_len = sizeof(global_stat_str);
-    char *stat_str = (char*)malloc(FIELDSTAT_OUTPUT_BUF_LEN);
-    size_t stat_str_len = sizeof(stat_str);
+    char *global_stat_str = NULL;
+    size_t global_stat_str_len = 0;
+    char *stat_str = NULL;
+    size_t stat_str_len = 0;
     
     shaper_thread_global_stat_refresh(&ctx->thread_ctx[0]);
     shaper_stat_refresh(&ctx->thread_ctx[0], sf, 1);
@@ -1583,6 +1584,9 @@ TEST(statistics, udp_queueing_pkt)
     shaping_stat_judge(stat_str, 0, 0, 0, 1, 10, 1000, 0, 90, 0, SHAPING_DIR_OUT, profile_type_primary);
     shaping_global_stat_judge(global_stat_str, 10, 1000, 0, 0, 90, 9000);
 
+    free(global_stat_str);
+    free(stat_str);
+
     //first 10 packets
     ASSERT_EQ(0, judge_packet_eq(&expec_tx_queue, actual_tx_queue, 10));
     ASSERT_TRUE(TAILQ_EMPTY(actual_tx_queue));