从 AWS SQS 接收消息并在 AWS S3 中批量存储的 Logstash 替代方案

Question

我需要能够将日志作为批处理存储在 AWS S3 中，作为格式适合 JSON-SerDe 的文本文件。

其中一个批处理日志文件在 S3 上的外观示例，非常重要的是日期时间格式为 yyyy-MM-dd HH:mm:ss

{"message":"Message number 1","datetime":"2020-12-01 14:37:00"}
{"message":"Message number 2","datetime":"2020-12-01 14:38:00"}
{"message":"Message number 3","datetime":"2020-12-01 14:39:00"}

理想情况下，这些信息将每 5 秒或当排队消息达到 50 时存储在 S3 上，但也可以配置。

---

我几乎已经设法使用以下配置使用 sqs input plugin 和 s3 output plugin 与 Logstash 一起工作

input {
  sqs {
    endpoint => "AWS_SQS_ENDPOINT"
    queue => "logs"
  }
}

output {
   s3 {
     access_key_id => "AWS_ACCESS_KEY_ID"
     secret_access_key => "AWS_SECRET_ACCESS_KEY"
     region => "AWS_REGION"
     bucket => "AWS_BUCKET"
     prefix => "audit/year=%{+YYYY}/month=%{+MM}/day=%{+dd}/"
     size_file => 128
     time_file => 5
     codec => "json_lines"
     encoding => "gzip"
     canned_acl => "private"
   }
}

问题是 S3 输出插件需要 @timestamp 字段，这与我们的查询工具不兼容。如果您使用 mutate 过滤器删除 @timestamp 或更改为 datetime，则它不会处理日志。我们不能为每条记录存储日期时间字段和@timestamp，因为这会大大增加我们需要存储的数据量（数百万条日志）。

是否有其他软件替代品可以实现这一结果？

---

感谢 [Badger][https://stackoverflow.com/users/11792977/badger] 更新了与 Logstash 配合使用的配置

input {
  sqs {
    endpoint => "http://AWS_SQS_ENDPOINT"
    queue => "logs"
  }
}

filter {
  mutate {
    add_field => {
      "[@metadata][year]" => "%{+YYYY}"
      "[@metadata][month]" => "%{+MM}"
      "[@metadata][day]" => "%{+dd}"
    }
    remove_field => [ "@timestamp" ]
  }
}

output {
   s3 {
     access_key_id => "AWS_ACCESS_KEY_ID"
     secret_access_key => "AWS_SECRET_ACCESS_KEY"
     region => "AWS_REGION"
     bucket => "AWS_BUCKET"
     prefix => "audit/year=%{[@metadata][year]}/month=%{[@metadata][month]}/day=%{[@metadata][day]}"
     # 1 MB
     size_file => 1024
     # 1 Minute
     time_file => 1
     codec => "json_lines"
     encoding => "gzip"
     canned_acl => "private"
   }
}

Answer 1

我在 s3 输出代码中没有看到对 @timestamp 的任何依赖。您已经通过在prefix => "audit/year=%{+YYYY}/month=%{+MM}/day=%{+dd}/" 中使用对它的 sprintf 引用创建了一个。您可以将这些 sprintf 引用移动到 mutate+add_field 过滤器，该过滤器将字段添加到 [@metadata]，然后删除 @timestamp，然后在前缀选项中引用 [@metadata] 字段。