feat(monitor): F6 日志告警引擎 — AlertEngine 支持 type=log 规则查询Loki/ES+触发/恢复 + 数据源认证升级(none/basic/bearer/custom) + 从查询创建告警真实API对接

backend/controller/alert_engine.go

@@ -11,7 +11,6 @@ import (
 	"log"
 	"net/http"
 	"net/url"
-	"strconv"
 	"strings"
 	"time"
 
@@ -67,6 +66,7 @@ func (e *AlertEngine) Check() {
 	e.checkPortAlerts()
 	e.checkLogAlertRules()
 	e.autoResolveAlerts()
+	e.checkLogAlertRecovery()
 }
 
 // tryAcquireLock 尝试获取分布式锁，返回是否获得 Leader 角色
@@ -386,6 +386,143 @@ func (e *AlertEngine) checkPortAlerts() {
 	}
 }
 
+// checkLogAlertRules 执行 type == "log" 的告警规则
+// 通过数据源查询 Loki/ES，结果条数 > 0 即触发。// Logan.Li
+func (e *AlertEngine) checkLogAlertRules() {
+	if e.ds == nil {
+		return
+	}
+	var rules []model.AlertRule
+	e.db.Where("enabled = ? AND type = ?", true, "log").Find(&rules)
+
+	for _, rule := range rules {
+		ds, err := loadLogDS(e.ds, rule.DatasourceID)
+		if err != nil || ds.Type != rule.Source {
+			log.Printf("[AlertEngine] 日志告警规则 %s 数据源无效: id=%d source=%s", rule.Name, rule.DatasourceID, rule.Source)
+			continue
+		}
+
+		count, err := e.queryLogDataSource(ds, rule.Expression)
+		if err != nil {
+			log.Printf("[AlertEngine] 日志告警规则 %s 查询失败: %v", rule.Name, err)
+			continue
+		}
+
+		// 简单策略：查询结果数量 > 0 即触发（后续可扩展为 count_over_time 阈值）// Logan.Li
+		if count == 0 {
+			continue
+		}
+
+		// 避免重复告警：最近 10 分钟已有同规则未恢复事件，不重复
+		var existing model.AlertEvent
+		tenMinAgo := time.Now().Add(-10 * time.Minute)
+		e.db.Where("rule_id = ? AND status = ? AND started_at > ?", rule.ID, model.AlertEventFiring, tenMinAgo).First(&existing)
+		if existing.ID > 0 {
+			continue
+		}
+
+		if e.isSilenced(rule.Name, "", rule.ID) {
+			continue
+		}
+
+		now := time.Now()
+		alert := model.AlertEvent{
+			RuleID:     rule.ID,
+			RuleName:   rule.Name,
+			Severity:   rule.Severity,
+			Status:     string(model.AlertEventFiring),
+			Target:     fmt.Sprintf("%s/%s", ds.Type, ds.Name),
+			Message:    fmt.Sprintf("日志告警触发: %s 在 %s 中匹配到 %d 条日志", rule.Name, ds.Name, count),
+			Value:      float64(count),
+			Threshold:  0,
+			StartedAt:  now,
+			LastEvalAt: now,
+		}
+		e.db.Create(&alert)
+
+		e.notifyAlertRule(rule.Name, alert.Target, "log", rule.Severity, alert.Message, float64(count), 0)
+
+		e.db.Model(&rule).Update("last_triggered", now)
+	}
+}
+
+// queryLogDataSource 根据数据源类型查询日志条数 // Logan.Li
+func (e *AlertEngine) queryLogDataSource(ds *model.LogDataSource, expression string) (int, error) {
+	end := time.Now()
+	start := end.Add(-5 * time.Minute)
+	client := newLogHTTPClient(ds.SkipTLSVerify, 30*time.Second)
+
+	switch strings.ToLower(ds.Type) {
+	case "loki":
+		params := url.Values{}
+		params.Set("query", expression)
+		params.Set("limit", "1")
+		params.Set("direction", "backward")
+		params.Set("start", fmt.Sprintf("%d", start.UnixNano()))
+		params.Set("end", fmt.Sprintf("%d", end.UnixNano()))
+		target := ds.URL + "/loki/api/v1/query_range?" + params.Encode()
+		req, _ := http.NewRequest(http.MethodGet, target, nil)
+		applyLogDSAuthV2(req, ds)
+		resp, err := client.Do(req)
+		if err != nil {
+			return 0, err
+		}
+		defer resp.Body.Close()
+		body, _ := io.ReadAll(resp.Body)
+		if resp.StatusCode >= 400 {
+			return 0, fmt.Errorf("HTTP %s: %s", resp.Status, string(body))
+		}
+		var raw map[string]interface{}
+		_ = json.Unmarshal(body, &raw)
+		entries := flattenLokiStreams(raw)
+		return len(entries), nil
+	case "elasticsearch":
+		searchBody := map[string]interface{}{
+			"size": 0,
+			"query": map[string]interface{}{
+				"bool": map[string]interface{}{
+					"must": []interface{}{
+						map[string]interface{}{"query_string": map[string]interface{}{"query": defaultESQuery(expression)}},
+					},
+				},
+			},
+		}
+		rangeCond := map[string]interface{}{
+			"bool": map[string]interface{}{
+				"should": []interface{}{
+					map[string]interface{}{"range": map[string]interface{}{"@timestamp": map[string]string{"gte": start.Format(time.RFC3339Nano), "lte": end.Format(time.RFC3339Nano)}}},
+					map[string]interface{}{"range": map[string]interface{}{"timestamp": map[string]string{"gte": start.Format(time.RFC3339Nano), "lte": end.Format(time.RFC3339Nano)}}},
+				},
+				"minimum_should_match": 1,
+			},
+		}
+		searchBody["query"].(map[string]interface{})["bool"].(map[string]interface{})["must"] = append(
+			searchBody["query"].(map[string]interface{})["bool"].(map[string]interface{})["must"].([]interface{}),
+			rangeCond,
+		)
+		payload, _ := json.Marshal(searchBody)
+		target := fmt.Sprintf("%s/%s/_search", ds.URL, "*")
+		req, _ := http.NewRequest(http.MethodPost, target, bytes.NewReader(payload))
+		req.Header.Set("Content-Type", "application/json")
+		applyLogDSAuthV2(req, ds)
+		resp, err := client.Do(req)
+		if err != nil {
+			return 0, err
+		}
+		defer resp.Body.Close()
+		body, _ := io.ReadAll(resp.Body)
+		if resp.StatusCode >= 400 {
+			return 0, fmt.Errorf("HTTP %s: %s", resp.Status, string(body))
+		}
+		var raw map[string]interface{}
+		_ = json.Unmarshal(body, &raw)
+		total := esHitsTotal(raw)
+		return int(total), nil
+	default:
+		return 0, fmt.Errorf("未知日志数据源类型: %s", ds.Type)
+	}
+}
+
 // 自动恢复告警（指标恢复正常）
 func (e *AlertEngine) autoResolveAlerts() {
 	var alerts []model.AgentAlert
@@ -478,6 +615,43 @@ func (e *AlertEngine) autoResolveAlerts() {
 	}
 }
 
+// checkLogAlertRecovery 恢复已解决的日志告警事件 // Logan.Li
+func (e *AlertEngine) checkLogAlertRecovery() {
+	if e.ds == nil {
+		return
+	}
+	var events []model.AlertEvent
+	e.db.Where("status = ?", model.AlertEventFiring).Find(&events)
+
+	for _, evt := range events {
+		if evt.RuleID == 0 {
+			continue
+		}
+		var rule model.AlertRule
+		if err := e.db.First(&rule, evt.RuleID).Error; err != nil || rule.Type != "log" {
+			continue
+		}
+		ds, err := loadLogDS(e.ds, rule.DatasourceID)
+		if err != nil {
+			continue
+		}
+		count, err := e.queryLogDataSource(ds, rule.Expression)
+		if err != nil {
+			continue
+		}
+		// 连续两个周期无日志匹配则恢复（简化策略）// Logan.Li
+		if count > 0 {
+			continue
+		}
+		now := time.Now()
+		e.db.Model(&evt).Updates(map[string]interface{}{
+			"status":      model.AlertEventResolved,
+			"resolved_at": now,
+		})
+		e.notifyRecovery(rule.Name, evt.Target, "log", evt.Message)
+	}
+}
+
 func (e *AlertEngine) compareOperator(op string, val, threshold float64) bool {
 	switch op {
 	case ">":

backend/controller/log_datasource.go

@@ -4,6 +4,7 @@ import (
 	"crypto/tls"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"io"
 	"net/http"
 	"strings"
@@ -68,12 +69,18 @@ func (h *LogDSHandler) Create(c *gin.Context) {
 	}
 	ds := model.LogDataSource{
 		Name: req.Name, Type: req.Type, URL: strings.TrimRight(req.URL, "/"),
+		AuthType: normalizeLogDSAuthType(req.AuthType),
 		BasicAuth: req.BasicAuth, BasicAuthUser: req.BasicAuthUser, BasicAuthPass: req.BasicAuthPass,
+		Token: req.Token, Headers: req.Headers,
 		SkipTLSVerify: req.SkipTLSVerify, DefaultIndex: req.DefaultIndex,
 		IsDefault: req.IsDefault, Status: 1,
 	}
 	if err := h.db.Create(&ds).Error; err != nil {
-		c.JSON(500, model.Response{Code: 500, Message: err.Error()})
+		if errors.Is(err, gorm.ErrDuplicatedKey) {
+			c.JSON(409, model.Response{Code: 409, Message: "数据源名称已存在"})
+			return
+		}
+		c.JSON(500, model.Response{Code: 500, Message: "创建失败: " + err.Error()})
 		return
 	}
 	c.JSON(200, model.Response{Code: 200, Message: "创建成功", Data: ds})
@@ -101,6 +108,9 @@ func (h *LogDSHandler) Update(c *gin.Context) {
 	if req.URL != nil {
 		updates["url"] = strings.TrimRight(*req.URL, "/")
 	}
+	if req.AuthType != nil {
+		updates["auth_type"] = normalizeLogDSAuthType(*req.AuthType)
+	}
 	if req.BasicAuth != nil {
 		updates["basic_auth"] = *req.BasicAuth
 	}
@@ -110,6 +120,12 @@ func (h *LogDSHandler) Update(c *gin.Context) {
 	if req.BasicAuthPass != nil && *req.BasicAuthPass != "" {
 		updates["basic_auth_pass"] = *req.BasicAuthPass
 	}
+	if req.Token != nil {
+		updates["token"] = *req.Token
+	}
+	if req.Headers != nil {
+		updates["headers"] = *req.Headers
+	}
 	if req.SkipTLSVerify != nil {
 		updates["skip_tls_verify"] = *req.SkipTLSVerify
 	}
@@ -126,7 +142,11 @@ func (h *LogDSHandler) Update(c *gin.Context) {
 		updates["is_default"] = false
 	}
 	if err := h.db.Model(&ds).Updates(updates).Error; err != nil {
-		c.JSON(500, model.Response{Code: 500, Message: err.Error()})
+		if errors.Is(err, gorm.ErrDuplicatedKey) {
+			c.JSON(409, model.Response{Code: 409, Message: "数据源名称已存在"})
+			return
+		}
+		c.JSON(500, model.Response{Code: 500, Message: "更新失败: " + err.Error()})
 		return
 	}
 	h.db.First(&ds, ds.ID)
@@ -172,9 +192,12 @@ func (h *LogDSHandler) TestConnection(c *gin.Context) {
 		}
 		req.Type = ds.Type
 		req.URL = ds.URL
+		req.AuthType = ds.AuthType
 		req.BasicAuth = ds.BasicAuth
 		req.BasicAuthUser = ds.BasicAuthUser
 		req.BasicAuthPass = ds.BasicAuthPass
+		req.Token = ds.Token
+		req.Headers = ds.Headers
 		req.SkipTLSVerify = ds.SkipTLSVerify
 	} else if req.Type == "" || req.URL == "" {
 		c.JSON(400, model.Response{Code: 400, Message: "datasource_id 与 (type,url) 至少提供一组"})
@@ -197,19 +220,24 @@ func (h *LogDSHandler) ListIndices(c *gin.Context) {
 		c.JSON(400, model.Response{Code: 400, Message: "url 参数必填"})
 		return
 	}
-	basicAuth := c.Query("basic_auth") == "true"
-	user := c.Query("basic_auth_user")
-	pass := c.Query("basic_auth_pass")
-	skipTLS := c.Query("skip_tls_verify") == "true"
+	tempDS := model.LogDataSource{
+		AuthType:      c.Query("auth_type"),
+		BasicAuth:     c.Query("basic_auth") == "true",
+		BasicAuthUser: c.Query("basic_auth_user"),
+		BasicAuthPass: c.Query("basic_auth_pass"),
+		Token:         c.Query("token"),
+		Headers:       c.Query("headers"),
+		SkipTLSVerify: c.Query("skip_tls_verify") == "true",
+	}
 
-	client := newLogHTTPClient(skipTLS, 10*time.Second)
+	client := newLogHTTPClient(tempDS.SkipTLSVerify, 10*time.Second)
 	target := strings.TrimRight(esURL, "/") + "/_cat/indices?format=json&h=index"
 	httpReq, err := http.NewRequest(http.MethodGet, target, nil)
 	if err != nil {
 		c.JSON(502, model.Response{Code: 502, Message: err.Error()})
 		return
 	}
-	applyLogDSAuth(httpReq, basicAuth, user, pass)
+	applyLogDSAuthV2(httpReq, &tempDS)
 	resp, err := client.Do(httpReq)
 	if err != nil {
 		c.JSON(502, model.Response{Code: 502, Message: "查询索引失败: " + err.Error()})
@@ -218,7 +246,8 @@ func (h *LogDSHandler) ListIndices(c *gin.Context) {
 	defer resp.Body.Close()
 	body, _ := io.ReadAll(resp.Body)
 	if resp.StatusCode >= 400 {
-		c.JSON(502, model.Response{Code: 502, Message: "ES 错误: " + string(body)})
+		status := logProxyErrStatus(resp.StatusCode)
+		c.JSON(status, model.Response{Code: status, Message: "ES 错误: " + string(body), Data: map[string]string{"category": logProxyErrCategory(resp.StatusCode)}})
 		return
 	}
 	var raw []map[string]interface{}
@@ -251,7 +280,16 @@ func testLogDSConnection(req model.TestLogDSReq) (bool, string) {
 	if err != nil {
 		return false, err.Error()
 	}
-	applyLogDSAuth(httpReq, req.BasicAuth, req.BasicAuthUser, req.BasicAuthPass)
+	// 用临时 LogDataSource 统一走 V2 认证逻辑 // Logan.Li
+	tempDS := model.LogDataSource{
+		AuthType:      req.AuthType,
+		BasicAuth:     req.BasicAuth,
+		BasicAuthUser: req.BasicAuthUser,
+		BasicAuthPass: req.BasicAuthPass,
+		Token:         req.Token,
+		Headers:       req.Headers,
+	}
+	applyLogDSAuthV2(httpReq, &tempDS)
 	resp, err := client.Do(httpReq)
 	if err != nil {
 		return false, "连接失败: " + err.Error()
@@ -284,4 +322,51 @@ func applyLogDSAuth(req *http.Request, basic bool, user, pass string) {
 	if basic && user != "" {
 		req.Header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(user+":"+pass)))
 	}
+}
+
+// applyLogDSAuthV2 根据完整数据源配置应用认证（支持 basic/bearer/custom headers）// Logan.Li
+func applyLogDSAuthV2(req *http.Request, ds *model.LogDataSource) {
+	authType := strings.ToLower(ds.AuthType)
+	// 兼容旧版：显式开启 BasicAuth 但没有 AuthType 时按 basic 处理 // Logan.Li
+	if authType == "" && ds.BasicAuth {
+		authType = "basic"
+	}
+	switch authType {
+	case "basic":
+		if ds.BasicAuthUser != "" {
+			req.Header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(ds.BasicAuthUser+":"+ds.BasicAuthPass)))
+		}
+	case "bearer":
+		if ds.Token != "" {
+			req.Header.Set("Authorization", "Bearer "+ds.Token)
+		}
+	case "custom":
+		if ds.Headers != "" {
+			var headers []struct {
+				Name  string `json:"name"`
+				Value string `json:"value"`
+			}
+			if err := json.Unmarshal([]byte(ds.Headers), &headers); err == nil {
+				for _, h := range headers {
+					if h.Name != "" {
+						req.Header.Set(h.Name, h.Value)
+					}
+				}
+			}
+		}
+	}
+}
+
+// normalizeLogDSAuthType 规范认证方式取值 // Logan.Li
+func normalizeLogDSAuthType(t string) string {
+	switch strings.ToLower(t) {
+	case "basic":
+		return "basic"
+	case "bearer":
+		return "bearer"
+	case "custom":
+		return "custom"
+	default:
+		return "none"
+	}
 }

backend/controller/log_proxy.go

@@ -79,7 +79,7 @@ func (h *LogProxyHandler) QueryLoki(c *gin.Context) {
 	client := newLogHTTPClient(ds.SkipTLSVerify, time.Duration(60)*time.Second)
 	target := ds.URL + "/loki/api/v1/query_range?" + params.Encode()
 	httpReq, _ := http.NewRequest(http.MethodGet, target, nil)
-	applyLogDSAuth(httpReq, ds.BasicAuth, ds.BasicAuthUser, ds.BasicAuthPass)
+	applyLogDSAuthV2(httpReq, ds)
 	resp, err := client.Do(httpReq)
 	if err != nil {
 		c.JSON(502, model.Response{Code: 502, Message: "查询失败: " + err.Error()})
@@ -88,7 +88,7 @@ func (h *LogProxyHandler) QueryLoki(c *gin.Context) {
 	defer resp.Body.Close()
 	body, _ := io.ReadAll(resp.Body)
 	if resp.StatusCode >= 400 {
-		c.JSON(502, model.Response{Code: 502, Message: "Loki 错误: " + string(body)})
+		c.JSON(logProxyErrStatus(resp.StatusCode), model.Response{Code: logProxyErrStatus(resp.StatusCode), Message: "Loki 错误: " + string(body), Data: map[string]string{"category": logProxyErrCategory(resp.StatusCode)}})
 		return
 	}
 	var raw map[string]interface{}
@@ -137,6 +137,8 @@ func (h *LogProxyHandler) QueryES(c *gin.Context) {
 	if searchBody == nil {
 		searchBody = map[string]interface{}{
 			"size": req.Size,
+			// ES 7+ 默认 track_total_hits 为 10000，必须显式开启才能返回真实命中总数 // Logan.Li
+			"track_total_hits": true,
 			"sort": []interface{}{
 				map[string]interface{}{"_score": map[string]string{"order": "desc"}},
 			},
@@ -210,7 +212,7 @@ func (h *LogProxyHandler) QueryES(c *gin.Context) {
 	target := fmt.Sprintf("%s/%s/_search", ds.URL, strings.Trim(index, "/"))
 	httpReq, _ := http.NewRequest(http.MethodPost, target, bytes.NewReader(payload))
 	httpReq.Header.Set("Content-Type", "application/json")
-	applyLogDSAuth(httpReq, ds.BasicAuth, ds.BasicAuthUser, ds.BasicAuthPass)
+	applyLogDSAuthV2(httpReq, ds)
 	resp, err := client.Do(httpReq)
 	if err != nil {
 		c.JSON(502, model.Response{Code: 502, Message: "查询失败: " + err.Error()})
@@ -219,17 +221,43 @@ func (h *LogProxyHandler) QueryES(c *gin.Context) {
 	defer resp.Body.Close()
 	body, _ := io.ReadAll(resp.Body)
 	if resp.StatusCode >= 400 {
-		c.JSON(502, model.Response{Code: 502, Message: "ES 错误: " + string(body)})
+		c.JSON(logProxyErrStatus(resp.StatusCode), model.Response{Code: logProxyErrStatus(resp.StatusCode), Message: "ES 错误: " + string(body), Data: map[string]string{"category": logProxyErrCategory(resp.StatusCode)}})
 		return
 	}
 	var raw map[string]interface{}
 	_ = json.Unmarshal(body, &raw)
 	entries := flattenESEntries(raw)
+	total := esHitsTotal(raw)
 	c.JSON(200, model.Response{Code: 200, Data: gin.H{
-		"entries": entries, "total": len(entries),
+		"entries": entries, "total": total,
 	}})
 }
 
+// logProxyErrStatus 把下游 HTTP 状态码映射为合适的 API 状态码 // Logan.Li
+func logProxyErrStatus(status int) int {
+	if status >= 400 && status < 500 {
+		return 400
+	}
+	return 502
+}
+
+// logProxyErrCategory 对下游错误进行分类，便于前端展示中文提示 // Logan.Li
+func logProxyErrCategory(status int) string {
+	switch status {
+	case 401:
+		return "auth_failed"
+	case 403:
+		return "forbidden"
+	case 400, 404:
+		return "bad_request"
+	default:
+		if status >= 500 {
+			return "server_error"
+		}
+		return "unknown"
+	}
+}
+
 func parseLogTime(s string) (time.Time, error) {
 	if n, err := strconv.ParseInt(s, 10, 64); err == nil {
 		if n > 1e15 {
@@ -314,6 +342,20 @@ func flattenESEntries(raw map[string]interface{}) []logEntryRow {
 	return out
 }
 
+// esHitsTotal 从 ES 响应中提取真实命中总数（兼容 ES 7+ total.value 与旧版 total）// Logan.Li
+func esHitsTotal(raw map[string]interface{}) int64 {
+	hitsWrap, _ := raw["hits"].(map[string]interface{})
+	if total, ok := hitsWrap["total"].(map[string]interface{}); ok {
+		if v, ok := total["value"].(float64); ok {
+			return int64(v)
+		}
+	}
+	if v, ok := hitsWrap["total"].(float64); ok {
+		return int64(v)
+	}
+	return 0
+}
+
 func pickESMessage(src map[string]interface{}) string {
 	for _, key := range []string{"message", "msg", "log", "content"} {
 		if v, ok := src[key]; ok {