gfwleak/tango-tsg-master
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Squashed commit of the following:

Browse Source 
commit 91574444c3
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Tue Dec 3 11:44:16 2019 +0800

    公共字段中的通信市时长单位是：ms；转换单位
    KNI发日志需等待TFE结束，需提前构造streaminfo日志字段，调整包数字节数统计位置
    细化调整日志

commit ff053e3e2c
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Wed Nov 27 11:38:35 2019 +0800

    增加调试日志信息

commit 2d1a530165
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Wed Nov 27 11:27:06 2019 +0800

    使用前置声明结构体，修改接口函数原型

commit 27f0cfc91a
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Wed Nov 27 10:33:48 2019 +0800

    提供TLD_append_streaminfo函数接口供KNI使用
    修改配置文件

commit ed0b17a49d
Merge: 132d807 04963fe
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Tue Nov 26 17:44:28 2019 +0800

    Merge branch 'feature-kni' into 'develop'

    增加http解析部分

    See merge request tango/tsg_master!3

commit 04963fe1bf
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Tue Nov 26 17:41:43 2019 +0800

    增加http解析部分

commit 132d807af1
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Tue Nov 26 16:58:54 2019 +0800

    调整调试日志内容

commit 8cfc6b32fd
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Mon Nov 25 19:30:08 2019 +0800

    增加调试日志信息

commit 4aef0628e5
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Mon Nov 25 18:36:48 2019 +0800

    增加调试信息

commit 7c9dabdab5
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Mon Nov 25 10:51:37 2019 +0800

    矫正tableinfo配置项

commit 34ec2059ea
Merge: 27cb2ad 2b96c05
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Mon Nov 25 10:00:18 2019 +0800

    Merge branch 'develop' of https://git.mesalab.cn/tango/tsg_master into develop

commit 27cb2ad3b2
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Thu Nov 21 19:21:08 2019 +0800

    获取包数、字节数与sapp头文件不一致，导致统计信息颠倒
    删除从vxlan里获取信息

commit 2b96c050d5
Merge: 0a11db6 8d954e4
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Thu Nov 21 10:22:42 2019 +0800

    Merge branch 'feature-kni' into 'develop'

    修改proto初始值为PROTO_UNKONWN

    See merge request tango/tsg_master!2

commit 46a233b782
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Thu Nov 21 10:18:08 2019 +0800

    KNI发送日志的线程不是sapp申请的，所以不能使用dictator_malloc申请内存
    增加user_region日志字段

commit 8d954e4cc5
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Wed Nov 20 19:55:59 2019 +0800

    修改proto初始值为PROTO_UNKONWN

commit 0a11db6d2e
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Wed Nov 20 18:02:54 2019 +0800

    增加总控配置文件

commit 5c04ba23aa
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Wed Nov 20 18:00:35 2019 +0800

    SSL_SNI和HTTP_HOST的deny动作由总控负责处理
    修复获取全局流ID时的错误

commit 291c05ea79
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Wed Nov 20 16:40:19 2019 +0800

    增加获取全局流ID的接口
    调整总控注册为TCP入口

commit 28da97b53b
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Wed Nov 20 14:45:41 2019 +0800

    支持平台提供的APP_STATE_KILL_OTHER
    修复BUG，天剑流标签时未对协议进行赋值，导致tfe颁发证书失败

commit 7912ec54b2
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Tue Nov 19 18:41:31 2019 +0800

    修复BUG，命中拦截策略时未对类型进行赋值，导致KNI获取不到命中结果
    添加总控配置文件
    添加FS2统计日志
    添加错误日志
    构造适用sapp4.0的rpm

commit 484feaed94
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Tue Nov 19 13:39:37 2019 +0800

    增加获取deny结果规则的接口
    修复自测试过程中遇到的BUG

commit 5165173025
Merge: 186e591 6b6cbef
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Mon Nov 18 15:46:03 2019 +0800

    Merge branch 'feature-fast-path' into feature-kni

commit 6b6cbeffa1
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Mon Nov 18 14:50:32 2019 +0800

    修复初始化错误

commit e7532d6280
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Mon Nov 18 13:33:29 2019 +0800

    添加遗漏头文件tsg_entry.h

commit f33d11c50b
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Mon Nov 18 13:29:28 2019 +0800

    修复编译错误

commit 88d9368cee
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Mon Nov 18 13:25:38 2019 +0800

    合并头文件，删除无用头文件

commit 382ca0a793
Author: 刘学利 <liuxueli@iie.ac.cn>
Date:   Mon Nov 18 13:21:25 2019 +0800

    Delete tsg_types.h

commit ab58f918e5
Author: 刘学利 <liuxueli@iie.ac.cn>
Date:   Mon Nov 18 13:21:05 2019 +0800

    Delete tsg_log_id.h

commit 3b13075701
Author: 刘学利 <liuxueli@iie.ac.cn>
Date:   Mon Nov 18 13:20:57 2019 +0800

    Delete tsg_entry.h

commit 186e591dbf
Merge: 39f82ed 91c511c
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Fri Nov 15 21:22:44 2019 +0800

    Merge branch 'master' into feature-kni

commit 39f82ed3f7
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Fri Nov 15 21:22:39 2019 +0800

    接口适配

commit 8913a7d22e
Author: liuxueli <liuxueli@iie.ac.cn>
Date:   Fri Nov 15 19:29:54 2019 +0800

    规则排序功能
    统一发送日志功能

commit 1cb2358fea
Merge: c76bc55 bd65acc
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Thu Nov 14 15:06:35 2019 +0800

    合并master修改

commit c76bc5534d
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Wed Nov 13 19:56:19 2019 +0800

    接口增加protocol参数

commit db7282dab7
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Wed Nov 13 19:24:22 2019 +0800

    扫描接口增加domain

commit 56fadd73b2
Merge: f9aee05 9e738ae
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Wed Nov 13 19:08:55 2019 +0800

    Merge branch 'master' into feature-kni

commit f9aee05bcc
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Wed Nov 13 16:17:59 2019 +0800

    修改ssl_utils文件名

commit 452a08790c
Merge: 3abe8a2 c1ffc53
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Wed Nov 13 16:12:14 2019 +0800

    Merge branch 'master' into feature-kni

commit 3abe8a2fc5
Author: 崔一鸣 <cuiyiming@iie.ac.cn>
Date:   Wed Nov 13 16:11:38 2019 +0800

    添加kni模块扫描部分
This commit is contained in:
杨威
2019-12-09 18:58:05 +08:00
parent 91c511c3ed
commit 64c1dcb159
22 changed files with 2184 additions and 357 deletions
Download Patch File Download Diff File Expand all files Collapse all files

531
src/tsg_send_log.cpp
View File

@@ -2,6 +2,7 @@
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include <time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
@@ -11,100 +12,502 @@
#include <unistd.h>
#include <MESA/stream.h>
#include <MESA/cabot_send_log.h>
#include <MESA/MESA_prof_load.h>
#include <MESA/MESA_handle_logger.h>
#include "tsg_types.h"
#include "tsg_entry.h"
#include "tsg_send_log.h"
#include "tsg_send_log_internal.h"
tsg_logger_info_t tsg_logger_info;
char TSG_SEND_LOG_VERSION_20191129=0;
struct tsg_log_instance_t *g_tsg_log_instance;
static unsigned int get_ip_by_eth_name(const char *ifname)
const id2field_t tld_type[TLD_TYPE_MAX]={{TLD_TYPE_UNKNOWN, TLD_TYPE_UNKNOWN, "UNKOWN"},
{TLD_TYPE_LONG, TLD_TYPE_LONG, "LONG"},
{TLD_TYPE_STRING, TLD_TYPE_STRING, "STRING"},
{TLD_TYPE_FILE, TLD_TYPE_FILE, "FILE"}
};
extern "C" int MESA_get_dev_ipv4(const char *device, int *ip_add);
unsigned long long tsg_get_stream_id(struct streaminfo * a_stream)
{
int sockfd;
struct ifreq ifr;
unsigned int ip;
int ret=0;
int device_id_size=sizeof(unsigned long long);
unsigned long long device_id=(unsigned long long)g_tsg_para.device_id;
sockfd = socket(AF_INET, SOCK_DGRAM, 0);
if (-1 == sockfd) {
goto error;
}
ret=MESA_get_stream_opt(a_stream, MSO_GLOBAL_STREAM_ID, (void *)&device_id, &device_id_size);
if(ret==0)
{
return device_id;
}
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;
error:
close(sockfd);
return INADDR_NONE;
return -1;
}
int steaminfo2opt()
int TLD_cancel(struct TLD_handle_t *handle)
{
struct TLD_handle_t *_handle=handle;
if(_handle!=NULL)
{
if(_handle->object!=NULL)
{
cJSON_Delete(_handle->object);
_handle->object=NULL;
}
free(handle);
handle=NULL;
}
return 0;
}
int tsg_sendlog_init(char *filename)
int TLD_delete(struct TLD_handle_t *handle, char *key)
{
int ret=0;
unsigned int local_ip_nr=0;
char nic_name[32];
struct TLD_handle_t *_handle=handle;
memset(&tsg_logger_info, 0, sizeof(tsg_logger_info));
MESA_load_profile_int_def(filename, "TSG_LOG", "MODE",&(tsg_logger_info.mode), 0);
MESA_load_profile_string_def(filename, "TSG_LOG", "FIELD_FILE", tsg_logger_info.field_file, sizeof(tsg_logger_info.field_file), NULL);
MESA_load_profile_string_def(filename, "TSG_LOG", "BROKER_LIST", tsg_logger_info.broker_list, sizeof(tsg_logger_info.broker_list), NULL);
MESA_load_profile_int_def(filename, "TSG_LOG", "LEVEL",&(tsg_logger_info.level), 30);
MESA_load_profile_string_def(filename, "TSG_LOG", "LOG_PATH", tsg_logger_info.log_path, sizeof(tsg_logger_info.log_path), NULL);
tsg_logger_info.logger=MESA_create_runtime_log_handle(tsg_logger_info.log_path, tsg_logger_info.level);
if(tsg_logger_info.logger==NULL)
if(_handle!=NULL && key!=NULL)
{
cJSON_DeleteItemFromObject(_handle->object, key);
}
return 0;
}
int TLD_append(struct TLD_handle_t *handle, char *key, void *value, TLD_TYPE type)
{
struct TLD_handle_t *_handle=handle;
if(_handle==NULL || key==NULL || (value==NULL && type!=TLD_TYPE_LONG))
{
printf("MESA_create_runtime_log_handle failed ..., path: %s level: %d", tsg_logger_info.log_path, tsg_logger_info.level);
return -1;
}
tsg_logger_info.cabot_handle=cabot_sendlog_create();
cabot_sendlog_set(tsg_logger_info.cabot_handle, SENDLOG_MODE,(void *)&(tsg_logger_info.mode));
cabot_sendlog_set(tsg_logger_info.cabot_handle, CONFIG_FILE,(void *)tsg_logger_info.field_file);
cabot_sendlog_set(tsg_logger_info.cabot_handle, BROKER_LIST,(void *)&(tsg_logger_info.mode));
ret=cabot_sendlog_init(tsg_logger_info.cabot_handle, tsg_logger_info.logger);
if(ret<0)
switch(type)
{
MESA_handle_runtime_log(tsg_logger_info.logger, RLOG_LV_FATAL, "CABOT_INIT", "cabot_sendlog_init failed ...");
return -2;
case TLD_TYPE_LONG:
cJSON_AddNumberToObject(_handle->object, key, (long)value);
break;
case TLD_TYPE_FILE:
break;
case TLD_TYPE_STRING:
cJSON_AddStringToObject(_handle->object, key, (char *)value);
break;
default:
return -1;
break;
}
MESA_load_profile_string_def(filename, "TSG_LOG", "NIC_NAME", nic_name, sizeof(nic_name), "eth0");
local_ip_nr=get_ip_by_eth_name(nic_name);
if(local_ip_nr==INADDR_NONE)
{
MESA_handle_runtime_log(tsg_logger_info.logger,RLOG_LV_FATAL, "GET_LOCAL_IP","get NIC_NAME: %s error.", nic_name);
return -3;
}
inet_ntop(AF_INET,&(local_ip_nr),tsg_logger_info.local_ip_str,sizeof(tsg_logger_info.local_ip_str));
//maat
return 0;
}
struct TLD_handle_t *TLD_create(int thread_id)
{
//struct _tld_handle *_handle=(struct _tld_handle *)dictator_malloc(thread_id, sizeof(struct _tld_handle));
struct TLD_handle_t *_handle=(struct TLD_handle_t *)calloc(1, sizeof(struct TLD_handle_t));
_handle->thread_id = thread_id;
_handle->object = cJSON_CreateObject();
return _handle;
}
char *log_field_id2name(struct tsg_log_instance_t *instance, tsg_log_field_id_t id)
{
struct tsg_log_instance_t *_instance=instance;
if(_instance!=NULL)
{
return _instance->id2field[id].name;
}
return NULL;
}
int TLD_append_streaminfo(struct tsg_log_instance_t *instance, struct TLD_handle_t *handle, struct streaminfo *a_stream)
{
int ret=0,addr_type=0;
unsigned short tunnel_type=0;
char nest_addr_buf[1024];
char *addr_proto=NULL;
time_t cur_time;
long common_con_duration_ms=0;
unsigned long long stream_id=0;
unsigned short c_port=0, s_port=0;
int tunnel_type_size=sizeof(tunnel_type);
struct layer_addr_ipv4 *ipv4=NULL;
struct layer_addr_ipv6 *ipv6=NULL;
char server_ip[MAX_IPV4_LEN*8]={0};
char client_ip[MAX_IPV4_LEN*8]={0};
struct TLD_handle_t *_handle=handle;
struct tsg_log_instance_t *_instance=instance;
if(_instance==NULL || _handle==NULL || a_stream==NULL)
{
MESA_handle_runtime_log(_instance->logger,
((a_stream==NULL) ? RLOG_LV_DEBUG: RLOG_LV_FATAL),
"TLD_APPEND_STREAM",
"instance==NULL || TLD_handle==NULL || addr: %s",
((a_stream==NULL) ? "NULL": (printaddr(&a_stream->addr, a_stream->threadnum)))
);
return -1;
}
switch(a_stream->addr.addrtype)
{
case ADDR_TYPE_IPV4:
case __ADDR_TYPE_IP_PAIR_V4:
ipv4=a_stream->addr.ipv4;
addr_type=4;
c_port=ntohs(ipv4->source);
s_port=ntohs(ipv4->dest);
inet_ntop(AF_INET, (void *)&ipv4->saddr, client_ip, sizeof(client_ip));
inet_ntop(AF_INET, (void *)&ipv4->daddr, server_ip, sizeof(server_ip));
break;
case ADDR_TYPE_IPV6:
case __ADDR_TYPE_IP_PAIR_V6:
ipv6=a_stream->addr.ipv6;
addr_type=6;
c_port=ntohs(ipv6->source);
s_port=ntohs(ipv6->dest);
inet_ntop(AF_INET6, (void *)ipv6->saddr, client_ip, sizeof(client_ip));
inet_ntop(AF_INET6, (void *)ipv6->daddr, server_ip, sizeof(server_ip));
break;
default:
break;
}
TLD_append(_handle, _instance->id2field[LOG_COMMON_SERVER_IP].name, (void *)server_ip, TLD_TYPE_STRING);
TLD_append(_handle, _instance->id2field[LOG_COMMON_CLIENT_IP].name, (void *)client_ip, TLD_TYPE_STRING);
TLD_append(_handle, _instance->id2field[LOG_COMMON_SERVER_PORT].name, (void *)(long)s_port, TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_CLIENT_PORT].name, (void *)(long)c_port, TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_STREAM_DIR].name, (void *)(long)a_stream->dir, TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_ADDRESS_TYPE].name, (void *)(long)addr_type, TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_S2C_PKT_NUM].name, (void *)(long)a_stream->ptcpdetail->clientpktnum, TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_S2C_BYTE_NUM].name, (void *)(long)a_stream->ptcpdetail->clientbytes, TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_C2S_PKT_NUM].name, (void *)(long)a_stream->ptcpdetail->serverpktnum, TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_C2S_BYTE_NUM].name, (void *)(long)a_stream->ptcpdetail->serverbytes, TLD_TYPE_LONG);
if(a_stream!=NULL && a_stream->ptcpdetail!=NULL)
{
TLD_append(_handle, _instance->id2field[LOG_COMMON_START_TIME].name, (void *)(a_stream->ptcpdetail->createtime), TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_END_TIME].name, (void *)(a_stream->ptcpdetail->lastmtime), TLD_TYPE_LONG);
common_con_duration_ms=(a_stream->ptcpdetail->lastmtime-a_stream->ptcpdetail->createtime)*1000;
TLD_append(_handle, _instance->id2field[LOG_COMMON_CON_DURATION_MS].name, (void *)(common_con_duration_ms), TLD_TYPE_LONG);
}
else
{
cur_time=time(NULL);
TLD_append(_handle, _instance->id2field[LOG_COMMON_START_TIME].name, (void *)cur_time, TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_END_TIME].name, (void *)cur_time, TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_CON_DURATION_MS].name, (void *)(common_con_duration_ms), TLD_TYPE_LONG);
}
stream_id=tsg_get_stream_id(a_stream);
TLD_append(_handle, _instance->id2field[LOG_COMMON_STREAM_TRACE_ID].name, (void *)(long)stream_id, TLD_TYPE_LONG);
addr_proto=(char *)layer_addr_prefix_ntop(a_stream);
TLD_append(_handle, _instance->id2field[LOG_COMMON_L4_PROTOCOL].name, (void *)addr_proto, TLD_TYPE_STRING);
ret=MESA_get_stream_opt(a_stream, MSO_STREAM_TUNNEL_TYPE, &tunnel_type, &tunnel_type_size);
assert(ret==0);
if(tunnel_type==STREAM_TUNNLE_NON)
{
layer_addr_ntop_r(a_stream,nest_addr_buf, sizeof(nest_addr_buf));
}
else
{
stream_addr_list_ntop(a_stream,nest_addr_buf, sizeof(nest_addr_buf));
}
TLD_append(_handle, _instance->id2field[LOG_COMMON_ADDRESS_LIST].name, (void *)nest_addr_buf, TLD_TYPE_STRING);
return 0;
}
void tsg_send_log(const tsg_log_t* log_msg, struct _opt_unit_t* log_opt, int opt_num, int thread_id)
static int load_log_common_field(const char *filename, id2field_t *id2field, id2field_t *service2topic)
{
int i=0;
int ret=0,id=0;
FILE *fp=NULL;
char line[1024]={0};
char field_name[64]={0};
char type_name[32]={0};
fp=fopen(filename, "r");
if(fp==NULL)
{
printf("Open %s failed ...", filename);
return -1;
}
memset(line, 0, sizeof(line));
while((fgets(line, sizeof(line), fp))!=NULL)
{
if(line[0]=='#' || line[0]=='\n' || line[0]=='\r' ||line[0]=='\0')
{
continue;
}
memset(type_name, 0, sizeof(type_name));
ret=sscanf(line, "%s %s %d", type_name, field_name, &id);
assert(ret==3 && id<LOG_COMMON_MAX);
for(i=0; i<LOG_COMMON_MAX; i++)
{
if((strncasecmp(tld_type[i].name, type_name, strlen(tld_type[i].name)))==0)
{
switch(tld_type[i].type)
{
case TLD_TYPE_FILE:
case TLD_TYPE_LONG:
case TLD_TYPE_STRING:
id2field[id].type = tld_type[i].type;
id2field[id].id = id;
memcpy(id2field[id].name, field_name, strlen(field_name));
break;
default:
if((strncasecmp("TOPIC", type_name, strlen("TOPIC")))==0)
{
service2topic[id].type = TLD_TYPE_MAX;
service2topic[id].id = id;
memcpy(service2topic[id].name, field_name, strlen(field_name));
}
break;
}
}
}
memset(line, 0, sizeof(line));
}
fclose(fp);
fp=NULL;
return 0;
}
struct tsg_log_instance_t *tsg_sendlog_init(const char *conffile)
{
int i=0,ret=0,level=30;
char nic_name[32]={0};
char kafka_errstr[1024]={0};
unsigned int local_ip_nr=0;
char log_path[MAX_STRING_LEN*4]={0};
rd_kafka_t *kafka_handle = NULL;
rd_kafka_conf_t *rdkafka_conf = NULL;
rd_kafka_topic_conf_t *topic_conf;
struct tsg_log_instance_t *_instance=NULL;
_instance=(struct tsg_log_instance_t *)calloc(1, sizeof(struct tsg_log_instance_t));
MESA_load_profile_int_def(conffile, "TSG_LOG", "MODE",&(_instance->mode), 0);
MESA_load_profile_string_def(conffile, "TSG_LOG", "COMMON_FIELD_FILE", _instance->common_field_file, sizeof(_instance->common_field_file), NULL);
MESA_load_profile_string_def(conffile, "TSG_LOG", "BROKER_LIST", _instance->broker_list, sizeof(_instance->broker_list), NULL);
MESA_load_profile_int_def(conffile, "TSG_LOG", "LOG_LEVEL",&(level), 30);
MESA_load_profile_string_def(conffile, "TSG_LOG", "LOG_PATH", log_path, sizeof(log_path), NULL);
_instance->logger=MESA_create_runtime_log_handle(log_path, level);
if(_instance->logger==NULL)
{
printf("MESA_create_runtime_log_handle failed ..., path: %s level: %d", log_path, level);
return NULL;
}
MESA_load_profile_string_def(conffile, "TSG_LOG", "NIC_NAME", nic_name, sizeof(nic_name), "eth0");
ret=MESA_get_dev_ipv4(nic_name, (int *)&local_ip_nr);
if(ret<0)
{
MESA_handle_runtime_log(_instance->logger, RLOG_LV_FATAL, "GET_LOCAL_IP", "MESA_get_dev_ipv4 is error, nic_name: %s", nic_name);
return NULL;
}
inet_ntop(AF_INET,&(local_ip_nr),_instance->local_ip_str,sizeof(_instance->local_ip_str));
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, "request.required.acks", "1", kafka_errstr, sizeof(kafka_errstr));
if(!(kafka_handle=rd_kafka_new(RD_KAFKA_PRODUCER, rdkafka_conf, kafka_errstr, sizeof(kafka_errstr))))
{
MESA_handle_runtime_log(_instance->logger, RLOG_LV_FATAL, "KAFKA_INIT", "rd_kafka_new is error");
return NULL;
}
if(rd_kafka_brokers_add(kafka_handle, _instance->broker_list) == 0)
{
MESA_handle_runtime_log(_instance->logger, RLOG_LV_FATAL, "KAFKA_INIT", "rd_kafka_brokers_add is error, broker_list: %s", _instance->broker_list);
return NULL;
}
MESA_load_profile_int_def(conffile, "TSG_LOG", "MAX_SERVICE",&(_instance->max_service), 0);
(_instance->topic_rkt)=(rd_kafka_topic_t **)calloc(1, sizeof(rd_kafka_topic_t*));
*(_instance->topic_rkt)=(rd_kafka_topic_t *)calloc(1, (1+_instance->max_service)*sizeof(rd_kafka_topic_t*));
_instance->service2topic=(id2field_t *)calloc(1, (1+_instance->max_service)*sizeof(id2field_t));
load_log_common_field(_instance->common_field_file, _instance->id2field, _instance->service2topic);
for(i=0; i<_instance->max_service+1; i++)
{
if(_instance->service2topic[i].type==TLD_TYPE_MAX)
{
topic_conf=rd_kafka_topic_conf_new();
_instance->topic_rkt[_instance->service2topic[i].id]=rd_kafka_topic_new(kafka_handle, _instance->service2topic[i].name, topic_conf);
}
}
return _instance;
}
int tsg_send_log(struct tsg_log_instance_t *instance, struct TLD_handle_t *handle, tsg_log_t *log_msg, int thread_id)
{
int i=0,ret=0,status=0;
char *payload=NULL;
struct TLD_handle_t *_handle=handle;
struct tsg_log_instance_t *_instance=instance;
if(_instance==NULL || _handle==NULL || log_msg==NULL)
{
TLD_cancel(handle);
MESA_handle_runtime_log(_instance->logger, RLOG_LV_FATAL, "TSG_SEND_LOG", " instance==NULL || TLD_handle==NULL || log_msg==NULL ");
return -1;
}
if(_instance->mode==CLOSE)
{
TLD_cancel(handle);
MESA_handle_runtime_log(_instance->logger, RLOG_LV_INFO, "TSG_SEND_LOG", "Disable tsg_send_log.");
return 0;
}
//TODO
//common_user_tags
//common_isp
//common_app_label
//common_app_id
//common_protocol_id
//common_has_dup_traffic
//common_stream_error
TLD_append_streaminfo(instance, handle, log_msg->a_stream);
TLD_append(_handle, _instance->id2field[LOG_COMMON_SLED_IP].name, (void *)(_instance->local_ip_str), TLD_TYPE_STRING);
#if 0
struct vxlan_info vinfo;
int opt_val_len = sizeof(vinfo);
status=MESA_get_stream_opt(log_msg->a_stream, MSO_STREAM_VXLAN_INFO, &vinfo, &opt_val_len);
if(status < 0)
{
MESA_handle_runtime_log(_instance->logger, RLOG_LV_DEBUG, "TSG_SEND_LOG", "tsg log: get vxlan info error, tuple4: %s", printaddr(&log_msg->a_stream->addr, thread_id));
}
else
{
TLD_append((TLD_handle_t)_handle, _instance->id2field[LOG_COMMON_LINK_ID].name, (void *)(long)vinfo.link_id, TLD_TYPE_LONG);
TLD_append((TLD_handle_t)_handle, _instance->id2field[LOG_COMMON_DIRECTION].name, (void *)(long)vinfo.link_dir, TLD_TYPE_LONG);
TLD_append((TLD_handle_t)_handle, _instance->id2field[LOG_COMMON_DEVICE_ID].name, (void *)(long)vinfo.dev_id, TLD_TYPE_LONG);
TLD_append((TLD_handle_t)_handle, _instance->id2field[LOG_COMMON_ENTRANCE_ID].name, (void *)(long)vinfo.entrance_id, TLD_TYPE_LONG);
TLD_append((TLD_handle_t)_handle, _instance->id2field[LOG_COMMON_ENCAPSULATION].name, (void *)(long)vinfo.encap_type, TLD_TYPE_LONG);
}
#endif
for(i=0;i<log_msg->result_num; i++)
{
switch(log_msg->result[i].do_log)
{
case LOG_ABORT:
MESA_handle_runtime_log(_instance->logger, RLOG_LV_INFO,
"TSG_SEND_LOG",
"tsg abort log:cfg_id=%d service=%d addr=%s",
log_msg->result[i].config_id,
log_msg->result[i].service_id,
printaddr(&(log_msg->a_stream->addr), thread_id));
continue;
break;
case LOG_ALL:
break;
case LOG_NOFILE:
break;
default:
break;
}
TLD_append(_handle, _instance->id2field[LOG_COMMON_POLICY_ID].name, (void *)(long)(log_msg->result[i].config_id), TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_SERVICE].name, (void *)(long)(log_msg->result[i].service_id), TLD_TYPE_LONG);
TLD_append(_handle, _instance->id2field[LOG_COMMON_ACTION].name, (void *)(long)((unsigned char)log_msg->result[i].action), TLD_TYPE_LONG);
if(log_msg->result[i].serv_def_len<128)
{
TLD_append(_handle, _instance->id2field[LOG_COMMON_USER_REGION].name, (void *)(log_msg->result[i].service_defined), TLD_TYPE_STRING);
}
else
{
char *service_defined=(char *)calloc(1, log_msg->result[i].serv_def_len+1);
ret=Maat_read_rule(g_tsg_maat_feather, &log_msg->result[i], MAAT_RULE_SERV_DEFINE, service_defined, log_msg->result[i].serv_def_len);
if(ret==log_msg->result[i].serv_def_len)
{
TLD_append(_handle, _instance->id2field[LOG_COMMON_USER_REGION].name, (void *)service_defined, TLD_TYPE_STRING);
}
else
{
MESA_handle_runtime_log(_instance->logger,
RLOG_LV_FATAL,
"TSG_SEND_LOG",
"Fetch service_defined failed, policy_id: %d service: %d action: %d addr: %s",
log_msg->result[i].config_id,
log_msg->result[i].service_id,
log_msg->result[i].action,
printaddr(&log_msg->a_stream->addr, thread_id));
}
free((void *)service_defined);
service_defined=NULL;
}
payload = cJSON_PrintUnformatted(_handle->object);
status = rd_kafka_produce(_instance->topic_rkt[log_msg->result[i].service_id], RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, payload, strlen(payload), NULL, 0, NULL);
if(status < 0)
{
MESA_handle_runtime_log(_instance->logger, RLOG_LV_INFO, "TSG_SEND_LOG",
"tsg_send_log to kafka is error, status: %d, topic: %s payload: %s",
status, _instance->service2topic[log_msg->result[i].service_id].name, payload);
}
else
{
MESA_handle_runtime_log(_instance->logger,RLOG_LV_INFO, "TSG_SEND_LOG",
"log send successfully %s: %s", _instance->service2topic[log_msg->result[i].service_id].name, payload);
}
free(payload);
payload=NULL;
TLD_delete(_handle, _instance->id2field[LOG_COMMON_POLICY_ID].name);
TLD_delete(_handle, _instance->id2field[LOG_COMMON_SERVICE].name);
TLD_delete(_handle, _instance->id2field[LOG_COMMON_ACTION].name);
TLD_delete(_handle, _instance->id2field[LOG_COMMON_USER_REGION].name);
FS_operate(g_tsg_para.fs2_handle, g_tsg_para.fs2_field_id[TSG_FS2_LOG], 0, FS_OP_ADD, 1);
}
TLD_cancel(handle);
return 0;
}