Merge pull request #5682 from DIRACGridBot/cherry-pick-2-3394339f4-integration

[sweep:integration] sweep Bdii2CSAgent features: bannedCEs, SingleCoreQueues

Author: Andrei Tsaregorodtsev

docs/source/AdministratorGuide/Resources/computingelements.rst

@@ -112,7 +112,10 @@ of CEs are describe in the subsections below
 
 Note that there's no absolute need to define a 1-to-1 relation between CEs and Queues in DIRAC and "in real".
 If for example you want to send, to the same queue, a mix of single processor and multiprocessor Pilots,
-you can define two queues identical but for the NumberOfProcessors parameter.
+you can define two queues identical but for the NumberOfProcessors parameter. To avoid sending single
+processor jobs to multiprocessor queues, add the ``RequiredTag=MultiProcessor`` option to a multiprocessor queue. To
+automatically create the equivalent single core queues, see the :mod:`~DIRAC.ConfigurationSystem.Agent.Bdii2CSAgent`
+configuration.
 
 
 CREAM Computing Element

src/DIRAC/ConfigurationSystem/Agent/Bdii2CSAgent.py

@@ -43,7 +43,7 @@ def __init__(self, *args, **kwargs):
         self.voBdiiCEDict = {}
         self.voBdiiSEDict = {}
         self.host = "cclcgtopbdii01.in2p3.fr:2170"
-
+        self.injectSingleCoreQueues = False
         self.csAPI = None
 
         # What to get
@@ -61,6 +61,7 @@ def initialize(self):
         # Create a list of alternative bdii urls
         self.alternativeBDIIs = self.am_getOption("AlternativeBDIIs", self.alternativeBDIIs)
         self.host = self.am_getOption("Host", self.host)
+        self.injectSingleCoreQueues = self.am_getOption("InjectSingleCoreQueues", self.injectSingleCoreQueues)
 
         # Check if the bdii url is appended by a port number, if not append the default 2170
         for index, url in enumerate(self.alternativeBDIIs):
@@ -232,13 +233,20 @@ def __updateCEs(self):
         """Update the Site/CE/queue settings in the CS if they were changed in the BDII"""
 
         bdiiChangeSet = set()
+        bannedCEs = self.am_getOption("BannedCEs", [])
 
         for vo in self.voName:
             result = self.__getGlue2CEInfo(vo)
             if not result["OK"]:
                 continue
             ceBdiiDict = result["Value"]
-            result = getSiteUpdates(vo, bdiiInfo=ceBdiiDict, log=self.log)
+
+            for _siteName, ceDict in ceBdiiDict.items():
+                for bannedCE in bannedCEs:
+                    ceDict["CEs"].pop(bannedCE, None)
+
+            result = getSiteUpdates(vo, bdiiInfo=ceBdiiDict, log=self.log, onecore=self.injectSingleCoreQueues)
+
             if not result["OK"]:
                 continue
             bdiiChangeSet = bdiiChangeSet.union(result["Value"])

src/DIRAC/ConfigurationSystem/Client/Utilities.py

@@ -10,6 +10,7 @@
 
 __RCSID__ = "$Id$"
 
+from copy import deepcopy
 import socket
 
 from DIRAC import gConfig, gLogger, S_OK, S_ERROR
@@ -103,12 +104,14 @@ def getGridCEs(vo, bdiiInfo=None, ceBlackList=None, hostURL=None):
     return result
 
 
-def getSiteUpdates(vo, bdiiInfo=None, log=None):
+def getSiteUpdates(vo, bdiiInfo=None, log=None, onecore=False):
     """Get all the necessary updates for the already defined sites and CEs
 
     :param str vo: VO name
     :param dict bdiiInfo: information from DBII
     :param object log: logger
+    :param bool onecore: whether to add single core copies of multicore queues, see the documentation about :ref:`CE`
+       and the :mod:`~DIRAC.ConfigurationSystem.Agent.Bdii2CSAgent` configuration for details
 
     :result: S_OK(set)/S_ERROR()
     """
@@ -123,6 +126,24 @@ def addToChangeSet(entry, changeSet):
         if new_value and new_value != value:
             changeSet.add(entry)
 
+    def dropTag(tags, tagToDrop):
+        """Remove tag from a comma-separated string of tags.
+
+        :param str tags: the string of current tags
+        :param str tagToDrop: the tag to potentially remove
+        :return: string of comma separated tags
+        """
+        return ",".join(sorted(set(tags.split(",")).difference({tagToDrop}))).strip(",")
+
+    def addTag(tags, tagToAdd):
+        """Add tag to a comma-separated string of tags.
+
+        :param str tags: the string of current tags
+        :param str tagToAdd: the tag to potentially add
+        :return: string of comma separated tags
+        """
+        return ",".join(sorted(set(tags.split(",")).union({tagToAdd}))).strip(",")
+
     if log is None:
         log = gLogger
 
@@ -133,6 +154,35 @@ def addToChangeSet(entry, changeSet):
             return result
         ceBdiiDict = result["Value"]
 
+    if onecore:
+        # If enabled this creates a copy of the queue with multiple processors and sets NumberOfProcessors to 1 for ARC
+        # and HTCondorCE entries
+
+        def makeNewQueueName(queueName, ceType):
+            """Create a new queueName for single core queues."""
+            if ceType == "HTCondorCE":
+                return queueName + "1core"
+            # we should have only ARC left, we add 1core to the middle part
+            queueNameSplit = queueName.split("-", 2)
+            queueNameSplit[1] = queueNameSplit[1] + "1core"
+            return "-".join(queueNameSplit)
+
+        for siteName, ceDict in ceBdiiDict.items():
+            for _ceName, ceInfo in ceDict["CEs"].items():
+                newQueues = dict()
+                for queueName, queueDict in ceInfo["Queues"].items():
+                    if (
+                        queueDict["GlueCEImplementationName"] not in ("ARC", "HTCondorCE")
+                        or int(queueDict.get("NumberOfProcessors", 1)) == 1
+                    ):
+                        continue
+                    newQueueName = makeNewQueueName(queueName, queueDict["GlueCEImplementationName"])
+                    newQueueDict = deepcopy(queueDict)
+                    newQueueDict["NumberOfProcessors"] = 1
+                    newQueues[newQueueName] = newQueueDict
+
+                ceInfo["Queues"].update(newQueues)
+
     changeSet = set()
     for site in ceBdiiDict:
         result = getDIRACSiteName(site)
@@ -266,6 +316,7 @@ def addToChangeSet(entry, changeSet):
 
                     # tags, processors, localCEType
                     tag = queueDict.get("Tag", "")
+                    reqTag = queueDict.get("RequiredTag", "")
                     # LocalCEType can be empty (equivalent to "InProcess")
                     # or "Pool", "Singularity", but also "Pool/Singularity"
                     localCEType = queueDict.get("LocalCEType", "")
@@ -280,12 +331,19 @@ def addToChangeSet(entry, changeSet):
                     # Adding queue info to the CS
                     addToChangeSet((queueSection, "maxCPUTime", maxCPUTime, newMaxCPUTime), changeSet)
                     addToChangeSet((queueSection, "SI00", si00, newSI00), changeSet)
+
+                    # add RequiredTag if onecore is enabled, do this here for previously created MultiCore queues
+                    if newNOP > 1 and onecore:
+                        addToChangeSet(
+                            (queueSection, "RequiredTag", reqTag, addTag(reqTag, "MultiProcessor")), changeSet
+                        )
+
                     if newNOP != numberOfProcessors:
                         addToChangeSet((queueSection, "NumberOfProcessors", numberOfProcessors, newNOP), changeSet)
                         if newNOP > 1:
                             # if larger than one, add MultiProcessor to site tags, and LocalCEType=Pool
-                            newTag = ",".join(sorted(set(tag.split(",")).union({"MultiProcessor"}))).strip(",")
-                            addToChangeSet((queueSection, "Tag", tag, newTag), changeSet)
+                            addToChangeSet((queueSection, "Tag", tag, addTag(tag, "MultiProcessor")), changeSet)
+
                             if localCEType_inner:
                                 newLocalCEType = "Pool/" + localCEType_inner
                             else:
@@ -294,8 +352,10 @@ def addToChangeSet(entry, changeSet):
                         else:
                             # if not larger than one, drop MultiProcessor Tag.
                             # Here we do not change the LocalCEType as Pool CE would still be perfectly valid.
-                            newTag = ",".join(sorted(set(tag.split(",")).difference({"MultiProcessor"}))).strip(",")
-                            changeSet.add((queueSection, "Tag", tag, newTag))
+                            changeSet.add((queueSection, "Tag", tag, dropTag(tag, "MultiProcessor")))
+                            if onecore:
+                                changeSet.add((queueSection, "RequiredTag", reqTag, dropTag(reqTag, "MultiProcessor")))
+
                     if maxTotalJobs == "Unknown":
                         newTotalJobs = min(1000, int(int(queueInfo.get("GlueCEInfoTotalCPUs", 0)) / 2))
                         newWaitingJobs = max(2, int(newTotalJobs * 0.1))

src/DIRAC/ConfigurationSystem/ConfigTemplate.cfg

@@ -42,6 +42,9 @@ Agents
     DryRun = True
     # Host to query, must include port
     Host = cclcgtopbdii01.in2p3.fr:2170
+    # If True, add single core queues for each Multi Core Queue and set
+    # RequiredTag=MultiProcessor for those
+    InjectSingleCoreQueues = False
   }
   ##END
   ##BEGIN VOMS2CSAgent

src/DIRAC/ConfigurationSystem/scripts/dirac_admin_add_resources.py

@@ -27,18 +27,23 @@
 
 def processScriptSwitches():
 
-    global vo, dry, doCEs, hostURL
+    global vo, dry, doCEs, hostURL, onecore
 
     Script.registerSwitch("V:", "vo=", "Virtual Organization")
     Script.registerSwitch("D", "dry", "Dry run")
     Script.registerSwitch("C", "ce", "Process Computing Elements")
     Script.registerSwitch("H:", "host=", "use this url for information querying")
+    Script.registerSwitch(
+        "", "onecore", "Add Single Core Queues for each MultiCore Queue, set RequiredTag for those Queues"
+    )
     Script.parseCommandLine(ignoreErrors=True)
 
     vo = ""
     dry = False
     doCEs = False
     hostURL = None
+    onecore = False
+
     for sw in Script.getUnprocessedSwitches():
         if sw[0] in ("V", "vo"):
             vo = sw[1]
@@ -48,6 +53,8 @@ def processScriptSwitches():
             doCEs = True
         if sw[0] in ("H", "host"):
             hostURL = sw[1]
+        if sw[0] in ("onecore",):
+            onecore = True
 
 
 ceBdiiDict = None
@@ -100,7 +107,7 @@ def checkUnusedCEs():
 
     inp = input("\nDo you want to add sites ? [default=yes] [yes|no]: ")
     inp = inp.strip()
-    if not inp and inp.lower().startswith("n"):
+    if inp and inp.lower().startswith("n"):
         return
 
     gLogger.notice("\nAdding new sites/CEs interactively\n")
@@ -241,9 +248,9 @@ def updateCS(changeSet):
 
 def updateSites():
 
-    global vo, dry, ceBdiiDict
+    global vo, dry, ceBdiiDict, onecore
 
-    result = getSiteUpdates(vo, bdiiInfo=ceBdiiDict)
+    result = getSiteUpdates(vo, bdiiInfo=ceBdiiDict, onecore=onecore)
     if not result["OK"]:
         gLogger.error("Failed to get site updates", result["Message"])
         DIRACExit(-1)