<commit_before>"""
  JobMover: top level mover to copy all Job files

  This functionality actually needs to be incorporated directly into RunJob,
  because it's just a workflow of RunJob
  but since RunJob at the moment is implemented as a Singleton
  it could be dangerous because of Job instance stored in the class object.
  :author: <NAME>
"""

from . import getSiteMover
from .trace_report import TraceReport

from FileStateClient import updateFileState, dumpFileStates
from PilotErrors import PilotException, PilotErrors

from pUtil import tolog, readpar

import sys
import os
import time
import traceback
from random import shuffle, uniform
from subprocess import Popen, PIPE, STDOUT

try:
    import json # python2.6
except ImportError:
    import simplejson as json


class JobMover(object):
    """
        Top level mover to copy Job files from/to storage.
    """

    job = None  # Job object
    si = None   # Site information

    MAX_STAGEIN_RETRY = 5
    MAX_STAGEOUT_RETRY = 10

    _stageoutretry = 2 # default value
    _stageinretry = 2  # devault value

    _stagein_sleeptime_min = 20       # seconds, min allowed sleep time in case of stagein failure
    _stagein_sleeptime_max = 50       # seconds, max allowed sleep time in case of stagein failure

    _stageout_sleeptime_min = 1*60  # seconds, min allowed sleep time in case of stageout failure
    _stageout_sleeptime_max = 5*60    # seconds, max allowed sleep time in case of stageout failure

    direct_remoteinput_allowed_schemas = ['root']
    remoteinput_allowed_schemas = ['root', 'gsiftp', 'dcap', 'davs', 'srm'] ## extend me later if need


    def __init__(self, job, si, **kwargs):

        self.job = job
        self.si = si

        self.workDir = kwargs.get('workDir', '')

        self.stageoutretry = kwargs.get('stageoutretry', self._stageoutretry)
        self.stageinretry = kwargs.get('stageinretry', self._stageinretry)
        self.protocols = {}
        self.ddmconf = {}
        self.objectstorekeys = {}
        self.trace_report = {}

        self.useTracingService = kwargs.get('useTracingService', self.si.getExperimentObject().useTracingService())


    def log(self, value): # quick stub
        #sys.stdout.flush()
        tolog(value)
        sys.stdout.flush() # quick hack until proper central logging will be implemented

    def calc_stagein_sleeptime(self):
        return uniform(self._stagein_sleeptime_min, self._stagein_sleeptime_max)

    def calc_stageout_sleeptime(self, attempt=1):
        return uniform(self._stageout_sleeptime_min, min(self._stageout_sleeptime_max, (attempt + 1) * self._stageout_sleeptime_min))

    @property
    def stageoutretry(self):
        return self._stageoutretry

    @stageoutretry.setter
    def stageoutretry(self, value):
        if value >= self.MAX_STAGEOUT_RETRY or value < 1:
            ival = value
            value = JobMover._stageoutretry
            self.log("WARNING: Unreasonable number of stage-out tries: %d, reset to default value=%s" % (ival, value))
        self._stageoutretry = value

    @property
    def stageinretry(self):
        return self._stageinretry

    @stageinretry.setter
    def stageinretry(self, value):
        if value >= self.MAX_STAGEIN_RETRY or value < 1:
            ival = value
            value = JobMover._stageinretry
            self.log("WARNING: Unreasonable number of stage-in tries: %d, reset to default value=%s" % (ival, value))
        self._stageinretry = value


    @classmethod
    def _prepare_input_ddm(self, ddm, localddms):
        """
            Sort and filter localddms for given preferred ddm entry
            :return: list of ordered ddmendpoint names
        """
        # set preferred ddm as first source
        # move is_tape ddm to the end

        ddms, tapes = [], []
        for e in localddms:
            ddmendpoint = e['name']
            if ddmendpoint == ddm['name']:
                continue
            if e['is_tape']:
                tapes.append(ddmendpoint)
            else:
                ddms.append(ddmendpoint)

        # randomize input
        shuffle(ddms)

        # TODO: apply more sort rules here

        ddms = [ddm['name']] + ddms + tapes

        return ddms


    def get_pfns(self, replica, protocol='root'):
        """ Extract the PFNs from the replicas dictionary"""

        pfns = {}  # FORMAT: { 'endpoint': [pfn1, ..], .. }

        for pfn in replica['pfns'].keys():
            if pfn.startswith(protocol):
                endpoint = replica['pfns'][pfn]['rse']
                if endpoint in pfns:
                    pfns[endpoint].append(pfn)
                else:
                    pfns[endpoint] = [pfn]
        return pfns


    def get_turl(self, pfns, endpoint=None):
        """ Get a turl from the pfns dictionary """

        turl = ""
        if not endpoint:
            keys = pfns.keys()
            turl = pfns[keys[0]][0]
        else:
            turl = pfns[endpoint][0]
        return turl

    def detect_client_location(self):
        """
        Open a UDP socket to a machine on the internet, to get the local IP address
        of the requesting client.
        Try to determine the sitename automatically from common environment variables,
        in this order: SITE_NAME, ATLAS_SITE_NAME, OSG_SITE_NAME. If none of these exist
        use the fixed string 'ROAMING'.
        Note: this is a modified Rucio function.

        :return: ip, fqdn, site dictionary
        """

        site = readpar('gocname')
        dic = {}
        try:
            import socket
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(("8.8.8.8", 80))
            ip = s.getsockname()[0]
            dic = {'ip': ip,
                   'fqdn': socket.getfqdn(),
                   'site': site}
        except Exception as e:
            self.log('socket() failed to lookup local IP')

        return dic

    def resolve_replicas(self, files):
        """
            populates fdat.replicas of each entry from `files` list
            fdat.replicas = [(ddmendpoint, replica, ddm_se, ddm_path)]
            ddm_se -- integration logic -- is used to manually form TURL
            (quick stub until all protocols are properly populated in Rucio from AGIS)
            :return: `files`
        """

        # build list of local ddmendpoints grouped by site
        ddms = {}
        for ddm, dat in self.ddmconf.iteritems():
            if dat.get('state') != 'ACTIVE': # skip DISABLED ddms
                continue
            ddms.setdefault(dat['site'], []).append(dat)

        xfiles = [] # consider only normal ddmendpoints (skip OS)

        for fdat in files:
            ddmdat = self.ddmconf.get(fdat.ddmendpoint)
            if not ddmdat:
                raise Exception("Failed to resolve input ddmendpoint by name=%s sent by Panda job, please check configuration. fdat=%s" % (fdat.ddmendpoint, fdat))
            if not ddmdat['site']:
                raise Exception("Failed to resolve input site name of ddmendpoint=%s. please check ddm declaration: ddmconf=%s ... fdat=%s" % (fdat.ddmendpoint, ddmconf, fdat))

            if fdat.storageId and fdat.storageId > 0 and ddmdat.get('type') in ['OS_ES']:  ## redundant check by storageId?
                self.log('file(%s:%s) in os ddms(%s), skip resolving replicas' % (fdat.scope, fdat.lfn, fdat.ddmendpoint))
                # skip OS ddms, storageId -1 means normal RSE
                #self.log("fdat.storageId: %s" % fdat.storageId)
                #fdat.inputddms = [fdat.ddmendpoint]         ### is it used for OS?
                continue

            if fdat.storageId in [0, -1]:
                fdat.storageId = None   ### redundant, is it required for further workflow outside the function?

            # build and order list of local ddms
            #localddms = ddms.get(ddmdat['site'])
            # sort and filter ddms (as possible input source)
            #fdat.inputddms = self._prepare_input_ddm(ddmdat, localddms)
            pandaqueue = self.si.getQueueName()
            fdat.inputddms = self.si.resolvePandaAssociatedStorages(pandaqueue).get(pandaqueue, {}).get('pr', {})

            #if fdat.storageId in [-1, None] or ddmdat.type not in ['OS_ES']: ## redundant, double check -- just ported from old workflow ?
            xfiles.append(fdat)

        if not xfiles: # no files for replica look-up
            return files

        # load replicas from Rucio
        from rucio.client import Client
        c = Client()

        ## for the time being until Rucio bug with geo-ip sorting is resolved
        ## do apply either simple query list_replicas() without geoip sort to resolve LAN replicas in case of directaccesstype=[None, LAN]
        # otherwise in case of directaccesstype=WAN mode do query geo sorted list_replicas() with location data passed

        bquery = {'schemes':['srm', 'root', 'davs', 'gsiftp', 'https'],
                  'dids': [dict(scope=e.scope, name=e.lfn) for e in xfiles]
                 }

        allowRemoteInputs = True in set(e.allowRemoteInputs for e in xfiles)

        try:
            query = bquery.copy()
            if allowRemoteInputs:
                location = self.detect_client_location()
                if not location:
                    raise Exception("Failed to get client location")
                query.update(sort='geoip', client_location=location)
                # query.update(sort='geoip', client_location=location, domain='lan') # remove lan again after testing

            try:
                self.log('Call rucio.list_replicas() with query=%s' % query)
                replicas = c.list_replicas(**query)
            except TypeError, e:
                if query == bquery:
                    raise
                self.log("WARNING: Detected outdated Rucio list_replicas(), cannot do geoip-sorting: %s .. fallback to old list_replicas() call" % e)
                replicas = c.list_replicas(**bquery)

        except Exception, e:
            raise PilotException("Failed to get replicas from Rucio: %s" % e, code=PilotErrors.ERR_FAILEDLFCGETREPS)

        replicas = list(replicas)
        self.log("replicas received from Rucio: %s" % replicas)

        files_lfn = dict(((e.scope, e.lfn), e) for e in xfiles)
        #self.log("files_lfn=%s" % files_lfn)

        for r in replicas:
            k = r['scope'], r['name']
            fdat = files_lfn.get(k)
            if not fdat: # not requested replica returned?
                continue

            fdat.replicas = [] # reset replicas list

            def get_preferred_replica(replicas, allowed_schemas):
                for schema in allowed_schemas:
                    for replica in replicas:
                        if replica and replica.startswith('%s://' % schema):
                            return replica
                return None

            has_direct_remoteinput_replicas = False

            # local replicas
            for ddm in fdat.inputddms: ## iterate over local ddms and check if replica is exist here

                if ddm not in r['rses']: # no replica found for given local ddm
                    continue

                ddm_se, ddm_path = '',''
                def_protocol = self.ddmconf[ddm].get('aprotocols', {}).get('r', [])  ## fix me later: use 'read_lan' then fallback to 'read_wan'
                def_protocol = def_protocol[0] if def_protocol else None  ## take first entry
                if def_protocol:
                    ddm_se, ddm_path = def_protocol[0], def_protocol[2]

                fdat.replicas.append((ddm, r['rses'][ddm], ddm_se, ddm_path))

                if not has_direct_remoteinput_replicas:
                    has_direct_remoteinput_replicas = bool(get_preferred_replica(r['rses'][ddm], self.direct_remoteinput_allowed_schemas))

            if ((not fdat.replicas or ( fdat.accessmode == 'direct' and not has_direct_remoteinput_replicas)) and fdat.allowRemoteInputs) or fdat.storageId > 0:
                if fdat.accessmode == 'direct':
                    allowed_schemas = self.direct_remoteinput_allowed_schemas
                else:
                    allowed_schemas = self.remoteinput_allowed_schemas
                self.log("No local replicas found for lfn=%s or direct access is set but no local direct access files, but allowRemoteInputs is set, looking for remote inputs" % (fdat.lfn))
                self.log("consider first/closest replica, accessmode=%s, remoteinput_allowed_schemas=%s" % (fdat.accessmode, allowed_schemas))
                #self.log('rses=%s' % r['rses'])
                for ddm, replicas in r['rses'].iteritems():
                    replica = get_preferred_replica(r['rses'][ddm], self.remoteinput_allowed_schemas)
                    if not replica:
                        continue

                    ddm_se, ddm_path = '', ''

                    # remoteinput supported replica (root) replica has been found
                    fdat.replicas.append((ddm, r['rses'][ddm], ddm_se, ddm_path))
                    # break # ignore other remote replicas/sites

            # verify filesize and checksum values

            if fdat.filesize in [None, 'NULL', '', 0]:
                self.log("WARNING: filesize is not defined, assigning info got from Rucio to it.")
                fdat.filesize = r['bytes']
            elif fdat.filesize != r['bytes']:
                self.log("WARNING: filesize value of input file=%s mismatched with info got from Rucio replica:  job.indata.filesize=%s, replica.filesize=%s, fdat=%s" % (fdat.lfn, fdat.filesize, r['bytes'], fdat))

            cc_ad = 'ad:%s' % r['adler32']
            cc_md = 'md:%s' % r['md5']
            if fdat.checksum in [None, 'NULL', '']:
                self.log("WARNING: checksum is not defined, assigning info got from Rucio to it.")
                if r['adler32']:
                    fdat.checksum = cc_ad
                elif r['md5']:
                    fdat.checksum = cc_md
            elif fdat.checksum not in [cc_ad, cc_md]:
                self.log("WARNING: checksum value of input file=%s mismatched with info got from Rucio replica:  job.indata.checksum=%s, replica.checksum=%s, fdat=%s" % (fdat.lfn, fdat.checksum, (cc_ad, cc_md), fdat))

        self.log('Number of resolved replicas:\n' + '\n'.join(["lfn=%s: replicas=%s, allowRemoteInputs=%s, is_directaccess=%s" % (e.lfn, len(e.replicas), e.allowRemoteInputs,  e.is_directaccess(ensure_replica=False)) for e in files]))

        return files


    def get_directaccess(self):
        """
            Check if direct access I/O is allowed by Site. It actually exposes direct_access_lan/wan schedconfig settings
            Also return the directaccess type (WAN or LAN).
            quick workaround: should be properly implemented in SiteInformation
            :return: (is_directaccess, directaccesstype)
        """

        try:
            from FileHandling import getDirectAccess
            return getDirectAccess()
        except Exception, e:
            self.log("mover.is_directaccess(): Failed to resolve direct access settings: exception=%s" % e)
            return False, None

    def handle_dbreleases(self, files):
        """
            Check if the DBRelease file(s) is locally available,
            if so, create the skeleton file and exclude file from stage-in
        """

        from DBReleaseHandler import DBReleaseHandler  ## partially reuse old implementation.. FIX ME LATER

        dbh = DBReleaseHandler(workdir=self.job.workdir)
        path = self.job.workdir

        # resolve dbreleases from files
        dbreleases = {} # by versions
        for fspec in files:
            ver = dbh.extractVersion(fspec.lfn)  ##
            if ver:
                dbreleases.setdefault(ver, []).append(fspec)

        # resolve dbrelease fom jobPars
        version = dbh.extractVersion(self.job.jobPars) ## quick hack -- fix me later (format of jobPars differs from lfn)
        if version:
            ## check if available or requested for stage-in
            if not dbh.isDBReleaseAvailable(version) and not version in dbreleases: # not available
                self.log("ERROR: Requested DBRelease from jobPars with version=%s is not available on CVMFS and not requested for stage in.. raise error." % version)
                raise PilotException("Requested DBRelease is not available: version=%s" % version, code=PilotErrors.ERR_MISSDBREL, state='ERR_MISSDBREL')
            dbreleases.setdefault(version, [])

        if not dbreleases:
            return False

        self.log("Detected DBRelease versions extracted from JobPars and input files: %s" % sorted(dbreleases))
        for version, fspecs in dbreleases.iteritems():
            if dbh.isDBReleaseAvailable(version):
                self.log("Creating the skeleton DBRelease tarball for version=%s" % version)
                if not dbh.createDBRelease(version, path):
                    self.log("Failed to create the skeleton file for DBRelease version=%s" % version)
                    raise PilotException("DBRelease: failed to create skeleton for version=%s" % version, code=PilotErrors.ERR_MISSDBREL, state='MISSDBREL_SKELFAIL')
                for fspec in fspecs:
                    fspec.status = 'no_transfer'  ## ignore for stage-in

        return True

    def get_objectstore_keys(self, ddmendpoint):
        if ddmendpoint in self.objectstorekeys:
            return

        endpoint_id = self.si.getObjectstoreEndpointID(ddmendpoint=ddmendpoint, label='read_wan', protocol='s3')
        os_access_key, os_secret_key, os_is_secure = self.si.getObjectstoreKeyInfo(endpoint_id, ddmendpoint=ddmendpoint)

        if os_access_key and os_access_key != "" and os_secret_key and os_secret_key != "":
            keyPair = self.si.getSecurityKey(os_secret_key, os_access_key)
            if "privateKey" not in keyPair or keyPair["privateKey"] is None:
                self.log("Failed to get the keyPair for S3 objectstore from panda")
                self.objectstorekeys[ddmendpoint] = {'status': False}
            else:
                self.objectstorekeys[ddmendpoint] = {'S3_ACCESS_KEY': keyPair["publicKey"],
                                                     'S3_SECRET_KEY': keyPair["privateKey"],
                                                     'S3_IS_SECURE': os_is_secure,
                                                     'status': True}
        else:
            self.log("Failed to get the keyPair name for S3 objectstore from ddm config")
            self.objectstorekeys[ddmendpoint] = {'status': False}

    def is_local_storage(self, ddmendpoint, activity='pr'):
        pandaqueue = self.si.getQueueName() # FIX ME LATER
        associate_storages = self.si.resolvePandaAssociatedStorages(pandaqueue).get(pandaqueue, {})
        ddmendpoints = associate_storages.get(activity, [])
        return ddmendpoint in ddmendpoints

    def stagein(self, files=None, analyjob=False):
        """
            :return: (transferred_files, failed_transfers)
        """

        if files is None:
            files = self.job.inData
        _files = files

        # only print out 'files' for debugging purposes since it might contain explicit pandaProxySecretKey values
        # self.log("To stagein files: %s" % files)

        normal_files, es_files, es_local_files = [], [], []
        transferred_files, failed_transfers = [], []

        self.ddmconf.update(self.si.resolveDDMConf([])) # load ALL ddmconf

        # resolve OS ddmendpoint
        os_ddms = {}
        for ddm, dat in self.ddmconf.iteritems():
            if dat.get('type') in ['OS_ES', 'OS_LOGS']:
                os_ddms.setdefault(int(dat.get('resource', {}).get('bucket_id', -1)), ddm)
        for ddm, dat in self.ddmconf.iteritems():
            os_ddms.setdefault(int(dat.get('id', -1)), ddm)

        for fspec in files:
            if fspec.pathConvention and fspec.pathConvention >= 1000:
                fspec.scope = 'transient'
            if fspec.prodDBlockToken and fspec.prodDBlockToken.isdigit() and int(fspec.prodDBlockToken) > 0:
                fspec.storageId = int(fspec.prodDBlockToken)
                fspec.ddmendpoint = os_ddms.get(fspec.storageId)
                self.get_objectstore_keys(fspec.ddmendpoint)
                es_files.append(fspec)
                if self.is_local_storage(fspec.ddmendpoint, activity='pr'):
                    es_local_files.append(fspec)
            elif fspec.prodDBlockToken and (fspec.prodDBlockToken.strip() == '-1' or fspec.prodDBlockToken.strip() == '0'):
                # es outputs in normal RSEs (not in objectstore) and registered in rucio
                fspec.allowRemoteInputs = True
                fspec.storageId = None  # resolve replicas needs to be called for it
                es_files.append(fspec)
            else:
                normal_files.append(fspec)

        allowRemoteInputs = True in set(e.allowRemoteInputs for e in normal_files)
        self.log("AllowRemoteInputs: %s" % allowRemoteInputs)
        if normal_files:
            self.log("Will stagin normal files: %s" % [f.lfn for f in normal_files])
            transferred_files, failed_transfers = self.stagein_real(files=normal_files, activity='pr', analyjob=analyjob, skip_transfer_failure=allowRemoteInputs)

        remain_normal_files = [e for e in normal_files if e.status not in ['remote_io', 'transferred', 'no_transfer']]
        self.log("Remain normal files: %s" % (remain_normal_files))
        if remain_normal_files and allowRemoteInputs:
            self.log("Will stagin remain normal files with remote inputs: %s, allowRemoteInputs: %s" % ([f.lfn for f in remain_normal_files], allowRemoteInputs))
            copytools = [('rucio', {'setup': ''})]
            transferred_files_normal, failed_transfers_normal = self.stagein_real(files=normal_files, activity='pr_remote', copytools=copytools, analyjob=analyjob)
            transferred_files += transferred_files_normal
            failed_transfers += failed_transfers_normal
            self.log("Failed to transfer files: %s" % failed_transfers)


        if es_local_files:
            self.log("Will stagin es local files: %s" % [f.lfn for f in es_local_files])
            self.trace_report.update(eventType='get_es')
            transferred_files_es, failed_transfers_es = self.stagein_real(files=es_local_files, activity='pr', analyjob=analyjob)
            transferred_files += transferred_files_es
            failed_transfers += failed_transfers_es
            self.log("Failed to transfer files: %s" % failed_transfers)

        remain_es_files = [e for e in es_files if e.status not in ['remote_io', 'transferred', 'no_transfer']]
        if remain_es_files:<commit_msg>fix for es merge stagein<commit_after>"""
  JobMover: top level mover to copy all Job files

  This functionality actually needs to be incorporated directly into RunJob,
  because it's just a workflow of RunJob
  but since RunJob at the moment is implemented as a Singleton
  it could be dangerous because of Job instance stored in the class object.
  :author: <NAME>
"""

from . import getSiteMover
from .trace_report import TraceReport

from FileStateClient import updateFileState, dumpFileStates
from PilotErrors import PilotException, PilotErrors

from pUtil import tolog, readpar

import sys
import os
import time
import traceback
from random import shuffle, uniform
from subprocess import Popen, PIPE, STDOUT

try:
    import json # python2.6
except ImportError:
    import simplejson as json


class JobMover(object):
    """
        Top level mover to copy Job files from/to storage.
    """

    job = None  # Job object
    si = None   # Site information

    MAX_STAGEIN_RETRY = 5
    MAX_STAGEOUT_RETRY = 10

    _stageoutretry = 2 # default value
    _stageinretry = 2  # devault value

    _stagein_sleeptime_min = 20       # seconds, min allowed sleep time in case of stagein failure
    _stagein_sleeptime_max = 50       # seconds, max allowed sleep time in case of stagein failure

    _stageout_sleeptime_min = 1*60  # seconds, min allowed sleep time in case of stageout failure
    _stageout_sleeptime_max = 5*60    # seconds, max allowed sleep time in case of stageout failure

    direct_remoteinput_allowed_schemas = ['root']
    remoteinput_allowed_schemas = ['root', 'gsiftp', 'dcap', 'davs', 'srm'] ## extend me later if need


    def __init__(self, job, si, **kwargs):

        self.job = job
        self.si = si

        self.workDir = kwargs.get('workDir', '')

        self.stageoutretry = kwargs.get('stageoutretry', self._stageoutretry)
        self.stageinretry = kwargs.get('stageinretry', self._stageinretry)
        self.protocols = {}
        self.ddmconf = {}
        self.objectstorekeys = {}
        self.trace_report = {}

        self.useTracingService = kwargs.get('useTracingService', self.si.getExperimentObject().useTracingService())


    def log(self, value): # quick stub
        #sys.stdout.flush()
        tolog(value)
        sys.stdout.flush() # quick hack until proper central logging will be implemented

    def calc_stagein_sleeptime(self):
        return uniform(self._stagein_sleeptime_min, self._stagein_sleeptime_max)

    def calc_stageout_sleeptime(self, attempt=1):
        return uniform(self._stageout_sleeptime_min, min(self._stageout_sleeptime_max, (attempt + 1) * self._stageout_sleeptime_min))

    @property
    def stageoutretry(self):
        return self._stageoutretry

    @stageoutretry.setter
    def stageoutretry(self, value):
        if value >= self.MAX_STAGEOUT_RETRY or value < 1:
            ival = value
            value = JobMover._stageoutretry
            self.log("WARNING: Unreasonable number of stage-out tries: %d, reset to default value=%s" % (ival, value))
        self._stageoutretry = value

    @property
    def stageinretry(self):
        return self._stageinretry

    @stageinretry.setter
    def stageinretry(self, value):
        if value >= self.MAX_STAGEIN_RETRY or value < 1:
            ival = value
            value = JobMover._stageinretry
            self.log("WARNING: Unreasonable number of stage-in tries: %d, reset to default value=%s" % (ival, value))
        self._stageinretry = value


    @classmethod
    def _prepare_input_ddm(self, ddm, localddms):
        """
            Sort and filter localddms for given preferred ddm entry
            :return: list of ordered ddmendpoint names
        """
        # set preferred ddm as first source
        # move is_tape ddm to the end

        ddms, tapes = [], []
        for e in localddms:
            ddmendpoint = e['name']
            if ddmendpoint == ddm['name']:
                continue
            if e['is_tape']:
                tapes.append(ddmendpoint)
            else:
                ddms.append(ddmendpoint)

        # randomize input
        shuffle(ddms)

        # TODO: apply more sort rules here

        ddms = [ddm['name']] + ddms + tapes

        return ddms


    def get_pfns(self, replica, protocol='root'):
        """ Extract the PFNs from the replicas dictionary"""

        pfns = {}  # FORMAT: { 'endpoint': [pfn1, ..], .. }

        for pfn in replica['pfns'].keys():
            if pfn.startswith(protocol):
                endpoint = replica['pfns'][pfn]['rse']
                if endpoint in pfns:
                    pfns[endpoint].append(pfn)
                else:
                    pfns[endpoint] = [pfn]
        return pfns


    def get_turl(self, pfns, endpoint=None):
        """ Get a turl from the pfns dictionary """

        turl = ""
        if not endpoint:
            keys = pfns.keys()
            turl = pfns[keys[0]][0]
        else:
            turl = pfns[endpoint][0]
        return turl

    def detect_client_location(self):
        """
        Open a UDP socket to a machine on the internet, to get the local IP address
        of the requesting client.
        Try to determine the sitename automatically from common environment variables,
        in this order: SITE_NAME, ATLAS_SITE_NAME, OSG_SITE_NAME. If none of these exist
        use the fixed string 'ROAMING'.
        Note: this is a modified Rucio function.

        :return: ip, fqdn, site dictionary
        """

        site = readpar('gocname')
        dic = {}
        try:
            import socket
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(("8.8.8.8", 80))
            ip = s.getsockname()[0]
            dic = {'ip': ip,
                   'fqdn': socket.getfqdn(),
                   'site': site}
        except Exception as e:
            self.log('socket() failed to lookup local IP')

        return dic

    def resolve_replicas(self, files):
        """
            populates fdat.replicas of each entry from `files` list
            fdat.replicas = [(ddmendpoint, replica, ddm_se, ddm_path)]
            ddm_se -- integration logic -- is used to manually form TURL
            (quick stub until all protocols are properly populated in Rucio from AGIS)
            :return: `files`
        """

        # build list of local ddmendpoints grouped by site
        ddms = {}
        for ddm, dat in self.ddmconf.iteritems():
            if dat.get('state') != 'ACTIVE': # skip DISABLED ddms
                continue
            ddms.setdefault(dat['site'], []).append(dat)

        xfiles = [] # consider only normal ddmendpoints (skip OS)

        for fdat in files:
            ddmdat = self.ddmconf.get(fdat.ddmendpoint)
            if not ddmdat:
                raise Exception("Failed to resolve input ddmendpoint by name=%s sent by Panda job, please check configuration. fdat=%s" % (fdat.ddmendpoint, fdat))
            if not ddmdat['site']:
                raise Exception("Failed to resolve input site name of ddmendpoint=%s. please check ddm declaration: ddmconf=%s ... fdat=%s" % (fdat.ddmendpoint, ddmconf, fdat))

            if fdat.storageId and fdat.storageId > 0 and ddmdat.get('type') in ['OS_ES']:  ## redundant check by storageId?
                self.log('file(%s:%s) in os ddms(%s), skip resolving replicas' % (fdat.scope, fdat.lfn, fdat.ddmendpoint))
                # skip OS ddms, storageId -1 means normal RSE
                #self.log("fdat.storageId: %s" % fdat.storageId)
                #fdat.inputddms = [fdat.ddmendpoint]         ### is it used for OS?
                continue

            if fdat.storageId in [0, -1]:
                fdat.storageId = None   ### redundant, is it required for further workflow outside the function?

            # build and order list of local ddms
            #localddms = ddms.get(ddmdat['site'])
            # sort and filter ddms (as possible input source)
            #fdat.inputddms = self._prepare_input_ddm(ddmdat, localddms)
            pandaqueue = self.si.getQueueName()
            fdat.inputddms = self.si.resolvePandaAssociatedStorages(pandaqueue).get(pandaqueue, {}).get('pr', {})

            #if fdat.storageId in [-1, None] or ddmdat.type not in ['OS_ES']: ## redundant, double check -- just ported from old workflow ?
            xfiles.append(fdat)

        if not xfiles: # no files for replica look-up
            return files

        # load replicas from Rucio
        from rucio.client import Client
        c = Client()

        ## for the time being until Rucio bug with geo-ip sorting is resolved
        ## do apply either simple query list_replicas() without geoip sort to resolve LAN replicas in case of directaccesstype=[None, LAN]
        # otherwise in case of directaccesstype=WAN mode do query geo sorted list_replicas() with location data passed

        bquery = {'schemes':['srm', 'root', 'davs', 'gsiftp', 'https'],
                  'dids': [dict(scope=e.scope, name=e.lfn) for e in xfiles]
                 }

        allowRemoteInputs = True in set(e.allowRemoteInputs for e in xfiles)

        try:
            query = bquery.copy()
            if allowRemoteInputs:
                location = self.detect_client_location()
                if not location:
                    raise Exception("Failed to get client location")
                query.update(sort='geoip', client_location=location)
                # query.update(sort='geoip', client_location=location, domain='lan') # remove lan again after testing

            try:
                self.log('Call rucio.list_replicas() with query=%s' % query)
                replicas = c.list_replicas(**query)
            except TypeError, e:
                if query == bquery:
                    raise
                self.log("WARNING: Detected outdated Rucio list_replicas(), cannot do geoip-sorting: %s .. fallback to old list_replicas() call" % e)
                replicas = c.list_replicas(**bquery)

        except Exception, e:
            raise PilotException("Failed to get replicas from Rucio: %s" % e, code=PilotErrors.ERR_FAILEDLFCGETREPS)

        replicas = list(replicas)
        self.log("replicas received from Rucio: %s" % replicas)

        files_lfn = dict(((e.scope, e.lfn), e) for e in xfiles)
        #self.log("files_lfn=%s" % files_lfn)

        for r in replicas:
            k = r['scope'], r['name']
            fdat = files_lfn.get(k)
            if not fdat: # not requested replica returned?
                continue

            fdat.replicas = [] # reset replicas list

            def get_preferred_replica(replicas, allowed_schemas):
                for schema in allowed_schemas:
                    for replica in replicas:
                        if replica and replica.startswith('%s://' % schema):
                            return replica
                return None

            has_direct_remoteinput_replicas = False

            # local replicas
            for ddm in fdat.inputddms: ## iterate over local ddms and check if replica is exist here

                if ddm not in r['rses']: # no replica found for given local ddm
                    continue

                ddm_se, ddm_path = '',''
                def_protocol = self.ddmconf[ddm].get('aprotocols', {}).get('r', [])  ## fix me later: use 'read_lan' then fallback to 'read_wan'
                def_protocol = def_protocol[0] if def_protocol else None  ## take first entry
                if def_protocol:
                    ddm_se, ddm_path = def_protocol[0], def_protocol[2]

                fdat.replicas.append((ddm, r['rses'][ddm], ddm_se, ddm_path))

                if not has_direct_remoteinput_replicas:
                    has_direct_remoteinput_replicas = bool(get_preferred_replica(r['rses'][ddm], self.direct_remoteinput_allowed_schemas))

            if ((not fdat.replicas or ( fdat.accessmode == 'direct' and not has_direct_remoteinput_replicas)) and fdat.allowRemoteInputs) or (not fdat.replicas and fdat.storageId >= 0):
                if fdat.accessmode == 'direct':
                    allowed_schemas = self.direct_remoteinput_allowed_schemas
                else:
                    allowed_schemas = self.remoteinput_allowed_schemas
                self.log("No local replicas found for lfn=%s or direct access is set but no local direct access files, but allowRemoteInputs is set, looking for remote inputs" % (fdat.lfn))
                self.log("consider first/closest replica, accessmode=%s, remoteinput_allowed_schemas=%s" % (fdat.accessmode, allowed_schemas))
                #self.log('rses=%s' % r['rses'])
                for ddm, replicas in r['rses'].iteritems():
                    replica = get_preferred_replica(r['rses'][ddm], self.remoteinput_allowed_schemas)
                    if not replica:
                        continue

                    ddm_se, ddm_path = '', ''

                    # remoteinput supported replica (root) replica has been found
                    fdat.replicas.append((ddm, r['rses'][ddm], ddm_se, ddm_path))
                    # break # ignore other remote replicas/sites

            # verify filesize and checksum values

            if fdat.filesize in [None, 'NULL', '', 0]:
                self.log("WARNING: filesize is not defined, assigning info got from Rucio to it.")
                fdat.filesize = r['bytes']
            elif fdat.filesize != r['bytes']:
                self.log("WARNING: filesize value of input file=%s mismatched with info got from Rucio replica:  job.indata.filesize=%s, replica.filesize=%s, fdat=%s" % (fdat.lfn, fdat.filesize, r['bytes'], fdat))

            cc_ad = 'ad:%s' % r['adler32']
            cc_md = 'md:%s' % r['md5']
            if fdat.checksum in [None, 'NULL', '']:
                self.log("WARNING: checksum is not defined, assigning info got from Rucio to it.")
                if r['adler32']:
                    fdat.checksum = cc_ad
                elif r['md5']:
                    fdat.checksum = cc_md
            elif fdat.checksum not in [cc_ad, cc_md]:
                self.log("WARNING: checksum value of input file=%s mismatched with info got from Rucio replica:  job.indata.checksum=%s, replica.checksum=%s, fdat=%s" % (fdat.lfn, fdat.checksum, (cc_ad, cc_md), fdat))

        self.log('Number of resolved replicas:\n' + '\n'.join(["lfn=%s: replicas=%s, allowRemoteInputs=%s, is_directaccess=%s" % (e.lfn, len(e.replicas), e.allowRemoteInputs,  e.is_directaccess(ensure_replica=False)) for e in files]))

        return files


    def get_directaccess(self):
        """
            Check if direct access I/O is allowed by Site. It actually exposes direct_access_lan/wan schedconfig settings
            Also return the directaccess type (WAN or LAN).
            quick workaround: should be properly implemented in SiteInformation
            :return: (is_directaccess, directaccesstype)
        """

        try:
            from FileHandling import getDirectAccess
            return getDirectAccess()
        except Exception, e:
            self.log("mover.is_directaccess(): Failed to resolve direct access settings: exception=%s" % e)
            return False, None

    def handle_dbreleases(self, files):
        """
            Check if the DBRelease file(s) is locally available,
            if so, create the skeleton file and exclude file from stage-in
        """

        from DBReleaseHandler import DBReleaseHandler  ## partially reuse old implementation.. FIX ME LATER

        dbh = DBReleaseHandler(workdir=self.job.workdir)
        path = self.job.workdir

        # resolve dbreleases from files
        dbreleases = {} # by versions
        for fspec in files:
            ver = dbh.extractVersion(fspec.lfn)  ##
            if ver:
                dbreleases.setdefault(ver, []).append(fspec)

        # resolve dbrelease fom jobPars
        version = dbh.extractVersion(self.job.jobPars) ## quick hack -- fix me later (format of jobPars differs from lfn)
        if version:
            ## check if available or requested for stage-in
            if not dbh.isDBReleaseAvailable(version) and not version in dbreleases: # not available
                self.log("ERROR: Requested DBRelease from jobPars with version=%s is not available on CVMFS and not requested for stage in.. raise error." % version)
                raise PilotException("Requested DBRelease is not available: version=%s" % version, code=PilotErrors.ERR_MISSDBREL, state='ERR_MISSDBREL')
            dbreleases.setdefault(version, [])

        if not dbreleases:
            return False

        self.log("Detected DBRelease versions extracted from JobPars and input files: %s" % sorted(dbreleases))
        for version, fspecs in dbreleases.iteritems():
            if dbh.isDBReleaseAvailable(version):
                self.log("Creating the skeleton DBRelease tarball for version=%s" % version)
                if not dbh.createDBRelease(version, path):
                    self.log("Failed to create the skeleton file for DBRelease version=%s" % version)
                    raise PilotException("DBRelease: failed to create skeleton for version=%s" % version, code=PilotErrors.ERR_MISSDBREL, state='MISSDBREL_SKELFAIL')
                for fspec in fspecs:
                    fspec.status = 'no_transfer'  ## ignore for stage-in

        return True

    def get_objectstore_keys(self, ddmendpoint):
        if ddmendpoint in self.objectstorekeys:
            return

        endpoint_id = self.si.getObjectstoreEndpointID(ddmendpoint=ddmendpoint, label='read_wan', protocol='s3')
        os_access_key, os_secret_key, os_is_secure = self.si.getObjectstoreKeyInfo(endpoint_id, ddmendpoint=ddmendpoint)

        if os_access_key and os_access_key != "" and os_secret_key and os_secret_key != "":
            keyPair = self.si.getSecurityKey(os_secret_key, os_access_key)
            if "privateKey" not in keyPair or keyPair["privateKey"] is None:
                self.log("Failed to get the keyPair for S3 objectstore from panda")
                self.objectstorekeys[ddmendpoint] = {'status': False}
            else:
                self.objectstorekeys[ddmendpoint] = {'S3_ACCESS_KEY': keyPair["publicKey"],
                                                     'S3_SECRET_KEY': keyPair["privateKey"],
                                                     'S3_IS_SECURE': os_is_secure,
                                                     'status': True}
        else:
            self.log("Failed to get the keyPair name for S3 objectstore from ddm config")
            self.objectstorekeys[ddmendpoint] = {'status': False}

    def is_local_storage(self, ddmendpoint, activity='pr'):
        pandaqueue = self.si.getQueueName() # FIX ME LATER
        associate_storages = self.si.resolvePandaAssociatedStorages(pandaqueue).get(pandaqueue, {})
        ddmendpoints = associate_storages.get(activity, [])
        return ddmendpoint in ddmendpoints

    def stagein(self, files=None, analyjob=False):
        """
            :return: (transferred_files, failed_transfers)
        """

        if files is None:
            files = self.job.inData
        _files = files

        # only print out 'files' for debugging purposes since it might contain explicit pandaProxySecretKey values
        # self.log("To stagein files: %s" % files)

        normal_files, es_files, es_local_files = [], [], []
        transferred_files, failed_transfers = [], []

        self.ddmconf.update(self.si.resolveDDMConf([])) # load ALL ddmconf

        # resolve OS ddmendpoint
        os_ddms = {}
        for ddm, dat in self.ddmconf.iteritems():
            if dat.get('type') in ['OS_ES', 'OS_LOGS']:
                os_ddms.setdefault(int(dat.get('resource', {}).get('bucket_id', -1)), ddm)
        for ddm, dat in self.ddmconf.iteritems():
            os_ddms.setdefault(int(dat.get('id', -1)), ddm)

        for fspec in files:
            if fspec.pathConvention and fspec.pathConvention >= 1000:
                fspec.scope = 'transient'
            if fspec.prodDBlockToken and fspec.prodDBlockToken.isdigit() and int(fspec.prodDBlockToken) > 0:
                fspec.storageId = int(fspec.prodDBlockToken)
                fspec.ddmendpoint = os_ddms.get(fspec.storageId)
                self.get_objectstore_keys(fspec.ddmendpoint)
                es_files.append(fspec)
                if self.is_local_storage(fspec.ddmendpoint, activity='pr'):
                    es_local_files.append(fspec)
            elif fspec.prodDBlockToken and (fspec.prodDBlockToken.strip() == '-1' or fspec.prodDBlockToken.strip() == '0'):
                # es outputs in normal RSEs (not in objectstore) and registered in rucio
                fspec.allowRemoteInputs = True
                fspec.storageId = None  # resolve replicas needs to be called for it
                es_files.append(fspec)
            else:
                normal_files.append(fspec)

        allowRemoteInputs = True in set(e.allowRemoteInputs for e in normal_files)
        self.log("AllowRemoteInputs: %s" % allowRemoteInputs)
        if normal_files:
            self.log("Will stagin normal files: %s" % [f.lfn for f in normal_files])
            transferred_files, failed_transfers = self.stagein_real(files=normal_files, activity='pr', analyjob=analyjob, skip_transfer_failure=allowRemoteInputs)

        remain_normal_files = [e for e in normal_files if e.status not in ['remote_io', 'transferred', 'no_transfer']]
        self.log("Remain normal files: %s" % (remain_normal_files))
        if remain_normal_files and allowRemoteInputs:
            self.log("Will stagin remain normal files with remote inputs: %s, allowRemoteInputs: %s" % ([f.lfn for f in remain_normal_files], allowRemoteInputs))
            copytools = [('rucio', {'setup': ''})]
            transferred_files_normal, failed_transfers_normal = self.stagein_real(files=normal_files, activity='pr_remote', copytools=copytools, analyjob=analyjob)
            transferred_files += transferred_files_normal
            failed_transfers += failed_transfers_normal
            self.log("Failed to transfer files: %s" % failed_transfers)


        if es_local_files:
            try:
                self.log("Will stagin es local files: %s" % [f.lfn for f in es_local_files])
                self.trace_report.update(eventType='get_es')
                transferred_files_es, failed_transfers_es = self.stagein_real(files=es_local_files, activity='pr', analyjob=analyjob)
                transferred_files += transferred_files_es
                failed_transfers += failed_transfers_es
                self.log("Failed to transfer files: %s" % failed_transfers)
            except Exception, e:
                self.log("Failed to stagein eventservice local files: %s" % traceback.format_exc())

        remain_es_files = [e for e in es_files if e.status not in ['remote_io', 'transferred', 'no_transfer']]
        if remain_es_files:
{'fraction_non_alphanumeric': 0.0946252508507111, 'fraction_numerical': 0.002748451269522729, 'mean_word_length': 2.2537260468417317, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6151259', 'n_tokens_mistral': 13679, 'n_tokens_neox': 13045, 'n_words': 3728}