Альтернатива Logstash для получения сообщений от AWS SQS и пакетного хранилища в AWS S3

Мне нужна возможность хранить журналы в виде пакетов в 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"}

В идеале они должны храниться на S3 каждые 5 секунд или когда сообщения в очереди достигают 50, но также могут быть настраиваемыми.

Мне почти удалось заставить это работать с Logstash, используя плагин ввода sqs и вывод s3 плагин, используя приведенную ниже конфигурацию

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 или изменения даты и времени, он не будет обрабатывать журналы. Мы не можем хранить поле даты и времени и @timestamp для каждой записи, так как это резко увеличивает объем данных, которые нам нужно хранить (миллионы журналов).

Существуют ли какие-либо другие программные альтернативы для достижения этого результата?

Обновлена ​​конфигурация, которая работает с Logstash благодаря [Badger][https://stackoverflow.com/users/11792977/badger]

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"
   }
}