opsx
419a81c2af
=== 问题 ===
max_query_parallelism 原来是硬编码值，改过两轮（8 → 6）：
- 8：4 核机器上 76% 调度周期被限流，label values 要 20 秒
- 6：仍然高于核数，25% 周期限流，查询并没有更快
根因不是"该填几"，而是**并行度只有一个正确值：等于 CPU 核数**。
把它做成可配置项反而是坑 —— 用户改了 CPU 忘了改并行度就出现错配。
=== 做法 ===
新增 queryParallelism(resourcesCPU)：
- 从部署记录的资源规格推导，模板渲染成 {{ .QueryParallelism }}
- 容错：支持 "4000m"/"500m" 毫核写法（向上取整）
- 边界：低于 2 核按 2，高于 12 核按 12（再高频只会互相抢 CPU）
- 非法值/空值回退 4
用户在「组件部署」里改 CPU，并行度自动跟随，页面上不需要新字段。
=== 实测（4 核 + 并行 4）===
by-app(15m) 11913ms 75 条序列 （并行 6 时 12438ms）
label/namespace 38ms
label/app 25ms
label/pod 25ms
限流 61/252 周期（仅重查询那几秒）
附带 TestQueryParallelism 覆盖 10 种输入（含毫核/越界/非法值）。
503 lines
15 KiB
Go
503 lines
15 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 高得离谱，一条重查询就能把单进程打满。
|
||
# 并行度**自动等于部署的 CPU 核数**（见 queryParallelism()）：
|
||
# 高于核数只会让多个子查询互相抢 CPU——实测 4 核配 6/8 路并行时
|
||
# 25%~76% 的调度周期处于限流状态，查询耗时并没有变快，反而 label values 要 20 秒。
|
||
# 所以这个值不需要单独暴露配置：用户在「组件部署」改 CPU，这里自动跟随。
|
||
max_query_parallelism: {{ .QueryParallelism }}
|
||
|
||
# ── 查询结果缓存（官方推荐；此前完全没配，重复查询要重算一遍）───────
|
||
# 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
|
||
`
|
||
|
||
// queryParallelism 把部署的 CPU 核数换算成查询并行度。
|
||
//
|
||
// 为什么不用独立配置项：并行度只有"跟核数对齐"一个正确值。
|
||
// 单独暴露一个配置，用户改了 CPU 忘了改并行度，就会出现
|
||
// "4 核配 8 路并行 → 限流 76%"这种自己坑自己的组合。
|
||
// 这里直接跟随 CPU，用户只需要在「组件部署」里改 CPU。
|
||
func queryParallelism(resourcesCPU string) int {
|
||
value := strings.TrimSpace(resourcesCPU)
|
||
value = strings.TrimSuffix(value, "m") // 容忍 "4000m" 这种写法
|
||
cores, err := strconv.Atoi(value)
|
||
if err != nil || cores <= 0 {
|
||
return defaultQueryParallelism
|
||
}
|
||
if strings.HasSuffix(strings.TrimSpace(resourcesCPU), "m") {
|
||
cores = (cores + 999) / 1000 // 毫核向上取整到核
|
||
}
|
||
if cores < 2 {
|
||
return 2
|
||
}
|
||
// 上限 12：再高单实例 Loki 只会互相抢 CPU，且并发护栏本身还有全局闸门
|
||
if cores > 12 {
|
||
return 12
|
||
}
|
||
return cores
|
||
}
|
||
|
||
const defaultQueryParallelism = 4
|
||
|
||
type lokiConfigData struct {
|
||
AuthEnabled bool
|
||
ReplicationFactor int
|
||
RetentionPeriod int
|
||
StorageConfigBlock string
|
||
ObjectStore string
|
||
RulerStoreType string
|
||
CompactorStoreType string
|
||
// CompactorDeleteWorkers 过期 chunk 删除并发，按对象存储类型取默认值
|
||
CompactorDeleteWorkers int
|
||
// QueryParallelism 查询并行度，自动等于部署 CPU 核数（2~12）
|
||
QueryParallelism 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),
|
||
QueryParallelism: queryParallelism(d.deployment.ResourcesCPU),
|
||
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
|
||
}
|