Utilities/python/tier0.py

0001 from __future__ import print_function
0002
0003 #-toDo: move this to common?
0004
0005 import logging
0006 import json
0007 import os
0008 import sys
0009 import time
0010 import subprocess
0011
0012 import pycurl
0013
0014 tier0Url = 'https://cmsweb.cern.ch/t0wmadatasvc/prod/'
0015
0016 class Tier0Error(Exception):
0017     '''Tier0 exception.
0018     '''
0019
0020     def __init__(self, message):
0021         self.args = (message, )
0022
0023
0024 def unique(seq, keepstr=True):
0025     t = type(seq)
0026     if t in (unicode, str):
0027         t = (list, t('').join)[bool(keepstr)]
0028     try:
0029         remaining = set(seq)
0030         seen = set()
0031         return t(c for c in seq if (c in remaining and not remaining.remove(c)))
0032     except TypeError: # hashing didn't work, see if seq is sortable
0033         try:
0034             from itertools import groupby
0035             s = sorted(enumerate(seq),key=lambda i_v1:(i_v1[1],i_v1[0]))
0036             return t(next(g) for k,g in groupby(s, lambda i_v: i_v[1]))
0037         except:  # not sortable, use brute force
0038             seen = []
0039             return t(c for c in seq if not (c in seen or seen.append(c)))
0040
0041 class ResponseError( Tier0Error ):
0042
0043     def __init__( self, curl, response, proxy, timeout ):
0044         super( ResponseError, self ).__init__( response )
0045         self.args += ( curl, proxy )
0046         self.timeout = timeout
0047
0048     def __str__( self ):
0049         errStr = """Wrong response for curl connection to Tier0DataSvc from URL \"%s\"""" %( self.args[1].getinfo( self.args[1].EFFECTIVE_URL ), )
0050         if self.args[ -1 ]:
0051             errStr += """ using proxy \"%s\"""" %( str( self.args[ -1 ] ), )
0052         errStr += """ with timeout \"%d\" with error code \"%d\".""" %( self.timeout, self.args[1].getinfo( self.args[1].RESPONSE_CODE) )
0053         if self.args[0].find( '<p>' ) != -1:
0054             errStr += """\nFull response: \"%s\".""" %( self.args[0].partition('<p>')[-1].rpartition('</p>')[0], )
0055         else:
0056             errStr += """\nFull response: \"%s\".""" %( self.args[0], )
0057         return errStr
0058
0059 #TODO: Add exceptions for each category of HTTP error codes
0060 #TODO: check response code and raise corresponding exceptions
0061
0062 def _raise_http_error( curl, response, proxy, timeout ):
0063     raise ResponseError( curl, response, proxy, timeout )
0064
0065 class Tier0Handler( object ):
0066
0067     def __init__( self, uri, timeOut, retries, retryPeriod, proxy, debug ):
0068         """
0069         Parameters:
0070         uri: Tier0DataSvc URI;
0071         timeOut: time out for Tier0DataSvc HTTPS calls;
0072         retries: maximum retries for Tier0DataSvc HTTPS calls;
0073         retryPeriod: sleep time between two Tier0DataSvc HTTPS calls;
0074         proxy: HTTP proxy for accessing Tier0DataSvc HTTPS calls;
0075         debug: if set to True, enables debug information.
0076         """
0077         self._uri = uri
0078         self._timeOut = timeOut
0079         self._retries = retries
0080         self._retryPeriod = retryPeriod
0081         self._proxy = proxy
0082         self._debug = debug
0083
0084     def setDebug( self ):
0085         self._debug = True
0086
0087     def unsetDebug( self ):
0088         self._debug = False
0089
0090     def setProxy( self, proxy ):
0091         self._proxy = proxy
0092
0093     def _queryTier0DataSvc( self, url ):
0094         """
0095         Queries Tier0DataSvc.
0096         url: Tier0DataSvc URL.
0097         @returns: dictionary, from whence the required information must be retrieved according to the API call.
0098         Raises if connection error, bad response, or timeout after retries occur.
0099         """
0100
0101         userAgent = "User-Agent: ConditionWebServices/1.0 python/%d.%d.%d PycURL/%s" % ( sys.version_info[ :3 ] + ( pycurl.version_info()[ 1 ], ) )
0102
0103         proxy = ""
0104         if self._proxy: proxy = ' --proxy=%s ' % self._proxy
0105
0106         debug = " -s -S "
0107         if self._debug: debug = " -v "
0108
0109         cmd = '/usr/bin/curl -k -L --user-agent "%s" %s --connect-timeout %i --retry %i %s %s ' % (userAgent, proxy, self._timeOut, self._retries, debug, url)
0110
0111         # time the curl to understand if re-tries have been carried out
0112         start = time.time()
0113         process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
0114         (stdoutdata, stderrdata) =  process.communicate()
0115         retcode = process.returncode
0116         end = time.time()
0117
0118         if retcode != 0 or stderrdata:
0119
0120            # if the first curl has failed, logg its stderror and prepare and independent retry
0121            msg = "looks like curl returned an error: retcode=%s and took %s seconds" % (retcode,(end-start),)
0122            msg += ' msg = "'+str(stderrdata)+'"'
0123            logging.error(msg)
0124
0125            time.sleep(10)
0126            cmd = '/usr/bin/curl -k -L --user-agent "%s" %s --connect-timeout %i --retry %i %s %s ' % (userAgent, proxy, self._timeOut, self._retries, "-v", url)
0127            process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
0128            (stdoutdata, stderrdata) =  process.communicate()
0129            retcode = process.returncode
0130            if retcode != 0:
0131               msg = "looks like curl returned an error for the second time: retcode=%s" % (retcode,)
0132               msg += ' msg = "'+str(stderrdata)+'"'
0133               logging.error(msg)
0134               raise Tier0Error(msg)
0135            else :
0136               msg = "curl returned ok upon the second try"
0137               logging.info(msg)
0138         resp = json.loads( ''.join(stdoutdata.decode()).replace( "'", '"').replace(' None', ' "None"') )
0139         return resp
0140
0141
0142     def getFirstSafeRun( self ):
0143         """
0144         Queries Tier0DataSvc to get the first condition safe run.
0145         Parameters:
0146         @returns: integer, the run number.
0147         Raises if connection error, bad response, timeout after retries occur, or if the run number is not available.
0148         """
0149         firstConditionSafeRunAPI = "firstconditionsaferun"
0150         safeRunDict = self._queryTier0DataSvc( os.path.join( self._uri, firstConditionSafeRunAPI ) )
0151         if safeRunDict is None:
0152             errStr = """First condition safe run is not available in Tier0DataSvc from URL \"%s\"""" %( os.path.join( self._uri, firstConditionSafeRunAPI ), )
0153             if self._proxy:
0154                 errStr += """ using proxy \"%s\".""" %( str( self._proxy ), )
0155             raise Tier0Error( errStr )
0156         return int(safeRunDict['result'][0])
0157
0158     def getGlobalTag( self, config ):
0159         """
0160         Queries Tier0DataSvc to get the most recent Global Tag for a given workflow.
0161         Parameters:
0162         config: Tier0DataSvc API call for the workflow to be looked for;
0163         @returns: a string with the Global Tag name.
0164         Raises if connection error, bad response, timeout after retries occur, or if no Global Tags are available.
0165         """
0166         data = self._queryTier0DataSvc( os.path.join( self._uri, config ) )
0167         gtnames = sorted(unique( [ str( di[ 'global_tag' ] ) for di in data['result'] if di[ 'global_tag' ] is not None ] ))
0168         try:
0169             recentGT = gtnames[-1]
0170             return recentGT
0171         except IndexError:
0172             errStr = """No Global Tags for \"%s\" are available in Tier0DataSvc from URL \"%s\"""" %( config, os.path.join( self._uri, config ) )
0173             if self._proxy:
0174                 errStr += """ using proxy \"%s\".""" %( str( self._proxy ), )
0175         raise Tier0Error( errStr )
0176
0177
0178 def test( url ):
0179     t0 = Tier0Handler( url, 1, 1, 1, None, debug=False)
0180
0181     print('   fcsr = %s (%s)' % (t0.getFirstSafeRun(), type(t0.getFirstSafeRun()) ))
0182     print('   reco_config = %s' % t0.getGlobalTag('reco_config'))
0183     print('   express_config = %s' % t0.getGlobalTag('express_config'))
0184     print('\n')
0185
0186
0187 if __name__ == '__main__':
0188     test( tier0Url )
0189