[80000] Instant profiling assets sharing in DevCloud (#11)

Author: Artyom Tugaryov

automation/bom/image_BOM.txt

@@ -38,6 +38,7 @@ migrations/versions/1e1c4fc30e7c_add_accuracy_import.py
 migrations/versions/2133a18b601e_load_text_dataset_support.py
 migrations/versions/241ced8ef5aa_create_project_report_pipeline.py
 migrations/versions/260ffba4b041_create_inference_report_pipeline.py
+migrations/versions/275bd4d6cd35_instant_assets_sharing.py
 migrations/versions/2a070ca81f20_tensor_distance_report_query_api.py
 migrations/versions/39bd11960375_edge_node_script_in_deployment.py
 migrations/versions/424ed0531227_add_annotate_dataset_job.py
@@ -536,13 +537,14 @@ wb/main/models/analyze_model_input_shape_job_model.py
 wb/main/models/apply_model_layout_model.py
 wb/main/models/artifacts_model.py
 wb/main/models/base_model.py
+wb/main/models/cloud_bundle_model.py
 wb/main/models/convert_dataset_jobs_model.py
 wb/main/models/convert_keras_job_model.py
 wb/main/models/cpu_info_model.py
 wb/main/models/create_accuracy_bundle_job_model.py
 wb/main/models/create_accuracy_scripts_job_model.py
+wb/main/models/create_int8_calibration_bundle_job_model.py
 wb/main/models/create_int8_calibration_scripts_job_model.py
-wb/main/models/create_int8calibration_bundle_job_model.py
 wb/main/models/create_profiling_bundle_job_model.py
 wb/main/models/create_profiling_scripts_job_model.py
 wb/main/models/create_reshape_model_scripts_model.py
@@ -599,6 +601,7 @@ wb/main/models/remote_target_model.py
 wb/main/models/reshape_model_job_model.py
 wb/main/models/setup_environment_job_model.py
 wb/main/models/setup_target_jobs_model.py
+wb/main/models/shared_artifact_model.py
 wb/main/models/single_inference_info_model.py
 wb/main/models/system_resources_model.py
 wb/main/models/target_model.py

config/constants.py

@@ -160,12 +160,16 @@
 CLOUD_SERVICE_URL = get_env_var(name='CLOUD_SERVICE_URL')
 CLOUD_SERVICE_HOST = None
 CLOUD_SERVICE_PORT = None
+CLOUD_SHARED_FOLDER = None
 # dev cloud session duration in minutes
 CLOUD_SERVICE_SESSION_TTL_MINUTES = None
 
 if CLOUD_SERVICE_URL:
     CLOUD_SERVICE_HOST, CLOUD_SERVICE_PORT = parse_host_port_from_url(CLOUD_SERVICE_URL)
     CLOUD_SERVICE_SESSION_TTL_MINUTES = get_env_var(name='CLOUD_SERVICE_SESSION_TTL_MINUTES', cast_function=int)
+    CLOUD_SHARED_FOLDER = os.path.join(ESSENTIAL_DATA_FOLDER, 'bundles')
+
+SETUP_BUNDLE_SUBFOLDER = 'setup_bundle'
 
 WORKBENCH_NETWORK_ALIAS = get_env_var(name='NETWORK_ALIAS', default='localhost')
 
@@ -251,6 +255,7 @@
 ENABLED_FEATURE_PREVIEW_FILE = Path(ROOT_FOLDER) / '.features.json'
 
 FOLDER_PERMISSION = 0o775  # rwxrwxr-x
+FULL_ACCESS_FOLDER_PERMISSION = 0o777  # rwxrwxrwx
 FILE_PERMISSION = 0o664  # rw-rw-r--
 
 DATASET_LABELS_PATH = os.path.join(ROOT_FOLDER, 'wb', 'main', 'accuracy_utils', 'yml_templates', 'datasets_labels')

migrations/versions/275bd4d6cd35_instant_assets_sharing.py

@@ -0,0 +1,85 @@
+"""Instant assets sharing
+
+Revision ID: c2f12b313b48
+Revises: f7fcb58b99cd
+Create Date: 2022-03-01 13:48:42.551162
+
+"""
+
+"""
+ OpenVINO DL Workbench
+ Migration: Instant assets sharing
+
+ Copyright (c) 2022 Intel Corporation
+
+ LEGAL NOTICE: Your use of this software and any required dependent software (the “Software Package”) is subject to
+ the terms and conditions of the software license agreements for Software Package, which may also include
+ notices, disclaimers, or license terms for third party or open source software
+ included in or with the Software Package, and your use indicates your acceptance of all such terms.
+ Please refer to the “third-party-programs.txt” or other similarly-named text file included with the Software Package
+ for additional details.
+ You may obtain a copy of the License at
+      https://software.intel.com/content/dam/develop/external/us/en/documents/intel-openvino-license-agreements.pdf
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = 'c2f12b313b48'
+down_revision = 'f7fcb58b99cd'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.rename_table('downloadable_artifacts', 'shared_artifacts')
+    op.drop_column('create_profiling_bundle_jobs', 'tab_id')
+
+    op.execute("UPDATE artifacts SET type='downloadable_artifact' WHERE type='downloadable_artifacts';")
+
+    op.execute("UPDATE shared_artifacts SET job_id=create_profiling_bundle_jobs.job_id FROM create_profiling_bundle_jobs WHERE shared_artifacts.id=create_profiling_bundle_jobs.bundle_id;")
+    # Connect bundles from downloadable_artifacts with jobs throw job_id instead of bundle_id
+    op.execute("UPDATE shared_artifacts SET job_id=create_profiling_bundle_jobs.job_id FROM create_profiling_bundle_jobs WHERE shared_artifacts.id=create_profiling_bundle_jobs.bundle_id;")
+    op.drop_constraint('create_profiling_bundle_jobs_bundle_id_fkey', 'create_profiling_bundle_jobs', type_='foreignkey')
+    op.drop_column('create_profiling_bundle_jobs', 'bundle_id')
+
+    op.execute("UPDATE shared_artifacts SET job_id=create_int8_calibration_bundle_jobs.job_id FROM create_int8_calibration_bundle_jobs WHERE shared_artifacts.id=create_int8_calibration_bundle_jobs.bundle_id;")
+    op.drop_constraint('create_int8_calibration_bundle_jobs_bundle_id_fkey', 'create_int8_calibration_bundle_jobs',
+                       type_='foreignkey')
+    op.drop_column('create_int8_calibration_bundle_jobs', 'bundle_id')
+
+    op.execute("UPDATE shared_artifacts SET job_id=create_accuracy_bundle_jobs.job_id FROM create_accuracy_bundle_jobs WHERE shared_artifacts.id=create_accuracy_bundle_jobs.bundle_id;")
+    op.drop_constraint('create_accuracy_bundle_jobs_bundle_id_fkey', 'create_accuracy_bundle_jobs', type_='foreignkey')
+    op.drop_column('create_accuracy_bundle_jobs', 'bundle_id')
+
+    op.execute("UPDATE shared_artifacts SET job_id=create_annotate_dataset_bundle_jobs.job_id FROM create_annotate_dataset_bundle_jobs WHERE shared_artifacts.id=create_annotate_dataset_bundle_jobs.bundle_id;")
+    op.drop_constraint('create_annotate_dataset_bundle_jobs_bundle_id_fkey', 'create_annotate_dataset_bundle_jobs',
+                       type_='foreignkey')
+    op.drop_column('create_annotate_dataset_bundle_jobs', 'bundle_id')
+
+    op.execute("UPDATE shared_artifacts SET job_id=create_per_tensor_bundle_jobs.job_id FROM create_per_tensor_bundle_jobs WHERE shared_artifacts.id=create_per_tensor_bundle_jobs.bundle_id;")
+    op.drop_constraint('create_per_tensor_bundle_jobs_bundle_id_fkey', 'create_per_tensor_bundle_jobs',
+                       type_='foreignkey')
+    op.drop_column('create_per_tensor_bundle_jobs', 'bundle_id')
+
+    op.add_column('parse_dev_cloud_result_jobs', sa.Column('are_results_obtained', sa.Boolean(), nullable=True))
+    op.alter_column('parse_dev_cloud_result_jobs', 'result_artifact_id', existing_type=sa.INTEGER(), nullable=True)
+
+    op.drop_constraint('parse_dev_cloud_result_jobs_result_artifact_id_fkey', 'parse_dev_cloud_result_jobs',
+                       type_='foreignkey')
+    op.create_foreign_key(None, 'parse_dev_cloud_result_jobs', 'shared_artifacts', ['result_artifact_id'], ['id'])
+
+    op.drop_constraint('trigger_dev_cloud_jobs_job_bundle_id_fkey', 'trigger_dev_cloud_jobs', type_='foreignkey')
+    op.drop_constraint('trigger_dev_cloud_profiling_jobs_setup_bundle_id_fkey', 'trigger_dev_cloud_jobs',
+                       type_='foreignkey')
+    op.create_foreign_key(None, 'trigger_dev_cloud_jobs', 'shared_artifacts', ['job_bundle_id'], ['id'])
+    op.create_foreign_key(None, 'trigger_dev_cloud_jobs', 'shared_artifacts', ['setup_bundle_id'], ['id'])
+    op.drop_constraint('upload_artifact_to_target_jobs_artifact_id_fkey', 'upload_artifact_to_target_jobs',
+                       type_='foreignkey')
+    op.create_foreign_key(None, 'upload_artifact_to_target_jobs', 'shared_artifacts', ['artifact_id'], ['id'])
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    raise NotImplementedError(f'Downgrade is not implemented for the {revision} migration')

wb/main/api_endpoints/utils.py

@@ -160,53 +160,9 @@ def delete_model_from_db(model_id: int):
 
 
 def delete_dataset_from_db(dataset_id: int):
-    for records in dataset_related_information(dataset_id):
-        DatasetsModel.delete_records(records, get_db_session_for_app())
-
     dataset = DatasetsModel.query.get(dataset_id)
-
     if dataset:
-        dataset_path = dataset.path
         dataset.delete_record(get_db_session_for_app())
-        remove_dir(dataset_path)
-
-
-def dataset_related_information(dataset_id: int):
-    projects = ProjectsModel.query.filter_by(dataset_id=dataset_id).all()
-    all_project_ids = [p.id for p in projects]
-
-    run_results, compound_configs = projects_related_information(all_project_ids)
-
-    all_accuracy_results = (
-        AccuracyJobsModel.query
-            .filter(AccuracyJobsModel.project_id.in_(all_project_ids))
-            .all()
-    )
-    all_int8_results = (
-        Int8CalibrationJobModel.query
-            .filter(Int8CalibrationJobModel.project_id.in_(all_project_ids))
-            .all()
-    )
-
-    return run_results, compound_configs, all_int8_results, all_accuracy_results, projects
-
-
-def projects_related_information(project_ids: List[int]) -> tuple:
-    compound_configs = (
-        ProfilingJobModel.query
-            .filter(ProfilingJobModel.project_id.in_(project_ids))
-            .all()
-    )
-
-    all_infer_config_ids = [i.job_id for i in compound_configs]
-
-    inference_results = (
-        SingleInferenceInfoModel.query
-            .filter(SingleInferenceInfoModel.profiling_job_id.in_(all_infer_config_ids))
-            .all()
-    )
-
-    return inference_results, compound_configs
 
 
 def find_projects(model_id: int, all_levels: bool) -> tuple:

wb/main/api_endpoints/v1/export.py

@@ -37,7 +37,7 @@
 def archive_model(model_id: int):
     download_job: ModelDownloadConfigsModel = ModelDownloadConfigsModel.query.filter_by(model_id=model_id).first()
     if download_job:
-        downloadable_artifact = download_job.downloadable_artifact
+        downloadable_artifact = download_job.shared_artifact
         exist, _ = downloadable_artifact.archive_exists()
         if exist:
             return jsonify({

wb/main/api_endpoints/v1/profiling.py

@@ -100,7 +100,7 @@ def check_sum_profiling_bundle():
     md5sum = None
     job = CreateProfilingBundleJobModel.query.filter_by(project_id=project_id).first()
     if job:
-        bundle = job.bundle.path
+        bundle = job.shared_artifact
         exists, _ = bundle.archive_exists()
         if exists:
             md5sum = md5(bundle.path)

wb/main/api_endpoints/v1/registry.py

@@ -26,7 +26,8 @@
                               SHORT_TRANSFORMATIONS_CONFIGS, UPLOAD_FOLDER_DATASETS,
                               UPLOAD_FOLDER_MODELS,
                               DISABLE_JUPYTER, ENABLE_AUTH, JUPYTER_NOTEBOOKS_FOLDER, PYTHON_WRAPPER,
-                              PRC_URL_TO_CHECK_CONNECTION, GENERAL_URL_TO_CHECK_CONNECTION)
+                              PRC_URL_TO_CHECK_CONNECTION, GENERAL_URL_TO_CHECK_CONNECTION, CLOUD_SHARED_FOLDER,
+                              ENVIRONMENTS_FOLDER)
 from wb.config.application import get_config
 from wb.error.code_registry import CodeRegistry
 from wb.extensions_factories.database import get_db_for_app
@@ -169,9 +170,10 @@ def clear_assets_paths():
     assets_paths = (
         ARTIFACTS_PATH, MODEL_DOWNLOADS_FOLDER, PROFILING_ARTIFACTS_REPORT_DIR,
         UPLOAD_FOLDER_DATASETS, UPLOAD_FOLDER_MODELS, JUPYTER_NOTEBOOKS_FOLDER,
+        ENVIRONMENTS_FOLDER, CLOUD_SHARED_FOLDER
     )
     for path in assets_paths:
-        if not os.path.exists(path):
+        if not path or not os.path.exists(path):
             continue
         shutil.rmtree(path)
 

wb/main/api_endpoints/v1/remote_job.py

@@ -17,6 +17,7 @@
 
 from flask import jsonify, request
 
+from wb.error.request_error import NotFoundRequestError
 from wb.extensions_factories.database import get_db_session_for_app
 from wb.main.api_endpoints.utils import save_artifact_chunk_upload
 from wb.main.api_endpoints.v1 import V1_REMOTE_JOB_API
@@ -29,6 +30,32 @@
 from wb.main.utils.utils import create_empty_dir, FileSizeConverter
 
 
+@V1_REMOTE_JOB_API.route('/remote-job/finish', methods=['POST'])
+@safe_run
+def set_remote_job_ready():
+    data = request.get_json()
+    wb_pipeline_id = data['wbPipelineId']
+    job_models: List[ParseDevCloudResultJobModel] = (
+        ParseDevCloudResultJobModel.query.filter_by(pipeline_id=wb_pipeline_id).all()
+    )
+    if not job_models:
+        raise NotFoundRequestError(f'No job found for provided pipeline id: {wb_pipeline_id}')
+
+    for job_model in job_models:
+        parent_job: JobsModel = JobsModel.query.get(job_model.parent_job)
+        if not parent_job:
+            raise NotFoundRequestError(f'No job found for provided pipeline id: {wb_pipeline_id}')
+        if parent_job.status == StatusEnum.running:
+            break
+    else:
+        raise NotFoundRequestError(f'No job found for provided pipeline id: {wb_pipeline_id}')
+
+    job_model.are_results_obtained = True
+    job_model.write_record(session=get_db_session_for_app())
+
+    return jsonify({})
+
+
 @V1_REMOTE_JOB_API.route('/remote-job-result/upload', methods=['POST'])
 @safe_run
 def upload_remote_job_result():
@@ -41,16 +68,16 @@ def upload_remote_job_result():
 
     # find a particular job from a big pipeline (int8+profiling) for this artifact
     if not job_models:
-        return no_job_found_response()
+        raise NotFoundRequestError(f'No job found for provided pipeline id: {wb_pipeline_id}')
 
     for job_model in job_models:
         parent_job: JobsModel = JobsModel.query.get(job_model.parent_job)
         if not parent_job:
-            return no_job_found_response()
+            raise NotFoundRequestError(f'No job found for provided pipeline id: {wb_pipeline_id}')
         if parent_job.status == StatusEnum.running:
             break
     else:
-        return no_job_found_response()
+        raise NotFoundRequestError(f'No job found for provided pipeline id: {wb_pipeline_id}')
 
     remote_job_result_artifact: DownloadableArtifactsModel = job_model.result_artifact
     if not remote_job_result_artifact:
@@ -65,15 +92,16 @@ def upload_remote_job_result():
     return jsonify({'artifactItem': remote_job_result_artifact.json(), 'files': files_ids})
 
 
-def no_job_found_response():
-    return 'No job found for provided pipeline id', 404
-
-
 @V1_REMOTE_JOB_API.route('/remote-job-result/upload/<int:file_id>', methods=['POST'])
 @safe_run
 def upload_remote_job_result_chunk(file_id: int):
     file_record = FilesModel.query.get(file_id)
     if not file_record:
-        return 'File record with id {} was not found on the database'.format(file_id), 404
+        raise NotFoundRequestError(f'File record with id {file_id} was not found on the database')
     save_artifact_chunk_upload(request, file_id)
+    artifact = file_record.artifact
+    if artifact.is_all_files_uploaded:
+        job_model = ParseDevCloudResultJobModel.query.filter_by(result_artifact_id=artifact.id).first()
+        job_model.are_results_obtained = True
+        job_model.write_record(session=get_db_session_for_app())
     return jsonify({})

wb/main/console_tool_wrapper/reshape/console_output_parser.py

@@ -15,7 +15,6 @@
 """
 import re
 
-from wb.error.job_error import Int8CalibrationError
 from wb.main.enumerates import StatusEnum
 from wb.main.jobs.interfaces.job_state import JobStateSubject
 from wb.main.jobs.tools_runner.console_output_parser import ConsoleToolOutputParser, skip_empty_line_decorator

wb/main/enumerates.py

@@ -518,6 +518,11 @@ class DevCloudRemoteJobTypeEnum(enum.Enum):
     accuracy = 'accuracy'
 
 
+class DevCloudAPIVersionEnum(enum.Enum):
+    v1 = 'v1'
+    v2 = 'v2'
+
+
 class BenchmarkAppReportTypesEnum(enum.Enum):
     no_counters = 'no_counters'
     average_counters = 'average_counters'

wb/main/jobs/accuracy_analysis/accuracy/create_accuracy_bundle_job.py

@@ -47,7 +47,7 @@ def run(self):
             project = job_model.project
             model_path = project.topology.path
             dataset_path = accuracy_job_model.target_dataset.path
-            bundle_id = job_model.bundle_id
+            bundle_id = job_model.shared_artifact.id
             accuracy_artifacts_path = Path(ACCURACY_ARTIFACTS_FOLDER) / str(job_model.pipeline_id)
 
         configuration_path = accuracy_artifacts_path / JOBS_SCRIPTS_FOLDER_NAME / ACCURACY_CONFIGURATION_FILE_NAME
@@ -72,8 +72,8 @@ def on_success(self):
         with closing(get_db_session_for_celery()) as session:
             job: CreateAccuracyBundleJobModel = self.get_job_model(session)
             # TODO: [61937] Move to separate DBObserver
-            bundle: DownloadableArtifactsModel = job.bundle
-            bundle_path = DownloadableArtifactsModel.get_archive_path(bundle.id)
+            bundle: DownloadableArtifactsModel = job.shared_artifact
+            bundle_path = bundle.build_full_artifact_path()
             bundle.update(bundle_path)
             bundle.write_record(session)
             set_status_in_db(DownloadableArtifactsModel, bundle.id, StatusEnum.ready, session, force=True)

wb/main/jobs/accuracy_analysis/annotate_datset/create_annotate_dataset_bundle_job.py

@@ -55,7 +55,7 @@ def run(self):
             project = job_model.project
             model_path = project.topology.original_model.path
             dataset_path = annotate_dataset_job_model.project.dataset.dataset_data_dir
-            bundle_id = job_model.bundle_id
+            bundle_id = job_model.shared_artifact.id
             annotate_dataset_artifacts_path = Path(DATASET_ANNOTATION_ARTIFACTS_FOLDER) / str(job_model.pipeline_id)
 
         configuration_path = annotate_dataset_artifacts_path / JOBS_SCRIPTS_FOLDER_NAME / DATASET_ANNOTATION_ACCURACY_CONFIGURATION_FILE_NAME
@@ -80,8 +80,8 @@ def on_success(self):
         with closing(get_db_session_for_celery()) as session:
             job: CreateAccuracyBundleJobModel = self.get_job_model(session)
             # TODO: [61937] Move to separate DBObserver
-            bundle: DownloadableArtifactsModel = job.bundle
-            bundle_path = DownloadableArtifactsModel.get_archive_path(bundle.id)
+            bundle: DownloadableArtifactsModel = job.shared_artifact
+            bundle_path = bundle.build_full_artifact_path()
             bundle.update(bundle_path)
             bundle.write_record(session)
             set_status_in_db(DownloadableArtifactsModel, bundle.id, StatusEnum.ready, session, force=True)

wb/main/jobs/accuracy_analysis/per_tensor/create_per_tensor_bundle_job.py

@@ -51,7 +51,7 @@ def run(self):
             optimized_model_path = project.topology.path
             parent_model_path = project.topology.optimized_from_record.path
             dataset_path = per_tensor_report_job_model.project.dataset.dataset_data_dir
-            bundle_id = job_model.bundle_id
+            bundle_id = job_model.shared_artifact.id
             accuracy_artifacts_path = Path(ACCURACY_ARTIFACTS_FOLDER) / str(job_model.pipeline_id)
 
         job_script_path = accuracy_artifacts_path / JOBS_SCRIPTS_FOLDER_NAME / JOB_SCRIPT_NAME
@@ -75,8 +75,8 @@ def on_success(self):
         with closing(get_db_session_for_celery()) as session:
             job: CreateAccuracyBundleJobModel = self.get_job_model(session)
             # TODO: [61937] Move to separate DBObserver
-            bundle: DownloadableArtifactsModel = job.bundle
-            bundle_path = DownloadableArtifactsModel.get_archive_path(bundle.id)
+            bundle: DownloadableArtifactsModel = job.shared_artifact
+            bundle_path = bundle.build_full_artifact_path()
             bundle.update(bundle_path)
             bundle.write_record(session)
             set_status_in_db(DownloadableArtifactsModel, bundle.id, StatusEnum.ready, session, force=True)