ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/Publisher.py
(Generate patch)

Comparing COMP/CRAB/python/Publisher.py (file contents):
Revision 1.5 by fanzago, Fri Dec 7 13:26:14 2007 UTC vs.
Revision 1.17 by afanfani, Wed Oct 15 09:28:21 2008 UTC

# Line 5 | Line 5 | from Actor import *
5   from crab_util import *
6   from crab_logger import Logger
7   from crab_exceptions import *
8 < from FwkJobRep.ReportParser import readJobReport
8 > from ProdCommon.FwkJobRep.ReportParser import readJobReport
9   from ProdCommon.MCPayloads.WorkflowSpec import WorkflowSpec
10   from ProdCommon.DataMgmt.DBS.DBSWriter import DBSWriter
11   from ProdCommon.DataMgmt.DBS.DBSErrors import DBSWriterError, formatEx,DBSReaderError
# Line 24 | Line 24 | class Publisher(Actor):
24          """
25  
26          try:
27 <            self.processedData = cfg_params['USER.publish_data_name']
27 >            userprocessedData = cfg_params['USER.publish_data_name']
28 >            self.processedData = None
29          except KeyError:
30              raise CrabException('Cannot publish output data, because you did not specify USER.publish_data_name parameter in the crab.cfg file')
31 +
32          try:
33              if (int(cfg_params['USER.copy_data']) != 1): raise KeyError
34          except KeyError:
# Line 41 | Line 43 | class Publisher(Actor):
43              self.globalDBS="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
44          try:
45              self.DBSURL=cfg_params['USER.dbs_url_for_publication']
46 <            common.logger.message('dbs url = '+self.DBSURL)
47 <            if (self.DBSURL == "http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"):
46 >            common.logger.message('<dbs_url_for_publication> = '+self.DBSURL)
47 >            if (self.DBSURL == "http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet") or (self.DBSURL == "https://cmsdbsprod.cern.ch:8443/cms_dbs_prod_global_writer/servlet/DBSServlet"):
48                  msg = "You can not publish your data in the globalDBS = " + self.DBSURL + "\n"
49                  msg = msg + "Please write your local one in the [USER] section 'dbs_url_for_publication'"
50                  raise CrabException(msg)
51          except KeyError:
52 <            msg = "Error. The [USER] section does not have 'dbs_url_for_publication'"
53 <            msg = msg + " entry, necessary to publish the data"
52 >            msg = "Warning. The [USER] section does not have 'dbs_url_for_publication'"
53 >            msg = msg + " entry, necessary to publish the data.\n"
54 >            msg = msg + "Use the command **crab -publish -USER.dbs_url_for_publication=dbs_url_for_publication*** \nwhere dbs_url_for_publication is your local dbs instance."
55              raise CrabException(msg)
56              
57          self.content=file(self.pset).read()
58          self.resDir = common.work_space.resDir()
59 +        
60 +        self.dataset_to_import=[]
61 +        
62          self.datasetpath=cfg_params['CMSSW.datasetpath']
63 +        if (self.datasetpath.upper() != 'NONE'):
64 +            self.dataset_to_import.append(self.datasetpath)
65 +        
66 +        ### Added PU dataset
67 +        tmp = cfg_params.get('CMSSW.dataset_pu',None)
68 +        if tmp :
69 +            datasets = tmp.split(',')
70 +            for dataset in datasets:
71 +                dataset=string.strip(dataset)
72 +                self.dataset_to_import.append(dataset)
73 +        ###        
74 +                
75          self.SEName=''
76          self.CMSSW_VERSION=''
77          self.exit_status=''
# Line 68 | Line 86 | class Publisher(Actor):
86          dbsWriter = DBSWriter(self.DBSURL,level='ERROR')
87          
88          try:
89 <            dbsWriter.importDataset(globalDBS, self.datasetpath, self.DBSURL)
89 >            dbsWriter.importDatasetWithoutParentage(globalDBS, self.datasetpath, self.DBSURL)
90          except DBSWriterError, ex:
91              msg = "Error importing dataset to be processed into local DBS\n"
92              msg += "Source Dataset: %s\n" % datasetpath
# Line 89 | Line 107 | class Publisher(Actor):
107              msg = "Error: Problem with "+file+" file"  
108              common.logger.message(msg)
109              return self.exit_status
110 <            
111 <        if (self.datasetpath != 'None'):
112 <            common.logger.message("--->>> Importing parent dataset in the dbs")
113 <            status_import=self.importParentDataset(self.globalDBS, self.datasetpath)
114 <            if (status_import == 1):
115 <                common.logger.message('Problem with parent import from the global DBS '+self.globalDBS+ 'to the local one '+self.DBSURL)
116 <                self.exit_status='1'
117 <                return self.exit_status
118 <            common.logger.message("Parent import ok")
110 >
111 >        if (len(self.dataset_to_import) != 0):
112 >           for dataset in self.dataset_to_import:
113 >               common.logger.message("--->>> Importing parent dataset in the dbs: " +dataset)
114 >               status_import=self.importParentDataset(self.globalDBS, dataset)
115 >               if (status_import == 1):
116 >                   common.logger.message('Problem with parent '+ dataset +' import from the global DBS '+self.globalDBS+ 'to the local one '+self.DBSURL)
117 >                   self.exit_status='1'
118 >                   return self.exit_status
119 >               else:    
120 >                   common.logger.message('Import ok of dataset '+dataset)
121              
122          #// DBS to contact
123          dbswriter = DBSWriter(self.DBSURL)                        
# Line 114 | Line 134 | class Publisher(Actor):
134          common.logger.debug(6,"FileInfo = " + str(fileinfo))
135          common.logger.debug(6,"DatasetInfo = " + str(datasets))
136          for dataset in datasets:
137 +            #### for production data
138 +            self.processedData = dataset['ProcessedDataset']
139 +            if (dataset['PrimaryDataset'] == 'null'):
140 +                dataset['PrimaryDataset'] = dataset['ProcessedDataset']
141 +            else: # add parentage from input dataset
142 +                dataset['ParentDataset']= self.datasetpath
143 +    
144              dataset['PSetContent']=self.content
118 –            #cfgMeta = {'name' : 'usercfg' , 'Type' : 'user' , 'annotation': 'user cfg', 'version' : 'private version'} # add real name of user cfg
145              cfgMeta = {'name' : self.pset , 'Type' : 'user' , 'annotation': 'user cfg', 'version' : 'private version'} # add real name of user cfg
146              common.logger.message("PrimaryDataset = %s"%dataset['PrimaryDataset'])
147              common.logger.message("ProcessedDataset = %s"%dataset['ProcessedDataset'])
148 <            common.logger.message("--->>> Inserting primary: %s processed : %s"%(dataset['PrimaryDataset'],dataset['ProcessedDataset']))
148 >            common.logger.message("<User Dataset Name> = /"+dataset['PrimaryDataset']+"/"+dataset['ProcessedDataset']+"/USER")
149 >            
150 >            common.logger.debug(6,"--->>> Inserting primary: %s processed : %s"%(dataset['PrimaryDataset'],dataset['ProcessedDataset']))
151              
152              primary = DBSWriterObjects.createPrimaryDataset( dataset, dbswriter.dbs)
153              common.logger.debug(6,"Primary:  %s "%primary)
# Line 130 | Line 158 | class Publisher(Actor):
158              processed = DBSWriterObjects.createProcessedDataset(primary, algo, dataset, dbswriter.dbs)
159              common.logger.debug(6,"Processed:  %s "%processed)
160              
161 <            common.logger.message("Inserted primary %s processed %s"%(primary,processed))
161 >            common.logger.debug(6,"Inserted primary %s processed %s"%(primary,processed))
162              
163          common.logger.debug(6,"exit_status = %s "%self.exit_status)
164          return self.exit_status    
# Line 158 | Line 186 | class Publisher(Actor):
186                  self.noLFN.append(file['PFN'])
187              else:
188                  if int(file['TotalEvents']) != 0 :
189 <                    file.lumisections = {}
189 >                    #file.lumisections = {}
190 >                    # lumi info are now in run hash
191 >                    file.runs = {}
192                      for ds in file.dataset:
193 <                        ds['ProcessedDataset']=procdataset
193 >                        ### FEDE FOR NEW LFN ###
194 >                        #ds['ProcessedDataset']=procdataset
195 >                        ########################
196 >                        ### Fede for production
197 >                        if (ds['PrimaryDataset'] == 'null'):
198 >                            ds['PrimaryDataset']=procdataset
199                      filestopublish.append(file)
200                  else:
201                      self.noEventsFiles.append(file['LFN'])
# Line 203 | Line 238 | class Publisher(Actor):
238                  common.logger.message("file = "+file)
239                  Blocks=self.publishAJobReport(file,self.processedData)
240                  if Blocks:
241 <                    [BlocksList.append(x) for x in Blocks]
241 >                    for x in Blocks: # do not allow multiple entries of the same block
242 >                        if x not in BlocksList:
243 >                           BlocksList.append(x)
244                      
245              # close the blocks
246 <            common.logger.message("BlocksList = %s"%BlocksList)
246 >            common.logger.debug(6, "BlocksList = %s"%BlocksList)
247              # dbswriter = DBSWriter(self.DBSURL,level='ERROR')
248              dbswriter = DBSWriter(self.DBSURL)
249              
250              for BlockName in BlocksList:
251                  try:  
252                      closeBlock=dbswriter.manageFileBlock(BlockName,maxFiles= 1)
253 <                    common.logger.message("closeBlock %s"%closeBlock)
253 >                    common.logger.debug(6, "closeBlock %s"%closeBlock)
254                      #dbswriter.dbs.closeBlock(BlockName)
255                  except DBSWriterError, ex:
256                      common.logger.message("Close block error %s"%ex)
257  
221 –            common.logger.message("--->>> End files publication")
258              if (len(self.noEventsFiles)>0):
259                  common.logger.message("--->>> WARNING: "+str(len(self.noEventsFiles))+" files not published because they contain 0 events are:")
260                  for lfn in self.noEventsFiles:
# Line 231 | Line 267 | class Publisher(Actor):
267                  common.logger.message("--->>> WARNING: "+str(len(self.problemFiles))+" files not published because they had problem with copy to SE")
268                  for lfn in self.problemFiles:
269                      common.logger.message("------ LFN: %s"%lfn)
270 +            common.logger.message("--->>> End files publication")
271 +            common.logger.message("--->>> To check data publication please use: InspectDBS2.py --DBSURL=<dbs_url_for_publication> --datasetPath=<User Dataset Name>")
272              return self.exit_status
273  
274          else:

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines