Merge pull request #5208 from fstagni/v7r3-fixes11

[v7r3] Hackathon fixes

environment-py3.yml

@@ -19,7 +19,7 @@ dependencies:
   - elasticsearch-dsl
   - future
   - gitpython >=2.1.0
-  - m2crypto >=0.36
+  - m2crypto >=0.36,!=0.38.0
   - matplotlib
   - numpy
   - pexpect >=4.0.1
@@ -93,5 +93,5 @@ dependencies:
     - git+https://gitlab.cern.ch/cburr/fts-rest.git@python3
     # Doesn't support Python 3.9
     # - rucio-clients >=1.25.6
-    - git+https://github.com/rucio/rucio.git@master
+    - git+https://github.com/rucio/rucio.git@eafde85bfbfe51fb22c7774e2af8deda3fb05e81
     - -e .

src/DIRAC/Core/Utilities/Subprocess.py

@@ -379,7 +379,11 @@ def __readFromFile(self, fd, baseLength):
           nB = fd.read(1)
         if not nB:
           break
-        dataString += nB.decode()
+        try:
+          dataString += nB.decode()
+        except UnicodeDecodeError as e:
+          self.log.warn("Unicode decode error in readFromFile", "(%r): %r" % (e, dataString))
+          dataString += nB.decode("utf-8", "replace")
         # break out of potential infinite loop, indicated by dataString growing beyond reason
         if len(dataString) + baseLength > self.bufferLimit:
           self.log.error("DataString is getting too long (%s): %s " % (len(dataString), dataString[-10000:]))
@@ -426,10 +430,7 @@ def systemCall(self, cmdSeq, callbackFunction=None, shell=False, env=None):
 
     self.cmdSeq = cmdSeq
     self.callback = callbackFunction
-    if sys.platform.find("win") == 0:
-      closefd = False
-    else:
-      closefd = True
+    closefd = sys.platform.find("win") != 0
     try:
       self.child = subprocess.Popen(self.cmdSeq,
                                     shell=shell,

src/DIRAC/Core/Utilities/test/Test_Subprocess.py

@@ -27,7 +27,7 @@
 from subprocess import Popen
 
 # SUT
-from DIRAC.Core.Utilities.Subprocess import systemCall, shellCall, pythonCall, getChildrenPIDs
+from DIRAC.Core.Utilities.Subprocess import systemCall, shellCall, pythonCall, getChildrenPIDs, Subprocess
 
 # Mark this entire module as slow
 pytestmark = pytest.mark.slow
@@ -66,3 +66,18 @@ def test_getChildrenPIDs():
     assert isinstance(p, int)
 
   mainProcess.wait()
+
+
+def test_decodingCommandOutput():
+  retVal = systemCall(10, ["echo", "-e", "-n", r"\xdf"])
+  assert retVal["OK"]
+  assert retVal["Value"] == (0, u"\ufffd", "")
+
+  retVal = systemCall(10, ["echo", "-e", r"\xdf"])
+  assert retVal["OK"]
+  assert retVal["Value"] == (0, u"\ufffd\n", "")
+
+  sp = Subprocess()
+  retVal = sp.systemCall(r"""python -c 'import os; os.fdopen(2, "wb").write(b"\xdf")'""", shell=True)
+  assert retVal["OK"]
+  assert retVal["Value"] == (0, "", u"\ufffd")

src/DIRAC/ResourceStatusSystem/Agent/ElementInspectorAgent.py

@@ -147,7 +147,7 @@ def execute(self):
       try:
         future.result()
       except Exception as exc:
-        self.log.error('%d generated an exception: %s' % (transID, exc))
+        self.log.exception('%s generated an exception: %s' % (transID, exc))
       else:
         self.log.info('Processed', transID)
 

src/DIRAC/ResourceStatusSystem/Agent/SiteInspectorAgent.py

@@ -51,27 +51,27 @@ def __init__(self, *args, **kwargs):
 
     AgentModule.__init__(self, *args, **kwargs)
 
-    self.siteClient = None
+    self.rsClient = None
     self.clients = {}
 
   def initialize(self):
     """ Standard initialize.
     """
 
-    res = ObjectLoader().loadObject('DIRAC.ResourceStatusSystem.Client.SiteStatus')
-    if not res['OK']:
-      self.log.error('Failed to load SiteStatus class: %s' % res['Message'])
-      return res
-    siteStatusClass = res['Value']
-
     res = ObjectLoader().loadObject('DIRAC.ResourceStatusSystem.Client.ResourceManagementClient')
     if not res['OK']:
       self.log.error('Failed to load ResourceManagementClient class: %s' % res['Message'])
       return res
     rmClass = res['Value']
 
-    self.siteClient = siteStatusClass()
-    self.clients['SiteStatus'] = siteStatusClass()
+    res = ObjectLoader().loadObject('DIRAC.ResourceStatusSystem.Client.ResourceStatusClient')
+    if not res['OK']:
+      self.log.error('Failed to load ResourceStatusClient class: %s' % res['Message'])
+      return res
+    rsClass = res['Value']
+
+    self.rsClient = rsClass()
+    self.clients['ResourceStatusClient'] = rsClass()
     self.clients['ResourceManagementClient'] = rmClass()
 
     maxNumberOfThreads = self.am_getOption('maxNumberOfThreads', 15)
@@ -85,15 +85,11 @@ def execute(self):
     It gets the sites from the Database which are eligible to be re-checked.
     """
 
-    res = self.siteClient.getSites('All')
-    if not res['OK']:
-      return res
-
     utcnow = datetime.datetime.utcnow().replace(microsecond=0)
     future_to_element = {}
 
     # get the current status
-    res = self.siteClient.getSiteStatuses(res['Value'])
+    res = self.rsClient.selectStatusElement('Site', 'Status')
     if not res['OK']:
       return res
 
@@ -119,7 +115,8 @@ def execute(self):
       self.log.verbose('"%s" # %s # %s' % (siteDict['Name'],
                                            siteDict['Status'],
                                            siteDict['LastCheckTime']))
-      lowerElementDict = {}
+
+      lowerElementDict = {'element': 'Site'}
       for key, value in siteDict.items():
         if len(key) >= 2:  # VO !
           lowerElementDict[key[0].lower() + key[1:]] = value
@@ -132,7 +129,7 @@ def execute(self):
       try:
         future.result()
       except Exception as exc:
-        self.log.error('%d generated an exception: %s' % (transID, exc))
+        self.log.exception('%s generated an exception: %s' % (transID, exc))
       else:
         self.log.info('Processed', transID)
 
@@ -147,8 +144,9 @@ def _execute(self, site):
     pep = PEP(clients=self.clients)
 
     self.log.verbose(
-        '%s ( status=%s / statusType=%s ) being processed' % (
+        '%s ( VO=%s / status=%s / statusType=%s ) being processed' % (
             site['name'],
+            site['vO'],
             site['status'],
             site['statusType']))
 

src/DIRAC/ResourceStatusSystem/Agent/test/Test_Agent_SiteInspectorAgent.py

@@ -22,6 +22,7 @@
 site = {
     'status': 'status',
     'name': 'site',
+    'vO': 'some_vo',
     'site': 'site',
     'element': 'Site',
     'statusType': 'all',

src/DIRAC/ResourceStatusSystem/Client/SiteStatus.py

@@ -41,6 +41,7 @@ class SiteStatus(object):
   * getUsableSites
   * getSites
   """
+
   def __init__(self):
     """
     Constructor, initializes the rssClient.

src/DIRAC/ResourceStatusSystem/Command/PropagationCommand.py

@@ -28,10 +28,10 @@ def doNew(self, masterParams=None):
 
   def doCache(self):
 
-    if not self.args['site']:
+    if not self.args['name']:
       return S_ERROR('site was not found in args')
 
-    site = self.args['site']
+    site = self.args['name']
 
     elements = CSHelpers.getSiteElements(site)
 

src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py

@@ -367,6 +367,18 @@ def execute(self):
         return self._rescheduleFailedJob(
             jobID, errorMsg, self.stopOnApplicationFailure)
 
+      gridCE = gConfig.getValue('/LocalSite/GridCE', '')
+      if gridCE:
+        jobReport.setJobParameter(par_name='GridCE',
+                                  par_value=gridCE,
+                                  sendFlag=False)
+
+      queue = gConfig.getValue('/LocalSite/CEQueue', '')
+      if queue:
+        jobReport.setJobParameter(par_name='CEQueue',
+                                  par_value=queue,
+                                  sendFlag=False)
+
       self.log.debug('Before self._submitJob() (%sCE)' % (self.ceName))
       result_submitJob = self._submitJob(
           jobID=jobID,
@@ -573,15 +585,6 @@ def _submitJob(self, jobID, jobParams, resourceParams, optimizerParams,
     jobReport.setJobStatus(status=JobStatus.MATCHED,
                            minorStatus='Submitting To CE')
 
-    gridCE = gConfig.getValue('/LocalSite/GridCE', '')
-    queue = gConfig.getValue('/LocalSite/CEQueue', '')
-    jobReport.setJobParameter(par_name='GridCE',
-                              par_value=gridCE,
-                              sendFlag=False)
-    jobReport.setJobParameter(par_name='CEQueue',
-                              par_value=queue,
-                              sendFlag=False)
-
     self.log.info('Submitting JobWrapper',
                   '%s to %sCE' % (os.path.basename(wrapperFile), self.ceName))
 

tests/Integration/WorkloadManagementSystem/Test_Client_WMS.py

@@ -356,7 +356,7 @@ def test_JobStateUpdateAndJobMonitoringMultuple(self):
                     resJG_olderThanOneYear)
     res = jobMonitor.getStates()
     self.assertTrue(res['OK'], res.get('Message'))
-    self.assertTrue(sorted(res['Value']) in [[JobStatus.RECEIVED], sorted([JobStatus.RECEIVED, JobStatus.WAITING])],
+    self.assertTrue(sorted(res['Value']) in [[JobStatus.RECEIVED], sorted([JobStatus.RECEIVED, JobStatus.KILLED])],
                     res['Value'])
     res = jobMonitor.getMinorStates()
     self.assertTrue(res['OK'], res.get('Message'))

tests/Jenkins/dirac_ci.sh

@@ -366,7 +366,7 @@ fullInstallDIRAC() {
   dirac-admin-allow-se SE-1 SE-2 S3-DIRECT S3-INDIRECT --All
 
   #agents
-  findAgents
+  findAgents 'exclude' 'Rucio'
   if ! diracAgents; then
     echo "ERROR: diracAgents failed"
     exit 1