日志采样与存储成本优化：不是所有日志都值得保留 30 天-平芜编程栈

日志采样与存储成本优化：不是所有日志都值得保留 30 天

一、月均日志存储账单 8000 元的反思

一个 20 个微服务的系统，每天产生约 200GB 日志。ELK 集群的存储费用月均 8000 元，其中超过 60% 的日志在过去 90 天内从未被查询过。运维团队翻了一遍，发现大量是 DEBUG 级别的"收到请求"、"参数校验通过"这类毫无价值的日志。

更隐蔽的问题是：日志量越大，ES 的索引和查询性能越差。P99 查询延迟从 200ms 涨到了 3 秒，告警已经不准了。

根本原因不是日志太多，而是对日志价值的判断机制缺失。不是所有日志都值得全量保留，也不是所有日志都需要 30 天保留周期。

二、日志分级采样策略

核心原则：日志级别×查詢頻率×合规要求= 保留策略

三、Go 实现自适应日志采样

package log import ( "crypto/sha256" "fmt" "sync" "time" ) // LogLevel 日志级别 type LogLevel int const ( DEBUG LogLevel = iota INFO WARN ERROR FATAL ) // SamplingConfig 采样配置 type SamplingConfig struct { // ERR/FATAL 级别：全量保留 ErrorRate float64 // 固定 1.0（100%） // WARN 级别：按比例保留 WarnRate float64 // 建议 0.5（50%） // INFO 级别：自适应采样 InfoWindow time.Duration // 时间窗口，默认 5 秒 InfoMaxPerWindow int // 窗口内最多保留条数 // DEBUG 级别：默认丢弃 } // AdaptiveSampler 自适应日志采样器 type AdaptiveSampler struct { config SamplingConfig // 去重缓存：相同日志模式在一个窗口内只保留固定条数 mu sync.Mutex counters map[string]*windowCounter } type windowCounter struct { count int resetTime time.Time } // 常见日志模式的哈希模板（用于去重） // 例如 "收到请求: %s" → 哈希后归入同一个计数器 type LogPattern string func NewAdaptiveSampler(config SamplingConfig) *AdaptiveSampler { if config.InfoWindow == 0 { config.InfoWindow = 5 * time.Second } if config.InfoMaxPerWindow == 0 { config.InfoMaxPerWindow = 5 // 默认每种日志 5 秒内最多保留 5 条 } return &AdaptiveSampler{ config: config, counters: make(map[string]*windowCounter), } } // ShouldSample 判断一条日志是否应该被保留 // msg: 日志内容 // level: 日志级别 // 返回 true 表示保留，false 表示丢弃 func (as *AdaptiveSampler) ShouldSample(msg string, level LogLevel) bool { switch level { case ERROR, FATAL: return true // 全量保留 case WARN: return as.sampleByRate(msg, as.config.WarnRate) case INFO: return as.adaptiveSample(msg) case DEBUG: return false // 生产环境默认丢弃 DEBUG default: return true } } // sampleByRate 基于固定比率的采样 func (as *AdaptiveSampler) sampleByRate(_ string, rate float64) bool { // 使用时间戳微秒部分的哈希做伪随机 hashInput := fmt.Sprintf("%d", time.Now().UnixNano()) hash := sha256.Sum256([]byte(hashInput)) // 取第一个字节映射到 [0, 1) sample := float64(hash[0]) / 256.0 return sample < rate } // adaptiveSample 自适应采样，同类日志去重 func (as *AdaptiveSampler) adaptiveSample(msg string) bool { pattern := extractPattern(msg) as.mu.Lock() defer as.mu.Unlock() now := time.Now() counter, exists := as.counters[pattern] // 窗口过期，重置计数器 if !exists || now.After(counter.resetTime) { as.counters[pattern] = &windowCounter{ count: 1, resetTime: now.Add(as.config.InfoWindow), } return true // 窗口第一条日志一定保留 } // 窗口内计数已达上限，丢弃 if counter.count >= as.config.InfoMaxPerWindow { return false } counter.count++ return true } // extractPattern 提取日志模式（去除变量部分） func extractPattern(msg string) string { // 生产环境建议使用正则提取变量部分并替换为占位符 // 简化版：取前 80 个字符的哈希作为模式标识 if len(msg) > 80 { msg = msg[:80] } hash := sha256.Sum256([]byte(msg)) return fmt.Sprintf("%x", hash[:8]) }

集成到日志库中

// SampledLogger 带采样的日志器 type SampledLogger struct { sampler *AdaptiveSampler writer io.Writer // 实际写入目标（ES、文件等） mu sync.Mutex } func (sl *SampledLogger) Info(format string, args ...interface{}) { msg := fmt.Sprintf(format, args...) if !sl.sampler.ShouldSample(msg, INFO) { return // 被采样丢弃 } sl.mu.Lock() defer sl.mu.Unlock() fmt.Fprintf(sl.writer, "[INFO] %s %s\n", time.Now().Format(time.RFC3339), msg) }