589 lines
16 KiB
Go
589 lines
16 KiB
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) {
|
||
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("数据源不存在")
|
||
}
|
||
|
||
// 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
|
||
}
|
||
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)
|
||
}
|
||
|
||
// 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])
|
||
}
|
||
}
|