-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #36 from lsst/tickets/DM-30965
Initial implementation of lazy commands creation in PanDA plugin
- Loading branch information
Showing
6 changed files
with
253 additions
and
115 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,126 @@ | ||
| # This file is part of ctrl_bps. | ||
| # | ||
| # Developed for the LSST Data Management System. | ||
| # This product includes software developed by the LSST Project | ||
| # (https://www.lsst.org). | ||
| # See the COPYRIGHT file at the top-level directory of this distribution | ||
| # for details of code ownership. | ||
| # | ||
| # This program is free software: you can redistribute it and/or modify | ||
| # it under the terms of the GNU General Public License as published by | ||
| # the Free Software Foundation, either version 3 of the License, or | ||
| # (at your option) any later version. | ||
| # | ||
| # This program is distributed in the hope that it will be useful, | ||
| # but WITHOUT ANY WARRANTY; without even the implied warranty of | ||
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | ||
| # GNU General Public License for more details. | ||
| # | ||
| # You should have received a copy of the GNU General Public License | ||
| # along with this program. If not, see <https://www.gnu.org/licenses/>. | ||
|
|
||
| import os | ||
| import logging | ||
|
|
||
|
|
||
| _LOG = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class CommandLineEmbedder: | ||
| """ | ||
| Class embeds static (constant across a task) values | ||
| into the pipeline execution command line | ||
| and resolves submission side environment variables | ||
| Parameters | ||
| ---------- | ||
| config : `~lsst.ctrl.bps.BPSConfig` | ||
| BPS configuration that includes the list of dynamic | ||
| (uniques per job) and submission side resolved variables | ||
| """ | ||
| def __init__(self, config): | ||
| self.leave_placeholder_params = config.get("placeholderParams", ['qgraphNodeId', 'qgraphId']) | ||
| self.submit_side_resolved = config.get("submitSideResolvedParams", ['USER']) | ||
|
|
||
| def replace_static_parameters(self, cmd_line, lazy_vars): | ||
| """ Substitutes the lazy parameters in the command line which | ||
| are static, the same for every job in the workflow and could be | ||
| defined once. This function offloads the edge node processing | ||
| and number of parameters transferred together with job | ||
| Parameters | ||
| ---------- | ||
| cmd_line: `str` command line to be processed | ||
| lazy_vars: `dict` of lazy variables and its values | ||
| Returns | ||
| ------- | ||
| processed command line | ||
| """ | ||
| for param_name, param_val in lazy_vars.items(): | ||
| if param_name not in self.leave_placeholder_params: | ||
| cmd_line = cmd_line.replace('{'+param_name+'}', param_val) | ||
| return cmd_line | ||
|
|
||
| def resolve_submission_side_env_vars(self, cmd_line): | ||
| """ Substitutes the lazy parameters in the command line | ||
| which are defined and resolved on the submission side | ||
| Parameters | ||
| ---------- | ||
| cmd_line: `str` command line to be processed | ||
| Returns | ||
| ------- | ||
| processed command line | ||
| """ | ||
|
|
||
| for param in self.submit_side_resolved: | ||
| if os.getenv(param): | ||
| cmd_line = cmd_line.replace('<ENV:'+param+'>', os.getenv(param)) | ||
| else: | ||
| _LOG.info("Expected parameter {0} is not found in the environment variables".format(param)) | ||
| return cmd_line | ||
|
|
||
| def attach_pseudo_file_params(self, lazy_vars): | ||
| """ Adds the parameters needed to finalize creation of a pseudo file | ||
| Parameters | ||
| ---------- | ||
| lazy_vars: `dict` of values of to be substituted | ||
| Returns | ||
| ------- | ||
| pseudo input file name suffix | ||
| """ | ||
|
|
||
| file_suffix = "" | ||
| for item in self.leave_placeholder_params: | ||
| file_suffix += '+' + item + ':' + lazy_vars.get(item, '') | ||
| return file_suffix | ||
|
|
||
| def substitute_command_line(self, bps_file_name, cmd_line, lazy_vars, job_name): | ||
| """ Preprocesses the command line leaving for the egde node evaluation | ||
| only parameters which are job / environment dependent | ||
| Parameters | ||
| ---------- | ||
| bps_file_name: `str` input file name proposed by BPS | ||
| cmd_line: `str` command line containing all lazy placeholders | ||
| lazy_vars: `dict` of lazy parameter name/values | ||
| job_name: `str` job name proposed by BPS | ||
| Returns | ||
| ------- | ||
| cmd_line: `str` | ||
| processed command line | ||
| file_name: `str` | ||
| job pseudo input file name | ||
| """ | ||
|
|
||
| cmd_line = self.replace_static_parameters(cmd_line, lazy_vars) | ||
| cmd_line = self.resolve_submission_side_env_vars(cmd_line) | ||
| file_name = bps_file_name + "+" + job_name+self.attach_pseudo_file_params(lazy_vars) | ||
| return cmd_line, file_name |
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
41 changes: 31 additions & 10 deletions
41
python/lsst/ctrl/bps/wms/panda/edgenode/cmd_line_decoder.py
100755 → 100644
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 |
|---|---|---|
| @@ -1,16 +1,37 @@ | ||
| #!/usr/bin/python | ||
| """ | ||
| This file is needed to decode the command line string sent from the BPS | ||
| plugin -> PanDA -> Edge node cluster management -> Edge node -> Container. | ||
| Once the command line hexified it can contain symbols often stripped such as | ||
| ',",` and multiline commands. | ||
| plugin -> PanDA -> Edge node cluster management | ||
| -> Edge node -> Container. This file is not a part | ||
| of the BPS but a part of the payload wrapper. | ||
| It decodes the hexified command line. | ||
| """ | ||
| import os | ||
| import re | ||
| import sys | ||
| import binascii | ||
| cmdline = str(binascii.unhexlify(sys.argv[1]).decode()) | ||
| dataparams = sys.argv[2].split("+") | ||
| cmdline = cmdline.replace("${filename}", dataparams[0]) | ||
| if len(dataparams) > 2: | ||
| cmdline = cmdline.replace("${qgraph-id}", dataparams[2]) | ||
| cmdline = cmdline.replace("${qgraph-node-id}", dataparams[3]) | ||
| print(cmdline) | ||
|
|
||
|
|
||
| def replace_environment_vars(cmdline): | ||
| """ | ||
| Replaces placeholders to the actual environment variables. | ||
| :param cmdline: `str` | ||
| Command line | ||
| :return: cmdline: `str` | ||
| Processed command line | ||
| """ | ||
| envs_to_replace = re.findall(r'<ENV:(.*?)>', cmdline) | ||
| for env_var in envs_to_replace: | ||
| if value := os.getenv(env_var): | ||
| cmdline = cmdline.replace('<ENV:' + env_var + '>', value) | ||
| return cmdline | ||
|
|
||
|
|
||
| cmd_line = str(binascii.unhexlify(sys.argv[1]).decode()) | ||
| data_params = sys.argv[2].split("+") | ||
| cmd_line = replace_environment_vars(cmd_line) | ||
| cmd_line = cmd_line.replace("<FILE:runQgraphFile>", data_params[0]) | ||
| for key_value_pair in data_params[1:]: | ||
| (key, value) = key_value_pair.split(":") | ||
| cmd_line = cmd_line.replace("{" + key + "}", value) | ||
| print(cmd_line) |
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 |
|---|---|---|
| @@ -1,11 +1,13 @@ | ||
| #!/bin/bash | ||
| # This is a starter script needed to initialize basic Rubin software environment inside the container and execute | ||
| # the actual command line after decoding from hexed strings | ||
| cd /tmp; | ||
| ls /; | ||
| export HOME=/tmp; | ||
| export HOME=${PILOT_HOME} | ||
| cd ${HOME} | ||
| export S3_ENDPOINT_URL=https://storage.googleapis.com | ||
| . /opt/lsst/software/stack/loadLSST.bash; | ||
| DIR="$(dirname "${BASH_SOURCE[0]}")" # Get the directory name | ||
| DIR="$(realpath "${DIR}")" # Resolve its full path | ||
| unset PYTHONPATH | ||
| . ${DIR}/../../../../../../../../../../../../loadLSST.bash; | ||
| setup lsst_distrib; | ||
| echo "$(/usr/bin/python $3/cmd_line_decoder.py $1 $2)" | ||
| eval "$(/usr/bin/python $3/cmd_line_decoder.py $1 $2) >&2" | ||
| echo "$(python3 $CTRL_BPS_DIR/python/lsst/ctrl/bps/wms/panda/edgenode/cmd_line_decoder.py $1 $2)" | ||
| eval "$(python3 $CTRL_BPS_DIR/python/lsst/ctrl/bps/wms/panda/edgenode/cmd_line_decoder.py $1 $2) >&2" |
Oops, something went wrong.