LokiStack/backend/services/loki_config.go

// LokiDeployer — loki.yaml 配置生成:模板 + 对象存储/规则存储/ring 拓扑。
package services

import (
	"bytes"
	"fmt"
	"text/template"

	"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 的「标记 → 真正删除」间隔。默认 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
	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,
		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()
}

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
}