Merge pull request #5937 from DIRACGridBot/cherry-pick-2-0fdafbc52-integration

[sweep:integration] using a DErrno for no match found

Author: fstagni

src/DIRAC/Core/Base/AgentModule.py

@@ -18,9 +18,9 @@
 from DIRAC.Core.Utilities import Time, MemStat, Network
 from DIRAC.Core.Utilities.Shifter import setupShifterProxyInEnv
 from DIRAC.Core.Utilities.ReturnValues import isReturnStructure
-from DIRAC.FrameworkSystem.Client.MonitoringClient import gMonitor
 from DIRAC.ConfigurationSystem.Client import PathFinder
 from DIRAC.FrameworkSystem.Client.MonitoringClient import MonitoringClient
+from DIRAC.FrameworkSystem.Client.MonitoringClient import gMonitor
 from DIRAC.Core.Utilities.ThreadScheduler import gThreadScheduler
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 
@@ -145,8 +145,6 @@ def __init__(self, agentName, loadName, baseAgentName=False, properties={}):
 
         self.__monitorLastStatsUpdate = -1
         self.monitor = None
-        self.__initializeMonitor()
-        self.__initialized = False
 
     def __getCodeInfo(self):
 
@@ -187,6 +185,8 @@ def am_initialize(self, *initArgs):
         # Set the work directory in an environment variable available to subprocesses if needed
         os.environ["AGENT_WORKDIRECTORY"] = workDirectory
 
+        self.__initializeMonitor()
+
         self.__moduleProperties["shifterProxy"] = self.am_getOption("shifterProxy")
         if self.am_monitoringEnabled() and not self.activityMonitoring:
             self.monitor.enable()
@@ -217,7 +217,6 @@ def am_initialize(self, *initArgs):
         else:
             self.log.notice(" Watchdog interval: disabled ")
         self.log.notice("=" * 40)
-        self.__initialized = True
         return S_OK()
 
     def am_getControlDirectory(self):

src/DIRAC/Core/Utilities/DErrno.py

@@ -119,6 +119,7 @@
 EWMSSUBM = 1503
 EWMSJMAN = 1504
 EWMSSTATUS = 1505
+EWMSNOMATCH = 1510
 EWMSNOPILOT = 1550
 
 # ## DMS/StorageManagement (16XX)
@@ -198,6 +199,7 @@
     1503: "EWMSSUBM",
     1504: "EWMSJMAN",
     1505: "EWMSSTATUS",
+    1510: "EWMSNOMATCH",
     1550: "EWMSNOPILOT",
     # DMS/StorageManagement
     1601: "EFILESIZE",
@@ -274,6 +276,7 @@
     EWMSJMAN: "Job management error",
     EWMSSTATUS: "Job status error",
     EWMSNOPILOT: "No pilots found",
+    EWMSNOMATCH: "No match found",
     # DMS/StorageManagement
     EFILESIZE: "Bad file size",
     EGFAL: "Error with the gfal call",

src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py

@@ -23,6 +23,7 @@
 from DIRAC.Core.Base.AgentModule import AgentModule
 from DIRAC.Core.Security.ProxyInfo import getProxyInfo
 from DIRAC.Core.Security import Properties
+from DIRAC.Core.Utilities import DErrno
 from DIRAC.FrameworkSystem.Client.ProxyManagerClient import gProxyManager
 from DIRAC.Resources.Computing.BatchSystems.TimeLeft.TimeLeft import TimeLeft
 from DIRAC.Resources.Computing.ComputingElementFactory import ComputingElementFactory
@@ -181,7 +182,10 @@ def execute(self):
             # if we don't match a job, independently from the reason,
             # we wait a bit longer before trying again
             self.am_setOption("PollingTime", int(self.am_getOption("PollingTime") * 1.5))
-            return self._checkMatchingIssues(jobRequest["Message"])
+            res = self._checkMatchingIssues(jobRequest)
+            if not res["OK"]:
+                self._finish(res["Message"])
+            return res
 
         # Reset the Counter
         self.matchFailedCount = 0
@@ -543,28 +547,29 @@ def _matchAJob(self, ceDictList):
                 break
         return jobRequest
 
-    def _checkMatchingIssues(self, issueMessage):
+    def _checkMatchingIssues(self, jobRequest):
         """Check the source of the matching issue
 
-        :param str issueMessage: message returned by the matcher
+        :param dict jobRequest: S_ERROR returned by the matcher
         :return: S_OK/S_ERROR
         """
-        if issueMessage.find("Pilot version does not match") != -1:
+
+        if jobRequest["Message"].find("Pilot version does not match") != -1:
             errorMsg = "Pilot version does not match the production version"
-            self.log.error(errorMsg, issueMessage.replace(errorMsg, ""))
-            return S_ERROR(issueMessage)
+            self.log.error(errorMsg, jobRequest["Message"].replace(errorMsg, ""))
+            return jobRequest
 
-        if re.search("No match found", issueMessage):
-            self.log.notice("Job request OK, but no match found", ": %s" % issueMessage)
-        elif issueMessage.find("seconds timeout") != -1:
-            self.log.error("Timeout while requesting job", issueMessage)
+        if DErrno.cmpError(jobRequest, DErrno.EWMSNOMATCH):
+            self.log.notice("Job request OK, but no match found", jobRequest["Message"])
+        elif jobRequest["Message"].find("seconds timeout") != -1:
+            self.log.error("Timeout while requesting job", jobRequest["Message"])
         else:
-            self.log.notice("Failed to get jobs", ": %s" % issueMessage)
+            self.log.notice("Failed to get jobs", jobRequest["Message"])
 
         self.matchFailedCount += 1
         if self.matchFailedCount > self.stopAfterFailedMatches:
             return self._finish("Nothing to do for more than %d cycles" % self.stopAfterFailedMatches)
-        return S_OK(issueMessage)
+        return S_OK()
 
     def _checkMatcherInfo(self, matcherInfo, matcherParams, jobReport):
         """Check that all relevant information about the job are available"""

src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py

@@ -11,13 +11,12 @@
 
 import sys
 import re
-import time
-import six
 import random
 from collections import defaultdict
 
 from DIRAC import S_OK, S_ERROR, gConfig
 from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader
+from DIRAC.Core.Utilities import DErrno
 from DIRAC.ConfigurationSystem.Client.PathFinder import getSystemInstance
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.FrameworkSystem.Client.ProxyManagerClient import gProxyManager
@@ -336,7 +335,7 @@ def execute(self):
                     break
 
             if not jobRequest["OK"]:
-                self._checkMatchingIssues(jobRequest["Message"])
+                self._checkMatchingIssues(jobRequest)
                 self.failedQueues[queueName] += 1
                 continue
 
@@ -402,18 +401,17 @@ def _setCEDict(self, ceDict):
         if project:
             ceDict["ReleaseProject"] = project
 
-    def _checkMatchingIssues(self, issueMessage):
+    def _checkMatchingIssues(self, jobRequest):
         """Check the source of the matching issue
 
-        :param str issueMessage: message returned by the matcher
-        :return: S_OK/S_ERROR
+        :param dict jobRequest: S_ERROR returned by the matcher
+        :return: S_OK
         """
-        matchingFailed = False
-        if re.search("No match found", issueMessage):
-            self.log.notice("Job request OK, but no match found", ": %s" % issueMessage)
-        elif issueMessage.find("seconds timeout") != -1:
-            self.log.error("Timeout while requesting job", issueMessage)
+        if DErrno.cmpError(jobRequest, DErrno.EWMSNOMATCH):
+            self.log.notice("Job request OK, but no match found", jobRequest["Message"])
+        elif jobRequest["Message"].find("seconds timeout") != -1:
+            self.log.error("Timeout while requesting job", jobRequest["Message"])
         else:
-            self.log.notice("Failed to get jobs", ": %s" % issueMessage)
+            self.log.notice("Failed to get jobs", jobRequest["Message"])
 
-        return S_OK(issueMessage)
+        return S_OK()

src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobAgent.py

@@ -7,7 +7,7 @@
 from DIRAC.WorkloadManagementSystem.Client.JobReport import JobReport
 from DIRAC.Resources.Computing.ComputingElementFactory import ComputingElementFactory
 from DIRAC.Resources.Computing.BatchSystems.TimeLeft.TimeLeft import TimeLeft
-from DIRAC import gLogger
+from DIRAC import gLogger, S_ERROR
 
 gLogger.setLevel("DEBUG")
 
@@ -215,7 +215,7 @@ def test__checkMatchingIssues(mocker, issueMessage, stopAfterFailedMatches, matc
     jobAgent.stopAfterFailedMatches = stopAfterFailedMatches
     jobAgent.matchFailedCount = matchFailedCount
 
-    result = jobAgent._checkMatchingIssues(issueMessage)
+    result = jobAgent._checkMatchingIssues(S_ERROR(issueMessage))
     assert result["OK"] == expectedResult
 
 

src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_PushJobAgent.py

@@ -6,11 +6,10 @@
 from collections import defaultdict
 
 # DIRAC Components
-from DIRAC.WorkloadManagementSystem.Agent.PushJobAgent import PushJobAgent
-from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
+from DIRAC.WorkloadManagementSystem.Agent.PushJobAgent import PushJobAgent
 
-from DIRAC import gLogger
+from DIRAC import gLogger, S_ERROR
 
 gLogger.setLevel("DEBUG")
 
@@ -98,5 +97,5 @@ def test__checkMatchingIssues(mocker, issueMessage, expectedResult):
     jobAgent.log = gLogger
     jobAgent.log.setLevel("DEBUG")
 
-    result = jobAgent._checkMatchingIssues(issueMessage)
+    result = jobAgent._checkMatchingIssues(S_ERROR(issueMessage))
     assert result["OK"] == expectedResult

src/DIRAC/WorkloadManagementSystem/Service/MatcherHandler.py

@@ -7,6 +7,7 @@
 
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.Core.DISET.RequestHandler import RequestHandler
+from DIRAC.Core.Utilities import DErrno
 from DIRAC.Core.Utilities.DEncode import ignoreEncodeWarning
 from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader
 
@@ -87,8 +88,7 @@ def export_requestJob(self, resourceDescription):
             gMonitor.addMark("matchesDone")
             gMonitor.addMark("matchesOK")
             return S_OK(result)
-        # FIXME: This is correctly interpreted by the JobAgent, but DErrno should be used instead
-        return S_ERROR("No match found")
+        return S_ERROR(DErrno.EWMSNOMATCH)
 
     ##############################################################################
     types_getActiveTaskQueues = []