// 日志数据源管理 — CRUD、默认源、部署自动注册(Managed 标记)。 package services import ( "encoding/json" "fmt" "regexp" "sort" "strings" "time" "opsxloki/models" "opsxloki/pkg/crypto" "opsxloki/repository" ) // ==================== 数据源 CRUD ==================== func GetLogSource(id uint) *models.LogSource { db := repository.GetDB() db.RLock() defer db.RUnlock() for i := range db.LogSources { if db.LogSources[i].ID == id { s := db.LogSources[i] return &s } } return nil } func ListLogSources() []map[string]interface{} { db := repository.GetDB() db.RLock() defer db.RUnlock() out := make([]map[string]interface{}, 0, len(db.LogSources)) for i := range db.LogSources { out = append(out, db.LogSources[i].Sanitized()) } return out } func ListVisibleLogSources(userID uint, admin bool) []map[string]interface{} { db := repository.GetDB() db.RLock() defer db.RUnlock() out := make([]map[string]interface{}, 0) for i := range db.LogSources { s := &db.LogSources[i] if admin || policyGroupAccess(db.LogPolicyGroups, s.ID, userID) != nil || sourceAccess(s, userID) != nil { item := s.Sanitized() if !admin { item = map[string]interface{}{"id": s.ID, "name": s.Name, "isDefault": s.IsDefault} } out = append(out, item) } } return out } func sourceAccess(src *models.LogSource, userID uint) *models.LogSourceAccess { for i := range src.Access { if src.Access[i].UserID == userID { return &src.Access[i] } } return nil } var labelNamePattern = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`) func ValidLogLabelName(name string) bool { return labelNamePattern.MatchString(name) } // AuthorizedLogQuery constrains the Loki stream selector server-side. Non-admins must be explicitly granted access. func AuthorizedLogQuery(src *models.LogSource, userID uint, admin bool, query string) (string, error) { if admin { return query, nil } db := repository.GetDB() var access *models.LogSourceAccess if db != nil { db.RLock() access = policyGroupAccess(db.LogPolicyGroups, src.ID, userID) db.RUnlock() } if access == nil { access = sourceAccess(src, userID) } if access == nil { return "", fmt.Errorf("无权查询此数据源") } q := strings.TrimSpace(query) if !strings.HasPrefix(q, "{") { return "", fmt.Errorf("普通用户查询必须以日志流选择器开头") } close := strings.IndexByte(q, '}') if close < 0 || strings.Contains(q[close+1:], "{") { return "", fmt.Errorf("查询只能包含一个日志流选择器") } if len(access.Labels) == 0 { return q, nil } keys := make([]string, 0, len(access.Labels)) for k := range access.Labels { if !labelNamePattern.MatchString(k) { return "", fmt.Errorf("访问策略包含无效标签名") } keys = append(keys, k) } sort.Strings(keys) matchers := make([]string, 0, len(keys)) for _, k := range keys { v, _ := json.Marshal(access.Labels[k]) matchers = append(matchers, k+"="+string(v)) } inner := strings.TrimSpace(q[1:close]) if inner != "" { inner += "," } return "{" + inner + strings.Join(matchers, ",") + "}" + q[close+1:], nil } func CreateLogSource(s *models.LogSource) (*models.LogSource, error) { if err := encryptSourceCredentials(s); err != nil { return nil, err } db := repository.GetDB() newID := db.NextID("logSources") db.Lock() defer db.Unlock() s.ID = newID if s.IsDefault { for i := range db.LogSources { db.LogSources[i].IsDefault = false } } // 首个数据源自动设默认 if len(db.LogSources) == 0 { s.IsDefault = true } db.LogSources = append(db.LogSources, *s) db.RequestSave() return s, nil } func encryptSourceCredentials(s *models.LogSource) error { if s.BasicPass != "" { v, err := crypto.EncryptField(s.BasicPass) if err != nil { return err } s.BasicPass = v } if s.BearerToken != "" { v, err := crypto.EncryptField(s.BearerToken) if err != nil { return err } s.BearerToken = v } for name, value := range s.Headers { if !validSourceHeader(name) { return fmt.Errorf("自定义 Header 名称无效: %s", name) } v, err := crypto.EncryptField(value) if err != nil { return err } s.Headers[name] = v } return nil } func validSourceHeader(name string) bool { if name == "" || strings.EqualFold(name, "Host") || strings.EqualFold(name, "Authorization") { return false } for _, c := range name { if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '-') { return false } } return true } func UpdateLogSource(id uint, patch map[string]interface{}) error { db := repository.GetDB() db.Lock() defer db.Unlock() for i := range db.LogSources { if db.LogSources[i].ID != id { continue } if v, ok := patch["name"].(string); ok && v != "" { db.LogSources[i].Name = v } if v, ok := patch["namespace"].(string); ok { db.LogSources[i].Namespace = v } if v, ok := patch["service"].(string); ok { db.LogSources[i].Service = v } if v, ok := patch["port"].(float64); ok { db.LogSources[i].Port = int(v) } if v, ok := patch["url"].(string); ok { db.LogSources[i].URL = v } if v, ok := patch["authType"].(string); ok { db.LogSources[i].AuthType = v } if v, ok := patch["basicUser"].(string); ok { db.LogSources[i].BasicUser = v } if v, ok := patch["bearerToken"].(string); ok { if v != "" { enc, err := crypto.EncryptField(v) if err != nil { return err } db.LogSources[i].BearerToken = enc } } if v, ok := patch["skipTLSVerify"].(bool); ok { db.LogSources[i].SkipTLSVerify = v } if raw, ok := patch["headers"]; ok { data, err := json.Marshal(raw) if err != nil { return err } var headers map[string]string if err := json.Unmarshal(data, &headers); err != nil { return err } for name, value := range headers { if !validSourceHeader(name) { return fmt.Errorf("自定义 Header 名称无效: %s", name) } if value == "" { if old, ok := db.LogSources[i].Headers[name]; ok { headers[name] = old } continue } enc, err := crypto.EncryptField(value) if err != nil { return err } headers[name] = enc } db.LogSources[i].Headers = headers } if v, ok := patch["basicPass"].(string); ok && v != "" { if enc, err := crypto.EncryptField(v); err == nil { db.LogSources[i].BasicPass = enc } } if v, ok := patch["isDefault"].(bool); ok && v { for j := range db.LogSources { db.LogSources[j].IsDefault = db.LogSources[j].ID == id } } if raw, ok := patch["access"]; ok { data, err := json.Marshal(raw) if err != nil { return err } var access []models.LogSourceAccess if err := json.Unmarshal(data, &access); err != nil { return err } if err := ValidateSourceAccess(access); err != nil { return err } db.LogSources[i].Access = access } db.LogSources[i].UpdatedAt = time.Now() db.RequestSave() return nil } return fmt.Errorf("数据源不存在") } func ValidateSourceAccess(access []models.LogSourceAccess) error { seen := map[uint]bool{} for _, rule := range access { if rule.UserID == 0 || seen[rule.UserID] { return fmt.Errorf("访问策略的用户 ID 无效或重复") } seen[rule.UserID] = true for k := range rule.Labels { if !labelNamePattern.MatchString(k) { return fmt.Errorf("访问策略包含无效标签名: %s", k) } } } return nil } func DeleteLogSource(id uint) error { db := repository.GetDB() db.Lock() defer db.Unlock() for i := range db.LogSources { if db.LogSources[i].ID == id { db.LogSources = append(db.LogSources[:i], db.LogSources[i+1:]...) groups := db.LogPolicyGroups[:0] for _, group := range db.LogPolicyGroups { if group.SourceID != id { groups = append(groups, group) } } db.LogPolicyGroups = groups db.RequestSave() return nil } } return fmt.Errorf("数据源不存在") } // EnsureLogSourceForDeployment 部署成功后自动注册查询数据源(host→url;k8s→cluster-svc)。 // 已存在同名数据源时:平台托管的自动刷新连接信息,用户手填的不覆盖。 func EnsureLogSourceForDeployment(d *models.LokiDeployment) { if d == nil || d.Endpoint == "" { return } svc := "loki" port := 3100 target := &models.LogSource{ Name: d.Name, TargetType: "cluster-svc", ClusterID: d.ClusterID, Namespace: d.Namespace, Service: svc, Port: port, AuthType: "none", Managed: true, } if d.Mode == "host-binary" || d.Mode == "host-docker" { target.TargetType = "url" target.URL = d.Endpoint target.ClusterID = nil } db := repository.GetDB() db.Lock() for i := range db.LogSources { if db.LogSources[i].Name != target.Name { continue } if !db.LogSources[i].Managed { db.Unlock() return } db.LogSources[i].TargetType = target.TargetType db.LogSources[i].ClusterID = target.ClusterID db.LogSources[i].Namespace = target.Namespace db.LogSources[i].Service = target.Service db.LogSources[i].Port = target.Port db.LogSources[i].URL = target.URL db.LogSources[i].UpdatedAt = time.Now() db.Unlock() db.RequestSave() return } db.Unlock() _, _ = CreateLogSource(target) } // EnsureLogSourcesForRunningDeployments 补齐运行中部署的数据源,避免查询页默认没有数据源。 func EnsureLogSourcesForRunningDeployments() { db := repository.GetDB() db.RLock() running := make([]models.LokiDeployment, 0, len(db.Deployments)) for i := range db.Deployments { d := db.Deployments[i] if d.Status == "running" && d.Endpoint != "" { running = append(running, d) } } db.RUnlock() for i := range running { EnsureLogSourceForDeployment(&running[i]) } }