ServerUtilities.py

"""
This contains some utility methods to share between the server and the client, or by the server components themself
"""
# this is a weird and unclear message from pylint, better to ignore it
# pylint: disable=W0715

from __future__ import print_function
from __future__ import division
from __future__ import absolute_import

import os
import sys
import re
import time
import fcntl
import hashlib
import calendar
import datetime
import traceback
import subprocess
import contextlib
try:
    from http.client import HTTPException  # Python 3 and Python 2 in modern CMSSW
except:  # pylint: disable=bare-except
    from httplib import HTTPException  # old Python 2 version in CMSSW_7
if sys.version_info >= (3, 0):
    from urllib.parse import urlencode, quote  # pylint: disable=no-name-in-module
    from past.builtins import basestring
if sys.version_info < (3, 0):
    from urllib import urlencode, quote

BOOTSTRAP_CFGFILE_DUMP = 'PSetDump.py'
FEEDBACKMAIL = 'hn-cms-computing-tools@cern.ch'

# Parameters for User File Cache
# 120 MB is the maximum allowed size of a single file
FILE_SIZE_LIMIT = 120*1048576
# 0.5MB is the maximum limit for file completely loaded into memory
FILE_MEMORY_LIMIT = 512*1024

# these are known, pre-defined nickames for "CRAB configuration" at large
# which correspond to a well known and specified REST host name and DataBase instance
SERVICE_INSTANCES = {'prod': {'restHost':'cmsweb.cern.ch', 'dbInstance':'prod'},
                     'preprod': {'restHost':'cmsweb-testbed.cern.ch', 'dbInstance':'preprod'},
                     'auth' : {'restHost':'cmsweb-auth.cern.ch', 'dbInstance':'preprod'},
                     'test1': {'restHost':'cmsweb-test1.cern.ch', 'dbInstance':'dev'},
                     'test2': {'restHost':'cmsweb-test2.cern.ch', 'dbInstance':'dev'},
                     'test3': {'restHost':'cmsweb-test3.cern.ch', 'dbInstance':'dev'},
                     'test4': {'restHost':'cmsweb-test4.cern.ch', 'dbInstance':'dev'},
                     'test5': {'restHost':'cmsweb-test5.cern.ch', 'dbInstance':'dev'},
                     'test6': {'restHost':'cmsweb-test6.cern.ch', 'dbInstance':'dev'},
                     'stefanovm': {'restHost':'stefanovm.cern.ch', 'dbInstance':'dev'},
                     'stefanovm2': {'restHost':'stefanovm2.cern.ch', 'dbInstance':'dev'},
                     'other': {'restHost':None, 'dbInstance':None},
                     }

# Fatal error limits for job resource usage
# Defaults are used if unable to load from .job.ad
# Otherwise it uses these values.
MAX_WALLTIME = 21*60*60 + 30*60
MAX_MEMORY = 2*1024
# see https://github.com/dmwm/CRABServer/issues/5995
MAX_MEMORY_PER_CORE = 2500
MAX_MEMORY_SINGLE_CORE = 5000
MAX_DISK_SPACE = 20000000 # Disk usage is not used from .job.ad as CRAB3 is not seeting it. 20GB is max.

MAX_IDLE_JOBS = 1000
MAX_POST_JOBS = 20

## Parameter used to set the LeaveJobInQueue and the PeriodicRemoveclassads.
## It's also used during resubmissions since we don't allow a resubmission during the last week
## Before changing this value keep in mind that old running DAGs have the old value in the CRAB_TaskSubmitTime
## classad expression but DagmanResubmitter uses this value to calculate if a resubmission is possible
TASKLIFETIME = 30*24*60*60
## Number of days where the resubmission is not possible if the task is expiring
NUM_DAYS_FOR_RESUBMITDRAIN = 7
## Maximum number of days a task can stay in TAPERECALL status for
MAX_DAYS_FOR_TAPERECALL = 14

## These are all possible statuses of a task in the TaskDB.
TASKDBSTATUSES_TMP = ['NEW', 'HOLDING', 'QUEUED']
TASKDBSTATUSES_FAILURES = ['SUBMITFAILED', 'KILLFAILED', 'RESUBMITFAILED', 'FAILED']
TASKDBSTATUSES_FINAL = ['UPLOADED', 'SUBMITTED', 'KILLED'] + TASKDBSTATUSES_FAILURES
TASKDBSTATUSES = TASKDBSTATUSES_TMP + TASKDBSTATUSES_FINAL

## These are all possible statuses of a task as returned by the `status' API.
TASKSTATUSES = TASKDBSTATUSES + ['COMPLETED', 'UNKNOWN', 'InTransition']


## These are all allowed transfer and publication status. P.S. See below for allowed statuses for user
## to put into database. For sure we don`t want to allow or make mistakes and add KILL status for files
## which were not transferred at all. Just a bit of security
TRANSFERDB_STATES = {0: "NEW",
                     1: "ACQUIRED",
                     2: "FAILED",
                     3: "DONE",
                     4: "RETRY",
                     5: "SUBMITTED",
                     6: "KILL",
                     7: "KILLED"}
TRANSFERDB_STATUSES = dict((v, k) for k, v in TRANSFERDB_STATES.items())

PUBLICATIONDB_STATES = {0: "NEW",
                        1: "ACQUIRED",
                        2: "FAILED",
                        3: "DONE",
                        4: "RETRY",
                        5: "NOT_REQUIRED"}
PUBLICATIONDB_STATUSES = dict((v, k) for k, v in PUBLICATIONDB_STATES.items())

## Whenever user is putting a new Doc, these statuses will be used for double checking
## and also for adding in database correct value
USER_ALLOWED_TRANSFERDB_STATES = {0: "NEW",
                                  3: "DONE"}
USER_ALLOWED_TRANSFERDB_STATUSES = dict((v, k) for k, v in TRANSFERDB_STATES.items())

USER_ALLOWED_PUBLICATIONDB_STATES = {0: "NEW",
                                     4: "RETRY",
                                     5: "NOT_REQUIRED"}
USER_ALLOWED_PUBLICATIONDB_STATUSES = dict((v, k) for k, v in PUBLICATIONDB_STATES.items())

def USER_SANDBOX_EXCLUSIONS(tarmembers):
    """ The function is used by both the client and the crabcache to get a list of files to exclude during the
        calculation of the checksum of the user input sandbox.

        In particular the client tries to put the process object obtained with dumpPython in the sandbox, so that
        this can be used to calculate the checksum. If we used the process object saved with pickle.dump we would
        run into a problem since the same process objects have different dumps, see:
        https://github.com/dmwm/CRABServer/issues/4948#issuecomment-132984687
    """
    if BOOTSTRAP_CFGFILE_DUMP in map(lambda x: x.name, tarmembers):  # pylint: disable=deprecated-lambda
        #exclude the pickle pset if the dumpPython PSet is there
        return ['PSet.py', 'PSet.pkl', 'debug/crabConfig.py', 'debug/originalPSet.py.py']
    else:
        return ['debug/crabConfig.py', 'debug/originalPSet.py.py']

def NEW_USER_SANDBOX_EXCLUSIONS(tarmembers):
    """ Exclusion function used with the new crabclient (>= 3.3.1607). Since the new client sandbox no longer
        contains the debug files, it's pointless to exclude them. Also, this function is used when getting
        the hash of the debug tarball (a new addition in 3.3.1607). If the debug files are excluded, the tarball
        would always have the same hash and stay the same, serving no purpose.
    """
    if BOOTSTRAP_CFGFILE_DUMP in map(lambda x: x.name, tarmembers): # pylint: disable=deprecated-lambda
        #exclude the pickle pset if the dumpPython PSet is there
        return ['PSet.py', 'PSet.pkl']
    else:
        return []


def getTestDataDirectory():
    """ Get the /path/to/repository/test/data direcotry to locate files for unit tests
    """
    testdirList = __file__.split(os.sep)[:-3] + ["test", "data"]
    return os.sep.join(testdirList)

def isCouchDBURL(url):
    """ Return True if the url proviced is a couchdb one
    """
    return 'couchdb' in url


def truncateError(msg):
    """Truncate the error message to the first 7400 chars if needed, and add a message if we truncate it.
       See https://github.com/dmwm/CRABServer/pull/4867#commitcomment-12086393
    """
    MSG_LIMIT = 7500
    if len(msg) > MSG_LIMIT:
        truncMsg = msg[:MSG_LIMIT - 100]
        truncMsg += "\n[... message truncated to the first 7400 chars ...]"
        return truncMsg
    else:
        return msg


def checkOutLFN(lfn, username):
    """ Check if the lfn provided contains the username or is /store/local/ or /store/local/
    """
    if lfn.startswith('/store/user/rucio/'):
        if lfn.split('/')[4] != username:
            return False 
    elif lfn.startswith('/store/user/'):
        if lfn.split('/')[3] != username:
            return False
    elif lfn.startswith('/store/test/rucio/user/'):
        if lfn.split('/')[5] != username:
            return False
    elif lfn.startswith('/store/test/rucio/int/user/'):
        if lfn.split('/')[6] != username:
            return False
    elif lfn.startswith('/store/group/') or lfn.startswith('/store/local/'):
        if lfn.split('/')[3] == '':
            return False
    else:
        return False
    return True


def getProxiedWebDir(crabserver=None, task=None, logFunction=print):
    """ The function simply queries the REST interface specified to get the proxied webdir to use
        for the task. Returns None in case the API could not find the url (either an error or the schedd
        is not configured)
    """
    data = {'subresource': 'webdirprx',
            'workflow': task,
           }
    res = None
    try:
        dictresult, _, _ = crabserver.get(api='task', data=data)  # the second and third parameters are deprecated
        if dictresult.get('result'):
            res = dictresult['result'][0]
    except HTTPException as hte:
        logFunction(traceback.format_exc())
        logFunction(hte.headers)
        logFunction(hte.result)

    return res

def insertJobIdSid(jinfo, jobid, workflow, jobretry):
    """ Modifies passed dictionary (jinfo) to contain jobId and sid keys and values.
        Used when creating dashboard reports.
    """
    jinfo['jobId'] = "%s_https://glidein.cern.ch/%s/%s_%s" % (jobid, jobid, workflow.replace("_", ":"), jobretry)
    jinfo['sid'] = "https://glidein.cern.ch/%s%s" % (jobid, workflow.replace("_", ":"))


def getWebdirForDb(reqname, storage_rules):
    """ Get the location of the webdir. This method is called on the schedd by AdjustSites
    """
    path = os.path.expanduser("~/%s" % reqname)
    sinfo = storage_rules.split(",")
    storage_re = re.compile(sinfo[0])
    val = storage_re.sub(sinfo[1], path)
    return val


def cmd_exist(cmd):
    """ Check if linux command exist
    """
    try:
        null = open("/dev/null", "w")
        p = subprocess.Popen("/usr/bin/which %s" % cmd, stdout=null, stderr=null, shell=True)
        p.communicate()
        null.close()
        if p.returncode == 0:
            return True
    except OSError:
        return False
    return False


def getCheckWriteCommand(proxy, logger):
    """ Return prepared gfal or lcg commands for checkwrite
    """
    cpCmd = None
    rmCmd = None
    append = ""
    if cmd_exist("gfal-copy") and cmd_exist("gfal-rm"):
        logger.info("Will use GFAL2 commands for checking write permissions")
        cpCmd = "env -i X509_USER_PROXY=%s gfal-copy -v -p -t 180 " % os.path.abspath(proxy)
        rmCmd = "env -i X509_USER_PROXY=%s gfal-rm -v -t 180 " % os.path.abspath(proxy)
        append = "file://"
    elif cmd_exist("lcg-cp") and cmd_exist("lcg-del"):
        logger.info("Will use LCG commands for checking write permissions")
        cpCmd = "lcg-cp -v -b -D srmv2 --connect-timeout 180 "
        rmCmd = "lcg-del --connect-timeout 180 -b -l -D srmv2 "
    return cpCmd, rmCmd, append


def createDummyFile(filename, logger):
    """ Create dummy file for checking write permissions
    """
    abspath = os.path.abspath(filename)
    try:
        with open(abspath, 'w') as fd:
            fd.write('This is a dummy file created by the crab checkwrite command on %s' % str(datetime.datetime.now().strftime('%d/%m/%Y at %H:%M:%S')))
    except IOError:
        logger.info('Error: Failed to create file %s localy.' % filename)
        raise


def removeDummyFile(filename, logger):
    """ Remove created dummy file
    """
    abspath = os.path.abspath(filename)
    try:
        os.remove(abspath)
    except Exception:
        logger.info('Warning: Failed to delete file %s' % filename)

def executeCommand(command):
    """ Execute passed bash command. There is no check for command success or failure. Who`s calling
        this command, has to check exitcode, out and err
    """
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    out, err = process.communicate()
    exitcode = process.returncode
    return out, err, exitcode

def execute_command(command=None, logger=None, timeout=None, redirect=True):
    """
    execute command with optional logging and timeout.
    Returns a 3-ple: stdout, stderr, rc
      rc=0 means success.
      rc=124 (SIGTERM) means that command timed out
    Redirection of std* can be turned off if the command will need to interact with caller
    writing messages and/or asking for input, like if needs to get a passphrase to access
    usercert/key for (my)proxy creation as in
    https://github.com/dmwm/WMCore/blob/75c5abd83738a6a3534027369cd6e109667de74e/src/python/WMCore/Credential/Proxy.py#L383-L385
    Should eventually replace executeCommand and be used everywhere.
    Imported here from WMCore/Credential/Proxy.py and then adapted to avoid deadlocks
    as per https://github.com/dmwm/CRABClient/issues/5026
    """

    stdout, stderr, rc = None, None, 99999
    if logger:
        logger.debug('Executing command :\n %s' % command)
    if timeout:
        if logger:
            logger.debug('add timeout at %s seconds', timeout)
        command = ('timeout %s ' % timeout ) + command
    if redirect:
        proc = subprocess.Popen(
            command, shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            stdin=subprocess.PIPE,
        )
    else:
        proc = subprocess.Popen(command, shell=True)

    out, err = proc.communicate()
    rc = proc.returncode
    if rc == 124 and timeout:
        if logger:
            logger.error('ERROR: Timeout after %s seconds in executing:\n %s' % (timeout,command))
    # for Py3 compatibility
    stdout = out.decode(encoding='UTF-8') if out else ''
    stderr = err.decode(encoding='UTF-8') if err else ''
    if logger:
        logger.debug('output : %s\n error: %s\n retcode : %s' % (stdout, stderr, rc))

    return stdout, stderr, rc


def isFailurePermanent(reason, gridJob=False):
    """ Method that decides whether a failure reason should be considered as a
        permanent failure and submit task or not.
    """
    from WMCore.WMExceptions import STAGEOUT_ERRORS

    checkQuota = " Please check that you have write access to destination site and that your quota is not exceeded, use crab checkwrite for more informations."
    refuseToSubmit = " Can't submit task because write check at destination site fails."
    if gridJob:
        refuseToSubmit = ""
    for exitCode in STAGEOUT_ERRORS:
        for error in STAGEOUT_ERRORS[exitCode]:
            if re.match(error['regex'].lower(), reason.lower()):
                reason = error['error-msg'] + refuseToSubmit + checkQuota
                return error['isPermanent'], reason, exitCode
    return False, "", None


def parseJobAd(filename):
    """ Parse the jobad file provided as argument and return a dict representing it
    """
    jobAd = {}
    with open(filename) as fd:
        #classad.parseOld(fd)
        for adline in fd.readlines():
            info = adline.split(' = ', 1)
            if len(info) != 2:
                continue
            if info[1].startswith('undefined'):
                val = info[1].strip()
            elif info[1].startswith('"'):
                val = info[1].strip().replace('"', '')
            else:
                try:
                    val = int(info[1].strip())
                except ValueError:
                    continue
            jobAd[info[0]] = val
    return jobAd


def mostCommon(lst, default=0):
    """ Return the most common error among the list
    """
    try:
        return max(set(lst), key=lst.count)
    except ValueError:
        return default


@contextlib.contextmanager
def getLock(name):
    """ Create a "name".lock file and, using fcnt, lock it (or wait for the lock to be released) before proceeding)
    """
    with open(name + '.lock', 'a+') as fd:
        fcntl.flock(fd, fcntl.LOCK_EX)
        yield fd


def getHashLfn(lfn):
    """ Provide a hashed lfn from an lfn.
    """
    return hashlib.sha224(lfn).hexdigest()


def generateTaskName(username, requestname, timestamp=None):
    """ Generate a taskName which is saved in database
    """
    if not timestamp:
        timestamp = time.strftime('%y%m%d_%H%M%S', time.gmtime())
    taskname = "%s:%s_%s" % (timestamp, username, requestname)
    return taskname

def getUsernameFromTaskname (taskname):
    """
    extract username from a taskname constructed as in generateTaskName above
    :param taskname:
    :return: username
    """
    username = taskname.split(':')[1].split('_')[0]
    return username

def getTimeFromTaskname(taskname):
    """ Get the submission time from the taskname and return the seconds since epoch
        corresponding to it. The function is not currently used.
    """

    #validate taskname. In principle not necessary, but..
    if not isinstance(taskname, str):
        raise TypeError('In ServerUtilities.getTimeFromTaskname: "taskname" parameter must be a string')
    stime = taskname.split(':')[0] #s stands for string
    stimePattern = '^\d{6}_\d{6}$'
    if not re.match(stimePattern, stime):
        raise ValueError('In ServerUtilities.getTimeFromTaskname: "taskname" parameter must match %s' % stimePattern)
    #convert the time
    dtime = time.strptime(stime, '%y%m%d_%H%M%S') #d stands for data structured
    return calendar.timegm(dtime)

# TODO: Remove this from CRABClient. This is kind of common for WMCore not only for CRAB. Maybe better place to have this in WMCore?
def encodeRequest(configreq, listParams=None):
    """ Used to encode the request from a dict to a string. Include the code needed for transforming lists in the format required by
        cmsweb, e.g.:   adduserfiles = ['file1','file2']  ===>  [...]adduserfiles=file1&adduserfiles=file2[...]
        The list of dictionary keys like adduserfiles above, which have a list as value, needs to be passed in the listParams argument  
    """
    listParams = listParams or []
    encodedLists = ''
    for lparam in listParams:
        if lparam in configreq:
            if len(configreq[lparam]) > 0:
                encodedLists += ('&%s=' % lparam) + ('&%s=' % lparam).join(map(quote, configreq[lparam]))
            del configreq[lparam]
    if isinstance(configreq, dict):
        #Make sure parameters we encode are streing. Otherwise u'IamAstring' may become 'uIamAstring' in the DB
        for k, v in configreq.items():
            if isinstance(v, basestring):
                configreq[k] = str(v)
            if isinstance(v, list):
                configreq[k] = [str(x) if isinstance(x, basestring) else x for x in v]
    encoded = urlencode(configreq) + encodedLists
    return str(encoded)


def oracleOutputMapping(result, key=None):
    """ If key is defined, it will use id as a key and will return dictionary which contains all items with this specific key
        Otherwise it will return a list of dictionaries.
        Up to caller to check for catch KeyError,ValueError and any other related failures.
        KeyError is raised if wrong key is specified.
        ValueError is raised if wrong format is provided. So far database provides tuple which has 1 dictionary.
    """
    if not(result, tuple):
        raise ValueError('Provided input is not valid tuple')
    if not(result[0], dict):
        raise ValueError('Provided input does not contain dictionary as first element')
    outputDict = {} if key else []
    for item in result[0]['result']:
        docInfo = dict(zip(result[0]['desc']['columns'], item))
        docOut = {}
        for dockey in docInfo:
            # Remove first 3 characters as they are tm_*
            docOut[dockey[3:]] = docInfo[dockey] # rm tm_ which is specific for database
        if key:
            if docOut[key] not in outputDict:
                outputDict[docOut[key]] = []
            outputDict[docOut[key]].append(docOut)
        else:
            outputDict.append(docOut)
    return outputDict

def checkTaskLifetime(submissionTime):
    """ Verify that at least 7 days are left before the task periodic remove expression
        evaluates to true. This is to let job finish and possibly not remove a task with
        running jobs.

        submissionTime are the seconds since epoch of the task submission time in the DB
    """

    msg = "ok"
    ## resubmitLifeTime is 23 days expressed in seconds
    resubmitLifeTime = TASKLIFETIME - NUM_DAYS_FOR_RESUBMITDRAIN * 24 * 60 * 60
    if time.time() > (submissionTime + resubmitLifeTime):
        msg = ("Resubmission of the task is not possble since less than %s days are left before the task is removed from the""schedulers.\n" % NUM_DAYS_FOR_RESUBMITDRAIN)
        msg += "A task expires %s days after its submission\n" % (TASKLIFETIME / (24 * 60 * 60))
        msg += "You can submit a 'recovery task' if you need to execute again the failed jobs\n"
        msg += "See https://twiki.cern.ch/twiki/bin/view/CMSPublic/CRAB3FAQ for more information about recovery tasks"
    return msg

def getEpochFromDBTime(startTime):
    """
    args:
    startTime: a time tuple such as returned by the gmtime() function in the time module
    returns:
    seconds from Epoch
    """
    return calendar.timegm(startTime.utctimetuple())

def getColumn(dictresult, columnName):
    """ Given a dict returned by REST calls, return the value of columnName
    """
    columnIndex = dictresult['desc']['columns'].index(columnName)
    value = dictresult['result'][columnIndex]
    if value == 'None':
        return None
    else:
        return value

class newX509env():
    """
    context manager to run some code with a new x509 environment
    exiting env is restored when code is exited, even if exited via exception
    The new environmnet will only contain X509_USER_.. variables explicitely
    listed as argument to the method.
    see: https://docs.python.org/2.7/reference/compound_stmts.html?highlight=try#the-with-statement
    and: http://preshing.com/20110920/the-python-with-statement-by-example/
    usage 1:
        with newX509env(X509_USER_PROXY=proxy,X509_USER_CERT=cert,X509_USER_KEY=key :
           do stuff
    usage 2:
        myEnv=newX509env(X509_USER_PROXY=myProxy,X509_USER_CERT=myCert,X509_USER_KEY=myKey)
        with myEnv:
            do stuff
    """

    def __init__(self, X509_USER_PROXY=None, X509_USER_CERT=None, X509_USER_KEY=None):
        # define the new environment
        self.oldProxy = None
        self.oldCert = None
        self.oldKey = None
        self.newProxy = X509_USER_PROXY
        self.newCert = X509_USER_CERT
        self.newKey = X509_USER_KEY

    def __enter__(self):
        # save current env
        self.oldProxy = os.getenv('X509_USER_PROXY')
        self.oldCert = os.getenv('X509_USER_CERT')
        self.oldKey = os.getenv('X509_USER_KEY')
        # Clean previous env. only delete env. vars if they were defined
        if self.oldProxy: del os.environ['X509_USER_PROXY']
        if self.oldCert:  del os.environ['X509_USER_CERT']
        if self.oldKey:   del os.environ['X509_USER_KEY']
        # set environment to whatever user wants
        if self.newProxy: os.environ['X509_USER_PROXY'] = self.newProxy
        if self.newCert:  os.environ['X509_USER_CERT'] = self.newCert
        if self.newKey:   os.environ['X509_USER_KEY'] = self.newKey

    def __exit__(self, a, b, c):
        # restore X509 environment
        if self.oldProxy:
            os.environ['X509_USER_PROXY'] = self.oldProxy
        else:
            if os.getenv('X509_USER_PROXY'): del os.environ['X509_USER_PROXY']
        if self.oldCert:
            os.environ['X509_USER_CERT'] = self.oldCert
        else:
            if os.getenv('X509_USER_CERT'): del os.environ['X509_USER_CERT']
        if self.oldKey:
            os.environ['X509_USER_KEY'] = self.oldKey
        else:
            if os.getenv('X509_USER_KEY'): del os.environ['X509_USER_KEY']


class tempSetLogLevel():
    """
        a simple context manager to temporarely change logging level
        making sure it is restored even if exception is raised
        USAGE:
            with tempSetLogLevel(logger=myLogger,level=logging.ERROR):
               do stuff
        both arguments are mandatory

        EXAMPLE:
        import logging
        logger=logging.getLogger()
        logger.addHandler(logging.StreamHandler())
        logger=logging.getLogger()
        logger.setLevel(logging.DEBUG)
        logger.debug("debug msg")
        with tempSetLogLevel(logger,logging.ERROR):
          try:
            i=1
            while True:
              logger.debug("msg inside loop at %d", i)
              if i==3: logger.error("got one error at %d", i)
              if i==5: raise
              i+=1
          except Exception as e:
            logger.error("got exception at %d", i)
            pass
        logger.debug("after the loop")

        RESULTS IN THIS PRINTOUT:
        debug msg
        got one error at 3
        got exception at 5
        after the loop
    """
    import logging
    def __init__(self, logger=None, level=None):
        self.previousLogLevel = None
        self.newLogLevel = level
        self.logger = logger
    def __enter__(self):
        self.previousLogLevel = self.logger.getEffectiveLevel()
        self.logger.setLevel(self.newLogLevel)
    def __exit__(self,a,b,c):
        self.logger.setLevel(self.previousLogLevel)

####################################################################
# Below a few convenience functions to access the S3 implementation of crabcache
# Clients are expected to use the following functions which all have the same signature:
#  downloadFromS3        to retrieve an object into a file
#  retrieveFromS3        to retrieve an object as JSON to use in the code
#  uploadToS3            to upload a file into an S3 object
#  getDownloadUrlFromS3  to obtain a PreSigned URL to access an existing object in S3
#    this can be used e.g. to share access to logs
# Common signature is: (crabserver, filepath, objecttype, taskname, username, tarballname, logger)
#  see the individual functions for more details
#
# Other functions are uploadToS3viaPSU and downloadFromS3viaPSU are for internal use.
####################################################################
def downloadFromS3(crabserver=None, filepath=None, objecttype=None, taskname=None,
                    username=None, tarballname=None, logger=None):
    """
    one call to make a 2-step operation:
    obtains a preSignedUrl from crabserver RESTCache and use it to download a file
    :param crabserver: a RESTInteraction/CRABRest object : points to CRAB Server to use
    :param filepath: string : the full path of the file to create with the downloaded content
        if file exists already, it is silently overwritten
    :param objecttype: string : the kind of object to dowbload: clientlog|twlog|sandbox|debugfiles|runtimefiles
    :param taskname: string : the task this object belongs to, if applicable
    :param username: string : the username this sandbox belongs to, in case objecttype=sandbox
    :param tarballname: string : for sandbox, taskname is not used but tarballname is needed
    :return: nothing. Raises an exception in case of error
    """
    preSignedUrl = getDownloadUrlFromS3 (crabserver=crabserver, objecttype=objecttype,
            taskname=taskname, username=username, tarballname=tarballname, logger=logger)
    downloadFromS3ViaPSU(filepath=filepath, preSignedUrl=preSignedUrl, logger=logger)
    return

def retrieveFromS3(crabserver=None, filepath=None, objecttype=None, taskname=None,
                   username=None, tarballname=None, logger=None):
    """
    obtains a preSignedUrl from crabserver RESTCache and use it to retrieve a file
    :param crabserver: a RESTInteraction/CRABRest object : points to CRAB Server to use
    :param objecttype: string : the kind of object to retrieve: clientlog|twlog|sandbox|debugfiles|runtimefiles
    :param taskname: string : the task this object belongs to, if applicable
    :param username: string : the username this sandbox belongs to, in case objecttype=sandbox
    :param tarballname: string : when retrieving sandbox taskname is not used but tarballname is needed
    :return: the content of the S3 object as a string object
    """
    api = 'cache'
    dataDict = {'subresource':'retrieve', 'objecttype':objecttype}
    if taskname:
        dataDict['taskname'] = taskname
    if username:
        dataDict['username'] = username
    if tarballname:
        dataDict['tarballname'] = tarballname
    data = encodeRequest(dataDict)
    try:
        # calls to restServer alway return a 3-ple ({'result':a-list}, HTTPcode, HTTPreason)
        res = crabserver.get(api, data)
        result = res[0]['result'][0]
    except Exception as e:
        raise Exception('Failed to retrieve S3 file content via CRABServer:\n%s' % str(e))
    logger.debug("%s retrieved OK", objecttype)
    return result

def uploadToS3(crabserver=None, filepath=None, objecttype=None, taskname=None,
               username=None, tarballname=None, logger=None):
    """
    one call to make a 2-step operation:
    obtains a preSignedUrl from crabserver RESTCache and use it to upload a file
    :param crabserver: a RESTInteraction/CRABRest object : points to CRAB Server to use
    :param filepath: string : the full path of the file to upload, if applicable
    :param objecttype: string : the kind of object to upload: clientlog|twlog|sandbox|debugfiles|runtimefiles
    :param username: string : the username this sandbox belongs to, in case objecttype=sandbox
      username is not needed for upload, it is here to have same signature as other methods
    :param tarballname: string : when uploading sandbox, taskname is not used but tarballname is needed
    :return: nothing. Raises an exception in case of error
    """
    api = 'cache'
    dataDict = {'subresource':'upload', 'objecttype':objecttype}
    if taskname:
        dataDict['taskname'] = taskname
    if username:
        dataDict['username'] = username
    if tarballname:
        dataDict['tarballname'] = tarballname
    data = encodeRequest(dataDict)
    try:
        # calls to restServer alway return a 3-ple ({'result':'string'}, HTTPcode, HTTPreason)
        # this returns in result.value a 2-element list (url, dictionay)
        res = crabserver.get(api, data)
        result = res[0]['result']
    except Exception as e:
        raise Exception('Failed to get PreSignedURL from CRAB Server:\n%s' % str(e))
    # prepare a single dictionary with curl arguments for the actual upload
    url = result[0]
    if not url and objecttype == 'sandbox':
        # in this case a null string as url indicates that sandbox with this name is there already
        logger.debug("%s %s is already in the S3 store, will not upload again", objecttype, tarballname)
        return
    fields = result[1]
    preSignedUrl = {'url':url}
    preSignedUrl.update(fields)
    try:
        uploadToS3ViaPSU(filepath=filepath, preSignedUrlFields=preSignedUrl, logger=logger)
    except Exception as e:
        raise Exception('Upload to S3 failed\n%s' % str(e))
    logger.debug('%s %s successully uploaded to S3', objecttype, filepath)
    return

def getDownloadUrlFromS3(crabserver=None, filepath=None, objecttype=None, taskname=None,
                         username=None, tarballname=None, logger=None):
    """
    obtains a PreSigned URL to access an existing object in S3
    :param crabserver: a RESTInteraction/CRABRest object : points to CRAB Server to use
    :param objecttype: string : the kind of object to retrieve: clientlog|twlog|sandbox|debugfiles|runtimefiles
    :param taskname: string : the task this object belongs to, if applicable
    :param username: string : the username this sandbox belongs to, in case objecttype=sandbox
    :param tarballname: string : for sandbox, taskname is not used but tarballname is needed
    :return: a (short lived) pre-signed URL to use e.g. in a wget
    """
    api = 'cache'
    dataDict = {'subresource':'download', 'objecttype':objecttype}
    if taskname:
        dataDict['taskname'] = taskname
    if username:
        dataDict['username'] = username
    if tarballname:
        dataDict['tarballname'] = tarballname
    data = encodeRequest(dataDict)
    try:
        # calls to restServer alway return a 3-ple ({'result':a-list}, HTTPcode, HTTPreason)
        res = crabserver.get(api, data)
        result = res[0]['result'][0]
    except Exception as e:
        raise Exception('Failed to retrieve S3 preSignedUrl via CRABServer:\n%s' % str(e))
    logger.debug("PreSignedUrl to download %s received OK", objecttype)
    return result

def uploadToS3ViaPSU (filepath=None, preSignedUrlFields=None, logger=None):
    """
    uploads a file to s3.cern.ch usign PreSigned URLs obtained e.g. from a call to
    crabserver RESTCache API /crabserver/prod/cache?subresource=upload
    based on env.variable 'CRAB_useGoCurl' presence, implementation can use plain curl
    or gocurl to execute upload to S3 command
    :param filepath: string : the full path to a file to upload
    :param preSignedUrlFields: dictionary: a dictionary with the needed fields as
             returned from boto3 client when generating preSignedUrls:
             dictionary keys are: AWSAccessKeyId, policy, signature, key, url
             see e.g. https://boto3.amazonaws.com/v1/documentation/api/1.9.185/guide/s3-presigned-urls.html
             and https://gist.github.com/henriquemenezes/61ab0a0e5b54d309194c
    :return: nothing. Raises an exception in case of error
    """

    # validate args
    if not filepath :
        raise Exception("mandatory filepath argument missing")
    if not os.path.exists(filepath) :
        raise Exception("filepath argument points to non existing file")
    if not preSignedUrlFields :
        raise Exception("mandatory preSignedUrlFields argument missing")

    # parse field disctionary into a list of arguments for curl
    AWSAccessKeyId = preSignedUrlFields['AWSAccessKeyId']
    signature = preSignedUrlFields['signature']
    policy = preSignedUrlFields['policy']
    key = preSignedUrlFields['key']
    url = preSignedUrlFields['url']   # N.B. this contains the bucket name e.g. https://s3.cern.ch/bucket1

    userAgent = 'CRAB'
    uploadCommand = ''

    # CRAB_useGoCurl env. variable is used to define how upload to S3 command should be executed.
    # If variable is set, then goCurl is used for command execution: https://github.com/vkuznet/gocurl
    if os.getenv('CRAB_useGoCurl'):
        uploadCommand += '/afs/cern.ch/user/v/valya/public/gocurl -verbose 2 -method POST'
        uploadCommand += ' -header "User-Agent:%s"' % userAgent
        uploadCommand += ' -form "key=%s"' % key
        uploadCommand += ' -form "AWSAccessKeyId=%s"' % AWSAccessKeyId
        uploadCommand += ' -form "policy=%s"' % policy
        uploadCommand += ' -form "signature=%s"' % signature
        uploadCommand += ' -form "file=@%s"' % filepath
        uploadCommand += ' -url "%s"' % url
    else:
        # am I running in the CMSSW SCRAM environment (e.g. CRABClient) ?
        arch = os.getenv('SCRAM_ARCH')
        if arch and arch.startswith('slc6'):
            # make sure to use latest curl version
            uploadCommand += 'source /cvmfs/cms.cern.ch/slc6_amd64_gcc700/external/curl/7.59.0/etc/profile.d/init.sh; '

        # curl: -f flag makes it fail if server return HTTP error
        uploadCommand += 'curl -f -v -X POST '
        uploadCommand += ' -H "User-Agent: %s"' % userAgent
        uploadCommand += ' -F "key=%s"' % key
        uploadCommand += ' -F "AWSAccessKeyId=%s"' % AWSAccessKeyId
        uploadCommand += ' -F "policy=%s"' % policy
        uploadCommand += ' -F "signature=%s"' % signature
        uploadCommand += ' -F "file=@%s"' % filepath
        uploadCommand += ' "%s"' % url

    logger.debug('Will execute:\n%s', uploadCommand)
    uploadProcess = subprocess.Popen(uploadCommand, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    stdout, stderr = uploadProcess.communicate()
    exitcode = uploadProcess.returncode
    logger.debug('exitcode: %s\nstdout: %s', exitcode, stdout)

    if exitcode != 0:
        raise Exception('Failed to upload file with %s. stderr is:\n%s' % (uploadCommand, stderr))
    return

def downloadFromS3ViaPSU(filepath=None, preSignedUrl=None, logger=None):
    """
    More generic than the name implies:
    downloads a file usign PreSigned URLs obtained e.g. from a call to
    crabserver RESTCache API /crabserver/prod/cache?subresource=download
    based on env.variable 'CRAB_useGoCurl' presence, implementation can use wget
    or gocurl to execute download from S3 command
    :param filepath: string : the full path to the file to be created
        if the file exists already, it will be silently overwritten
    :param preSignedUrl: string : the URL to use (includes the host name)
    :return: nothing. Raises an exception in case of error
    """

    # validate args
    if not filepath :
        raise Exception("mandatory filepath argument missing")
    if not preSignedUrl :
        raise Exception("mandatory preSignedUrl argument missing")

    downloadCommand = ''

    # CRAB_useGoCurl env. variable is used to define how download from S3 command should be executed.
    # If variable is set, then goCurl is used for command execution: https://github.com/vkuznet/gocurl
    if os.getenv('CRAB_useGoCurl'):
        downloadCommand += '/afs/cern.ch/user/v/valya/public/gocurl -verbose 2 -method GET'
        downloadCommand += ' -out "%s"' % filepath
        downloadCommand += ' -url "%s"' % preSignedUrl
    else:
        downloadCommand += 'wget -Sq -O %s ' % filepath
        downloadCommand += ' "%s"' % preSignedUrl

    downloadProcess = subprocess.Popen(downloadCommand, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    logger.debug("Will execute:\n%s", downloadCommand)
    stdout, stderr = downloadProcess.communicate()
    exitcode = downloadProcess.returncode
    logger.debug('exitcode: %s\nstdout: %s', exitcode, stdout)

    if exitcode != 0:
        raise Exception('Download command %s failed. stderr is:\n%s' % (downloadCommand, stderr))
    if not os.path.exists(filepath) :
        raise Exception("Download failure with %s. File %s was not created" % (downloadCommand, filepath))

    return