Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
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
10 changes: 9 additions & 1 deletion src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ def __init__(self, agentName, loadName, baseAgentName=False, properties=None):
# Localsite options
self.siteName = "Unknown"
self.pilotReference = "Unknown"
self.pilotStamp = self.pilotReference
self.defaultProxyLength = 86400 * 5

# Agent options
Expand Down Expand Up @@ -116,6 +117,7 @@ def initialize(self):
# Localsite options
self.siteName = siteName()
self.pilotReference = gConfig.getValue("/LocalSite/PilotReference", self.pilotReference)
self.pilotStamp = os.environ.get("DIRAC_PILOT_STAMP", self.pilotReference)
self.defaultProxyLength = gConfig.getValue("/Registry/DefaultProxyLifeTime", self.defaultProxyLength)
# Agent options
# This is the factor to convert raw CPU to Normalized units (based on the CPU Model)
Expand Down Expand Up @@ -895,7 +897,13 @@ def finalize(self):
gridCE = gConfig.getValue("/LocalSite/GridCE", "")
queue = gConfig.getValue("/LocalSite/CEQueue", "")
result = PilotManagerClient().setPilotStatus(
str(self.pilotReference), PilotStatus.DONE, gridCE, "Report from JobAgent", self.siteName, queue
str(self.pilotReference),
PilotStatus.DONE,
gridCE,
"Report from JobAgent",
self.siteName,
queue,
self.pilotStamp,
)
if not result["OK"]:
self.log.warn("Issue setting the pilot status", result["Message"])
Expand Down
Original file line number Diff line number Diff line change
@@ -1,13 +1,17 @@
""" Module that contains client access to the Pilots handler.
"""
"""Module that contains client access to the Pilots handler."""

from DIRAC.Core.Base.Client import Client, createClient
from DIRAC.WorkloadManagementSystem.FutureClient.PilotManagerClient import (
PilotManagerClient as futurePilotManagerClient,
)


@createClient("WorkloadManagement/PilotManager")
class PilotManagerClient(Client):
"""PilotManagerClient sets url for the PilotManagerHandler."""

diracxClient = futurePilotManagerClient

def __init__(self, url=None, **kwargs):
"""
Sets URL for PilotManager handler
Expand Down
15 changes: 11 additions & 4 deletions src/DIRAC/WorkloadManagementSystem/DB/PilotAgentsDB.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@ def setPilotStatus(
benchmark=None,
currentJob=None,
updateTime=None,
pilotStamp=None,
conn=False,
):
"""Set pilot job status"""
Expand Down Expand Up @@ -123,17 +124,23 @@ def setPilotStatus(
setList.append("CurrentJobID=%s")
args.append(str(int(currentJob)))
if destination:
setList.append(f"DestinationSite=%s")
setList.append("DestinationSite=%s")
args.append(destination)
if not gridSite:
res = getCESiteMapping(destination)
if res["OK"] and res["Value"]:
setList.append(f"GridSite=%s")
setList.append("GridSite=%s")
args.append(res["Value"][destination])

set_string = ",".join(setList)
req = f"UPDATE PilotAgents SET {set_string} WHERE PilotJobReference=%s" # nosec
args.append(pilotRef)
req = f"UPDATE PilotAgents SET {set_string}" # nosec

if pilotStamp:
req += " WHERE PilotStamp=%s"
args.append(pilotStamp)
else:
req += " WHERE PilotJobReference=%s"
args.append(pilotRef)
return self._update(req, args=args, conn=conn)

def selectPilots(
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
from DIRAC.Core.Security.DiracX import DiracXClient, FutureClient
from DIRAC.Core.Utilities.ReturnValues import convertToReturnValue


class PilotManagerClient(FutureClient):
@convertToReturnValue
def addPilotReferences(
self,
pilot_reference: str,
VO: str,
grid_type: str = "DIRAC",
pilot_stamp_dict: dict = {},
grid_site: str = "Unknown",
destination_site: str = "NotAssigned",
):
"""Add a new pilot to the database.
Uses pilots/management/register_pilot
"""

with DiracXClient() as api:
api.pilots.register_pilot(
pilot_stamp=pilot_stamp_dict[pilot_reference],
vo=VO,
grid_type=grid_type,
grid_site=grid_site,
destination_site=destination_site,
)

@convertToReturnValue
def getPilotInfo(self, pilot_reference: str):
"""Get the info about a given pilot job reference"""

# with DiracXClient() as api:

@convertToReturnValue
def selectPilots(self, conditions_dict: dict):
"""Select pilots given the selection conditions"""

# with DiracXClient() as api:

@convertToReturnValue
def getPilotSummary(self, start_date: str, end_date: str):
"""Get summary of the status of the Pilot Jobs"""

# with DiracXClient() as api:

@convertToReturnValue
def getPilots(self, job_id: str | int):
"""Get pilots executing/having executed a Job"""

# with DiracXClient() as api:

@convertToReturnValue
def setPilotStatus(
self,
pilot_reference: str,
status: str,
destination: str | None = None,
reason: str | None = None,
grid_site: str | None = None,
queue: str | None = None,
pilot_stamp: str | None = None,
):
"""Set the pilot status"""

with DiracXClient() as api:
values_dict = (
{
"PilotStamp": pilot_stamp,
"Status": status,
"DestinationSite": destination,
"StatusReason": reason,
"GridSite": grid_site,
"Queue": queue,
},
)

return api.pilots.update_pilot_metadata(pilot_stamps_to_fields_mapping=[values_dict])
Original file line number Diff line number Diff line change
Expand Up @@ -73,9 +73,17 @@ def export_getPilots(cls, jobID):
types_setPilotStatus = [str, str]

@classmethod
def export_setPilotStatus(cls, pilotRef, status, destination=None, reason=None, gridSite=None, queue=None):
def export_setPilotStatus(
cls, pilotRef, status, destination=None, reason=None, gridSite=None, queue=None, pilotStamp=None
):
"""Set the pilot agent status"""

return cls.pilotAgentsDB.setPilotStatus(
pilotRef, status, destination=destination, statusReason=reason, gridSite=gridSite, queue=queue
pilotRef,
status,
destination=destination,
statusReason=reason,
gridSite=gridSite,
queue=queue,
pilotStamp=pilotStamp,
)
Loading