waf log >> db.logs

2026-07-25 21:54:30 +08:00
1 parent 2da504cdc4
commit 49c0dca8cf
@@ -11,6 +11,8 @@ import (

 	"github.com/corazawaf/coraza/v3"
 	"github.com/corazawaf/coraza/v3/types"
+
+	"github.com/micro_servers/waf/internal/logclient"
 )

 // Engine 持有当前生效的 WAF 实例，支持通过 Reload 热替换。
@@ -257,8 +259,8 @@ func TestFile(file string) error {
 	return nil
 }

-// LogError 默认错误回调，输出到 stderr。
+// LogError 默认错误回调，推送到日志库。
 func LogError(mr types.MatchedRule) {
-	fmt.Fprintf(os.Stderr, "[coraza] rule_id=%d severity=%s msg=%s\n",
-		mr.Rule().ID(), mr.Rule().Severity(), mr.ErrorLog())
+	logclient.Error("waf", fmt.Sprintf("coraza rule_id=%d severity=%s", mr.Rule().ID(), mr.Rule().Severity()),
+		map[string]any{"rule_id": mr.Rule().ID(), "msg": mr.ErrorLog()})
 }
@@ -0,0 +1,184 @@
+// Package logclient 提供异步批量日志推送客户端，将日志发送到 db 服务的 /log/ingest 接口。
+//
+// 设计要点（与 Python log_client.py 对齐）：
+//   - 异步非阻塞：内存 channel + 后台 goroutine 批量推送，不拖慢业务
+//   - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
+//   - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
+package logclient
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"sync"
+	"time"
+)
+
+const (
+	defaultURL     = "http://127.0.0.1:8005/log/ingest"
+	batchSize      = 50
+	flushInterval  = 2 * time.Second
+	requestTimeout = 5 * time.Second
+)
+
+// Entry 单条日志。
+type Entry struct {
+	Service string         `json:"service"`
+	Level   string         `json:"level"`
+	Message string         `json:"message"`
+	TraceID string         `json:"trace_id,omitempty"`
+	Extra   map[string]any `json:"extra,omitempty"`
+}
+
+// Client 异步批量日志推送客户端。
+type Client struct {
+	url     string
+	ch      chan Entry
+	client  *http.Client
+	stop    chan struct{}
+	wg      sync.WaitGroup
+	stopped bool
+	mu      sync.Mutex
+}
+
+// global 全局单例。
+var global *Client
+
+// Init 初始化全局日志客户端并启动后台 flush goroutine。
+func Init() {
+	global = New(defaultURL)
+}
+
+// New 创建客户端实例。
+func New(url string) *Client {
+	c := &Client{
+		url:    url,
+		ch:     make(chan Entry, 256),
+		client: &http.Client{Timeout: requestTimeout},
+		stop:   make(chan struct{}),
+	}
+	c.wg.Add(1)
+	go c.flushLoop()
+	return c
+}
+
+// Push 推送一条日志（非阻塞，队列满时丢弃）。
+func Push(service, level, message string, extra map[string]any) {
+	if global == nil {
+		return
+	}
+	global.push(Entry{
+		Service: service,
+		Level:   level,
+		Message: message,
+		Extra:   extra,
+	})
+}
+
+// Info 推送 INFO 级别日志。
+func Info(service, message string, extra map[string]any) {
+	Push(service, "INFO", message, extra)
+}
+
+// Warning 推送 WARNING 级别日志。
+func Warning(service, message string, extra map[string]any) {
+	Push(service, "WARNING", message, extra)
+}
+
+// Error 推送 ERROR 级别日志。
+func Error(service, message string, extra map[string]any) {
+	Push(service, "ERROR", message, extra)
+}
+
+// Flush 同步 flush 队列中剩余日志（用于进程退出前）。
+func Flush() {
+	if global == nil {
+		return
+	}
+	global.flush()
+}
+
+func (c *Client) push(e Entry) {
+	select {
+	case c.ch <- e:
+	default:
+		// 队列满，静默丢弃（fail-safe）
+	}
+}
+
+func (c *Client) flushLoop() {
+	defer c.wg.Done()
+	batch := make([]Entry, 0, batchSize)
+	ticker := time.NewTicker(flushInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case e := <-c.ch:
+			batch = append(batch, e)
+			// 非阻塞地继续取，直到满 batchSize
+			for len(batch) < batchSize {
+				select {
+				case e2 := <-c.ch:
+					batch = append(batch, e2)
+				default:
+					goto send
+				}
+			}
+		send:
+			if len(batch) >= batchSize {
+				c.send(batch)
+				batch = batch[:0]
+			}
+		case <-ticker.C:
+			if len(batch) > 0 {
+				c.send(batch)
+				batch = batch[:0]
+			}
+		case <-c.stop:
+			// 退出前 drain channel
+			for {
+				select {
+				case e := <-c.ch:
+					batch = append(batch, e)
+				default:
+					if len(batch) > 0 {
+						c.send(batch)
+					}
+					return
+				}
+			}
+		}
+	}
+}
+
+func (c *Client) send(batch []Entry) {
+	payload := map[string]any{"logs": batch}
+	data, err := json.Marshal(payload)
+	if err != nil {
+		return
+	}
+	req, err := http.NewRequest("POST", c.url, bytes.NewReader(data))
+	if err != nil {
+		return
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return // fail-safe：静默丢弃
+	}
+	resp.Body.Close()
+}
+
+func (c *Client) flush() {
+	c.mu.Lock()
+	if c.stopped {
+		c.mu.Unlock()
+		return
+	}
+	c.stopped = true
+	c.mu.Unlock()
+
+	close(c.stop)
+	c.wg.Wait()
+}
@@ -3,9 +3,8 @@ package proxy

 import (
 	"bytes"
-	"encoding/json"
+	"fmt"
 	"io"
-	"log"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
@@ -14,6 +13,7 @@ import (
 	"time"

 	"github.com/micro_servers/waf/internal/engine"
+	"github.com/micro_servers/waf/internal/logclient"
 )

 // Proxy 持有 ReverseProxy 与 WAF 引擎引用。
@@ -46,7 +46,8 @@ func New(upstream string, waf *engine.Engine) (*Proxy, error) {
 	}
 	// 透传错误，不暴露内部堆栈
 	rp.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
-		log.Printf("[proxy] upstream_error ip=%s path=%s err=%v", r.RemoteAddr, r.URL.Path, err)
+		logclient.Error("waf", fmt.Sprintf("upstream_error path=%s err=%v", r.URL.Path, err),
+			map[string]any{"ip": r.RemoteAddr})
 		http.Error(w, "upstream unavailable", http.StatusBadGateway)
 	}
 	return p, nil
@@ -111,27 +112,20 @@ func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 	tx.ProcessResponseHeaders(rw.status, "HTTP/1.1")
-
-	log.Printf("[proxy] pass ip=%s method=%s path=%s status=%d rule=- cost=%s",
-		r.RemoteAddr, r.Method, r.URL.Path, rw.status, time.Since(start))
 }

-// block 命中规则时统一返回 403，并打 JSON 日志便于 jq 分析。
+// block 命中规则时统一返回 403，并将关键输入推送到日志库。
 func (p *Proxy) block(w http.ResponseWriter, r *http.Request, ruleID int, start time.Time) {
-	rec := map[string]any{
-		"ts":      time.Now().UTC().Format(time.RFC3339),
-		"action":  "block",
-		"rule_id": ruleID,
-		"ip":      r.RemoteAddr,
-		"method":  r.Method,
-		"path":    r.URL.Path,
-		"query":   r.URL.RawQuery,
-		"ua":      r.UserAgent(),
-		"cost_ms": time.Since(start).Milliseconds(),
-	}
-	if raw, err := json.Marshal(rec); err == nil {
-		log.Println(string(raw))
-	}
+	logclient.Warning("waf", fmt.Sprintf("block rule_id=%d %s %s", ruleID, r.Method, r.URL.Path),
+		map[string]any{
+			"rule_id": ruleID,
+			"ip":      r.RemoteAddr,
+			"method":  r.Method,
+			"path":    r.URL.Path,
+			"query":   r.URL.RawQuery,
+			"ua":      r.UserAgent(),
+			"cost_ms": time.Since(start).Milliseconds(),
+		})
 	w.WriteHeader(http.StatusForbidden)
 	_, _ = w.Write([]byte("blocked by WAF\n"))
 }
@@ -16,6 +16,7 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
+	"io"
 	"log"
 	"net/http"
 	"os"
@@ -25,6 +26,7 @@ import (
 	"gopkg.in/yaml.v3"

 	"github.com/micro_servers/waf/internal/engine"
+	"github.com/micro_servers/waf/internal/logclient"
 	"github.com/micro_servers/waf/internal/proxy"
 )

@@ -82,9 +84,14 @@ func main() {
 	}

 	// 服务模式
+	logclient.Init()
+	log.SetOutput(io.Discard) // 静默标准 log
+
 	p, err := proxy.New(cfg.Upstream, eng)
 	if err != nil {
-		log.Fatalf("代理初始化失败: %v", err)
+		logclient.Error("waf", fmt.Sprintf("代理初始化失败: %v", err), nil)
+		logclient.Flush()
+		os.Exit(1)
 	}

 	// SIGHUP 热加载
@@ -92,11 +99,10 @@ func main() {
 	signal.Notify(sigCh, syscall.SIGHUP)
 	go func() {
 		for range sigCh {
-			log.Println("[reload] 收到 SIGHUP，重新加载规则")
 			if err := eng.Reload(); err != nil {
-				log.Printf("[reload] 失败，保留旧规则: %v", err)
+				logclient.Error("waf", fmt.Sprintf("热加载失败，保留旧规则: %v", err), nil)
 			} else {
-				log.Println("[reload] 成功")
+				logclient.Info("waf", "规则热加载成功", nil)
 			}
 		}
 	}()
@@ -109,13 +115,13 @@ func main() {
 		},
 	}

-	log.Printf("waf listening on %s, upstream=%s, rules=%s",
-		cfg.Listen, cfg.Upstream, cfg.RulesDir)
-	log.Printf("  - 热加载: kill -HUP %d", os.Getpid())
-	log.Printf("  - 列规则: %s -c %s -list", os.Args[0], *configFile)
+	logclient.Info("waf", fmt.Sprintf("服务启动 listen=%s upstream=%s rules=%s pid=%d",
+		cfg.Listen, cfg.Upstream, cfg.RulesDir, os.Getpid()), nil)

 	if err := server.ListenAndServeTLS(cfg.TLSCert, cfg.TLSKey); err != nil {
-		log.Fatal(err)
+		logclient.Error("waf", fmt.Sprintf("TLS 监听失败: %v", err), nil)
+		logclient.Flush()
+		os.Exit(1)
 	}
 }