Merge pull request #11995 from vkuznet/fix-issue-19994

Check status of attachDIDs API and act upon it

src/python/Utils/IteratorTools.py

@@ -18,6 +18,18 @@ def grouper(iterable, n):
     return iter(lambda: list(islice(iterable, n)), [])
 
 
+def getChunk(arr, step):
+    """
+    Return chunk of entries from given array and step, it is similar in behavior to grouper
+    function but instead of returning new list it provides a generator iterable object.
+    :param arr: input array of data
+    :param step: step to iterate
+    :return: generator, set of slices with number of entries equal to step of iteration
+    """
+    for i in range(0, len(arr), step):
+        yield arr[i:i + step]
+
+
 def flattenList(doubleList):
     """
     Make flat a list of lists.

src/python/WMCore/MicroService/MSPileup/MSPileupTasks.py

@@ -220,7 +220,10 @@ def partialPileupTask(self):
             newRules = []
             self.logger.info("Attaching %d blocks to custom pileup name: %s", len(customBlocks), cname)
             if len(customBlocks):
-                self.rucioClient.attachDIDs(None, cname, customBlocks, scope=self.customRucioScope)
+                status = self.rucioClient.attachDIDs(None, cname, customBlocks, scope=self.customRucioScope)
+                if not status:
+                    self.logger.error("Failed to attach DIDs to custom container %s with scope %s", cname, self.customRucioScope)
+                    continue
             for rse in doc['expectedRSEs']:
                 # create new rule for custom DID using pileup document rse
                 ruleIds = self.rucioClient.createReplicationRule(cname, rse, scope=self.customRucioScope)
@@ -264,7 +267,7 @@ def partialPileupTask(self):
 
                     # update MSPileup document in MongoDB
                     # here we update our new document in database with document validation
-                    self.mgr.updatePileupDocumentInDatabase(doc, rseList=doc['currentRSEs'])
+                    self.mgr.updatePileupDocumentInDatabase(doc, rseList=doc['expectedRSEs'])
                     self.logger.info("Pileup name %s had its fraction updated in the partialPileupTask function", doc['pileupName'])
                 except Exception as exp:
                     msg = f"Failed to update MSPileup document, {exp}"

src/python/WMCore/Services/Rucio/Rucio.py

@@ -18,6 +18,7 @@
                                     DataIdentifierAlreadyExists, DuplicateContent, InvalidRSEExpression,
                                     UnsupportedOperation, FileAlreadyExists, RuleNotFound, RSENotFound)
 from Utils.MemoryCache import MemoryCache
+from Utils.IteratorTools import grouper
 from WMCore.Services.Rucio.RucioUtils import (validateMetaData, weightedChoice,
                                               isTapeRSE, dropTapeRSEs)
 from WMCore.WMException import WMException
@@ -346,7 +347,7 @@ def createBlock(self, name, scope='cms', attach=True, **kwargs):
             response = self.attachDIDs(kwargs.get('rse'), container, name, scope)
         return response
 
-    def attachDIDs(self, rse, superDID, dids, scope='cms'):
+    def attachDIDs(self, rse, superDID, dids, scope='cms', chunkSize=1000):
         """
         _attachDIDs_
 
@@ -356,28 +357,38 @@ def attachDIDs(self, rse, superDID, dids, scope='cms'):
              then it's a container name; if attaching files, then it's a block name)
         :param dids: either a string or a list of data identifiers (can be block or files)
         :param scope: string with the scope name
+        :param chunkSize: maximum number of dids to be attached in any single Rucio server call
         :return: a boolean to represent whether it succeeded or not
         """
         if not isinstance(dids, list):
             dids = [dids]
         # NOTE: the attaching dids do not create new container within a scope
         # and it is safe to use cms scope for it
-        dids = [{'scope': 'cms', 'name': did} for did in dids]
+        alldids = [{'scope': 'cms', 'name': did} for did in sorted(dids)]
+
+        # report if we use chunk size in rucio attach_dids API call
+        if len(alldids) > chunkSize:
+            self.logger.info("Attaching a total of %d DIDs in chunk size of: %d", len(alldids), chunkSize)
+
+        for dids in grouper(alldids, chunkSize):
+            response = False
+            try:
+                response = self.cli.attach_dids(scope, superDID, dids=dids, rse=rse)
+            except DuplicateContent:
+                self.logger.warning("Dids: %s already attached to: %s", dids, superDID)
+                response = True
+            except FileAlreadyExists:
+                self.logger.warning("Failed to attach files already existent on block: %s", superDID)
+                response = True
+            except DataIdentifierNotFound:
+                self.logger.error("Failed to attach dids: %s. Parent DID %s does not exist.", dids, superDID)
+            except Exception as ex:
+                self.logger.error("Exception attaching %s dids to: %s. Error: %s. First 10 dids: %s",
+                                  len(dids), superDID, str(ex), dids[:10])
+            if not response:
+                # if we had failure with specific chunk of dids we'll return immediately
+                return response
 
-        response = False
-        try:
-            response = self.cli.attach_dids(scope, superDID, dids=dids, rse=rse)
-        except DuplicateContent:
-            self.logger.warning("Dids: %s already attached to: %s", dids, superDID)
-            response = True
-        except FileAlreadyExists:
-            self.logger.warning("Failed to attach files already existent on block: %s", superDID)
-            response = True
-        except DataIdentifierNotFound:
-            self.logger.error("Failed to attach dids: %s. Parent DID %s does not exist.", dids, superDID)
-        except Exception as ex:
-            self.logger.error("Exception attaching dids: %s to: %s. Error: %s",
-                              dids, superDID, str(ex))
         return response
 
     def createReplicas(self, rse, files, block, scope='cms', ignoreAvailability=True):

test/python/Utils_t/IteratorTools_t.py

@@ -7,7 +7,7 @@
 import itertools
 import unittest
 
-from Utils.IteratorTools import grouper, flattenList, makeListElementsUnique
+from Utils.IteratorTools import grouper, flattenList, makeListElementsUnique, getChunk
 
 
 class IteratorToolsTest(unittest.TestCase):
@@ -73,6 +73,13 @@ def testNoDupListOfLists(self):
         data = [(1, 2), (1, 3), (2, 4), (2, 5), (1, 3), (2, 5)]
         self.assertListEqual([(1, 2), (1, 3), (2, 4), (2, 5)], makeListElementsUnique(data))
 
+    def testGetChunk(self):
+        """
+        Test the `getChunk` function.
+        """
+        arr = range(10)
+        for chunk in getChunk(arr, 2):
+            self.assertTrue(len(chunk) == 2)
 
 if __name__ == '__main__':
     unittest.main()