refactor: adapt ResourceStatus/SiteStatus consumers to binary model

src/DIRAC/DataManagementSystem/scripts/dirac_admin_allow_se.py

@@ -61,6 +61,7 @@ def main():
     from DIRAC.DataManagementSystem.Utilities.DMSHelpers import resolveSEGroup
     from DIRAC.Interfaces.API.DiracAdmin import DiracAdmin
     from DIRAC.ResourceStatusSystem.Client.ResourceStatus import ResourceStatus
+    from DIRAC.ResourceStatusSystem.Client.ResourceStatusClient import ResourceStatusClient
 
     if not (read or write or check or remove):
         # No switch was specified, means we need all of them
@@ -98,7 +99,6 @@ def main():
         DIRAC.exit()
 
     STATUS_TYPES = ["ReadAccess", "WriteAccess", "CheckAccess", "RemoveAccess"]
-    ALLOWED_STATUSES = ["Unknown", "InActive", "Banned", "Probing", "Degraded", "Error"]
 
     statusAllowedDict = {}
     for statusType in STATUS_TYPES:
@@ -112,37 +112,31 @@ def main():
 
     resourceStatus = ResourceStatus()
 
-    res = resourceStatus.getElementStatus(ses, "StorageElement")
-    if not res["OK"]:
-        gLogger.error(f"Storage Element {ses} does not exist")
+    resDB = ResourceStatusClient().selectStatusElement(
+        "Resource", "Status", ses, elementType="StorageElement", meta={"columns": ["Name", "StatusType", "Status"]}
+    )
+    if not resDB["OK"]:
+        gLogger.error("Failed to get the status of the storage elements", resDB["Message"])
         DIRAC.exit(-1)
+    dbStatus = {}
+    for name, statusType, status in resDB["Value"]:
+        dbStatus.setdefault(name, {})[statusType] = status
 
     reason = f"Forced with dirac-admin-allow-se by {userName}"
 
-    for se, seOptions in res["Value"].items():
-        # InActive is used on the CS model, Banned is the equivalent in RSS
-        for statusType in STATUS_TYPES:
-            if statusFlagDict[statusType]:
-                if seOptions.get(statusType) == "Active":
-                    gLogger.notice(f"{statusType} status of {se} is already Active")
-                    continue
-                if statusType in seOptions:
-                    if not seOptions[statusType] in ALLOWED_STATUSES:
-                        gLogger.notice(
-                            "%s option for %s is %s, instead of %s"
-                            % (statusType, se, seOptions["ReadAccess"], ALLOWED_STATUSES)
-                        )
-                        gLogger.notice("Try specifying the command switches")
-                    else:
-                        resR = resourceStatus.setElementStatus(
-                            se, "StorageElement", statusType, "Active", reason, userName
-                        )
-                        if not resR["OK"]:
-                            gLogger.fatal(f"Failed to update {se} {statusType} to Active, exit -", resR["Message"])
-                            DIRAC.exit(-1)
-                        else:
-                            gLogger.notice(f"Successfully updated {se} {statusType} to Active")
-                            statusAllowedDict[statusType].append(se)
+    for se, seOptions in dbStatus.items():
+        for statusType in (s for s in STATUS_TYPES if statusFlagDict[s]):
+            if seOptions.get(statusType) == "Active":
+                gLogger.notice(f"{statusType} status of {se} is already Active")
+                continue
+            if statusType in seOptions:
+                resR = resourceStatus.setElementStatus(se, "StorageElement", statusType, "Active", reason, userName)
+                if not resR["OK"]:
+                    gLogger.fatal(f"Failed to update {se} {statusType} to Active, exit -", resR["Message"])
+                    DIRAC.exit(-1)
+                else:
+                    gLogger.notice(f"Successfully updated {se} {statusType} to Active")
+                    statusAllowedDict[statusType].append(se)
 
     totalAllowed = 0
     totalAllowedSEs = []

src/DIRAC/DataManagementSystem/scripts/dirac_admin_ban_se.py

@@ -68,6 +68,7 @@ def main():
     from DIRAC.DataManagementSystem.Utilities.DMSHelpers import DMSHelpers, resolveSEGroup
     from DIRAC.Interfaces.API.DiracAdmin import DiracAdmin
     from DIRAC.ResourceStatusSystem.Client.ResourceStatus import ResourceStatus
+    from DIRAC.ResourceStatusSystem.Client.ResourceStatusClient import ResourceStatusClient
 
     ses = resolveSEGroup(ses)
     diracAdmin = DiracAdmin()
@@ -94,135 +95,82 @@ def main():
         gLogger.error("There were no SEs provided")
         DIRAC.exit(-1)
 
-    readBanned = []
-    writeBanned = []
-    checkBanned = []
-    removeBanned = []
+    STATUS_TYPES = ["ReadAccess", "WriteAccess", "CheckAccess", "RemoveAccess"]
+
+    statusBannedDict = {}
+    for statusType in STATUS_TYPES:
+        statusBannedDict[statusType] = []
+
+    statusFlagDict = {}
+    statusFlagDict["ReadAccess"] = read
+    statusFlagDict["WriteAccess"] = write
+    statusFlagDict["CheckAccess"] = check
+    statusFlagDict["RemoveAccess"] = remove
 
     resourceStatus = ResourceStatus()
 
-    res = resourceStatus.getElementStatus(ses, "StorageElement")
-    if not res["OK"]:
-        gLogger.error(f"Storage Element {ses} does not exist")
+    resDB = ResourceStatusClient().selectStatusElement(
+        "Resource", "Status", ses, elementType="StorageElement", meta={"columns": ["Name", "StatusType", "Status"]}
+    )
+    if not resDB["OK"]:
+        gLogger.error("Failed to get the status of the storage elements", resDB["Message"])
         DIRAC.exit(-1)
+    dbStatus = {}
+    for name, statusType, status in resDB["Value"]:
+        dbStatus.setdefault(name, {})[statusType] = status
 
     reason = f"Forced with dirac-admin-ban-se by {userName}"
 
-    for se, seOptions in res["Value"].items():
-        resW = resC = resR = {"OK": False}
-
-        # Eventually, we will get rid of the notion of InActive, as we always write Banned.
-        if read and "ReadAccess" in seOptions:
-            if seOptions["ReadAccess"] == "Banned":
-                gLogger.notice("Read access already banned", se)
-                resR["OK"] = True
-            elif not seOptions["ReadAccess"] in ["Active", "Degraded", "Probing", "Error"]:
-                gLogger.notice(
-                    "Read option for %s is %s, instead of %s"
-                    % (se, seOptions["ReadAccess"], ["Active", "Degraded", "Probing", "Error"])
-                )
-                gLogger.notice("Try specifying the command switches")
-            else:
-                resR = resourceStatus.setElementStatus(se, "StorageElement", "ReadAccess", "Banned", reason, userName)
-                # res = csAPI.setOption( "%s/%s/ReadAccess" % ( storageCFGBase, se ), "InActive" )
+    for se, seOptions in dbStatus.items():
+        for statusType in (s for s in STATUS_TYPES if statusFlagDict[s]):
+            if seOptions.get(statusType) == "Banned":
+                gLogger.notice(f"{statusType} status of {se} is already Banned")
+                continue
+            if statusType in seOptions:
+                resR = resourceStatus.setElementStatus(se, "StorageElement", statusType, "Banned", reason, userName)
                 if not resR["OK"]:
-                    gLogger.error(f"Failed to update {se} read access to Banned")
-                else:
-                    gLogger.notice(f"Successfully updated {se} read access to Banned")
-                    readBanned.append(se)
-
-        # Eventually, we will get rid of the notion of InActive, as we always write Banned.
-        if write and "WriteAccess" in seOptions:
-            if seOptions["WriteAccess"] == "Banned":
-                gLogger.notice("Write access already banned", se)
-                resW["OK"] = True
-            elif not seOptions["WriteAccess"] in ["Active", "Degraded", "Probing"]:
-                gLogger.notice(
-                    "Write option for %s is %s, instead of %s"
-                    % (se, seOptions["WriteAccess"], ["Active", "Degraded", "Probing"])
-                )
-                gLogger.notice("Try specifying the command switches")
-            else:
-                resW = resourceStatus.setElementStatus(se, "StorageElement", "WriteAccess", "Banned", reason, userName)
-                # res = csAPI.setOption( "%s/%s/WriteAccess" % ( storageCFGBase, se ), "InActive" )
-                if not resW["OK"]:
-                    gLogger.error(f"Failed to update {se} write access to Banned")
-                else:
-                    gLogger.notice(f"Successfully updated {se} write access to Banned")
-                    writeBanned.append(se)
-
-        # Eventually, we will get rid of the notion of InActive, as we always write Banned.
-        if check and "CheckAccess" in seOptions:
-            if seOptions["CheckAccess"] == "Banned":
-                gLogger.notice("Check access already banned", se)
-                resC["OK"] = True
-            elif not seOptions["CheckAccess"] in ["Active", "Degraded", "Probing"]:
-                gLogger.notice(
-                    "Check option for %s is %s, instead of %s"
-                    % (se, seOptions["CheckAccess"], ["Active", "Degraded", "Probing"])
-                )
-                gLogger.notice("Try specifying the command switches")
-            else:
-                resC = resourceStatus.setElementStatus(se, "StorageElement", "CheckAccess", "Banned", reason, userName)
-                # res = csAPI.setOption( "%s/%s/CheckAccess" % ( storageCFGBase, se ), "InActive" )
-                if not resC["OK"]:
-                    gLogger.error(f"Failed to update {se} check access to Banned")
+                    gLogger.fatal(f"Failed to update {se} {statusType} to Banned, exit -", resR["Message"])
+                    DIRAC.exit(-1)
                 else:
-                    gLogger.notice(f"Successfully updated {se} check access to Banned")
-                    checkBanned.append(se)
-
-        # Eventually, we will get rid of the notion of InActive, as we always write Banned.
-        if remove and "RemoveAccess" in seOptions:
-            if seOptions["RemoveAccess"] == "Banned":
-                gLogger.notice("Remove access already banned", se)
-                resC["OK"] = True
-            elif not seOptions["RemoveAccess"] in ["Active", "Degraded", "Probing"]:
-                gLogger.notice(
-                    "Remove option for %s is %s, instead of %s"
-                    % (se, seOptions["RemoveAccess"], ["Active", "Degraded", "Probing"])
-                )
-                gLogger.notice("Try specifying the command switches")
-            else:
-                resC = resourceStatus.setElementStatus(se, "StorageElement", "RemoveAccess", "Banned", reason, userName)
-                # res = csAPI.setOption( "%s/%s/CheckAccess" % ( storageCFGBase, se ), "InActive" )
-                if not resC["OK"]:
-                    gLogger.error(f"Failed to update {se} remove access to Banned")
-                else:
-                    gLogger.notice(f"Successfully updated {se} remove access to Banned")
-                    removeBanned.append(se)
-
-        if not (resR["OK"] or resW["OK"] or resC["OK"]):
-            DIRAC.exit(-1)
-
-    if not (writeBanned or readBanned or checkBanned or removeBanned):
-        gLogger.notice("No storage elements were banned")
+                    gLogger.notice(f"Successfully updated {se} {statusType} to Banned")
+                    statusBannedDict[statusType].append(se)
+
+    totalBanned = 0
+    totalBannedSEs = []
+    for statusType in STATUS_TYPES:
+        totalBanned += len(statusBannedDict[statusType])
+        totalBannedSEs += statusBannedDict[statusType]
+    totalBannedSEs = list(set(totalBannedSEs))
+
+    if not totalBanned:
+        gLogger.info("No storage elements were Banned")
         DIRAC.exit(-1)
 
     if mute:
         gLogger.notice("Email is muted by script switch")
         DIRAC.exit(0)
 
-    subject = f"{len(writeBanned + readBanned + checkBanned + removeBanned)} storage elements banned for use"
+    subject = f"{len(totalBannedSEs)} storage elements banned for use"
     addressPath = "EMail/Production"
     address = Operations().getValue(addressPath, "")
     fromAddress = Operations().getValue("ResourceStatus/Config/FromAddress", "")
 
     body = ""
     if read:
         body = f"{body}\n\nThe following storage elements were banned for reading:"
-        for se in readBanned:
+        for se in statusBannedDict["ReadAccess"]:
             body = f"{body}\n{se}"
     if write:
         body = f"{body}\n\nThe following storage elements were banned for writing:"
-        for se in writeBanned:
+        for se in statusBannedDict["WriteAccess"]:
             body = f"{body}\n{se}"
     if check:
         body = f"{body}\n\nThe following storage elements were banned for check access:"
-        for se in checkBanned:
+        for se in statusBannedDict["CheckAccess"]:
             body = f"{body}\n{se}"
     if remove:
         body = f"{body}\n\nThe following storage elements were banned for remove access:"
-        for se in removeBanned:
+        for se in statusBannedDict["RemoveAccess"]:
             body = f"{body}\n{se}"
 
     if not address:

src/DIRAC/Interfaces/API/DiracAdmin.py

@@ -107,7 +107,7 @@ def getSiteMask(self, printOutput=False, status="Active"):
 
     #############################################################################
     def getBannedSites(self, printOutput=False):
-        """Retrieve current list of banned  and probing sites.
+        """Retrieve current list of banned sites.
 
         Example usage:
 
@@ -122,16 +122,12 @@ def getBannedSites(self, printOutput=False):
         if not bannedSites["OK"]:
             return bannedSites
 
-        probingSites = self.sitestatus.getSites(siteState="Probing")
-        if not probingSites["OK"]:
-            return probingSites
-
-        mergedList = sorted(bannedSites["Value"] + probingSites["Value"])
+        bannedList = sorted(bannedSites["Value"])
 
         if printOutput:
-            gLogger.notice("\n".join(mergedList))
+            gLogger.notice("\n".join(bannedList))
 
-        return S_OK(mergedList)
+        return S_OK(bannedList)
 
     #############################################################################
     def getSiteSection(self, site, printOutput=False):
@@ -153,6 +149,20 @@ def getSiteSection(self, site, printOutput=False):
             gLogger.notice(self.pPrint.pformat(result["Value"]))
         return result
 
+    #############################################################################
+    def _getSiteDBStatus(self, site):
+        """Get the status of a site as stored in the RSS Database.
+
+        :param str site: DIRAC site name
+        :return: S_OK,S_ERROR
+        """
+        result = ResourceStatusClient().selectStatusElement("Site", "Status", site, meta={"columns": ["Status"]})
+        if not result["OK"]:
+            return result
+        if not result["Value"]:
+            return S_ERROR(f"Site {site} not found in the RSS database")
+        return S_OK(result["Value"][0][0])
+
     #############################################################################
     def allowSite(self, site, comment, printOutput=False, days=1):
         """Adds the site to the site mask. The site must be a valid DIRAC site name
@@ -170,10 +180,10 @@ def allowSite(self, site, comment, printOutput=False, days=1):
         if not (result := self._checkSiteIsValid(site))["OK"]:
             return result
 
-        if not (result := self.getSiteMask(status="Active"))["OK"]:
+        if not (result := self._getSiteDBStatus(site))["OK"]:
             return result
-        siteMask = result["Value"]
-        if site in siteMask:
+
+        if result["Value"] == "Active":
             if printOutput:
                 gLogger.notice(f"Site {site} is already Active")
             return S_OK(f"Site {site} is already Active")
@@ -246,11 +256,11 @@ def banSite(self, site, comment, printOutput=False, days=1):
         """
         if not (result := self._checkSiteIsValid(site))["OK"]:
             return result
-        mask = self.getSiteMask(status="Banned")
-        if not mask["OK"]:
-            return mask
-        siteMask = mask["Value"]
-        if site in siteMask:
+
+        if not (result := self._getSiteDBStatus(site))["OK"]:
+            return result
+
+        if result["Value"] == "Banned":
             if printOutput:
                 gLogger.notice(f"Site {site} is already Banned")
             return S_OK(f"Site {site} is already Banned")

src/DIRAC/ResourceStatusSystem/Client/ResourceStatus.py

@@ -17,6 +17,7 @@
 from DIRAC.ResourceStatusSystem.Utilities.InfoGetter import getPoliciesThatApply
 from DIRAC.ResourceStatusSystem.Utilities.RSSCacheNoThread import RSSCache
 from DIRAC.ResourceStatusSystem.Utilities.RssConfiguration import RssConfiguration
+from DIRAC.ResourceStatusSystem.PolicySystem.StateMachine import resolveElementStatus
 
 
 class ResourceStatus(metaclass=DIRACSingleton):
@@ -66,13 +67,13 @@ def getElementStatus(self, elementName, elementType, statusType=None, default=No
         >>> getElementStatus('SE1', 'StorageElement', ['ReadAccess', 'WriteAccess'])
             S_OK( { 'SE1': { 'ReadAccess': 'Banned' , 'WriteAccess': 'Active'} } } )
         >>> getElementStatus('SE1', 'StorageElement')
-            S_OK( { 'SE1': { 'ReadAccess': 'Probing' ,
+            S_OK( { 'SE1': { 'ReadAccess': 'Banned' ,
                              'WriteAccess': 'Active',
-                             'CheckAccess': 'Degraded',
+                             'CheckAccess': 'Active',
                              'RemoveAccess': 'Banned'} } } )
         >>> getElementStatus(['CE1', 'CE2'], 'ComputingElement')
             S_OK( {'CE1': {'all': 'Active'},
-                   'CE2': {'all': 'Probing'}}}
+                   'CE2': {'all': 'Banned'}}}
         """
 
         allowedParameters = ["StorageElement", "ComputingElement", "FTS"]
@@ -257,9 +258,7 @@ def getCacheDictFromRawData(rawList):
 
     :return: dict of the form { ( elementName, elementType, statusType, vO ) : status, ... }
     """
-
     res = {}
     for entry in rawList:
-        res.update({(entry[0], entry[1], entry[2], entry[4]): entry[3]})
-
+        res[(entry[0], entry[1], entry[2], entry[4])] = resolveElementStatus(entry[3])
     return res