mirror of
https://github.com/bjdgyc/anylink.git
synced 2025-08-08 12:24:30 +08:00
优化审计日志异步写入
This commit is contained in:
lanrenwo
@@ -9,6 +9,7 @@ import (
|
||||
"github.com/bjdgyc/anylink/dbdata"
|
||||
"github.com/bjdgyc/anylink/pkg/utils"
|
||||
"github.com/bjdgyc/anylink/sessdata"
|
||||
"github.com/ivpusic/grpool"
|
||||
"github.com/songgao/water/waterutil"
|
||||
)
|
||||
|
||||
@@ -19,33 +20,49 @@ const (
|
||||
acc_proto_http
|
||||
)
|
||||
|
||||
var (
|
||||
logAuditSink *LogSink
|
||||
auditPayload *AuditPayload
|
||||
)
|
||||
|
||||
// 审计日志
|
||||
type AuditPayload struct {
|
||||
Pool *grpool.Pool
|
||||
IpAuditMap utils.IMaps
|
||||
}
|
||||
|
||||
// 保存批量的审计日志
|
||||
type LogBatch struct {
|
||||
Logs []dbdata.AccessAudit
|
||||
}
|
||||
|
||||
// 日志池
|
||||
// 批量日志池
|
||||
type LogSink struct {
|
||||
logChan chan dbdata.AccessAudit
|
||||
autoCommitChan chan *LogBatch // 超时通知
|
||||
}
|
||||
|
||||
var logAuditSink *LogSink
|
||||
|
||||
// 写入日志通道
|
||||
func logAuditWrite(aa dbdata.AccessAudit) {
|
||||
logAuditSink.logChan <- aa
|
||||
func (p *AuditPayload) Add(userName string, pl *sessdata.Payload) {
|
||||
bPlData := getByteFull()
|
||||
copy(*bPlData, pl.Data)
|
||||
p.Pool.JobQueue <- func() {
|
||||
logAudit(userName, bPlData)
|
||||
}
|
||||
}
|
||||
|
||||
// 批量写入数据表
|
||||
// 开启批量写入数据功能
|
||||
func logAuditBatch() {
|
||||
if base.Cfg.AuditInterval < 0 {
|
||||
return
|
||||
}
|
||||
logAuditSink = &LogSink{
|
||||
logChan: make(chan dbdata.AccessAudit, 1000),
|
||||
logChan: make(chan dbdata.AccessAudit, 5000),
|
||||
autoCommitChan: make(chan *LogBatch, 10),
|
||||
}
|
||||
auditPayload = &AuditPayload{
|
||||
Pool: grpool.NewPool(10, 500),
|
||||
IpAuditMap: utils.NewMap("cmap", 0),
|
||||
}
|
||||
var (
|
||||
limit = 100 // 超过上限批量写入数据表
|
||||
logAudit dbdata.AccessAudit
|
||||
@@ -53,6 +70,7 @@ func logAuditBatch() {
|
||||
commitTimer *time.Timer // 超时自动提交
|
||||
timeOutBatch *LogBatch
|
||||
)
|
||||
|
||||
for {
|
||||
select {
|
||||
case logAudit = <-logAuditSink.logChan:
|
||||
@@ -85,8 +103,11 @@ func logAuditBatch() {
|
||||
}
|
||||
|
||||
// 解析IP包的数据
|
||||
func logAudit(cSess *sessdata.ConnSession, pl *sessdata.Payload) {
|
||||
ipProto := waterutil.IPv4Protocol(pl.Data)
|
||||
func logAudit(userName string, bPlData *[]byte) {
|
||||
defer putByte(bPlData)
|
||||
|
||||
plData := *bPlData
|
||||
ipProto := waterutil.IPv4Protocol(plData)
|
||||
// 访问协议
|
||||
var accessProto uint8
|
||||
// 只统计 tcp和udp 的访问
|
||||
@@ -99,9 +120,9 @@ func logAudit(cSess *sessdata.ConnSession, pl *sessdata.Payload) {
|
||||
return
|
||||
}
|
||||
|
||||
ipSrc := waterutil.IPv4Source(pl.Data)
|
||||
ipDst := waterutil.IPv4Destination(pl.Data)
|
||||
ipPort := waterutil.IPv4DestinationPort(pl.Data)
|
||||
ipSrc := waterutil.IPv4Source(plData)
|
||||
ipDst := waterutil.IPv4Destination(plData)
|
||||
ipPort := waterutil.IPv4DestinationPort(plData)
|
||||
|
||||
b := getByte51()
|
||||
key := *b
|
||||
@@ -114,79 +135,42 @@ func logAudit(cSess *sessdata.ConnSession, pl *sessdata.Payload) {
|
||||
info := ""
|
||||
nu := utils.NowSec().Unix()
|
||||
if ipProto == waterutil.TCP {
|
||||
plData := waterutil.IPv4Payload(pl.Data)
|
||||
if len(plData) < 14 {
|
||||
tcpPlData := waterutil.IPv4Payload(plData)
|
||||
// 24 (ACK PSH)
|
||||
if len(tcpPlData) < 14 || tcpPlData[13] != 24 {
|
||||
return
|
||||
}
|
||||
flags := plData[13]
|
||||
switch flags {
|
||||
case flags & 0x20:
|
||||
// URG
|
||||
return
|
||||
case flags & 0x14:
|
||||
// RST ACK
|
||||
return
|
||||
case flags & 0x12:
|
||||
// SYN ACK
|
||||
return
|
||||
case flags & 0x11:
|
||||
// Client FIN
|
||||
return
|
||||
case flags & 0x10:
|
||||
// ACK
|
||||
return
|
||||
case flags & 0x08:
|
||||
// PSH
|
||||
return
|
||||
case flags & 0x04:
|
||||
// RST
|
||||
return
|
||||
case flags & 0x02:
|
||||
// SYN
|
||||
return
|
||||
case flags & 0x01:
|
||||
// FIN
|
||||
return
|
||||
case flags & 0x18:
|
||||
// PSH ACK
|
||||
accessProto, info = onTCP(plData)
|
||||
// HTTPS or HTTP
|
||||
if accessProto != acc_proto_tcp {
|
||||
// 提前存储只含ip数据的key, 避免即记录域名又记录一笔IP数据的记录
|
||||
ipKey := make([]byte, 51)
|
||||
copy(ipKey, key)
|
||||
ipS := utils.BytesToString(ipKey)
|
||||
cSess.IpAuditMap.Set(ipS, nu)
|
||||
accessProto, info = onTCP(tcpPlData)
|
||||
// HTTPS or HTTP
|
||||
if accessProto != acc_proto_tcp {
|
||||
// 提前存储只含ip数据的key, 避免即记录域名又记录一笔IP数据的记录
|
||||
ipKey := make([]byte, 51)
|
||||
copy(ipKey, key)
|
||||
ipS := utils.BytesToString(ipKey)
|
||||
auditPayload.IpAuditMap.Set(ipS, nu)
|
||||
|
||||
key[34] = byte(accessProto)
|
||||
// 存储含域名的key
|
||||
if info != "" {
|
||||
md5Sum := md5.Sum([]byte(info))
|
||||
copy(key[35:51], md5Sum[:])
|
||||
}
|
||||
key[34] = byte(accessProto)
|
||||
// 存储含域名的key
|
||||
if info != "" {
|
||||
md5Sum := md5.Sum([]byte(info))
|
||||
copy(key[35:51], md5Sum[:])
|
||||
}
|
||||
case flags & 0x19:
|
||||
// URG
|
||||
return
|
||||
case flags & 0xC2:
|
||||
// SYN-ECE-CWR
|
||||
return
|
||||
}
|
||||
}
|
||||
s := utils.BytesToString(key)
|
||||
|
||||
// 判断已经存在，并且没有过期
|
||||
v, ok := cSess.IpAuditMap.Get(s)
|
||||
v, ok := auditPayload.IpAuditMap.Get(s)
|
||||
if ok && nu-v.(int64) < int64(base.Cfg.AuditInterval) {
|
||||
// 回收byte对象
|
||||
putByte51(b)
|
||||
return
|
||||
}
|
||||
|
||||
cSess.IpAuditMap.Set(s, nu)
|
||||
auditPayload.IpAuditMap.Set(s, nu)
|
||||
|
||||
audit := dbdata.AccessAudit{
|
||||
Username: cSess.Username,
|
||||
Username: userName,
|
||||
Protocol: uint8(ipProto),
|
||||
Src: ipSrc.String(),
|
||||
Dst: ipDst.String(),
|
||||
@@ -195,5 +179,5 @@ func logAudit(cSess *sessdata.ConnSession, pl *sessdata.Payload) {
|
||||
AccessProto: accessProto,
|
||||
Info: info,
|
||||
}
|
||||
logAuditWrite(audit)
|
||||
logAuditSink.logChan <- audit
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user