突破多行日志解析难题：TKE容器内Java堆栈采集的方案

graph TD
    A[应用日志] --> B(Sidecar Agent)
    B --> C{日志分类器}
    C -->|单行| D[标准输出流]
    C -->|多行| E[临时缓冲区]
    E --> F[批量拼接器]
    F --> G[结构化输出]

// 自定义日志分隔符检测算法核心逻辑
public class MultiLineDetector {
    private static final Pattern STACK_TRACE_PATTERN = 
        Pattern.compile("^\\s+at\\s+\\w+\\.\\w+\\(\\w+\\.java:\\d+\\)$");
    
    public boolean isContinuation(String line) {
        return line.trim().startsWith("at ") || 
               line.contains("Caused by:") ||
               line.contains("... ");
    }
}

# daemonset.yaml 核心配置片段
apiVersion: apps/v1
kind: DaemonSet
spec:
  template:
    spec:
      containers:
      - name: log-sidecar
        image: custom/log-agent:1.2.3
        env:
        - name: LOG_PATTERN
          value: "%d{ISO8601} [%t] %-5level %logger{36} - %msg%n"
        resources:
          limits:
            cpu: 200m
            memory: 256Mi

# logstash pipeline 配置示例
filter {
  if [message] =~ /^\s+at\s/ {
    grok {
      match => {
        "message" => "%{SPACE}at %{JAVACLASS:class}\.%{JAVAMETHOD:method}\(%{JAVAFILE:file}\:%{NUMBER:line}\)"
      }
      tag_on_failure => []
    }
    mutate {
      add_field => { "log_type" => "stacktrace" }
    }
  }
}

# 动态缓冲区调整算法
def adjust_buffer_size(current_size, error_rate):
    if error_rate > 0.2:
        return min(current_size * 1.5, MAX_BUFFER_SIZE)
    elif error_rate < 0.05 and current_size > MIN_BUFFER_SIZE:
        return max(int(current_size * 0.8), MIN_BUFFER_SIZE)
    return current_size

# 配置日志轮转策略
/app/logs/*.log {
    daily
    missingok
    rotate 7
    compress
    delaycompress
    sharedscripts
    postrotate
        # 发送SIGHUP信号刷新采集器
        pkill -HUP log-agent
    endscript
}

# 条件路由配置示例
if [log_type] == "stacktrace" {
    # 多行处理管道
} else {
    # 标准处理管道
}

// 自定义压缩算法伪代码
public class ContextAwareCompressor {
    public byte[] compress(List<String> logs) {
        // 1. 识别堆栈边界
        // 2. 对完整堆栈使用DEFLATE
        // 3. 对单行日志使用Snappy
        // 4. 添加元数据头
    }
}

# 自适应采样算法
def adaptive_sampling(current_rate, error_count):
    if error_count > 100:
        return max(current_rate * 0.8, MIN_SAMPLE_RATE)
    elif error_count < 10:
        return min(current_rate * 1.2, MAX_SAMPLE_RATE)
    return current_rate