feat: parallelize download (#169)

* feat: starting parallel job manager

* fix: get parallel process working for download

* feat: update testcases

* chore: update version

* chore: update readme

* feat: add configurable parallel count

Author: wphyojpl

CHANGELOG.md

@@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [5.2.0] - 2023-07-05
+### Added
+- [#169](https://github.com/unity-sds/unity-data-services/pull/169) feat: parallelize download
+
 ## [5.1.0] - 2023-06-08
 ### Added
 - [#156](https://github.com/unity-sds/unity-data-services/pull/156) feat: added filter keyword in granules endpoint + repeatedly checking with time boundary for cataloging result

cumulus_lambda_functions/lib/processing_jobs/__init__.py

@@ -0,0 +1,11 @@
+from abc import ABCMeta, abstractmethod
+
+
+class JobExecutorAbstract(metaclass=ABCMeta):
+    @abstractmethod
+    def validate_job(self, job_obj):
+        return
+
+    @abstractmethod
+    def execute_job(self, job_obj, lock) -> bool:
+        return

cumulus_lambda_functions/lib/processing_jobs/job_executor_abstract.py

@@ -0,0 +1,122 @@
+from abc import ABCMeta, abstractmethod
+
+from cumulus_lambda_functions.lib.utils.fake_lock import FakeLock
+
+
+class JobManagerProps:
+    def __init__(self):
+        self.__job_bucket = None
+        self.__job_path = None
+        self.__job_file_postfix = ''
+        self.__processing_job_path = None
+        self.__lock = FakeLock()
+        self.__memory_job_dict = {}
+
+    @property
+    def memory_job_dict(self):
+        return self.__memory_job_dict
+
+    @memory_job_dict.setter
+    def memory_job_dict(self, val):
+        """
+        :param val:
+        :return: None
+        """
+        self.__memory_job_dict = val
+        return
+
+    def load_from_json(self, input_json: dict):
+        if 'memory_job_dict' in input_json:
+            self.memory_job_dict = input_json['memory_job_dict']
+        if 'job_bucket' in input_json:
+            self.job_bucket = input_json['job_bucket']
+        if 'job_path' in input_json:
+            self.job_path = input_json['job_path']
+        if 'processing_job_path' in input_json:
+            self.processing_job_path = input_json['processing_job_path']
+        if 'job_file_postfix' in input_json:
+            self.job_file_postfix = input_json['job_file_postfix']
+        return self
+
+    @property
+    def job_file_postfix(self):
+        return self.__job_file_postfix
+
+    @job_file_postfix.setter
+    def job_file_postfix(self, val):
+        """
+        :param val:
+        :return: None
+        """
+        self.__job_file_postfix = val
+        return
+
+    @property
+    def lock(self):
+        return self.__lock
+
+    @lock.setter
+    def lock(self, val):
+        """
+        :param val:
+        :return: None
+        """
+        self.__lock = val
+        return
+
+    @property
+    def job_bucket(self):
+        return self.__job_bucket
+
+    @job_bucket.setter
+    def job_bucket(self, val):
+        """
+        :param val:
+        :return: None
+        """
+        self.__job_bucket = val
+        return
+
+    @property
+    def job_path(self):
+        return self.__job_path
+
+    @job_path.setter
+    def job_path(self, val):
+        """
+        :param val:
+        :return: None
+        """
+        self.__job_path = val
+        return
+
+    @property
+    def processing_job_path(self):
+        return self.__processing_job_path
+
+    @processing_job_path.setter
+    def processing_job_path(self, val):
+        """
+        :param val:
+        :return: None
+        """
+        self.__processing_job_path = val
+        return
+
+
+class JobManagerAbstract(metaclass=ABCMeta):
+    @abstractmethod
+    def get_all_job_files(self):
+        return
+
+    @abstractmethod
+    def get_job_file(self, job_path, validate_job_content=lambda x: True):
+        return
+
+    @abstractmethod
+    def remove_from_processing(self, job_path):
+        return
+
+    @abstractmethod
+    def put_back_failed_job(self, original_job_path: str):
+        return

cumulus_lambda_functions/lib/processing_jobs/job_manager_abstract.py

@@ -0,0 +1,18 @@
+
+from cumulus_lambda_functions.lib.processing_jobs.job_manager_abstract import JobManagerProps
+from cumulus_lambda_functions.lib.processing_jobs.job_manager_local_filesystem import JobManagerLocalFileSystem
+from cumulus_lambda_functions.lib.processing_jobs.job_manager_memory import JobManagerMemory
+from cumulus_lambda_functions.lib.utils.factory_abstract import FactoryAbstract
+
+
+class JobManagerFactory(FactoryAbstract):
+    MEMORY = 'MEMORY'
+    LOCAL = 'LOCAL'
+    def get_instance(self, class_type, **kwargs):
+        props = JobManagerProps().load_from_json(kwargs)
+        fr = class_type.upper()
+        if fr == self.LOCAL:
+            return JobManagerLocalFileSystem(props)
+        if fr == self.MEMORY:
+            return JobManagerMemory(props)
+        raise ModuleNotFoundError(f'cannot find JobManagerFactory class for {class_type}')

cumulus_lambda_functions/lib/processing_jobs/job_manager_factory.py

@@ -0,0 +1,52 @@
+import logging
+import os
+from glob import glob
+
+from cumulus_lambda_functions.lib.processing_jobs.job_manager_abstract import JobManagerAbstract, JobManagerProps
+from cumulus_lambda_functions.lib.utils.file_utils import FileUtils
+
+LOGGER = logging.getLogger(__name__)
+
+
+class JobManagerLocalFileSystem(JobManagerAbstract):
+    def __init__(self, props=JobManagerProps()) -> None:
+        super().__init__()
+        self.__props = props
+
+    def get_all_job_files(self):
+        return [k.replace(self.__props.job_bucket, '')[1:] for k in glob(os.path.join(self.__props.job_bucket, self.__props.job_path, f'*{self.__props.job_file_postfix}'))]
+
+    def get_job_file(self, job_path, validate_job_content=lambda x: True):
+        job_abs_path = os.path.join(self.__props.job_bucket, job_path)
+        with self.__props.lock:
+            try:
+                job_content = FileUtils.read_json(job_abs_path)
+            except:
+                LOGGER.exception(f'job in this path is not in JSON: {job_path}')
+                return None
+            FileUtils.remove_if_exists(job_abs_path)
+        if validate_job_content(job_content) is False:
+            LOGGER.error(f'{job_path} does not have valid json: {job_content}. Putting back')
+            FileUtils.write_json(job_abs_path, job_content, overwrite=True)
+            return None
+        LOGGER.debug(f'writing job to processing path')
+        FileUtils.mk_dir_p(os.path.join(self.__props.job_bucket, self.__props.processing_job_path))
+        FileUtils.write_json(os.path.join(self.__props.job_bucket, self.__props.processing_job_path, os.path.basename(job_path)), job_content, overwrite=True)
+        return job_content
+
+    def put_back_failed_job(self, original_job_path: str):
+        processing_job_path = os.path.join(self.__props.processing_job_path, os.path.basename(original_job_path))
+        LOGGER.debug(f'reading failed job file: {processing_job_path}')
+        try:
+            job_json: dict = FileUtils.read_json(os.path.join(self.__props.job_bucket, processing_job_path))
+            LOGGER.debug(f'deleting failed job file from processing dir: {processing_job_path}')
+            FileUtils.remove_if_exists(os.path.join(self.__props.job_bucket, processing_job_path))
+            LOGGER.debug(f'uploading failed job back to folder')
+            FileUtils.write_json(os.path.join(self.__props.job_bucket, original_job_path), job_json)
+        except Exception as e:
+            LOGGER.exception(f'error while reading failed job file from from processing dir: {processing_job_path}')
+        return
+
+    def remove_from_processing(self, job_path):
+        FileUtils.remove_if_exists(os.path.join(self.__props.job_bucket, self.__props.processing_job_path, os.path.basename(job_path)))
+        return True

cumulus_lambda_functions/lib/processing_jobs/job_manager_local_filesystem.py

@@ -0,0 +1,33 @@
+from cumulus_lambda_functions.lib.processing_jobs.job_manager_abstract import JobManagerAbstract, JobManagerProps
+
+
+class JobManagerMemory(JobManagerAbstract):
+    def __init__(self, props=JobManagerProps()) -> None:
+        self.__props = props
+        self.__processing_job_dict = {}
+
+    def get_all_job_files(self):
+        return [k for k in self.__props.memory_job_dict.keys()]
+
+    def get_job_file(self, job_path, validate_job_content=lambda x: True):
+        if job_path not in self.__props.memory_job_dict:
+            return None
+        job = self.__props.memory_job_dict.get(job_path)
+        del self.__props.memory_job_dict[job_path]
+        if not validate_job_content(job):
+            return None
+        self.__processing_job_dict[job_path] = job
+        return job
+
+    def remove_from_processing(self, job_path):
+        if job_path not in self.__processing_job_dict:
+            return
+        del self.__processing_job_dict[job_path]
+        pass
+
+    def put_back_failed_job(self, original_job_path: str):
+        if original_job_path not in self.__processing_job_dict:
+            return
+        self.__props.memory_job_dict[original_job_path] = self.__processing_job_dict[original_job_path]
+        del self.__processing_job_dict[original_job_path]
+        return