FIX:set label_buflen as -1 when no label founded

1 files changed, 111 insertions, 117 deletions

diff --git a/src/ntc_ip_comm.c b/src/ntc_ip_comm.c
index 2e4e914..b5a02e7 100644
--- a/src/ntc_ip_comm.c
+++ b/src/ntc_ip_comm.c
@@ -1,11 +1,3 @@
-/*

- *  t1_rawpkt

- *  author:yangwei

- *  time:2016-09-03

- *  last_version:1.20160903

- *

- */

-

 #include "ntc_ip_comm.h"

 #include "soq_sendlog.h"

 

@@ -33,7 +25,7 @@
 #include <rdkafka.h>

 #include <cJSON.h>

 

-#define PLUGIN_NAME "NTC_IP_COMMs"

+#define PLUGIN_NAME "NTC_IP_COMM"

 #define PROFILE_PATH "./ntcconf/main.conf"

 #define NTC_IP_COMM_TOPIC "ntc_ip_comm_log"

 

@@ -48,27 +40,27 @@ rd_kafka_topic_t *g_ntc_ip_comm_kafka_topic = NULL;
 

 static unsigned int get_ip_by_eth_name(const char *ifname)

 {

-    int sockfd;

-    struct ifreq ifr;

-    unsigned int ip;

+	int sockfd;

+	struct ifreq ifr;

+	unsigned int ip;

 

-    sockfd = socket(AF_INET, SOCK_DGRAM, 0);

-    if (-1 == sockfd) {

-        goto error;

-    }

+	sockfd = socket(AF_INET, SOCK_DGRAM, 0);

+	if (-1 == sockfd) {

+		goto error;

+	}

 

-    strcpy(ifr.ifr_name,ifname);

-    if (ioctl(sockfd, SIOCGIFADDR, &ifr) < 0) {

-        goto error;

-    }

+	strcpy(ifr.ifr_name,ifname);

+	if (ioctl(sockfd, SIOCGIFADDR, &ifr) < 0) {

+		goto error;

+	}

 

-    ip = ((struct sockaddr_in*)&(ifr.ifr_addr))->sin_addr.s_addr;

-    close(sockfd);

-    return ip;

+	ip = ((struct sockaddr_in*)&(ifr.ifr_addr))->sin_addr.s_addr;

+	close(sockfd);

+	return ip;

 

 error:

-    close(sockfd);

-    return INADDR_NONE;

+	close(sockfd);

+	return INADDR_NONE;

 }

 

 static int soq_addStreamInfo_to_jsonObj(cJSON *json_obj, const struct streaminfo *a_stream)

@@ -143,7 +135,7 @@ static int soq_addStreamInfo_to_jsonObj(cJSON *json_obj, const struct streaminfo
 	{		

 		stream_addr_list_ntop(a_stream,nest_addr_buf, sizeof(nest_addr_buf));

 	}

-	

+

 	cJSON_AddStringToObject(json_obj, "addr_list", nest_addr_buf);

 

 	return 0;

@@ -154,89 +146,92 @@ void ntc_ip_comm_send_kafka_log(rd_kafka_topic_t *topic, struct streaminfo *a_st
 {

 	cJSON *log_obj = cJSON_CreateObject();

 

-    cJSON_AddNumberToObject(log_obj, "service", g_ntc_ip_comm_item.service);

-    cJSON_AddStringToObject(log_obj, "cap_ip", g_ntc_ip_comm_item.local_ip_str);

-    cJSON_AddNumberToObject(log_obj, "entrance_id", g_ntc_ip_comm_item.entry_id);

+	cJSON_AddNumberToObject(log_obj, "service", g_ntc_ip_comm_item.service);

+	cJSON_AddStringToObject(log_obj, "cap_ip", g_ntc_ip_comm_item.local_ip_str);

+	cJSON_AddNumberToObject(log_obj, "entrance_id", g_ntc_ip_comm_item.entry_id);

 

 	cJSON_AddNumberToObject(log_obj, "found_time", g_CurrentTime);

 	cJSON_AddNumberToObject(log_obj, "recv_time", g_CurrentTime);    

-    soq_addStreamInfo_to_jsonObj(log_obj, a_stream);

-	

-	cJSON_AddNumberToObject(log_obj, "c2s_pkt_num", a_stream->ptcpdetail->clientpktnum);
	

+	soq_addStreamInfo_to_jsonObj(log_obj, a_stream);

+

+	cJSON_AddNumberToObject(log_obj, "c2s_pkt_num", a_stream->ptcpdetail->clientpktnum);	

 	cJSON_AddNumberToObject(log_obj, "s2c_pkt_num", a_stream->ptcpdetail->serverpktnum);

 	cJSON_AddNumberToObject(log_obj, "c2s_byte_num", a_stream->ptcpdetail->clientbytes);

 	cJSON_AddNumberToObject(log_obj, "s2c_byte_num", a_stream->ptcpdetail->serverbytes);

-    if(dpkt_buf != NULL && dpkt_buflen > 0)

-    {

-        cJSON_AddStringToObject(log_obj, "app_label", dpkt_buf);

-    }

-    char *payload = cJSON_Print(log_obj);

-    int paylen = strlen(payload);

-    rd_kafka_produce(g_ntc_ip_comm_kafka_topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, payload, paylen, NULL, 0, NULL);

-

-    free(payload);

-    cJSON_Delete(log_obj);

-    log_obj = NULL;

-    return ;

+	if(dpkt_buf != NULL && dpkt_buflen > 0)

+	{

+		cJSON_AddStringToObject(log_obj, "app_label", dpkt_buf);

+	}

+	char *payload = cJSON_Print(log_obj);

+	int paylen = strlen(payload);

+	rd_kafka_produce(g_ntc_ip_comm_kafka_topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, payload, paylen, NULL, 0, NULL);

+

+	free(payload);

+	cJSON_Delete(log_obj);

+	log_obj = NULL;

+	return ;

 }

 

 void ntc_ip_comm_send_ntc_log(struct streaminfo *a_stream, char *dpkt_buf, int dpkt_buflen)

 {

-    soq_log_t log_msg;

-    Maat_rule_t maat_rule;

-    struct opt_unit_t opts[5];

-    int opt_num = 4;

-

-    memset(&maat_rule, 0, sizeof(maat_rule));

-    maat_rule.action = SOQ_ACTION_MONITOR;

-    maat_rule.config_id = 0;

-    maat_rule.service_id = g_ntc_ip_comm_item.service;

-    maat_rule.do_log = 1;

-    log_msg.stream = a_stream;

-    log_msg.result = &maat_rule;

-    log_msg.result_num =1;

-

-    opts[0].opt_len = sizeof(a_stream->ptcpdetail->clientbytes);

-    opts[0].opt_type =LOG_OPT_C2S_BYTE_NUM ;

-    opts[0].opt_value = &(a_stream->ptcpdetail->clientbytes);    

-

-    opts[1].opt_len = sizeof(a_stream->ptcpdetail->serverbytes);

-    opts[1].opt_type =LOG_OPT_S2C_BYTE_NUM ;

-    opts[1].opt_value = &(a_stream->ptcpdetail->serverbytes);    

-

-    opts[2].opt_len = sizeof(a_stream->ptcpdetail->clientpktnum);

-    opts[2].opt_type =LOG_OPT_C2S_PKT_NUM ;

-    opts[2].opt_value = &(a_stream->ptcpdetail->clientpktnum);    

-

-    opts[3].opt_len = sizeof(a_stream->ptcpdetail->serverpktnum);

-    opts[3].opt_type =LOG_OPT_S2C_PKT_NUM ;

-    opts[3].opt_value = &(a_stream->ptcpdetail->serverpktnum);    

-

-    if(dpkt_buf != NULL && dpkt_buflen > 0)

-    {

-        opts[4].opt_len = dpkt_buflen;

-        opts[4].opt_type =LOG_OPT_APP_LABEL ;

-        opts[4].opt_value =dpkt_buf;        

-        opt_num+=1;

-    }

-    soq_send_log(&log_msg, opts, opt_num, a_stream->threadnum);

-    return;

+	soq_log_t log_msg;

+	Maat_rule_t maat_rule;

+	struct opt_unit_t opts[5];

+	int opt_num = 4;

+

+	memset(&maat_rule, 0, sizeof(maat_rule));

+	maat_rule.action = SOQ_ACTION_MONITOR;

+	maat_rule.config_id = 0;

+	maat_rule.service_id = g_ntc_ip_comm_item.service;

+	maat_rule.do_log = 1;

+	log_msg.stream = a_stream;

+	log_msg.result = &maat_rule;

+	log_msg.result_num =1;

+

+	opts[0].opt_len = sizeof(a_stream->ptcpdetail->clientbytes);

+	opts[0].opt_type =LOG_OPT_C2S_BYTE_NUM ;

+	opts[0].opt_value = &(a_stream->ptcpdetail->clientbytes);    

+

+	opts[1].opt_len = sizeof(a_stream->ptcpdetail->serverbytes);

+	opts[1].opt_type =LOG_OPT_S2C_BYTE_NUM ;

+	opts[1].opt_value = &(a_stream->ptcpdetail->serverbytes);    

+

+	opts[2].opt_len = sizeof(a_stream->ptcpdetail->clientpktnum);

+	opts[2].opt_type =LOG_OPT_C2S_PKT_NUM ;

+	opts[2].opt_value = &(a_stream->ptcpdetail->clientpktnum);    

+

+	opts[3].opt_len = sizeof(a_stream->ptcpdetail->serverpktnum);

+	opts[3].opt_type =LOG_OPT_S2C_PKT_NUM ;

+	opts[3].opt_value = &(a_stream->ptcpdetail->serverpktnum);    

+

+	if(dpkt_buf != NULL && dpkt_buflen > 0)

+	{

+		opts[4].opt_len = dpkt_buflen;

+		opts[4].opt_type =LOG_OPT_APP_LABEL ;

+		opts[4].opt_value =dpkt_buf;        

+		opt_num+=1;

+	}

+	soq_send_log(&log_msg, opts, opt_num, a_stream->threadnum);

+	return;

 }

 

 

 int ntc_ip_comm_get_dpkt_label(struct streaminfo *a_stream, const char* label_name, char *label_buf, int *label_buflen)

 {

-    dpkt_lable_t *dpkt_info = (dpkt_lable_t*)project_req_get_struct(a_stream,g_ntc_ip_comm_item.dpkt_project_id);

-	if(dpkt_info == NULL)return -1;

-	

+	dpkt_lable_t *dpkt_info = (dpkt_lable_t*)project_req_get_struct(a_stream,g_ntc_ip_comm_item.dpkt_project_id);

+	if(dpkt_info == NULL)

+	{

+		*label_buflen = 0;

+		return -1;

+	}

 	snprintf(label_buf, *label_buflen, "serv=%u;proto=%u;app=%u;os=%u;brow=%u;web=%u;beh=%u;",

-	        dpkt_info->dpkt_service_type,

-	        dpkt_info->dpkt_proto_type,	        

-	        dpkt_info->dpkt_app_type,	        

-	        dpkt_info->dpkt_op_type,	        

-	        dpkt_info->dpkt_browser_type,	        

-	        dpkt_info->dpkt_web_type,	        

-	        dpkt_info->dpkt_behavior_type);

+			dpkt_info->dpkt_service_type,

+			dpkt_info->dpkt_proto_type,	        

+			dpkt_info->dpkt_app_type,	        

+			dpkt_info->dpkt_op_type,	        

+			dpkt_info->dpkt_browser_type,	        

+			dpkt_info->dpkt_web_type,	        

+			dpkt_info->dpkt_behavior_type);

 	*label_buflen = strlen(label_buf);

 	return 0;

 }

@@ -254,21 +249,21 @@ extern "C" UCHAR ntc_ip_comm_transfer_entry(struct streaminfo *a_stream,  void *
 	}

 	if(state == OP_STATE_CLOSE)

 	{

-	    char label_buf[128] = {0};

-	    int label_buflen = sizeof(label_buf);

-	    ntc_ip_comm_get_dpkt_label(a_stream, g_ntc_ip_comm_item.dpkt_label, label_buf, &label_buflen);

-	    if(a_stream->ptcpdetail->clientbytes+ a_stream->ptcpdetail->serverbytes >= g_ntc_ip_comm_item.min_bytes && a_stream->ptcpdetail->clientpktnum+ a_stream->ptcpdetail->serverpktnum >= g_ntc_ip_comm_item.min_pkts)

-        {

-    	    if(g_ntc_ip_comm_item.using_kafka == 1)

-    	    {

-                ntc_ip_comm_send_kafka_log(g_ntc_ip_comm_kafka_topic, a_stream, label_buf, label_buflen);

-    	    }

-    	    else

-    	    {

-                ntc_ip_comm_send_ntc_log(a_stream, label_buf, label_buflen);

-    	    }

-	    }

-	    return APP_STATE_DROPME;	

+		char label_buf[128] = {0};

+		int label_buflen = sizeof(label_buf);

+		ntc_ip_comm_get_dpkt_label(a_stream, g_ntc_ip_comm_item.dpkt_label, label_buf, &label_buflen);

+		if(a_stream->ptcpdetail->clientbytes+ a_stream->ptcpdetail->serverbytes >= g_ntc_ip_comm_item.min_bytes && a_stream->ptcpdetail->clientpktnum+ a_stream->ptcpdetail->serverpktnum >= g_ntc_ip_comm_item.min_pkts)

+		{

+			if(g_ntc_ip_comm_item.using_kafka == 1)

+			{

+				ntc_ip_comm_send_kafka_log(g_ntc_ip_comm_kafka_topic, a_stream, label_buf, label_buflen);

+			}

+			else

+			{

+				ntc_ip_comm_send_ntc_log(a_stream, label_buf, label_buflen);

+			}

+		}

+		return APP_STATE_DROPME;	

 	}

 	return APP_STATE_GIVEME;

 }

@@ -278,12 +273,12 @@ void ntc_ip_comm_load_profile()
 	MESA_load_profile_string_def(PROFILE_PATH, PLUGIN_NAME, "log_path", g_ntc_ip_comm_item.log_path, sizeof(g_ntc_ip_comm_item.log_path), "./log/t1_rawpkt_log");

 	MESA_load_profile_uint_def(PROFILE_PATH,PLUGIN_NAME, "log_level", &g_ntc_ip_comm_item.log_level, 30);

 

-    char nic_name[64];

+	char nic_name[64];

 	MESA_load_profile_string_def(PROFILE_PATH,"SYSTEM", "NIC_NAME",nic_name,sizeof(nic_name),"eth0");

 	g_ntc_ip_comm_item.local_ip_nr=get_ip_by_eth_name(nic_name);

 	if(g_ntc_ip_comm_item.local_ip_nr==INADDR_NONE)

 	{

-	    printf("get_ip_by_eth_name in %s return NULL, exit!\n", nic_name);

+		printf("get_ip_by_eth_name in %s return NULL, exit!\n", nic_name);

 		exit(0);

 	}

 	inet_ntop(AF_INET,&(g_ntc_ip_comm_item.local_ip_nr),g_ntc_ip_comm_item.local_ip_str,sizeof(g_ntc_ip_comm_item.local_ip_str));

@@ -294,16 +289,16 @@ void ntc_ip_comm_load_profile()
 	MESA_load_profile_uint_def(PROFILE_PATH,PLUGIN_NAME, "min_bytes", &g_ntc_ip_comm_item.min_bytes, 5);

 

 	MESA_load_profile_int_def(PROFILE_PATH,PLUGIN_NAME, "service", &g_ntc_ip_comm_item.service, 0);

-	

+

 	MESA_load_profile_uint_def(PROFILE_PATH,PLUGIN_NAME, "using_kafka", &g_ntc_ip_comm_item.using_kafka, 0);

-	

+

 	MESA_load_profile_string_def(PROFILE_PATH, PLUGIN_NAME, "dpkt_label", g_ntc_ip_comm_item.dpkt_label, sizeof(g_ntc_ip_comm_item.dpkt_label), "DPKT_PROJECT");

 	return ;

 }

 

 int ntc_ip_comm_kaka_init()

 {

-    char kafka_errstr[1024];

+	char kafka_errstr[1024];

 	MESA_load_profile_uint_def(PROFILE_PATH,PLUGIN_NAME, "indie_kafka", &g_ntc_ip_comm_item.indie_kafka, 0);

 	MESA_load_profile_string_def(PROFILE_PATH, PLUGIN_NAME, "kafka_topic", g_ntc_ip_comm_item.kafka_topic, sizeof(g_ntc_ip_comm_item.kafka_topic), NTC_IP_COMM_TOPIC);

 	if(g_ntc_ip_comm_item.indie_kafka == 1)

@@ -366,9 +361,9 @@ extern "C" int ntc_ip_comm_init()
 	g_ntc_ip_comm_item.dpkt_project_id = project_customer_register(g_ntc_ip_comm_item.dpkt_label, "struct");	

 	if(g_ntc_ip_comm_item.using_kafka == 1)

 	{

-    	if(0 != ntc_ip_comm_kaka_init())

-    		return ERROR;

-    }

+		if(0 != ntc_ip_comm_kaka_init())

+			return ERROR;

+	}

 	return OK;

 }

 

@@ -379,4 +374,3 @@ extern "C" void ntc_ip_comm_destroy(void)
 	MESA_destroy_runtime_log_handle(g_ntc_ip_comm_item.log_handle);

 	return;

 }

-