Skip to content

[9.0] fix: JobWrapper does not manage to send job parameters #8399

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aldbr wants to merge 1 commit into
base: integration
Choose a base branch
from
+15 −11
Open

[9.0] fix: JobWrapper does not manage to send job parameters #8399

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/DIRAC/Resources/Computing/ComputingElement.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@
getNumberOfProcessors,
)

INTEGER_PARAMETERS = ["CPUTime", "NumberOfProcessors", "NumberOfPayloadProcessors", "MaxRAM"]
INTEGER_PARAMETERS = ["CPUTime", "CPUNormalizationFactor", "NumberOfProcessors", "NumberOfPayloadProcessors", "MaxRAM"]
FLOAT_PARAMETERS = ["WaitingToRunningRatio"]
LIST_PARAMETERS = ["Tag", "RequiredTag"]
WAITING_TO_RUNNING_RATIO = 0.5
Expand Down Expand Up @@ -450,7 +450,7 @@ def getCEConfigDict(section: str) -> dict:
ceOptions = result["Value"]
for key in ceOptions:
if key in INTEGER_PARAMETERS:
ceOptions[key] = int(ceOptions[key])
ceOptions[key] = int(float(ceOptions[key]))
Copy link
Contributor Author

@aldbr aldbr Dec 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In DIRACGrid/diracx#653 we said that CPUNormalizationFactor should be treated as a float.
I still process it as an int here, because I would also need to modify diracx and OS before adding it to FLOAT_PARAMETERS.
Once merged I will open an issue to keep track of it of course.

if key in FLOAT_PARAMETERS:
ceOptions[key] = float(ceOptions[key])
if key in LIST_PARAMETERS:
Expand Down
16 changes: 10 additions & 6 deletions src/DIRAC/WorkloadManagementSystem/Client/JobReport.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,16 +115,20 @@ def sendStoredJobParameters(self):

def commit(self):
"""Send all the accumulated information"""
messages = []

success = True
result = self.sendStoredStatusInfo()
success &= result["OK"]
if not result["OK"]:
messages.append(result["Message"])
result = self.sendStoredJobParameters()
success &= result["OK"]
if not result["OK"]:
messages.append(result["Message"])

if success:
return S_OK()
return S_ERROR("Information upload to JobStateUpdate service failed")
if messages:
gLogger.warn("Some information could not be uploaded to JobStateUpdate service:", "; ".join(messages))
return S_ERROR("Information upload to JobStateUpdate service failed")

return S_OK()

def dump(self):
"""Print out the contents of the internal cached information"""
Expand Down
4 changes: 2 additions & 2 deletions src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -546,7 +546,7 @@ def postProcess(
self.__report(status=JobStatus.FAILED, minorStatus=JobMinorStatus.APP_THREAD_FAILED, sendFlag=True)
applicationErrorStatus = "None reported"
if payloadStatus:
applicationErrorStatus = payloadStatus
applicationErrorStatus = str(payloadStatus)
self.__setJobParam("ApplicationError", applicationErrorStatus, sendFlag=True)

# This might happen if process() and postProcess() are called on different machines
Expand Down Expand Up @@ -1544,7 +1544,7 @@ def __report(self, status="", minorStatus="", sendFlag=False):
#############################################################################
def __setJobParam(self, name, value, sendFlag=False):
"""Wraps around setJobParameter of JobReport client"""
jobParam = self.jobReport.setJobParameter(str(name), str(value), sendFlag)
jobParam = self.jobReport.setJobParameter(str(name), value, sendFlag)
if not jobParam["OK"]:
self.log.warn("Failed setting job parameter", jobParam["Message"])
if self.jobID:
Expand Down
2 changes: 1 addition & 1 deletion src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -565,7 +565,7 @@ def test_postProcess_executor_failed_status_defined(setup_job_wrapper, mocker, m
assert result["OK"]
assert report_args[-1]["status"] == JobStatus.COMPLETING
assert report_args[-1]["minorStatus"] == JobMinorStatus.APP_ERRORS
assert set_param_args[-3][0][1] == 126
assert set_param_args[-3][0][1] == "126"


def test_postProcess_subprocess_not_complete(setup_job_wrapper, mocker, mock_report_and_set_param):
Expand Down
Loading