robin/waf-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

日常查询由mysql改为clickhouse

Browse Source
This commit is contained in:
robin
2026-02-08 02:00:51 +08:00
parent bc223fd1aa
commit b7388d83b0
43 changed files with 657 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
EdgeAPI/internal/clickhouse/logs_ingest_store.go
View File

@@ -56,6 +56,10 @@ type ListFilter struct {
LastRequestId string
ServerIds []int64
NodeIds []int64
// 搜索条件
Keyword string
Ip string
Domain string
}
// LogsIngestStore 封装对 logs_ingest 的只读列表查询
@@ -126,10 +130,35 @@ func (s *LogsIngestStore) List(ctx context.Context, f ListFilter) (rows []*LogsI
conditions = append(conditions, "firewall_policy_id = "+strconv.FormatInt(f.FirewallPolicyId, 10))
}
// 搜索条件
if f.Keyword != "" {
keyword := escapeString(f.Keyword)
// 在 host, path, ip, ua 中模糊搜索
conditions = append(conditions, fmt.Sprintf("(host LIKE '%%%s%%' OR path LIKE '%%%s%%' OR ip LIKE '%%%s%%' OR ua LIKE '%%%s%%')", keyword, keyword, keyword, keyword))
}
if f.Ip != "" {
conditions = append(conditions, "ip = '"+escapeString(f.Ip)+"'")
}
if f.Domain != "" {
conditions = append(conditions, "host LIKE '%"+escapeString(f.Domain)+"%'")
}
// 游标分页：使用 trace_id 作为游标
// Reverse=false：历史向后翻页，查询更早的数据
// Reverse=true：实时增量拉新，查询更新的数据
if f.LastRequestId != "" {
if f.Reverse {
conditions = append(conditions, "trace_id > '"+escapeString(f.LastRequestId)+"'")
} else {
conditions = append(conditions, "trace_id < '"+escapeString(f.LastRequestId)+"'")
}
}
where := strings.Join(conditions, " AND ")
orderDir := "ASC"
// 默认按时间倒序（最新的在前面），与前端默认行为一致
orderDir := "DESC"
if f.Reverse {
orderDir = "DESC"
orderDir = "ASC"
}
limit := f.Size
if limit <= 0 {
@@ -138,7 +167,7 @@ func (s *LogsIngestStore) List(ctx context.Context, f ListFilter) (rows []*LogsI
if limit > 1000 {
limit = 1000
}
orderBy := fmt.Sprintf("timestamp %s, node_id %s, server_id %s, trace_id %s", orderDir, orderDir, orderDir, orderDir)
orderBy := fmt.Sprintf("timestamp %s, trace_id %s", orderDir, orderDir)
query := fmt.Sprintf("SELECT timestamp, node_id, cluster_id, server_id, host, ip, method, path, status, bytes_in, bytes_out, cost_ms, ua, referer, log_type, trace_id, firewall_policy_id, firewall_rule_group_id, firewall_rule_set_id, firewall_rule_id, request_headers, request_body, response_headers, response_body FROM %s WHERE %s ORDER BY %s LIMIT %d",
table, where, orderBy, limit+1)
@@ -155,13 +184,51 @@ func (s *LogsIngestStore) List(ctx context.Context, f ListFilter) (rows []*LogsI
rows = append(rows, r)
}
}
if !f.Reverse {
if len(rows) > int(limit) {
nextCursor = rows[limit].TraceId
rows = rows[:limit]
}
return rows, nextCursor, nil
}
if len(rows) > int(limit) {
nextCursor = rows[limit].TraceId
rows = rows[:limit]
}
if len(rows) > 0 {
nextCursor = rows[len(rows)-1].TraceId
}
// 实时模式统一返回为“最新在前”，与前端显示和 MySQL 语义一致。
for left, right := 0, len(rows)-1; left < right; left, right = left+1, right-1 {
rows[left], rows[right] = rows[right], rows[left]
}
return rows, nextCursor, nil
}
// FindByTraceId 按 trace_id 查询单条日志详情
func (s *LogsIngestStore) FindByTraceId(ctx context.Context, traceId string) (*LogsIngestRow, error) {
if !s.client.IsConfigured() {
return nil, fmt.Errorf("clickhouse: not configured")
}
if traceId == "" {
return nil, nil
}
table := quoteIdent("logs_ingest")
query := fmt.Sprintf("SELECT timestamp, node_id, cluster_id, server_id, host, ip, method, path, status, bytes_in, bytes_out, cost_ms, ua, referer, log_type, trace_id, firewall_policy_id, firewall_rule_group_id, firewall_rule_set_id, firewall_rule_id, request_headers, request_body, response_headers, response_body FROM %s WHERE trace_id = '%s' LIMIT 1",
table, escapeString(traceId))
var rawRows []map[string]interface{}
if err := s.client.Query(ctx, query, &rawRows); err != nil {
return nil, err
}
if len(rawRows) == 0 {
return nil, nil
}
return mapToLogsIngestRow(rawRows[0]), nil
}
func quoteIdent(name string) string {
return "`" + strings.ReplaceAll(name, "`", "``") + "`"
}
@@ -244,7 +311,7 @@ func mapToLogsIngestRow(m map[string]interface{}) *LogsIngestRow {
return r
}
// RowToPB 将 logs_ingest 一行转为 pb.HTTPAccessLog（列表展示用）
// RowToPB 将 logs_ingest 一行转为 pb.HTTPAccessLog（列表展示用+详情展示）
func RowToPB(r *LogsIngestRow) *pb.HTTPAccessLog {
if r == nil {
return nil
@@ -259,6 +326,9 @@ func RowToPB(r *LogsIngestRow) *pb.HTTPAccessLog {
RemoteAddr: r.IP,
RequestMethod: r.Method,
RequestPath: r.Path,
RequestURI: r.Path, // 前端使用 requestURI 显示完整路径
Scheme: "http", // 默认 http，日志中未存储实际值
Proto: "HTTP/1.1", // 默认值，日志中未存储实际值
Status: int32(r.Status),
RequestLength: int64(r.BytesIn),
BytesSent: int64(r.BytesOut),
@@ -273,6 +343,39 @@ func RowToPB(r *LogsIngestRow) *pb.HTTPAccessLog {
if r.TimeISO8601() != "" {
a.TimeISO8601 = r.TimeISO8601()
}
// TimeLocal: 用户友好的时间格式 (e.g., "2026-02-07 23:17:12")
if !r.Timestamp.IsZero() {
a.TimeLocal = r.Timestamp.Format("2006-01-02 15:04:05")
}
// 解析请求头 (JSON -> map[string]*pb.Strings)
// ClickHouse 中存储的是 map[string]string 格式
if r.RequestHeaders != "" {
var headers map[string]string
if err := json.Unmarshal([]byte(r.RequestHeaders), &headers); err == nil {
a.Header = make(map[string]*pb.Strings)
for k, v := range headers {
a.Header[k] = &pb.Strings{Values: []string{v}}
}
}
}
// 解析响应头 (JSON -> map[string]*pb.Strings)
if r.ResponseHeaders != "" {
var headers map[string]string
if err := json.Unmarshal([]byte(r.ResponseHeaders), &headers); err == nil {
a.SentHeader = make(map[string]*pb.Strings)
for k, v := range headers {
a.SentHeader[k] = &pb.Strings{Values: []string{v}}
}
}
}
// 请求体
if r.RequestBody != "" {
a.RequestBody = []byte(r.RequestBody)
}
return a
}