В одном из проектов встала следующая задача: пользователь загружает пачку файлов через клиента (CloudBerry Explorer, к примеру) в S3 бакет, мы копируем эти файлы в архив и шлем SNS уведомление о том, что все сделано. Перекладывать файлы в архив нужно начинать только тогда, когда пользователь загрузит все, что хотел. Пользователей мало и загружают батчи они довольно редко. Но файлов может быть много.
Чтобы понять, что пора начинать архивацию, зададим определенную структуру каталогов и будем просить пользователя загружать триггер-файлы с расширением .trigger когда он закончит. Этакая эмуляция кнопки Done. Структура каталогов будет такой:
<batch_name>/done.trigger
<batch_name>/files/<file_key_1>
<batch_name>/files/<file_key_2>
...
<batch_name>/files/<file_key_n>
Как видим, для каждой пачки создается свой каталог <batch_name> с подкаталогом files, в который и заливаются уже пользовательские файлы с каталогами и именами, которые он хочет. Триггер-файл загружается в <batch_name> и по ключу этого файла можно понять какие конкретно файлы нужно отправить в архив. Но здесь есть один нюанс, мы хотим при копировании в архив вырезать каталог files. Т.е. файл <batch_name>/files/<file_key_1> скопировать в <batch_name>/<file_key_1>.
К счастью, S3 позволяет отслеживать загрузку файлов с определенным суффиксом и отправлять уведомления при наструплении этого события. В качестве получаетеля этих уведомлений можно указать аж 3 сервиса: SNS, SQS и Lambda-функцию. Но тут не без нюансов. Так, первые 2 типа поддерживают только стандартные очереди и SNS, а FIFO не поддерживают, увы.
Мы же, в случае загрузки триггера, будем ставить задачу в стандартную очередь SQS и в приложении будем эту очередь опрашивать. На питоне это выглядит примерно так:
…
session = boto3.Session(profile_name=profile)
queue = session.resource('sqs', region_name=region).Queue(queue_url)
while True:
# Здесь можно усыпить процесс, если хочется
messages = queue.receive_messages(AttributeNames=('ALL',), WaitTimeSeconds=20)
if not messages:
continue
message = messages[0]
# Process message
Окей, мы разобрались как понять, что пользователь закончил заливать свои файлы и пора их архивировать.
Теперь нам нужно скопировать файлы в архив и отправить SNS уведомление. Уведомление нужно отправлять только после того как все файлы уедут в архив. Что нам может предложить для такого случая Амазон?
S3 Batch Operations
Амазон позволяет копировать объекты батчами между бакетами и даже между аккаунтами. Но функционал этот довольно ограничен. Рассмотрим чуть подробнее.
Поставить задачу на копирование можно двумя способами: с использованием отчета об инвентаризации и с помощью CSV манифеста.
Оба способа очень похожи друг на друга и обладают одним и тем же недостатком - они оба недостаточно гибкие. Так к примеру, при постановке задачи на копирование мы можем указать список копируемых файлов из батча в CSV манифесте, а так же можем указать каталог куда все эти файлы скопировать, но выбрать произвольное имя или вообще как-то модифицировать ключ, увы, нельзя. К примеру, если у нас в файле указаны загруженные пользователем файлы из примера выше,<batch_name>/done.trigger
<batch_name>/files/<file_key_1>
<batch_name>/files/<file_key_2>
...
<batch_name>/files/<file_key_n>
то можно все эти файлы загрузить в каталог, скажем, archive и получить пути вида archive/<batch_name>/files/<file_key_1> , но вот каталог files из пути не выдрать, т.е. файл <batch_name>/files/<file_key_1> скопировать в archive/<batch_name>/<file_key_1>, увы, нельзя.
В случае с использованием сервиса инвентаризации все то же самое, только вместо манифеста - отчет с похожим содержимым. А очень бы хотелось на самом деле.
К слову, задачи на копирование батчами можно ставить как программно, так и через Консоль.
SDK Python
Когда гибкости стандартных S3 Batch Operations не хватает, Амазон советует пользоваться SDK, что мы и сделаем. У нас питон, а для питона синхронная либа boto3. boto3 под капотом делает вызовы к REST API амазоновских сервисов, поэтому операция копирования файла сводится к отправке HTTP запроса, а это значит, что мы можем немного распараллелить процесс копирования. Однопоточный синхронный код значительно проще конечно, что мне очень импонирует, но его производительности, к сожалению, будет не достаточно, т.к. файлов может быть много, а очень долго ждать не хочется. В итоге имеем вот такой простенький скриптик:
from concurrent.futures import as_completed
from concurrent.futures.thread import ThreadPoolExecutor
import boto3
import click
def get_source_prefix(message_body: str) -> str:
...
def modify_key(key) -> str:
...
@click.command()
@click.option('--queue-url')
@click.option('--source-bucket')
@click.option('--target-bucket')
@click.option('--target-prefix')
@click.option('--max-workers')
@click.option('--region')
@click.option('--profile')
def main(queue_url, source_bucket, target_bucket, target_prefix, max_workers, region, profile):
session = boto3.Session(profile_name=profile)
queue = session.resource('sqs', region_name=region).Queue(queue_url)
s3_client = session.client('s3')
paginator = s3_client.get_paginator('list_objects_v2')
while True:
# Здесь можно усыпить процесс, если очень хочется
# По умолчанию, метод возвращает одно сообщение
messages = queue.receive_messages(AttributeNames=('ALL',), WaitTimeSeconds=20)
if not messages:
continue
message = messages[0]
kwargs = {'Bucket': source_bucket, 'Prefix': f'{get_source_prefix(message.body)}/files'}
keys = [k for page in paginator.paginate(**kwargs) for k in page.get('Contents', [])]
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures_to_keys = {
executor.submit(
s3_client.copy,
{'Bucket': source_bucket, 'Key': key['Key']},
target_bucket,
f'{target_prefix}/{modify_key(key["Key"])}'
): key
for key in keys if not key['Key'].endswith('.trigger')
}
as_completed(futures_to_keys)
# шлем SNS уведомление
...
message.delete()
if __name__ == '__main__':
main()
