sweep: DIRACGrid#6682 Implement a recursive listDirectory for the DFC

Author: chaen

src/DIRAC/DataManagementSystem/Client/DataManager.py

@@ -160,24 +160,22 @@ def __cleanDirectory(self, folder):
             errStr = "Write access not permitted for this credential."
             log.debug(errStr, folder)
             return S_ERROR(errStr)
-        res = self.__getCatalogDirectoryContents([folder], includeDirectories=True)
+
+        res = returnSingleResult(self.fileCatalog.getDirectoryDump([folder]))
+
         if not res["OK"]:
             return res
 
-        if not res["Value"]:
+        if not (res["Value"]["Files"] or res["Value"]["SubDirs"]):
             # folder is empty, just remove it and return
             res = returnSingleResult(self.fileCatalog.removeDirectory(folder, recursive=True))
             return res
 
         # create a list of folders so that empty folders are also deleted
-        areDirs = self.fileCatalog.isDirectory(res["Value"])
-        if not areDirs["OK"]:
-            return areDirs
-        listOfFolders = [aDir for aDir in areDirs["Value"]["Successful"] if areDirs["Value"]["Successful"][aDir]]
-        for lfn in listOfFolders:
-            res["Value"].pop(lfn)
-
-        res = self.removeFile(res["Value"])
+        listOfFolders = res["Value"]["SubDirs"]
+        listOfFiles = res["Value"]["Files"]
+
+        res = self.removeFile(listOfFiles)
         if not res["OK"]:
             return res
         for lfn, reason in res["Value"]["Failed"].items():  # can be an iterator
@@ -238,34 +236,6 @@ def __removeStorageDirectory(self, directory, storageElement):
         )
         return S_OK()
 
-    def __getCatalogDirectoryContents(self, directories, includeDirectories=False):
-        """ls recursively all files in directories
-
-        :param self: self reference
-        :param list directories: folder names
-        :param bool includeDirectories: if True includes directories in the return dictionary
-        :return: S_OK with dict of LFNs and their attribute dictionary
-        """
-        log = self.log.getSubLogger("__getCatalogDirectoryContents")
-        log.debug("Obtaining the catalog contents for %d directories:" % len(directories))
-        activeDirs = directories
-        allFiles = {}
-        while len(activeDirs) > 0:
-            currentDir = activeDirs[0]
-            res = returnSingleResult(self.fileCatalog.listDirectory(currentDir, verbose=True))
-            activeDirs.remove(currentDir)
-
-            if not res["OK"]:
-                log.debug("Problem getting the %s directory content" % currentDir, res["Message"])
-            else:
-                dirContents = res["Value"]
-                activeDirs.extend(dirContents["SubDirs"])
-                allFiles.update(dirContents["Files"])
-                if includeDirectories:
-                    allFiles.update(dirContents["SubDirs"])
-        log.debug("Found %d files" % len(allFiles))
-        return S_OK(allFiles)
-
     def getReplicasFromDirectory(self, directory):
         """get all replicas from a given directory
 
@@ -276,11 +246,16 @@ def getReplicasFromDirectory(self, directory):
             directories = [directory]
         else:
             directories = directory
-        res = self.__getCatalogDirectoryContents(directories)
+        res = returnSingleResult(self.fileCatalog.getDirectoryDump(directories))
         if not res["OK"]:
             return res
-        allReplicas = {lfn: metadata["Replicas"] for lfn, metadata in res["Value"].items()}  # can be an iterator
-        return S_OK(allReplicas)
+
+        lfns = res["Value"]["Files"]
+        res = self.fileCatalog.getReplicas(lfns, allStatus=True)
+        if not res["OK"]:
+            return res
+        res["Value"] = res["Value"]["Successful"]
+        return res
 
     def getFilesFromDirectory(self, directory, days=0, wildcard="*"):
         """get all files from :directory: older than :days: days matching to :wildcard:

src/DIRAC/DataManagementSystem/Client/test/Test_Client_DataManagementSystem.py

@@ -100,83 +100,6 @@ def test__getFileTypesCount(self):
     #   res = self.ci.catalogDirectoryToSE(lfnDir)
     #   self.assertTrue(res['OK'])
 
-    def test__getCatalogDirectoryContents(self):
-        lfnDirs = ["/this/is/dir1/", "/this/is/dir2/"]
-
-        res = self.ci._getCatalogDirectoryContents(lfnDirs)
-        self.assertTrue(res["OK"])
-
-        resExpected = {
-            "Metadata": {
-                "/this/is/dir1/file1.txt": {
-                    "MetaData": {
-                        "Checksum": "7149ed85",
-                        "ChecksumType": "Adler32",
-                        "CreationDate": datetime.datetime(2014, 12, 4, 12, 16, 56),
-                        "FileID": 156301805,
-                        "GID": 2695,
-                        "GUID": "6A5C6C86-AD7B-E411-9EDB",
-                        "Mode": 436,
-                        "ModificationDate": datetime.datetime(2014, 12, 4, 12, 16, 56),
-                        "Owner": "phicharp",
-                        "OwnerGroup": "lhcb_prod",
-                        "Size": 206380531,
-                        "Status": "AprioriGood",
-                        "Type": "File",
-                        "UID": 19503,
-                    }
-                },
-                "/this/is/dir1/file2.foo.bar": {
-                    "MetaData": {
-                        "Checksum": "7149ed86",
-                        "ChecksumType": "Adler32",
-                        "CreationDate": datetime.datetime(2014, 12, 4, 12, 16, 56),
-                        "FileID": 156301805,
-                        "GID": 2695,
-                        "GUID": "6A5C6C86-AD7B-E411-9EDB",
-                        "Mode": 436,
-                        "ModificationDate": datetime.datetime(2014, 12, 4, 12, 16, 56),
-                        "Owner": "phicharp",
-                        "OwnerGroup": "lhcb_prod",
-                        "Size": 206380532,
-                        "Status": "AprioriGood",
-                        "Type": "File",
-                        "UID": 19503,
-                    }
-                },
-                "/this/is/dir2/subdir1/file3.pippo": {
-                    "MetaData": {
-                        "Checksum": "7149ed86",
-                        "ChecksumType": "Adler32",
-                        "CreationDate": datetime.datetime(2014, 12, 4, 12, 16, 56),
-                        "FileID": 156301805,
-                        "GID": 2695,
-                        "GUID": "6A5C6C86-AD7B-E411-9EDB",
-                        "Mode": 436,
-                        "ModificationDate": datetime.datetime(2014, 12, 4, 12, 16, 56),
-                        "Owner": "phicharp",
-                        "OwnerGroup": "lhcb_prod",
-                        "Size": 206380532,
-                        "Status": "AprioriGood",
-                        "Type": "File",
-                        "UID": 19503,
-                    }
-                },
-            },
-            "Replicas": {
-                "/this/is/dir1/file1.txt": {
-                    "SE1": "smr://srm.SE1.ch:8443/srm/v2/server?SFN=/this/is/dir1/file1.txt",
-                    "SE2": "smr://srm.SE2.fr:8443/srm/v2/server?SFN=/this/is/dir1/file1.txt",
-                },
-                "/this/is/dir1/file2.foo.bar": {
-                    "SE1": "smr://srm.SE1.ch:8443/srm/v2/server?SFN=/this/is/dir1/file2.foo.bar",
-                    "SE3": "smr://srm.SE3.es:8443/srm/v2/server?SFN=/this/is/dir1/file2.foo.bar",
-                },
-            },
-        }
-
-        self.assertEqual(res["Value"], resExpected)
-
 
 if __name__ == "__main__":
     suite = unittest.defaultTestLoader.loadTestsFromTestCase(UtilitiesTestCase)

src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/DirectoryManager/DirectoryClosure.py

@@ -7,6 +7,7 @@
                      you do it several times within 1 second, then there will be no changed, and affected = 0
 
 """
+import errno
 import os
 
 from DIRAC import S_OK, S_ERROR
@@ -657,3 +658,37 @@ def _changeDirectoryParameter(self, paths, directoryFunction, _fileFunction, rec
                 successful[path] = True
 
         return S_OK({"Successful": successful, "Failed": failed})
+
+    def _getDirectoryDump(self, path):
+        """Recursively dump all the content of a directory
+
+        :param str path: directory to dump
+
+        :returns: dictionary with `Files` and `SubDirs` as keys
+                    `Files` is a dict containing files metadata.
+                    `SubDirs` is a list of directory
+        """
+
+        result = self.findDir(path)
+        if not result["OK"]:
+            return result
+        dirID = result["Value"]
+        if not dirID:
+            return S_ERROR(errno.ENOENT, f"{path} does not exist")
+
+        result = self.db.executeStoredProcedureWithCursor("ps_get_directory_dump", (dirID,))
+
+        if not result["OK"]:
+            return result
+
+        rows = result["Value"]
+        files = {}
+        subDirs = []
+
+        for lfn, size, creationDate in rows:
+            if size is None:
+                subDirs.append(lfn)
+            else:
+                files[lfn] = {"Size": int(size), "CreationDate": creationDate}
+
+        return S_OK({"Files": files, "SubDirs": subDirs})

src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/DirectoryManager/DirectoryTreeBase.py

@@ -1,4 +1,5 @@
 """ DIRAC DirectoryTree base class """
+import errno
 import time
 import threading
 import os
@@ -676,6 +677,89 @@ def listDirectory(self, lfns, verbose=False):
 
         return S_OK({"Successful": successful, "Failed": failed})
 
+    def getDirectoryDump(self, lfns):
+        """Get the dump of the directories in lfns"""
+        successful = {}
+        failed = {}
+        for path in lfns:
+            result = self._getDirectoryDump(path)
+            if not result["OK"]:
+                failed[path] = result["Message"]
+            else:
+                successful[path] = result["Value"]
+
+        return S_OK({"Successful": successful, "Failed": failed})
+
+    def _getDirectoryDump(self, path):
+        """
+        Recursively dump all the content of a directory
+
+        :param str path: directory to dump
+
+        :returns: dictionary with `Files` and `SubDirs` as keys
+                  `Files` is a dict containing files metadata.
+                  `SubDirs` is a list of directory
+        """
+        result = self.findDir(path)
+        if not result["OK"]:
+            return result
+        directoryID = result["Value"]
+        if not directoryID:
+            return S_ERROR(errno.ENOENT, f"{path} does not exist")
+        directories = []
+
+        result = self.db.fileManager.getFilesInDirectory(directoryID)
+        if not result["OK"]:
+            return result
+
+        filesInDir = result["Value"]
+        files = {
+            os.path.join(path, fileName): {
+                "Size": fileMetadata["MetaData"]["Size"],
+                "CreationDate": fileMetadata["MetaData"]["CreationDate"],
+            }
+            for fileName, fileMetadata in filesInDir.items()
+        }
+
+        dirIDList = [directoryID]
+
+        while dirIDList:
+            curDirID = dirIDList.pop()
+            result = self.getChildren(curDirID)
+            if not result["OK"]:
+                return result
+            newDirIDList = result["Value"]
+            for dirID in newDirIDList:
+                result = self.getDirectoryPath(dirID)
+                if not result["OK"]:
+                    return result
+                dirName = result["Value"]
+
+                directories.append(dirName)
+
+                result = self.db.fileManager.getFilesInDirectory(dirID)
+                if not result["OK"]:
+                    return result
+
+                filesInDir = result["Value"]
+
+                files.update(
+                    {
+                        os.path.join(dirName, fileName): {
+                            "Size": fileMetadata["MetaData"]["Size"],
+                            "CreationDate": fileMetadata["MetaData"]["CreationDate"],
+                        }
+                        for fileName, fileMetadata in filesInDir.items()
+                    }
+                )
+
+            # Add to this list to get subdirectories of these directories
+            dirIDList.extend(newDirIDList)
+
+        pathDict = {"Files": files, "SubDirs": directories}
+
+        return S_OK(pathDict)
+
     def getDirectoryReplicas(self, lfns, allStatus=False):
         """Get replicas for files in the given directories"""
         successful = {}

src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/SecurityManager/SecurityManagerBase.py

@@ -17,6 +17,8 @@
     "getDirectoryReplicas",
     "getDirectorySize",
     "getDirectoryMetadata",
+    "getSEDump",
+    "getDirectoryDump",
 ]
 
 _writeMethods = [

src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/SecurityManager/VOMSSecurityManager.py

@@ -565,6 +565,7 @@ def hasAccess(self, opType, paths, credDict):
             "exists",
             "getFileAncestors",
             "getFileDescendents",
+            "getDirectoryDump",
         ]:
             policyToExecute = self.__policyReadForFileAndDirectory
 

src/DIRAC/DataManagementSystem/DB/FileCatalogDB.py

@@ -913,6 +913,33 @@ def listDirectory(self, lfns, credDict, verbose=False):
         successful = res["Value"]["Successful"]
         return S_OK({"Successful": successful, "Failed": failed})
 
+    def getDirectoryDump(self, lfns, credDict):
+        """
+        Get a dump of the directories
+
+        :param list lfns: list of directories
+        :param creDict: credential
+
+        :return: Successful/Failed dict.
+           The successful values are dictionaries indexed "Files", "Subdirs"
+        """
+
+        res = self._checkPathPermissions("getDirectoryDump", lfns, credDict)
+        if not res["OK"]:
+            return res
+        failed = res["Value"]["Failed"]
+
+        # if no successful, just return
+        if not res["Value"]["Successful"]:
+            return S_OK({"Successful": {}, "Failed": failed})
+
+        res = self.dtree.getDirectoryDump(res["Value"]["Successful"])
+        if not res["OK"]:
+            return res
+        failed.update(res["Value"]["Failed"])
+        successful = res["Value"]["Successful"]
+        return S_OK({"Successful": successful, "Failed": failed})
+
     def isDirectory(self, lfns, credDict):
         """
         Checks whether a list of LFNS are directories or not