Add metadata to rucio registration

fixedTarget/batch/README.md

@@ -1,3 +1,16 @@
+# Production submission
+
+1. Make sure the FairShip installation in `bashScript.sh` points to the version you want
+
+2. Make sure the MassStorageFile location in `gangaScript.py` points to wherever you want the data to go
+
+3. Make sure you have created a voms proxy for the file registration
+
+4. Use the gangaSubmit.sh script to submit the job. This is important - you need to use the ship version of ganga on CVMFS (not latest)
+
+5. To monitor your jobs use this ganga: /cvmfs/ganga.cern.ch/Ganga/install/ship/bin/ganga . Don't forget to have a valid voms proxy!
+
+
 # Batch submission
 
 1. Download the FairShip software:

fixedTarget/batch/bashScript.sh

@@ -1,18 +1,15 @@
 #!/bin/bash
 
-FS_INSTALL=/user/kskovpen/analysis/SHiP/FairShip
+FS_INSTALL=/cvmfs/ship.cern.ch/25.11/
 
-source /cvmfs/ship.cern.ch/24.10/setUp.sh
+export WORK_DIR=${FS_INSTALL}/sw/
+source ${FS_INSTALL}/sw/slc9_x86-64/FairShip/latest/etc/profile.d/init.sh
 
-export ALIBUILD_WORK_DIR=${FS_INSTALL}/sw
+echo "INFO: Environment set up for FairShip located at " $FS_INSTALL
 
-source ${FS_INSTALL}/../htcondor_submission_scripts/fixedTarget/batch/test_config.sh
+echo "INFO: Executing: python ${FS_INSTALL}/sw/slc9_x86-64/FairShip/latest/muonShieldOptimization/run_fixedTarget.py" $@
 
-echo "INFO: Environment set up for FairShip located at " $FAIRSHIP
-
-echo "INFO: Executing: python ${FAIRSHIP_INSTALL}/muonShieldOptimization/run_fixedTarget.py" $@
-
-python ${FS_INSTALL}/muonShieldOptimization/run_fixedTarget.py $@
+python ${FS_INSTALL}/sw/slc9_x86-64/FairShip/latest/muonShieldOptimization/run_fixedTarget.py $@
 
 echo "INFO: Finished running. These files are on the WN:"
 ls -lh

fixedTarget/batch/gangaScript.py

@@ -1,20 +1,37 @@
 import time
 import os
 import random
+
+# Set this path to wherever you want the output to go
+#config['Output']['MassStorageFile']['uploadOptions']['path'] = '/eos/experiment/ship/test/masmith_test/'
+config['Output']['MassStorageFile']['uploadOptions']['defaultProtocol'] = 'root://eospublic.cern.ch'
+
+# Now set up the random seed, how many events per subjobs and how many events total
 random.seed(os.environ.get("USER"))
 startRun = int(time.time()) + random.randint(0,10000)
 evtsPerJob = 10 #100000
-evtsToGen = 100
+evtsToGen = 20
 nSJ = int(evtsToGen/evtsPerJob)
 
-
 j = Job(name = f'run fixed target production - {evtsToGen} events')
 j.application = Executable(exe = File('bashScript.sh'), args = ['-o', '"./"', '-n', evtsPerJob])
+
+# IMPORTANT: Only put the run seed in the splitter arguments
 j.splitter = ArgSplitter(args = [['-r', startRun + _i] for _i in range(nSJ)], append = True)
-j.outputfiles = [LocalFile('*.root')]
+j.outputfiles = [MassStorageFile('pythia8_evtgen_Geant4_*.root')]
+#j.backend = Local()
 j.backend = Condor()
-j.backend.cdf_options['+MaxRuntime'] = '1000'
-cc = CustomChecker(moduel = 'postprocessor.py')
+#j.backend.cdf_options['+MaxRuntime'] = '2000'
+j.backend.cdf_options['+JobFlavour'] = '"longlunch"'
+
+# For running at CERN only
+j.backend.cdf_options['accounting_group'] = 'group_u_SHIP.u_ship_cg'
+
+# Add in the postprocessor to do the file registration
+cc = CustomChecker(module = 'postprocessor.py')
+#cc = CustomChecker(module = 'postprocessor_master.py')
+fc = FileChecker(files = ['stdout'], searchStrings = ['Macro finished successfully.'], failIfFound = False)
+j.postprocessors.append(fc)
 j.postprocessors.append(cc)
 j.comment = f'{evtsPerJob} events in each of {nSJ} subjobs'
 j.submit()

fixedTarget/batch/gangaSubmit.sh

@@ -1,4 +1,4 @@
 #!/bin/bash
 
-/cvmfs/ganga.cern.ch/runGanga.sh gangaScript.py
+/cvmfs/ganga.cern.ch/Ganga/install/ship/bin/ganga gangaScript.py
 

fixedTarget/batch/postprocessor.py

@@ -1,20 +1,62 @@
+from GangaCore.GPIDev.Lib.File.MassStorageFile import MassStorageFile
 import rucio_it_tools.rucio_it_register
+import os
+import uproot
+
+# This is a config file set up for SHiP
+if 'RUCIO_CONFIG' not in os.environ:
+    os.environ['RUCIO_CONFIG'] = '/cvmfs/ganga.cern.ch/Ganga/install/ship/rucio/etc/rucio.cfg'
 
 def check(j):
+    # We don't want to run on the master job 
+    if j.subjobs:
+        print("this is a master job")
+        return True
     file_list = []
     for _f in j.outputfiles:
         if isinstance(_f, MassStorageFile):
-            _loc = _f.location()
-            _size = rucio_it_tools.rucio_it_register.get_file_on_disk_size_in_bytes(_f)
-            _checksum = rucio_it_tools.rucio_it_register.get_file_on_disk_size_in_bytes(_f)
+            _loc = _f.locations[0]
+            # Check you can open it
+#            try:
+#                _tfile = uproot.open(_loc)
+#                _tfile["cbmsim"]
+#                _tfile.close()
+#            except uproot.deserialization.DeserializationError:
+#                print(f"ERROR: Unable to open {_loc}! Job likely failed!")
+#                _tfile.close()
+#                return False
+            _size = rucio_it_tools.rucio_it_register.get_file_on_disk_size_in_bytes(_loc)
+            _checksum = rucio_it_tools.rucio_it_register.get_file_on_disk_adler32_checksum(_loc)
+#                print(_loc,' - ', _size, ' - ', _checksum)
             file_list.append({
                 "path": _loc,
                 "size": _size,
                 "adler32": _checksum
             })
+    metadata = {
+            "title" : j.name,
+            "ganga_id": str(j.fqid),
+            "completion_time" : str(j.time.backend_final()),
+            "job_args" : str([_a for _a in j.application.args]),
+            "run_nos" : str((j.fqid, j.application.args[-1])),
+            "n_jobs" : str(len(j.master.subjobs)),
+            "creator": os.environ.get("USER"),
+            "comment": j.comment
+            }
+#    print(f"INFO: File list - {file_list}")
+#    print(f"INFO: metadata - {metadata}")
 
-    rucio_it_tools.rucio_it_register.register_files_with_structure(
-        rse_name = "SHIP_TIER_0_DISK",
-        files = file_list
-    )
+    if len(file_list)>0:
+        try:
+            rucio_it_tools.rucio_it_register.register_files_with_structure(
+                rse_name = "SHIP_TIER_0_DISK",
+                files = file_list,
+                metadata = metadata,
+                dry_run = True # Comment this when you run it for real
+            )
+        except Exception as e:
+            print(f"ERROR: Not able to register file {file_list} with rucio: {e}")
+            j.force_status("failed")
+            j.comment += " - Rucio registration failed"
+            return False
     return True

fixedTarget/batch/postprocessor_master.py

@@ -0,0 +1,64 @@
+from GangaCore.GPIDev.Lib.File.MassStorageFile import MassStorageFile
+import rucio_it_tools.rucio_it_register
+import os
+
+# This is a config file set up for SHiP
+if 'RUCIO_CONFIG' not in os.environ:
+    os.environ['RUCIO_CONFIG'] = '/cvmfs/ganga.cern.ch/Ganga/install/ship/rucio/etc/rucio.cfg'
+
+def check(j):
+    # Only do this on the master job
+    if j.master:
+        return True
+    file_list = []
+    for _sj in j.subjobs:
+        if not _sj.status in ['completed', 'completing']:
+            print(f"WARNING: Subjobs {_sj.id} did not complete")
+            continue
+        for _f in _sj.outputfiles:
+            if isinstance(_f, MassStorageFile):
+                _loc = _f.locations[0]
+                # Check you can open it
+#               try:
+#                    _tfile = uproot.open(_loc)
+#                    _tfile.values()
+#                    _tfile.close()
+#                except uproot.deserialization.DeserializationError:
+#                    print(f"ERROR: Unable to open {_loc} from job {_sj.fqid}! Job likely failed!")
+#                    _tfile.close()
+#                    _sj.force_status('failed')
+                _size = rucio_it_tools.rucio_it_register.get_file_on_disk_size_in_bytes(_loc)
+                _checksum = rucio_it_tools.rucio_it_register.get_file_on_disk_adler32_checksum(_loc)
+#                print(_loc,' - ', _size, ' - ', _checksum)
+                file_list.append({
+                    "path": _loc,
+                    "size": _size,
+                    "adler32": _checksum
+                })
+    metadata = {
+                "title" : j.name,
+                "ganga_id": str(j.id),
+                "completion_time" : str(j.time.backend_final()),
+                "job_args" : str([_a for _a in j.application.args]),
+                "run_nos" : str([(_sj.id, _sj.application.args[-1]) for _sj in j.subjobs if _sj.status=='completed']),
+                "n_jobs" : str(len(j.subjobs)),
+                "creator": os.environ.get("USER"),
+                "comment": j.comment
+               }
+#    print(f"INFO: File list - {file_list}")
+#    print(f"INFO: metadata - {metadata}")
+
+    if len(file_list)>0:
+        try:
+            rucio_it_tools.rucio_it_register.register_files_with_structure(
+                rse_name = "SHIP_TIER_0_DISK",
+                files = file_list,
+                metadata = metadata,
+    #            dry_run = True
+            )
+        except Exception as e:
+            print(f"ERROR: Not able to register file {file_list} with rucio: {e}")
+            j.force_status("failed")
+            j.comment += " - Rucio registration failed"
+            return False        
+    return True