修改日志字段，按account入partition

1 files changed, 33 insertions, 31 deletions

diff --git a/src/ntc_radius_plug.cpp b/src/ntc_radius_plug.cpp
index 28f1c51..3c5dc09 100644
--- a/src/ntc_radius_plug.cpp
+++ b/src/ntc_radius_plug.cpp
@@ -25,7 +25,7 @@
 
 #include "ntc_radius_plug.h"
 
-static int NTC_RADIUS_PLUG_VERSION_20191022=0;
+static int NTC_RADIUS_PLUG_VERSION_20191108=0;
 const char *config_file="t1conf/main.conf";
 g_ntc_radius_plug_t g_ntc_radius_plug;
 
@@ -65,88 +65,89 @@ error:
 }
 
 
-int get_radius_object_element(cJSON *radius_info_object, radius_header_t *radius_header, radius_body_t *radius_body)
+int get_radius_object_element(cJSON *radius_info_object, radius_header_t *radius_header, radius_body_t *radius_body, int *account_index)
 {
 	int i=0;
 	cJSON *radius_json_obj = radius_info_object;
 		
-	cJSON_AddNumberToObject(radius_json_obj, "CODE", radius_header->code);
+	cJSON_AddNumberToObject(radius_json_obj, "packet_type", radius_header->code);
 
 	for(i=0; i<radius_body->attribute_num; i++)
 	{
 		switch(radius_body->attribute[i].type)
 		{
 			case	RADIUS_USER_NAME:
-				cJSON_AddStringToObject(radius_json_obj, "user_name", (char *)(radius_body->attribute[i].value));
+				cJSON_AddStringToObject(radius_json_obj, "account", (char *)(radius_body->attribute[i].value));
+				*account_index=i;
 				break;
 			case	CALLBACK_NUMBER:
- 				cJSON_AddStringToObject(radius_json_obj, "CALLBACK_NUMBER", (char *)(radius_body->attribute[i].value));
+ 				cJSON_AddStringToObject(radius_json_obj, "callback_number", (char *)(radius_body->attribute[i].value));
 				break;
 			case	CALLBACK_ID:
-				cJSON_AddStringToObject(radius_json_obj, "CALLBACK_ID", (char *)(radius_body->attribute[i].value));
+				cJSON_AddStringToObject(radius_json_obj, "callback_id", (char *)(radius_body->attribute[i].value));
 				break;
 			case	CALLED_STATION_ID:
- 				cJSON_AddStringToObject(radius_json_obj, "CALLED_STATION_ID", (char *)(radius_body->attribute[i].value));
+ 				cJSON_AddStringToObject(radius_json_obj, "called_station_id", (char *)(radius_body->attribute[i].value));
 				break;
 			case	CALLING_STATION_ID:
-  				cJSON_AddStringToObject(radius_json_obj, "CALLING_STATION_ID", (char *)(radius_body->attribute[i].value));
+  				cJSON_AddStringToObject(radius_json_obj, "calling_station_id", (char *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_SESSION_ID:
-   				cJSON_AddStringToObject(radius_json_obj, "ACCT_SESSION_ID", (char *)(radius_body->attribute[i].value));
+   				cJSON_AddStringToObject(radius_json_obj, "acct_session_id", (char *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_MULTI_SESSION_ID:
-    			cJSON_AddStringToObject(radius_json_obj, "ACCT_MULTI_SESSION_ID", (char *)(radius_body->attribute[i].value));
+				cJSON_AddStringToObject(radius_json_obj, "acct_multi_session_id", (char *)(radius_body->attribute[i].value));
 				break;
 			case	NAS_IP_ADDRESS:
-     			cJSON_AddStringToObject(radius_json_obj, "NAS_IP_ADDRESS", (char *)(radius_body->attribute[i].value));
+				cJSON_AddStringToObject(radius_json_obj, "nas_ip", (char *)(radius_body->attribute[i].value));
 				break;
 			case	FRAMED_IP_ADDRESS:
-      			cJSON_AddStringToObject(radius_json_obj, "FRAMED_IP_ADDRESS", (char *)(radius_body->attribute[i].value));
+				cJSON_AddStringToObject(radius_json_obj, "framed_ip", (char *)(radius_body->attribute[i].value));
 				break;
 			case	FRAMED_IP_NETMASK:
-       			cJSON_AddStringToObject(radius_json_obj, "FRAMED_IP_NETMASK", (char *)(radius_body->attribute[i].value));
+				cJSON_AddStringToObject(radius_json_obj, "framed_ip_netmask", (char *)(radius_body->attribute[i].value));
 				break;
 			case	SERVICE_TYPE:
- 				cJSON_AddNumberToObject(radius_json_obj, "SERVICE_TYPE", *(unsigned int *)(radius_body->attribute[i].value));
+ 				cJSON_AddNumberToObject(radius_json_obj, "service_type", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	FRAMED_MTU:
-  				cJSON_AddNumberToObject(radius_json_obj, "FRAMED_MTU", *(unsigned int *)(radius_body->attribute[i].value));
+  				cJSON_AddNumberToObject(radius_json_obj, "framed_mtu", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	SESSION_TIMEOUT:
-   				cJSON_AddNumberToObject(radius_json_obj, "SESSION_TIMEOUT", *(unsigned int *)(radius_body->attribute[i].value));
+   				cJSON_AddNumberToObject(radius_json_obj, "session_timeout", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	IDLE_TIMEOUT:
-    			cJSON_AddNumberToObject(radius_json_obj, "IDLE_TIMEOUT", *(unsigned int *)(radius_body->attribute[i].value));
+				cJSON_AddNumberToObject(radius_json_obj, "idle_timeout", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	TERMINATION_ACTION:
-				cJSON_AddNumberToObject(radius_json_obj, "TERMINATION_ACTION", *(unsigned int *)(radius_body->attribute[i].value));
+				cJSON_AddNumberToObject(radius_json_obj, "termination_action", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	PROXY_STATE:
-				cJSON_AddNumberToObject(radius_json_obj, "PROXY_STATE", *(unsigned int *)(radius_body->attribute[i].value));
+				cJSON_AddNumberToObject(radius_json_obj, "proxy_state", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_STATUS_TYPE:
-				cJSON_AddNumberToObject(radius_json_obj, "ACCT_STATUS_TYPE", *(unsigned int *)(radius_body->attribute[i].value));
+				cJSON_AddNumberToObject(radius_json_obj, "acct_status_type", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_INPUT_OCTETS:
-				cJSON_AddNumberToObject(radius_json_obj, "ACCT_INPUT_OCTETS", *(unsigned int *)(radius_body->attribute[i].value));
+				cJSON_AddNumberToObject(radius_json_obj, "acct_input_octets", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_OUTPUT_OCTETS:
- 				cJSON_AddNumberToObject(radius_json_obj, "ACCT_OUTPUT_OCTETS", *(unsigned int *)(radius_body->attribute[i].value));
+ 				cJSON_AddNumberToObject(radius_json_obj, "acct_output_octets", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_INPUT_PACKETS:
-  				cJSON_AddNumberToObject(radius_json_obj, "ACCT_INPUT_PACKETS", *(unsigned int *)(radius_body->attribute[i].value));
+  				cJSON_AddNumberToObject(radius_json_obj, "acct_input_packets", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_OUTPUT_PACKETS:
-   				cJSON_AddNumberToObject(radius_json_obj, "ACCT_OUTPUT_PACKETS", *(unsigned int *)(radius_body->attribute[i].value));
+   				cJSON_AddNumberToObject(radius_json_obj, "acct_output_packets", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_TERMINATE_CAUSE:
-    			cJSON_AddNumberToObject(radius_json_obj, "ACCT_TERMINATE_CAUSE", *(unsigned int *)(radius_body->attribute[i].value));
+				cJSON_AddNumberToObject(radius_json_obj, "acct_terminate_cause", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_LINK_COUNT:
-     			cJSON_AddNumberToObject(radius_json_obj, "ACCT_LINK_COUNT", *(unsigned int *)(radius_body->attribute[i].value));
+				cJSON_AddNumberToObject(radius_json_obj, "acct_link_count", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			case	ACCT_INTERIM_INTERVAL:
-      			cJSON_AddNumberToObject(radius_json_obj, "ACCT_INTERIM_INTERVAL", *(unsigned int *)(radius_body->attribute[i].value));
+      			cJSON_AddNumberToObject(radius_json_obj, "acct_interim_interval", *(unsigned int *)(radius_body->attribute[i].value));
 				break;
 			default:
 				continue;
@@ -219,7 +220,7 @@ char NTC_RADIUS_PLUG_ENTRY(stSessionInfo *session_info, void **pme, int thread_s
 {
 	int ret=0,status=0;
 	int config_id=0;
-	int payload_len=0;
+	int payload_len=0,index=0;
 	time_t cur_time;
 	char *payload=NULL;
 	struct vxlan_info vinfo;
@@ -228,7 +229,7 @@ char NTC_RADIUS_PLUG_ENTRY(stSessionInfo *session_info, void **pme, int thread_s
 	cJSON *radius_info_object=cJSON_CreateObject();
 	
 	streamInfo2jsonObject(radius_info_object, stream);
-	get_radius_object_element(radius_info_object, &radius_info->header, &radius_info->body);
+	get_radius_object_element(radius_info_object, &radius_info->header, &radius_info->body, &index);
 
 	cJSON_AddNumberToObject(radius_info_object, "cfg_id", config_id);
 	cJSON_AddNumberToObject(radius_info_object, "service", g_ntc_radius_plug.service_id);
@@ -268,7 +269,8 @@ char NTC_RADIUS_PLUG_ENTRY(stSessionInfo *session_info, void **pme, int thread_s
 	payload_len = strlen(payload);
 
 	
-	status = rd_kafka_produce(g_ntc_radius_plug.topic_rkt, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, payload, payload_len, NULL, 0, NULL); 		
+	status = rd_kafka_produce(g_ntc_radius_plug.topic_rkt, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, payload, payload_len,
+										radius_info->body.attribute[index].value, radius_info->body.attribute[index].len, NULL); 		
 	
 	if(status < 0)
 	{
@@ -345,7 +347,7 @@ int NTC_RADIUS_PLUG_INIT(void)
 		return -1;
 	}	
 
-	printf("INIT NTC_RADIUS_PLUG SUCCESS, VERSION: %d\n", NTC_RADIUS_PLUG_VERSION_20191022);
+	printf("INIT NTC_RADIUS_PLUG SUCCESS, VERSION: %d\n", NTC_RADIUS_PLUG_VERSION_20191108);
 	
 	return 0;
 }