Coverage for python/lsst/ctrl/bps/parsl/site.py: 45%

1# This file is part of ctrl_bps_parsl. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org) and the LSST DESC (https://www.lsstdesc.org/). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <https://www.gnu.org/licenses/>. 

27 

28from abc import ABC, abstractmethod 

29from types import ModuleType 

30from typing import TYPE_CHECKING 

31 

32import parsl.config 

33from parsl.addresses import address_by_hostname 

34from parsl.executors.base import ParslExecutor 

35from parsl.monitoring import MonitoringHub 

36 

37from lsst.ctrl.bps import BpsConfig 

38from lsst.utils import doImport 

39 

40from .configuration import get_bps_config_value, get_workflow_name 

41from .environment import export_environment 

42 

43if TYPE_CHECKING: 

44 from .job import ParslJob 

45 

46__all__ = ("SiteConfig",) 

47 

48 

49class SiteConfig(ABC): 

50 """Base class for site configuration. 

51 

52 Subclasses need to override at least the ``get_executors`` and 

53 ``select_executor`` methods. 

54 

55 Parameters 

56 ---------- 

57 config : `BpsConfig` 

58 BPS configuration. 

59 add_resources : `bool` 

60 Add resource specification when submitting the job? This is only 

61 appropriate for the ``WorkQueue`` executor; other executors will 

62 raise an exception. 

63 """ 

64 

65 def __init__(self, config: BpsConfig, add_resources: bool = False): 

66 self.config = config 

67 self.site = self.get_site_subconfig(config) 

68 self.add_resources = add_resources 

69 

70 @staticmethod 

71 def get_site_subconfig(config: BpsConfig) -> BpsConfig: 

72 """Get BPS configuration for the site of interest. 

73 

74 We return the BPS sub-configuration for the site indicated by the 

75 ``computeSite`` value, which is ``site.<computeSite>``. 

76 

77 Parameters 

78 ---------- 

79 config : `BpsConfig` 

80 BPS configuration. 

81 

82 Returns 

83 ------- 

84 site : `BpsConfig` 

85 Site sub-configuration. 

86 """ 

87 computeSite = get_bps_config_value(config, "computeSite", str, required=True) 

88 return get_bps_config_value(config, f".site.{computeSite}", BpsConfig, required=True) 

89 

90 @classmethod 

91 def from_config(cls, config: BpsConfig) -> "SiteConfig": 

92 """Get the site configuration nominated in the BPS config. 

93 

94 The ``computeSite`` (`str`) value in the BPS configuration is used to 

95 select a site configuration. The site configuration class to use is 

96 specified by the BPS configuration as ``site.<computeSite>.class`` 

97 (`str`), which should be the fully-qualified name of a python class 

98 that inherits from `SiteConfig`. 

99 

100 Parameters 

101 ---------- 

102 config : `BpsConfig` 

103 BPS configuration. 

104 

105 Returns 

106 ------- 

107 site_config : subclass of `SiteConfig` 

108 Site configuration. 

109 """ 

110 site = cls.get_site_subconfig(config) 

111 name = get_bps_config_value(site, "class", str, required=True) 

112 site_config = doImport(name) 

113 if isinstance(site_config, ModuleType) or not issubclass(site_config, SiteConfig): 

114 raise RuntimeError(f"Site class={name} is not a SiteConfig subclass") 

115 return site_config(config) 

116 

117 @abstractmethod 

118 def get_executors(self) -> list[ParslExecutor]: 

119 """Get a list of executors to be used in processing. 

120 

121 Each executor should have a unique ``label``. 

122 """ 

123 raise NotImplementedError("Subclasses must define") 

124 

125 @abstractmethod 

126 def select_executor(self, job: "ParslJob") -> str: 

127 """Get the ``label`` of the executor to use to execute a job. 

128 

129 Parameters 

130 ---------- 

131 job : `ParslJob` 

132 Job to be executed. 

133 

134 Returns 

135 ------- 

136 label : `str` 

137 Label of executor to use to execute ``job``. 

138 """ 

139 raise NotImplementedError("Subclasses must define") 

140 

141 def get_address(self) -> str: 

142 """Return the IP address of the machine hosting the driver/submission. 

143 

144 This address should be accessible from the workers. This should 

145 generally by the return value of one of the functions in 

146 ``parsl.addresses``. 

147 

148 This is used by the default implementation of ``get_monitor``, but will 

149 generally be used by ``get_executors`` too. 

150 

151 This default implementation gets the address from the hostname, but 

152 that will not work if the workers don't access the driver/submission 

153 node by that address. 

154 """ 

155 return address_by_hostname() 

156 

157 def get_command_prefix(self) -> str: 

158 """Return command(s) to add before each job command. 

159 

160 These may be used to configure the environment for the job. 

161 

162 This default implementation respects the BPS configuration elements: 

163 

164 - ``site.<computeSite>.commandPrefix`` (`str`): command(s) to use as a 

165 prefix to executing a job command on a worker. 

166 - ``site.<computeSite>.environment`` (`bool`): add bash commands that 

167 replicate the environment on the driver/submit machine? 

168 """ 

169 prefix = get_bps_config_value(self.site, "commandPrefix", str, "") 

170 if get_bps_config_value(self.site, "environment", bool, False): 

171 prefix += "\n" + export_environment() 

172 return prefix 

173 

174 def get_monitor(self) -> MonitoringHub | None: 

175 """Get parsl monitor. 

176 

177 The parsl monitor provides a database that tracks the progress of the 

178 workflow and the use of resources on the workers. 

179 

180 This implementation respects the BPS configuration elements: 

181 

182 - ``site.<computeSite>.monitorEnable`` (`bool`): enable monitor? 

183 - ``site.<computeSite>.monitorInterval`` (`float`): time interval (sec) 

184 between logging of resource usage. 

185 - ``site.<computeSite>.monitorFilename`` (`str`): name of file to use 

186 for the monitor sqlite database. 

187 

188 Returns 

189 ------- 

190 monitor : `MonitoringHub` or `None` 

191 Parsl monitor, or `None` for no monitor. 

192 """ 

193 if not get_bps_config_value(self.site, "monitorEnable", bool, False): 

194 return None 

195 return MonitoringHub( 

196 workflow_name=get_workflow_name(self.config), 

197 hub_address=self.get_address(), 

198 resource_monitoring_interval=get_bps_config_value(self.site, "monitorInterval", float, 30.0), 

199 logging_endpoint="sqlite:///" 

200 + get_bps_config_value(self.site, "monitorFilename", str, "monitor.sqlite"), 

201 ) 

202 

203 def get_parsl_config(self) -> parsl.config.Config: 

204 """Get Parsl configuration for this site. 

205 

206 Subclasses can overwrite this method to build a more specific Parsl 

207 configuration, if required. 

208 

209 The retries are set from the ``site.<computeSite>.retries`` value 

210 found in the BPS configuration file. 

211 

212 Returns 

213 ------- 

214 config : `parsl.config.Config` 

215 The configuration to be used for Parsl. 

216 """ 

217 executors = self.get_executors() 

218 monitor = self.get_monitor() 

219 retries = get_bps_config_value(self.site, "retries", int, 1) 

220 # Path to Parsl run directory. The default set by Parsl is 

221 # 'runinfo' which is not explicit enough for end users given that 

222 # we are using BPS + Parsl + Slurm to execute a workflow. 

223 run_dir = get_bps_config_value(self.site, "run_dir", str, "runinfo") 

224 return parsl.config.Config( 

225 executors=executors, 

226 monitoring=monitor, 

227 retries=retries, 

228 run_dir=run_dir, 

229 checkpoint_mode="task_exit", 

230 )