問題

AWS S3 によって Presigned URL を発行すれば、サーバーに直接ファイルをアップロードせずとも、S3 に対して直接ファイルをアップロードできる。
しかし、1回のリクエストでアップロードできる最大ファイルサイズは 5GB であり、それ以上のファイルをアップロードするためにはマルチパートアップロードによって分割アップロードを行う必要がある。
ここでは、ブラウザ側に Credential を発行することなく、サーバー側で発行された Presigned URL を利用してファイルをマルチパートアップロードする方法についてを記す。

プロジェクト全体

プロジェクト全体で利用するソースコードを以下で公開している。
今回は、サーバーサイドは AWS Chalice (Python) で、クライアントサイドでは JavaScript を Webpack でビルドしたものを利用している。

ローカルで自分で利用する場合は README.md を参考に環境を構築すること。

S3 バケットの設定

上記プロジェクトではアップロード対象となる S3 バケットを aws-cdk v2 にて構築している。

aws-cdk/pycdk/__init__.pyから抜粋

# Define S3 Bucket and settings to allow access from CloudFront
prefix = conf.get('StageName', 'local')
allowed_origins = conf['S3'].get('AllowOrigins', [])
bucket_cors = cdk_s3.CorsRule(
    allowed_headers=['*'],
    allowed_methods=[
        cdk_s3.HttpMethods.PUT
    ],
    allowed_origins=allowed_origins,
    exposed_headers=['ETag']
)
bucket_name = '.'.join([
    prefix, conf['S3'].get('BucketName')
])

lifecycle_rules = [
    # マルチパートアップロード失敗を時間経過で自動消去
    cdk_s3.LifecycleRule(
        abort_incomplete_multipart_upload_after=cdk.Duration.days(3),
        prefix='uploads/'
    ),
]
self.bucket = cdk_s3.Bucket(
    self, 'UploadsBucket',
    bucket_name=bucket_name,
    encryption=cdk_s3.BucketEncryption.S3_MANAGED,
    block_public_access=cdk_s3.BlockPublicAccess.BLOCK_ALL,
    cors=[bucket_cors],
    lifecycle_rules=lifecycle_rules
)
cdk.CfnOutput(self, 'UploadBucketName',
              value=self.bucket.bucket_name)

構築コードで重要な点は以下の通り。これらを忘れるとブラウザからのアップロードに失敗したり、アップロードの完了ができなかったりする。

バケットに対してCORSルールを設定し、特定ドメインからの PUT リクエストを許可するような設定を行う (allowed_origins)
- この設定がないオリジンから PUT を行った場合、403 エラーとなる
presigned url を発行した後、クライアントサイドから S3 に対して PUT をする場合、その結果として ETag が発行される。アップロード完了通知を行うときにこの ETag の値が必要となるため、この値をクライアントサイドから参照可能な状態にしている (exposed_headers)
- マルチパートアップロード完了の API 実行には ETag の値が必要である。一応、uploadIdが分かれば list_parts によってサーバーサイドから取得可能であるようなので、ETag を expose できない場合はこちらを利用することになりそう
- 元々の S3 のデータ保持戦略は結果整合性だったので、レスポンスによる ETag を用いる方が良いと考えていたが、2020年のアップデートで強い一貫性をサポートするようになっているため、こちらを使う方法で実装しても良さそう

また、特定 prefix が付与されている部分に対して、マルチパートアップロードが不完全となった場合、それらのファイルを自動的に消去するためのライフサイクルポリシーを設定している。これを設定しないと、マルチアップロードに失敗したファイル（途中までアップロードされたファイル）はそのまま維持され続ける (=課金され続ける) ため注意が必要。

アップロードの流れ

Presigned URL を利用したマルチパートアップロードの流れは以下のように行われる。 S3に対するアクセス権限はサーバーサイドのみが保有しており、権限のあるサーバーが発行した Presigned URL をクライアントサイドで利用することで、クライアント側から直接アップロードを行う。

C1: input type="file" を用意し、ファイルが選択された段階でサーバーサイドにアップロードに利用する Presigned URL の一覧を要求する
- この時、アップロードを行うファイルのサイズはクライアント側で取得できるので、マルチアップロードのために何分割するかを合わせて渡している
- このコードでは 100MB 単位で分割している。そのため、例えば 450MB のファイルをアップロードする場合は5つのアップロード用URLを要求する
S1: 要求を受け取った後、マルチパートアップロードのためのURL一覧を生成する
C2: それぞれの URL に対して、ファイルを分割してアップロードする
C3: 全てのアップロードが完了した後、アップロード完了時に発行される ETag 情報を合わせて、サーバーサイド側にアップロード完了通知を行う
S2: サーバー側で ETag 情報を突き合わせて、正常なアップロードが行われたかを S3 に確認する

クライアントサイドの実装

js/upload.jsから抜粋

  // C1: ファイルからアップロードURL群を要求
  const totalPartNumber = Math.ceil(filedata.size / FILE_CHUNK_SIZE);
  let url = "/api/multipart_upload/start";
  const ext = getExt(filedata.name);
  if (ext) {
    url = url + "/" + ext;
  }
  // S1: /api/multipart_upload/start に対応するサーバーサイドで
  // マルチパートアップロード用の情報を生成
  const { data } = await axios.put(url, { TotalPart: totalPartNumber });

  const key = data.UploadKey;
  const uploadId = data.UploadId;
  const totalPartCount = data.Parts.length;

  let completed = 0;
  const updateProgress = () => {
    // 進捗バー更新 (略)
  };

  // ファイルを基準バイト単位で分割して
  // それぞれに対して S1 で生成された Presigned URL に対してアップロード
  const partUploads = data.Parts.map((part, index) => {
    return () => {
      return readPartData(index, filedata).then(async (partdata) => {
        const content = partdata.data;
        const uploadUrl = part.url;
        // アップロード開始
        const res = await axios.put(uploadUrl, content, {
          headers: {
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Credentials": "true",
            "Access-Control-Expose-Headers": "etag",
            "Content-Type": "", // no content-type required
          },
        });

        // アップロード完了
        // 完了通知に必要な情報を作成して返す
        return {
          ETag: res.headers.etag.replaceAll('"', ""),
          PartNumber: data.Parts[index].part,
        };
      });
    };
  });

  // C2: 順次実行して解決
  // await をかけずに大量の平行アップロードすることも可能だが、
  // 本数が多い場合ブラウザで CPU ビジーとなるため、ここでは効率が悪いが１本ずつアップロードしている
  let uploadedParts = [];
  for (const part of partUploads) {
    const uploaded = await part();
    completed++;
    updateProgress();
    uploadedParts = uploadedParts.concat([uploaded]);
  }

  // C3. 全て完了まで待機して完了処理をサーバーに通知
  const completeUrl = "/api/multipart_upload/complete";
  // S2. サーバーサイドでアップロード完了処理を行う
  await axios.put(completeUrl, {
    Key: key,
    Parts: uploadedParts,
    UploadId: uploadId,
  });
  
  // 完了処理
  bar.textContent = "アップロードが完了しました。";

サーバーサイドの実装

マルチアップロードの開始・完了処理実装は以下の通り。
開始時に複数のアップロード用 Presigned URL を生成し、完了時は発行した各 Part に対応する正しい ETag が発行されているかを S3 に問い合わせて、問題なければアップロードが完了するようになっている。
アップロードの失敗時は {"Success": false} が返ってくるので、アップロードの失敗時のハンドリングも可能となっている。

apis/app.pyから抜粋

@app.route('/api/multipart_upload/start/{ext}',
           methods=['PUT'])
def start_multipart_upload(ext: Optional[str] = None):
    """ S1: マルチアップロードの開始処理 """
    params = app.current_request.json_body or {}
    total_part = params.get('TotalPart', 0)
    if total_part < 0:
        return _json({
            'message': 'TotalPart: int required',
        }, status=400)
    if ext:
        filename = '.'.join([str(uuid.uuid4()), ext])
    else:
        filename = str(uuid.uuid4())

    res = s3.start_multipart_upload(filename, total_part, suffix=ext)
    return _json(res)


@app.route('/api/multipart_upload/complete',
           methods=['PUT'])
def complete_multipart_upload():
    """ S2: マルチアップロードの完了通知 """
    params = app.current_request.json_body or {}
    res = s3.complete_multipart_upload(
        params['Key'], params['UploadId'], params['Parts'])
    return _json({
        'Success': res,
    })

apis/chalicelib/s3.pyから抜粋

    def start_multipart_upload(
        self, filename: str, total_part: int,
        suffix: str = None, expires_seconds: int = 7200,
        temporary_prefix: str = None
    ):
        """ マルチパートアップロードを開始します """
        prefix = temporary_prefix or self.upload_prefix
        key = f'{prefix}{filename}'
        content_type = self._content_type(suffix)
        res = self.client.create_multipart_upload(
            Bucket=self.bucketname, Key=key
        )
        parts = [
            {
                'part': idx+1,
                'url': self._create_multipart_upload_url(
                    res['Key'], res['UploadId'], idx+1
                ),
            } for idx in range(total_part)
        ]

        return {
            'UploadId': res['UploadId'],
            'UploadKey': key,
            'ContentType': content_type,
            'Parts': parts,
        }

    def complete_multipart_upload(self, key: str, upload_id: str,
                                  parts: list):
        """ マルチパートアップロードを完了させます """
        try:
            self.client.complete_multipart_upload(
                Bucket=self.bucketname,
                Key=key,
                MultipartUpload={'Parts': parts},
                UploadId=upload_id)
            return True
        except Exception as err:
            logging.exception(err)
            return False

動作例

ローカルの開発サーバーを起動、あるいは AWS にデプロイしてサイトのトップページにアクセスすると以下のようなページが表示される。

手元にあった大容量のファイルのアップロード検証として、ファイルを選択する部分で Ubuntu の ISO イメージを選択した。その後、100MB ずつ順にアップロードが進んでいき、全てのファイルのアップロードが終了すると、マルチパートアップロードの完了処理を呼び出し、S3上にファイルがアップロードされる。

現時点実装に対する課題

現時点の実装には以下の課題があるため、利用するプロダクト次第ではちゃんとした実装が必要となる。

各URLに対して await 無しで並列アップロードを行った場合、URL数が増えるとブラウザがビジーとなってしまった。そのため、ここでは順次処理でアップロードするようにした。本来は並列処理で高速にできる部分が順次処理として低速になっているため、効率を求める場合はそのためのコードの修正が必要
アップロードに失敗した時、途中から再開するということも理論上は可能だが、今回記事で提示したコードにはそのための処理が含まれていないため、ユーザビリティを求める場合はそのための仕組みを別途作成する必要がある
Presigned URL を生成するタイミングで Presigned URL の有効期限を設定する必要がある。今回のコードでは URL 生成から 7200秒 (=2時間) 固定となっているため、生成から2時間以内に全URLに対するアップロードを開始すれば正常に終了するが、利用者の環境、あるいは、想定するファイルサイズ次第では7200秒以内にアップロードに着手できない可能性があるため、十分な時間を設定する必要がある
マルチパートアップロードの完了については、この complete_multipart_upload を Presignd URL にすることでサーバーサイドへの問い合わせなしに解決可能かもしれないが、これは未検証

参考

Appendix: 開発時に色々と失敗した余談

Chalice でのデプロイ時には requirements.txt を元にライブラリを取得して zip ファイルを作成し、これをデプロイする。また、AWS Lambda の Python 環境ではデフォルトで boto3 が利用可能なのだが、かつて requirements.txt の中に boto3 の古いバージョンを固定で指定していたことがあった。同様の状況をローカルの pip で再現しようとしても失敗したが、何故か Chalice でデプロイした環境では古い boto3 のバージョンで動いてしまった。 その結果、S3 Presigned URL の不正なドメインが返ってくるという事象が起こっていたため、この問題の特定に非常に苦労した
S3 に put した後のレスポンスについて Chrome の Developer Tool で確認していた。結果、Developer Tools ではレスポンスに ETag が含まれていることを確認したにも関わらず、JavaScript では ETag を取得できないという事象に遭遇した。 これは、S3 の ETag の expose 設定を行っていなかったためだが、かなり混乱した