perf(loki): 查询并行度改为自动跟随 CPU 核数(不再硬编码)
CI / Backend (go vet + test) (push) Canceled after 0s
CI / Frontend (lint + build + test) (push) Canceled after 0s

=== 问题 ===
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 种输入(含毫核/越界/非法值)。
This commit is contained in:
opsx committed 2026-10-30 15:41:38 +00:00
1 parent ad0d35f72a
commit 419a81c2af
2 files changed
+74 -10

No files matched your search

+43 -10
View File
@@ -69,10 +69,12 @@ compactor:
limits_config:
retention_period: {{ .RetentionPeriod }}h
# 单实例 Loki 的并发护栏:默认 32 高得离谱,一条重查询就能把单进程打满
# 并行度不要超过实际 CPU 核数:超了只是让多个查询互相抢 CPU、
# 每个都变慢(实测 4 核 + 并行 8 时限流 76%,label values 都要 20 秒)。
max_query_parallelism: 6
# 单实例 Loki 的并发护栏:默认 32 高得离谱,一条重查询就能把单进程打满。
# 并行度**自动等于部署的 CPU 核数**(见 queryParallelism()):
# 高于核数只会让多个子查询互相抢 CPU——实测 4 核配 6/8 路并行时
# 25%~76% 的调度周期处于限流状态,查询耗时并没有变快,反而 label values 要 20 秒。
# 所以这个值不需要单独暴露配置:用户在「组件部署」改 CPU,这里自动跟随。
max_query_parallelism: {{ .QueryParallelism }}
# ── 查询结果缓存(官方推荐;此前完全没配,重复查询要重算一遍)───────
# align_queries_with_step 让不同时间范围的同类查询对齐到同一 step,
@@ -115,6 +117,34 @@ 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
@@ -125,12 +155,14 @@ type lokiConfigData struct {
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
// 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 {
@@ -191,6 +223,7 @@ func (d *LokiDeployer) generateLokiConfig() string {
RulerStoreType: rulerStore,
CompactorStoreType: compactorStore,
CompactorDeleteWorkers: compactorDeleteWorkers(d.deployment.StorageConfig),
QueryParallelism: queryParallelism(d.deployment.ResourcesCPU),
RingInstanceAddr: ringAddr,
RingKVStore: ringStore,
MemberlistBlock: memberlistBlock,
@@ -0,0 +1,31 @@
package services
import "testing"
// 查询并行度必须跟部署的 CPU 核数对齐。
//
// 事故背景:曾经写死 8,而部署只有 4 核 → 多个子查询互相抢 CPU,
// 实测 76% 的调度周期被限流,连 label values 都要 20 秒。
// 现在改成从 resourcesCPU 推导,用户在界面改 CPU 时并行度自动跟随。
func TestQueryParallelism(t *testing.T) {
cases := []struct {
cpu string
want int
}{
{"4", 4},
{"8", 8},
{"2", 2},
{"1", 2}, // 低于 2 核按 2 兜底
{"16", 12}, // 上限 12
{"", 4}, // 空值走默认
{"abc", 4}, // 非法值走默认
{"4000m", 4}, // 毫核写法
{"500m", 2}, // 0.5 核 → 向上取整 1 → 不低于 2
{"6000m", 6},
}
for _, c := range cases {
if got := queryParallelism(c.cpu); got != c.want {
t.Errorf("queryParallelism(%q) = %d, want %d", c.cpu, got, c.want)
}
}
}