path: root/src/comm_audit.cpp

/*
 * comm_audit.cpp
 *
 *  Created on: 2017-8-14
 *  Last Modified: 2017-11-24
 *  Author: zhangxi
 */

#include <fstream>
#include <iostream>
#include <string>
#include <vector>
#include <atomic>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>


#include "project_requirement.h"
#include "identify.h"
#include "cJSON.h"
#include "comm_audit.h"
#include "maxminddb.h"
//#define LOG_DEBUG 0

using namespace std;

int COMM_AUDIT_VERSION_20171124 = 0;

comm_prog_configure_t             g_comm_prog_configure;
comm_prog_runtime_parameter_t     g_comm_prog_para;
atomic_flag                       lock = ATOMIC_FLAG_INIT;

static int log_nums;
static int times = 0;
static int log = 0;
static int now = 0;
// static char FDIR[12]={0};


//获取IP归属地
int get_ip_coun(char * filename, char *ip_address, char ** name_en)
{
    MMDB_s mmdb;
    int status = MMDB_open(filename, MMDB_MODE_MMAP, &mmdb);

    if(MMDB_SUCCESS != status)
    {
        //fprintf(stderr, "\n  Can't open %s - %s\n",filename, MMDB_strerror(status));
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "ERROR", "Can't open %s - %s\n",filename, MMDB_strerror(status));
        if(MMDB_IO_ERROR == status)
        {
            //fprintf(stderr, "IO error: %s\n", strerror(errno));
            MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "ERROR","IO error: %s\n", strerror(errno));
        }
        return 1; //error
    }

    int gai_error, mmdb_error;

    MMDB_lookup_result_s result =
        MMDB_lookup_string(&mmdb, ip_address, &gai_error, &mmdb_error);

    if (0 != gai_error)
    {
        fprintf(stderr, "\n  Error from getaddrinfo for %s - %s\n\n", ip_address, gai_strerror(gai_error));
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "ERROR","Error from getaddrinfo for %s - %s", ip_address, gai_strerror(gai_error));
        return 2; //error
    }

    if (MMDB_SUCCESS != mmdb_error)
    {
        fprintf(stderr,"\n  Got an error from libmaxminddb: %s\n\n", MMDB_strerror(mmdb_error));
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "ERROR","Got an error from libmaxminddb: %s", MMDB_strerror(mmdb_error));
        return 3; //error
    }

    MMDB_entry_data_s entry_data;

    int exit_code = 0;
    if (result.found_entry)
    {

        int status = MMDB_get_value(&result.entry, &entry_data, "COUNTRY", NULL);

        if (MMDB_SUCCESS != status)
        {
            fprintf(stderr,"Got an error looking up the entry data - %s\n", MMDB_strerror(status));
            MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "ERROR","Got an error looking up the entry data - %s\n", MMDB_strerror(status));
            exit_code = 4;
            goto end;
        }

        if (entry_data.has_data)
        {
            *name_en = (char *)malloc(entry_data.data_size + 1);
            memset(*name_en, '\0', (entry_data.data_size + 1));
            memcpy(*name_en, entry_data.utf8_string, entry_data.data_size);
            //printf ("%s\n", name_en);
        }
    } 
    else
    {
        //fprintf(stderr,"\n  No entry for this IP address (%s) was found\n\n", ip_address);
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "ERROR","No entry for this IP address (%s) was found", ip_address);
        exit_code = 5;
    }

end:
    MMDB_close(&mmdb);
    //exit(exit_code);
    return exit_code;
}


//读取json格式文件
void parseJson(char * buf)
{
    if(NULL == buf)
    {
        return;
    }
    cJSON * json = cJSON_Parse(buf);
    if (!json)
    {
        printf("JSON FILE READ FAIL! PLEASE CHECK net_log_upload_pz.txt's FORMAT.\n");
        printf("USE LAST UPLOAD_PZ NOW.\n");
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "UPLOAD_PZ JSON FILE READ FAIL!ERROR NUM IS %s.", strerror(errno));
    }
    else
    {
        int size = cJSON_GetArraySize(json);
        cJSON *object;
        for(int i = 0; i < size; i++)
        {
            object = cJSON_GetArrayItem(json,i);
            int interval = cJSON_GetObjectItem(object, "interval")->valueint;
            int max_num = cJSON_GetObjectItem(object, "num")->valueint;
            if(interval != g_comm_prog_configure.write_interval || max_num != g_comm_prog_configure.write_max_num)
            {
                g_comm_prog_configure.write_interval = interval;
                g_comm_prog_configure.write_max_num = max_num;
                MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "INFO", "UPLOAD_PZ JSON FILE READ SUCC! interval is %d, num is %d.", g_comm_prog_configure.write_interval, g_comm_prog_configure.write_max_num);
            }
        }
    }

    cJSON_Delete(json);
}


void read_json_file(int num)
{

    string read_path = g_comm_prog_configure.upload_pz_path;
    if (read_path[read_path.size() - 1] != '/') 
        read_path+='/';

    fstream handle;
    string in_file_name = read_path + "net_log_upload_pz.txt";
    handle.open(in_file_name.c_str(), ios::binary|ios::in);
    if(handle.is_open())
    {
        char *buf = new char[MAX_LEN];
        handle.read(buf, MAX_LEN);
        parseJson(buf);
        handle.close();

        delete buf;
    }

}

// 定时读取json线程
void *set_timer(void * pParam)
{
    signal(SIGALRM, read_json_file);

    struct itimerval tick;
    tick.it_value.tv_sec = 1;  //1秒钟后将启动定时器
    tick.it_value.tv_usec = 0;
    tick.it_interval.tv_sec  = g_comm_prog_configure.read_interval; //定时器启动后，每隔指定秒数将执行相应的函数
    tick.it_interval.tv_usec = 0;

    //setitimer将触发SIGALRM信号
    int ret = setitimer(ITIMER_REAL, &tick, NULL);

    if ( ret != 0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "SET TIMER ERROR! ERROR NUM IS %s.", strerror(errno));
        return NULL;
    }

    while(1){sleep(1);}

}

// 写数据
int send_data(htable_data_t* send_data, FILE *fp)
{
    log = 1;
    long long EndTime;
    EndTime = time(NULL);

    char stime[24];
    struct tm stm;
    time_t sTime = (time_t)send_data->start_time;
    localtime_r(&sTime, &stm);
    sprintf(stime, "%d-%02d-%02d %d:%d:%d", stm.tm_year+1900, stm.tm_mon+1, stm.tm_mday, stm.tm_hour, stm.tm_min, stm.tm_sec);

    char etime[24];
    struct tm ptm;
    time_t eTime;
    if(send_data->end_time == 0)
        eTime = (time_t)EndTime;
    else
        eTime = (time_t)send_data->end_time;
    localtime_r(&eTime, &ptm);
    sprintf(etime, "%d-%02d-%02d %d:%d:%d", ptm.tm_year+1900, ptm.tm_mon+1, ptm.tm_mday, ptm.tm_hour, ptm.tm_min, ptm.tm_sec);
    // memcpy(stat_info->time_data.end_time, etime, 24);

    char sendInnerIP[IPLEN]={0};
    inet_ntop(AF_INET, &send_data->inner_ip, sendInnerIP ,IPLEN);
    // sprintf(sendInnerIP, "%s", inet_ntop(AF_INET, &send_data->inner_ip, sendInnerIP ,IPLEN));
    char sendExtIP[IPLEN]={0};
    inet_ntop(AF_INET, &send_data->ext_ip, sendExtIP ,IPLEN);
    // sprintf(sendExtIP, "%s", inet_ntop(AF_INET, &send_data->ext_ip, sendExtIP ,IPLEN));

    string app = APP_TYPE(send_data->app);
    string protocol = STREAM_TYPE(send_data->protocol);
    string tcp_flag = TCP_FLAG(send_data->tcp_flag);

    fprintf(fp, "%s#%u#%s#%s#%u#%s#%s#%s#%s#%llu#%llu#%llu#%llu#%s#%s\n",
            sendInnerIP, 
            ntohs(send_data->inner_port), 
            send_data->inner_mac,
            sendExtIP, 
            ntohs(send_data->ext_port), 
            send_data->ext_mac,
            protocol.c_str(),
            app.c_str(),
            tcp_flag.c_str(),
            send_data->in_bytes,
            send_data->out_bytes,
            send_data->in_pkts,
            send_data->out_pkts,
            stime,
            etime
           );
    // log_nums条记录写入一个txt
    if(log_nums < g_comm_prog_configure.write_max_num-1)
        log_nums++;
    else
    {
        log_nums = 0;
        times++;
    }

    return 0;
}

// 遍历哈希表写文件
void iterate_cb(const uchar * key, uint size, void * data, void *arg)
{
    htable_data_t* info = (htable_data_t*)data;
    char * writetime = ((del_para_t*)arg)->writetime;
    int thread_seq = ((del_para_t*)arg)->thread_seq;
    char writefile[50];
    writefile[0]='\0';

    string tmp_log_path = g_comm_prog_configure.tmp_log_path;
    if (tmp_log_path[tmp_log_path.size() - 1] != '/') 
        tmp_log_path+='/';
#ifdef LOG_DEBUG
    cout<<"tmp_log_path:"<<tmp_log_path<<endl;
#endif

    sprintf(writefile, "%s%s-%d.txt", tmp_log_path.c_str(), writetime, times);
#ifdef LOG_DEBUG
    cout<<"writefile:"<<writefile<<endl;
#endif

    FILE *fp = fopen(writefile, "a+");
    if(fp != NULL)
    {
        send_data(info, fp);
    }
    else
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "FOPEN ERROR! ERROR NUM IS %s.", strerror(errno));
    }
    fclose(fp);
    fp = NULL;


    if(MESA_htable_del(g_comm_prog_para.htable_handle[thread_seq][1], key, size, NULL) < 0)
    {
        perror("MESA_htable_del()"); 
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "MESA_htable_del error!");
    }
}

// 写日志线程
void * thread_time(void * pParam)
{
    while(1)
    {

        int last;
        if(g_comm_prog_configure.write_interval <= 0)
        {
            g_comm_prog_configure.write_interval = 1;
            MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_INFO, "INFO", "write_interval arg is not more than 0! reset interval to 1.");
        }

        time_t sTime = (time_t)g_comm_prog_para.last_send_time;
        struct tm stm;
        localtime_r(&sTime, &stm);
        last = stm.tm_hour * 10000 + stm.tm_min * 100 + stm.tm_sec;

        char FNAME[10];
        char TXT[8];
        sprintf(FNAME, "%d-%d-%d", stm.tm_hour, stm.tm_min, stm.tm_sec);
        //sprintf(TXT, "%d-%d-%d", stm.tm_hour, stm.tm_min, stm.tm_sec);
        sprintf(TXT, "%d%02d%02d", stm.tm_year+1900, stm.tm_mon+1, stm.tm_mday);

        // times = 1;
        log_nums = 0;
        if(g_comm_prog_configure.write_max_num <= 0)
        {
            g_comm_prog_configure.write_max_num = 1000;
            MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_INFO, "INFO", "write_interval arg is not more than 0! reset interval to 1000.");
        }

        sleep(g_comm_prog_configure.sleep);

        // pthread_mutex_lock(&g_comm_prog_para.mutex);
        while (lock.test_and_set(std::memory_order_acquire)); // spin
        for(int i = 0; i < g_comm_prog_para.thread_num; i++)
        {
            MESA_htable_handle tmp = g_comm_prog_para.htable_handle[i][0];
            g_comm_prog_para.htable_handle[i][0] = g_comm_prog_para.htable_handle[i][1];
            g_comm_prog_para.htable_handle[i][1] = tmp;
        }
        lock.clear(std::memory_order_release);               // release lock
        // pthread_mutex_unlock(&g_comm_prog_para.mutex);

        for(int i = 0; i < g_comm_prog_para.thread_num; i++)
        {
            del_para_t arg;
            arg.writetime = TXT;
            arg.thread_seq = i;
            MESA_htable_iterate(g_comm_prog_para.htable_handle[i][1], iterate_cb, (void *)&arg);
        }

        sTime = (time_t)g_comm_prog_para.last_send_time;
        localtime_r(&sTime, &stm);
        last = stm.tm_hour * 10000 + stm.tm_min * 100 + stm.tm_sec;

        long long NowTime = time(NULL);
        time_t eTime = (time_t)NowTime;
        struct tm ptm;
        localtime_r(&eTime, &ptm);
        now = ptm.tm_hour * 10000 + ptm.tm_min * 100 + ptm.tm_sec;

        if(log == 1)
        {
            if(now - last >= g_comm_prog_configure.write_interval * 100)
            {
                times++;
                g_comm_prog_para.last_send_time = time(NULL);
            }
            else if(now < last)
            {
                times = 1;
                g_comm_prog_para.last_send_time = time(NULL);
            }
        }   
    }
    return NULL;
}


static long search_cb(void* data, const uchar* key, uint size, void* arg)
{
    if(data == NULL)
        return -1;

    struct streaminfo * a_stream = (struct streaminfo *)((htable_search_arg *)arg)->stream;
    void * a_packet = (void *)((htable_search_arg *)arg)->packet;
    int is_end = ((htable_search_arg *)arg)->is_end;
    htable_data_t * info = (htable_data_t *)data;

    if(a_stream->type == STREAM_TYPE_TCP)
    {
        struct tcpdetail* a_tcp = (struct tcpdetail*)(a_stream->ptcpdetail);
        if(a_stream->curdir == DIR_C2S)//C2S:INNER TO EXTERN
        {
            info->out_bytes += a_tcp->datalen;
            info->out_pkts++;
        }
        else
        {
            info->in_bytes += a_tcp->datalen;
            info->in_pkts++;
        }
        // struct tcpdetail* a_tcp = (struct tcpdetail*)(a_stream->ptcpdetail);
        // info->tcp_flag = info->tcp_flag & a_stream->stream_state;
        if(a_packet != NULL)
        {
            struct tcphdr* tcp_header = (struct tcphdr*)(a_packet + sizeof(struct iphdr));
            info->tcp_flag = 32*tcp_header->urg + 16*tcp_header->ack + 8*tcp_header->psh + 4*tcp_header->rst + \
                             2*tcp_header->syn + tcp_header->fin;
        }
    }
    else
    {
        struct udpdetail* a_udp = (struct udpdetail*)(a_stream->pudpdetail);
        if(a_stream->curdir == DIR_C2S)//C2S:INNER TO EXTERN
        {
            info->out_bytes += a_udp->datalen;
            info->out_pkts++;
        }
        else
        {
            info->in_bytes += a_udp->datalen;
            info->in_pkts++;
        }
        info->tcp_flag = 0;
    }
    if(is_end)
        info->end_time = time(NULL);

    return 0;
}


int judge_ip_location(unsigned int inner_ip, char ** inner_name_en, unsigned int ext_ip, char ** ext_name_en)
{
    char innerIP[IPLEN] = {0};
    char extIP[IPLEN] = {0};
    int res1 = 0;
    int res2 = 0;
    //char * inner_name_en = NULL;
    //char * ext_name_en = NULL;
    inet_ntop(AF_INET, &inner_ip, innerIP ,IPLEN);
    inet_ntop(AF_INET, &ext_ip, extIP ,IPLEN);
    res1 = get_ip_coun(g_comm_prog_configure.mmdb_path, innerIP, inner_name_en);
    res2 = get_ip_coun(g_comm_prog_configure.mmdb_path, extIP, ext_name_en);
    if(0 != res1 && 0 != res2)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_INFO, "JUDGE_IP", "judge_innerip_location INNER[%s] EXT[%s] error!", innerIP, extIP);
        return 0;    
    }
    if ((NULL!=*inner_name_en && 0!=strncmp(*inner_name_en, "China", 5)) || (NULL!=*ext_name_en && 0 != strncmp(*ext_name_en, "China", 5))) //有境外IP
    {
        return 1;
    }
    else //全部都是境内IP
    {
        return 2;
    }
}


void time_mode_op(struct streaminfo* a_stream, int thread_seq, void * a_packet, int state, int is_end)
{
    stDntyProInfo * pro_info = (stDntyProInfo *)project_req_get_struct(a_stream,g_comm_prog_para.project_id_pro);
    if(pro_info == NULL)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "project_req_get_struct is NULL. LACK OF IDENTIFY'S CONF!");
        return ;
    }

    long ret;
    int judge_res = 0;

    htable_search_arg arg;
    arg.stream = a_stream;
    arg.packet = a_packet;

    // pthread_mutex_lock(&g_comm_prog_para.mutex);
    if(NULL == MESA_htable_search_cb(g_comm_prog_para.htable_handle[thread_seq][0], (unsigned char*)&a_stream->addr, sizeof(a_stream->addr), search_cb, (void *)&arg, &ret))
    {
        htable_data_t * value = (htable_data_t*)calloc(1, sizeof(htable_data_t));
        if(a_stream->addr.addrtype == ADDR_TYPE_IPV4)
        {
            value->inner_ip = a_stream->addr.tuple4_v4->saddr;
            value->inner_port = a_stream->addr.tuple4_v4->source;
            value->ext_ip = a_stream->addr.tuple4_v4->daddr;
            value->ext_port = a_stream->addr.tuple4_v4->dest;
            if(g_comm_prog_configure.flag == 0) //flag == 0, 只记录包含境外IP的通联数据
            {
                char * inner_name_en = NULL;
                char * ext_name_en = NULL;
                judge_res = judge_ip_location(value->inner_ip, &inner_name_en, value->ext_ip, &ext_name_en);
                if(judge_res == 0)  //查找IP归属失败，仍然记录
                {
                    MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "find IP[%u or %u] location error!.", value->inner_ip, value->ext_ip);
                }
                else if(judge_res == 1) //有境外IP，需要记录
                {
                    MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "find IP1[%u:%s]  IP2[%u:%s].", value->inner_ip,inner_name_en, value->ext_ip, ext_name_en);
                }
                else if(judge_res == 2) //全都是境内IP，不做记录，直接返回
                {
					if(NULL != inner_name_en)
					{
						free(inner_name_en);
						inner_name_en = NULL;
					}
					if(NULL != ext_name_en)
					{
						free(ext_name_en);
						ext_name_en = NULL;
					}
					free(value);
					value = NULL;
                    return ;
                }      
		if(NULL != inner_name_en)
		{
		    free(inner_name_en);
		    inner_name_en = NULL;
		}
		if(NULL != ext_name_en)
		{
		    free(ext_name_en);
		    ext_name_en = NULL;
		}
            }
        }
        if(a_stream->type == STREAM_TYPE_TCP)
        {
            struct tcpdetail* a_tcp = (struct tcpdetail*)(a_stream->ptcpdetail);
            if(a_stream->routedir == DIR_C2S)//C2S:INNER TO EXTERN
            {
                value->in_bytes = 0;
                value->in_pkts = 0;
                value->out_bytes = a_tcp->datalen;
                value->out_pkts = 1;
            }
            else
            {
                value->in_bytes = a_tcp->datalen;
                value->in_pkts = 1;
                value->out_bytes = 0;
                value->out_pkts = 0;
            }
            if(a_packet != NULL)
            {
                struct tcphdr* tcp_header = (struct tcphdr*)(a_packet + sizeof(struct iphdr));
                value->tcp_flag = 32*tcp_header->urg + 16*tcp_header->ack + 8*tcp_header->psh + 4*tcp_header->rst + 2*tcp_header->syn + tcp_header->fin;
            }
        }
        else
        {
            struct udpdetail* a_udp = (struct udpdetail*)(a_stream->pudpdetail);
            if(a_stream->routedir == DIR_C2S)//C2S:INNER TO EXTERN
            {
                value->in_bytes = 0;
                value->in_pkts = 0;
                value->out_bytes = a_udp->datalen;
                value->out_pkts = 1;
            }
            else
            {
                value->in_bytes = a_udp->datalen;
                value->in_pkts = 1;
                value->out_bytes = 0;
                value->out_pkts = 0;
            }
            value->tcp_flag = 0;
        }

        value->protocol = a_stream->type;
        value->app = pro_info->protocol_type;

        struct streaminfo *father = a_stream->pfather;
        while(father)
        {
            if(father->addr.addrtype != ADDR_TYPE_MAC)   
            {      
                father = father->pfather;      
                continue;   
            }   
            else   
            {      
                memset(value->inner_mac, 0, 60);      
                sprintf(value->inner_mac, "%02x-%02x-%02x-%02x-%02x-%02x", father->addr.mac->src_mac[0], father->addr.mac->src_mac[1],\
                        father->addr.mac->src_mac[2], father->addr.mac->src_mac[3], father->addr.mac->src_mac[4], father->addr.mac->src_mac[5]);      
                memset(value->ext_mac, 0, 60);      
                sprintf(value->ext_mac, "%02x-%02x-%02x-%02x-%02x-%02x", father->addr.mac->dst_mac[0], father->addr.mac->dst_mac[1],\
                        father->addr.mac->dst_mac[2], father->addr.mac->dst_mac[3], father->addr.mac->dst_mac[4], father->addr.mac->dst_mac[5]);      
                break;   
            }
        }

        value->start_time = time(NULL);
        value->end_time = 0;

        if(MESA_htable_add(g_comm_prog_para.htable_handle[thread_seq][0], (const unsigned char *)&a_stream->addr, \
                    sizeof(a_stream->addr), (void *)value) < 0)
        {
            MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "htable_add error.ERROR IS %s.", strerror(errno));
            free(value);
            value = NULL;
        }
    }
    // pthread_mutex_unlock(&g_comm_prog_para.mutex);
    // free(arg);
    // arg = NULL;
}


char op_state_data(struct streaminfo *a_stream,  void **pme, int thread_seq, void *a_packet, int state, int is_end)
{
    // struct tcpdetail *pdetail = (struct tcpdetail*)a_stream->ptcpdetail;
    // comm_audit_t* stat_info = (comm_audit_t*)(*pme);
    // assert(stat_info != NULL);

    if(a_packet == NULL && state != OP_STATE_CLOSE)
        return APP_STATE_GIVEME;

    time_mode_op(a_stream, thread_seq, a_packet, state, is_end);

    return APP_STATE_GIVEME;
}

// char op_state_pending(struct streaminfo *a_stream,  void **pme, int thread_seq,void *a_packet, int state, int is_end)
// {
//     // init_pme_info(pme);

//     op_state_data(a_stream, pme, thread_seq, a_packet, state, is_end);

//     return APP_STATE_GIVEME;
// }


char op_state_close(struct streaminfo *a_stream,  void **pme, int thread_seq,void *a_packet, int state, int is_end)
{
    op_state_data(a_stream, pme, thread_seq, a_packet, state, is_end);

    // free_pme_info(pme);

    return APP_STATE_DROPME;
}


char COMM_AUDIT_ENTRY(struct streaminfo *a_stream, void **pme, int thread_seq, void *a_packet)
{
    if(g_comm_prog_configure.flag == 2) //flag == 2,不记录任何通联数据，直接返回
    {
        return APP_STATE_DROPME;
    }
    char ret = APP_STATE_GIVEME;
    int state;
    if(a_stream->type == STREAM_TYPE_TCP)
        state = a_stream->pktstate;
    else
        state = a_stream->opstate;
    int is_end = 0;
    switch(state)
    {
        case OP_STATE_PENDING:
        case OP_STATE_DATA:
            ret = op_state_data(a_stream, pme, thread_seq, a_packet, state, is_end);
            break;

        case OP_STATE_CLOSE:
            is_end=1;
            ret = op_state_close(a_stream, pme, thread_seq, a_packet, state, is_end);
            break;

        default:
            ret = APP_STATE_DROPME;
            break;
    }
    return ret;
}

void htable_del1(const uchar * key, uint size, void * data, void * arg)
{
    int thread_seq = *(int*)arg;
    if(MESA_htable_del(g_comm_prog_para.htable_handle[thread_seq][0], key, size, NULL) < 0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "MESA_htable_del error!");
    }
}

void htable_del2(const uchar * key, uint size, void * data, void * arg)
{
    int thread_seq = *(int*)arg;
    if(MESA_htable_del(g_comm_prog_para.htable_handle[thread_seq][1], key, size, NULL) < 0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "MESA_htable_del error!");
    }
}

void COMM_AUDIT_DESTROY()
{

    // free(g_comm_prog_configure.upload_log_path);
    free(g_comm_prog_configure.tmp_log_path);
    free(g_comm_prog_configure.upload_pz_path);

    for(int i = 0; i < g_comm_prog_para.thread_num; i++)
    {
        if(MESA_htable_iterate(g_comm_prog_para.htable_handle[i][0], htable_del1, (void*)&i) < 0)
            MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "COMM_AUDIT_DESTROY. MESA_htable_iterate error!");
        // printf("iterate error!\n");
        if(MESA_htable_destroy(g_comm_prog_para.htable_handle[i][0], NULL) < 0)
            MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "COMM_AUDIT_DESTROY. MESA_htable_destroy error!");
        // printf("destroy error!\n");

        if(MESA_htable_iterate(g_comm_prog_para.htable_handle[i][1], htable_del2, (void*)&i) < 0)
            MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "COMM_AUDIT_DESTROY. MESA_htable_iterate error!");
        // printf("iterate error!\n");
        if(MESA_htable_destroy(g_comm_prog_para.htable_handle[i][1], NULL) < 0)
            MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "COMM_AUDIT_DESTROY. MESA_htable_destroy error!");
        // printf("destroy error!\n");
    }
    for(int i = 0; i < g_comm_prog_para.thread_num; i++)
        free(g_comm_prog_para.htable_handle[i]);
    free(g_comm_prog_para.htable_handle);

    pthread_exit(NULL);
    //printf("destroy COMM_AUDIT_DESTROY success!\n");

}

void htable_free(void * data)
{
    htable_data_t * info = (htable_data_t *)data;
    free(info);
    data = NULL;
}

int init_hash_table()
{
    MESA_htable_handle ** handle = (MESA_htable_handle**)malloc(g_comm_prog_para.thread_num*sizeof(MESA_htable_handle*));
    for(int i = 0; i < g_comm_prog_para.thread_num; i++)
        handle[i] = (MESA_htable_handle*)malloc(2 * sizeof(MESA_htable_handle));
    for(int i = 0; i < g_comm_prog_para.thread_num; i++)
    {
        MESA_htable_create_args_t htable_args1;
        memset(&htable_args1, 0, sizeof(htable_args1));
        htable_args1.recursive = 1;
        htable_args1.thread_safe = 0;
        htable_args1.hash_slot_size = 4096*4096;
        htable_args1.max_elem_num = g_comm_prog_configure.htable_size;
        htable_args1.data_free = htable_free;
        htable_args1.data_expire_with_condition = NULL;
        htable_args1.eliminate_type = HASH_ELIMINATE_ALGO_LRU;
        htable_args1.expire_time = 0;
        handle[i][0] = MESA_htable_create(&htable_args1, sizeof(htable_args1));
        if(handle[i][0] == NULL)
            return -1;

        MESA_htable_create_args_t htable_args2;
        memset(&htable_args2, 0, sizeof(htable_args2));
        htable_args2.recursive = 1;
        htable_args2.thread_safe = 0;
        htable_args2.hash_slot_size = 4096*4096;
        htable_args2.max_elem_num = g_comm_prog_configure.htable_size;
        htable_args2.data_free = htable_free;
        htable_args2.data_expire_with_condition = NULL;
        htable_args2.eliminate_type = HASH_ELIMINATE_ALGO_LRU;
        htable_args2.expire_time = 0;
        handle[i][1] = MESA_htable_create(&htable_args2, sizeof(htable_args2));
        if(handle[i][1] == NULL)
            return -1;
    }

    g_comm_prog_para.htable_handle = handle;
    return 0;
}

int comm_read_conf(const char *file_name)
{   
    short   log_level = 30;
    char    log_filename[256];
    log_filename[0]='\0';
    int ret;

    if(MESA_load_profile_short_nodef(file_name, "DEFAULT", "log_level", (short*)&log_level)<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [DEFAULT] log_level error!");
        return -1;
    } 
    if(MESA_load_profile_string_nodef(file_name, "DEFAULT", "log_path", log_filename, sizeof(log_filename))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [DEFAULT] log_path error!");
        return -1;
    }
    g_comm_prog_para.log_handle = MESA_create_runtime_log_handle(log_filename, log_level);
    if(g_comm_prog_para.log_handle == NULL)
    {       
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "Create log_handle error!");
        return -1;
    }

    if((ret=MESA_load_profile_int_nodef(file_name, "DEFAULT", "htable_size", (int*)&g_comm_prog_configure.htable_size))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [DEFAULT] htable_size error!");
        return -1;
    }
    if((ret=MESA_load_profile_int_nodef(file_name, "DEFAULT", "sleep", (int*)&g_comm_prog_configure.sleep))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [DEFAULT] sleep error!");
        return -1;
    }
    if(g_comm_prog_configure.sleep <= 0)
    {
        g_comm_prog_configure.sleep = 10;
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_INFO, "comm_read_conf",\
                "conf [DEFAULT] sleep is not more than 0! Reset interval to 10.");
    }


    // g_comm_prog_configure.upload_log_path = (char *)calloc(1, 256*sizeof(char));
    // if((ret=MESA_load_profile_string_nodef(file_name, "UPLOAD_LOG", "upload_log_path", g_comm_prog_configure.upload_log_path, 256*sizeof(char)))<0)
    // {
    //     MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [UPLOAD_LOG] upload_log_path error!");
    //     return -1;
    // }
    // char upload_log_path[50]={0};
    // sprintf(upload_log_path, "mkdir -p %s", g_comm_prog_configure.upload_log_path);
    // FILE * pret = popen(upload_log_path, "w");
    // if(!pret)
    //     MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "%s IS FAILED! ERROR IS %s",upload_log_path, strerror(errno));
    // pclose(pret);

    g_comm_prog_configure.tmp_log_path = (char *)calloc(1, 256*sizeof(char));
    if((ret=MESA_load_profile_string_nodef(file_name, "UPLOAD_LOG", "tmp_log_path", g_comm_prog_configure.tmp_log_path, 256*sizeof(char)))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [UPLOAD_LOG] tmp_log_path error!");
        return -1;
    }
    char tmp_log_path[50]={0};
    sprintf(tmp_log_path, "mkdir -p %s", g_comm_prog_configure.tmp_log_path);
    FILE * pret = popen(tmp_log_path, "w");
    if(!pret)
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "%s IS FAILED! ERROR IS %s",tmp_log_path, strerror(errno));
    pclose(pret);


    g_comm_prog_configure.upload_pz_path = (char *)calloc(1, 256*sizeof(char));
    if((ret=MESA_load_profile_string_nodef(file_name, "UPLOAD_PZ", "upload_pz_path", g_comm_prog_configure.upload_pz_path,256*sizeof(char)))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [UPLOAD_PZ] upload_pz_path error!");
        return -1;
    }

    if((ret=MESA_load_profile_int_nodef(file_name,"UPLOAD_PZ","read_upload_pz_interval",&g_comm_prog_configure.read_interval))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [UPLOAD_PZ] read_upload_pz_interval error!");
        return -1;
    }

    if((ret=MESA_load_profile_int_nodef(file_name,"UPLOAD_PZ","default_upload_interval",&g_comm_prog_configure.write_interval))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [UPLOAD_PZ] default_upload_interval error!");
        return -1;
    }

    if((ret=MESA_load_profile_int_nodef(file_name,"UPLOAD_PZ","defailt_upload_max_num",&g_comm_prog_configure.write_max_num))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [UPLOAD_PZ] defailt_upload_max_num error!");
        return -1;
    }

    if((ret=MESA_load_profile_int_nodef(file_name,"IPDB_PZ","flag",&g_comm_prog_configure.flag))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [IPDB_PZ] flag error!");
        return -1;
    }

    g_comm_prog_configure.mmdb_path = (char *)calloc(1, 256*sizeof(char));
    if((ret=MESA_load_profile_string_nodef(file_name, "IPDB_PZ", "mmdb_path", g_comm_prog_configure.mmdb_path, 256*sizeof(char)))<0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "comm_read_conf", "conf [IPDB_PZ] mmdb_path error!");
        return -1;
    }
    return 0;
}

int COMM_AUDIT_INIT()
{
    int ret;    
    if(-1 == comm_read_conf("./djconf/comm_audit.conf"))    
        return -1;
    MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_INFO, "COMM_AUDIT_INIT", "COMM_AUDIT_INIT conf read over!");

    g_comm_prog_para.project_id_pro = project_customer_register(DNTY_PPROJECT_PRO, DNTY_PPROJECT_TYPE);
    if(g_comm_prog_para.project_id_pro < 0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "COMM_AUDIT_INIT. project_req_register error. LACK OF IDENTIFY'S CONF!");
        return -1;
    }


    int thread_num = get_thread_count();
    g_comm_prog_para.thread_num = thread_num;
#ifdef LOG_DEBUG
    printf("thread_num:%d\n", g_comm_prog_para.thread_num);
#endif
    ret = init_hash_table();
    if(ret < 0)
    {
        MESA_handle_runtime_log(g_comm_prog_para.log_handle, RLOG_LV_FATAL, "FATAL", "COMM_AUDIT_INIT. init_hash_table error!");
        //printf("init_hash_table() error!\n");
        return -1;
    }

    g_comm_prog_para.last_send_time = time(NULL);
    // pthread_mutex_init(&g_comm_prog_para.mutex, NULL);


    //创建写文件线程
    pthread_t hd_write;
    pthread_create(&hd_write, NULL, thread_time, NULL);

    //创建读json配置线程
    pthread_t hd_read;
    pthread_create(&hd_read, NULL, set_timer, NULL);

    // printf("init COMM_AUDIT_INIT success!\n");

    return 0;
}