Merge pull request #6004 from fstagni/cherry-pick-2-dce63b79f-integration

[sweep:integration] fix: StalledJobAgent can force status from Submitting to Failed

Author: fstagni

src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py

@@ -562,7 +562,7 @@ def _failSubmittingJobs(self):
             return result
 
         for jobID in result["Value"]:
-            result = self._updateJobStatus(jobID, JobStatus.FAILED)
+            result = self._updateJobStatus(jobID, JobStatus.FAILED, force=True)
             if not result["OK"]:
                 self.log.error("Failed to update job status", result["Message"])
                 continue

src/DIRAC/WorkloadManagementSystem/DB/ElasticJobParametersDB.py

@@ -28,6 +28,8 @@
       - setJobParameter()
       - deleteJobParameters()
 """
+import hashlib
+
 from DIRAC import S_OK, gConfig
 from DIRAC.ConfigurationSystem.Client.PathFinder import getDatabaseSection
 from DIRAC.ConfigurationSystem.Client.Helpers import CSGlobals
@@ -125,7 +127,11 @@ def setJobParameter(self, jobID, key, value):
 
         self.log.debug("Inserting data in %s:%s" % (self.indexName, data))
 
-        result = self.index(self.indexName, body=data, docID=str(jobID) + key)
+        # The _id in ES can't exceed 512 bytes, this is a ES hard-coded limitation.
+        docID = str(jobID) + key
+        if len(docID) > 505:
+            docID = docID[:475] + hashlib.md5(docID.encode()).hexdigest()[:35]
+        result = self.index(self.indexName, body=data, docID=docID)
         if not result["OK"]:
             self.log.error("ERROR: Couldn't insert data", result["Message"])
         return result
@@ -142,10 +148,12 @@ def setJobParameters(self, jobID, parameters):
         """
         self.log.debug("Inserting parameters", "in %s: for job %s : %s" % (self.indexName, jobID, parameters))
 
-        parametersListDict = [
-            {"JobID": jobID, "Name": parName, "Value": parValue, "_id": str(jobID) + str(parName) + str(parValue)}
-            for parName, parValue in parameters
-        ]
+        parametersListDict = []
+        for parName, parValue in parameters:
+            docID = str(jobID) + parName
+            if len(docID) > 505:
+                docID = docID[:475] + hashlib.md5(docID.encode()).hexdigest()[:35]
+            parametersListDict.append({"JobID": jobID, "Name": parName, "Value": parValue, "_id": docID})
 
         result = self.bulk_index(self.indexName, data=parametersListDict, period=None, withTimeStamp=False)
         if not result["OK"]: