Source code for smartsim.settings.containers

import shutil
from ..log import get_logger

logger = get_logger(__name__)

class Container():
    '''Base class for container types in SmartSim.

    Container types are used to embed all the information needed to
    launch a workload within a container into a single object.

    :param image: local or remote path to container image
    :type image: str
    :param args: arguments to container command
    :type args: str | list[str], optional
    :param mount: paths to mount (bind) from host machine into image.
    :type mount: str | list[str] | dict[str, str], optional
    :param working_directory: path of the working directory within the container
    :type working_directory: str

    def __init__(self, image, args='', mount='', working_directory=''):
        # Validate types
        if not isinstance(image, str):
            raise TypeError('image must be a str')
        elif not isinstance(args, (str, list)):
            raise TypeError('args must be a str | list')
        elif not isinstance(mount, (str, list, dict)):
            raise TypeError('mount must be a str | list | dict')
        elif not isinstance(working_directory, str):
            raise TypeError('working_directory must be a str')

        self.image = image
        self.args = args
        self.mount = mount
        self.working_directory = working_directory

    def _containerized_run_command(self, run_command: str):
        '''Return modified run_command with container commands prepended.

        :param run_command: run command from a RunSettings class
        :type run_command: str
        raise NotImplementedError(f"Containerized run command specification not implemented for this Container type: {type(self)}")

[docs]class Singularity(Container): '''Singularity (apptainer) container type. To be passed into a ``RunSettings`` class initializer or ``Experiment.create_run_settings``. .. note:: Singularity integration is currently tested with `Apptainer 1.0 <>`_ with slurm and PBS workload managers only. Also, note that user-defined bind paths (``mount`` argument) may be disabled by a `system administrator <>`_ :param image: local or remote path to container image, e.g. ``docker://sylabsio/lolcow`` :type image: str :param args: arguments to 'singularity exec' command :type args: str | list[str], optional :param mount: paths to mount (bind) from host machine into image. :type mount: str | list[str] | dict[str, str], optional ''' def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) def _container_cmds(self, default_working_directory=''): '''Return list of container commands to be inserted before exe. Container members are validated during this call. :raises TypeError: if object members are invalid types ''' serialized_args = '' if self.args: # Serialize args into a str if isinstance(self.args, str): serialized_args = self.args elif isinstance(self.args, list): serialized_args = ' '.join(self.args) else: raise TypeError('self.args must be a str | list') serialized_mount = '' if self.mount: if isinstance(self.mount, str): serialized_mount = self.mount elif isinstance(self.mount, list): serialized_mount = ','.join(self.mount) elif isinstance(self.mount, dict): paths = [] for host_path,img_path in self.mount.items(): if img_path: paths.append(f'{host_path}:{img_path}') else: paths.append(host_path) serialized_mount = ','.join(paths) else: raise TypeError('self.mount must be str | list | dict') working_directory = default_working_directory if self.working_directory: working_directory = self.working_directory if not (working_directory in serialized_mount): serialized_mount = ','.join([working_directory, serialized_mount]) logger.warning( f'Working directory not specified in mount: \n {working_directory}'+ '\nAutomatically adding it to the list of bind points' ) # Find full path to singularity singularity = shutil.which('singularity') # Some systems have singularity available on compute nodes only, # so warn instead of error if not singularity: logger.warning('Unable to find singularity. Continuing in case singularity is available on compute node') # Construct containerized launch command cmd_list = [singularity, 'exec'] if working_directory: cmd_list.extend(['--pwd', working_directory]) if serialized_args: cmd_list.append(serialized_args) if serialized_mount: cmd_list.extend(['--bind', serialized_mount]) cmd_list.append(self.image) return cmd_list