<commit_before>import time
import datetime
import uuid
import os
import sys
import stat
import os.path
import threading
import tarfile
import hashlib
import string
import shutil
import errno

from future.utils import iteritems

from rucio.client import Client as RucioClient
from rucio.common.exception import DataIdentifierNotFound, DuplicateRule, DataIdentifierAlreadyExists, \
    FileAlreadyExists


from pandaharvester.harvestercore import core_utils
from pandaharvester.harvestercore.plugin_base import PluginBase
from pandaharvester.harvesterconfig import harvester_config
from pandaharvester.harvestermover import mover_utils
from pandaharvester.harvestercore.queue_config_mapper import QueueConfigMapper
from pandaharvester.harvesterstager.base_stager import BaseStager

from .base_stager import BaseStager

def dump(tmpLog,obj):
   for attr in dir(obj):
       if hasattr( obj, attr ):
           tmpLog.debug( "obj.%s = %s" % (attr, getattr(obj, attr)))



# logger
baseLogger = core_utils.setup_logger('yoda_rucio_rse_direct_stager')

class Error(EnvironmentError):
    pass

class SpecialFileError(EnvironmentError):
    """Raised when trying to do a kind of operation (e.g. copying) which is
    not supported on a special file (e.g. a named pipe)"""

class ExecError(EnvironmentError):
    """Raised when a command could not be executed"""



# stager plugin with RSE + local site move behaviour for Yoda zip files
class YodaRucioRseDirectStager(BaseStager):
    """In the workflow for RseDirectStager, workers directly upload output files to RSE
    and thus there is no data motion in Harvester."""
    # constructor
    def __init__(self, **kwarg):
        BaseStager.__init__(self, **kwarg)
        tmpLog = self.make_logger(baseLogger, 'ThreadID={0}'.format(threading.current_thread().ident),
                                  method_name='YodaRucioRseDirectStager __init__ ')
        tmpLog.debug('start')
        self.Yodajob = False 
        self.pathConvention = None
        self.objstoreID = None
        tmpLog.debug('stop')

    # set FileSpec.objstoreID 
    def set_FileSpec_objstoreID(self,jobspec, objstoreID, pathConvention):
        # loop over all output files
        for fileSpec in jobspec.outFiles:
            fileSpec.objstoreID = objstoreID
            fileSpec.pathConvention = pathConvention

    # set FileSpec.status 
    def set_FileSpec_status(self,jobspec,status):
        # loop over all output files
        for fileSpec in jobspec.outFiles:
            fileSpec.status = status

    # check status
    def check_status(self, jobspec):
        tmpStat = True
        tmpMsg = ''
        # make logger
        tmpLog = self.make_logger(baseLogger, 'PandaID={0} ThreadID={1}'.format(jobspec.PandaID,threading.current_thread().ident),
                                  method_name='check_status')
        tmpLog.debug('start')
        # Get the files grouped by Rucio Rule ID 
        groups = jobspec.get_groups_of_output_files()
        if len(groups) == 0:
            tmpLog.debug('No Rucio Rules')
            return None,'No Rucio Rules'
        tmpLog.debug('#Rucio Rules - {0} - Rules - {1}'.format(len(groups),groups)) 
        
        try:
            rucioAPI = RucioClient()
        except:
            tmpLog.error('failure to get Rucio Client try again later')
            return None,'failure to get Rucio Client try again later'

        # loop over the Rucio rules 
        for rucioRule in groups:
            if rucioRule is None:
                continue
            # lock
            have_db_lock = self.dbInterface.get_object_lock(rucioRule, lock_interval=120)
            if not have_db_lock:
                msgStr = 'escape since {0} is locked by another thread'.format(rucioRule)
                tmpLog.debug(msgStr)
                return None, msgStr
            # get transfer status
            groupStatus = self.dbInterface.get_file_group_status(rucioRule)
            tmpLog.debug('rucioRule - {0} - groupStatus - {1}'.format(rucioRule,groupStatus))
            if 'transferred' in groupStatus:
                # already succeeded
                pass
            elif 'failed' in groupStatus :
                # transfer failure
                tmpStat = False
                tmpMsg = 'rucio rule for {0}:{1} already failed'.format(datasetScope, datasetName)
            elif 'transferring' in groupStatus or 'pending' in groupStatus:
                # transfer started in Rucio check status
                try:
                    result = rucioAPI.get_replication_rule(rucioRule,False)
                    if result['state'] == "OK" :
                        # files transfered to nucleus
                        tmpLog.debug('Files for Rucio Rule {0} successfully transferred'.format(rucioRule))
                        self.dbInterface.update_file_group_status(rucioRule, 'transferred')
                        # set the fileSpec status for these files 
                        self.set_FileSpec_status(jobspec,'finished')
                    elif result['state'] == "FAILED" :
                        # failed Rucio Transfer
                        tmpStat = False
                        tmpMsg = 'Failed Rucio Transfer - Rucio Rule - {0}'.format(rucioRule)
                        tmpLog.debug(tmpMsg)
                        self.set_FileSpec_status(jobspec,'failed')
                    elif result['state'] == 'STUCK' :
                        tmpStat = None
                        tmpMsg = 'Rucio Transfer Rule {0} Stuck'.format(rucioRule)
                        tmpLog.debug(tmpMsg)
                except:
                        tmpStat = None<commit_msg>add PathConvention and objstoreID to FileSpec<commit_after>import time
import datetime
import uuid
import os
import sys
import stat
import os.path
import threading
import tarfile
import hashlib
import string
import shutil
import errno

from future.utils import iteritems

from rucio.client import Client as RucioClient
from rucio.common.exception import DataIdentifierNotFound, DuplicateRule, DataIdentifierAlreadyExists, \
    FileAlreadyExists


from pandaharvester.harvestercore import core_utils
from pandaharvester.harvestercore.plugin_base import PluginBase
from pandaharvester.harvesterconfig import harvester_config
from pandaharvester.harvestermover import mover_utils
from pandaharvester.harvestercore.queue_config_mapper import QueueConfigMapper
from pandaharvester.harvesterstager.base_stager import BaseStager

from .base_stager import BaseStager

def dump(tmpLog,obj):
   for attr in dir(obj):
       if hasattr( obj, attr ):
           tmpLog.debug( "obj.%s = %s" % (attr, getattr(obj, attr)))



# logger
baseLogger = core_utils.setup_logger('yoda_rucio_rse_direct_stager')

class Error(EnvironmentError):
    pass

class SpecialFileError(EnvironmentError):
    """Raised when trying to do a kind of operation (e.g. copying) which is
    not supported on a special file (e.g. a named pipe)"""

class ExecError(EnvironmentError):
    """Raised when a command could not be executed"""



# stager plugin with RSE + local site move behaviour for Yoda zip files
class YodaRucioRseDirectStager(BaseStager):
    """In the workflow for RseDirectStager, workers directly upload output files to RSE
    and thus there is no data motion in Harvester."""
    # constructor
    def __init__(self, **kwarg):
        BaseStager.__init__(self, **kwarg)
        tmpLog = self.make_logger(baseLogger, 'ThreadID={0}'.format(threading.current_thread().ident),
                                  method_name='YodaRucioRseDirectStager __init__ ')
        tmpLog.debug('start')
        self.Yodajob = False 
        self.pathConvention = None
        self.objstoreID = None
        tmpLog.debug('stop')

    # set FileSpec.objstoreID 
    def set_FileSpec_objstoreID(self,jobspec, objstoreID, pathConvention):
        # loop over all output files
        for fileSpec in jobspec.outFiles:
            fileSpec.objstoreID = objstoreID
            fileSpec.pathConvention = pathConvention

    # set FileSpec.status 
    def set_FileSpec_status(self,jobspec,status):
        # loop over all output files
        for fileSpec in jobspec.outFiles:
            fileSpec.status = status

    # check status
    def check_status(self, jobspec):
        tmpStat = True
        tmpMsg = ''
        # make logger
        tmpLog = self.make_logger(baseLogger, 'PandaID={0} ThreadID={1}'.format(jobspec.PandaID,threading.current_thread().ident),
                                  method_name='check_status')
        tmpLog.debug('start')
        # Get the files grouped by Rucio Rule ID 
        groups = jobspec.get_groups_of_output_files()
        if len(groups) == 0:
            tmpLog.debug('No Rucio Rules')
            return None,'No Rucio Rules'
        tmpLog.debug('#Rucio Rules - {0} - Rules - {1}'.format(len(groups),groups)) 
        
        try:
            rucioAPI = RucioClient()
        except:
            tmpLog.error('failure to get Rucio Client try again later')
            return None,'failure to get Rucio Client try again later'

        # loop over the Rucio rules 
        for rucioRule in groups:
            if rucioRule is None:
                continue
            # lock
            have_db_lock = self.dbInterface.get_object_lock(rucioRule, lock_interval=120)
            if not have_db_lock:
                msgStr = 'escape since {0} is locked by another thread'.format(rucioRule)
                tmpLog.debug(msgStr)
                return None, msgStr
            # get transfer status
            groupStatus = self.dbInterface.get_file_group_status(rucioRule)
            tmpLog.debug('rucioRule - {0} - groupStatus - {1}'.format(rucioRule,groupStatus))
            if 'transferred' in groupStatus:
                # already succeeded
                pass
            elif 'failed' in groupStatus :
                # transfer failure
                tmpStat = False
                tmpMsg = 'rucio rule for {0}:{1} already failed'.format(datasetScope, datasetName)
            elif 'transferring' in groupStatus or 'pending' in groupStatus:
                # transfer started in Rucio check status
                try:
                    result = rucioAPI.get_replication_rule(rucioRule,False)
                    if result['state'] == "OK" :
                        # files transfered to nucleus
                        tmpLog.debug('Files for Rucio Rule {0} successfully transferred'.format(rucioRule))
                        self.dbInterface.update_file_group_status(rucioRule, 'transferred')
                        # set the fileSpec status for these files 
                        self.set_FileSpec_objstoreID(jobspec, self.objstoreID, self.pathConvention)
                        self.set_FileSpec_status(jobspec,'finished')
                    elif result['state'] == "FAILED" :
                        # failed Rucio Transfer
                        tmpStat = False
                        tmpMsg = 'Failed Rucio Transfer - Rucio Rule - {0}'.format(rucioRule)
                        tmpLog.debug(tmpMsg)
                        self.set_FileSpec_status(jobspec,'failed')
                    elif result['state'] == 'STUCK' :
                        tmpStat = None
                        tmpMsg = 'Rucio Transfer Rule {0} Stuck'.format(rucioRule)
                        tmpLog.debug(tmpMsg)
                except:
                        tmpStat = None
{'fraction_non_alphanumeric': 0.0760964529093133, 'fraction_numerical': 0.002970470033199371, 'mean_word_length': 2.2154494382022474, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6201408', 'n_tokens_mistral': 3200, 'n_tokens_neox': 3072, 'n_words': 841}