Merge pull request #5954 from chaen/cherry-pick-2-16f27a4f6-integration

[sweep:integration] DFC allows to dump more than one SE at the time

Author: fstagni

src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/FileManager/FileManagerBase.py

@@ -1333,12 +1333,12 @@ def setFileMode(self, path, mode):
         """
         return self._setFileParameter(path, "Mode", mode)
 
-    def getSEDump(self, seName):
+    def getSEDump(self, seNames):
         """
          Return all the files at a given SE, together with checksum and size
 
-        :param seName: name of the StorageElement
+        :param seName: list of storageElement names
 
-        :returns: S_OK with list of tuples (lfn, checksum, size)
+        :returns: S_OK with list of tuples (SEName, lfn, checksum, size)
         """
         return S_ERROR("To be implemented on derived class")

src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/FileManager/FileManagerPs.py

@@ -921,18 +921,23 @@ def _getFileLFNs(self, fileIDs):
 
         return S_OK({"Successful": successful, "Failed": failed})
 
-    def getSEDump(self, seName):
+    def getSEDump(self, seNames):
         """
          Return all the files at a given SE, together with checksum and size
 
-        :param seName: name of the StorageElement
+        :param seName: list of StorageElement names
 
-        :returns: S_OK with list of tuples (lfn, checksum, size)
+        :returns: S_OK with list of tuples (SEName, lfn, checksum, size)
         """
 
-        res = self.db.seManager.findSE(seName)
-        if not res["OK"]:
-            return res
-        seID = res["Value"]
+        seIDs = []
+
+        for seName in seNames:
+            res = self.db.seManager.findSE(seName)
+            if not res["OK"]:
+                return res
+            seIDs.append(res["Value"])
+
+        formatedSEIds = intListToString(seIDs)
 
-        return self.db.executeStoredProcedureWithCursor("ps_get_se_dump", (seID,))
+        return self.db.executeStoredProcedureWithCursor("ps_get_se_dump", (formatedSEIds,))

src/DIRAC/DataManagementSystem/DB/FileCatalogDB.py

@@ -1171,12 +1171,12 @@ def _checkPathPermissions(self, operation, lfns, credDict):
                 successful[lfn] = lfns[lfn]
         return S_OK({"Successful": successful, "Failed": failed})
 
-    def getSEDump(self, seName):
+    def getSEDump(self, seNames):
         """
-         Return all the files at a given SE, together with checksum and size
+         Return all the files at given SEs, together with checksum and size
 
-        :param seName: name of the StorageElement
+        :param seName: list of StorageElement names
 
-        :returns: S_OK with list of tuples (lfn, checksum, size)
+        :returns: S_OK with list of tuples (SEName, lfn, checksum, size)
         """
-        return self.fileManager.getSEDump(seName)
+        return self.fileManager.getSEDump(seNames)

src/DIRAC/DataManagementSystem/DB/FileCatalogWithFkAndPsDB.sql

@@ -1941,21 +1941,26 @@ DELIMITER ;
 
 
 
--- ps_get_se_dump : dump all the lfns in an SE, with checksum and size
--- se_id : storageElement's ID
--- output : LFN, Checksum, Size
+-- ps_get_se_dump : dump all the lfns in list of SEs, with checksum and size
+-- se_id : storageElement IDs
+-- output : SEName, LFN, Checksum, Size
 
 DROP PROCEDURE IF EXISTS ps_get_se_dump;
 DELIMITER //
 CREATE PROCEDURE ps_get_se_dump
-(IN se_id INT)
+(IN se_ids TEXT)
 BEGIN
 
-  SELECT SQL_NO_CACHE CONCAT(d.Name, '/', f.FileName), f.Checksum, f.Size
-         FROM FC_Files f
-         JOIN FC_Replicas r on f.FileID = r.FileID
-         JOIN FC_DirectoryList d on d.DirID = f.DirID
-         WHERE SEID = se_id;
+  SET @sql = CONCAT('SELECT SEName, CONCAT(d.Name, "/", f.FileName), f.Checksum, f.Size
+        FROM FC_Files f
+        JOIN FC_Replicas r on f.FileID = r.FileID
+        JOIN FC_DirectoryList d on d.DirID = f.DirID
+        JOIN FC_StorageElements s on r.SEID = s.SEID
+        WHERE s.SEID IN (', se_ids, ')');
+
+  PREPARE stmt FROM @sql;
+  EXECUTE stmt;
+  DEALLOCATE PREPARE stmt;
 
 END //
 DELIMITER ;

src/DIRAC/DataManagementSystem/Service/FileCatalogHandler.py

@@ -4,6 +4,7 @@
 """
 
 import csv
+import json
 import os
 from io import StringIO
 
@@ -695,19 +696,19 @@ def export_getDatasetFiles(self, datasets):
         """Get lfns in the given dataset"""
         return self.fileCatalogDB.datasetManager.getDatasetFiles(datasets, self.getRemoteCredentials())
 
-    def getSEDump(self, seName):
+    def getSEDump(self, seNames):
         """
-         Return all the files at a given SE, together with checksum and size
+         Return all the files at given SEs, together with checksum and size
 
-        :param seName: name of the StorageElement
+        :param seNames: StorageElement names
 
-        :returns: S_OK with list of tuples (lfn, checksum, size)
+        :returns: S_OK with list of tuples (SEName, lfn, checksum, size)
         """
-        return self.fileCatalogDB.getSEDump(seName)["Value"]
+        return self.fileCatalogDB.getSEDump(seNames)
 
 
 class FileCatalogHandler(FileCatalogHandlerMixin, RequestHandler):
-    def transfer_toClient(self, seName, token, fileHelper):
+    def transfer_toClient(self, jsonSENames, token, fileHelper):
         """This method used to transfer the SEDump to the client,
         formated as CSV with '|' separation
 
@@ -718,9 +719,17 @@ def transfer_toClient(self, seName, token, fileHelper):
 
         """
 
-        retVal = self.getSEDump(seName)
+        seNames = json.loads(jsonSENames)
+        csvOutput = None
+        res = self.getSEDump(seNames)
 
         try:
+            if not res["OK"]:
+                ret = fileHelper.stringToNetwork(json.dumps(res))
+                return ret
+
+            retVal = res["Value"]
+
             csvOutput = StringIO()
             writer = csv.writer(csvOutput, delimiter="|")
             for lfn in retVal:
@@ -735,4 +744,5 @@ def transfer_toClient(self, seName, token, fileHelper):
             self.log.exception("Exception while sending seDump", repr(e))
             return S_ERROR("Exception while sending seDump: %s" % repr(e))
         finally:
-            csvOutput.close()
+            if csvOutput is not None:
+                csvOutput.close()

src/DIRAC/DataManagementSystem/Service/TornadoFileCatalogHandler.py

@@ -10,13 +10,15 @@
 
 """
 # imports
+import json
 import csv
 
 from io import StringIO
 
 # from DIRAC
 
 from DIRAC import gLogger, S_ERROR
+from DIRAC.Core.Utilities.ReturnValues import returnValueOrRaise
 from DIRAC.DataManagementSystem.Service.FileCatalogHandler import FileCatalogHandlerMixin
 
 from DIRAC.Core.Tornado.Server.TornadoService import TornadoService
@@ -35,20 +37,22 @@ class TornadoFileCatalogHandler(FileCatalogHandlerMixin, TornadoService):
     # This is needed because the mixin class uses `cls.log`
     log = sLog
 
-    def export_streamToClient(self, seName):
-        """This method used to transfer the SEDump to the client,
+    def export_streamToClient(self, jsonSENames):
+        """This method is used to transfer the SEDump to the client,
         formated as CSV with '|' separation
 
-        :param seName: name of the se to dump
+        :param jsonSENames: json formated names of the SEs to dump
 
         :returns: the result of the FileHelper
 
 
         """
-
-        retVal = self.getSEDump(seName)
+        seNames = json.loads(jsonSENames)
+        csvOutput = None
 
         try:
+            retVal = returnValueOrRaise(self.getSEDump(seNames))
+
             csvOutput = StringIO()
             writer = csv.writer(csvOutput, delimiter="|")
             writer.writerows(retVal)
@@ -60,4 +64,5 @@ def export_streamToClient(self, seName):
             sLog.exception("Exception while sending seDump", repr(e))
             return S_ERROR("Exception while sendind seDump: %s" % repr(e))
         finally:
-            csvOutput.close()
+            if csvOutput is not None:
+                csvOutput.close()

src/DIRAC/Resources/Catalog/FileCatalogClient.py

@@ -1,5 +1,6 @@
 """ The FileCatalogClient is a class representing the client of the DIRAC File Catalog
 """
+import json
 import os
 
 from DIRAC import S_OK, S_ERROR
@@ -45,6 +46,7 @@ class FileCatalogClient(FileCatalogClientBase):
         "getDatasetParameters",
         "getDatasetFiles",
         "getDatasetAnnotation",
+        "getSEDump",
     ]
 
     WRITE_METHODS = [
@@ -655,17 +657,21 @@ def getDatasetFiles(self, datasets, timeout=120):
 
     #############################################################################
 
-    def getSEDump(self, seName, outputFilename):
+    def getSEDump(self, seNames, outputFilename):
         """
-        Dump the content of an SE in the given file.
-        The file contains a list of [lfn,checksum,size] dumped as csv,
+        Dump the content of SEs in the given file.
+        The file contains a list of [SEName, lfn,checksum,size] dumped as csv,
         separated by '|'
 
-        :param seName: name of the StorageElement
+        :param seName: list of StorageElement names
         :param outputFilename: path to the file where to dump it
 
         :returns: result from the TransferClient
         """
+        if isinstance(seNames, str):
+            seNames = seNames.split(",")
 
-        dfc = TransferClient(self.serverURL)
-        return dfc.receiveFile(outputFilename, seName)
+        seNames = json.dumps(seNames)
+
+        dfc = TransferClient(self.serverURL, timeout=3600)
+        return dfc.receiveFile(outputFilename, seNames)

tests/Integration/DataManagementSystem/Test_Client_DFC.py

@@ -339,7 +339,7 @@ def test_fileOperations(self):
 
         with open(expectedDumpFn, "w") as expectedDumpFd:
             csvWriter = csv.writer(expectedDumpFd, delimiter="|")
-            csvWriter.writerow([testFile, "0", 123])
+            csvWriter.writerow(["testSE", testFile, "0", 123])
 
         actualDumpFn = expectedDumpFn + "real"
         result = self.dfc.getSEDump("testSE", actualDumpFn)

tests/Integration/DataManagementSystem/Test_FileCatalogDB.py

@@ -435,10 +435,10 @@ def test_fileOperations(self):
 
         #    ADD SOMETHING ABOUT FILE ANCESTORS AND DESCENDENTS
 
-        result = self.db.getSEDump("testSE")
+        result = self.db.getSEDump(["testSE"])
         self.assertTrue(result["OK"], "Error when getting SE dump %s" % result)
         self.assertEqual(
-            result["Value"], ((testFile, "0", 123),), "Did not get the expected SE Dump %s" % result["Value"]
+            result["Value"], (("testSE", testFile, "0", 123),), "Did not get the expected SE Dump %s" % result["Value"]
         )
 
         result = self.db.removeFile([testFile, nonExistingFile], credDict)