package fwh_es_mqtt import ( "bytes" "crypto/tls" "encoding/json" "fmt" "fwh-toolbox-server/config" "io" "net/http" "strings" "time" "go.uber.org/zap" ) // 服务端是一个 Elasticsearch,接口遵循 Elasticsearch API 技术规范。 // 该模块使用独立的 HTTP 客户端(basic auth + 可选 TLS 跳过校验), // 与项目通用 httpClient 解耦,原因: // 1. ES 使用 basic auth,通用客户端不支持; // 2. ES 走 HTTPS 自签证书(curl -k),需要跳过 TLS 校验; // 3. ES 的非 200(如 400/404) 返回体仍包含有用的错误信息,需要读取。 const ( esTimeout time.Duration = 3 * time.Minute // ES 聚合查询可能较慢,超时 3min ) var ( esHost string // 形如 "alb-prod.franklinwh.click:5603" esUser string esPassword string esIndexPrefix string // 形如 "iotcore-stat" esClient *http.Client ) // EsHttpInit 由启动流程调用,注入 ES 连接配置。 func EsHttpInit(cfg *config.FwhEsConfig) { esHost = cfg.EsHost esUser = cfg.EsUser esPassword = cfg.EsPassword esIndexPrefix = cfg.EsIndexPrefix transport := &http.Transport{ TLSClientConfig: &tls.Config{ InsecureSkipVerify: cfg.EsInsecureSkipVerify, // 对应 curl -k }, } esClient = &http.Client{ Timeout: esTimeout, Transport: transport, } } // getFullEsUrl 拼接完整的 ES 请求 URL。 // relativePath 形如 "/iotcore-stat-2026.09.07/_search" func getFullEsUrl(relativePath string) string { if !strings.HasPrefix(relativePath, "/") { relativePath = "/" + relativePath } return "https://" + esHost + relativePath } // esDo 发送请求并读取响应体。 // 非 200 时同样返回响应体(供上层解析 ES 错误信息),并附带 error。 func esDo(method, relativePath string, body any) ([]byte, int, error) { if esClient == nil { return nil, 0, fmt.Errorf("[esDo] ES client 未初始化,请先调用 EsHttpInit") } var reader io.Reader if body != nil { jsonData, err := json.Marshal(body) if err != nil { zap.L().Error("[esDo] body marshal failed.", zap.Any("body", body), zap.String("err", err.Error())) return nil, 0, err } reader = bytes.NewReader(jsonData) } url := getFullEsUrl(relativePath) zap.L().Debug("[esDo]", zap.String("method", method), zap.String("url", url)) req, err := http.NewRequest(method, url, reader) if err != nil { zap.L().Error("[esDo] NewRequest failed.", zap.String("err", err.Error())) return nil, 0, err } req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "application/json") req.SetBasicAuth(esUser, esPassword) resp, err := esClient.Do(req) if err != nil { zap.L().Error("[esDo] Do Send failed.", zap.String("err", err.Error())) return nil, 0, err } defer resp.Body.Close() respBody, err := io.ReadAll(resp.Body) if err != nil { return nil, resp.StatusCode, err } if resp.StatusCode != http.StatusOK { zap.L().Warn("[esDo] Response failed", zap.String("status", resp.Status), zap.String("body", string(respBody))) return respBody, resp.StatusCode, fmt.Errorf("[esDo] ES response failed, status: %s", resp.Status) } return respBody, resp.StatusCode, nil } // esGet 发送 GET 请求(如 _count)。 func esGet(relativePath string) ([]byte, int, error) { return esDo(http.MethodGet, relativePath, nil) } // esSearch 向指定索引发送 _search POST 请求,body 为聚合查询 DSL。 func esSearch(index string, query any) ([]byte, error) { respBody, _, err := esDo(http.MethodPost, "/"+index+"/_search", query) if err != nil { return respBody, err } return respBody, nil } // esIndexName 根据日期(UTC 日期,格式 yyyy.MM.dd)拼接索引名。 // 例如 date="2026.09.07" -> "iotcore-stat-2026.09.07" func esIndexName(date string) string { return esIndexPrefix + "-" + date } // esMonthIndexPattern 根据年月拼接“该月所有日索引”的通配模式。 // 例如 year=2026, month=9 -> "iotcore-stat-2026.09.*" // ES 支持索引名通配,可一次跨该月全部日索引做聚合。 func esMonthIndexPattern(year, month int) string { return fmt.Sprintf("%s-%04d.%02d.*", esIndexPrefix, year, month) }