LokiStack/backend/services/promtail_host_config.go

package services

import (
	"fmt"
	"strconv"
	"strings"

	"lokistack/models"
)

// 系统日志时间戳格式（安装期探测结果）。
// 不同发行版/rsyslog 版本写法不同，正则必须跟着变，否则 timestamp 阶段不生效。
const (
	// SyslogFormatISO8601 2026-07-17T15:38:24.923784+08:00（TencentOS 4 / 新版 rsyslog）
	SyslogFormatISO8601 = "iso8601"
	// SyslogFormatClassic Oct 28 18:11:06（Debian / 老 rsyslog）
	SyslogFormatClassic = "rfc3164"
)

// HostPromtailOptions 生成主机模式 promtail.yaml 的入参。
type HostPromtailOptions struct {
	Endpoint   string
	Cluster    string
	LabelsJSON string
	// Hostname 真机主机名（Promtail 不展开 ${HOSTNAME}，必须写具体值）
	Hostname string
	// SyslogFormat 安装时探测：iso8601 / rfc3164
	SyslogFormat string
	// SystemLogPaths 安装时探测到的系统日志绝对路径。
	// Ubuntu/Debian 是 syslog+auth.log，RHEL 系是 messages+secure；
	// 留空时退回 RHEL 系默认值（兼容旧调用）。
	SystemLogPaths []string
	LogPath        string
	Jobs           []models.ScrapeJob
}

// defaultSystemLogPaths 是探测失败时的兜底（RHEL/CentOS/TencentOS/AWS Linux）。
var defaultSystemLogPaths = []string{
	"/var/log/messages", "/var/log/secure", "/var/log/cron",
}

// PromtailConfigForHost 渲染主机模式配置。
//
// 跨云/跨发行版的三个关键点：
//  1. 系统日志路径来自安装期探测（Ubuntu 没有 /var/log/messages，写死就采不到）。
//  2. 时间戳格式按探测结果选：ISO8601 → RFC3339Nano；传统 syslog → Jan _2 15:04:05，
//     location 用 Local（跟随进程所在主机时区，不写死某个地区）。
//  3. host 标签写真实主机名。
func PromtailConfigForHost(o HostPromtailOptions) (string, error) {
	hostname := strings.TrimSpace(o.Hostname)
	if hostname == "" {
		hostname = "unknown-host"
	}
	cluster := o.Cluster
	if cluster == "" {
		cluster = "host"
	}

	var b strings.Builder
	fmt.Fprintf(&b, `server:
  http_listen_port: 9080
  grpc_listen_port: 0
  log_level: info
  log_format: logfmt

positions:
  filename: /var/lib/promtail/positions.yaml
  sync_period: 10s

clients:
  - url: %s/loki/api/v1/push
    batchwait: 1s
    batchsize: 1048576
    timeout: 10s
    backoff_config:
      min_period: 500ms
      max_period: 5m
      max_retries: 10
    external_labels:
      job: host-logs
%s

limits_config:
  readline_rate_enabled: false
  readline_rate: 1000
  readline_burst: 10000
  max_streams: 0

scrape_configs:
`, strings.TrimSuffix(o.Endpoint, "/"), promtailExternalLabels(cluster, o.LabelsJSON, "      "))

	// 自定义采集项优先；一个都没有时用「应用日志 + 系统日志」的默认组合
	if len(o.Jobs) > 0 {
		jobs := make([]models.ScrapeJob, len(o.Jobs))
		copy(jobs, o.Jobs)
		for i := range jobs {
			if jobs[i].System && len(jobs[i].Paths) == 0 {
				jobs[i].Paths = pickSystemPaths(o.SystemLogPaths)
			}
		}
		// 用户填的正则/路径/标签必须校验：非法名字会拼出坏 YAML，
		// 敏感路径（/etc/shadow、私钥）会把主机凭据推到日志平台里。
		if err := models.ValidateScrapeJobs(jobs); err != nil {
			return "", err
		}
		for _, job := range jobs {
			if job.System && len(job.Paths) == 0 {
				job.Paths = pickSystemPaths(o.SystemLogPaths)
			}
			rendered, err := renderScrapeJob(job, hostname, o.SyslogFormat)
			if err != nil {
				return "", err
			}
			b.WriteString(rendered)
		}
		return b.String(), nil
	}

	// 默认 1：应用日志
	if path := strings.TrimSpace(o.LogPath); path != "" {
		rendered, err := renderScrapeJob(models.ScrapeJob{
			Name:  "app",
			Paths: splitPaths(path),
		}, hostname, o.SyslogFormat)
		if err != nil {
			return "", err
		}
		b.WriteString(rendered)
	}

	// 默认 2：系统日志
	rendered, err := renderScrapeJob(models.ScrapeJob{
		Name:   "system",
		System: true,
		Paths:  pickSystemPaths(o.SystemLogPaths),
	}, hostname, o.SyslogFormat)
	if err != nil {
		return "", err
	}
	b.WriteString(rendered)
	return b.String(), nil
}

// yamlScalar 输出 YAML 标量：简单值不加引号，含特殊字符时用双引号包裹并转义。
// external_labels / labels 里的值（env=pre、app=cex-api）加引号会变成
// `env: "pre"`，与用户预期和既有配置风格不一致。
func yamlScalar(v string) string {
	safe := true
	for _, r := range v {
		switch {
		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9':
		case r == '_' || r == '-' || r == '.' || r == '/' || r == ':' || r == '@':
		default:
			safe = false
		}
	}
	if safe && v != "" {
		return v
	}
	return strconv.Quote(v)
}

// pickSystemPaths 选系统日志路径：优先用安装期探测结果，否则退回默认。
func pickSystemPaths(detected []string) []string {
	if len(detected) > 0 {
		return detected
	}
	return defaultSystemLogPaths
}

// splitPaths 兼容调用方用逗号或空白分隔多个路径。
func splitPaths(raw string) []string {
	fields := strings.FieldsFunc(raw, func(r rune) bool { return r == ',' || r == '\n' })
	out := make([]string, 0, len(fields))
	for _, f := range fields {
		if p := strings.TrimSpace(f); p != "" {
			out = append(out, p)
		}
	}
	return out
}

// renderScrapeJob 把一条 ScrapeJob 渲染成 promtail 的 scrape_config。
func renderScrapeJob(job models.ScrapeJob, hostname, syslogFormat string) (string, error) {
	if len(job.Paths) == 0 {
		return "", fmt.Errorf("采集项 %q 没有配置日志路径", job.Name)
	}
	name := job.Name
	if name == "" {
		name = "custom"
	}
	isCRI := pathsContainCRI(job.Paths)

	var b strings.Builder
	fmt.Fprintf(&b, "  - job_name: %s\n", name)
	// encoding 是 scrape_config 级字段（不是 static_config），写错位置 promtail 直接起不来
	if job.Encoding != "" {
		fmt.Fprintf(&b, "    encoding: %s\n", job.Encoding)
	}
	// 一个 job 下多条路径要写成 static_configs 的**列表**，
	// 不能重复出现 static_configs 键（YAML 会报 duplicate key）。
	b.WriteString("    static_configs:\n")
	for _, p := range job.Paths {
		fmt.Fprintf(&b, "      - targets: [localhost]\n        labels:\n          job: %s\n          host: %s\n          __path__: %s\n",
			yamlScalar(name), yamlScalar(hostname), yamlScalar(p))
		for k, v := range job.Labels {
			fmt.Fprintf(&b, "          %s: %s\n", sanitizePromLabel(k), yamlScalar(v))
		}
	}

	stages := make([]string, 0, 8)
	// cri 必须排在最前：容器日志行是 "时间 stdout F 内容"，
	// 不先剥离前缀，后面的 multiline/json/regex 全都匹配不上。
	if isCRI {
		stages = append(stages, "      - cri: {}")
	}
	if job.Multiline != nil && strings.TrimSpace(job.Multiline.FirstLine) != "" {
		maxLines := job.Multiline.MaxLines
		if maxLines <= 0 {
			maxLines = 128
		}
		stages = append(stages, fmt.Sprintf(`      - multiline:
          firstline: '%s'
          max_wait_time: 3s
          max_lines: %d`, strings.ReplaceAll(job.Multiline.FirstLine, "'", "''"), maxLines))
	}
	if job.JSON {
		stages = append(stages, "      - json: {}")
	}
	if job.System {
		switch syslogFormat {
		case SyslogFormatISO8601:
			stages = append(stages,
				`      - regex:
          expression: '^(?P<ts>\d{4}-\d{2}-\d{2}T\S+)'`,
				`      - timestamp:
          source: ts
          format: RFC3339Nano`)
		case SyslogFormatClassic:
			stages = append(stages,
				`      - regex:
          expression: '^(?P<ts>[A-Z][a-z]{2}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})\s+(?P<hostname>\S+)\s+(?P<unit>[^:\[]+)(?:\[(?P<pid>\d+)\])?:\s*(?P<msg>.*)$'`,
				`      - timestamp:
          source: ts
          format: Jan _2 15:04:05
          location: Local`)
		}
	} else if job.Timestamp != nil && strings.TrimSpace(job.Timestamp.Source) != "" {
		loc := ""
		if job.Timestamp.Location != "" {
			loc = fmt.Sprintf("\n          location: %s", job.Timestamp.Location)
		}
		stages = append(stages, fmt.Sprintf(`      - timestamp:
          source: %s
          format: '%s'%s`, job.Timestamp.Source, strings.ReplaceAll(job.Timestamp.Format, "'", "''"), loc))
	} else if isCRI {
		// CRI 行自带时间字段，不解析它就会用采集时间（回填历史日志时间全错）
		stages = append(stages, `      - timestamp:
          source: time
          format: RFC3339Nano`)
	}
	for _, expr := range job.DropIfMatch {
		if strings.TrimSpace(expr) == "" {
			continue
		}
		stages = append(stages, fmt.Sprintf(`      - drop:
          expression: '%s'
          drop_counter_reason: drop_if_match`, strings.ReplaceAll(expr, "'", "''")))
	}

	rate, burst := 2000, 4000
	if job.RateLimit != nil {
		if job.RateLimit.Rate > 0 {
			rate = job.RateLimit.Rate
		}
		if job.RateLimit.Burst > 0 {
			burst = job.RateLimit.Burst
		}
	}
	stages = append(stages, fmt.Sprintf(`      - limit:
          rate: %d
          burst: %d
          drop: true`, rate, burst))

	fmt.Fprintf(&b, "    pipeline_stages:\n%s\n", strings.Join(stages, "\n"))
	return b.String(), nil
}

// pathsContainCRI 判断是否有容器运行时日志路径（需要 cri 阶段解析）。
func pathsContainCRI(paths []string) bool {
	for _, p := range paths {
		if isCRILogPath(p) {
			return true
		}
	}
	return false
}