-
Notifications
You must be signed in to change notification settings - Fork 26
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement filesystem storage for PFCON in-network
- Loading branch information
1 parent
18b4124
commit f0bf5e3
Showing
8 changed files
with
508 additions
and
24 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| """ | ||
| Handle filesystem-based (eg. mount directory) storage. This is used when pfcon is | ||
| in-network and configured to directly copy the data from a filesystem. | ||
| """ | ||
|
|
||
| import logging | ||
| import datetime | ||
| import os | ||
| import json | ||
| import io | ||
| import shutil | ||
|
|
||
|
|
||
| from .base_storage import BaseStorage | ||
|
|
||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class FileSystemStorage(BaseStorage): | ||
|
|
||
| def __init__(self, config): | ||
|
|
||
| super().__init__(config) | ||
|
|
||
| self.base_dir = config.get('FILESYSTEM_BASEDIR') | ||
|
|
||
|
|
||
| def store_data(self, job_id, job_incoming_dir, data, **kwargs): | ||
| """ | ||
| Copy all the files/folders under each input folder in the specified data list | ||
| into the specified incoming directory. | ||
| """ | ||
| nfiles = 0 | ||
| for rel_path in data: | ||
| abs_path = os.path.join(self.base_dir, rel_path.strip('/')) | ||
|
|
||
| for root, dirs, files in os.walk(abs_path): | ||
| local_path = root.replace(abs_path, job_incoming_dir, 1) | ||
| os.makedirs(local_path, exist_ok=True) | ||
|
|
||
| for filename in files: | ||
| fs_file_path = os.path.join(root, filename) | ||
| try: | ||
| shutil.copy(fs_file_path, local_path) | ||
| except Exception as e: | ||
| logger.error(f'Failed to copy file {fs_file_path} for ' | ||
| f'job {job_id}, detail: {str(e)}') | ||
| raise | ||
| nfiles += 1 | ||
|
|
||
| logger.info(f'{nfiles} files copied from file system for job {job_id}') | ||
| return { | ||
| 'jid': job_id, | ||
| 'nfiles': nfiles, | ||
| 'timestamp': f'{datetime.datetime.now()}', | ||
| 'path': job_incoming_dir | ||
| } | ||
|
|
||
| def get_data(self, job_id, job_outgoing_dir, **kwargs): | ||
| """ | ||
| Copy output files/folders from the specified outgoing directory into the folder | ||
| specified by job_output_path keyword argument (relative to the FS base dir). | ||
| Then create job json file ready for transmission to a remote origin. The json | ||
| file contains the job_output_path prefix and the list of relative file paths. | ||
| """ | ||
| job_output_path = kwargs['job_output_path'] | ||
| fs_output_path = os.path.join(self.base_dir, job_output_path) | ||
| fs_rel_file_paths = [] | ||
|
|
||
| for root, dirs, files in os.walk(job_outgoing_dir): | ||
| rel_path = os.path.relpath(root, job_outgoing_dir) | ||
| if rel_path == '.': | ||
| rel_path = '' | ||
| fs_path = os.path.join(fs_output_path, rel_path) | ||
| os.makedirs(fs_path, exist_ok=True) | ||
|
|
||
| for filename in files: | ||
| local_file_path = os.path.join(root, filename) | ||
| if not os.path.islink(local_file_path): | ||
| try: | ||
| shutil.copy(local_file_path, fs_path) | ||
| except Exception as e: | ||
| logger.error(f'Failed to copy file {local_file_path} for ' | ||
| f'job {job_id}, detail: {str(e)}') | ||
| raise | ||
| fs_rel_file_paths.append(os.path.join(rel_path, filename)) | ||
|
|
||
| data = {'job_output_path': job_output_path, | ||
| 'rel_file_paths': fs_rel_file_paths} | ||
| return io.BytesIO(json.dumps(data).encode()) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| # https://docs.docker.com/compose/yml/ | ||
| # Each service defined in docker-compose.yml must specify exactly one of | ||
| # image or build. Other keys are optional, and are analogous to their | ||
| # docker run command-line counterparts. | ||
| # | ||
| # As with docker run, options specified in the Dockerfile (e.g., CMD, | ||
| # EXPOSE, VOLUME, ENV) are respected by default - you don't need to | ||
| # specify them again in docker-compose.yml. | ||
| # | ||
|
|
||
| version: '3.7' | ||
|
|
||
| services: | ||
| pfcon: | ||
| image: localhost:5000/fnndsc/pfcon:dev | ||
| build: | ||
| context: .. | ||
| args: | ||
| ENVIRONMENT: local | ||
| stdin_open: true # docker run -i | ||
| tty: true # docker run -t | ||
| # We need to mount a physical dir in the HOST onto the key store in pfcon. This dir | ||
| # is given by the STOREBASE env variable substitution. The keystore can be specified | ||
| # by the --storeBase flag during development. | ||
| command: ["python", "-m", "pfcon"] | ||
| environment: | ||
| - APPLICATION_MODE=development | ||
| - PFCON_INNETWORK=true | ||
| - STORAGE_ENV=filesystem | ||
| volumes: | ||
| - fs_storage_dev:/filesystem | ||
| - ${STOREBASE:?}:/var/local/storeBase:z | ||
| - ../pfcon:/app/pfcon:z | ||
| - ../tests:/app/tests:z | ||
| ports: | ||
| - "30006:5005" | ||
| depends_on: | ||
| - pman | ||
| - swift_service | ||
| networks: | ||
| - remote | ||
| labels: | ||
| name: "pfcon" | ||
| role: "pfcon service" | ||
|
|
||
| pman: | ||
| image: ${PMANREPO:?}/pman | ||
| # Since pman spins off containers of its own it needs to mount storeBase dir (where | ||
| # pfcon shares the data) into the spawned container. This directory is passed in the | ||
| # STOREBASE env variable. | ||
| environment: | ||
| - STORAGE_TYPE=host | ||
| - STOREBASE | ||
| - SECRET_KEY="w1kxu^l=@pnsf!5piqz6!!5kdcdpo79y6jebbp+2244yjm*#+k" | ||
| - CONTAINER_ENV=swarm | ||
| volumes: | ||
| - /var/run/docker.sock:/var/run/docker.sock:z | ||
| deploy: | ||
| placement: | ||
| constraints: | ||
| - "node.role==manager" | ||
| networks: | ||
| - remote | ||
| labels: | ||
| name: "pman" | ||
| role: "pman service" | ||
|
|
||
|
|
||
| networks: | ||
| remote: | ||
|
|
||
| volumes: | ||
| fs_storage_dev: |
Oops, something went wrong.