Cloudian 객체 스토리지/멀티스레드로 파일 업로드

이번은 Python (boto3)의 샘플 편입니다.

멀티스레드로 파일 업로드

지정된 디렉토리 (이 샘플에서는 "./log")에있는 모든 파일을 Cloudian의 버킷 "logs"의 실행 날짜 폴더 (YYYY-MM-DD)에 업로드하는 Python 프로그램입니다.

업로드할 각 파일에 대해 스레드를 만들어 멀티스레드로 Cloudian에 파일을 업로드합니다. 업로드시 파일에 대해 ACL에 "private"를 설정하고 메타데이터로 업로드한 날짜를 부여하며 AES256의 서버측 암호화를 사용하도록 설정했습니다.

MultipartUpload.py

import os
import glob
import threading
import sys
from datetime import datetime

import boto3, botocore
from boto3.s3.transfer import S3Transfer
from boto3.s3.transfer import TransferConfig


######### マルチスレッド関数 ##########
def multithreads_upload(upfile):

    print("Thread %s is uploading file: %s" % (threading.current_thread(), upfile))

    transfer.upload_file(
        upfile,
       bucket,
       today + '/' + os.path.basename(upfile),

       extra_args={
            'ACL': 'private',
            'Metadata': {'Stored': today},
            'ServerSideEncryption': 'AES256'
       }
    )

    print("Thread %s done uploading file: %s" % (threading.current_thread(), upfile))



########## メイン処理 ##########
# 変数セット
dir = './fileup/'

bucket = 'logs'
region = 'region1'
today = datetime.now().strftime('%Y-%m-%d')


# boto3オブジェクト作成
client = boto3.client(
    's3',
    endpoint_url='http://s3-region1.admin-tech.tokyo',

    # 認証情報の直接記述は推奨されない（テスト目的）
    aws_access_key_id='アクセスキーを記述',
    aws_secret_access_key='シークレットキーを記述'
)

# 転送オブジェクト作成
transfer = S3Transfer(client)



try:
    # バケットの存在確認
    client.head_bucket(Bucket = bucket)

except botocore.exceptions.ClientError:
    # バケットが存在しなければ、新規作成
    client.create_bucket(
        ACL='private',
        Bucket=bucket,
        CreateBucketConfiguration={
            'LocationConstraint': 'region1'
        }
)


# マルチスレッドによるファイルのアップロード
upfiles = [dir + file for file in os.listdir(dir) if os.path.isfile(dir + file)]
threads = []

for upfile in upfiles:
    t = threading.Thread(target=multithreads_upload, args=(upfile,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()

프로그램 실행의 출력 로그로부터, 업로드하는 파일 사이즈에 의해 각 thread의 개시와 종료에 차이가 있는 것을 알 수 있습니다.
이 샘플에서는 각 thread의 종료를 threading.join()를 사용해, 모든 파일의 업로드가 종료하는 것을 대기합니다.

아래 그림은 이 샘플 프로그램에 의해 업로드된 파일을 CMC에서 확인한 화면입니다.