如何将CloudWatch缺失的历史日志导入Elasticsearch且不产生重复数据？

最优实现方式为独立运行一次性定向同步任务，与现有生产实时同步的Logstash实例完全隔离，既不影响正常实时日志同步，也可精准拉取目标时间范围数据，从根源避免重复数据生成，具体操作如下： #

操作步骤 #

1. 准备独立的一次性同步配置文件 #

和现有生产用的Logstash配置完全分开，避免影响实时业务，配置分为三个模块：

input 模块（精准限定拉取范围）

使用logstash-input-cloudwatch_logs插件的时间范围参数直接锁定缺失区间，不用修改原有的sincedb配置：

input {
  cloudwatch_logs {
    log_group_name => ["你要同步的日志组1", "你要同步的日志组2"]
    # 替换为你实际缺失的时间范围，格式为ISO8601标准UTC时间
    start_time => "202X-XX-XXT00:00:00Z"
    end_time => "202X-XX-XXT23:59:59Z"
    # 数值设置为大于你缺失的时间跨度即可，单位为秒，示例为31天的秒数，不要设为0
    ignore_older => 2678400
    region => "你的AWS区域"
    # 其他和生产一致的access_key/secret_key配置保留即可
  }
}

filter 模块（幂等去重配置）

基于CloudWatch原生的唯一事件ID生成ES文档ID，保证同一条日志写入时只会覆盖不会重复：

filter {
  fingerprint {
    source => ["[cloudwatch][log_group]", "[cloudwatch][event_id]"]
    target => "[@metadata][document_id]"
    method => "SHA256"
  }
}

output 模块（保持和生产索引格式一致）

和现有生产的索引命名规则完全匹配，且指定上述生成的唯一ID作为文档ID：

output {
  elasticsearch {
    hosts => ["你的ES集群地址:9200"]
    # 替换为你生产环境用的日志索引前缀，保持格式一致
    index => "your-log-index-prefix-%{+YYYY.MM.dd}"
    document_id => "%{[@metadata][document_id]}"
    # 如有ES认证，补充user、password配置即可
  }
  # 调试阶段可打开stdout输出查看同步进度，正式运行可注释
  stdout { codec => rubydebug }
}

2. 单独启动一次性同步容器 #

不要修改或重启现有跑实时同步的Logstash容器，使用上述准备好的配置文件单独启动新的Logstash容器执行同步任务，任务跑完后容器会自动退出，完全不影响现有业务。

3. 同步完成校验 #

在Kibana或者ES接口中统计目标时间范围的日志条数，和CloudWatch控制台对应时间范围的日志条数对比，一致即同步完成。 #

方案优势 #

• 与现有生产链路完全隔离，无操作风险，不会影响正常实时日志同步
• 通过start_time/end_time参数精准锁定缺失范围，不会出现拉取范围不符合预期的问题
• 基于CloudWatch原生事件ID实现幂等写入，从机制上完全避免重复数据生成
• 无需修改现有生产的sincedb配置，不需要调整实时同步链路的任何参数

内容的提问来源于stack exchange，提问作者Ezdz