130 lines
2.7 KiB
Go
130 lines
2.7 KiB
Go
package nodes
|
|
|
|
import (
|
|
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
|
|
"github.com/TeaOSLab/EdgeDNS/internal/accesslogs"
|
|
"github.com/TeaOSLab/EdgeDNS/internal/goman"
|
|
"github.com/TeaOSLab/EdgeDNS/internal/remotelogs"
|
|
"github.com/TeaOSLab/EdgeDNS/internal/rpc"
|
|
"strconv"
|
|
"time"
|
|
)
|
|
|
|
var sharedNSAccessLogQueue = NewNSAccessLogQueue()
|
|
|
|
// NSAccessLogQueue NS访问日志队列
|
|
type NSAccessLogQueue struct {
|
|
queue chan *pb.NSAccessLog
|
|
}
|
|
|
|
// NewNSAccessLogQueue 获取新对象
|
|
func NewNSAccessLogQueue() *NSAccessLogQueue {
|
|
// 队列中最大的值，超出此数量的访问日志会被抛弃
|
|
// TODO 需要可以在界面中设置
|
|
var maxSize = 20000
|
|
queue := &NSAccessLogQueue{
|
|
queue: make(chan *pb.NSAccessLog, maxSize),
|
|
}
|
|
goman.New(func() {
|
|
queue.Start()
|
|
})
|
|
|
|
return queue
|
|
}
|
|
|
|
// Start 开始处理访问日志
|
|
func (this *NSAccessLogQueue) Start() {
|
|
var ticker = time.NewTicker(1 * time.Second)
|
|
for range ticker.C {
|
|
err := this.loop()
|
|
if err != nil {
|
|
if rpc.IsConnError(err) {
|
|
remotelogs.Debug("ACCESS_LOG_QUEUE", err.Error())
|
|
} else {
|
|
remotelogs.Error("ACCESS_LOG_QUEUE", err.Error())
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Push 加入新访问日志
|
|
func (this *NSAccessLogQueue) Push(accessLog *pb.NSAccessLog) {
|
|
select {
|
|
case this.queue <- accessLog:
|
|
default:
|
|
|
|
}
|
|
}
|
|
|
|
// 上传访问日志
|
|
func (this *NSAccessLogQueue) loop() error {
|
|
var accessLogs = []*pb.NSAccessLog{}
|
|
var count = 0
|
|
var timestamp int64
|
|
var requestId = 10_000_000
|
|
|
|
Loop:
|
|
for {
|
|
select {
|
|
case accessLog := <-this.queue:
|
|
if accessLog.Timestamp > timestamp {
|
|
requestId = 10_000_000
|
|
timestamp = accessLog.Timestamp
|
|
} else {
|
|
requestId++
|
|
}
|
|
|
|
// timestamp + nodeId + requestId
|
|
accessLog.RequestId = strconv.FormatInt(accessLog.Timestamp, 10) + strconv.FormatInt(accessLog.NsNodeId, 10) + strconv.Itoa(requestId)
|
|
|
|
accessLogs = append(accessLogs, accessLog)
|
|
count++
|
|
|
|
// 每次只提交 N 条访问日志，防止网络拥堵
|
|
if count > 2000 {
|
|
break Loop
|
|
}
|
|
default:
|
|
break Loop
|
|
}
|
|
}
|
|
|
|
if len(accessLogs) == 0 {
|
|
return nil
|
|
}
|
|
|
|
var clusterId int64
|
|
var needWriteFile = true
|
|
var needReportAPI = true
|
|
var cfg = dnsNodeConfig()
|
|
if cfg != nil {
|
|
clusterId = cfg.ClusterId
|
|
if cfg.AccessLogWriteTargets != nil {
|
|
targets := cfg.AccessLogWriteTargets
|
|
needWriteFile = targets.File || targets.ClickHouse
|
|
needReportAPI = targets.MySQL
|
|
}
|
|
}
|
|
|
|
if needWriteFile {
|
|
accesslogs.SharedDNSFileWriter().WriteBatch(accessLogs, clusterId)
|
|
}
|
|
|
|
if !needReportAPI {
|
|
return nil
|
|
}
|
|
|
|
// 发送到API
|
|
client, err := rpc.SharedRPC()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
_, err = client.NSAccessLogRPC.CreateNSAccessLogs(client.Context(), &pb.CreateNSAccessLogsRequest{NsAccessLogs: accessLogs})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|