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.7 by fanzago, Tue Dec 18 10:58:20 2007 UTC vs.
Revision 1.35 by fanzago, Tue Jun 16 17:07:14 2009 UTC

# Line 3 | Line 3 | import common
3   import time, glob
4   from Actor import *
5   from crab_util import *
6 – from crab_logger import Logger
6   from crab_exceptions import *
7 < from FwkJobRep.ReportParser import readJobReport
7 > from ProdCommon.FwkJobRep.ReportParser import readJobReport
8 > from ProdCommon.FwkJobRep.ReportState import checkSuccess
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
12   from ProdCommon.DataMgmt.DBS.DBSReader import DBSReader
13   from ProdCommon.DataMgmt.DBS.DBSWriter import DBSWriter,DBSWriterObjects
14   import sys
15 + from DBSAPI.dbsApiException import DbsException
16 + from DBSAPI.dbsApi import DbsApi
17  
18   class Publisher(Actor):
19      def __init__(self, cfg_params):
# Line 23 | Line 25 | class Publisher(Actor):
25          - publishes output data on DBS and DLS
26          """
27  
28 <        try:
29 <            self.processedData = cfg_params['USER.publish_data_name']
30 <        except KeyError:
28 >        self.cfg_params=cfg_params
29 >      
30 >        if not cfg_params.has_key('USER.publish_data_name'):
31              raise CrabException('Cannot publish output data, because you did not specify USER.publish_data_name parameter in the crab.cfg file')
32 <        try:
33 <            if (int(cfg_params['USER.copy_data']) != 1): raise KeyError
34 <        except KeyError:
35 <            raise CrabException('You can not publish data because you did not selected *** copy_data = 1  *** in the crab.cfg file')
36 <        try:
37 <            self.pset = cfg_params['CMSSW.pset']
38 <        except KeyError:
32 >        self.userprocessedData = cfg_params['USER.publish_data_name']
33 >        self.processedData = None
34 >
35 >        if (not cfg_params.has_key('USER.copy_data') or int(cfg_params['USER.copy_data']) != 1 ) or \
36 >            (not cfg_params.has_key('USER.publish_data') or int(cfg_params['USER.publish_data']) != 1 ):
37 >            msg  = 'You can not publish data because you did not selected \n'
38 >            msg += '\t*** copy_data = 1 and publish_data = 1  *** in the crab.cfg file'
39 >            raise CrabException(msg)
40 >
41 >        if not cfg_params.has_key('CMSSW.pset'):
42              raise CrabException('Cannot publish output data, because you did not specify the psetname in [CMSSW] of your crab.cfg file')
43 <        try:
44 <            self.globalDBS=cfg_params['CMSSW.dbs_url']
45 <        except KeyError:
46 <            self.globalDBS="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
47 <        try:
48 <            self.DBSURL=cfg_params['USER.dbs_url_for_publication']
49 <            common.logger.message('<dbs_url_for_publication> = '+self.DBSURL)
50 <            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"):
51 <                msg = "You can not publish your data in the globalDBS = " + self.DBSURL + "\n"
52 <                msg = msg + "Please write your local one in the [USER] section 'dbs_url_for_publication'"
53 <                raise CrabException(msg)
54 <        except KeyError:
55 <            msg = "Error. The [USER] section does not have 'dbs_url_for_publication'"
56 <            msg = msg + " entry, necessary to publish the data"
43 >        self.pset = cfg_params['CMSSW.pset']
44 >
45 >        self.globalDBS=cfg_params.get('CMSSW.dbs_url',"http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet")
46 >
47 >        if not cfg_params.has_key('USER.dbs_url_for_publication'):
48 >            msg = "Warning. The [USER] section does not have 'dbs_url_for_publication'"
49 >            msg = msg + " entry, necessary to publish the data.\n"
50 >            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."
51 >            raise CrabException(msg)
52 >
53 >        self.DBSURL=cfg_params['USER.dbs_url_for_publication']
54 >        common.logger.info('<dbs_url_for_publication> = '+self.DBSURL)
55 >        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"):
56 >            msg = "You can not publish your data in the globalDBS = " + self.DBSURL + "\n"
57 >            msg = msg + "Please write your local one in the [USER] section 'dbs_url_for_publication'"
58              raise CrabException(msg)
59              
60          self.content=file(self.pset).read()
61          self.resDir = common.work_space.resDir()
62 +        
63 +        self.dataset_to_import=[]
64 +        
65          self.datasetpath=cfg_params['CMSSW.datasetpath']
66 +        if (self.datasetpath.upper() != 'NONE'):
67 +            self.dataset_to_import.append(self.datasetpath)
68 +        
69 +        ### Added PU dataset
70 +        tmp = cfg_params.get('CMSSW.dataset_pu',None)
71 +        if tmp :
72 +            datasets = tmp.split(',')
73 +            for dataset in datasets:
74 +                dataset=string.strip(dataset)
75 +                self.dataset_to_import.append(dataset)
76 +        ###        
77 +            
78 +        self.skipOcheck=cfg_params.get('CMSSW.publish_zero_event',0)
79 +    
80          self.SEName=''
81          self.CMSSW_VERSION=''
82          self.exit_status=''
# Line 64 | Line 87 | class Publisher(Actor):
87          
88      def importParentDataset(self,globalDBS, datasetpath):
89          """
90 <        """
90 >        """
91 >        print " patch for importParentDataset: datasetpath = ", datasetpath
92 >        """
93          dbsWriter = DBSWriter(self.DBSURL,level='ERROR')
94          
95          try:
96 <            dbsWriter.importDataset(globalDBS, self.datasetpath, self.DBSURL)
96 >            #dbsWriter.importDatasetWithoutParentage(globalDBS, datasetpath, self.DBSURL)
97 >            dbsWriter.importDataset(globalDBS, datasetpath, self.DBSURL)
98          except DBSWriterError, ex:
99              msg = "Error importing dataset to be processed into local DBS\n"
100              msg += "Source Dataset: %s\n" % datasetpath
101              msg += "Source DBS: %s\n" % globalDBS
102              msg += "Destination DBS: %s\n" % self.DBSURL
103 <            common.logger.message(msg)
103 >            common.logger.info(msg)
104 >            common.logger.debug(str(ex))
105 >            return 1
106 >        return 0
107 >        """
108 >        try:
109 >            args={}
110 >            args['url']=self.DBSURL
111 >            args['mode']='POST'
112 >            block = ""
113 >            api = DbsApi(args)
114 >            #api.migrateDatasetContents(srcURL, dstURL, path, block , False)
115 >            api.migrateDatasetContents(globalDBS, self.DBSURL, datasetpath, block , False)
116 >
117 >        except DbsException, ex:
118 >            print "Caught API Exception %s: %s "  % (ex.getClassName(), ex.getErrorMessage() )
119 >            if ex.getErrorCode() not in (None, ""):
120 >                print "DBS Exception Error Code: ", ex.getErrorCode()
121              return 1
122 +        print "Done"
123          return 0
124            
125      def publishDataset(self,file):
# Line 87 | Line 131 | class Publisher(Actor):
131          except IndexError:
132              self.exit_status = '1'
133              msg = "Error: Problem with "+file+" file"  
134 <            common.logger.message(msg)
134 >            common.logger.info(msg)
135              return self.exit_status
136 <            
137 <        if (self.datasetpath != 'None'):
138 <            common.logger.message("--->>> Importing parent dataset in the dbs")
139 <            status_import=self.importParentDataset(self.globalDBS, self.datasetpath)
140 <            if (status_import == 1):
141 <                common.logger.message('Problem with parent import from the global DBS '+self.globalDBS+ 'to the local one '+self.DBSURL)
142 <                self.exit_status='1'
143 <                return self.exit_status
144 <            common.logger.message("Parent import ok")
136 >
137 >        if (len(self.dataset_to_import) != 0):
138 >           for dataset in self.dataset_to_import:
139 >               common.logger.info("--->>> Importing parent dataset in the dbs: " +dataset)
140 >               status_import=self.importParentDataset(self.globalDBS, dataset)
141 >               if (status_import == 1):
142 >                   common.logger.info('Problem with parent '+ dataset +' import from the global DBS '+self.globalDBS+ 'to the local one '+self.DBSURL)
143 >                   self.exit_status='1'
144 >                   return self.exit_status
145 >               else:    
146 >                   common.logger.info('Import ok of dataset '+dataset)
147              
148          #// DBS to contact
149          dbswriter = DBSWriter(self.DBSURL)                        
# Line 107 | Line 153 | class Publisher(Actor):
153          except IndexError:
154              self.exit_status = '1'
155              msg = "Error: No file to publish in xml file"+file+" file"  
156 <            common.logger.message(msg)
156 >            common.logger.info(msg)
157              return self.exit_status
158  
159          datasets=fileinfo.dataset
160 <        common.logger.debug(6,"FileInfo = " + str(fileinfo))
161 <        common.logger.debug(6,"DatasetInfo = " + str(datasets))
160 >        common.logger.log(10-1,"FileInfo = " + str(fileinfo))
161 >        common.logger.log(10-1,"DatasetInfo = " + str(datasets))
162 >        if len(datasets)<=0:
163 >           self.exit_status = '1'
164 >           msg = "Error: No info about dataset in the xml file "+file
165 >           common.logger.info(msg)
166 >           return self.exit_status
167          for dataset in datasets:
168              #### for production data
169 +            self.processedData = dataset['ProcessedDataset']
170              if (dataset['PrimaryDataset'] == 'null'):
171 <                dataset['PrimaryDataset'] = dataset['ProcessedDataset']
172 <                
171 >                #dataset['PrimaryDataset'] = dataset['ProcessedDataset']
172 >                dataset['PrimaryDataset'] = self.userprocessedData
173 >            #else: # add parentage from input dataset
174 >            elif self.datasetpath.upper() != 'NONE':
175 >                dataset['ParentDataset']= self.datasetpath
176 >    
177              dataset['PSetContent']=self.content
178              cfgMeta = {'name' : self.pset , 'Type' : 'user' , 'annotation': 'user cfg', 'version' : 'private version'} # add real name of user cfg
179 <            common.logger.message("PrimaryDataset = %s"%dataset['PrimaryDataset'])
180 <            common.logger.message("ProcessedDataset = %s"%dataset['ProcessedDataset'])
181 <            common.logger.message("<User Dataset Name> = /"+dataset['PrimaryDataset']+"/"+dataset['ProcessedDataset']+"/USER")
179 >            common.logger.info("PrimaryDataset = %s"%dataset['PrimaryDataset'])
180 >            common.logger.info("ProcessedDataset = %s"%dataset['ProcessedDataset'])
181 >            common.logger.info("<User Dataset Name> = /"+dataset['PrimaryDataset']+"/"+dataset['ProcessedDataset']+"/USER")
182 >            self.dataset_to_check="/"+dataset['PrimaryDataset']+"/"+dataset['ProcessedDataset']+"/USER"
183              
184 <            common.logger.debug(6,"--->>> Inserting primary: %s processed : %s"%(dataset['PrimaryDataset'],dataset['ProcessedDataset']))
184 >            common.logger.log(10-1,"--->>> Inserting primary: %s processed : %s"%(dataset['PrimaryDataset'],dataset['ProcessedDataset']))
185              
186              primary = DBSWriterObjects.createPrimaryDataset( dataset, dbswriter.dbs)
187 <            common.logger.debug(6,"Primary:  %s "%primary)
187 >            common.logger.log(10-1,"Primary:  %s "%primary)
188              
189              algo = DBSWriterObjects.createAlgorithm(dataset, cfgMeta, dbswriter.dbs)
190 <            common.logger.debug(6,"Algo:  %s "%algo)
190 >            common.logger.log(10-1,"Algo:  %s "%algo)
191  
192              processed = DBSWriterObjects.createProcessedDataset(primary, algo, dataset, dbswriter.dbs)
193 <            common.logger.debug(6,"Processed:  %s "%processed)
193 >            common.logger.log(10-1,"Processed:  %s "%processed)
194              
195 <            common.logger.debug(6,"Inserted primary %s processed %s"%(primary,processed))
195 >            common.logger.log(10-1,"Inserted primary %s processed %s"%(primary,processed))
196              
197 <        common.logger.debug(6,"exit_status = %s "%self.exit_status)
197 >        common.logger.log(10-1,"exit_status = %s "%self.exit_status)
198          return self.exit_status    
199  
200      def publishAJobReport(self,file,procdataset):
# Line 162 | Line 219 | class Publisher(Actor):
219              elif (file['LFN'] == ''):
220                  self.noLFN.append(file['PFN'])
221              else:
222 <                if int(file['TotalEvents']) != 0 :
223 <                    file.lumisections = {}
222 >                if  self.skipOcheck==0:
223 >                    if int(file['TotalEvents']) != 0:
224 >                        #file.lumisections = {}
225 >                        # lumi info are now in run hash
226 >                        file.runs = {}
227 >                        for ds in file.dataset:
228 >                            ### Fede for production
229 >                            if (ds['PrimaryDataset'] == 'null'):
230 >                                #ds['PrimaryDataset']=procdataset
231 >                                ds['PrimaryDataset']=self.userprocessedData
232 >                        filestopublish.append(file)
233 >                    else:
234 >                        self.noEventsFiles.append(file['LFN'])
235 >                else:
236 >                    file.runs = {}
237                      for ds in file.dataset:
168 –                        ds['ProcessedDataset']=procdataset
238                          ### Fede for production
239                          if (ds['PrimaryDataset'] == 'null'):
240 <                            ds['PrimaryDataset']=procdataset
240 >                            #ds['PrimaryDataset']=procdataset
241 >                            ds['PrimaryDataset']=self.userprocessedData
242                      filestopublish.append(file)
243 <                else:
174 <                    self.noEventsFiles.append(file['LFN'])
243 >      
244          jobReport.files = filestopublish
245          ### if all files of FJR have number of events = 0
246          if (len(filestopublish) == 0):
# Line 183 | Line 252 | class Publisher(Actor):
252          Blocks=None
253          try:
254              Blocks=dbswriter.insertFiles(jobReport)
255 <            common.logger.message("Blocks = %s"%Blocks)
255 >            common.logger.info("Inserting file in blocks = %s"%Blocks)
256          except DBSWriterError, ex:
257 <            common.logger.message("Insert file error: %s"%ex)
257 >            common.logger.info("Insert file error: %s"%ex)
258          return Blocks
259  
260      def run(self):
# Line 194 | Line 263 | class Publisher(Actor):
263          """
264          
265          file_list = glob.glob(self.resDir+"crab_fjr*.xml")
266 <        common.logger.debug(6, "file_list = "+str(file_list))
267 <        common.logger.debug(6, "len(file_list) = "+str(len(file_list)))
266 >        ## Select only those fjr that are succesfull
267 >        good_list=[]
268 >        for fjr in file_list:
269 >            reports = readJobReport(fjr)
270 >            if len(reports)>0:
271 >               if reports[0].status == "Success":
272 >                  good_list.append(fjr)
273 >        file_list=good_list
274 >        ##
275 >        common.logger.log(10-1, "file_list = "+str(file_list))
276 >        common.logger.log(10-1, "len(file_list) = "+str(len(file_list)))
277              
278          if (len(file_list)>0):
279              BlocksList=[]
280 <            common.logger.message("--->>> Start dataset publication")
280 >            common.logger.info("--->>> Start dataset publication")
281              self.exit_status=self.publishDataset(file_list[0])
282              if (self.exit_status == '1'):
283                  return self.exit_status
284 <            common.logger.message("--->>> End dataset publication")
284 >            common.logger.info("--->>> End dataset publication")
285  
286  
287 <            common.logger.message("--->>> Start files publication")
287 >            common.logger.info("--->>> Start files publication")
288              for file in file_list:
289 <                common.logger.message("file = "+file)
289 >                common.logger.debug( "file = "+file)
290                  Blocks=self.publishAJobReport(file,self.processedData)
291                  if Blocks:
292 <                    [BlocksList.append(x) for x in Blocks]
292 >                    for x in Blocks: # do not allow multiple entries of the same block
293 >                        if x not in BlocksList:
294 >                           BlocksList.append(x)
295                      
296              # close the blocks
297 <            common.logger.debug(6, "BlocksList = %s"%BlocksList)
297 >            common.logger.log(10-1, "BlocksList = %s"%BlocksList)
298              # dbswriter = DBSWriter(self.DBSURL,level='ERROR')
299              dbswriter = DBSWriter(self.DBSURL)
300              
301              for BlockName in BlocksList:
302                  try:  
303                      closeBlock=dbswriter.manageFileBlock(BlockName,maxFiles= 1)
304 <                    common.logger.debug(6, "closeBlock %s"%closeBlock)
304 >                    common.logger.log(10-1, "closeBlock %s"%closeBlock)
305                      #dbswriter.dbs.closeBlock(BlockName)
306                  except DBSWriterError, ex:
307 <                    common.logger.message("Close block error %s"%ex)
307 >                    common.logger.info("Close block error %s"%ex)
308  
309              if (len(self.noEventsFiles)>0):
310 <                common.logger.message("--->>> WARNING: "+str(len(self.noEventsFiles))+" files not published because they contain 0 events are:")
310 >                common.logger.info("--->>> WARNING: "+str(len(self.noEventsFiles))+" files not published because they contain 0 events are:")
311                  for lfn in self.noEventsFiles:
312 <                    common.logger.message("------ LFN: %s"%lfn)
312 >                    common.logger.info("------ LFN: %s"%lfn)
313              if (len(self.noLFN)>0):
314 <                common.logger.message("--->>> WARNING: there are "+str(len(self.noLFN))+" files not published because they have empty LFN")
314 >                common.logger.info("--->>> WARNING: there are "+str(len(self.noLFN))+" files not published because they have empty LFN")
315                  for pfn in self.noLFN:
316 <                    common.logger.message("------ pfn: %s"%pfn)
316 >                    common.logger.info("------ pfn: %s"%pfn)
317              if (len(self.problemFiles)>0):
318 <                common.logger.message("--->>> WARNING: "+str(len(self.problemFiles))+" files not published because they had problem with copy to SE")
318 >                common.logger.info("--->>> WARNING: "+str(len(self.problemFiles))+" files not published because they had problem with copy to SE")
319                  for lfn in self.problemFiles:
320 <                    common.logger.message("------ LFN: %s"%lfn)
321 <            common.logger.message("--->>> End files publication")
322 <            common.logger.message("--->>> To check data publication please use: InspectDBS2.py --DBSURL=<dbs_url_for_publication> --datasetPath=<User Dataset Name>")
320 >                    common.logger.info("------ LFN: %s"%lfn)
321 >            common.logger.info("--->>> End files publication")
322 >          
323 >            self.cfg_params['USER.dataset_to_check']=self.dataset_to_check
324 >            from InspectDBS import InspectDBS
325 >            check=InspectDBS(self.cfg_params)
326 >            check.checkPublication()
327              return self.exit_status
328  
329          else:
330 <            common.logger.message("--->>> "+self.resDir+" empty: no file to publish on DBS")
330 >            common.logger.info("--->>> "+self.resDir+" empty: no file to publish on DBS")
331              self.exit_status = '1'
332              return self.exit_status
333      

Diff Legend

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