Merge pull request #5927 from fstagni/80_fixes21

[8.0] new TornadoDataIntegrityHandler

Author: fstagni

docs/source/AdministratorGuide/ServerInstallations/scalingAndLimitations.rst

@@ -77,7 +77,7 @@ Services
 +--------------------+---------------------------------------------------------------------------------------------------+-------------+---------------------------------------------------------------------------+-----------+
 | Configuration      | :mod:`Configuration <DIRAC.ConfigurationSystem.Service.ConfigurationHandler>`                     | PARTIAL     | One master (rw) and slaves (ro). It's advised to have several CS slaves   | YES       +
 +--------------------+---------------------------------------------------------------------------------------------------+-------------+---------------------------------------------------------------------------+-----------+
-| DataManagement     | :mod:`DataIntegrity <DIRAC.DataManagementSystem.Service.DataIntegrityHandler>`                    | YES         |                                                                           |           +
+| DataManagement     | :mod:`DataIntegrity <DIRAC.DataManagementSystem.Service.DataIntegrityHandler>`                    | YES         |                                                                           | YES       +
 +                    +---------------------------------------------------------------------------------------------------+-------------+---------------------------------------------------------------------------+-----------+
 |                    | :mod:`FileCatalog <DIRAC.DataManagementSystem.Service.FileCatalogHandler>`                        | YES         |                                                                           | YES       +
 +                    +---------------------------------------------------------------------------------------------------+-------------+---------------------------------------------------------------------------+-----------+

src/DIRAC/DataManagementSystem/ConfigTemplate.cfg

@@ -8,6 +8,16 @@ Services
       Default = authenticated
     }
   }
+  ##BEGIN TornadoDataIntegrity
+  TornadoDataIntegrity
+  {
+    Protocol = https
+    Authorization
+    {
+      Default = authenticated
+    }
+  }
+  ##END
   ##BEGIN FTS3Manager
   FTS3Manager
   {
@@ -56,7 +66,7 @@ Services
     }
   }
 
-  # Caution: tech preview, with LHCb specific managers
+  # Caution: LHCb specific managers
   TornadoFileCatalog
   {
     Protocol = https
@@ -76,7 +86,6 @@ Services
       Default = authenticated
     }
   }
-  ######################
 
   ##BEGIN StorageElement
   StorageElement

src/DIRAC/DataManagementSystem/Service/DataIntegrityHandler.py

@@ -8,16 +8,13 @@
   the DISET framework
 
 """
-# imports
-import six
-
 # from DIRAC
 from DIRAC import S_OK
 from DIRAC.Core.DISET.RequestHandler import RequestHandler
 from DIRAC.DataManagementSystem.DB.DataIntegrityDB import DataIntegrityDB
 
 
-class DataIntegrityHandler(RequestHandler):
+class DataIntegrityHandlerMixin:
     """
     .. class:: DataIntegrityHandler
 
@@ -31,7 +28,7 @@ def initializeHandler(cls, serviceInfoDict):
         cls.dataIntegrityDB = DataIntegrityDB()
         return S_OK()
 
-    types_removeProblematic = [list(six.integer_types) + [list]]
+    types_removeProblematic = [[int, list]]
 
     def export_removeProblematic(self, fileID):
         """Remove the file with the supplied FileID from the database"""
@@ -57,7 +54,7 @@ def export_getProblematic(self):
             )
         return res
 
-    types_getPrognosisProblematics = [list(six.string_types)]
+    types_getPrognosisProblematics = [str]
 
     def export_getPrognosisProblematics(self, prognosis):
         """Get problematic files from the problematics table of the IntegrityDB"""
@@ -69,7 +66,7 @@ def export_getPrognosisProblematics(self, prognosis):
             )
         return res
 
-    types_setProblematicStatus = [list(six.integer_types), list(six.string_types)]
+    types_setProblematicStatus = [int, str]
 
     def export_setProblematicStatus(self, fileID, status):
         """Update the status of the problematics with the provided fileID"""
@@ -79,7 +76,7 @@ def export_setProblematicStatus(self, fileID, status):
             self.log.error("DataIntegrityHandler.setProblematicStatus: Failed to set status.", res["Message"])
         return res
 
-    types_incrementProblematicRetry = [list(six.integer_types)]
+    types_incrementProblematicRetry = [int]
 
     def export_incrementProblematicRetry(self, fileID):
         """Update the retry count for supplied file ID."""
@@ -91,7 +88,7 @@ def export_incrementProblematicRetry(self, fileID):
             )
         return res
 
-    types_insertProblematic = [list(six.string_types), dict]
+    types_insertProblematic = [str, dict]
 
     def export_insertProblematic(self, source, fileMetadata):
         """Insert problematic files into the problematics table of the IntegrityDB"""
@@ -111,7 +108,7 @@ def export_changeProblematicPrognosis(self, fileID, newPrognosis):
             self.log.error("DataIntegrityHandler.changeProblematicPrognosis: Failed to update.", res["Message"])
         return res
 
-    types_getTransformationProblematics = [list(six.integer_types)]
+    types_getTransformationProblematics = [int]
 
     def export_getTransformationProblematics(self, transID):
         """Get the problematics for a given transformation"""
@@ -148,3 +145,7 @@ def export_getDistinctPrognosis(self):
         else:
             self.log.error("DataIntegrityHandler.getDistinctPrognosis: Failed to get unique prognosis.", res["Message"])
         return res
+
+
+class DataIntegrityHandler(DataIntegrityHandlerMixin, RequestHandler):
+    pass

src/DIRAC/DataManagementSystem/Service/FTS3ManagerHandler.py

@@ -8,9 +8,6 @@
   :caption: FTS3Manager options
 
 """
-import six
-
-# from DIRAC
 from DIRAC import S_OK, S_ERROR, gLogger
 from DIRAC.ConfigurationSystem.Client.Helpers.Registry import getDNForUsername
 from DIRAC.Core.DISET.RequestHandler import RequestHandler, getServiceOption
@@ -25,7 +22,7 @@
 ########################################################################
 
 
-class FTS3ManagerHandlerMixin(object):
+class FTS3ManagerHandlerMixin:
     """
     .. class:: FTS3ManagerHandlerMixin
 
@@ -86,7 +83,7 @@ def _isAllowed(opObj, remoteCredentials):
 
         return False
 
-    types_persistOperation = [six.string_types]
+    types_persistOperation = [str]
 
     def export_persistOperation(self, opJSON):
         """update or insert request into db
@@ -105,7 +102,7 @@ def export_persistOperation(self, opJSON):
 
         return self.fts3db.persistOperation(opObj)
 
-    types_getOperation = [six.integer_types]
+    types_getOperation = [int]
 
     @classmethod
     def export_getOperation(cls, operationID):
@@ -117,14 +114,14 @@ def export_getOperation(cls, operationID):
         """
         getOperation = cls.fts3db.getOperation(operationID)
         if not getOperation["OK"]:
-            gLogger.error("getOperation: %s" % getOperation["Message"])
+            gLogger.error("getOperation:", getOperation["Message"])
             return getOperation
 
         getOperation = getOperation["Value"]
         opJSON = encode(getOperation)
         return S_OK(opJSON)
 
-    types_getActiveJobs = [six.integer_types, [type(None)] + [six.string_types], six.string_types]
+    types_getActiveJobs = [int, [type(None)] + [str], str]
 
     @classmethod
     def export_getActiveJobs(cls, limit, lastMonitor, jobAssignmentTag):
@@ -168,7 +165,7 @@ def export_updateJobStatus(cls, jobStatusDict):
         jobStatusDict = strToIntDict(jobStatusDict)
         return cls.fts3db.updateJobStatus(jobStatusDict)
 
-    types_getNonFinishedOperations = [six.integer_types, six.string_types]
+    types_getNonFinishedOperations = [int, str]
 
     @classmethod
     def export_getNonFinishedOperations(cls, limit, operationAssignmentTag):
@@ -188,7 +185,7 @@ def export_getNonFinishedOperations(cls, limit, operationAssignmentTag):
 
         return S_OK(nonFinishedOperationsJSON)
 
-    types_getOperationsFromRMSOpID = [six.integer_types]
+    types_getOperationsFromRMSOpID = [int]
 
     @classmethod
     def export_getOperationsFromRMSOpID(cls, rmsOpID):
@@ -200,7 +197,7 @@ def export_getOperationsFromRMSOpID(cls, rmsOpID):
         """
         res = cls.fts3db.getOperationsFromRMSOpID(rmsOpID)
         if not res["OK"]:
-            gLogger.error("getOperationsFromRMSOpID: %s" % res["Message"])
+            gLogger.error("getOperationsFromRMSOpID", res["Message"])
             return res
 
         operations = res["Value"]

src/DIRAC/DataManagementSystem/Service/FileCatalogHandler.py

@@ -13,7 +13,7 @@
 from DIRAC.DataManagementSystem.DB.FileCatalogDB import FileCatalogDB
 
 
-class FileCatalogHandlerMixin(object):
+class FileCatalogHandlerMixin:
     """
     A simple Replica and Metadata Catalog service.
     """

src/DIRAC/DataManagementSystem/Service/S3GatewayHandler.py

@@ -14,7 +14,6 @@
 import errno
 
 # from DIRAC
-import six
 from DIRAC import S_OK, S_ERROR, gLogger
 
 from DIRAC.Core.DISET.RequestHandler import RequestHandler
@@ -93,7 +92,7 @@ def _hasAccess(self, lfn, s3_method):
 
         return returnSingleResult(self._fc.hasAccess(lfn, opType))
 
-    types_createPresignedUrl = [six.string_types, six.string_types, (dict, list), six.integer_types]
+    types_createPresignedUrl = [str, str, (dict, list), int]
 
     def export_createPresignedUrl(self, storageName, s3_method, urls, expiration):
         """Generate a presigned URL for a given object, given method, and given storage

src/DIRAC/DataManagementSystem/Service/StorageElementHandler.py

@@ -28,7 +28,6 @@
 import re
 import errno
 import shlex
-import six
 
 # from DIRAC
 from DIRAC import gLogger, S_OK, S_ERROR
@@ -226,15 +225,15 @@ def __getFileStat(path):
 
         return S_OK(resultDict)
 
-    types_exists = [six.string_types]
+    types_exists = [str]
 
     def export_exists(self, fileID):
         """Check existence of the fileID"""
         if os.path.exists(self.__resolveFileID(fileID)):
             return S_OK(True)
         return S_OK(False)
 
-    types_getMetadata = [six.string_types]
+    types_getMetadata = [str]
 
     def export_getMetadata(self, fileID):
         """Get metadata for the file or directory specified by fileID"""
@@ -260,7 +259,7 @@ def export_getTotalDiskSpace():
         """
         return getTotalDiskSpace()
 
-    types_createDirectory = [six.string_types]
+    types_createDirectory = [str]
 
     def export_createDirectory(self, dir_path):
         """Creates the directory on the storage"""
@@ -282,7 +281,7 @@ def export_createDirectory(self, dir_path):
             gLogger.error("StorageElementHandler.createDirectory: %s" % errStr, repr(x))
             return S_ERROR(errStr)
 
-    types_listDirectory = [six.string_types, six.string_types]
+    types_listDirectory = [str, str]
 
     def export_listDirectory(self, dir_path, mode):
         """Return the dir_path directory listing"""
@@ -419,7 +418,7 @@ def transfer_bulkToClient(self, fileId, token, fileHelper):
             gLogger.error("Failed to send bulk to network", res["Message"])
         return res
 
-    types_remove = [six.string_types, six.string_types]
+    types_remove = [str, str]
 
     def export_remove(self, fileID, token):
         """Remove fileID from the storage. token is used for access rights confirmation."""
@@ -440,7 +439,7 @@ def __removeFile(self, fileID, token):
         else:
             return S_ERROR("File removal %s not authorized" % fileID)
 
-    types_getDirectorySize = [six.string_types]
+    types_getDirectorySize = [str]
 
     def export_getDirectorySize(self, fileID):
         """Get the size occupied by the given directory"""
@@ -455,7 +454,7 @@ def export_getDirectorySize(self, fileID):
         else:
             return S_ERROR("Directory does not exists")
 
-    types_removeDirectory = [six.string_types, six.string_types]
+    types_removeDirectory = [str, str]
 
     def export_removeDirectory(self, fileID, token):
         """Remove the given directory from the storage"""
@@ -475,7 +474,7 @@ def export_removeDirectory(self, fileID, token):
                     gLogger.error(str(error))
                     return S_ERROR("Failed to remove directory %s" % dir_path)
 
-    types_removeFileList = [list, six.string_types]
+    types_removeFileList = [list, str]
 
     def export_removeFileList(self, fileList, token):
         """Remove files in the given list"""

src/DIRAC/DataManagementSystem/Service/TornadoDataIntegrityHandler.py

@@ -0,0 +1,22 @@
+"""
+Service handler for DataIntegrity using https
+
+.. literalinclude:: ../ConfigTemplate.cfg
+  :start-after: ##BEGIN TornadoDataIntegrity
+  :end-before: ##END
+  :dedent: 2
+  :caption: TornadoDataIntegrity options
+
+"""
+
+from DIRAC import gLogger
+from DIRAC.Core.Tornado.Server.TornadoService import TornadoService
+from DIRAC.DataManagementSystem.Service.DataIntegrityHandler import DataIntegrityHandlerMixin
+
+sLog = gLogger.getSubLogger(__name__)
+
+
+class TornadoDataIntegrityHandler(DataIntegrityHandlerMixin, TornadoService):
+    """Tornado handler for the DataIntegrityHandler"""
+
+    log = sLog