272 lines
7.6 KiB
Go
272 lines
7.6 KiB
Go
package db
|
||
|
||
// influx.go —— InfluxDB 2.x 原生接入（HTTP API，零第三方依赖）。
|
||
//
|
||
// 字段映射（连接表单）：
|
||
//
|
||
// 主机/端口 → 服务地址（默认 :8086）
|
||
// 用户 → org（如 opsx）
|
||
// 密码 → admin token（如 opsxdb-demo-token）
|
||
// 库名 → bucket（如 demo）
|
||
//
|
||
// 能力：对象树 bucket → measurement（Flux schema.measurements）；
|
||
// SQL 编辑器直写 Flux（from(bucket: ...) |> range(...) ...），结果按注解 CSV 解析为表格。
|
||
// Flux 只读（写走 /write 专用通道），危险语句拦截在 ops.go 入口统一过一遍。
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
func isInflux(d string) bool { return driverKey(d) == "influxdb" }
|
||
|
||
func ifxBase(c ConnReq) string {
|
||
host := strings.TrimSpace(c.Host)
|
||
if host == "" {
|
||
host = "127.0.0.1"
|
||
}
|
||
port := c.Port
|
||
if port == 0 {
|
||
port = 8086
|
||
}
|
||
return fmt.Sprintf("http://%s:%d", host, port)
|
||
}
|
||
|
||
func ifxToken(c ConnReq) string { return c.Password }
|
||
|
||
// ifxDo 统一请求：Token 鉴权 + org 查询参数。
|
||
func ifxDo(ctx context.Context, c ConnReq, method, path string, body io.Reader, out any) error {
|
||
req, err := http.NewRequestWithContext(ctx, method, ifxBase(c)+path, body)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if ifxToken(c) != "" {
|
||
req.Header.Set("Authorization", "Token "+ifxToken(c))
|
||
}
|
||
if body != nil {
|
||
req.Header.Set("Content-Type", "application/json")
|
||
}
|
||
resp, err := (&http.Client{Timeout: 30 * time.Second}).Do(req)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode >= 400 {
|
||
msg, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
||
return fmt.Errorf("InfluxDB HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(msg)))
|
||
}
|
||
if out != nil {
|
||
return json.NewDecoder(resp.Body).Decode(out)
|
||
}
|
||
_, _ = io.Copy(io.Discard, resp.Body)
|
||
return nil
|
||
}
|
||
|
||
func ifxOrg(c ConnReq) (string, error) {
|
||
org := strings.TrimSpace(c.User)
|
||
if org == "" {
|
||
return "", fmt.Errorf("InfluxDB 2.x 需要填「用户」= org 名（如 opsx）")
|
||
}
|
||
return org, nil
|
||
}
|
||
|
||
func ifxPing(c ConnReq) error {
|
||
org, err := ifxOrg(c)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
|
||
defer cancel()
|
||
var out struct {
|
||
Buckets []struct {
|
||
Name string `json:"name"`
|
||
} `json:"buckets"`
|
||
}
|
||
return ifxDo(ctx, c, http.MethodGet, "/api/v2/buckets?org="+org, nil, &out)
|
||
}
|
||
|
||
func ifxListDatabases(ctx context.Context, c ConnReq) ([]string, error) {
|
||
org, err := ifxOrg(c)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var out struct {
|
||
Buckets []struct {
|
||
Name string `json:"name"`
|
||
} `json:"buckets"`
|
||
}
|
||
if err := ifxDo(ctx, c, http.MethodGet, "/api/v2/buckets?org="+org, nil, &out); err != nil {
|
||
return nil, err
|
||
}
|
||
names := []string{}
|
||
for _, b := range out.Buckets {
|
||
names = append(names, b.Name)
|
||
}
|
||
return names, nil
|
||
}
|
||
|
||
// ifxQueryFlux 跑 Flux 并把注解 CSV 解析为表格。
|
||
func ifxQueryFlux(ctx context.Context, c ConnReq, flux string, maxRows int) (ExecResult, error) {
|
||
org, err := ifxOrg(c)
|
||
if err != nil {
|
||
return ExecResult{}, err
|
||
}
|
||
body := map[string]string{"query": flux, "type": "flux"}
|
||
b, err := json.Marshal(body)
|
||
if err != nil {
|
||
return ExecResult{}, err
|
||
}
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, ifxBase(c)+"/api/v2/query?org="+org, strings.NewReader(string(b)))
|
||
if err != nil {
|
||
return ExecResult{}, err
|
||
}
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("Accept", "application/csv")
|
||
if ifxToken(c) != "" {
|
||
req.Header.Set("Authorization", "Token "+ifxToken(c))
|
||
}
|
||
resp, err := (&http.Client{Timeout: 60 * time.Second}).Do(req)
|
||
if err != nil {
|
||
return ExecResult{}, err
|
||
}
|
||
defer resp.Body.Close()
|
||
raw, err := io.ReadAll(resp.Body)
|
||
if err != nil {
|
||
return ExecResult{}, err
|
||
}
|
||
if resp.StatusCode >= 400 {
|
||
return ExecResult{}, fmt.Errorf("InfluxDB HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(raw)))
|
||
}
|
||
return parseAnnotatedCSV(string(raw), maxRows)
|
||
}
|
||
|
||
// parseAnnotatedCSV 解析 InfluxDB 注解 CSV：
|
||
// #group/#datatype/#default 注释行 + 空行 + header 行 + 数据行。
|
||
func parseAnnotatedCSV(text string, maxRows int) (ExecResult, error) {
|
||
lines := strings.Split(strings.ReplaceAll(text, "\r\n", "\n"), "\n")
|
||
headerIdx := -1
|
||
var header []string
|
||
for i, ln := range lines {
|
||
s := strings.TrimSpace(ln)
|
||
if s == "" || strings.HasPrefix(s, "#") {
|
||
continue
|
||
}
|
||
headerIdx = i
|
||
header = strings.Split(ln, ",")
|
||
break
|
||
}
|
||
if headerIdx < 0 {
|
||
return ExecResult{Columns: []string{}, Rows: []map[string]interface{}{}}, nil
|
||
}
|
||
res := ExecResult{Columns: header, Rows: []map[string]interface{}{}}
|
||
for _, ln := range lines[headerIdx+1:] {
|
||
if strings.TrimSpace(ln) == "" {
|
||
continue
|
||
}
|
||
cells := strings.Split(ln, ",")
|
||
row := make(map[string]interface{}, len(header))
|
||
for i, col := range header {
|
||
v := ""
|
||
if i < len(cells) {
|
||
v = cells[i]
|
||
}
|
||
row[col] = v
|
||
}
|
||
res.Rows = append(res.Rows, row)
|
||
if len(res.Rows) >= maxRows {
|
||
res.Truncated = true
|
||
break
|
||
}
|
||
}
|
||
res.RowCount = len(res.Rows)
|
||
return res, nil
|
||
}
|
||
|
||
func ifxListTables(ctx context.Context, c ConnReq) ([]string, error) {
|
||
bucket := strings.TrimSpace(c.Database)
|
||
if bucket == "" {
|
||
return nil, fmt.Errorf("InfluxDB 列表需要库名（bucket）")
|
||
}
|
||
flux := fmt.Sprintf("import \"influxdata/influxdb/schema\"\nschema.measurements(bucket: \"%s\")", strings.ReplaceAll(bucket, "\"", ""))
|
||
res, err := ifxQueryFlux(ctx, c, flux, 10000)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
out := []string{}
|
||
if len(res.Columns) == 0 {
|
||
return out, nil
|
||
}
|
||
// schema.measurements 输出列为 [,result,table,_value]：优先 _value 列，退化为最后一列
|
||
idx := len(res.Columns) - 1
|
||
for i, col := range res.Columns {
|
||
if col == "_value" {
|
||
idx = i
|
||
break
|
||
}
|
||
}
|
||
col := res.Columns[idx]
|
||
for _, row := range res.Rows {
|
||
if v, ok := row[col].(string); ok && v != "" {
|
||
out = append(out, v)
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// ifxExecSQL 编辑器内容按 Flux 处理（只读）。
|
||
func ifxExecSQL(ctx context.Context, c ConnReq, start time.Time) ExecResult {
|
||
if c.Offset > 0 {
|
||
return ExecResult{Columns: []string{}, Rows: []map[string]interface{}{}, Error: "InfluxDB Flux 结果暂不支持翻页（可缩小 range 重查）", TimeMs: time.Since(start).Milliseconds()}
|
||
}
|
||
limit := rowLimit(c)
|
||
if limit > MaxRowLimit {
|
||
limit = MaxRowLimit
|
||
}
|
||
res, err := ifxQueryFlux(ctx, c, strings.TrimSpace(c.SQL), limit)
|
||
if err != nil {
|
||
return execFail(start, err)
|
||
}
|
||
res.TimeMs = time.Since(start).Milliseconds()
|
||
return res
|
||
}
|
||
|
||
// ifxListColumns measurement 的字段探查：Flux schema.fieldKeys + tagKeys 合并。
|
||
func ifxListColumns(ctx context.Context, c ConnReq) ([]ColumnInfo, error) {
|
||
bucket := strings.TrimSpace(c.Database)
|
||
table := strings.TrimSpace(c.Table)
|
||
if bucket == "" || table == "" {
|
||
return nil, fmt.Errorf("InfluxDB 列探查需要库名与 measurement")
|
||
}
|
||
safeB := strings.ReplaceAll(bucket, "\"", "")
|
||
safeM := strings.ReplaceAll(table, "\"", "")
|
||
flux := fmt.Sprintf(`import "influxdata/influxdb/schema"
|
||
union(tables: [
|
||
schema.fieldKeys(bucket: "%s", predicate: (r) => r._measurement == "%s"),
|
||
schema.tagKeys(bucket: "%s", predicate: (r) => r._measurement == "%s")
|
||
])`, safeB, safeM, safeB, safeM)
|
||
res, err := ifxQueryFlux(ctx, c, flux, 1000)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
cols := []ColumnInfo{}
|
||
seen := map[string]bool{}
|
||
if len(res.Columns) == 0 {
|
||
return cols, nil
|
||
}
|
||
last := res.Columns[len(res.Columns)-1]
|
||
for _, row := range res.Rows {
|
||
name, _ := row[last].(string)
|
||
if name == "" || seen[name] {
|
||
continue
|
||
}
|
||
seen[name] = true
|
||
cols = append(cols, ColumnInfo{Name: name, Type: "tag/field", Nullable: true})
|
||
}
|
||
return cols, nil
|
||
}
|