From 8e6aedfdf5e5f74b3bedc350a96177a589125d07 Mon Sep 17 00:00:00 2001 From: zhiren Date: Tue, 11 Apr 2023 17:36:46 -0400 Subject: [PATCH 1/9] update the item_path in attribute attachment to item_id --- app/commands/file.py | 7 +++++-- app/services/file_manager/file_manifests.py | 8 ++++---- app/services/file_manager/file_upload/file_upload.py | 6 +++++- 3 files changed, 14 insertions(+), 7 deletions(-) diff --git a/app/commands/file.py b/app/commands/file.py index f1defc0d..5f21a22d 100644 --- a/app/commands/file.py +++ b/app/commands/file.py @@ -215,9 +215,12 @@ def file_put(**kwargs): # noqa: C901 if source_file: upload_event['valid_source'] = src_file_info - simple_upload(upload_event, num_of_thread=thread, resumable_id=resumable_id, job_id=job_id, item_id=item_id) + file_objects = simple_upload( + upload_event, num_of_thread=thread, resumable_id=resumable_id, job_id=job_id, item_id=item_id + ) - srv_manifest.attach_manifest(attribute, result_file, zone) if attribute else None + # since only file upload can attach manifest, take the first file object + srv_manifest.attach_manifest(attribute, file_objects[0], zone) if attribute else None message_handler.SrvOutPutHandler.all_file_uploaded() diff --git a/app/services/file_manager/file_manifests.py b/app/services/file_manager/file_manifests.py index 7756a48f..e6e98093 100644 --- a/app/services/file_manager/file_manifests.py +++ b/app/services/file_manager/file_manifests.py @@ -61,9 +61,9 @@ def validate_template(self, manifest_json): return False, res_json @require_valid_token() - def attach(self, manifest_json, file_name, zone): + def attach(self, manifest_json: dict, item_id: str, zone: str): url = self.app_config.Connections.url_bff + '/v1/manifest/attach' - manifest_json['file_name'] = file_name + manifest_json['item_id'] = item_id manifest_json['zone'] = zone headers = { 'Authorization': 'Bearer ' + self.user.access_token, @@ -151,8 +151,8 @@ def validate_manifest(self, manifest, raise_error=True): validation_error = '' return validation, validation_error - def attach_manifest(self, manifest, file_name, zone): - res = self.attach(manifest, file_name, zone) + def attach_manifest(self, manifest: dict, item_id: str, zone: str): + res = self.attach(manifest, item_id, zone) if res.get('code') != 200: error = res.get('error_msg') if self.interactive: diff --git a/app/services/file_manager/file_upload/file_upload.py b/app/services/file_manager/file_upload/file_upload.py index 93680427..5e5d251a 100644 --- a/app/services/file_manager/file_upload/file_upload.py +++ b/app/services/file_manager/file_upload/file_upload.py @@ -8,6 +8,7 @@ import zipfile from multiprocessing.pool import ThreadPool from typing import Dict +from typing import List from typing import Tuple import click @@ -15,6 +16,7 @@ import app.services.logger_services.log_functions as logger import app.services.output_manager.message_handler as mhandler from app.configs.app_config import AppConfig +from app.services.file_manager.file_upload.models import FileObject from app.services.file_manager.file_upload.models import UploadType from app.services.file_manager.file_upload.upload_client import UploadClient from app.services.output_manager.error_handler import ECustomizedError @@ -108,7 +110,7 @@ def simple_upload( # noqa: C901 resumable_id: str = None, job_id: str = None, item_id: str = None, -): +) -> List[FileObject]: upload_start_time = time.time() my_file = upload_event.get('file') project_code = upload_event.get('project_code') @@ -215,3 +217,5 @@ def simple_upload( # noqa: C901 num_of_file = len(upload_file_path) logger.info(f'Upload Time: {time.time() - upload_start_time:.2f}s for {num_of_file:d} files') + + return pre_upload_infos From 8e46af004f0b79c351131a394e69d3da37a119ae Mon Sep 17 00:00:00 2001 From: zhiren Date: Wed, 12 Apr 2023 11:45:50 -0400 Subject: [PATCH 2/9] fixup the token fresh will immediately exit after the async function --- app/configs/app_config.py | 2 +- .../file_manager/file_upload/file_upload.py | 23 +++++++++++++++---- 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/app/configs/app_config.py b/app/configs/app_config.py index 3d2ba3e8..12b2bf45 100644 --- a/app/configs/app_config.py +++ b/app/configs/app_config.py @@ -13,7 +13,7 @@ class Env(object): msg_path = ConfigClass.custom_path user_config_file = f'{user_config_path}/config.ini' token_warn_need_refresh = 250 # refresh token if token is about to expire - token_refresh_interval = 120 # auto refresh token every 2 minutes + token_refresh_interval = 10 # auto refresh token every 2 minutes # NOTE: there is a limitation on minio that # the multipart number is 10000. so we set diff --git a/app/services/file_manager/file_upload/file_upload.py b/app/services/file_manager/file_upload/file_upload.py index 8b1b7e2e..be4e747b 100644 --- a/app/services/file_manager/file_upload/file_upload.py +++ b/app/services/file_manager/file_upload/file_upload.py @@ -181,14 +181,22 @@ def simple_upload( # noqa: C901 pool = ThreadPool(num_of_thread + 1) pool.apply_async(upload_client.upload_token_refresh) + on_success_res = [] for file_object in pre_upload_infos: chunk_res = upload_client.stream_upload(file_object, pool) # NOTE: if there is some racing error make the combine chunks # out of thread pool. - pool.apply_async( + res = pool.apply_async( upload_client.on_succeed, args=(file_object, tags, chunk_res), ) + on_success_res.append(res) + + # finish the upload once all on success api return + # otherwise wait for 1 second and check again + for res in on_success_res: + while res.get() is None: + time.sleep(1) upload_client.set_finish_upload() pool.close() @@ -263,15 +271,22 @@ def resume_upload( pool = ThreadPool(num_of_thread + 1) pool.apply_async(upload_client.upload_token_refresh) + on_success_res = [] for file_object in unfinished_items: - upload_client.stream_upload(file_object, pool) + chunk_res = upload_client.stream_upload(file_object, pool) # NOTE: if there is some racing error make the combine chunks # out of thread pool. - pool.apply_async( + res = pool.apply_async( upload_client.on_succeed, - args=(file_object, manifest_json.get('tags')), + args=(file_object, manifest_json.get('tags'), chunk_res), ) + on_success_res.append(res) + # finish the upload once all on success api return + # otherwise wait for 1 second and check again + for res in on_success_res: + while res.get() is None: + time.sleep(1) upload_client.set_finish_upload() pool.close() From 605d83a4eefdff88a4f7fb76ca0fe1b7f4b70d7f Mon Sep 17 00:00:00 2001 From: zhiren Date: Wed, 12 Apr 2023 15:22:36 -0400 Subject: [PATCH 3/9] adding the item id as identifier for manifest attaching --- app/commands/file.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/commands/file.py b/app/commands/file.py index 5f21a22d..a8fd5224 100644 --- a/app/commands/file.py +++ b/app/commands/file.py @@ -220,7 +220,7 @@ def file_put(**kwargs): # noqa: C901 ) # since only file upload can attach manifest, take the first file object - srv_manifest.attach_manifest(attribute, file_objects[0], zone) if attribute else None + srv_manifest.attach_manifest(attribute, file_objects[0].item_id, zone) if attribute else None message_handler.SrvOutPutHandler.all_file_uploaded() From e03209b745dc0c56f1a941fe82a8deef59666fb8 Mon Sep 17 00:00:00 2001 From: zhiren Date: Wed, 12 Apr 2023 16:31:56 -0400 Subject: [PATCH 4/9] add the test case for normal upload w/o attribute --- tests/app/commands/test_file.py | 54 +++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/tests/app/commands/test_file.py b/tests/app/commands/test_file.py index d560be53..160299b1 100644 --- a/tests/app/commands/test_file.py +++ b/tests/app/commands/test_file.py @@ -2,11 +2,65 @@ # # Contact Indoc Research for any questions regarding the use of this source code. +import click + +from app.commands.file import file_put from app.commands.file import file_resume +from app.services.file_manager.file_upload.models import FileObject from app.services.output_manager.error_handler import ECustomizedError from app.services.output_manager.error_handler import customized_error_msg +def test_file_upload_command_success(mocker, cli_runner): + project_code = 'test_project' + target_folder = 'admin' + + mocker.patch('app.commands.file.identify_target_folder', return_value=(project_code, target_folder)) + mocker.patch('app.commands.file.validate_upload_event', return_value={'source_file': '', 'attribute': ''}) + mocker.patch('app.commands.file.assemble_path', return_value=('test', {'id': 'id'}, True, 'test')) + simple_upload_mock = mocker.patch('app.commands.file.simple_upload', return_value=None) + + # create a test file + runner = click.testing.CliRunner() + with runner.isolated_filesystem(): + with open('test.txt', 'w') as f: + f.write('test.txt') + + result = cli_runner.invoke(file_put, ['--project-path', 'test', '--thread', 1, 'test.txt']) + assert result.exit_code == 0 + simple_upload_mock.assert_called_once() + + +def test_file_upload_command_success_with_attribute(mocker, cli_runner): + project_code = 'test_project' + target_folder = 'admin' + + mocker.patch('app.commands.file.identify_target_folder', return_value=(project_code, target_folder)) + mocker.patch('app.commands.file.validate_upload_event', return_value={'source_file': '', 'attribute': 'test'}) + mocker.patch('app.commands.file.assemble_path', return_value=('test', {'id': 'id'}, True, 'test')) + + mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(1, 1)) + test_obj = FileObject('resumable_id', 'job_id', 'item_id', 'object/path', 'local_path', []) + + simple_upload_mock = mocker.patch('app.commands.file.simple_upload', return_value=[test_obj]) + attribute_mock = mocker.patch( + 'app.services.file_manager.file_manifests.SrvFileManifests.attach_manifest', return_value=None + ) + + # create a test file + runner = click.testing.CliRunner() + with runner.isolated_filesystem(): + with open('test.txt', 'w') as f: + f.write('test.txt') + + result = cli_runner.invoke( + file_put, ['--project-path', 'test', '--thread', 1, '--attribute', 'test.json', 'test.txt'] + ) + assert result.exit_code == 0 + simple_upload_mock.assert_called_once() + attribute_mock.assert_called_once() + + def test_resumable_upload_command_success(mocker, cli_runner): mocker.patch('os.path.exists', return_value=True) # mock the open function From d1184e88c92f26529a1e5bc1d5b6dd5344b01d0a Mon Sep 17 00:00:00 2001 From: zhiren Date: Thu, 13 Apr 2023 10:20:36 -0400 Subject: [PATCH 5/9] add back the toekn refresh interval --- app/configs/app_config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/configs/app_config.py b/app/configs/app_config.py index 12b2bf45..3d2ba3e8 100644 --- a/app/configs/app_config.py +++ b/app/configs/app_config.py @@ -13,7 +13,7 @@ class Env(object): msg_path = ConfigClass.custom_path user_config_file = f'{user_config_path}/config.ini' token_warn_need_refresh = 250 # refresh token if token is about to expire - token_refresh_interval = 10 # auto refresh token every 2 minutes + token_refresh_interval = 120 # auto refresh token every 2 minutes # NOTE: there is a limitation on minio that # the multipart number is 10000. so we set From 052e304ed5e8eeb424b2e3acace4f283b2ebe741 Mon Sep 17 00:00:00 2001 From: zhiren Date: Thu, 13 Apr 2023 11:39:11 -0400 Subject: [PATCH 6/9] update the while loop into wait() function --- app/commands/file.py | 4 ++-- .../file_manager/file_upload/file_upload.py | 13 +++++-------- 2 files changed, 7 insertions(+), 10 deletions(-) diff --git a/app/commands/file.py b/app/commands/file.py index 516777df..39b287af 100644 --- a/app/commands/file.py +++ b/app/commands/file.py @@ -200,10 +200,10 @@ def file_put(**kwargs): # noqa: C901 if source_file: upload_event['valid_source'] = src_file_info - file_objects = simple_upload(upload_event, num_of_thread=thread, output_path=output_path) + item_ids = simple_upload(upload_event, num_of_thread=thread, output_path=output_path) # since only file upload can attach manifest, take the first file object - srv_manifest.attach_manifest(attribute, file_objects[0].item_id, zone) if attribute else None + srv_manifest.attach_manifest(attribute, item_ids[0].item_id, zone) if attribute else None message_handler.SrvOutPutHandler.all_file_uploaded() diff --git a/app/services/file_manager/file_upload/file_upload.py b/app/services/file_manager/file_upload/file_upload.py index 92207b1a..eea78d61 100644 --- a/app/services/file_manager/file_upload/file_upload.py +++ b/app/services/file_manager/file_upload/file_upload.py @@ -9,6 +9,7 @@ from multiprocessing.pool import ThreadPool from typing import Any from typing import Dict +from typing import List from typing import Tuple import click @@ -107,7 +108,7 @@ def simple_upload( # noqa: C901 upload_event, num_of_thread: int = 1, output_path: str = None, -): +) -> List[str]: upload_start_time = time.time() my_file = upload_event.get('file') project_code = upload_event.get('project_code') @@ -194,9 +195,7 @@ def simple_upload( # noqa: C901 # finish the upload once all on success api return # otherwise wait for 1 second and check again - for res in on_success_res: - while res.get() is None: - time.sleep(1) + [res.wait() for res in on_success_res] upload_client.set_finish_upload() pool.close() @@ -216,7 +215,7 @@ def simple_upload( # noqa: C901 num_of_file = len(upload_file_path) logger.info(f'Upload Time: {time.time() - upload_start_time:.2f}s for {num_of_file:d} files') - return pre_upload_infos + return [file_object.item_id for file_object in pre_upload_infos] def resume_upload( @@ -286,9 +285,7 @@ def resume_upload( # finish the upload once all on success api return # otherwise wait for 1 second and check again - for res in on_success_res: - while res.get() is None: - time.sleep(1) + [res.wait() for res in on_success_res] upload_client.set_finish_upload() pool.close() From 3c6a0afae2b3ef14f0a4a472af19673b6fad8e9d Mon Sep 17 00:00:00 2001 From: zhiren Date: Thu, 13 Apr 2023 11:39:48 -0400 Subject: [PATCH 7/9] remove the unnecessary test --- tests/app/commands/test_file.py | 20 -------------------- 1 file changed, 20 deletions(-) diff --git a/tests/app/commands/test_file.py b/tests/app/commands/test_file.py index 160299b1..d6f25f49 100644 --- a/tests/app/commands/test_file.py +++ b/tests/app/commands/test_file.py @@ -11,26 +11,6 @@ from app.services.output_manager.error_handler import customized_error_msg -def test_file_upload_command_success(mocker, cli_runner): - project_code = 'test_project' - target_folder = 'admin' - - mocker.patch('app.commands.file.identify_target_folder', return_value=(project_code, target_folder)) - mocker.patch('app.commands.file.validate_upload_event', return_value={'source_file': '', 'attribute': ''}) - mocker.patch('app.commands.file.assemble_path', return_value=('test', {'id': 'id'}, True, 'test')) - simple_upload_mock = mocker.patch('app.commands.file.simple_upload', return_value=None) - - # create a test file - runner = click.testing.CliRunner() - with runner.isolated_filesystem(): - with open('test.txt', 'w') as f: - f.write('test.txt') - - result = cli_runner.invoke(file_put, ['--project-path', 'test', '--thread', 1, 'test.txt']) - assert result.exit_code == 0 - simple_upload_mock.assert_called_once() - - def test_file_upload_command_success_with_attribute(mocker, cli_runner): project_code = 'test_project' target_folder = 'admin' From 1783d53a92caba2a65330654cb8a92ed8577b5d7 Mon Sep 17 00:00:00 2001 From: zhiren Date: Thu, 13 Apr 2023 12:09:31 -0400 Subject: [PATCH 8/9] fixup the test case --- test | 0 tests/app/commands/test_file.py | 2 +- 2 files changed, 1 insertion(+), 1 deletion(-) create mode 100644 test diff --git a/test b/test new file mode 100644 index 00000000..e69de29b diff --git a/tests/app/commands/test_file.py b/tests/app/commands/test_file.py index d6f25f49..a82fd154 100644 --- a/tests/app/commands/test_file.py +++ b/tests/app/commands/test_file.py @@ -20,7 +20,7 @@ def test_file_upload_command_success_with_attribute(mocker, cli_runner): mocker.patch('app.commands.file.assemble_path', return_value=('test', {'id': 'id'}, True, 'test')) mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(1, 1)) - test_obj = FileObject('resumable_id', 'job_id', 'item_id', 'object/path', 'local_path', []) + test_obj = FileObject('resumable_id', 'job_id', 'item_id', 'object/path', 'local_path') simple_upload_mock = mocker.patch('app.commands.file.simple_upload', return_value=[test_obj]) attribute_mock = mocker.patch( From 20cc11221a3d19d37ff6fc14d8cd47de7e3a099a Mon Sep 17 00:00:00 2001 From: zhiren Date: Thu, 13 Apr 2023 15:22:04 -0400 Subject: [PATCH 9/9] fixup the item_id issue --- app/commands/file.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/commands/file.py b/app/commands/file.py index 39b287af..4d4aacca 100644 --- a/app/commands/file.py +++ b/app/commands/file.py @@ -203,7 +203,7 @@ def file_put(**kwargs): # noqa: C901 item_ids = simple_upload(upload_event, num_of_thread=thread, output_path=output_path) # since only file upload can attach manifest, take the first file object - srv_manifest.attach_manifest(attribute, item_ids[0].item_id, zone) if attribute else None + srv_manifest.attach_manifest(attribute, item_ids[0], zone) if attribute else None message_handler.SrvOutPutHandler.all_file_uploaded()