【系统架构设计百科】日志架构：从 printf 到结构化日志管道

{
  "timestamp": "2024-03-15T14:23:07.342Z",
  "level": "ERROR",
  "logger": "OrderService",
  "message": "Failed to create order",
  "trace_id": "abc123def456",
  "span_id": "span789",
  "request_id": "req-001",
  "user_id": "12345",
  "order_id": "ORD-98765",
  "error_code": "PAYMENT_TIMEOUT",
  "timeout_ms": 3000,
  "items": ["SKU001", "SKU002"],
  "service": "order-service",
  "instance": "order-service-pod-7b8c9",
  "environment": "production"
}

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "required": ["timestamp", "level", "message", "service"],
  "properties": {
    "timestamp": {
      "type": "string",
      "format": "date-time",
      "description": "ISO 8601 格式，UTC 时区，精确到毫秒"
    },
    "level": {
      "type": "string",
      "enum": ["TRACE", "DEBUG", "INFO", "WARN", "ERROR", "FATAL"]
    },
    "message": {
      "type": "string",
      "description": "简短的人类可读描述，不包含变量值"
    },
    "service": {
      "type": "string",
      "description": "服务名称，与 Kubernetes 的 Deployment 名称一致"
    },
    "instance": {
      "type": "string",
      "description": "实例标识，Kubernetes 环境下为 Pod 名称"
    },
    "trace_id": {
      "type": "string",
      "description": "分布式追踪的 trace ID，遵循 W3C Trace Context 规范"
    },
    "span_id": {
      "type": "string",
      "description": "当前 span 的 ID"
    },
    "request_id": {
      "type": "string",
      "description": "业务请求 ID，由网关层生成"
    },
    "environment": {
      "type": "string",
      "enum": ["development", "staging", "production"]
    }
  }
}

{
  "responseTime": 234,
  "response_time_ms": 234,
  "resp_time": 234,
  "duration": 0.234,
  "elapsed": "234ms"
}

package main

import (
    "context"
    "log/slog"
    "os"
)

type contextKey string

const (
    traceIDKey   contextKey = "trace_id"
    spanIDKey    contextKey = "span_id"
    requestIDKey contextKey = "request_id"
)

// ContextHandler 自动从 context 中提取追踪字段并注入日志
type ContextHandler struct {
    inner slog.Handler
}

func NewContextHandler(inner slog.Handler) *ContextHandler {
    return &ContextHandler{inner: inner}
}

func (h *ContextHandler) Enabled(ctx context.Context, level slog.Level) bool {
    return h.inner.Enabled(ctx, level)
}

func (h *ContextHandler) Handle(ctx context.Context, r slog.Record) error {
    if traceID, ok := ctx.Value(traceIDKey).(string); ok {
        r.AddAttrs(slog.String("trace_id", traceID))
    }
    if spanID, ok := ctx.Value(spanIDKey).(string); ok {
        r.AddAttrs(slog.String("span_id", spanID))
    }
    if requestID, ok := ctx.Value(requestIDKey).(string); ok {
        r.AddAttrs(slog.String("request_id", requestID))
    }
    return h.inner.Handle(ctx, r)
}

func (h *ContextHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
    return NewContextHandler(h.inner.WithAttrs(attrs))
}

func (h *ContextHandler) WithGroup(name string) slog.Handler {
    return NewContextHandler(h.inner.WithGroup(name))
}

func main() {
    jsonHandler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
        Level: slog.LevelInfo,
    })
    logger := slog.New(NewContextHandler(jsonHandler))
    slog.SetDefault(logger)

    // 模拟中间件注入追踪字段
    ctx := context.Background()
    ctx = context.WithValue(ctx, traceIDKey, "abc123def456")
    ctx = context.WithValue(ctx, spanIDKey, "span789")
    ctx = context.WithValue(ctx, requestIDKey, "req-001")

    // 开发者只需关注业务字段，追踪字段自动注入
    slog.InfoContext(ctx, "order created",
        slog.String("order_id", "ORD-98765"),
        slog.Int("item_count", 3),
    )
}

{
  "time": "2024-03-15T14:23:07.342Z",
  "level": "INFO",
  "msg": "order created",
  "order_id": "ORD-98765",
  "item_count": 3,
  "trace_id": "abc123def456",
  "span_id": "span789",
  "request_id": "req-001"
}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import javax.servlet.*;
import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
import java.util.UUID;

// Servlet Filter 自动注入追踪字段到 MDC
public class TraceContextFilter implements Filter {
    private static final String TRACE_ID = "trace_id";
    private static final String REQUEST_ID = "request_id";

    @Override
    public void doFilter(ServletRequest request, ServletResponse response,
                         FilterChain chain) throws IOException, ServletException {
        HttpServletRequest httpReq = (HttpServletRequest) request;
        try {
            // 从上游传递的 Header 中获取 trace_id，没有则生成
            String traceId = httpReq.getHeader("X-Trace-Id");
            if (traceId == null || traceId.isEmpty()) {
                traceId = UUID.randomUUID().toString().replace("-", "");
            }
            String requestId = UUID.randomUUID().toString().replace("-", "");

            MDC.put(TRACE_ID, traceId);
            MDC.put(REQUEST_ID, requestId);

            chain.doFilter(request, response);
        } finally {
            MDC.clear();
        }
    }
}

<configuration>
    <appender name="JSON_STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="net.logstash.logback.encoder.LogstashEncoder">
            <includeMdcKeyName>trace_id</includeMdcKeyName>
            <includeMdcKeyName>request_id</includeMdcKeyName>
            <customFields>
                {"service":"order-service","environment":"production"}
            </customFields>
        </encoder>
    </appender>

    <root level="INFO">
        <appender-ref ref="JSON_STDOUT" />
    </root>
</configuration>

slog.Info(fmt.Sprintf("User %s created order %s with %d items totaling $%.2f",
    userID, orderID, itemCount, total))

slog.Info("order created",
    slog.String("user_id", userID),
    slog.String("order_id", orderID),
    slog.Int("item_count", itemCount),
    slog.Float64("total_usd", total),
)

filebeat.autodiscover:
  providers:
    - type: kubernetes
      node: ${NODE_NAME}
      hints.enabled: true
      hints.default_config:
        type: container
        paths:
          - /var/log/containers/*-${data.kubernetes.container.id}.log
        # 多行合并：Java 异常堆栈
        multiline.pattern: '^[[:space:]]+(at|\.{3}|Caused by)'
        multiline.negate: false
        multiline.match: after

processors:
  - add_kubernetes_metadata:
      host: ${NODE_NAME}
      matchers:
        - logs_path:
            logs_path: "/var/log/containers/"
  - decode_json_fields:
      fields: ["message"]
      target: ""
      overwrite_keys: true
  - drop_fields:
      fields: ["agent", "ecs", "host.name"]

output.kafka:
  hosts: ["kafka-0:9092", "kafka-1:9092", "kafka-2:9092"]
  topic: "logs-%{[kubernetes.namespace]}"
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  compression: lz4

# 数据源：从文件采集
[sources.app_logs]
type = "file"
include = ["/var/log/app/*.log"]
read_from = "beginning"

# 转换：解析 JSON 并添加字段
[transforms.parse_json]
type = "remap"
inputs = ["app_logs"]
source = '''
. = parse_json!(.message)
.hostname = get_hostname!()
.pipeline_ts = now()

# 敏感信息脱敏：手机号中间四位替换为 ****
if exists(.phone) {
    .phone = redact(.phone, filters: [r'\d{3}\d{4}\d{4}'], redactor: {"type": "text", "replacement": "***"})
}
'''

# 转换：按日志级别路由
[transforms.route_by_level]
type = "route"
inputs = ["parse_json"]

[transforms.route_by_level.route]
error = '.level == "ERROR" || .level == "FATAL"'
normal = '.level == "INFO" || .level == "WARN"'
debug = '.level == "DEBUG" || .level == "TRACE"'

# 输出：ERROR 日志发到专用 Topic
[sinks.error_logs]
type = "kafka"
inputs = ["route_by_level.error"]
bootstrap_servers = "kafka:9092"
topic = "logs-error"
encoding.codec = "json"
compression = "lz4"

# 输出：普通日志发到通用 Topic
[sinks.normal_logs]
type = "kafka"
inputs = ["route_by_level.normal"]
bootstrap_servers = "kafka:9092"
topic = "logs-normal"
encoding.codec = "json"
compression = "lz4"

# 输出：DEBUG 日志采样后发送（只保留 10%）
[transforms.sample_debug]
type = "sample"
inputs = ["route_by_level.debug"]
rate = 10

[sinks.debug_logs]
type = "kafka"
inputs = ["sample_debug"]
bootstrap_servers = "kafka:9092"
topic = "logs-debug"
encoding.codec = "json"

# 按命名空间分 Topic，而不是按服务分
# 按服务分会导致 Topic 数量爆炸（200 个微服务 = 200 个 Topic）
topics:
  - name: logs-production
    partitions: 30
    replication_factor: 3
    config:
      retention.ms: 259200000        # 3 天
      retention.bytes: 107374182400  # 100 GB per partition
      compression.type: lz4
      segment.bytes: 1073741824      # 1 GB per segment
      message.max.bytes: 1048576     # 1 MB 单条上限

  - name: logs-error
    partitions: 10
    replication_factor: 3
    config:
      retention.ms: 604800000        # 7 天，ERROR 日志保留更久
      compression.type: lz4

package asynclog

import (
    "context"
    "log/slog"
    "sync"
)

type AsyncHandler struct {
    inner  slog.Handler
    ch     chan slog.Record
    wg     sync.WaitGroup
    done   chan struct{}
}

func NewAsyncHandler(inner slog.Handler, bufferSize int) *AsyncHandler {
    h := &AsyncHandler{
        inner: inner,
        ch:    make(chan slog.Record, bufferSize),
        done:  make(chan struct{}),
    }
    h.wg.Add(1)
    go h.drain()
    return h
}

func (h *AsyncHandler) drain() {
    defer h.wg.Done()
    for {
        select {
        case r := <-h.ch:
            _ = h.inner.Handle(context.Background(), r)
        case <-h.done:
            // 排空缓冲区
            for {
                select {
                case r := <-h.ch:
                    _ = h.inner.Handle(context.Background(), r)
                default:
                    return
                }
            }
        }
    }
}

func (h *AsyncHandler) Handle(ctx context.Context, r slog.Record) error {
    select {
    case h.ch <- r:
        // 成功放入缓冲区
    default:
        // 缓冲区满，丢弃日志（不阻塞业务）
        // 这里可以增加一个 metrics 计数器来监控丢弃量
    }
    return nil
}

func (h *AsyncHandler) Enabled(ctx context.Context, level slog.Level) bool {
    return h.inner.Enabled(ctx, level)
}

func (h *AsyncHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
    return &AsyncHandler{inner: h.inner.WithAttrs(attrs), ch: h.ch, done: h.done}
}

func (h *AsyncHandler) WithGroup(name string) slog.Handler {
    return &AsyncHandler{inner: h.inner.WithGroup(name), ch: h.ch, done: h.done}
}

func (h *AsyncHandler) Close() {
    close(h.done)
    h.wg.Wait()
}

{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_size": "50gb",
            "max_age": "1d"
          },
          "set_priority": {
            "priority": 100
          }
        }
      },
      "warm": {
        "min_age": "2d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "forcemerge": {
            "max_num_segments": 1
          },
          "set_priority": {
            "priority": 50
          }
        }
      },
      "cold": {
        "min_age": "7d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "logs-s3-repo"
          },
          "set_priority": {
            "priority": 0
          }
        }
      },
      "delete": {
        "min_age": "30d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

# Loki 配置示例
auth_enabled: false

server:
  http_listen_port: 3100

common:
  path_prefix: /loki
  storage:
    filesystem:
      chunks_directory: /loki/chunks
      rules_directory: /loki/rules
  replication_factor: 1
  ring:
    kvstore:
      store: inmemory

schema_config:
  configs:
    - from: 2024-01-01
      store: tsdb
      object_store: s3
      schema: v13
      index:
        prefix: loki_index_
        period: 24h

storage_config:
  tsdb_shipper:
    active_index_directory: /loki/index
    cache_location: /loki/cache
  aws:
    s3: s3://us-east-1/loki-logs-bucket
    bucketnames: loki-logs-bucket
    region: us-east-1

limits_config:
  retention_period: 720h   # 30 天
  ingestion_rate_mb: 10
  ingestion_burst_size_mb: 20
  max_streams_per_user: 10000
  max_label_name_length: 1024
  max_label_value_length: 2048
  max_label_names_per_series: 30

chunk_store_config:
  chunk_cache_config:
    embedded_cache:
      enabled: true
      max_size_mb: 500

{
  "query": {
    "bool": {
      "must": [
        {"term": {"level": "ERROR"}},
        {"range": {"@timestamp": {"gte": "now-24h"}}}
      ]
    }
  },
  "aggs": {
    "error_distribution": {
      "terms": {
        "field": "error_code",
        "size": 20
      }
    }
  }
}

package sampling

import (
    "context"
    "log/slog"
    "sync/atomic"
)

// RateSamplingHandler 按固定比率采样日志
type RateSamplingHandler struct {
    inner    slog.Handler
    rate     uint64
    counter  atomic.Uint64
    minLevel slog.Level  // 高于此级别的日志不采样，全部保留
}

func NewRateSamplingHandler(inner slog.Handler, rate int, minLevel slog.Level) *RateSamplingHandler {
    return &RateSamplingHandler{
        inner:    inner,
        rate:     uint64(rate),
        minLevel: minLevel,
    }
}

func (h *RateSamplingHandler) Enabled(ctx context.Context, level slog.Level) bool {
    return h.inner.Enabled(ctx, level)
}

func (h *RateSamplingHandler) Handle(ctx context.Context, r slog.Record) error {
    // WARN 及以上级别不采样
    if r.Level >= h.minLevel {
        return h.inner.Handle(ctx, r)
    }

    // 对 INFO 和 DEBUG 进行采样
    n := h.counter.Add(1)
    if n%h.rate == 0 {
        r.AddAttrs(slog.Uint64("sample_rate", h.rate))
        return h.inner.Handle(ctx, r)
    }
    return nil
}

func (h *RateSamplingHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
    return &RateSamplingHandler{
        inner:    h.inner.WithAttrs(attrs),
        rate:     h.rate,
        minLevel: h.minLevel,
    }
}

func (h *RateSamplingHandler) WithGroup(name string) slog.Handler {
    return &RateSamplingHandler{
        inner:    h.inner.WithGroup(name),
        rate:     h.rate,
        minLevel: h.minLevel,
    }
}

package sampling

import (
    "context"
    "math/rand"
)

type sampledKey struct{}

// ShouldSample 在请求入口调用，决定是否采样
func ShouldSample(ctx context.Context, rate float64) context.Context {
    sampled := rand.Float64() < rate
    return context.WithValue(ctx, sampledKey{}, sampled)
}

// IsSampled 在日志处理器中调用，判断当前请求是否被采样
func IsSampled(ctx context.Context) bool {
    v, ok := ctx.Value(sampledKey{}).(bool)
    if !ok {
        return true // 没有采样标记的日志默认保留
    }
    return v
}

package sampling

import (
    "sync"
    "time"
)

// AdaptiveSampler 根据日志速率动态调整采样率
type AdaptiveSampler struct {
    mu          sync.Mutex
    threshold   int     // 每秒日志条数阈值
    count       int     // 当前窗口的计数
    windowStart time.Time
    currentRate float64 // 当前采样率，1.0 表示全保留
}

func NewAdaptiveSampler(threshold int) *AdaptiveSampler {
    return &AdaptiveSampler{
        threshold:   threshold,
        windowStart: time.Now(),
        currentRate: 1.0,
    }
}

func (s *AdaptiveSampler) ShouldLog() bool {
    s.mu.Lock()
    defer s.mu.Unlock()

    now := time.Now()
    elapsed := now.Sub(s.windowStart)

    // 每秒重置窗口
    if elapsed >= time.Second {
        ratePerSec := float64(s.count) / elapsed.Seconds()
        if ratePerSec > float64(s.threshold) {
            s.currentRate = float64(s.threshold) / ratePerSec
        } else {
            s.currentRate = 1.0
        }
        s.count = 0
        s.windowStart = now
    }

    s.count++

    if s.currentRate >= 1.0 {
        return true
    }
    // 使用计数器取模来决定是否采样，避免 rand 的锁竞争
    return s.count%int(1.0/s.currentRate) == 0
}

package main

import (
    "encoding/json"
    "log/slog"
    "net/http"
    "os"
)

var logLevel = new(slog.LevelVar) // 默认 INFO

func init() {
    handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
        Level: logLevel,
    })
    slog.SetDefault(slog.New(handler))
}

type levelRequest struct {
    Level string `json:"level"`
}

func handleSetLevel(w http.ResponseWriter, r *http.Request) {
    if r.Method != http.MethodPut {
        http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
        return
    }

    var req levelRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, "invalid request body", http.StatusBadRequest)
        return
    }

    var level slog.Level
    if err := level.UnmarshalText([]byte(req.Level)); err != nil {
        http.Error(w, "invalid log level", http.StatusBadRequest)
        return
    }

    logLevel.Set(level)
    slog.Info("log level changed", slog.String("new_level", req.Level))

    w.WriteHeader(http.StatusOK)
    json.NewEncoder(w).Encode(map[string]string{
        "level": logLevel.Level().String(),
    })
}

func main() {
    http.HandleFunc("/admin/log-level", handleSetLevel)

    slog.Info("server starting", slog.String("addr", ":8080"))
    http.ListenAndServe(":8080", nil)
}

# 临时开启 DEBUG 级别
curl -X PUT http://order-service:8080/admin/log-level \
  -H "Content-Type: application/json" \
  -d '{"level": "DEBUG"}'

# 恢复为 INFO 级别
curl -X PUT http://order-service:8080/admin/log-level \
  -H "Content-Type: application/json" \
  -d '{"level": "INFO"}'

# application.yml
management:
  endpoints:
    web:
      exposure:
        include: loggers
  endpoint:
    loggers:
      enabled: true

# 查看当前日志级别
curl http://order-service:8080/actuator/loggers/com.example.order

# 修改日志级别
curl -X POST http://order-service:8080/actuator/loggers/com.example.order \
  -H "Content-Type: application/json" \
  -d '{"configuredLevel": "DEBUG"}'

package conditional

import (
    "context"
    "log/slog"
)

type debugUserKey struct{}

// WithDebugUser 在中间件中为特定用户开启详细日志
func WithDebugUser(ctx context.Context, userID string, debugUsers map[string]bool) context.Context {
    if debugUsers[userID] {
        return context.WithValue(ctx, debugUserKey{}, true)
    }
    return ctx
}

// ConditionalHandler 根据上下文条件决定是否记录 DEBUG 日志
type ConditionalHandler struct {
    inner slog.Handler
}

func (h *ConditionalHandler) Enabled(ctx context.Context, level slog.Level) bool {
    if level >= slog.LevelInfo {
        return h.inner.Enabled(ctx, level)
    }
    // DEBUG 级别只对标记的请求开启
    if v, ok := ctx.Value(debugUserKey{}).(bool); ok && v {
        return true
    }
    return false
}

func (h *ConditionalHandler) Handle(ctx context.Context, r slog.Record) error {
    return h.inner.Handle(ctx, r)
}

func (h *ConditionalHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
    return &ConditionalHandler{inner: h.inner.WithAttrs(attrs)}
}

func (h *ConditionalHandler) WithGroup(name string) slog.Handler {
    return &ConditionalHandler{inner: h.inner.WithGroup(name)}
}

package main

import (
    "go.uber.org/zap"
    "go.uber.org/zap/zapcore"
    "time"
)

func main() {
    config := zap.Config{
        Level:       zap.NewAtomicLevelAt(zap.InfoLevel),
        Development: false,
        Encoding:    "json",
        EncoderConfig: zapcore.EncoderConfig{
            TimeKey:        "timestamp",
            LevelKey:       "level",
            NameKey:        "logger",
            CallerKey:      "caller",
            MessageKey:     "message",
            StacktraceKey:  "stacktrace",
            LineEnding:     zapcore.DefaultLineEnding,
            EncodeLevel:    zapcore.CapitalLevelEncoder,
            EncodeTime:     zapcore.ISO8601TimeEncoder,
            EncodeDuration: zapcore.MillisDurationEncoder,
            EncodeCaller:   zapcore.ShortCallerEncoder,
        },
        OutputPaths:      []string{"stdout"},
        ErrorOutputPaths: []string{"stderr"},
    }

    logger, _ := config.Build()
    defer logger.Sync()

    // 强类型结构化日志
    logger.Info("order created",
        zap.String("order_id", "ORD-98765"),
        zap.String("user_id", "12345"),
        zap.Int("item_count", 3),
        zap.Float64("total_usd", 99.99),
        zap.Duration("processing_time", 23*time.Millisecond),
        zap.String("trace_id", "abc123def456"),
    )
}

package sanitize

import (
    "regexp"
    "strings"
)

var (
    phonePattern    = regexp.MustCompile(`1[3-9]\d{9}`)
    idCardPattern   = regexp.MustCompile(`\d{17}[\dXx]`)
    emailPattern    = regexp.MustCompile(`[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}`)
    creditCardPattern = regexp.MustCompile(`\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}`)
)

// SanitizeField 对字段值进行脱敏
func SanitizeField(key, value string) string {
    switch {
    case strings.Contains(key, "phone") || strings.Contains(key, "mobile"):
        return maskPhone(value)
    case strings.Contains(key, "id_card") || strings.Contains(key, "identity"):
        return maskIDCard(value)
    case strings.Contains(key, "email"):
        return maskEmail(value)
    case strings.Contains(key, "card_number") || strings.Contains(key, "credit_card"):
        return maskCreditCard(value)
    case strings.Contains(key, "password") || strings.Contains(key, "secret") || strings.Contains(key, "token"):
        return "***REDACTED***"
    }
    return value
}

func maskPhone(phone string) string {
    if len(phone) != 11 {
        return "***"
    }
    return phone[:3] + "****" + phone[7:]
}

func maskIDCard(id string) string {
    if len(id) < 8 {
        return "***"
    }
    return id[:4] + "**********" + id[len(id)-4:]
}

func maskEmail(email string) string {
    parts := strings.Split(email, "@")
    if len(parts) != 2 {
        return "***"
    }
    name := parts[0]
    if len(name) <= 2 {
        return "**@" + parts[1]
    }
    return name[:2] + "***@" + parts[1]
}

func maskCreditCard(card string) string {
    clean := strings.ReplaceAll(strings.ReplaceAll(card, " ", ""), "-", "")
    if len(clean) < 8 {
        return "***"
    }
    return clean[:4] + " **** **** " + clean[len(clean)-4:]
}

[transforms.sanitize_pii]
type = "remap"
inputs = ["parse_json"]
source = '''
# 手机号脱敏
if exists(.phone) {
    .phone = redact(.phone, filters: [r'(1[3-9]\d)\d{4}(\d{4})'], redactor: {"type": "text", "replacement": "$1****$2"})
}

# password 字段直接删除
if exists(.password) {
    del(.password)
}

# 请求体中的敏感信息
if exists(.request_body) {
    .request_body = redact(.request_body, filters: [
        r'"password"\s*:\s*"[^"]*"',
        r'"token"\s*:\s*"[^"]*"'
    ], redactor: {"type": "text", "replacement": "\"***REDACTED***\""})
}
'''

{
  "Rules": [
    {
      "ObjectLockEnabled": "Enabled",
      "DefaultRetention": {
        "Mode": "COMPLIANCE",
        "Days": 365
      }
    }
  ]
}

package auditlog

import (
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "sync"
    "time"
)

type AuditEntry struct {
    Sequence  uint64 `json:"seq"`
    Timestamp string `json:"ts"`
    Event     string `json:"event"`
    Actor     string `json:"actor"`
    Detail    string `json:"detail"`
    PrevHash  string `json:"prev_hash"`
    Hash      string `json:"hash"`
}

type AuditLogger struct {
    mu       sync.Mutex
    seq      uint64
    prevHash string
}

func NewAuditLogger() *AuditLogger {
    return &AuditLogger{
        prevHash: "0000000000000000000000000000000000000000000000000000000000000000",
    }
}

func (l *AuditLogger) Log(event, actor, detail string) AuditEntry {
    l.mu.Lock()
    defer l.mu.Unlock()

    l.seq++
    entry := AuditEntry{
        Sequence:  l.seq,
        Timestamp: time.Now().UTC().Format(time.RFC3339Nano),
        Event:     event,
        Actor:     actor,
        Detail:    detail,
        PrevHash:  l.prevHash,
    }

    payload, _ := json.Marshal(map[string]interface{}{
        "seq":       entry.Sequence,
        "ts":        entry.Timestamp,
        "event":     entry.Event,
        "actor":     entry.Actor,
        "detail":    entry.Detail,
        "prev_hash": entry.PrevHash,
    })
    hash := sha256.Sum256(payload)
    entry.Hash = hex.EncodeToString(hash[:])
    l.prevHash = entry.Hash

    return entry
}

# elasticsearch-roles.yml
payment_team:
  cluster: []
  indices:
    - names: ["logs-payment-*"]
      privileges: ["read"]
      field_security:
        grant: ["*"]
        except: ["credit_card_number", "cvv"]

general_engineer:
  cluster: []
  indices:
    - names: ["logs-*"]
      privileges: ["read"]
      field_security:
        grant: ["timestamp", "level", "message", "service",
                "trace_id", "span_id", "request_id",
                "http_method", "http_path", "http_status_code",
                "duration_ms", "error_code"]

# kubelet 配置
containerLogMaxSize: "100Mi"
containerLogMaxFiles: 5

# Vector 配置：移除高体积低价值的字段
[transforms.trim_fields]
type = "remap"
inputs = ["parse_json"]
source = '''
# 移除大体积的请求/响应体（只保留前 500 字符的摘要）
if exists(.request_body) {
    body = string!(.request_body)
    if length(body) > 500 {
        .request_body_preview = slice!(body, 0, 500)
        del(.request_body)
    }
}

# 移除冗余的 Kubernetes 元数据
del(.kubernetes.labels)
del(.kubernetes.annotations)
del(.kubernetes.node_labels)

# 移除重复的时间戳字段（保留 timestamp，移除 @timestamp 和 time）
del(."@timestamp")
del(.time)
'''

# 日志配额配置示例
quotas:
  default:
    max_bytes_per_day: 10GB
    action_on_exceed: sample  # sample | drop | alert
    sample_rate: 0.1          # 超额后只保留 10%

  overrides:
    - service: payment-service
      max_bytes_per_day: 50GB  # 支付服务配额更高
      action_on_exceed: alert  # 超额只告警不丢弃

    - service: health-check
      max_bytes_per_day: 1GB
      action_on_exceed: drop   # 健康检查日志超额直接丢弃

# Grafana 数据源配置
apiVersion: 1
datasources:
  - name: Loki
    type: loki
    url: http://loki:3100
    jsonData:
      derivedFields:
        # 从日志中提取 trace_id，生成到 Tempo 的跳转链接
        - datasourceUid: tempo
          matcherRegex: '"trace_id":"(\w+)"'
          name: TraceID
          url: '$${__value.raw}'
          urlDisplayLabel: "View Trace"

  - name: Tempo
    type: tempo
    uid: tempo
    url: http://tempo:3200
    jsonData:
      tracesToLogs:
        datasourceUid: loki
        tags: ["service"]
        mappedTags:
          - key: service.name
            value: service
        filterByTraceID: true
        filterBySpanID: true

[transforms.extract_metrics]
type = "log_to_metric"
inputs = ["parse_json"]

[[transforms.extract_metrics.metrics]]
type = "counter"
field = "level"
name = "log_lines_total"
tags.service = "{{service}}"
tags.level = "{{level}}"

[[transforms.extract_metrics.metrics]]
type = "histogram"
field = "duration_ms"
name = "request_duration_ms"
tags.service = "{{service}}"
tags.http_method = "{{http_method}}"

[sinks.prometheus_metrics]
type = "prometheus_exporter"
inputs = ["extract_metrics"]
address = "0.0.0.0:9598"

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluent-bit
  namespace: logging
  labels:
    app: fluent-bit
spec:
  selector:
    matchLabels:
      app: fluent-bit
  template:
    metadata:
      labels:
        app: fluent-bit
    spec:
      serviceAccountName: fluent-bit
      tolerations:
        - operator: Exists
      containers:
        - name: fluent-bit
          image: fluent/fluent-bit:3.0
          resources:
            requests:
              cpu: 100m
              memory: 128Mi
            limits:
              cpu: 500m
              memory: 256Mi
          volumeMounts:
            - name: varlog
              mountPath: /var/log
              readOnly: true
            - name: containers
              mountPath: /var/lib/docker/containers
              readOnly: true
            - name: config
              mountPath: /fluent-bit/etc/
      volumes:
        - name: varlog
          hostPath:
            path: /var/log
        - name: containers
          hostPath:
            path: /var/lib/docker/containers
        - name: config
          configMap:
            name: fluent-bit-config

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluent-bit-config
  namespace: logging
data:
  fluent-bit.conf: |
    [SERVICE]
        Flush         5
        Log_Level     info
        Daemon        off
        Parsers_File  parsers.conf
        HTTP_Server   On
        HTTP_Listen   0.0.0.0
        HTTP_Port     2020

    [INPUT]
        Name              tail
        Tag               kube.*
        Path              /var/log/containers/*.log
        Parser            cri
        DB                /var/log/flb_kube.db
        Mem_Buf_Limit     10MB
        Skip_Long_Lines   On
        Refresh_Interval  10

    [FILTER]
        Name                kubernetes
        Match               kube.*
        Kube_URL            https://kubernetes.default.svc:443
        Kube_CA_File        /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
        Kube_Token_File     /var/run/secrets/kubernetes.io/serviceaccount/token
        Merge_Log           On
        Merge_Log_Key       log_processed
        K8S-Logging.Parser  On
        K8S-Logging.Exclude On

    [FILTER]
        Name    modify
        Match   kube.*
        Remove  stream
        Remove  logtag

    [OUTPUT]
        Name            kafka
        Match           kube.*
        Brokers         kafka-0.kafka:9092,kafka-1.kafka:9092,kafka-2.kafka:9092
        Topics          logs-kubernetes
        Format          json
        Timestamp_Key   @timestamp
        rdkafka.compression.type lz4
        rdkafka.queue.buffering.max.messages 100000
        rdkafka.queue.buffering.max.ms 1000

  parsers.conf: |
    [PARSER]
        Name        cri
        Format      regex
        Regex       ^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<logtag>[^ ]*) (?<log>.*)$
        Time_Key    time
        Time_Format %Y-%m-%dT%H:%M:%S.%L%z

logger.Error("request failed",
    zap.Error(err),                    // 自动捕获 error message
    zap.String("stacktrace", string(debug.Stack())), // 堆栈作为字段
)