# PROJECT LONGBOW - LIB FOR TRANSMISSION BETWEEN AMAZON S3 import datetime import logging import hashlib import concurrent.futures import threading import base64 import urllib.request import urllib.parse import boto3 from botocore.config import Config from botocore.exceptions import ClientError # from boto3.dynamodb import conditions import json import os import sys import time from fnmatch import fnmatchcase from pathlib import PurePosixPath, Path logger = logging.getLogger() Max_md5_retry = 2 # Configure logging def set_log(LoggingLevel, this_file_name): logger.setLevel(logging.WARNING) if LoggingLevel == 'INFO': logger.setLevel(logging.INFO) elif LoggingLevel == 'DEBUG': logger.setLevel(logging.DEBUG) # File logging log_path = Path(__file__).parent.parent / 'amazon-s3-migration-log' if not Path.exists(log_path): Path.mkdir(log_path) start_time = datetime.datetime.now().isoformat().replace(':', '-')[:19] _log_file_name = str(log_path / f'{this_file_name}-{start_time}.log') print('Log file:', _log_file_name) fileHandler = logging.FileHandler(filename=_log_file_name) fileHandler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s - %(message)s')) logger.addHandler(fileHandler) return logger, _log_file_name # Set environment def set_env(*, JobType, LocalProfileMode, table_queue_name, sqs_queue_name, ssm_parameter_credentials, MaxRetry): s3_config = Config(max_pool_connections=200, retries={'max_attempts': MaxRetry}) # boto default 10 if os.uname()[0] == 'Linux' and not LocalProfileMode: # on EC2, use EC2 role logger.info('Get instance-id and running region') instance_id = urllib.request.urlopen(urllib.request.Request( "http://169.254.169.254/latest/meta-data/instance-id" )).read().decode('utf-8') region = json.loads(urllib.request.urlopen(urllib.request.Request( "http://169.254.169.254/latest/dynamic/instance-identity/document" )).read().decode('utf-8'))['region'] sqs = boto3.client('sqs', region) dynamodb = boto3.resource('dynamodb', region) ssm = boto3.client('ssm', region) # 取另一个Account的credentials logger.info(f'Get ssm_parameter_credentials: {ssm_parameter_credentials}') try: credentials = json.loads(ssm.get_parameter( Name=ssm_parameter_credentials, WithDecryption=True )['Parameter']['Value']) except Exception as e: logger.error(f'Fail to get {ssm_parameter_credentials} in SSM Parameter store. ' f'Fix and restart Jobsender. {str(e)}') sys.exit(0) credentials_session = boto3.session.Session( aws_access_key_id=credentials["aws_access_key_id"], aws_secret_access_key=credentials["aws_secret_access_key"], region_name=credentials["region"] ) if JobType.upper() == "PUT": s3_src_client = boto3.client('s3', region, config=s3_config) s3_des_client = credentials_session.client('s3', config=s3_config) elif JobType.upper() == "GET": s3_des_client = boto3.client('s3', region, config=s3_config) s3_src_client = credentials_session.client('s3', config=s3_config) else: logger.error('Wrong JobType setting in config.ini file') sys.exit(0) # 在没有Role的环境运行，例如本地Mac测试 else: instance_id = "local" src_session = boto3.session.Session(profile_name='iad') des_session = boto3.session.Session(profile_name='zhy') sqs = src_session.client('sqs') dynamodb = src_session.resource('dynamodb') ssm = src_session.client('ssm') s3_src_client = src_session.client('s3', config=s3_config) s3_des_client = des_session.client('s3', config=s3_config) # 在当前屏幕也输出，便于local debug监控。 streamHandler = logging.StreamHandler() streamHandler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s - %(message)s')) logger.addHandler(streamHandler) table = dynamodb.Table(table_queue_name) table.wait_until_exists() sqs_queue = wait_sqs_available( sqs=sqs, sqs_queue_name=sqs_queue_name ) return sqs, sqs_queue, table, s3_src_client, s3_des_client, instance_id, ssm def wait_sqs_available(*, sqs, sqs_queue_name): while True: try: return sqs.get_queue_url(QueueName=sqs_queue_name)['QueueUrl'] except Exception as e: logger.warning(f'Waiting for SQS availability. {str(e)}') time.sleep(10) def check_sqs_empty(sqs, sqs_queue): try: sqs_in_flight = sqs.get_queue_attributes( QueueUrl=sqs_queue, AttributeNames=['ApproximateNumberOfMessagesNotVisible', 'ApproximateNumberOfMessages'] ) except Exception as e: logger.error(f'Fail to get_queue_attributes: {str(e)}') return False # Can't get sqs status, then consider it is not empty NotVisible = sqs_in_flight['Attributes']['ApproximateNumberOfMessagesNotVisible'] Visible = sqs_in_flight['Attributes']['ApproximateNumberOfMessages'] logger.info(f'ApproximateNumberOfMessagesNotVisible: {NotVisible}, ApproximateNumberOfMessages: {Visible}') if NotVisible == '0' and (Visible == '0' or Visible == '1'): # In init state, the new created bucket trigger SQS will send one test message to SQS. # So here to ignore the case Visible == '1' return True # sqs is empty return False # sqs is not empty # Get source S3 bucket file list with versionId def get_src_file_list(*, s3_client, bucket, S3Prefix, JobsenderCompareVersionId): # get s3 file list file_list = [] if JobsenderCompareVersionId: paginator = s3_client.get_paginator('list_object_versions') else: paginator = s3_client.get_paginator('list_objects_v2') # 速度比 list_object_versions 快很多 try: if S3Prefix == '/': S3Prefix = '' logger.info(f'Get s3 file list from: {bucket}/{S3Prefix}') response_iterator = paginator.paginate( Bucket=bucket, Prefix=S3Prefix ) for page in response_iterator: if "Versions" in page: # JobsenderCompareVersionId==True logger.info(f'Got list_object_versions {bucket}/{S3Prefix}: {len(page["Versions"])}') for n in page["Versions"]: # 只拿最新版本的列表 if n["IsLatest"]: file_list.append({ "Key": n["Key"], "Size": n["Size"], "versionId": n["VersionId"] if JobsenderCompareVersionId else 'null' }) elif "Contents" in page: # JobsenderCompareVersionId==False logger.info(f'Got list_objects_v2 {bucket}/{S3Prefix}: {len(page["Contents"])}') for n in page["Contents"]: file_list.append({ "Key": n["Key"], "Size": n["Size"], "versionId": 'null' }) except Exception as err: logger.error(f'Fail to get s3 list versions {bucket}/{S3Prefix}: {str(err)}') logger.info(f'Bucket list length(get_src_file_list)：{str(len(file_list))}') return file_list # Get Destination S3 bucket file list with versionId from DDB def get_des_file_list(*, s3_client, bucket, S3Prefix, table, JobsenderCompareVersionId): if JobsenderCompareVersionId: ver_list = get_versionid_from_ddb( des_bucket=bucket, table=table ) else: ver_list = {} # For delete prefix in des_prefix if S3Prefix == '' or S3Prefix == '/': # 目的bucket没有设置 Prefix dp_len = 0 else: # 目的bucket的 "prefix/"长度 dp_len = len(S3Prefix) + 1 # get s3 file list file_list = [] paginator = s3_client.get_paginator('list_objects_v2') try: if S3Prefix == '/': S3Prefix = '' logger.info(f'Get s3 file list from: {bucket}/{S3Prefix}') response_iterator = paginator.paginate( Bucket=bucket, Prefix=S3Prefix ) for page in response_iterator: if "Contents" in page: logger.info(f'Got list_objects_v2 {bucket}/{S3Prefix}: {len(page["Contents"])}') for n in page["Contents"]: # 取这个key的versionId，如果是ver_list是空，则会全部都是None ver = ver_list.get(n["Key"]) # 目的桶要去掉prefix file_list.append({ "Key": n["Key"][dp_len:], "Size": n["Size"], "versionId": ver if (ver is not None and JobsenderCompareVersionId) else 'null' }) except Exception as err: logger.error(f'Fail to get s3 list_objects_v2 {bucket}/{S3Prefix}: {str(err)}') logger.info(f'Bucket list length(get_des_file_list)：{str(len(file_list))}') return file_list # Get s3 object versionId record in DDB def get_versionid_from_ddb(*, des_bucket, table): logger.info(f'Get des_bucket versionId list from DDB') ver_list = {} try: r = table.query( IndexName='desBucket-index', KeyConditionExpression='desBucket=:b', ExpressionAttributeValues={":b": des_bucket} ) if 'Items' in r: for i in r['Items']: ver_list[i['desKey']] = i['versionId'] logger.info(f'Got versionId list {des_bucket}: {len(ver_list)}') except Exception as e: logger.error(f'Fail to query DDB for versionId {des_bucket}- {str(e)}') return ver_list # Jobsender compare source and destination bucket list def delta_job_list(*, src_file_list, des_file_list, src_bucket, src_prefix, des_bucket, des_prefix, ignore_list, JobsenderCompareVersionId): # Delta list，只对比key&size，version不做对比，只用来发Job logger.info(f'Compare source s3://{src_bucket}/{src_prefix} and destination s3://{des_bucket}/{des_prefix}') start_time = int(time.time()) job_list = [] ignore_records = [] for src in src_file_list: # 排除掉 ignore_list 里面列的 bucket/key src_bucket_key = src_bucket + '/' + src['Key'] ignore_match = False # 每个 ignore key 匹配一次，匹配上任何一个就跳过这个scr_key for ignore_key in ignore_list: if fnmatchcase(src_bucket_key, ignore_key): ignore_match = True break # 匹配不上，循环下一个 ignore_key if ignore_match: ignore_records.append(src_bucket_key) continue # 跳过当前 src # 比对源文件是否在目标中，会连带versionId一起比较 if src in des_file_list: continue # 在List中，下一个源文件 # 不在List中，把源文件加入job list else: Des_key = str(PurePosixPath(des_prefix) / src["Key"]) if src["Key"][-1] == '/': # 源Key是个目录的情况，需要额外加 / Des_key += '/' job_list.append( { "Src_bucket": src_bucket, "Src_key": src["Key"], # Src_key已经包含了Prefix "Des_bucket": des_bucket, "Des_key": Des_key, "Size": src["Size"], "versionId": src['versionId'] } ) spent_time = int(time.time()) - start_time if JobsenderCompareVersionId: logger.info(f'Finish compare key/size/versionId in {spent_time} Seconds (JobsenderCompareVersionId is enable)') else: logger.info(f'Finish compare key/size in {spent_time} Seconds (JobsenderCompareVersionId is disable)') logger.info(f'Delta Job List: {len(job_list)} - Ignore List: {len(ignore_records)}') return job_list, ignore_records def job_upload_sqs_ddb(*, sqs, sqs_queue, job_list): sqs_batch = 0 sqs_message = [] logger.info(f'Start uploading jobs to queue: {sqs_queue}') # create ddb writer # with table.batch_writer() as ddb_batch: for job in job_list: # construct sqs messages sqs_message.append({ "Id": str(sqs_batch), "MessageBody": json.dumps(job), }) sqs_batch += 1 # write to sqs in batch 10 or is last one if sqs_batch == 10 or job == job_list[-1]: try: sqs.send_message_batch(QueueUrl=sqs_queue, Entries=sqs_message) except Exception as e: logger.error(f'Fail to send sqs message: {str(sqs_message)}, {str(e)}') sqs_batch = 0 sqs_message = [] logger.info(f'Complete upload job to queue: {sqs_queue}') return # Split one file size into list of start byte position list def split(Size, ChunkSize): partnumber = 1 indexList = [0] if int(Size / ChunkSize) + 1 > 10000: ChunkSize = int(Size / 10000) + 1024 # 对于大于10000分片的大文件，自动调整Chunksize logger.info(f'Size excess 10000 parts limit. Auto change ChunkSize to {ChunkSize}') while ChunkSize * partnumber < Size: # 如果刚好是"="，则无需再分下一part，所以这里不能用"<=" indexList.append(ChunkSize * partnumber) partnumber += 1 return indexList, ChunkSize # Get S3 versionID def head_s3_version(*, s3_src_client, Src_bucket, Src_key): logger.info(f'Try to update VersionId: {Src_bucket}/{Src_key}') try: head = s3_src_client.head_object( Bucket=Src_bucket, Key=Src_key ) versionId = head['VersionId'] Size = head['ContentLength'] logger.info(f'Got VersionId: {versionId} - {Src_bucket}/{Src_key} - Size:{Size}') except Exception as e: logger.error(f'Fail to head s3 - {Src_bucket}/{Src_key}, {str(e)}') return "ERR", 0 return versionId, Size # Get unfinished multipart upload id from s3 def get_uploaded_list(*, s3_client, Des_bucket, Des_key): multipart_uploaded_list = [] paginator = s3_client.get_paginator('list_multipart_uploads') try: logger.info(f'Getting unfinished upload id list - {Des_bucket}/{Des_key}...') response_iterator = paginator.paginate( Bucket=Des_bucket, Prefix=Des_key ) for page in response_iterator: if "Uploads" in page: for i in page["Uploads"]: if i["Key"] == Des_key or Des_key == '': multipart_uploaded_list.append({ "Key": i["Key"], "Initiated": i["Initiated"], "UploadId": i["UploadId"] }) logger.info(f'Unfinished upload, Key: {i["Key"]}, Time: {i["Initiated"]}') except Exception as e: logger.error(f'Fail to list multipart upload - {Des_bucket}/{Des_key} - {str(e)}') return multipart_uploaded_list # Check file on the list from get_uploaded_list and get created multipart id def check_file_exist(*, prefix_and_key, UploadIdList): # 查Key是否有未完成的UploadID keyIDList = [] for u in UploadIdList: if u["Key"] == prefix_and_key: keyIDList.append(u) # 如果找不到上传过的Upload，则从头开始传 if not keyIDList: return 'UPLOAD' # 对同一个Key（文件）的不同Upload找出时间最晚的值 UploadID_latest = keyIDList[0] for u in keyIDList: if u["Initiated"] > UploadID_latest["Initiated"]: UploadID_latest = u # pick last one upload id with latest Initiated time logger.info(f"Pick UploadId Initiated Time: {UploadID_latest['Initiated']}") return UploadID_latest["UploadId"] # Check uploaded part number list on Des_bucket def checkPartnumberList(*, Des_bucket, Des_key, uploadId, s3_des_client): partnumberList = [] logger.info(f'Get partnumber list - {Des_bucket}/{Des_key}') paginator = s3_des_client.get_paginator('list_parts') try: response_iterator = paginator.paginate( Bucket=Des_bucket, Key=Des_key, UploadId=uploadId ) for page in response_iterator: if "Parts" in page: logger.info(f'Got list_parts: {len(page["Parts"])} - {Des_bucket}/{Des_key}') for p in page["Parts"]: partnumberList.append(p["PartNumber"]) except Exception as e: logger.error(f'Fail to list parts in checkPartnumberList - {Des_bucket}/{Des_key} - {str(e)}') return [] if partnumberList: # 如果空则表示没有查到已上传的Part logger.info(f"Found uploaded partnumber {len(partnumberList)} - {json.dumps(partnumberList)}" f" - {Des_bucket}/{Des_key}") else: logger.info(f'Part number list is empty - {Des_bucket}/{Des_key}') return partnumberList # Process one job def job_processor(*, uploadId, indexList, partnumberList, job, s3_src_client, s3_des_client, MaxThread, ChunkSize, MaxRetry, JobTimeout, ifVerifyMD5Twice, GetObjectWithVersionId): # 线程生成器，配合thread pool给出每个线程的对应关系，便于设置超时控制 def thread_gen(woker_thread, pool, stop_signal, partnumber, total, md5list, partnumberList, complete_list): for partStartIndex in indexList: # start to upload part if partnumber not in partnumberList: dryrun = False # dryrun 是为了沿用现有的流程做出完成列表，方便后面计算 MD5 else: dryrun = True th = pool.submit(woker_thread, stop_signal=stop_signal, partnumber=partnumber, partStartIndex=partStartIndex, total=total, md5list=md5list, dryrun=dryrun, complete_list=complete_list ) partnumber += 1 yield th # download part from src. s3 and upload to dest. s3 def woker_thread(*, stop_signal, partnumber, partStartIndex, total, md5list, dryrun, complete_list): if stop_signal.is_set(): return "TIMEOUT" Src_bucket = job['Src_bucket'] Src_key = job['Src_key'] Des_bucket = job['Des_bucket'] Des_key = job['Des_key'] versionId = job['versionId'] getBody, chunkdata_md5 = b'', b'' # init # 下载文件 if ifVerifyMD5Twice or not dryrun: # 如果 ifVerifyMD5Twice 则无论是否已有上传过都重新下载，作为校验整个文件用 if GetObjectWithVersionId: logger.info(f"--->Downloading {ChunkSize} Bytes {Src_bucket}/{Src_key} - {partnumber}/{total}" f" - versionId: {versionId}") else: logger.info(f"--->Downloading {ChunkSize} Bytes {Src_bucket}/{Src_key} - {partnumber}/{total}") retryTime = 0 # 正常工作情况下出现 stop_signal 需要退出 Thread while retryTime <= MaxRetry and not stop_signal.is_set(): retryTime += 1 try: if GetObjectWithVersionId: # 按VersionId获取Object response_get_object = s3_src_client.get_object( Bucket=Src_bucket, Key=Src_key, VersionId=versionId, Range="bytes=" + str(partStartIndex) + "-" + str(partStartIndex + ChunkSize - 1) ) else: # 不带VersionId，即获取最新对象 response_get_object = s3_src_client.get_object( Bucket=Src_bucket, Key=Src_key, Range="bytes=" + str(partStartIndex) + "-" + str(partStartIndex + ChunkSize - 1) ) getBody = response_get_object["Body"].read() chunkdata_md5 = hashlib.md5(getBody) md5list[partnumber - 1] = chunkdata_md5 break # 完成下载，不用重试 except ClientError as err: if err.response['Error']['Code'] in ['NoSuchKey', 'AccessDenied']: # 没这个ID，文件已经删除，或者无权限访问 logger.error(f"ClientError: Fail to access {Src_bucket}/{Src_key} - ERR: {str(err)}.") stop_signal.set() return "QUIT" logger.warning(f"ClientError: Fail to download {Src_bucket}/{Src_key} - ERR: {str(err)}. " f"Retry part: {partnumber} - Attempts: {retryTime}") if retryTime >= MaxRetry: # 超过次数退出 logger.error(f"ClientError: Quit for Max Download retries: {retryTime} - " f"{Src_bucket}/{Src_key}") stop_signal.set() return "TIMEOUT" # 退出Thread else: time.sleep(5 * retryTime) continue # 递增延迟，返回重试 except Exception as e: logger.warning(f"Fail to download {Src_bucket}/{Src_key} - ERR: {str(e)}. " f"Retry part: {partnumber} - Attempts: {retryTime}") if retryTime >= MaxRetry: # 超过次数退出 logger.error(f"Quit for Max Download retries: {retryTime} - {Src_bucket}/{Src_key}") stop_signal.set() return "TIMEOUT" # 退出Thread else: time.sleep(5 * retryTime) continue # 上传文件 if not dryrun: # 这里就不用考虑 ifVerifyMD5Twice 了， retryTime = 0 while retryTime <= MaxRetry and not stop_signal.is_set(): retryTime += 1 try: logger.info(f'--->Uploading {len(getBody)} Bytes {Des_bucket}/{Des_key} - {partnumber}/{total}') s3_des_client.upload_part( Body=getBody, Bucket=Des_bucket, Key=Des_key, PartNumber=partnumber, UploadId=uploadId, ContentMD5=base64.b64encode(chunkdata_md5.digest()).decode('utf-8') ) # 请求已经带上md5，如果s3校验是错的就Exception break except ClientError as err: if err.response['Error']['Code'] == 'NoSuchUpload': # 没这个ID，则是别人已经完成这个Job了。 logger.warning(f'ClientError: Fail to upload part - might be duplicated job:' f' {Des_bucket}/{Des_key}, {str(err)}') stop_signal.set() return "QUIT" logger.warning(f"ClientError: Fail to upload part - {Des_bucket}/{Des_key} - {str(err)}, " f"retry part: {partnumber} Attempts: {retryTime}") if retryTime >= MaxRetry: logger.error(f"ClientError: Quit for Max Upload retries: {retryTime} - {Des_bucket}/{Des_key}") # 改为跳下一个文件 stop_signal.set() return "TIMEOUT" else: time.sleep(5 * retryTime) # 递增延迟重试 continue except Exception as e: logger.warning(f"Fail to upload part - {Des_bucket}/{Des_key} - {str(e)}, " f"retry part: {partnumber} Attempts: {retryTime}") if retryTime >= MaxRetry: logger.error(f"Quit for Max Upload retries: {retryTime} - {Des_bucket}/{Des_key}") # 改为跳下一个文件 stop_signal.set() return "TIMEOUT" else: time.sleep(5 * retryTime) continue if not stop_signal.is_set(): complete_list.append(partnumber) if not dryrun: logger.info( f'--->Complete {len(getBody)} Bytes {Src_bucket}/{Src_key}' f' - {partnumber}/{total} {len(complete_list) / total:.2%}') else: return "TIMEOUT" return "COMPLETE" # woker_thread END # job_processor Main partnumber = 1 # 当前循环要上传的Partnumber total = len(indexList) md5list = [hashlib.md5(b'')] * total complete_list = [] # 线程池 try: stop_signal = threading.Event() # 用于JobTimeout终止当前文件的所有线程 with concurrent.futures.ThreadPoolExecutor(max_workers=MaxThread) as pool: # 这里要用迭代器拿到threads对象 threads = list(thread_gen(woker_thread, pool, stop_signal, partnumber, total, md5list, partnumberList, complete_list)) result = concurrent.futures.wait(threads, timeout=JobTimeout, return_when="ALL_COMPLETED") # 异常退出 if "QUIT" in [t.result() for t in result[0]]: # result[0] 是函数done logger.warning(f'QUIT. Canceling {len(result[1])} waiting threads in pool ...') stop_signal.set() for t in result[1]: t.cancel() logger.warning(f'QUIT Job: {job["Src_bucket"]}/{job["Src_key"]}') return "QUIT" # 超时 if len(result[1]) > 0: # # result[0] 是函数not_done, 即timeout有未完成的 logger.warning(f'TIMEOUT. Canceling {len(result[1])} waiting threads in pool ...') stop_signal.set() for t in result[1]: t.cancel() logger.warning(f'TIMEOUT {JobTimeout}S Job: {job["Src_bucket"]}/{job["Src_key"]}') return "TIMEOUT" # 线程池End logger.info(f'All parts uploaded: {job["Src_bucket"]}/{job["Src_key"]} - Size:{job["Size"]}') # 计算所有分片列表的总etag: cal_etag digests = b"".join(m.digest() for m in md5list) md5full = hashlib.md5(digests) cal_etag = '"%s-%s"' % (md5full.hexdigest(), len(md5list)) except Exception as e: logger.error(f'Exception in job_processor: {str(e)}') return "ERR" return cal_etag # Complete multipart upload # 通过查询回来的所有Part列表uploadedListParts来构建completeStructJSON def completeUpload(*, uploadId, Des_bucket, Des_key, len_indexList, s3_des_client): # 查询S3的所有Part列表uploadedListParts构建completeStructJSON uploadedListPartsClean = [] logger.info(f'Get partnumber list - {Des_bucket}/{Des_key}') paginator = s3_des_client.get_paginator('list_parts') try: response_iterator = paginator.paginate( Bucket=Des_bucket, Key=Des_key, UploadId=uploadId ) # 把 ETag 加入到 Part List for page in response_iterator: if "Parts" in page: logger.info(f'Got list_parts: {len(page["Parts"])} - {Des_bucket}/{Des_key}') for p in page["Parts"]: uploadedListPartsClean.append({ "ETag": p["ETag"], "PartNumber": p["PartNumber"] }) except ClientError as e: if e.response['Error']['Code'] == 'NoSuchUpload': # 没这个ID，则是别人已经完成这个Job了。 logger.warning(f'ClientError: Fail to list parts while completeUpload - might be duplicated job:' f' {Des_bucket}/{Des_key}, {str(err)}') return "QUIT" logger.error(f'ClientError: Fail to list parts while completeUpload - {Des_bucket}/{Des_key} - {str(e)}') return "ERR" except Exception as e: logger.error(f'Fail to list parts while completeUpload - {Des_bucket}/{Des_key} - {str(e)}') return "ERR" if len(uploadedListPartsClean) != len_indexList: logger.error(f'Uploaded parts size not match - {Des_bucket}/{Des_key}') return "ERR" completeStructJSON = {"Parts": uploadedListPartsClean} # S3合并multipart upload任务 try: logger.info(f'Try to merge multipart upload {Des_bucket}/{Des_key}') response_complete = s3_des_client.complete_multipart_upload( Bucket=Des_bucket, Key=Des_key, UploadId=uploadId, MultipartUpload=completeStructJSON ) result = response_complete['ETag'] logger.info(f'Merged: {Des_bucket}/{Des_key}') except Exception as e: logger.error(f'Fail to complete multipart upload {Des_bucket}/{Des_key}, {str(e)}') return "ERR" logger.info(f'Complete merge file {Des_bucket}/{Des_key}') return result # Continuely get job message to invoke one processor per job def job_looper(*, sqs, sqs_queue, table, s3_src_client, s3_des_client, instance_id, StorageClass, ChunkSize, MaxRetry, MaxThread, ResumableThreshold, JobTimeout, ifVerifyMD5Twice, CleanUnfinishedUpload, Des_bucket_default, Des_prefix_default, UpdateVersionId, GetObjectWithVersionId): while True: # Get Job from sqs try: logger.info('Get Job from sqs queue...') sqs_job_get = sqs.receive_message(QueueUrl=sqs_queue) # Empty queue message available if 'Messages' not in sqs_job_get: # No message on sqs queue logger.info('No message in queue available, wait...') time.sleep(60) # 拿到 Job message else: # TODO: 尚未完整处理 SQS 存在多条消息的情况的意外中断处理，建议只用于针对一次取一个SQS消息 for sqs_job in sqs_job_get["Messages"]: job = json.loads(sqs_job["Body"]) job_receipt = sqs_job["ReceiptHandle"] # 用于后面删除message # 判断是S3来的消息，而不是jodsender来的就转换一下 if 'Records' in job: # S3来的消息带着'Records' for One_record in job['Records']: if 's3' in One_record: Src_bucket = One_record['s3']['bucket']['name'] Src_key = One_record['s3']['object']['key'] Src_key = urllib.parse.unquote_plus(Src_key) Size = One_record['s3']['object']['size'] if "versionId" in One_record['s3']['object']: versionId = One_record['s3']['object']['versionId'] else: versionId = 'null' Des_bucket, Des_prefix = Des_bucket_default, Des_prefix_default Des_key = str(PurePosixPath(Des_prefix) / Src_key) if Src_key[-1] == '/': # 针对空目录对象 Des_key += '/' job = { 'Src_bucket': Src_bucket, 'Src_key': Src_key, 'Size': Size, 'Des_bucket': Des_bucket, 'Des_key': Des_key, 'versionId': versionId } if 'Des_bucket' not in job and 'Event' not in job: logger.warning(f'Wrong sqs job: {json.dumps(job, default=str)}') logger.warning('Try to handle next message') time.sleep(1) continue if 'versionId' not in job: job['versionId'] = 'null' # 主流程 if 'Event' not in job: if job['Size'] > ResumableThreshold: upload_etag_full = step_function( job=job, table=table, s3_src_client=s3_src_client, s3_des_client=s3_des_client, instance_id=instance_id, StorageClass=StorageClass, ChunkSize=ChunkSize, MaxRetry=MaxRetry, MaxThread=MaxThread, JobTimeout=JobTimeout, ifVerifyMD5Twice=ifVerifyMD5Twice, CleanUnfinishedUpload=CleanUnfinishedUpload, UpdateVersionId=UpdateVersionId, GetObjectWithVersionId=GetObjectWithVersionId ) else: upload_etag_full = step_fn_small_file( job=job, table=table, s3_src_client=s3_src_client, s3_des_client=s3_des_client, instance_id=instance_id, StorageClass=StorageClass, MaxRetry=MaxRetry, UpdateVersionId=UpdateVersionId, GetObjectWithVersionId=GetObjectWithVersionId ) else: if job['Event'] == 's3:TestEvent': logger.info('Skip s3:TestEvent') upload_etag_full = "s3:TestEvent" else: upload_etag_full = "OtherEvent" # Del Job on sqs logger.info(f'upload_etag_full={upload_etag_full}, job={str(job)}') if upload_etag_full != "TIMEOUT": # 如果是超时的就不删SQS消息，是正常结束或QUIT就删 # QUIT 是 NoSuchUpload, NoSuchKey, AccessDenied，可以认为没必要再让下一个worker再试了 # 直接删除SQS，并且DDB并不会记录结束状态 try: logger.info(f'Try to finsh job message on sqs. {str(job)}') sqs.delete_message( QueueUrl=sqs_queue, ReceiptHandle=job_receipt ) except Exception as e: logger.error(f'Fail to delete sqs message: {str(sqs_job)}, {str(e)}') except Exception as e: logger.error(f'Fail. Wait for 5 seconds. ERR: {str(e)}') time.sleep(5) # Finish Job, go back to get next job in queue # 清理S3上现有未完成的Multipart Upload ID（当前Job） def clean_multipart_upload(*, s3_client, multipart_uploaded_list, Des_bucket): for clean_i in multipart_uploaded_list: try: s3_client.abort_multipart_upload( Bucket=Des_bucket, Key=clean_i["Key"], UploadId=clean_i["UploadId"] ) logger.info(f'CLEAN FINISHED - {str(multipart_uploaded_list)}') except Exception as e: logger.error(f'Fail to clean - {str(multipart_uploaded_list)} - {str(e)}') # Steps func for multipart upload for one job def step_function(*, job, table, s3_src_client, s3_des_client, instance_id, StorageClass, ChunkSize, MaxRetry, MaxThread, JobTimeout, ifVerifyMD5Twice, CleanUnfinishedUpload, UpdateVersionId, GetObjectWithVersionId): # 正常开始处理 Src_bucket = job['Src_bucket'] Src_key = job['Src_key'] Size = job['Size'] Des_bucket = job['Des_bucket'] Des_key = job['Des_key'] versionId = job['versionId'] upload_etag_full = "" logger.info(f'Start multipart: {Src_bucket}/{Src_key}, Size: {Size}, versionId: {versionId}') # Get dest s3 unfinish multipart upload of this file multipart_uploaded_list = get_uploaded_list( s3_client=s3_des_client, Des_bucket=Des_bucket, Des_key=Des_key ) # 设置了CleanUnfinishedUpload，就清理S3上现有未完成的Multipart Upload ID（当前Job），不做断点续传 if multipart_uploaded_list and CleanUnfinishedUpload: logger.warning(f'You set CleanUnfinishedUpload. Now clean: {str(multipart_uploaded_list)}') clean_multipart_upload( s3_client=s3_des_client, multipart_uploaded_list=multipart_uploaded_list, Des_bucket=Des_bucket ) multipart_uploaded_list = [] # 开始 Job 步骤 # 循环重试3次（如果MD5计算的ETag，或VersionID不一致） for md5_retry in range(Max_md5_retry + 1): # Job 准备 # 检查文件没Multipart UploadID要新建, 有则 return UploadID response_check_upload = check_file_exist( prefix_and_key=Des_key, UploadIdList=multipart_uploaded_list ) if response_check_upload == 'UPLOAD': try: logger.info(f'Create multipart upload - {Des_bucket}/{Des_key}') response_new_upload = s3_des_client.create_multipart_upload( Bucket=Des_bucket, Key=Des_key, StorageClass=StorageClass ) # If update versionID enabled, update s3 versionID # 但可能会出现中断重传的时候，拿到了另一个新version，从而导致文件半老半新，所以需要在最后完成时候校验一次versionId if UpdateVersionId: versionId, Size = head_s3_version( s3_src_client=s3_src_client, Src_bucket=Src_bucket, Src_key=Src_key ) if versionId == "ERR": break job['versionId'] = versionId job['Size'] = Size reponse_uploadId = response_new_upload["UploadId"] partnumberList = [] # Write log to DDB in first round of job # ddb_first_round(table, Src_bucket, Src_key, Size, versionId) except Exception as e: logger.error(f'Fail to create new multipart upload - {Des_bucket}/{Des_key} - {str(e)}') upload_etag_full = "ERR" break else: reponse_uploadId = response_check_upload logger.info(f'Resume upload id: {Des_bucket}/{Des_key}') # 获取已上传partnumberList partnumberList = checkPartnumberList( Des_bucket=Des_bucket, Des_key=Des_key, uploadId=reponse_uploadId, s3_des_client=s3_des_client ) # Get versionId/Size from DDB，如果传输前或中断后文件被替换，Size可能改变，所以重新取启动时候的versionId和当时的Size if UpdateVersionId: versionId, Size = ddb_get( table=table, Src_bucket=Src_bucket, Src_key=Src_key ) if versionId == "ERR": break job['versionId'] = versionId job['Size'] = Size # 获取文件拆分片索引列表，例如[0, 10, 20] indexList, ChunkSize_auto = split( Size, ChunkSize ) # 对于大于10000分片的大文件，自动调整为Chunksize_auto # Write log to DDB in first round of job percent = int(len(partnumberList) / len(indexList) * 100) # ddb_this_round(table, percent, Src_bucket, Src_key, instance_id) ddb_start(table=table, percent=percent, job=job, instance_id=instance_id, new_upload=(response_check_upload == 'UPLOAD')) # Job Thread: uploadPart, 超时或Key不对返回 TIMEOUT/QUIT upload_etag_full = job_processor( uploadId=reponse_uploadId, indexList=indexList, partnumberList=partnumberList, job=job, s3_src_client=s3_src_client, s3_des_client=s3_des_client, MaxThread=MaxThread, ChunkSize=ChunkSize_auto, # 对单个文件使用自动调整的 Chunksize_auto MaxRetry=MaxRetry, JobTimeout=JobTimeout, ifVerifyMD5Twice=ifVerifyMD5Twice, GetObjectWithVersionId=GetObjectWithVersionId ) if upload_etag_full == "TIMEOUT" or upload_etag_full == "QUIT": logger.warning(f'Quit job upload_etag_full == {upload_etag_full} - {str(job)}') break # 退出处理该Job elif upload_etag_full == "ERR": # 清掉已上传id，以便重新上传 logger.error(f'upload_etag_full ERR - {str(job)}') clean_multipart_upload( s3_client=s3_des_client, multipart_uploaded_list=multipart_uploaded_list, Des_bucket=Des_bucket ) multipart_uploaded_list = [] if md5_retry >= Max_md5_retry: logger.error(f'Quit job upload_etag_full ERR - {upload_etag_full} - {str(job)}') return 'ERR' continue # 循环重试Job # 合并S3上的文件 complete_etag = completeUpload( uploadId=reponse_uploadId, Des_bucket=Des_bucket, Des_key=Des_key, len_indexList=len(indexList), s3_des_client=s3_des_client ) if complete_etag == "ERR": # 清掉已上传id，以便重新上传 logger.error(f'complete_etag ERR - {str(job)}') clean_multipart_upload( s3_client=s3_des_client, multipart_uploaded_list=multipart_uploaded_list, Des_bucket=Des_bucket ) multipart_uploaded_list = [] if md5_retry >= Max_md5_retry: logger.error(f'Quit job complete_etag - {upload_etag_full} - {str(job)}') return 'ERR' continue # 循环重试Job # 检查文件MD5 if ifVerifyMD5Twice and complete_etag != "QUIT": # QUIT是因为重复ID，所以就不检查了 if complete_etag == upload_etag_full: logger.info(f'MD5 ETag Matched - {Des_bucket}/{Des_key} - {complete_etag}') break # 结束本文件，下一个sqs job else: # ETag 不匹配，删除目的S3的文件，重试 logger.error(f'MD5 ETag NOT MATCHED {Des_bucket}/{Des_key}( Destination / Origin ): ' f'{complete_etag} - {upload_etag_full}') del_des_s3_object( s3_des_client=s3_des_client, Des_bucket=Des_bucket, Des_key=Des_key ) # 清掉已上传id，以便重新上传 logger.error(f'complete_etag != upload_etag_full ERR - {str(job)}') clean_multipart_upload( s3_client=s3_des_client, multipart_uploaded_list=multipart_uploaded_list, Des_bucket=Des_bucket ) multipart_uploaded_list = [] if md5_retry >= Max_md5_retry: logger.error(f'Quit job ifVerifyMD5Twice - {upload_etag_full} - {str(job)}') return 'ERR' continue # 重新执行Job # DynamoDB log: ADD status: DONE/ERR(upload_etag_full) ddb_complete( upload_etag_full=upload_etag_full, table=table, Src_bucket=Src_bucket, Src_key=Src_key ) # 正常结束 md5_retry 循环 break # END md5_retry 超过次数 # complete one job return upload_etag_full # delete des s3 object def del_des_s3_object(*, s3_des_client, Des_bucket, Des_key): logger.info(f'Delete {Des_bucket}/{Des_key}') try: s3_des_client.delete_object( Bucket=Des_bucket, Key=Des_key ) except Exception as e: logger.error(f'Fail to delete S3 object - {Des_bucket}/{Des_key} - {str(e)}') # Get iteam versionId and Size from DDB def ddb_get(*, table, Src_bucket, Src_key): logger.info(f'Get versionId and Size from DDB - {Src_bucket}/{Src_key}') table_key = str(PurePosixPath(Src_bucket) / Src_key) if Src_key[-1] == '/': # 针对空目录对象 table_key += '/' try: response = table.get_item( Key={"Key": table_key}, AttributesToGet=["versionId", "Size"] ) versionId = response['Item']['versionId'] Size = response['Item']['Size'] logger.info(f'Got VersionId: {versionId} - {Src_bucket}/{Src_key} - Size:{Size}') except Exception as e: logger.error(f'Fail to Get versionId and Size from DDB - {Src_bucket}/{Src_key} - {str(e)}') return 'ERR', 0 return versionId, Size # # Write log to DDB in first round of job def ddb_start(*, table, percent, job, instance_id, new_upload): Src_bucket = job['Src_bucket'] Src_key = job['Src_key'] Size = job['Size'] Des_bucket = job['Des_bucket'] Des_key = job['Des_key'] versionId = job['versionId'] logger.info(f'Write log to DDB start job - {Src_bucket}/{Src_key}') cur_time = time.time() table_key = str(PurePosixPath(Src_bucket) / Src_key) if Src_key[-1] == '/': # 针对空目录对象 table_key += '/' UpdateExpression = "ADD instanceID :id, tryTimes :t, startTime_f :s_format " \ "SET lastTimeProgress=:p, Size=:size, desBucket=:b, desKey=:k" ExpressionAttributeValues = { ":t": 1, ":id": {instance_id}, ":s_format": {time.asctime(time.localtime(cur_time))}, ":size": Size, ":p": percent, ":b": Des_bucket, ":k": Des_key } if new_upload: logger.info(f'Update DDB - {Src_bucket}/{Src_key}') ExpressionAttributeValues[":s"] = int(cur_time) ExpressionAttributeValues[":v"] = versionId UpdateExpression += ", firstTime =:s, versionId =:v" try: table.update_item( Key={"Key": table_key}, UpdateExpression=UpdateExpression, ExpressionAttributeValues=ExpressionAttributeValues ) except Exception as e: # 日志写不了 logger.error(f'Fail to put log to DDB at start job - {Src_bucket}/{Src_key} - {str(e)}') return # DynamoDB log: ADD status: DONE/ERR(upload_etag_full) def ddb_complete(*, upload_etag_full, table, Src_bucket, Src_key): if upload_etag_full not in ["TIMEOUT", "ERR", "QUIT"]: status = "DONE" else: status = upload_etag_full cur_time = time.time() table_key = str(PurePosixPath(Src_bucket) / Src_key) if Src_key[-1] == '/': # 针对空目录对象 table_key += '/' UpdateExpression = "ADD jobStatus :done SET totalSpentTime=:s-firstTime, endTime=:s, endTime_f=:e" ExpressionAttributeValues = { ":done": {status}, ":s": int(cur_time), ":e": time.asctime(time.localtime(cur_time)) } # 正常写DDB，如果是异常的就不加这个lastTimeProgress=100 if status == "DONE": UpdateExpression += ", lastTimeProgress=:p" ExpressionAttributeValues[":p"] = 100 # update DDB logger.info(f'Write job complete status to DDB: {status} - {Src_bucket}/{Src_key}') try: table.update_item( Key={"Key": table_key}, UpdateExpression=UpdateExpression, ExpressionAttributeValues=ExpressionAttributeValues ) except Exception as e: logger.error(f'Fail to put log to DDB at end - {Src_bucket}/{Src_key} - {str(e)}') return def step_fn_small_file(*, job, table, s3_src_client, s3_des_client, instance_id, StorageClass, MaxRetry, UpdateVersionId, GetObjectWithVersionId): # 开始处理小文件 Src_bucket = job['Src_bucket'] Src_key = job['Src_key'] Size = job['Size'] Des_bucket = job['Des_bucket'] Des_key = job['Des_key'] versionId = job['versionId'] # If update versionID enabled, update s3 versionID if UpdateVersionId: versionId = head_s3_version( s3_src_client=s3_src_client, Src_bucket=Src_bucket, Src_key=Src_key ) job['versionId'] = versionId logger.info(f'Start small: {Src_bucket}/{Src_key}, Size: {Size}, versionId: {versionId}') # Write DDB log for first round ddb_start(table=table, percent=0, job=job, instance_id=instance_id, new_upload=True) upload_etag_full = [] for retryTime in range(MaxRetry + 1): try: # Get object logger.info(f'--->Downloading {Size} Bytes {Src_bucket}/{Src_key} - Small file 1/1') if GetObjectWithVersionId: response_get_object = s3_src_client.get_object( Bucket=Src_bucket, Key=Src_key, VersionId=versionId ) else: response_get_object = s3_src_client.get_object( Bucket=Src_bucket, Key=Src_key ) getBody = response_get_object["Body"].read() chunkdata_md5 = hashlib.md5(getBody) ContentMD5 = base64.b64encode(chunkdata_md5.digest()).decode('utf-8') # Put object logger.info(f'--->Uploading {Size} Bytes {Des_bucket}/{Des_key} - Small file 1/1') response_put_object = s3_des_client.put_object( Body=getBody, Bucket=Des_bucket, Key=Des_key, ContentMD5=ContentMD5, StorageClass=StorageClass ) # 请求已经带上md5，如果s3校验是错的就Exception upload_etag_full = response_put_object['ETag'] # 结束 Upload/download break except ClientError as e: if e.response['Error']['Code'] in ['NoSuchKey', 'AccessDenied']: logger.error(f"Fail to access {Src_bucket}/{Src_key} - ERR: {str(e)}.") return "QUIT" logger.warning(f'Download/Upload small file Fail: {Src_bucket}/{Src_key}, ' f'{str(e)}, Attempts: {retryTime}') if retryTime >= MaxRetry: logger.error(f'Fail MaxRetry Download/Upload small file: {Des_bucket}/{Des_key}') return "TIMEOUT" else: time.sleep(5 * retryTime) except Exception as e: logger.error(f'Fail in step_fn_small - {Des_bucket}/{Des_key} - {str(e)}') return "TIMEOUT" # Write DDB log for complete ddb_complete( upload_etag_full=upload_etag_full, table=table, Src_bucket=Src_bucket, Src_key=Src_key ) # complete one job return upload_etag_full