test/lsst/lsstSubmit.py

0001 import sys
0002 import time
0003 import uuid
0004
0005 from pandaserver.taskbuffer.FileSpec import FileSpec
0006 from pandaserver.taskbuffer.JobSpec import JobSpec
0007 from pandaserver.userinterface import Client
0008
0009 prodUserNameDefault = "unknown-user"
0010 prodUserName = None
0011 prodUserNameDP = None
0012 prodUserNamePipeline = None
0013 site = "ANALY_BNL-LSST"
0014 PIPELINE_TASK = None
0015 PIPELINE_PROCESSINSTANCE = None
0016 PIPELINE_EXECUTIONNUMBER = None
0017 PIPELINE_STREAM = None
0018 lsstJobParams = ""
0019
0020 for idx, argv in enumerate(sys.argv):
0021     if argv == "--site":
0022         try:
0023             site = sys.argv[idx + 1]
0024         except Exception:
0025             site = "ANALY_BNL-LSST"
0026     if argv == "-DP_USER":
0027         try:
0028             prodUserNameDP = sys.argv[idx + 1]
0029             if len(lsstJobParams):
0030                 lsstJobParams += "|"
0031             lsstJobParams += f"DP_USER={str(prodUserNameDP)}"
0032         except Exception:
0033             prodUserNameDP = None
0034     if argv == "-PIPELINE_USER":
0035         try:
0036             prodUserNamePipeline = sys.argv[idx + 1]
0037             if len(lsstJobParams):
0038                 lsstJobParams += "|"
0039             lsstJobParams += f"PIPELINE_USER={str(prodUserNamePipeline)}"
0040         except Exception:
0041             prodUserNamePipeline = None
0042     if argv == "-PIPELINE_TASK":
0043         try:
0044             PIPELINE_TASK = sys.argv[idx + 1]
0045             if len(lsstJobParams):
0046                 lsstJobParams += "|"
0047             lsstJobParams += f"PIPELINE_TASK={str(PIPELINE_TASK)}"
0048         except Exception:
0049             PIPELINE_TASK = None
0050     if argv == "-PIPELINE_PROCESSINSTANCE":
0051         try:
0052             PIPELINE_PROCESSINSTANCE = int(sys.argv[idx + 1])
0053             if len(lsstJobParams):
0054                 lsstJobParams += "|"
0055             lsstJobParams += f"PIPELINE_PROCESSINSTANCE={str(PIPELINE_PROCESSINSTANCE)}"
0056         except Exception:
0057             PIPELINE_PROCESSINSTANCE = None
0058     if argv == "-PIPELINE_EXECUTIONNUMBER":
0059         try:
0060             PIPELINE_EXECUTIONNUMBER = int(sys.argv[idx + 1])
0061             if len(lsstJobParams):
0062                 lsstJobParams += "|"
0063             lsstJobParams += f"PIPELINE_EXECUTIONNUMBER={str(PIPELINE_EXECUTIONNUMBER)}"
0064         except Exception:
0065             PIPELINE_EXECUTIONNUMBER = None
0066     if argv == "-PIPELINE_STREAM":
0067         try:
0068             PIPELINE_STREAM = int(sys.argv[idx + 1])
0069             if len(lsstJobParams):
0070                 lsstJobParams += "|"
0071             lsstJobParams += f"PIPELINE_STREAM={str(PIPELINE_STREAM)}"
0072         except Exception:
0073             PIPELINE_STREAM = None
0074
0075 # DP_USER and PIPELINE_USER preference
0076 if prodUserNameDP is not None:
0077     prodUserName = prodUserNameDP
0078 elif prodUserNamePipeline is not None:
0079     prodUserName = prodUserNamePipeline
0080
0081
0082 # site = sys.argv[1]
0083 # site = 'ANALY_BNL-LSST'  #orig
0084 # site = 'BNL-LSST'
0085 # site = 'SWT2_CPB-LSST'
0086 # site = 'UTA_SWT2-LSST'
0087 # site = 'ANALY_SWT2_CPB-LSST'
0088
0089 destName = None
0090
0091 if prodUserName is not None and PIPELINE_TASK is not None and PIPELINE_PROCESSINSTANCE is not None:
0092     datasetName = f"panda.lsst.user.{str(PIPELINE_PROCESSINSTANCE)}.{str(PIPELINE_TASK)}.{str(prodUserName)}"
0093 else:
0094     datasetName = f"panda.lsst.user.jschovan.{str(uuid.uuid4())}"
0095
0096 if prodUserName is not None and PIPELINE_TASK is not None and PIPELINE_EXECUTIONNUMBER is not None and PIPELINE_STREAM is not None:
0097     jobName = f"job.{str(PIPELINE_PROCESSINSTANCE)}.{str(PIPELINE_TASK)}.{str(PIPELINE_EXECUTIONNUMBER)}.{str(prodUserName)}.{str(PIPELINE_STREAM)}"
0098 else:
0099     jobName = f"{str(uuid.uuid4())}"
0100
0101 if PIPELINE_STREAM is not None:
0102     jobDefinitionID = PIPELINE_STREAM
0103 else:
0104     jobDefinitionID = int(time.time()) % 10000
0105 job = JobSpec()
0106 job.jobDefinitionID = jobDefinitionID
0107 job.jobName = jobName
0108 job.transformation = "http://pandawms.org/pandawms-jobcache/lsst-trf.sh"
0109 job.destinationDBlock = datasetName
0110 job.destinationSE = "local"
0111 job.currentPriority = 1000
0112 job.prodSourceLabel = "panda"
0113 job.jobParameters = f' --lsstJobParams="{lsstJobParams}" '
0114 if prodUserName is not None:
0115     job.prodUserName = prodUserName
0116 else:
0117     job.prodUserName = prodUserNameDefault
0118 if PIPELINE_PROCESSINSTANCE is not None:
0119     job.taskID = PIPELINE_PROCESSINSTANCE
0120 if PIPELINE_EXECUTIONNUMBER is not None:
0121     job.attemptNr = PIPELINE_EXECUTIONNUMBER
0122 if PIPELINE_TASK is not None:
0123     job.processingType = PIPELINE_TASK
0124 job.computingSite = site
0125 job.VO = "lsst"
0126
0127 fileOL = FileSpec()
0128 fileOL.lfn = f"{job.jobName}.job.log.tgz"
0129 fileOL.destinationDBlock = job.destinationDBlock
0130 fileOL.destinationSE = job.destinationSE
0131 fileOL.dataset = job.destinationDBlock
0132 fileOL.type = "log"
0133 job.addFile(fileOL)
0134
0135
0136 s, o = Client.submit_jobs([job])
0137 print(s)
0138 for x in o:
0139     print(f"PandaID={x[0]}")