主分支代码

2026-02-07 20:30:31 +08:00
parent 3b042d1dad
commit bc223fd1aa
65 changed files with 1969 additions and 188 deletions
--- a/EdgeAPI/internal/clickhouse/client.go
+++ b/EdgeAPI/internal/clickhouse/client.go
@@ -0,0 +1,134 @@
+package clickhouse
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+)
+
+// Client 通过 HTTP 接口执行只读查询（SELECT），返回 JSONEachRow 解析为 map 或结构体
+type Client struct {
+	cfg     *Config
+	httpCli *http.Client
+}
+
+// NewClient 使用共享配置创建客户端
+func NewClient() *Client {
+	cfg := SharedConfig()
+	return &Client{
+		cfg: cfg,
+		httpCli: &http.Client{
+			Timeout: 30 * time.Second,
+		},
+	}
+}
+
+// IsConfigured 是否已配置
+func (c *Client) IsConfigured() bool {
+	return c.cfg != nil && c.cfg.IsConfigured()
+}
+
+// Query 执行 SELECT，将每行 JSON 解析到 dest 切片；dest 元素类型需为 *struct 或 map
+func (c *Client) Query(ctx context.Context, query string, dest interface{}) error {
+	if !c.IsConfigured() {
+		return fmt.Errorf("clickhouse: not configured")
+	}
+	// 强制 JSONEachRow 便于解析
+	q := strings.TrimSpace(query)
+	if !strings.HasSuffix(strings.ToUpper(q), "FORMAT JSONEACHROW") {
+		query = q + " FORMAT JSONEachRow"
+	}
+	u := c.buildURL(query)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return err
+	}
+	if c.cfg.User != "" || c.cfg.Password != "" {
+		req.SetBasicAuth(c.cfg.User, c.cfg.Password)
+	}
+	resp, err := c.httpCli.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("clickhouse HTTP %d: %s", resp.StatusCode, string(body))
+	}
+	dec := json.NewDecoder(resp.Body)
+	return decodeRows(dec, dest)
+}
+
+// QueryRow 执行仅返回一行的查询，将结果解析到 dest（*struct 或 *map）
+func (c *Client) QueryRow(ctx context.Context, query string, dest interface{}) error {
+	if !c.IsConfigured() {
+		return fmt.Errorf("clickhouse: not configured")
+	}
+	q := strings.TrimSpace(query)
+	if !strings.HasSuffix(strings.ToUpper(q), "FORMAT JSONEACHROW") {
+		query = q + " FORMAT JSONEachRow"
+	}
+	u := c.buildURL(query)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return err
+	}
+	if c.cfg.User != "" || c.cfg.Password != "" {
+		req.SetBasicAuth(c.cfg.User, c.cfg.Password)
+	}
+	resp, err := c.httpCli.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("clickhouse HTTP %d: %s", resp.StatusCode, string(body))
+	}
+	dec := json.NewDecoder(resp.Body)
+	return decodeOneRow(dec, dest)
+}
+
+func (c *Client) buildURL(query string) string {
+	rawURL := fmt.Sprintf("http://%s:%d/?query=%s&database=%s",
+		c.cfg.Host, c.cfg.Port, url.QueryEscape(query), url.QueryEscape(c.cfg.Database))
+	return rawURL
+}
+
+// decodeRows 将 JSONEachRow 流解析到 slice；元素类型须为 *struct 或 *map[string]interface{}
+func decodeRows(dec *json.Decoder, dest interface{}) error {
+	// dest 应为 *[]*SomeStruct 或 *[]map[string]interface{}
+	switch d := dest.(type) {
+	case *[]map[string]interface{}:
+		*d = (*d)[:0]
+		for {
+			var row map[string]interface{}
+			if err := dec.Decode(&row); err != nil {
+				if err == io.EOF {
+					return nil
+				}
+				return err
+			}
+			*d = append(*d, row)
+		}
+	default:
+		return fmt.Errorf("clickhouse: unsupported dest type for Query (use *[]map[string]interface{} or implement decoder)")
+	}
+}
+
+func decodeOneRow(dec *json.Decoder, dest interface{}) error {
+	switch d := dest.(type) {
+	case *map[string]interface{}:
+		if err := dec.Decode(d); err != nil {
+			return err
+		}
+		return nil
+	default:
+		return fmt.Errorf("clickhouse: unsupported dest type for QueryRow (use *map[string]interface{})")
+	}
+}
--- a/EdgeAPI/internal/clickhouse/config.go
+++ b/EdgeAPI/internal/clickhouse/config.go
@@ -0,0 +1,111 @@
+// Package clickhouse 提供 ClickHouse 只读客户端，用于查询 logs_ingest（Fluent Bit 写入）。
+// 配置优先从后台页面（edgeSysSettings.clickhouseConfig）读取，其次 api.yaml，最后环境变量。
+package clickhouse
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/configs"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"os"
+	"strconv"
+	"sync"
+)
+
+const (
+	envHost     = "CLICKHOUSE_HOST"
+	envPort     = "CLICKHOUSE_PORT"
+	envUser     = "CLICKHOUSE_USER"
+	envPassword = "CLICKHOUSE_PASSWORD"
+	envDatabase = "CLICKHOUSE_DATABASE"
+	defaultPort = 8123
+	defaultDB   = "default"
+)
+
+var (
+	sharedConfig *Config
+	configOnce   sync.Once
+	configLocker sync.Mutex
+)
+
+// Config ClickHouse 连接配置（仅查询，不从代码写库）
+type Config struct {
+	Host     string
+	Port     int
+	User     string
+	Password string
+	Database string
+}
+
+// SharedConfig 返回全局配置（优先从后台 DB 读取，其次 api.yaml，最后环境变量）
+func SharedConfig() *Config {
+	configLocker.Lock()
+	defer configLocker.Unlock()
+	if sharedConfig != nil {
+		return sharedConfig
+	}
+	sharedConfig = loadConfig()
+	return sharedConfig
+}
+
+// ResetSharedConfig 清空缓存，下次 SharedConfig() 时重新从 DB/文件/环境变量加载（后台保存 ClickHouse 配置后调用）
+func ResetSharedConfig() {
+	configLocker.Lock()
+	defer configLocker.Unlock()
+	sharedConfig = nil
+}
+
+func loadConfig() *Config {
+	cfg := &Config{Port: defaultPort, Database: defaultDB}
+	// 1) 优先从后台页面配置（DB）读取
+	if models.SharedSysSettingDAO != nil {
+		if dbCfg, err := models.SharedSysSettingDAO.ReadClickHouseConfig(nil); err == nil && dbCfg != nil && dbCfg.Host != "" {
+			cfg.Host = dbCfg.Host
+			cfg.Port = dbCfg.Port
+			cfg.User = dbCfg.User
+			cfg.Password = dbCfg.Password
+			cfg.Database = dbCfg.Database
+			if cfg.Port <= 0 {
+				cfg.Port = defaultPort
+			}
+			if cfg.Database == "" {
+				cfg.Database = defaultDB
+			}
+			return cfg
+		}
+	}
+	// 2) 其次 api.yaml
+	apiConfig, err := configs.SharedAPIConfig()
+	if err == nil && apiConfig != nil && apiConfig.ClickHouse != nil && apiConfig.ClickHouse.Host != "" {
+		ch := apiConfig.ClickHouse
+		cfg.Host = ch.Host
+		cfg.Port = ch.Port
+		cfg.User = ch.User
+		cfg.Password = ch.Password
+		cfg.Database = ch.Database
+		if cfg.Port <= 0 {
+			cfg.Port = defaultPort
+		}
+		if cfg.Database == "" {
+			cfg.Database = defaultDB
+		}
+		return cfg
+	}
+	// 3) 最后环境变量
+	cfg.Host = os.Getenv(envHost)
+	cfg.User = os.Getenv(envUser)
+	cfg.Password = os.Getenv(envPassword)
+	cfg.Database = os.Getenv(envDatabase)
+	if cfg.Database == "" {
+		cfg.Database = defaultDB
+	}
+	if p := os.Getenv(envPort); p != "" {
+		if v, err := strconv.Atoi(p); err == nil {
+			cfg.Port = v
+		}
+	}
+	return cfg
+}
+
+// IsConfigured 是否已配置（Host 非空即视为启用 ClickHouse 查询）
+func (c *Config) IsConfigured() bool {
+	return c != nil && c.Host != ""
+}
--- a/EdgeAPI/internal/clickhouse/logs_ingest_store.go
+++ b/EdgeAPI/internal/clickhouse/logs_ingest_store.go
@@ -0,0 +1,285 @@
+// Package clickhouse 提供 logs_ingest 表的只读查询（列表分页），用于访问日志列表优先走 ClickHouse。
+
+package clickhouse
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+)
+
+// LogsIngestRow 对应 ClickHouse logs_ingest 表的一行（用于 List 结果与 RowToPB）
+type LogsIngestRow struct {
+	Timestamp           time.Time
+	NodeId              uint64
+	ClusterId           uint64
+	ServerId            uint64
+	Host                string
+	IP                  string
+	Method              string
+	Path                string
+	Status              uint16
+	BytesIn             uint64
+	BytesOut            uint64
+	CostMs              uint32
+	UA                  string
+	Referer             string
+	LogType             string
+	TraceId             string
+	FirewallPolicyId    uint64
+	FirewallRuleGroupId uint64
+	FirewallRuleSetId   uint64
+	FirewallRuleId      uint64
+	RequestHeaders      string
+	RequestBody         string
+	ResponseHeaders     string
+	ResponseBody        string
+}
+
+// ListFilter 列表查询条件（与 ListHTTPAccessLogsRequest 对齐）
+type ListFilter struct {
+	Day               string
+	HourFrom          string
+	HourTo            string
+	Size              int64
+	Reverse           bool
+	HasError          bool
+	HasFirewallPolicy bool
+	FirewallPolicyId  int64
+	NodeId            int64
+	ClusterId         int64
+	LastRequestId     string
+	ServerIds         []int64
+	NodeIds           []int64
+}
+
+// LogsIngestStore 封装对 logs_ingest 的只读列表查询
+type LogsIngestStore struct {
+	client *Client
+}
+
+// NewLogsIngestStore 创建 store，内部使用共享 Client
+func NewLogsIngestStore() *LogsIngestStore {
+	return &LogsIngestStore{client: NewClient()}
+}
+
+// Client 返回底层 Client，供调用方判断 IsConfigured()
+func (s *LogsIngestStore) Client() *Client {
+	return s.client
+}
+
+// List 按条件分页查询 logs_ingest，返回行、下一页游标（trace_id）与错误
+func (s *LogsIngestStore) List(ctx context.Context, f ListFilter) (rows []*LogsIngestRow, nextCursor string, err error) {
+	if !s.client.IsConfigured() {
+		return nil, "", fmt.Errorf("clickhouse: not configured")
+	}
+	if f.Day == "" {
+		return nil, "", fmt.Errorf("clickhouse: day required")
+	}
+	table := "logs_ingest"
+	if s.client.cfg.Database != "" && s.client.cfg.Database != "default" {
+		table = quoteIdent(s.client.cfg.Database) + "." + quoteIdent("logs_ingest")
+	} else {
+		table = quoteIdent(table)
+	}
+
+	conditions := []string{"toDate(timestamp) = '" + escapeString(f.Day) + "'"}
+	if f.HourFrom != "" {
+		if _, err := strconv.Atoi(f.HourFrom); err == nil {
+			conditions = append(conditions, "toHour(timestamp) >= "+f.HourFrom)
+		}
+	}
+	if f.HourTo != "" {
+		if _, err := strconv.Atoi(f.HourTo); err == nil {
+			conditions = append(conditions, "toHour(timestamp) <= "+f.HourTo)
+		}
+	}
+	if len(f.ServerIds) > 0 {
+		parts := make([]string, 0, len(f.ServerIds))
+		for _, id := range f.ServerIds {
+			parts = append(parts, strconv.FormatInt(id, 10))
+		}
+		conditions = append(conditions, "server_id IN ("+strings.Join(parts, ",")+")")
+	}
+	if len(f.NodeIds) > 0 {
+		parts := make([]string, 0, len(f.NodeIds))
+		for _, id := range f.NodeIds {
+			parts = append(parts, strconv.FormatInt(id, 10))
+		}
+		conditions = append(conditions, "node_id IN ("+strings.Join(parts, ",")+")")
+	}
+	if f.NodeId > 0 {
+		conditions = append(conditions, "node_id = "+strconv.FormatInt(f.NodeId, 10))
+	}
+	if f.ClusterId > 0 {
+		conditions = append(conditions, "cluster_id = "+strconv.FormatInt(f.ClusterId, 10))
+	}
+	if f.HasFirewallPolicy {
+		conditions = append(conditions, "firewall_policy_id > 0")
+	}
+	if f.FirewallPolicyId > 0 {
+		conditions = append(conditions, "firewall_policy_id = "+strconv.FormatInt(f.FirewallPolicyId, 10))
+	}
+
+	where := strings.Join(conditions, " AND ")
+	orderDir := "ASC"
+	if f.Reverse {
+		orderDir = "DESC"
+	}
+	limit := f.Size
+	if limit <= 0 {
+		limit = 20
+	}
+	if limit > 1000 {
+		limit = 1000
+	}
+	orderBy := fmt.Sprintf("timestamp %s, node_id %s, server_id %s, trace_id %s", orderDir, orderDir, 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)
+
+	var rawRows []map[string]interface{}
+	if err = s.client.Query(ctx, query, &rawRows); err != nil {
+		return nil, "", err
+	}
+
+	rows = make([]*LogsIngestRow, 0, len(rawRows))
+	for _, m := range rawRows {
+		r := mapToLogsIngestRow(m)
+		if r != nil {
+			rows = append(rows, r)
+		}
+	}
+	if len(rows) > int(limit) {
+		nextCursor = rows[limit].TraceId
+		rows = rows[:limit]
+	}
+	return rows, nextCursor, nil
+}
+
+func quoteIdent(name string) string {
+	return "`" + strings.ReplaceAll(name, "`", "``") + "`"
+}
+
+func escapeString(s string) string {
+	return strings.ReplaceAll(s, "'", "''")
+}
+
+func mapToLogsIngestRow(m map[string]interface{}) *LogsIngestRow {
+	r := &LogsIngestRow{}
+	u64 := func(key string) uint64 {
+		v, ok := m[key]
+		if !ok || v == nil {
+			return 0
+		}
+		switch x := v.(type) {
+		case float64:
+			return uint64(x)
+		case string:
+			n, _ := strconv.ParseUint(x, 10, 64)
+			return n
+		case json.Number:
+			n, _ := x.Int64()
+			return uint64(n)
+		}
+		return 0
+	}
+	u32 := func(key string) uint32 {
+		return uint32(u64(key))
+	}
+	str := func(key string) string {
+		if v, ok := m[key]; ok && v != nil {
+			if s, ok := v.(string); ok {
+				return s
+			}
+		}
+		return ""
+	}
+	ts := func(key string) time.Time {
+		v, ok := m[key]
+		if ok && v != nil {
+			switch x := v.(type) {
+			case string:
+				t, _ := time.Parse("2006-01-02 15:04:05", x)
+				return t
+			case float64:
+				return time.Unix(int64(x), 0)
+			case json.Number:
+				n, _ := x.Int64()
+				return time.Unix(n, 0)
+			}
+		}
+		return time.Time{}
+	}
+
+	r.Timestamp = ts("timestamp")
+	r.NodeId = u64("node_id")
+	r.ClusterId = u64("cluster_id")
+	r.ServerId = u64("server_id")
+	r.Host = str("host")
+	r.IP = str("ip")
+	r.Method = str("method")
+	r.Path = str("path")
+	r.Status = uint16(u64("status"))
+	r.BytesIn = u64("bytes_in")
+	r.BytesOut = u64("bytes_out")
+	r.CostMs = u32("cost_ms")
+	r.UA = str("ua")
+	r.Referer = str("referer")
+	r.LogType = str("log_type")
+	r.TraceId = str("trace_id")
+	r.FirewallPolicyId = u64("firewall_policy_id")
+	r.FirewallRuleGroupId = u64("firewall_rule_group_id")
+	r.FirewallRuleSetId = u64("firewall_rule_set_id")
+	r.FirewallRuleId = u64("firewall_rule_id")
+	r.RequestHeaders = str("request_headers")
+	r.RequestBody = str("request_body")
+	r.ResponseHeaders = str("response_headers")
+	r.ResponseBody = str("response_body")
+	return r
+}
+
+// RowToPB 将 logs_ingest 一行转为 pb.HTTPAccessLog（列表展示用）
+func RowToPB(r *LogsIngestRow) *pb.HTTPAccessLog {
+	if r == nil {
+		return nil
+	}
+	a := &pb.HTTPAccessLog{
+		RequestId:           r.TraceId,
+		ServerId:            int64(r.ServerId),
+		NodeId:              int64(r.NodeId),
+		Timestamp:           r.Timestamp.Unix(),
+		Host:                r.Host,
+		RawRemoteAddr:       r.IP,
+		RemoteAddr:          r.IP,
+		RequestMethod:       r.Method,
+		RequestPath:         r.Path,
+		Status:              int32(r.Status),
+		RequestLength:       int64(r.BytesIn),
+		BytesSent:           int64(r.BytesOut),
+		RequestTime:         float64(r.CostMs) / 1000,
+		UserAgent:           r.UA,
+		Referer:             r.Referer,
+		FirewallPolicyId:    int64(r.FirewallPolicyId),
+		FirewallRuleGroupId: int64(r.FirewallRuleGroupId),
+		FirewallRuleSetId:   int64(r.FirewallRuleSetId),
+		FirewallRuleId:      int64(r.FirewallRuleId),
+	}
+	if r.TimeISO8601() != "" {
+		a.TimeISO8601 = r.TimeISO8601()
+	}
+	return a
+}
+
+// TimeISO8601 便于 RowToPB 使用
+func (r *LogsIngestRow) TimeISO8601() string {
+	if r.Timestamp.IsZero() {
+		return ""
+	}
+	return r.Timestamp.UTC().Format("2006-01-02T15:04:05Z07:00")
+}