LokiStack/backend/services/log_source.go

// 日志数据源管理 — CRUD、默认源、部署自动注册(Managed 标记)。
package services

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"regexp"
	"sort"
	"strings"
	"time"

	"lokistack/models"
	"lokistack/pkg/crypto"
	"lokistack/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
}

// UserAccessLabels 返回某用户在指定数据源上的授权 labels。
//
// 返回值语义：
//   - nil      → 该用户对此数据源完全无权限
//   - 空 map   → 有权限但不限定范围（可查全部）
//   - 非空 map → 只能查匹配这些 label 的日志（如 {"namespace":"astralx"}）
func UserAccessLabels(src *models.LogSource, userID uint) map[string]string {
	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 nil
	}
	if access.Labels == nil {
		return map[string]string{}
	}
	return access.Labels
}

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) {
	defer InvalidateLabelCache()
	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 {
	defer InvalidateLabelCache()
	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 {
	defer InvalidateLabelCache()
	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("数据源不存在")
}

// reachableEndpoint 把部署记录里的访问地址转换成「平台侧能真正连上」的地址。
//
// 主机模式创建部署时存的是容器的监听地址（0.0.0.0:3100 / [::]:3100），
// 那是容器内部视角，平台进程在别的机器/集群里，用它必然 connection refused。
// 换成该主机的真实 IP 再返回；主机信息缺失时原样返回（至少不会更差）。
func reachableEndpoint(d *models.LokiDeployment) string {
	endpoint := strings.TrimSpace(d.Endpoint)
	if endpoint == "" {
		return endpoint
	}
	if !strings.Contains(endpoint, "0.0.0.0") && !strings.Contains(endpoint, "[::]") {
		return endpoint
	}
	hostIP := ""
	if d.Host != nil {
		hostIP = strings.TrimSpace(d.Host.HostIP)
	}
	if hostIP == "" {
		// 没带 Host 关联时回库查一次，避免调用方漏预加载
		db := repository.GetDB()
		if db != nil && d.HostID != nil {
			db.RLock()
			for i := range db.Hosts {
				if db.Hosts[i].ID == *d.HostID {
					hostIP = strings.TrimSpace(db.Hosts[i].HostIP)
					break
				}
			}
			db.RUnlock()
		}
	}
	if hostIP == "" {
		return endpoint
	}
	endpoint = strings.Replace(endpoint, "0.0.0.0", hostIP, 1)
	endpoint = strings.Replace(endpoint, "[::]", hostIP, 1)
	return endpoint
}

// CleanupManagedLogSource 删除某部署名对应的**平台托管**数据源。
//
// 为什么必须有：部署删除后它自动注册的数据源若不清掉，「日志查询」下拉里会
// 留一个指向已删实例的源，选中必然连接失败（线上残留过 host-verify-loki）。
//
// 连带处理：
//   - 该源上的查询授权策略组：DeleteLogSource 内部会一并删除
//   - 绑定了该源的告警规则：置为禁用（保留配置便于审计，但不再触发，
//     否则引擎每次评估都会因为连不上而写失败记录）
//   - 手填的数据源（Managed=false）不动：那是用户自己建的，可能指向别的实例
func CleanupManagedLogSource(name string) {
	db := repository.GetDB()
	if db == nil || name == "" {
		return
	}
	db.RLock()
	var sourceID uint
	for _, src := range db.LogSources {
		if src.Managed && src.Name == name {
			sourceID = src.ID
			break
		}
	}
	db.RUnlock()
	if sourceID == 0 {
		// 部署本来就没注册数据源（如部署失败），无需处理
		return
	}

	if err := DeleteLogSource(sourceID); err != nil {
		log.Printf("[deploy-delete] 清理数据源 %q(id=%d) 失败: %v", name, sourceID, err)
	}

	db.Lock()
	defer db.Unlock()
	disabled := 0
	for i := range db.AlertRules {
		if db.AlertRules[i].SourceID == sourceID && db.AlertRules[i].Enabled {
			db.AlertRules[i].Enabled = false
			disabled++
		}
	}
	if disabled > 0 {
		db.RequestSave()
		log.Printf("[deploy-delete] %d 条告警规则因数据源被删而禁用", disabled)
	}
}

// CleanupOrphanManagedLogSources 启动时清理孤儿托管数据源：
// 部署记录已删除、或所属集群已删除的 managed 源。
//
// 为什么需要：CleanupManagedLogSource 只在「删除部署」的代码路径里执行；
// 如果部署是在旧版本（没有清理逻辑）删掉的，或删除过程中进程退出，
// 托管源就会永久残留，在数据源列表里指向一个不存在的实例。
func CleanupOrphanManagedLogSources() {
	db := repository.GetDB()
	if db == nil {
		return
	}
	db.Lock()
	names := make(map[string]bool, len(db.Deployments))
	for _, d := range db.Deployments {
		names[d.Name] = true
	}
	clusters := make(map[uint]bool, len(db.Clusters))
	for _, c := range db.Clusters {
		clusters[c.ID] = true
	}
	kept := make([]models.LogSource, 0, len(db.LogSources))
	removed, removedDefault := 0, false
	for _, s := range db.LogSources {
		orphan := false
		if s.Managed {
			if !names[s.Name] {
				orphan = true
			}
			if s.ClusterID != nil && !clusters[*s.ClusterID] {
				orphan = true
			}
		}
		if orphan {
			removed++
			if s.IsDefault {
				removedDefault = true
			}
			continue
		}
		kept = append(kept, s)
	}
	if removed > 0 {
		if removedDefault && len(kept) > 0 {
			kept[0].IsDefault = true
		}
		db.LogSources = kept
		db.RequestSave()
		log.Printf("[startup] 清理 %d 个孤儿托管数据源", removed)
	}
	db.Unlock()
}

// AutoDetectLogSource 探测数据源的可用连接方式，返回最终选择（cluster-svc / url）。
//
// 交付环境组合很多：平台可能在主机或集群内，Loki 也可能在主机或集群内。
// cluster-svc 走 kube-apiserver 的 service proxy，不依赖集群 DNS；url 直连
// 则需要平台能访问到该地址。这里按「代理优先、URL 兜底」实测，避免按部署
// 模式硬编码导致某种组合连不上。
func AutoDetectLogSource(ctx context.Context, src *models.LogSource) string {
	if src.ClusterID != nil {
		probe := *src
		probe.TargetType = "cluster-svc"
		if _, status, err := lokiProxyGET(ctx, &probe, "/ready", nil); err == nil && status == 200 {
			src.TargetType = "cluster-svc"
			return "cluster-svc"
		}
	}
	if src.URL != "" {
		probe := *src
		probe.TargetType = "url"
		if _, status, err := lokiProxyGET(ctx, &probe, "/ready", nil); err == nil && status == 200 {
			src.TargetType = "url"
			return "url"
		}
	}
	return ""
}

// EnsureLogSourceForDeployment 部署成功后自动注册查询数据源(host→url;k8s→cluster-svc)。
// 已存在同名数据源时：平台托管的自动刷新连接信息，用户手填的不覆盖。
func EnsureLogSourceForDeployment(d *models.LokiDeployment) {
	if d == nil || d.Endpoint == "" {
		return
	}
	// 用户主动删过这个自动数据源 → 不再补回来（重新部署/编辑部署会重置该开关）
	if d.AutoSourceDisabled {
		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"
		// 主机模式部署的 endpoint 记的是**容器监听地址**（0.0.0.0:3100），
		// 直接拿来当数据源 URL 会得到 dial tcp 0.0.0.0:3100: connection refused ——
		// 平台进程不在那台主机上，0.0.0.0 只在本机有意义。
		// 这里换成主机真实 IP，平台才能连上。
		target.URL = reachableEndpoint(d)
		target.ClusterID = nil
	} else {
		// k8s 部署：集群内 Service 地址也记下来，作为 URL 直连候选
		target.URL = d.Endpoint
	}

	// 实测可用性后决定连接方式：主机/集群内平台、主机/集群内 Loki 都兼容
	detectCtx, cancel := context.WithTimeout(context.Background(), 6*time.Second)
	if method := AutoDetectLogSource(detectCtx, target); method != "" {
		log.Printf("[log-source] %q 自动选择连接方式: %s", target.Name, method)
	}
	cancel()

	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)
}

// DisableAutoSourceFor 标记某个部署不再自动注册数据源。
//
// 触发场景：用户在「日志查询 → 数据源」里删掉了平台自动创建的数据源。
// 不置位的话，EnsureLogSourcesForRunningDeployments 会立刻把它补回来。
func DisableAutoSourceFor(name string) {
	db := repository.GetDB()
	if db == nil {
		return
	}
	db.Lock()
	defer db.Unlock()
	for i := range db.Deployments {
		if db.Deployments[i].Name == name && !db.Deployments[i].AutoSourceDisabled {
			db.Deployments[i].AutoSourceDisabled = true
			log.Printf("[log-source] %q 的自动数据源已被用户删除，不再自动重建（重新部署可恢复）", name)
			db.RequestSave()
		}
	}
}

// EnableAutoSourceFor 重新允许自动注册（部署被编辑/重新部署时调用）。
func EnableAutoSourceFor(name string) {
	db := repository.GetDB()
	if db == nil {
		return
	}
	db.Lock()
	defer db.Unlock()
	for i := range db.Deployments {
		if db.Deployments[i].Name == name && db.Deployments[i].AutoSourceDisabled {
			db.Deployments[i].AutoSourceDisabled = false
			db.RequestSave()
		}
	}
}

// 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])
	}
}