package controller import ( "crypto/tls" "encoding/base64" "encoding/json" "errors" "io" "log" "net/http" "strings" "time" "ezops/model" "github.com/gin-gonic/gin" "gorm.io/gorm" ) // LogDSHandler 日志数据源管理 // Logan.Li type LogDSHandler struct { db *gorm.DB } func NewLogDSHandler(db *gorm.DB) *LogDSHandler { return &LogDSHandler{db: db} } // List GET /api/v1/log-datasources?type=loki func (h *LogDSHandler) List(c *gin.Context) { q := h.db.Order("is_default DESC, id ASC") if t := c.Query("type"); t != "" { q = q.Where("type = ?", t) } var sources []model.LogDataSource if err := q.Find(&sources).Error; err != nil { c.JSON(500, model.Response{Code: 500, Message: err.Error()}) return } if sources == nil { sources = []model.LogDataSource{} } // 按项目惯例返回 {list,total},前端 listDatasources 取 .list // Logan.Li c.JSON(200, model.Response{Code: 200, Data: model.ListResponse{List: sources, Total: int64(len(sources))}}) } // Get GET /api/v1/log-datasources/:id func (h *LogDSHandler) Get(c *gin.Context) { var ds model.LogDataSource if err := h.db.First(&ds, c.Param("id")).Error; err != nil { c.JSON(404, model.Response{Code: 404, Message: "数据源不存在"}) return } c.JSON(200, model.Response{Code: 200, Data: ds}) } // Create POST /api/v1/log-datasources func (h *LogDSHandler) Create(c *gin.Context) { var req model.CreateLogDSReq if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, model.Response{Code: 400, Message: err.Error()}) return } req.Type = strings.ToLower(req.Type) if req.Type != "loki" && req.Type != "elasticsearch" { c.JSON(400, model.Response{Code: 400, Message: "type 必须为 loki 或 elasticsearch"}) return } if req.IsDefault { h.db.Model(&model.LogDataSource{}).Where("type = ? AND is_default = ?", req.Type, true).Update("is_default", false) } 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 { // 记录详细错误,便于排查字段缺失/唯一约束等数据库问题 // Logan.Li log.Printf("[LogDataSource] 创建数据源失败: %v", err) 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}) } // Update PUT /api/v1/log-datasources/:id func (h *LogDSHandler) Update(c *gin.Context) { var ds model.LogDataSource if err := h.db.First(&ds, c.Param("id")).Error; err != nil { c.JSON(404, model.Response{Code: 404, Message: "数据源不存在"}) return } var req model.UpdateLogDSReq if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, model.Response{Code: 400, Message: err.Error()}) return } updates := map[string]interface{}{} if req.Name != nil { updates["name"] = *req.Name } if req.Type != nil { updates["type"] = strings.ToLower(*req.Type) } 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 } if req.BasicAuthUser != nil { updates["basic_auth_user"] = *req.BasicAuthUser } 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 } if req.DefaultIndex != nil { updates["default_index"] = *req.DefaultIndex } if req.Status != nil { updates["status"] = *req.Status } if req.IsDefault != nil && *req.IsDefault { h.db.Model(&model.LogDataSource{}).Where("type = ? AND is_default = ?", ds.Type, true).Update("is_default", false) updates["is_default"] = true } else if req.IsDefault != nil { updates["is_default"] = false } if err := h.db.Model(&ds).Updates(updates).Error; err != nil { 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) c.JSON(200, model.Response{Code: 200, Message: "更新成功", Data: ds}) } // Delete DELETE /api/v1/log-datasources/:id func (h *LogDSHandler) Delete(c *gin.Context) { if err := h.db.Delete(&model.LogDataSource{}, c.Param("id")).Error; err != nil { c.JSON(500, model.Response{Code: 500, Message: "删除失败"}) return } c.JSON(200, model.Response{Code: 200, Message: "已删除"}) } // SetDefault POST /api/v1/log-datasources/:id/default func (h *LogDSHandler) SetDefault(c *gin.Context) { var ds model.LogDataSource if err := h.db.First(&ds, c.Param("id")).Error; err != nil { c.JSON(404, model.Response{Code: 404, Message: "数据源不存在"}) return } h.db.Model(&model.LogDataSource{}).Where("type = ? AND is_default = ?", ds.Type, true).Update("is_default", false) h.db.Model(&ds).Update("is_default", true) c.JSON(200, model.Response{Code: 200, Message: "已设为默认"}) } // TestConnection POST /api/v1/log-datasources/test // 若请求体携带 datasource_id,则读取数据库完整配置进行测试,避免前端泄露/丢失密码。// Logan.Li func (h *LogDSHandler) TestConnection(c *gin.Context) { var req model.TestLogDSReq if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, model.Response{Code: 400, Message: err.Error()}) return } // 已保存数据源优先按 ID 读取完整配置 // Logan.Li if req.DatasourceID > 0 { ds, err := loadLogDS(h, req.DatasourceID) if err != nil { c.JSON(404, model.Response{Code: 404, Message: "数据源不存在"}) return } 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) 至少提供一组"}) return } ok, msg := testLogDSConnection(req) if !ok { c.JSON(200, model.Response{Code: 200, Message: msg, Data: map[string]interface{}{"status": "failed", "error": msg}}) return } c.JSON(200, model.Response{Code: 200, Message: "连接成功", Data: map[string]string{"status": "ok"}}) } // ListIndices GET /api/v1/log-datasources/indices?url=...&type=elasticsearch // 通过代理查询 ES 集群的索引列表 // Logan.Li func (h *LogDSHandler) ListIndices(c *gin.Context) { esURL := c.Query("url") if esURL == "" { c.JSON(400, model.Response{Code: 400, Message: "url 参数必填"}) return } 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(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 } applyLogDSAuthV2(httpReq, &tempDS) resp, err := client.Do(httpReq) if err != nil { c.JSON(502, model.Response{Code: 502, Message: "查询索引失败: " + err.Error()}) return } defer resp.Body.Close() body, _ := io.ReadAll(resp.Body) if resp.StatusCode >= 400 { 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{} if err := json.Unmarshal(body, &raw); err != nil { c.JSON(502, model.Response{Code: 502, Message: "解析索引列表失败"}) return } indices := make([]string, 0, len(raw)) for _, entry := range raw { if idx, ok := entry["index"].(string); ok { indices = append(indices, idx) } } c.JSON(200, model.Response{Code: 200, Data: indices}) } func testLogDSConnection(req model.TestLogDSReq) (bool, string) { client := newLogHTTPClient(req.SkipTLSVerify, 10*time.Second) url := strings.TrimRight(req.URL, "/") var testURL string switch strings.ToLower(req.Type) { case "loki": testURL = url + "/loki/api/v1/labels" case "elasticsearch": testURL = url + "/_cluster/health" default: return false, "未知数据源类型" } httpReq, err := http.NewRequest(http.MethodGet, testURL, nil) if err != nil { return false, err.Error() } // 用临时 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() } defer resp.Body.Close() if resp.StatusCode >= 400 { body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) return false, "HTTP " + resp.Status + ": " + string(body) } return true, "" } func loadLogDS(h *LogDSHandler, id uint) (*model.LogDataSource, error) { var ds model.LogDataSource if err := h.db.First(&ds, id).Error; err != nil { return nil, err } return &ds, nil } func newLogHTTPClient(skipTLS bool, timeout time.Duration) *http.Client { tr := &http.Transport{} if skipTLS { tr.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} //nolint:gosec } return &http.Client{Timeout: timeout, Transport: tr} } 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" } }