opsx
5e59347c40
=== 0. 【严重】修复上一版引入的死锁（平台写路径全卡死）===
上一版把保留期读取写成 `GetRetentionSettings()`（内部 db.RLock()），
却在 `db.Lock()` 已持有的情况下调用（recordDelivery / RecordAudit 的裁剪路径）。
Go 的 sync.RWMutex **不可重入** → 第一次写投递历史/审计日志就永久死锁。
现象：登录 28 秒返回空（登录也要写审计）、所有写接口挂住。
go build / go vet / 之前的单测（不触发写路径）全都查不出来。
修法：
- 拆出 retentionFromDBLocked(db) / alertDeliveryRetentionLocked(db) /
auditLogRetentionLocked(db)，调用方已持锁时用无锁版本
- upsertSettingLocked 里**不能**用 db.NextID()（内部 d.mu.Lock()，同样死锁），
改为锁内自算 max(id)+1
- 新增 retention_test.go：持写锁调用裁剪函数，2 秒超时未返回即判死锁
实测修复后：登录 103ms（原 28000ms）、panic/deadlock 命中 0。
=== 1. 平台数据保留期可配（系统设置 → 通用配置 → 日志保留时间）===
新增 GET/PUT /api/v1/settings/retention：
alertDeliveryDays 告警发送历史保留天数，默认 7（原写死"最近 50 条"）
auditLogDays 审计日志保留天数，默认 90（原写死常量）
保存后立即按新策略裁剪一次，不必等下次触发。
校验：发送历史 1~365 天、审计 1~3650 天，超出返回明确中文原因（不是 500）。
注意与「Loki 日志保留期」区分：那个是部署级配置（组件部署 → 保留策略）。
=== 2. 发送历史：50 条硬上限 → 保留 7 天 ===
后端 Deliveries 接口原来写死 `len(out) < 50`，现在按保留期取（响应上限 500 可调），
标题改为「发送历史（保留 N 天）」，N 从设置里读。
=== 3. 菜单「日志服务」→「组件部署」===
=== 4. 分析页：去掉「全部命名空间」+ 改手动刷新 ===
- 移除「全部命名空间」选项：{namespace=~".+"} 要扫全集群（约 370GB/天），
一次分析几十秒，得不偿失；历史值 __all__ 也会被忽略并回退到策略/第一个 ns
- 筛选改为「选择值 / 已生效值」分离：切换命名空间或时间范围**不再自动查询**，
只把刷新按钮点亮并提示「筛选条件已改，点刷新生效」；
点一次才应用并取数。
原因：一次分析要跑多条 count_over_time，切一下下拉就重查一轮，
快速连点会把 Loki 打满（并发护栏 429），还拖慢别人。
=== 实测（9739c7c092e7）===
登录 103ms / token 273 字符
GET retention → {alertDeliveryDays:7, auditLogDays:90}
PUT 0 天 → 400「告警发送历史保留天数需在 1~365 之间」
平台日志 panic/deadlock 命中 0
go test ./... 全绿（含 2 个新死锁回归用例）
468 lines
14 KiB
Go
468 lines
14 KiB
Go
// LokiDeployer — loki.yaml 配置生成:模板 + 对象存储/规则存储/ring 拓扑。
|
||
package services
|
||
|
||
import (
|
||
"bytes"
|
||
"fmt"
|
||
"os"
|
||
"strconv"
|
||
"strings"
|
||
"text/template"
|
||
|
||
"lokistack/models"
|
||
"lokistack/pkg/crypto"
|
||
)
|
||
|
||
const lokiConfigTemplate = `auth_enabled: {{ .AuthEnabled }}
|
||
|
||
common:
|
||
path_prefix: /var/lib/loki
|
||
{{- if .FilesystemStorage }}
|
||
storage:
|
||
filesystem:
|
||
chunks_directory: /var/lib/loki/chunks
|
||
rules_directory: /var/lib/loki/rules
|
||
{{- end }}
|
||
replication_factor: {{ .ReplicationFactor }}
|
||
ring:
|
||
instance_addr: {{ .RingInstanceAddr }}
|
||
kvstore:
|
||
store: {{ .RingKVStore }}
|
||
|
||
{{- if .StorageConfigBlock }}
|
||
|
||
storage_config:
|
||
{{ .StorageConfigBlock }}
|
||
{{- end }}
|
||
{{- if .MemberlistBlock }}
|
||
|
||
{{ .MemberlistBlock }}
|
||
{{- end }}
|
||
|
||
ingester:
|
||
wal:
|
||
dir: /var/lib/loki/wal
|
||
flush_on_shutdown: true
|
||
replay_memory_ceiling: 1GB
|
||
lifecycler:
|
||
ring:
|
||
kvstore:
|
||
store: {{ .RingKVStore }}
|
||
replication_factor: {{ .ReplicationFactor }}
|
||
{{- if .MultiReplica }}
|
||
autoforget_unhealthy: true
|
||
{{- end }}
|
||
|
||
compactor:
|
||
working_directory: /var/lib/loki/compactor
|
||
retention_enabled: true
|
||
delete_request_store: {{ .CompactorStoreType }}
|
||
# 过期 chunk 的删除并发。Loki 默认 150，但各云对象存储的 QPS 限制差别很大，
|
||
# 打太猛会被限流（腾讯云 COS 实测返回 503 SlowDown，整轮 apply_retention 失败，
|
||
# 残留 21,354 个对象 / 24.8 GB 删不掉）。该值按 provider 取值，可用
|
||
# 环境变量 LOKI_COMPACTOR_DELETE_WORKERS 覆盖。
|
||
retention_delete_worker_count: {{ .CompactorDeleteWorkers }}
|
||
# 过期 chunk 的「标记 → 真正删除」间隔。默认 2h 会让对象存储里
|
||
# 多留 2 小时的数据（实测桶内超过保留期的对象一度占到 24.6 GB）。
|
||
# 预发环境取 10m：既保留短暂的安全窗口，又不会长期占用存储成本。
|
||
retention_delete_delay: 10m
|
||
|
||
limits_config:
|
||
retention_period: {{ .RetentionPeriod }}h
|
||
# 单实例 Loki 的并发护栏：默认 32 高得离谱，一条重查询就能把单进程打满
|
||
max_query_parallelism: 8
|
||
|
||
# ── 查询结果缓存（官方推荐；此前完全没配，重复查询要重算一遍）───────
|
||
# align_queries_with_step 让不同时间范围的同类查询对齐到同一 step，
|
||
# 显著提高缓存命中率；命中后同样的"日志量统计"直接从缓存返回。
|
||
query_range:
|
||
align_queries_with_step: true
|
||
cache_results: true
|
||
results_cache:
|
||
cache:
|
||
embedded_cache:
|
||
enabled: true
|
||
max_size_mb: 100
|
||
|
||
schema_config:
|
||
configs:
|
||
- from: "2024-01-01"
|
||
store: tsdb
|
||
object_store: {{ .ObjectStore }}
|
||
schema: v13
|
||
index:
|
||
prefix: tsdb_index_
|
||
period: 24h
|
||
|
||
ruler:
|
||
storage:
|
||
type: {{ .RulerStoreType }}
|
||
{{- if .RulerStorageBlock }}
|
||
{{ .RulerStorageBlock }}
|
||
{{- end }}
|
||
{{- if eq .RulerStoreType "local" }}
|
||
local:
|
||
directory: /var/lib/loki/rules
|
||
{{- end }}
|
||
rule_path: /var/lib/loki/rules-temp
|
||
ring:
|
||
kvstore:
|
||
store: {{ .RingKVStore }}
|
||
|
||
analytics:
|
||
reporting_enabled: false
|
||
`
|
||
|
||
type lokiConfigData struct {
|
||
AuthEnabled bool
|
||
ReplicationFactor int
|
||
RetentionPeriod int
|
||
StorageConfigBlock string
|
||
ObjectStore string
|
||
RulerStoreType string
|
||
CompactorStoreType string
|
||
// CompactorDeleteWorkers 过期 chunk 删除并发，按对象存储类型取默认值
|
||
CompactorDeleteWorkers int
|
||
RingInstanceAddr string // 单副本 127.0.0.1；多副本 0.0.0.0（须可被对端寻址）
|
||
RingKVStore string // 单副本 inmemory；多副本 memberlist
|
||
MemberlistBlock string // 多副本时非空，注入 memberlist 顶级配置
|
||
RulerStorageBlock string // 对象存储时非空，注入 ruler.storage.s3/gcs/azure 嵌套配置
|
||
MultiReplica bool
|
||
FilesystemStorage bool
|
||
}
|
||
|
||
func (d *LokiDeployer) generateLokiConfig() string {
|
||
retentionHours := d.deployment.RetentionDays * 24
|
||
if retentionHours <= 0 {
|
||
retentionHours = DefaultRetentionDays * 24
|
||
}
|
||
|
||
storageType := d.effectiveStorageType()
|
||
storageBlock := d.buildStorageConfigBlock()
|
||
objectStore := "filesystem"
|
||
rulerStore := "local"
|
||
compactorStore := "filesystem"
|
||
|
||
switch storageType {
|
||
case "s3", "oss", "minio", "cos", "obs":
|
||
objectStore = "s3"
|
||
rulerStore = "s3"
|
||
// 删除请求存本地盘，避免各云 S3 兼容实现差异导致 compactor 启动失败
|
||
compactorStore = "filesystem"
|
||
case "gcs":
|
||
objectStore = "gcs"
|
||
rulerStore = "gcs"
|
||
compactorStore = "gcs"
|
||
case "azure":
|
||
objectStore = "azure"
|
||
rulerStore = "azure"
|
||
compactorStore = "azure"
|
||
}
|
||
|
||
// Ring 拓扑：单副本用 inmemory + 127.0.0.1（无需跨实例通信）；
|
||
// 多副本必须切 memberlist + 0.0.0.0，否则 ring 无法收敛、写入失败。
|
||
ringAddr := "127.0.0.1"
|
||
ringStore := "inmemory"
|
||
memberlistBlock := ""
|
||
if d.deployment.Replicas >= 2 {
|
||
ringAddr = "0.0.0.0"
|
||
ringStore = "memberlist"
|
||
ns := d.deployment.Namespace
|
||
if ns == "" {
|
||
ns = "log"
|
||
}
|
||
memberlistBlock = fmt.Sprintf(`memberlist:
|
||
join_members:
|
||
- loki-headless.%s.svc.cluster.local
|
||
bind_port: 7946
|
||
dead_node_reclaim_time: 30s
|
||
gossip_to_dead_nodes_time: 2m
|
||
left_ingesters_timeout: 5m`, ns)
|
||
}
|
||
|
||
data := lokiConfigData{
|
||
AuthEnabled: d.deployment.AuthEnabled,
|
||
ReplicationFactor: d.replicationFactor(),
|
||
RetentionPeriod: retentionHours,
|
||
StorageConfigBlock: storageBlock,
|
||
ObjectStore: objectStore,
|
||
RulerStoreType: rulerStore,
|
||
CompactorStoreType: compactorStore,
|
||
CompactorDeleteWorkers: compactorDeleteWorkers(d.deployment.StorageConfig),
|
||
RingInstanceAddr: ringAddr,
|
||
RingKVStore: ringStore,
|
||
MemberlistBlock: memberlistBlock,
|
||
RulerStorageBlock: d.buildRulerStorageBlock(),
|
||
MultiReplica: d.deployment.Replicas >= 2,
|
||
FilesystemStorage: objectStore == "filesystem",
|
||
}
|
||
|
||
tmpl, err := template.New("loki-config").Parse(lokiConfigTemplate)
|
||
if err != nil {
|
||
return d.generateLokiConfigFallback(retentionHours)
|
||
}
|
||
|
||
var buf bytes.Buffer
|
||
if err := tmpl.Execute(&buf, data); err != nil {
|
||
return d.generateLokiConfigFallback(retentionHours)
|
||
}
|
||
return buf.String()
|
||
}
|
||
|
||
// defaultCompactorDeleteWorkers 是 Loki 自身的默认值。
|
||
const defaultCompactorDeleteWorkers = 150
|
||
|
||
// compactorDeleteWorkers 按对象存储类型给出「过期 chunk 删除并发」。
|
||
//
|
||
// 为什么不能写死一个值：这个并发直接决定 compactor 对对象存储发出的 DELETE 速率，
|
||
// 而各云的限流差别很大。打满会被限流并让整轮 apply_retention 失败 ——
|
||
// 腾讯云 COS 实测默认 150 就返回 503 SlowDown，过期 chunk 残留 21,354 个 / 24.8 GB
|
||
// 删不掉；降到 16 后删除稳定推进。
|
||
//
|
||
// cos 16 实测 150 触发 503 SlowDown
|
||
// obs 32 华为 OBS 未给明确单桶 QPS，取保守值
|
||
// minio 32 自建，取决于后端磁盘 IO，别打满
|
||
// azure 32 存储账户限流较严
|
||
// oss 64 阿里云 OSS 单桶 QPS 万级，留余量
|
||
// gcs 64 GCS 高 QPS，但有 per-bucket 操作速率
|
||
// s3 96 AWS S3 单前缀约 3,500 DELETE/s，150 也能跑，留余量
|
||
//
|
||
// 任意 provider 都可用环境变量 LOKI_COMPACTOR_DELETE_WORKERS 覆盖（运维手调无需改代码）。
|
||
func compactorDeleteWorkers(sc *models.StorageConfig) int {
|
||
if raw := strings.TrimSpace(os.Getenv("LOKI_COMPACTOR_DELETE_WORKERS")); raw != "" {
|
||
if n, err := strconv.Atoi(raw); err == nil && n > 0 {
|
||
return n
|
||
}
|
||
}
|
||
if sc == nil {
|
||
return defaultCompactorDeleteWorkers
|
||
}
|
||
switch strings.ToLower(strings.TrimSpace(sc.StorageType)) {
|
||
case "cos":
|
||
return 16
|
||
case "obs", "minio", "azure":
|
||
return 32
|
||
case "oss", "gcs":
|
||
return 64
|
||
case "s3":
|
||
return 96
|
||
default:
|
||
return defaultCompactorDeleteWorkers
|
||
}
|
||
}
|
||
|
||
func (d *LokiDeployer) buildStorageConfigBlock() string {
|
||
if d.deployment.StorageConfig == nil {
|
||
return ""
|
||
}
|
||
|
||
sc := d.deployment.StorageConfig
|
||
accessKey, secretKey := d.storageCredentialValues()
|
||
switch sc.StorageType {
|
||
case "s3":
|
||
endpoint := sc.Endpoint
|
||
if endpoint == "" {
|
||
endpoint = fmt.Sprintf("s3.%s.amazonaws.com", sc.Region)
|
||
}
|
||
insecure := "false"
|
||
if sc.Insecure {
|
||
insecure = "true"
|
||
}
|
||
return fmt.Sprintf(` aws:
|
||
bucketnames: %s
|
||
endpoint: %s
|
||
region: %s
|
||
access_key_id: %s
|
||
secret_access_key: %s
|
||
insecure: %s
|
||
s3forcepathstyle: true`, sc.Bucket, endpoint, sc.Region, accessKey, secretKey, insecure)
|
||
case "oss", "cos", "obs":
|
||
insecure := "false"
|
||
if sc.Insecure {
|
||
insecure = "true"
|
||
}
|
||
// 云厂商 S3 兼容端点有两种写法：
|
||
// 区域端点 cos.ap-hongkong.myqcloud.com ← 正确
|
||
// 虚拟主机式 mybucket-1250000000.cos.<region>... ← 桶名已在 host 里
|
||
// Loki 的 S3 客户端会自己把 bucketnames 拼到 endpoint 前面（s3forcepathstyle=false
|
||
// 即虚拟主机风格），所以 endpoint 必须是区域端点。若用户在平台里填了带桶名的写法，
|
||
// 这里剥掉桶名前缀，否则会拼出 <bucket>.<bucket>.cos... 导致 TLS 证书校验失败、
|
||
// 日志写入失败（ruler/ingester 持续报 x509 证书不匹配）。
|
||
endpoint := stripBucketPrefix(sc.Endpoint, sc.Bucket)
|
||
return fmt.Sprintf(` aws:
|
||
bucketnames: %s
|
||
endpoint: %s
|
||
region: %s
|
||
access_key_id: %s
|
||
secret_access_key: %s
|
||
insecure: %s
|
||
s3forcepathstyle: false`, sc.Bucket, endpoint, sc.Region, accessKey, secretKey, insecure)
|
||
case "minio":
|
||
insecure := "false"
|
||
if sc.Insecure {
|
||
insecure = "true"
|
||
}
|
||
return fmt.Sprintf(` aws:
|
||
bucketnames: %s
|
||
endpoint: %s
|
||
access_key_id: %s
|
||
secret_access_key: %s
|
||
insecure: %s
|
||
s3forcepathstyle: true`, sc.Bucket, stripBucketPrefix(sc.Endpoint, sc.Bucket), accessKey, secretKey, insecure)
|
||
case "gcs":
|
||
return fmt.Sprintf(` gcs:
|
||
bucket_name: %s`, sc.Bucket)
|
||
case "azure":
|
||
return fmt.Sprintf(` azure:
|
||
container_name: %s
|
||
account_name: %s
|
||
account_key: %s`, sc.Bucket, accessKey, secretKey)
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// buildRulerStorageBlock 生成 ruler.storage 下与存储类型同名的嵌套配置
|
||
// (Loki 3.x 的 ruler 存储客户端不回退读取 storage_config,缺少此块会以空 bucket 启动失败)
|
||
func (d *LokiDeployer) buildRulerStorageBlock() string {
|
||
if d.deployment.StorageConfig == nil {
|
||
return ""
|
||
}
|
||
sc := d.deployment.StorageConfig
|
||
accessKey, secretKey := d.storageCredentialValues()
|
||
switch sc.StorageType {
|
||
case "s3":
|
||
endpoint := sc.Endpoint
|
||
if endpoint == "" {
|
||
endpoint = fmt.Sprintf("s3.%s.amazonaws.com", sc.Region)
|
||
}
|
||
insecure := "false"
|
||
if sc.Insecure {
|
||
insecure = "true"
|
||
}
|
||
return fmt.Sprintf(` s3:
|
||
bucketnames: %s
|
||
endpoint: %s
|
||
region: %s
|
||
access_key_id: %s
|
||
secret_access_key: %s
|
||
insecure: %s
|
||
s3forcepathstyle: true`, sc.Bucket, endpoint, sc.Region, accessKey, secretKey, insecure)
|
||
case "oss", "cos", "obs":
|
||
insecure := "false"
|
||
if sc.Insecure {
|
||
insecure = "true"
|
||
}
|
||
return fmt.Sprintf(` s3:
|
||
bucketnames: %s
|
||
endpoint: %s
|
||
region: %s
|
||
access_key_id: %s
|
||
secret_access_key: %s
|
||
insecure: %s
|
||
s3forcepathstyle: false`, sc.Bucket, sc.Endpoint, sc.Region, accessKey, secretKey, insecure)
|
||
case "minio":
|
||
insecure := "false"
|
||
if sc.Insecure {
|
||
insecure = "true"
|
||
}
|
||
return fmt.Sprintf(` s3:
|
||
bucketnames: %s
|
||
endpoint: %s
|
||
access_key_id: %s
|
||
secret_access_key: %s
|
||
insecure: %s
|
||
s3forcepathstyle: true`, sc.Bucket, sc.Endpoint, accessKey, secretKey, insecure)
|
||
case "gcs":
|
||
return fmt.Sprintf(` gcs:
|
||
bucket_name: %s`, sc.Bucket)
|
||
case "azure":
|
||
return fmt.Sprintf(` azure:
|
||
container_name: %s
|
||
account_name: %s
|
||
account_key: %s`, sc.Bucket, accessKey, secretKey)
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func (d *LokiDeployer) storageCredentialValues() (string, string) {
|
||
if d.deployment.StorageConfig == nil {
|
||
return "", ""
|
||
}
|
||
if d.deployment.Mode == "k8s-yaml" || d.deployment.Mode == "host-docker" {
|
||
return "${LOKI_STORAGE_ACCESS_KEY}", "${LOKI_STORAGE_SECRET_KEY}"
|
||
}
|
||
// host-docker 模式没有 Secret 注入，必须把库里 ENC: 密文解密后再写进配置
|
||
return crypto.DecryptField(d.deployment.StorageConfig.AccessKey),
|
||
crypto.DecryptField(d.deployment.StorageConfig.SecretKey)
|
||
}
|
||
|
||
// generateLokiConfigFallback 当模板解析失败时的降级配置
|
||
func (d *LokiDeployer) generateLokiConfigFallback(retentionHours int) string {
|
||
return fmt.Sprintf(`auth_enabled: %v
|
||
server:
|
||
http_listen_port: 3100
|
||
grpc_listen_port: 9095
|
||
http_server_read_timeout: 3m
|
||
http_server_write_timeout: 3m
|
||
log_level: info
|
||
common:
|
||
path_prefix: /var/lib/loki
|
||
storage:
|
||
filesystem:
|
||
chunks_directory: /var/lib/loki/chunks
|
||
rules_directory: /var/lib/loki/rules
|
||
replication_factor: %d
|
||
ring:
|
||
instance_addr: 127.0.0.1
|
||
kvstore:
|
||
store: inmemory
|
||
ingester:
|
||
wal:
|
||
enabled: true
|
||
dir: /var/lib/loki/wal
|
||
lifecycler:
|
||
ring:
|
||
kvstore:
|
||
store: inmemory
|
||
replication_factor: %d
|
||
final_sleep: 0s
|
||
chunk_idle_period: 30m
|
||
max_chunk_age: 2h
|
||
querier:
|
||
max_concurrent: 10
|
||
compactor:
|
||
working_directory: /var/lib/loki/compactor
|
||
retention_enabled: true
|
||
retention_delete_delay: 2h
|
||
limits_config:
|
||
retention_period: %dh
|
||
ingestion_rate_mb: 64
|
||
ingestion_burst_size_mb: 128
|
||
schema_config:
|
||
configs:
|
||
- from: "2024-01-01"
|
||
store: tsdb
|
||
object_store: filesystem
|
||
schema: v13
|
||
index:
|
||
prefix: tsdb_index_
|
||
period: 24h
|
||
analytics:
|
||
reporting_enabled: false
|
||
`, d.deployment.AuthEnabled, d.replicationFactor(), d.replicationFactor(), retentionHours)
|
||
}
|
||
|
||
// ==================== 辅助函数 ====================
|
||
|
||
func (d *LokiDeployer) replicationFactor() int {
|
||
if d.deployment.Replicas >= 3 {
|
||
return 3
|
||
}
|
||
if d.deployment.Replicas >= 2 {
|
||
return 2
|
||
}
|
||
return 1
|
||
}
|