Pass mount limits from master to autoscaled SGE workers
@@ -537,7 +537,7 @@ class GridEngineScaleUpHandler:
def __init__(self, cmd_executor, pipe, grid_engine, host_storage, instance_helper, parent_run_id, default_hostfile, instance_disk,
instance_image, cmd_template, price_type, region_id, polling_timeout=_POLL_TIMEOUT, polling_delay=_POLL_DELAY,
- instance_family=None, shared_fs_type='lfs'):
+ instance_family=None, shared_fs_type='lfs', limit_mounts=None):
"""
Grid engine scale up implementation. It handles additional nodes launching and hosts configuration (/etc/hosts
and self.default_hostfile).
@@ -559,6 +559,7 @@ class GridEngineScaleUpHandler:
:param instance_family: Instance family for launching additional instance,
e.g. c5 means that you can run instances like c5.large, c5.xlarge etc.
:param shared_fs_type: Type of shared fs to initialize
+ :param shared_fs_type: Storage ids to be mounted to workers
"""
self.executor = cmd_executor
self.pipe = pipe
@@ -576,6 +577,7 @@ class GridEngineScaleUpHandler:
self.polling_delay = polling_delay
self.instance_family = instance_family
self.shared_fs_type = shared_fs_type
+ self.limit_mounts = limit_mounts
def scale_up(self, resource):
"""
@@ -628,6 +630,8 @@ class GridEngineScaleUpHandler:
'CP_CAP_SHARE_FS_TYPE %s ' \
% (self.instance_disk, instance, self.instance_image, self.cmd_template, self.parent_run_id,
self._pipe_cli_price_type(self.price_type), self.region_id, self.shared_fs_type)
+ if self.limit_mounts:
+ pipe_run_command = pipe_run_command + ' CP_CAP_LIMIT_MOUNTS "%s"' % self.limit_mounts
run_id = int(self.executor.execute_to_lines(pipe_run_command)[0])
Logger.info('Additional worker run id is %s.' % run_id)
return run_id
@@ -1332,6 +1336,7 @@ if __name__ == '__main__':
instance_family = os.getenv('CP_CAP_AUTOSCALE_HYBRID_FAMILY',
CloudPipelineInstanceHelper.get_family_from_type(cloud_provider, instance_type))
shared_fs_type = os.getenv('CP_CAP_SHARE_FS_TYPE', 'lfs')
+ limit_mounts = os.getenv('CP_CAP_LIMIT_MOUNTS', None)
# TODO: Replace all the usages of PipelineAPI raw client with an actual CloudPipelineAPI client
pipe = PipelineAPI(api_url=pipeline_api, log_dir=os.path.join(shared_work_dir, '.pipe.log'))
@@ -1371,7 +1376,8 @@ if __name__ == '__main__':
price_type=price_type, region_id=region_id,
polling_timeout=scale_up_polling_timeout,
instance_family=instance_family,
- shared_fs_type=shared_fs_type)
+ shared_fs_type=shared_fs_type,
+ limit_mounts=limit_mounts)
scale_down_handler = GridEngineScaleDownHandler(cmd_executor=cmd_executor, grid_engine=grid_engine,
default_hostfile=default_hostfile)
worker_validator = GridEngineWorkerValidator(cmd_executor=cmd_executor, api=api, host_storage=host_storage,
