Облачное хранилище

Доступ к S3-совместимому объектному хранилищу. Загрузка, скачивание, перечисление и управление объектами, presigned URL для скачивания, загрузки и частей multipart, а также чтение объектов с произвольным доступом.

Настройку хранилища см. в Cloud Storage.

Загрузка

local cloudstorage = require("cloudstorage")

Получение хранилища

Получить ресурс облачного хранилища по ID реестра:

local storage, err = cloudstorage.get("app.infra:files")
if err then
    return nil, err
end

storage:upload_object("data/file.txt", "content")
storage:release()

Возвращает: Storage, error

Загрузка объектов

Загрузка содержимого из строки или файла:

local storage = cloudstorage.get("app.infra:files")

-- Загрузка строкового содержимого
local ok, err = storage:upload_object("reports/daily.json", json.encode({
    date = "2024-01-15",
    total = 1234
}))

-- Загрузка из файла
local fs = require("fs")
local vol = fs.get("app:data")
local file = vol:open("/large-file.bin", "r")

storage:upload_object("backups/large-file.bin", file)
file:close()

storage:release()

Возвращает: boolean, error

Опции загрузки

Прикрепите метаданные или защитите запись с помощью таблицы опций:

storage:upload_object("reports/daily.json", body, {
    content_type = "application/json",
    cache_control = "max-age=3600",
    metadata = { owner = "team-a", run_id = "1234" },  -- stored as x-amz-meta-*
    only_if_absent = true                              -- fail if the key already exists
})

Условная запись, не прошедшая своё предусловие, возвращает ошибку precondition_failed.

Скачивание объектов

Скачать объект в файловый writer:

local storage = cloudstorage.get("app.infra:files")
local fs = require("fs")
local vol = fs.get("app:temp")

local file = vol:open("/downloaded.json", "w")
local ok, err = storage:download_object("reports/daily.json", file)
file:close()

-- Скачивание части (первый 1KB)
local partial = vol:open("/partial.bin", "w")
storage:download_object("backups/large-file.bin", partial, {
    range = "bytes=0-1023"
})
partial:close()

storage:release()

Возвращает: boolean, error

Непройденное предусловие (if_match/if_none_match) возвращает ошибку precondition_failed.

Перечисление объектов

Список объектов с опциональной фильтрацией по префиксу:

local storage = cloudstorage.get("app.infra:files")

local result, err = storage:list_objects({
    prefix = "reports/2024/",
    max_keys = 100
})

for _, obj in ipairs(result.objects) do
    print(obj.key, obj.size, obj.etag)
end

-- Пагинация для больших результатов
local token = nil
repeat
    local result = storage:list_objects({
        prefix = "logs/",
        max_keys = 1000,
        continuation_token = token
    })
    for _, obj in ipairs(result.objects) do
        process(obj)
    end
    token = result.next_continuation_token
until not result.is_truncated

storage:release()

Возвращает: table, error

Результат содержит objects, is_truncated, next_continuation_token. Каждый объект имеет key, size, etag, storage_class, а также опциональные last_modified, version_id и owner.

Метаданные объекта

Получить метаданные одного объекта без скачивания его тела:

local storage = cloudstorage.get("app.infra:files")

local meta, err = storage:head_object("reports/daily.json")
if err then
    return nil, err
end

print(meta.size, meta.etag, meta.content_type)
for k, v in pairs(meta.metadata) do
    print("meta", k, v)
end

storage:release()

Возвращает: table, error

Поля результата:

Отсутствующий объект возвращает ошибку not_found.

Удаление объектов

Удалить несколько объектов:

local storage = cloudstorage.get("app.infra:files")

storage:delete_objects({
    "temp/file1.txt",
    "temp/file2.txt",
    "temp/file3.txt"
})

storage:release()

Возвращает: boolean, error

Попытка выполняется для каждого ключа. Удаление несуществующего ключа ошибкой не является. Если провайдер сообщает о сбоях по отдельным ключам, вызов возвращает одну ошибку, перечисляющую каждый сбойный ключ и код ошибки провайдера.

URL для скачивания

Создать временный URL для скачивания объекта без учётных данных. Полезно для передачи файлов внешним пользователям или отдачи контента через приложение.

local storage, err = cloudstorage.get("app.infra:files")
if err then
    return nil, err
end

local url, err = storage:presigned_get_url("reports/quarterly.pdf", {
    expiration = 3600
})

storage:release()

if err then
    return nil, err
end

-- Вернуть URL клиенту для прямого скачивания
return {download_url = url}

Возвращает: string, error

URL для загрузки

Создать временный URL для загрузки объекта без учётных данных. Позволяет клиентам загружать файлы напрямую в хранилище без проксирования через сервер.

local storage, err = cloudstorage.get("app.infra:files")
if err then
    return nil, err
end

local url, err = storage:presigned_put_url("uploads/user-123/avatar.jpg", {
    expiration = 600,
    content_type = "image/jpeg",
    content_length = 1024 * 1024
})

storage:release()

if err then
    return nil, err
end

-- Вернуть URL клиенту для прямой загрузки
return {upload_url = url}

Возвращает: string, error

Multipart-загрузки

Одиночный presigned PUT ограничивает объект 5 ГиБ. Presigned multipart-загрузка разбивает более крупный объект на части, которые клиент загружает напрямую, а затем собирает их на стороне сервера. Multipart — это возможность провайдера: S3 её реализует, а провайдеры без неё возвращают errors.UNAVAILABLE.

local storage = cloudstorage.get("app.infra:files")

local mp, err = storage:create_multipart_upload("backups/huge.zip", {
    content_type = "application/zip",
    metadata = { source = "uploader" },
})
if err then return nil, err end

local urls, err = storage:presigned_part_urls("backups/huge.zip", mp.upload_id, {
    count = 3,
    expiration = 900,
})
if err then
    storage:abort_multipart_upload("backups/huge.zip", mp.upload_id)
    return nil, err
end

-- Клиент выполняет PUT по каждому url и возвращает ETag из заголовков ответа.
local done, err = storage:complete_multipart_upload("backups/huge.zip", mp.upload_id, {
    { part_number = 1, etag = etag1 },
    { part_number = 2, etag = etag2 },
    { part_number = 3, etag = etag3 },
})

storage:release()

create_multipart_upload

Начинает multipart-загрузку для ключа.

Возвращает: table, error — таблица несёт upload_id, идентифицирующий загрузку для всех последующих вызовов частей, завершения и отмены.

Условные записи (if_match, if_none_match, only_if_absent) не входят в протокол multipart и здесь не принимаются.

presigned_part_urls

Генерирует presigned PUT URL для частей выполняющейся загрузки. На каждый URL выполняется обычный HTTP PUT; загружающая сторона должна сохранить заголовок ответа ETag каждой части для complete_multipart_upload.

Ровно один из parts или count обязателен, и один вызов подписывает не более 1000 URL — для очень крупных объектов подписывайте постранично.

Возвращает: table, error — массив { part_number, url }.

Каждая часть, кроме последней, должна быть не меньше 5 МиБ; провайдер проверяет это при завершении.

complete_multipart_upload

Собирает итоговый объект из загруженных частей. Части можно сообщать в любом порядке — перед завершением они сортируются по номеру.

Возвращает: table, error — etag, а также version_id и location, если провайдер их сообщает. Неизвестный upload ID возвращает errors.NOT_FOUND.

abort_multipart_upload

Отбрасывает выполняющуюся загрузку и освобождает её сохранённые части.

Возвращает: boolean, error

Загрузка, которая так и не была завершена, хранит свои части — и тарифицируется — пока не будет отменена. Отменяйте её на каждом пути сбоя и настройте правило жизненного цикла бакета как подстраховку — см. Cloud Storage.

Чтение по диапазонам

open_reader открывает произвольный доступ к объекту через ranged GET — без локального промежуточного хранения и без полной загрузки. Основной потребитель — archive.open, который читает многогигабайтные архивы прямо из объектного хранилища с ограниченным потреблением памяти.

local archive = require("archive")
local storage = cloudstorage.get("app.infra:files")

local reader, err = storage:open_reader("uploads/huge.zip", {
    block_size = 8 * 1024 * 1024,
    cache_blocks = 4,
})
if err then return nil, err end

local r = assert(archive.open(reader))
for e in r:entries() do
    print(e.name, e.size)
end
r:close()
reader:close()

storage:release()

block_size * cache_blocks не может превышать 256 МиБ. Отсутствующий объект возвращает errors.NOT_FOUND.

Возвращает: Reader, error

ETag объекта закрепляется при открытии reader'а и отправляется как If-Match при каждом чтении диапазона, поэтому объект, перезаписанный посреди чтения, приводит к отказу чтения с ошибкой предусловия от провайдера вместо выдачи смеси двух поколений объекта; archive отдаёт её как errors.INTERNAL. Провайдер, не способный предоставить ETag, возвращает errors.UNAVAILABLE; reader никогда не отдаёт незакреплённый объект.

Промахи кэша выполняют блокирующий сетевой ввод-вывод в вызывающей задаче и сериализуют параллельных читателей, поэтому последовательный доступ по записям — как в архиве — и есть предполагаемая форма использования.

Методы Reader

Reader закрывается автоматически в области задачи, если не закрыт явно.

Методы Storage

Разрешения

Операции облачного хранилища подчиняются вычислению политики безопасности.

Ошибки

См. Обработка ошибок для работы с ошибками.