303 lines
7.0 KiB
Go
303 lines
7.0 KiB
Go
package accesslogs
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
|
||
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
|
||
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
|
||
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
|
||
"gopkg.in/natefinch/lumberjack.v2"
|
||
)
|
||
|
||
var (
|
||
sharedFileWriter *FileWriter
|
||
sharedOnce sync.Once
|
||
)
|
||
|
||
// SharedFileWriter 返回全局本地日志文件写入器（单例）
|
||
func SharedFileWriter() *FileWriter {
|
||
sharedOnce.Do(func() {
|
||
sharedFileWriter = NewFileWriter()
|
||
})
|
||
return sharedFileWriter
|
||
}
|
||
|
||
const (
|
||
defaultLogDir = "/var/log/edge/edge-node"
|
||
envLogDir = "EDGE_LOG_DIR"
|
||
)
|
||
|
||
// FileWriter 将访问/WAF/错误日志以 JSON Lines 写入本地文件，便于 Fluent Bit 采集。
|
||
// 文件轮转由 lumberjack 内建完成。
|
||
type FileWriter struct {
|
||
dir string
|
||
mu sync.Mutex
|
||
files map[string]*lumberjack.Logger // access.log, waf.log, error.log
|
||
rotateConfig *serverconfigs.AccessLogRotateConfig
|
||
inited bool
|
||
}
|
||
|
||
// NewFileWriter 创建本地日志文件写入器
|
||
func NewFileWriter() *FileWriter {
|
||
dir := resolveDefaultLogDir()
|
||
return &FileWriter{
|
||
dir: dir,
|
||
files: make(map[string]*lumberjack.Logger),
|
||
rotateConfig: serverconfigs.NewDefaultAccessLogRotateConfig(),
|
||
}
|
||
}
|
||
|
||
func resolveDefaultLogDir() string {
|
||
dir := strings.TrimSpace(os.Getenv(envLogDir))
|
||
if dir == "" {
|
||
return defaultLogDir
|
||
}
|
||
return dir
|
||
}
|
||
|
||
func resolveDirFromPolicyPath(policyPath string) string {
|
||
policyPath = strings.TrimSpace(policyPath)
|
||
if policyPath == "" {
|
||
return ""
|
||
}
|
||
|
||
if strings.HasSuffix(policyPath, "/") || strings.HasSuffix(policyPath, "\\") {
|
||
return filepath.Clean(policyPath)
|
||
}
|
||
|
||
baseName := filepath.Base(policyPath)
|
||
if strings.Contains(baseName, ".") || strings.Contains(baseName, "${") {
|
||
return filepath.Clean(filepath.Dir(policyPath))
|
||
}
|
||
|
||
return filepath.Clean(policyPath)
|
||
}
|
||
|
||
// Dir 返回当前配置的日志目录
|
||
func (w *FileWriter) Dir() string {
|
||
return w.dir
|
||
}
|
||
|
||
// SetDirByPolicyPath 使用公用日志策略 path 更新目录，空值时回退到 EDGE_LOG_DIR/default。
|
||
func (w *FileWriter) SetDirByPolicyPath(policyPath string) {
|
||
dir := resolveDirFromPolicyPath(policyPath)
|
||
w.SetDir(dir)
|
||
}
|
||
|
||
// SetDir 更新日志目录并重置文件句柄。
|
||
func (w *FileWriter) SetDir(dir string) {
|
||
if strings.TrimSpace(dir) == "" {
|
||
dir = resolveDefaultLogDir()
|
||
}
|
||
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
|
||
if dir == w.dir {
|
||
return
|
||
}
|
||
|
||
for name, file := range w.files {
|
||
if file != nil {
|
||
_ = file.Close()
|
||
}
|
||
w.files[name] = nil
|
||
}
|
||
w.inited = false
|
||
w.dir = dir
|
||
}
|
||
|
||
// SetRotateConfig 更新日志轮转配置并重建 writer。
|
||
func (w *FileWriter) SetRotateConfig(config *serverconfigs.AccessLogRotateConfig) {
|
||
normalized := config.Normalize()
|
||
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
|
||
if equalRotateConfig(w.rotateConfig, normalized) {
|
||
return
|
||
}
|
||
|
||
for name, file := range w.files {
|
||
if file != nil {
|
||
_ = file.Close()
|
||
}
|
||
w.files[name] = nil
|
||
}
|
||
w.inited = false
|
||
w.rotateConfig = normalized
|
||
}
|
||
|
||
// IsEnabled 是否启用落盘（目录非空即视为启用）
|
||
func (w *FileWriter) IsEnabled() bool {
|
||
return w.dir != ""
|
||
}
|
||
|
||
// EnsureInit 在启动时预创建日志目录和空文件，便于 Fluent Bit 立即 tail，无需等首条访问日志
|
||
func (w *FileWriter) EnsureInit() error {
|
||
if w.dir == "" {
|
||
return nil
|
||
}
|
||
return w.init()
|
||
}
|
||
|
||
// init 确保目录存在并打开三个日志文件（仅首次或 Reopen 时）
|
||
func (w *FileWriter) init() error {
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
if w.inited && len(w.files) > 0 {
|
||
return nil
|
||
}
|
||
if w.dir == "" {
|
||
return nil
|
||
}
|
||
if err := os.MkdirAll(w.dir, 0755); err != nil {
|
||
remotelogs.Error("ACCESS_LOG_FILE", "mkdir log dir failed: "+err.Error())
|
||
return err
|
||
}
|
||
for _, name := range []string{"access.log", "waf.log", "error.log"} {
|
||
if w.files[name] != nil {
|
||
continue
|
||
}
|
||
w.files[name] = w.newLogger(name)
|
||
}
|
||
w.inited = true
|
||
return nil
|
||
}
|
||
|
||
func (w *FileWriter) newLogger(fileName string) *lumberjack.Logger {
|
||
rotateConfig := w.rotateConfig.Normalize()
|
||
return &lumberjack.Logger{
|
||
Filename: filepath.Join(w.dir, fileName),
|
||
MaxSize: rotateConfig.MaxSizeMB,
|
||
MaxBackups: rotateConfig.MaxBackups,
|
||
MaxAge: rotateConfig.MaxAgeDays,
|
||
Compress: *rotateConfig.Compress,
|
||
LocalTime: *rotateConfig.LocalTime,
|
||
}
|
||
}
|
||
|
||
// Write 将一条访问日志按 log_type 写入对应文件（access.log / waf.log / error.log）
|
||
func (w *FileWriter) Write(l *pb.HTTPAccessLog, clusterId int64) {
|
||
if w.dir == "" {
|
||
return
|
||
}
|
||
if err := w.init(); err != nil || len(w.files) == 0 {
|
||
return
|
||
}
|
||
ingest, logType := FromHTTPAccessLog(l, clusterId)
|
||
line, err := json.Marshal(ingest)
|
||
if err != nil {
|
||
remotelogs.Error("ACCESS_LOG_FILE", "marshal ingest log: "+err.Error())
|
||
return
|
||
}
|
||
var fileName string
|
||
switch logType {
|
||
case LogTypeWAF:
|
||
fileName = "waf.log"
|
||
case LogTypeError:
|
||
fileName = "error.log"
|
||
default:
|
||
fileName = "access.log"
|
||
}
|
||
w.mu.Lock()
|
||
file := w.files[fileName]
|
||
w.mu.Unlock()
|
||
if file == nil {
|
||
return
|
||
}
|
||
_, err = file.Write(append(line, '\n'))
|
||
if err != nil {
|
||
remotelogs.Error("ACCESS_LOG_FILE", "write "+fileName+" failed: "+err.Error())
|
||
}
|
||
}
|
||
|
||
// WriteBatch 批量写入，减少锁竞争
|
||
func (w *FileWriter) WriteBatch(logs []*pb.HTTPAccessLog, clusterId int64) {
|
||
if w.dir == "" || len(logs) == 0 {
|
||
return
|
||
}
|
||
if err := w.init(); err != nil || len(w.files) == 0 {
|
||
return
|
||
}
|
||
w.mu.Lock()
|
||
accessFile := w.files["access.log"]
|
||
wafFile := w.files["waf.log"]
|
||
errorFile := w.files["error.log"]
|
||
w.mu.Unlock()
|
||
if accessFile == nil && wafFile == nil && errorFile == nil {
|
||
return
|
||
}
|
||
for _, logItem := range logs {
|
||
ingest, logType := FromHTTPAccessLog(logItem, clusterId)
|
||
line, err := json.Marshal(ingest)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
line = append(line, '\n')
|
||
var file *lumberjack.Logger
|
||
switch logType {
|
||
case LogTypeWAF:
|
||
file = wafFile
|
||
case LogTypeError:
|
||
file = errorFile
|
||
default:
|
||
file = accessFile
|
||
}
|
||
if file != nil {
|
||
_, _ = file.Write(line)
|
||
}
|
||
}
|
||
}
|
||
|
||
// Reopen 关闭并重建所有日志 writer（供 SIGHUP 兼容调用）。
|
||
func (w *FileWriter) Reopen() error {
|
||
if w.dir == "" {
|
||
return nil
|
||
}
|
||
w.mu.Lock()
|
||
for name, file := range w.files {
|
||
if file != nil {
|
||
_ = file.Close()
|
||
w.files[name] = nil
|
||
}
|
||
}
|
||
w.inited = false
|
||
w.mu.Unlock()
|
||
return w.init()
|
||
}
|
||
|
||
// Close 关闭所有已打开的文件
|
||
func (w *FileWriter) Close() error {
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
var lastErr error
|
||
for name, file := range w.files {
|
||
if file != nil {
|
||
if err := file.Close(); err != nil {
|
||
lastErr = err
|
||
remotelogs.Error("ACCESS_LOG_FILE", fmt.Sprintf("close %s: %v", name, err))
|
||
}
|
||
w.files[name] = nil
|
||
}
|
||
}
|
||
w.inited = false
|
||
return lastErr
|
||
}
|
||
|
||
func equalRotateConfig(left *serverconfigs.AccessLogRotateConfig, right *serverconfigs.AccessLogRotateConfig) bool {
|
||
if left == nil || right == nil {
|
||
return left == right
|
||
}
|
||
return left.MaxSizeMB == right.MaxSizeMB &&
|
||
left.MaxBackups == right.MaxBackups &&
|
||
left.MaxAgeDays == right.MaxAgeDays &&
|
||
*left.Compress == *right.Compress &&
|
||
*left.LocalTime == *right.LocalTime
|
||
}
|