Large objects and multipart uploads

Applies to VAST Cluster 5.5.

Large objects are uploaded in parts and read in ranges. This guide shows the managed transfers in boto3, which do both for you, then the multipart calls underneath: create, upload parts, list, complete and abort. It also explains the ETag a multipart object gets, and what to clean up when an upload fails. Everything happens under the key prefix docs-large-objects-and-multipart/ and in a temporary local directory, and the last block removes both.

Each code block is complete on its own. Below each block is the output you should see.

Before you start

You need the bucket, key pair and environment variables from the quickstart, about 50 MB of free space in the bucket and in your local temporary directory, and permission to list and abort multipart uploads (the bucket owner has it).

Create a 40 MiB local file to upload:

import os
import tempfile

workdir = os.path.join(tempfile.gettempdir(), "docs-large-objects-and-multipart")
os.makedirs(workdir, exist_ok=True)
path = os.path.join(workdir, "big.bin")

with open(path, "wb") as f:
    chunk = b"0123456789abcdef" * 65536  # 1 MiB
    for _ in range(40):
        f.write(chunk)
print(os.path.getsize(path), "bytes")

41943040 bytes

Upload and download with managed transfers

upload_file and download_file split a large transfer into parts and send them in parallel threads. TransferConfig sets when to switch to multipart (multipart_threshold), the part size (multipart_chunksize) and the number of parallel requests (max_concurrency). The download uses parallel ranged GETs in the same way:

import hashlib
import os
import tempfile

import boto3
from boto3.s3.transfer import TransferConfig
from botocore.config import Config

s3 = boto3.client(
    "s3",
    endpoint_url=os.environ["AWS_S3_ENDPOINT_URL"],
    region_name="us-east-1",
    config=Config(signature_version="s3v4", s3={"addressing_style": "path"}, max_pool_connections=8),
)
bucket = os.environ["DATASTORE_BUCKET"]
workdir = os.path.join(tempfile.gettempdir(), "docs-large-objects-and-multipart")
key = "docs-large-objects-and-multipart/big.bin"

MiB = 1024 * 1024
transfer = TransferConfig(multipart_threshold=8 * MiB, multipart_chunksize=8 * MiB, max_concurrency=4)

s3.upload_file(os.path.join(workdir, "big.bin"), bucket, key, Config=transfer)
etag = s3.head_object(Bucket=bucket, Key=key)["ETag"].strip('"')
print("uploaded in", etag.rsplit("-", 1)[1], "parts")

s3.download_file(bucket, key, os.path.join(workdir, "big.copy"), Config=transfer)

def md5(p):
    h = hashlib.md5()
    with open(p, "rb") as f:
        for block in iter(lambda: f.read(MiB), b""):
            h.update(block)
    return h.hexdigest()

print("download matches:", md5(os.path.join(workdir, "big.bin")) == md5(os.path.join(workdir, "big.copy")))

uploaded in 5 parts
download matches: True

Give the client at least as many connections (max_pool_connections) as max_concurrency, or threads wait for a free connection. For how part size and concurrency affect speed, see performance.

The multipart calls

The three multipart calls, and what happens to an upload that's never completed.

Underneath, a multipart upload is three steps:

create_multipart_upload returns an upload ID. Metadata, content type and tags (Tagging=) are given here, not on the parts.
upload_part sends each part with a part number. The cluster checks each part against its Content-MD5 header if you send one, and returns the part's ETag.
complete_multipart_upload sends the list of part numbers and ETags. The cluster checks the list and assembles the object.

Until step 3 succeeds, the object doesn't exist. Upload parts in any order and in parallel; only the list in step 3 has to be in order:

import hashlib
import os

import boto3
from botocore.config import Config

s3 = boto3.client(
    "s3",
    endpoint_url=os.environ["AWS_S3_ENDPOINT_URL"],
    region_name="us-east-1",
    config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
)
bucket = os.environ["DATASTORE_BUCKET"]
key = "docs-large-objects-and-multipart/assembled.bin"

MiB = 1024 * 1024
parts_data = [b"a" * (5 * MiB), b"b" * (5 * MiB), b"c" * MiB]

upload = s3.create_multipart_upload(Bucket=bucket, Key=key, ContentType="application/octet-stream")
upload_id = upload["UploadId"]
parts = []
for number, data in enumerate(parts_data, start=1):
    resp = s3.upload_part(Bucket=bucket, Key=key, UploadId=upload_id, PartNumber=number, Body=data)
    parts.append({"PartNumber": number, "ETag": resp["ETag"]})

listed = s3.list_parts(Bucket=bucket, Key=key, UploadId=upload_id)["Parts"]
print("parts on the cluster:", [(p["PartNumber"], p["Size"]) for p in listed])

done = s3.complete_multipart_upload(Bucket=bucket, Key=key, UploadId=upload_id, MultipartUpload={"Parts": parts})
etag = done["ETag"].strip('"')
print("ETag ends with the part count:", etag.endswith("-3"))

digests = [hashlib.md5(d).digest() for d in parts_data]
from_binary = hashlib.md5(b"".join(digests)).hexdigest() + "-3"
from_hex = hashlib.md5("".join(d.hex() for d in digests).encode()).hexdigest() + "-3"
print("ETag = MD5 of the parts' binary MD5s:", etag == from_binary)
print("ETag = MD5 of the parts' hex ETags:", etag == from_hex)
print("size:", s3.head_object(Bucket=bucket, Key=key)["ContentLength"])

parts on the cluster: [(1, 5242880), (2, 5242880), (3, 1048576)]
ETag ends with the part count: True
ETag = MD5 of the parts' binary MD5s: True
ETag = MD5 of the parts' hex ETags: False
size: 11534336

So a multipart object's ETag isn't the MD5 of its data. As on AWS, it's the MD5 of the parts' binary MD5 digests joined together (not of their hex ETag strings), followed by - and the number of parts. To check a multipart object's integrity, compare it with an ETag computed the same way with the same part size, or keep your own checksum in the object's metadata.

Keep to part sizes your S3 tools already use: at least 5 MiB for every part except the last, as AWS requires. No minimum or maximum part size is documented for the cluster in 5.5, so don't rely on smaller parts working.

Two more things to know:

Create once with multipart too. IfNoneMatch="*" on complete_multipart_upload completes only if no object with that key exists yet (not in versioned buckets).
Copy large objects in ranges. upload_part_copy takes a part from an existing object (CopySourceRange="bytes=...") without the data passing through your client. boto3's managed copy uses it for large objects.

Abandoned uploads

Parts of an upload that was neither completed nor aborted stay on the cluster and use capacity, and they don't show up in an ordinary listing. Find them with list_multipart_uploads, and abort them:

import os

import boto3
from botocore.config import Config

s3 = boto3.client(
    "s3",
    endpoint_url=os.environ["AWS_S3_ENDPOINT_URL"],
    region_name="us-east-1",
    config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
)
bucket = os.environ["DATASTORE_BUCKET"]
prefix = "docs-large-objects-and-multipart/"

# An upload that a crashed job left behind.
left = s3.create_multipart_upload(Bucket=bucket, Key=prefix + "crashed.bin")
s3.upload_part(Bucket=bucket, Key=prefix + "crashed.bin", UploadId=left["UploadId"], PartNumber=1, Body=b"x" * 1024)

def in_progress():
    return s3.list_multipart_uploads(Bucket=bucket, Prefix=prefix).get("Uploads", [])

print("in progress:", [u["Key"] for u in in_progress()])
for u in in_progress():
    s3.abort_multipart_upload(Bucket=bucket, Key=u["Key"], UploadId=u["UploadId"])
print("after abort:", [u["Key"] for u in in_progress()])

in progress: ['docs-large-objects-and-multipart/crashed.bin']
after abort: []

Aborting frees the parts. A part that was still being uploaded when you aborted may or may not complete, so list again after a busy upload is aborted. Your platform team can also set a lifecycle rule that aborts incomplete multipart uploads after a number of days; on VAST, lifecycle rules are an admin setting (Overview of Lifecycle Rules).

If you write your own S3 client: send CompleteMultipartUpload to the usual path (/<bucket>/<key>?uploadId=...). An absolute URL in the request line isn't supported in 5.5.

Clean up

import os
import shutil
import tempfile

import boto3
from botocore.config import Config

s3 = boto3.client(
    "s3",
    endpoint_url=os.environ["AWS_S3_ENDPOINT_URL"],
    region_name="us-east-1",
    config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
)
bucket = os.environ["DATASTORE_BUCKET"]
prefix = "docs-large-objects-and-multipart/"

for u in s3.list_multipart_uploads(Bucket=bucket, Prefix=prefix).get("Uploads", []):
    s3.abort_multipart_upload(Bucket=bucket, Key=u["Key"], UploadId=u["UploadId"])
keys = [{"Key": o["Key"]} for o in s3.list_objects_v2(Bucket=bucket, Prefix=prefix).get("Contents", [])]
if keys:
    s3.delete_objects(Bucket=bucket, Delete={"Objects": keys})
shutil.rmtree(os.path.join(tempfile.gettempdir(), "docs-large-objects-and-multipart"), ignore_errors=True)
print("deleted", len(keys), "objects")

deleted 2 objects