Merge branch 'RESTAPI-902-internal-transfer-constraint' into 'master'

Add constraint in internal transfer jobs

See merge request firecrest/firecrest!241

CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.14.0]
+
+### Added
+
+- Add constraint in xfer-internal job script when provided by the configuration. The associated environment variable is `F7T_XFER_CONSTRAINT` and can be empty, when no machine needs it. Otherwise, the different constraints should be separated by `;`.
 
 ## [1.13.1]
 

deploy/demo/common/common.env

@@ -87,6 +87,8 @@ F7T_STORAGE_MAX_FILE_SIZE=5120
 F7T_OBJECT_STORAGE='s3v4'
 # partition for internal transfer
 F7T_XFER_PARTITION=xfer
+# constraint for internal transfer
+F7T_XFER_CONSTRAINT=''
 # set if account is needed for SLURM job submission
 F7T_USE_SCHED_PROJECT=True
 #-------

deploy/k8s/config/templates/cm.common.yaml

@@ -26,6 +26,7 @@ data:
   F7T_UTILITIES_TIMEOUT: "{{ .Values.common.F7T_UTILITIES_TIMEOUT }}"
   F7T_UTILITIES_URL: "{{ .Values.common.F7T_UTILITIES_URL }}"
   F7T_XFER_PARTITION: "{{ .Values.common.F7T_XFER_PARTITION }}"
+  F7T_XFER_CONSTRAINT: "{{ .Values.common.F7T_XFER_CONSTRAINT }}"
 kind: ConfigMap
 metadata:
   name: common-env-file

deploy/test-build/environment/storage.env

@@ -25,6 +25,7 @@ F7T_OS_PROJECT_ID=
 
 # xfer partition (scheduler partition where internal data movement is done)
 F7T_XFER_PARTITION=xfer
+F7T_XFER_CONSTRAINT=''
 
 # Storage polling interval for uploads in seconds
 F7T_STORAGE_POLLING_INTERVAL=60

src/common/common.env.template

@@ -81,6 +81,7 @@ F7T_STORAGE_MAX_FILE_SIZE=5368709120
 F7T_OBJECT_STORAGE='s3v4'
 # partition for internal transfer
 F7T_XFER_PARTITION=xfer
+F7T_XFER_CONSTRAINT=''
 #-------
 # STATUS: microservices & systems to pool:
 F7T_STATUS_SERVICES='certificator;utilities;compute;tasks;storage'

src/common/schedulers/__init__.py

@@ -23,14 +23,22 @@ class JobScript:
     """
 
     def __init__(
-        self, name, time, partition, command, dependency_id=None, account=None
+        self,
+        name,
+        time,
+        partition,
+        command,
+        dependency_id=None,
+        account=None,
+        constraint=None
     ):
         self.name = name
         self.time = time
         self.partition = partition
         self.command = command
         self.dependency_id = dependency_id
         self.account = account
+        self.constraint = constraint
 
 
 class JobScheduler(abc.ABC):

src/common/schedulers/slurm.py

@@ -80,6 +80,9 @@ def script_template(self, id, script_spec):
             f"#SBATCH --ntasks=1\n"
             f"#SBATCH --partition={script_spec.partition}\n"
         )
+        if script_spec.constraint:
+            script += f"#SBATCH --constraint='{script_spec.constraint}'\n"
+
         if script_spec.dependency_id:
             script += f"#SBATCH --dependency=afterok:{script_spec.dependency_id}\n"
 

src/storage/storage.py

@@ -71,6 +71,9 @@
 # Scheduller partition used for internal transfers, one per public system
 XFER_PARTITION = os.environ.get("F7T_XFER_PARTITION", "").strip('\'"').split(";")
 
+# Scheduller constraint used for internal transfers, one per public system
+XFER_CONSTRAINT = os.environ.get("F7T_XFER_CONSTRAINT", "").strip('\'"').split(";")
+
 # use project account in submission
 # F7T_USE_SLURM_ACCOUNT is deprecated so we use it only in case F7T_USE_SCHED_PROJECT is not set
 USE_SCHED_PROJECT = get_boolean_var(
@@ -435,13 +438,13 @@ def os_to_fs(task_id):
         cert_list = [f"{td}/user-key-cert.pub", f"{td}/user-key.pub", f"{td}/user-key", td]
 
         # start download from OS to FS
-        update_task(task_id, headers, async_task.ST_DWN_BEG, 
+        update_task(task_id, headers, async_task.ST_DWN_BEG,
                     msg={
-                        "source": objectname, 
-                        "target": upl_file["target"], 
+                        "source": objectname,
+                        "target": upl_file["target"],
                         "system_name": system_name,
                         "system_addr": system_addr
-                        }, 
+                        },
                     is_json=True)
 
         # execute download
@@ -451,11 +454,11 @@ def os_to_fs(task_id):
         if result["error"] == 0:
             update_task(task_id, headers, async_task.ST_DWN_END,
                         msg={
-                            "source": objectname, 
-                            "target": upl_file["target"], 
+                            "source": objectname,
+                            "target": upl_file["target"],
                             "system_name": system_name,
                             "system_addr": system_addr
-                            }, 
+                            },
                         is_json=True)
 
             # No need to delete the dictionary, it will be cleaned on next iteration
@@ -505,7 +508,7 @@ def download_task(headers, system_name, system_addr, sourcePath, task_id):
         app.logger.error(f"Couldn't extract username from JWT token: {is_username_ok['reason']}")
         msg = {"error": is_username_ok['reason'], "source": sourcePath, "system_name": system_name}
         update_task(task_id, headers, async_task.ERROR, msg, is_json=True)
-        
+
     container_name = is_username_ok["username"]
 
     if not staging.is_container_created(container_name):
@@ -534,7 +537,7 @@ def download_task(headers, system_name, system_addr, sourcePath, task_id):
     # if upload to SWIFT fails:
     if res["error"] != 0:
         error_msg = f"Upload to Staging area has failed. Object: {object_name}"
-        
+
 
         error_str = res["msg"]
         if in_str(error_str,"OPENSSH"):
@@ -791,7 +794,7 @@ def upload_request():
         system_name = request.headers["X-Machine-Name"]
     except KeyError as e:
         return jsonify(description="No machine name given"), 400
-    
+
     # public endpoints from Kong to users
     if system_name not in SYSTEMS_PUBLIC:
         header = {"X-Machine-Does-Not-Exists": "Machine does not exists"}
@@ -810,14 +813,14 @@ def upload_request():
     v = validate_input(sourcePath)
     if v != "":
         return jsonify(description="Failed to upload file", error=f"'sourcePath' {v}"), 400
-        
+
     [headers, ID] = get_tracing_headers(request)
     # checks if the targetPath is a directory (targetPath = "/path/to/dir") that can be accessed by the user
     check_is_dir = is_valid_dir(targetPath, headers, system_name, system_addr)
-    
+
     # if so, then the actual targetPath has to include the name extracted from sourcePath
     _targetPath = f"{targetPath}/{sourcePath.split('/')[-1]}"
-    
+
     if not check_is_dir["result"]:
 
         # check if targetPath is a file path by extracting last part of the path
@@ -828,7 +831,7 @@ def upload_request():
             return jsonify(description="Failed to upload file", error="'targetPath' directory not allowed"), 400, check_is_dir["headers"]
         # if targetPath is a file, then no changes need to be done
         _targetPath = targetPath
-    
+
     # obtain new task from Tasks microservice
     task_id = create_task(headers, service="storage", system=system_name, init_data={"source":sourcePath, "target": _targetPath})
 
@@ -873,13 +876,19 @@ def exec_internal_command(headers, system_idx, command, jobName, jobTime, stageO
     try:
         td = tempfile.mkdtemp(prefix="job")
         ID = headers.get(TRACER_HEADER, '')
+        if XFER_CONSTRAINT == []:
+            constraint = None
+        else:
+            constraint = XFER_CONSTRAINT[system_idx]
+
         script_spec = JobScript(
             name=jobName,
             time=jobTime,
             partition=XFER_PARTITION[system_idx],
             command=command,
             dependency_id=stageOutJobId,
-            account=account
+            account=account,
+            constraint=constraint
         )
 
         with open(td + "/sbatch-job.sh", "w") as sbatch_file: