LokiStack/backend/services/collector_templates.go

package services

import (
	"encoding/json"
	"fmt"
	"regexp"
	"sort"
	"strings"
)

// PromtailConfig 主机 Promtail 配置（推送至 lokiEndpoint，采集 logPath）
// 生产级配置：持久化位置文件、批处理、重试、CRI日志格式解析
// promtailExternalLabels 组并 cluster 标识与用户自定义标签(JSON)为 external_labels YAML 行
// promtailExternalLabels 生成 external_labels 的子项。
// indent 必须与该模板里 `external_labels:` 的层级匹配：
//   - 主机模板（4 空格）→ 子项 6 空格
//   - DaemonSet 模板（6 空格）→ 子项 8 空格
//
// 早先写死 8 空格，主机模式一旦配了标签就会生成非法 YAML，Promtail 直接装不上。
func promtailExternalLabels(cluster string, labelsJSON string, indent string) string {
	lines := []string{fmt.Sprintf("%scluster: %s", indent, yamlScalar(cluster))}
	if labelsJSON != "" {
		var m map[string]string
		if err := json.Unmarshal([]byte(labelsJSON), &m); err == nil {
			keys := make([]string, 0, len(m))
			for k := range m {
				keys = append(keys, k)
			}
			sort.Strings(keys)
			for _, k := range keys {
				lines = append(lines, fmt.Sprintf("%s%s: %s", indent, k, yamlScalar(m[k])))
			}
		}
	}
	return strings.Join(lines, "\n")
}

// isCRILogPath 判断路径是否属于容器运行时日志（K8s pod 日志或 docker 容器日志）。
// 只有这类日志才是 CRI 格式，才能用 cri 阶段解析。
func isCRILogPath(logPath string) bool {
	p := strings.ToLower(logPath)
	return strings.Contains(p, "/pods/") ||
		strings.Contains(p, "/containers/") ||
		strings.Contains(p, "/docker/")
}

// PromtailDaemonsetYAML DaemonSet 模式纯 YAML 安装(SA/RBAC/ConfigMap/DaemonSet),
// 经 Kubernetes API 直接提交,后端无需任何额外二进制。
// relabel 体系与主流 Promtail 部署惯例对齐:namespace/pod/container/node_name + __path__ 按 pod uid 精确匹配。
// sanitizePromLabel 把 K8s 标签名转成 Prometheus/Promtail 的 meta label 片段。
//
//	k8s.kuboard.cn/name  → k8s_kuboard_cn_name
//	app.kubernetes.io/name → app_kubernetes_io_name
func sanitizePromLabel(name string) string {
	var b strings.Builder
	for _, r := range strings.TrimSpace(name) {
		switch {
		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '_':
			b.WriteRune(r)
		default:
			b.WriteRune('_')
		}
	}
	return b.String()
}

// promtailAppSourceLabel 把用户配置的 AppLabelSource 转成 Promtail 的 meta label 名。
// 返回空串表示「用默认的四级回退链」。
func promtailAppSourceLabel(src string) string {
	src = strings.TrimSpace(src)
	if src == "" || src == "controller_name" {
		return ""
	}
	if strings.HasPrefix(src, "label:") {
		src = strings.TrimPrefix(src, "label:")
	}
	// 已经是 __meta_ 开头就直接用
	if strings.HasPrefix(src, "__meta_kubernetes_") {
		return src
	}
	return "__meta_kubernetes_pod_label_" + sanitizePromLabel(src)
}

// promtailAppIncludeFilter 生成 app 白名单的 keep 规则（为空则不限）。
func promtailAppIncludeFilter(apps []string) string {
	clean := make([]string, 0, len(apps))
	for _, a := range apps {
		if a = strings.TrimSpace(a); a != "" {
			clean = append(clean, regexp.QuoteMeta(a))
		}
	}
	if len(clean) == 0 {
		return ""
	}
	return fmt.Sprintf("\n      - source_labels: [app]\n        regex: ^(%s)$\n        action: keep",
		strings.Join(clean, "|"))
}

// promtailNamespaceFilter 生成 kubernetes_sd_configs 的命名空间白名单。
// namespaces 为空 → 返回空串（采集全集群所有命名空间）。
func promtailNamespaceFilter(namespaces []string) string {
	clean := make([]string, 0, len(namespaces))
	for _, n := range namespaces {
		if n = strings.TrimSpace(n); n != "" {
			clean = append(clean, n)
		}
	}
	if len(clean) == 0 {
		return ""
	}
	var b strings.Builder
	b.WriteString("        namespaces:\n          names:\n")
	for _, n := range clean {
		b.WriteString("          - ")
		b.WriteString(n)
		b.WriteString("\n")
	}
	return b.String()
}

func PromtailDaemonsetYAML(ns, lokiURL, clusterName, labelsJSON string, targetNamespaces []string, appLabelSource string, appInclude []string) string {
	promtailImage := ResolveImageFor("promtail", "grafana/promtail", "3.4.0")
	extLabels := promtailExternalLabels(clusterName, labelsJSON, "        ")
	nsFilter := promtailNamespaceFilter(targetNamespaces)
	// 自定义 app 标签来源（用户可用自己注入的私有标签），插到回退链最前面
	appSourceLine := ""
	if lbl := promtailAppSourceLabel(appLabelSource); lbl != "" {
		appSourceLine = "        - " + lbl + "\n"
	}
	appIncludeFilter := promtailAppIncludeFilter(appInclude)
	return fmt.Sprintf(`apiVersion: v1
kind: ServiceAccount
metadata:
  name: promtail
  namespace: %s
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: loki-promtail
rules:
- apiGroups: [""]
  resources: ["pods", "nodes"]
  verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: loki-promtail
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: loki-promtail
subjects:
- kind: ServiceAccount
  name: promtail
  namespace: %s
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: promtail
  namespace: %s
data:
  promtail.yaml: |
    server:
      http_listen_port: 3101
      grpc_listen_port: 0
    positions:
      filename: /run/promtail/positions.yaml
    clients:
    - url: %s/loki/api/v1/push
      batchwait: 1s
      batchsize: 1048576
      backoff_config:
        min_period: 500ms
        max_period: 5m
        max_retries: 10
      external_labels:
%s
    scrape_configs:
    - job_name: kubernetes-pods
      kubernetes_sd_configs:
      - role: pod
%s
      pipeline_stages:
      - cri: {}
      - multiline:
          firstline: '^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}[+\\-]\\d{2}:\\d{2}'
          max_wait_time: 3s
          max_lines: 500
      - timestamp:
          source: time
          format: RFC3339Nano
      relabel_configs:
      # 跳过已终止的 Pod（Evicted / Failed / Succeeded）。
      # 实证：某服务残留 387 个 Evicted Pod，日志文件还在节点上，
      # Promtail 会把它们的历史异常堆栈全量采回来，导致 15 分钟 27 万条的假象，
      # 也会污染 Loki 存储和查询结果。
      - source_labels: [__meta_kubernetes_pod_phase]
        regex: (Succeeded|Failed)
        action: drop
      # 从 controller name 提取 deployment 名（去掉 replicaset 哈希）
      - source_labels: [__meta_kubernetes_pod_controller_name]
        regex: ([0-9a-z-.]+?)(-[0-9a-f]{8,10})?
        action: replace
        target_label: __tmp_controller_name
      # app 四级回退 —— 全部基于 Kubernetes 原生元数据，跨集群通用：
      #   1) app.kubernetes.io/name   ← K8s 标准标签约定
      #   2) app                      ← 最常见的简写
      #   3) controller name 去 hash  ← Pod 的 owner(ReplicaSet) → Deployment 名
      #                                  __meta_kubernetes_pod_controller_name 是
      #                                  Promtail 从 K8s API 直接拿的，任何集群都有，
      #                                  不依赖任何第三方工具注入的标签
      #   4) pod name                 ← 最后兜底，保证 app 永不为空
      #
      # 不使用 k8s.kuboard.cn/name 之类的平台私有标签：那只是 Kuboard 管理的集群才有，
      # 换其他方式部署的集群就不存在，会破坏通用性。
      # 如集群确实有私有规范标签，可在「采集器 → 编辑」里配置 AppLabelSource，
      # 它会作为最高优先级插到下面列表最前（其余仍兜底）。
      - source_labels:
%s        - __meta_kubernetes_pod_label_app_kubernetes_io_name
        - __meta_kubernetes_pod_label_app
        - __tmp_controller_name
        - __meta_kubernetes_pod_name
        regex: ^;*([^;]+)(;.*)?$
        action: replace
        target_label: app
      # instance：app.kubernetes.io/instance → instance
      - source_labels: [__meta_kubernetes_pod_label_app_kubernetes_io_instance, __meta_kubernetes_pod_label_instance]
        regex: ^;*([^;]+)(;.*)?$
        action: replace
        target_label: instance
      # component：app.kubernetes.io/component → component
      - source_labels: [__meta_kubernetes_pod_label_app_kubernetes_io_component, __meta_kubernetes_pod_label_component]
        regex: ^;*([^;]+)(;.*)?$
        action: replace
        target_label: component
      - action: replace
        source_labels: [__meta_kubernetes_pod_node_name]
        target_label: node_name
      - action: replace
        source_labels: [__meta_kubernetes_namespace]
        target_label: namespace
      - action: replace
        source_labels: [__meta_kubernetes_pod_name]
        target_label: pod
      - action: replace
        source_labels: [__meta_kubernetes_pod_container_name]
        target_label: container
      # job: namespace/app（必须放在 app 之后，否则取到的是未回退的空值）
      - action: replace
        replacement: $1
        separator: /
        source_labels: [namespace, app]
        target_label: job
      - action: replace
        replacement: /var/log/pods/*$1/*.log
        separator: /
        source_labels: [__meta_kubernetes_pod_uid, __meta_kubernetes_pod_container_name]
        target_label: __path__%s
---
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: promtail
  namespace: %s
  labels:
    app: promtail
spec:
  selector:
    matchLabels:
      app: promtail
  template:
    metadata:
      labels:
        app: promtail
    spec:
      serviceAccountName: promtail
      tolerations:
      - effect: NoSchedule
        operator: Exists
      containers:
      - name: promtail
        image: %s
        imagePullPolicy: IfNotPresent
        args: ["-config.file=/etc/promtail/promtail.yaml"]
        env:
        - name: HOSTNAME
          valueFrom:
            fieldRef:
              fieldPath: spec.nodeName
        ports:
        - name: http-metrics
          containerPort: 3101
        readinessProbe:
          httpGet:
            path: /ready
            port: http-metrics
          initialDelaySeconds: 10
          periodSeconds: 10
          timeoutSeconds: 3
          failureThreshold: 5
        resources:
          requests:
            cpu: 100m
            memory: 128Mi
          limits:
            cpu: 500m
            memory: 512Mi
        volumeMounts:
        - name: config
          mountPath: /etc/promtail
        - name: run
          mountPath: /run/promtail
        - name: pods
          mountPath: /var/log/pods
          readOnly: true
        - name: containers
          mountPath: /var/lib/docker/containers
          readOnly: true
      volumes:
      - name: config
        configMap:
          name: promtail
      - name: run
        hostPath:
          path: /run/promtail
      - name: pods
        hostPath:
          path: /var/log/pods
      - name: containers
        hostPath:
          path: /var/lib/docker/containers
`, ns, ns, ns, lokiURL, extLabels, nsFilter, appSourceLine, appIncludeFilter, ns, promtailImage)
}

// FluentbitConfig Sidecar 模式 Fluent Bit 配置（采集 logPath，推送至 lokiHost:lokiPort）
// Sidecar 已明确知道应用与命名空间，无需 Kubernetes 元数据过滤器。
func FluentbitConfig(lokiHost, lokiPort, logPath, app, namespace string) string {
	return fmt.Sprintf(`[SERVICE]
    Flush          1
    Log_Level      info
    HTTP_Server    On
    HTTP_Listen    0.0.0.0
    HTTP_Port      2020
    storage.path   /tmp/flb-storage/
    storage.sync   normal
    storage.checksum off
    storage.backlog.mem_limit 5M
    Health_Check   On
    HC_Errors_Count 5
    HC_Retry_Failure_Count 5
    HC_Period      60

[INPUT]
    Name              tail
    Path              %s
    Tag               app.%s
    Refresh_Interval  5
    Mem_Buf_Limit     10MB
    Skip_Long_Lines   On
    Read_from_Head    True
    DB                /tmp/flb.db
    DB.Sync           Normal
    Buffer_Max_Size   1MB
    storage.type      filesystem

[FILTER]
    Name    modify
    Match   app.*
    Add     namespace %s
    Add     app %s
    Add     source fluent-bit

[OUTPUT]
    Name                loki
    Match               *
    Host                %s
    Port                %s
    Labels              job=fluent-bit, app=%s, namespace=%s
    Line_Format         json
    Retry_Limit         10
`, logPath, app, namespace, app, lokiHost, lokiPort, app, namespace)
}

// FluentbitConfigMapYAML 为每个采集策略生成独立 ConfigMap，避免同命名空间的 Sidecar 相互覆盖。
func FluentbitConfigMapYAML(conf, namespace, name string) string {
	return fmt.Sprintf(`apiVersion: v1
kind: ConfigMap
metadata:
  name: %s
  namespace: %s
data:
  fluent-bit.conf: |
%s
`, name, namespace, indentBlock(conf, "    "))
}

// SidecarPatchYAML 将业务容器与 Fluent Bit 挂到同一日志卷和目录。
func SidecarPatchYAML(configMapName string, mount sidecarLogMount) string {
	volume := ""
	if mount.createVolume {
		volume = fmt.Sprintf("      - name: %s\n        emptyDir: {}\n", mount.volumeName)
	}
	appMount := ""
	if mount.mountApplication {
		appMount = fmt.Sprintf("      - name: %s\n        volumeMounts:\n        - name: %s\n          mountPath: %q\n", mount.containerName, mount.volumeName, mount.mountPath)
	}
	subPath := ""
	if mount.subPath != "" {
		subPath = fmt.Sprintf("          subPath: %q\n", mount.subPath)
	}
	return fmt.Sprintf(`spec:
  template:
    spec:
      volumes:
%s      - name: fluent-bit-config
        configMap:
          name: %s
      - name: fluent-bit-state
        emptyDir: {}
      containers:
%s      - name: fluent-bit
        image: %s
        imagePullPolicy: IfNotPresent
        ports:
        - name: fluent-bit-http
          containerPort: 2020
          protocol: TCP
        volumeMounts:
        - name: %s
          mountPath: %q
%s          readOnly: true
        - name: fluent-bit-config
          mountPath: /fluent-bit/etc/
        - name: fluent-bit-state
          mountPath: /tmp
        resources:
          limits:
            cpu: 500m
            memory: 512Mi
          requests:
            cpu: 100m
            memory: 128Mi
        securityContext:
          allowPrivilegeEscalation: false
          readOnlyRootFilesystem: true
          capabilities:
            drop:
            - ALL
        livenessProbe:
          httpGet:
            path: /
            port: fluent-bit-http
          initialDelaySeconds: 10
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /
            port: fluent-bit-http
          initialDelaySeconds: 5
          periodSeconds: 5
`, volume, configMapName, appMount, ResolveImageFor("fluentbit", "fluent/fluent-bit", "5.1.5"), mount.volumeName, mount.mountPath, subPath)
}

// indentBlock 给每行加前缀（ConfigMap 嵌入配置需缩进）
func indentBlock(s, pad string) string {
	lines := strings.Split(s, "\n")
	for i, l := range lines {
		if l != "" {
			lines[i] = pad + l
		}
	}
	return strings.Join(lines, "\n")
}

// parseLokiURL 从 http://host:port 提取 host 与 port（Fluent Bit OUTPUT 需要）
func parseLokiURL(endpoint string) (host, port string) {
	u := strings.TrimSpace(endpoint)
	u = strings.TrimPrefix(u, "http://")
	u = strings.TrimPrefix(u, "https://")
	if i := strings.Index(u, "/"); i >= 0 {
		u = u[:i]
	}
	if i := strings.LastIndex(u, ":"); i >= 0 {
		return u[:i], u[i+1:]
	}
	return u, "3100"
}