""" TransformationAgent processes transformations found in the transformation database.
The following options can be set for the TransformationAgent.
.. literalinclude:: ../ConfigTemplate.cfg
:start-after: ##BEGIN TransformationAgent
:end-before: ##END
:dedent: 2
:caption: TransformationAgent options
"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import time
import os
import datetime
import pickle
import concurrent.futures
from DIRAC import S_OK, S_ERROR
from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
from DIRAC.Core.Base.AgentModule import AgentModule
from DIRAC.Core.Utilities.ThreadSafe import Synchronizer
from DIRAC.Core.Utilities.List import breakListIntoChunks, randomize
from DIRAC.DataManagementSystem.Client.DataManager import DataManager
from DIRAC.TransformationSystem.Client import TransformationFilesStatus
from DIRAC.TransformationSystem.Client.TransformationClient import TransformationClient
from DIRAC.TransformationSystem.Agent.TransformationAgentsUtilities import TransformationAgentsUtilities
__RCSID__ = "64080842f (2021-08-19 10:29:23 +0200) Chris Burr <christopher.burr@cern.ch>"
AGENT_NAME = 'Transformation/TransformationAgent'
gSynchro = Synchronizer()
class TransformationAgent(AgentModule, TransformationAgentsUtilities):
""" Usually subclass of AgentModule
"""
def __init__(self, *args, **kwargs):
""" c'tor
"""
AgentModule.__init__(self, *args, **kwargs)
TransformationAgentsUtilities.__init__(self)
# few parameters
self.pluginLocation = ''
self.transformationStatus = []
self.maxFiles = 0
self.transformationTypes = []
# clients (out of the threads)
self.transfClient = None
# parameters for caching
self.workDirectory = ''
self.cacheFile = ''
self.controlDirectory = ''
self.lastFileOffset = {}
# Validity of the cache
self.replicaCache = None
self.replicaCacheValidity = None
self.writingCache = False
self.removedFromCache = 0
self.noUnusedDelay = 0
self.unusedFiles = {}
self.unusedTimeStamp = {}
self.debug = False
self.pluginTimeout = {}
def initialize(self):
""" standard initialize
"""
# few parameters
self.pluginLocation = self.am_getOption('PluginLocation',
'DIRAC.TransformationSystem.Agent.TransformationPlugin')
self.transformationStatus = self.am_getOption('transformationStatus', ['Active', 'Completing', 'Flush'])
# Prepare to change the name of the CS option as MaxFiles is ambiguous
self.maxFiles = self.am_getOption('MaxFilesToProcess', self.am_getOption('MaxFiles', 5000))
agentTSTypes = self.am_getOption('TransformationTypes', [])
if agentTSTypes:
self.transformationTypes = sorted(agentTSTypes)
else:
dataProc = Operations().getValue('Transformations/DataProcessing', ['MCSimulation', 'Merge'])
dataManip = Operations().getValue('Transformations/DataManipulation', ['Replication', 'Removal'])
self.transformationTypes = sorted(dataProc + dataManip)
# clients
self.transfClient = TransformationClient()
# for caching using a pickle file
self.workDirectory = self.am_getWorkDirectory()
self.cacheFile = os.path.join(self.workDirectory, 'ReplicaCache.pkl')
self.controlDirectory = self.am_getControlDirectory()
# remember the offset if any in TS
self.lastFileOffset = {}
# Validity of the cache
self.replicaCache = {}
self.replicaCacheValidity = self.am_getOption('ReplicaCacheValidity', 2)
self.noUnusedDelay = self.am_getOption('NoUnusedDelay', 6)
# Instantiating the ThreadPoolExecutor
maxNumberOfThreads = self.am_getOption('maxThreadsInPool', 15)
self.log.info("Multithreaded with %d threads" % maxNumberOfThreads)
self.threadPoolExecutor = concurrent.futures.ThreadPoolExecutor(max_workers=maxNumberOfThreads)
self.log.info("Will treat the following transformation types: %s" % str(self.transformationTypes))
return S_OK()
def finalize(self):
""" graceful finalization
"""
method = 'finalize'
self._logInfo("Wait for threads to get empty before terminating the agent", method=method)
self.threadPoolExecutor.shutdown()
self._logInfo("Threads are empty, terminating the agent...", method=method)
self.__writeCache()
return S_OK()
def execute(self):
""" Just puts transformations in the queue, and spawns threads if there's work to do.
"""
# Get the transformations to process
res = self.getTransformations()
if not res['OK']:
self._logError("Failed to obtain transformations:", res['Message'])
return S_OK()
# Process the transformations
count = 0
future_to_transID = {}
for transDict in res['Value']:
transID = int(transDict['TransformationID'])
if transDict.get('InheritedFrom'):
# Try and move datasets from the ancestor production
res = self.transfClient.moveFilesToDerivedTransformation(transDict)
if not res['OK']:
self._logError("Error moving files from an inherited transformation", res['Message'], transID=transID)
else:
parentProd, movedFiles = res['Value']
if movedFiles:
self._logInfo("Successfully moved files from %d to %d:" % (parentProd, transID), transID=transID)
for status, val in movedFiles.items():
self._logInfo("\t%d files to status %s" % (val, status), transID=transID)
count += 1
future = self.threadPoolExecutor.submit(self._execute, transDict)
future_to_transID[future] = transID
self._logInfo("Out of %d transformations, %d put in thread queue" % (len(res['Value']), count))
for future in concurrent.futures.as_completed(future_to_transID):
transID = future_to_transID[future]
try:
future.result()
except Exception as exc:
self._logError('%d generated an exception: %s' % (transID, exc))
else:
self._logInfo('Processed %d' % transID)
return S_OK()
def getTransformations(self):
""" Obtain the transformations to be executed - this is executed at the start of every loop (it's really the
only real thing in the execute()
"""
transName = self.am_getOption('Transformation', 'All')
method = 'getTransformations'
if transName == 'All':
self._logInfo("Getting all transformations%s, status %s." %
(' of type %s' % str(self.transformationTypes) if self.transformationTypes else '',
str(self.transformationStatus)),
method=method)
transfDict = {'Status': self.transformationStatus}
if self.transformationTypes:
transfDict['Type'] = self.transformationTypes
res = self.transfClient.getTransformations(transfDict, extraParams=True)
if not res['OK']:
return res
transformations = res['Value']
self._logInfo("Obtained %d transformations to process" % len(transformations), method=method)
else:
self._logInfo("Getting transformation %s." % transName, method=method)
res = self.transfClient.getTransformation(transName, extraParams=True)
if not res['OK']:
self._logError("Failed to get transformation:", res['Message'], method=method)
return res
transformations = [res['Value']]
return S_OK(transformations)
def _getClients(self):
""" returns the clients used in the threads
"""
threadTransformationClient = TransformationClient()
threadDataManager = DataManager()
return {'TransformationClient': threadTransformationClient,
'DataManager': threadDataManager}
def _execute(self, transDict):
""" thread - does the real job: processing the transformation to be processed
"""
self._logDebug('Starting _execute')
# Each thread will have its own clients
clients = self._getClients()
try:
transID = int(transDict['TransformationID'])
self._logInfo("Processing transformation %s." % transID, transID=transID)
startTime = time.time()
res = self.processTransformation(transDict, clients)
if not res['OK']:
self._logInfo("Failed to process transformation:", res['Message'], transID=transID)
except Exception as x: # pylint: disable=broad-except
self._logException('Exception in plugin', lException=x, transID=transID)
finally:
if not transID:
transID = 'None'
self._logInfo("Processed transformation in %.1f seconds" % (time.time() - startTime), transID=transID)
self._logDebug('Exiting _execute')
def processTransformation(self, transDict, clients):
""" process a single transformation (in transDict)
"""
method = 'processTransformation'
transID = transDict['TransformationID']
forJobs = transDict['Type'].lower() not in ('replication', 'removal')
# First get the LFNs associated to the transformation
transFiles = self._getTransformationFiles(transDict, clients, replicateOrRemove=not forJobs)
if not transFiles['OK']:
return transFiles
if not transFiles['Value']:
return S_OK()
if transID not in self.replicaCache:
self.__readCache(transID)
transFiles = transFiles['Value']
unusedLfns = [f['LFN'] for f in transFiles]
unusedFiles = len(unusedLfns)
plugin = transDict.get('Plugin', 'Standard')
# Limit the number of LFNs to be considered for replication or removal as they are treated individually
if not forJobs:
maxFiles = Operations().getValue('TransformationPlugins/%s/MaxFilesToProcess' % plugin, 0)
# Get plugin-specific limit in number of files (0 means no limit)
totLfns = len(unusedLfns)
lfnsToProcess = self.__applyReduction(unusedLfns, maxFiles=maxFiles)
if len(lfnsToProcess) != totLfns:
self._logInfo("Reduced number of files from %d to %d" % (totLfns, len(lfnsToProcess)),
method=method, transID=transID)
transFiles = [f for f in transFiles if f['LFN'] in lfnsToProcess]
else:
lfnsToProcess = unusedLfns
# Check the data is available with replicas
res = self.__getDataReplicas(transDict, lfnsToProcess, clients, forJobs=forJobs)
if not res['OK']:
self._logError("Failed to get data replicas:", res['Message'],
method=method, transID=transID)
return res
dataReplicas = res['Value']
# Get the plug-in type and create the plug-in object
self._logInfo("Processing transformation with '%s' plug-in." % plugin,
method=method, transID=transID)
res = self.__generatePluginObject(plugin, clients)
if not res['OK']:
return res
oPlugin = res['Value']
# Get the plug-in and set the required params
oPlugin.setParameters(transDict)
oPlugin.setInputData(dataReplicas)
oPlugin.setTransformationFiles(transFiles)
res = oPlugin.run()
if not res['OK']:
self._logError("Failed to generate tasks for transformation:", res['Message'],
method=method, transID=transID)
return res
tasks = res['Value']
self.pluginTimeout[transID] = res.get('Timeout', False)
# Create the tasks
allCreated = True
created = 0
lfnsInTasks = []
for se, lfns in tasks:
res = clients['TransformationClient'].addTaskForTransformation(transID, lfns, se)
if not res['OK']:
self._logError("Failed to add task generated by plug-in:", res['Message'],
method=method, transID=transID)
allCreated = False
else:
created += 1
lfnsInTasks += [lfn for lfn in lfns if lfn in lfnsToProcess]
if created:
self._logInfo("Successfully created %d tasks for transformation." % created,
method=method, transID=transID)
else:
self._logInfo("No new tasks created for transformation.",
method=method, transID=transID)
self.unusedFiles[transID] = unusedFiles - len(lfnsInTasks)
# If not all files were obtained, move the offset
lastOffset = self.lastFileOffset.get(transID)
if lastOffset:
self.lastFileOffset[transID] = max(0, lastOffset - len(lfnsInTasks))
self.__removeFilesFromCache(transID, lfnsInTasks)
# If this production is to Flush
if transDict['Status'] == 'Flush' and allCreated:
res = clients['TransformationClient'].setTransformationParameter(transID, 'Status', 'Active')
if not res['OK']:
self._logError("Failed to update transformation status to 'Active':", res['Message'],
method=method, transID=transID)
else:
self._logInfo("Updated transformation status to 'Active'.",
method=method, transID=transID)
return S_OK()
######################################################################
#
# Internal methods used by the agent
#
def _getTransformationFiles(self, transDict, clients, statusList=None, replicateOrRemove=False):
""" get the data replicas for a certain transID
"""
# By default, don't skip if no new Unused for DM transformations
skipIfNoNewUnused = not replicateOrRemove
transID = transDict['TransformationID']
plugin = transDict.get('Plugin', 'Standard')
# Check if files should be sorted and limited in number
operations = Operations()
sortedBy = operations.getValue('TransformationPlugins/%s/SortedBy' % plugin, None)
maxFiles = operations.getValue('TransformationPlugins/%s/MaxFilesToProcess' % plugin, 0)
# If the NoUnuse delay is explicitly set, we want to take it into account, and skip if no new Unused
if operations.getValue('TransformationPlugins/%s/NoUnusedDelay' % plugin, 0):
skipIfNoNewUnused = True
noUnusedDelay = 0 if self.pluginTimeout.get(transID, False) else \
operations.getValue('TransformationPlugins/%s/NoUnusedDelay' % plugin, self.noUnusedDelay)
method = '_getTransformationFiles'
lastOffset = self.lastFileOffset.setdefault(transID, 0)
# Files that were problematic (either explicit or because SE was banned) may be recovered,
# and always removing the missing ones
if not statusList:
statusList = [TransformationFilesStatus.UNUSED, TransformationFilesStatus.PROB_IN_FC]
statusList += [TransformationFilesStatus.MISSING_IN_FC] if transDict['Type'] == 'Removal' else []
transClient = clients['TransformationClient']
res = transClient.getTransformationFiles(condDict={'TransformationID': transID,
'Status': statusList},
orderAttribute=sortedBy,
offset=lastOffset, maxfiles=maxFiles)
if not res['OK']:
self._logError("Failed to obtain input data:", res['Message'],
method=method, transID=transID)
return res
transFiles = res['Value']
if maxFiles and len(transFiles) == maxFiles:
self.lastFileOffset[transID] += maxFiles
else:
del self.lastFileOffset[transID]
if not transFiles:
self._logInfo("No '%s' files found for transformation." % ','.join(statusList),
method=method, transID=transID)
if transDict['Status'] == 'Flush':
res = transClient.setTransformationParameter(transID, 'Status', 'Active')
if not res['OK']:
self._logError("Failed to update transformation status to 'Active':", res['Message'],
method=method, transID=transID)
else:
self._logInfo("Updated transformation status to 'Active'.",
method=method, transID=transID)
return S_OK()
# Check if transformation is kicked
kickFile = os.path.join(self.controlDirectory, 'KickTransformation_%s' % str(transID))
try:
kickTrans = os.path.exists(kickFile)
if kickTrans:
os.remove(kickFile)
except OSError:
pass
# Check if something new happened
now = datetime.datetime.utcnow()
if not kickTrans and skipIfNoNewUnused and noUnusedDelay:
nextStamp = self.unusedTimeStamp.setdefault(transID, now) + datetime.timedelta(hours=noUnusedDelay)
skip = now < nextStamp
if len(transFiles) == self.unusedFiles.get(transID, 0) and transDict['Status'] != 'Flush' and skip:
self._logInfo("No new '%s' files found for transformation." % ','.join(statusList),
method=method, transID=transID)
return S_OK()
self.unusedTimeStamp[transID] = now
# If files are not Unused, set them Unused
notUnused = [trFile['LFN'] for trFile in transFiles if trFile['Status'] != TransformationFilesStatus.UNUSED]
otherStatuses = sorted(set([trFile['Status'] for trFile in transFiles]) - set([TransformationFilesStatus.UNUSED]))
if notUnused:
res = transClient.setFileStatusForTransformation(
transID, TransformationFilesStatus.UNUSED, notUnused, force=True
)
if not res['OK']:
self._logError("Error setting %d files Unused:" % len(notUnused), res['Message'],
method=method, transID=transID)
else:
self._logInfo("Set %d files from %s to Unused" % (len(notUnused), ','.join(otherStatuses)))
self.__removeFilesFromCache(transID, notUnused)
return S_OK(transFiles)
def __applyReduction(self, lfns, maxFiles=None):
""" eventually remove the number of files to be considered
"""
if maxFiles is None:
maxFiles = self.maxFiles
if not maxFiles or len(lfns) <= maxFiles:
return lfns
return randomize(lfns)[:maxFiles]
def __getDataReplicas(self, transDict, lfns, clients, forJobs=True):
""" Get the replicas for the LFNs and check their statuses. It first looks within the cache.
"""
method = '__getDataReplicas'
transID = transDict['TransformationID']
if 'RemoveFile' in transDict['Body']:
# When removing files, we don't care about their replicas
return S_OK(dict.fromkeys(lfns, ['None']))
clearCacheFile = os.path.join(self.controlDirectory, 'ClearCache_%s' % str(transID))
try:
clearCache = os.path.exists(clearCacheFile)
if clearCache:
os.remove(clearCacheFile)
except Exception:
pass
if clearCache or transDict['Status'] == 'Flush':
self._logInfo("Replica cache cleared", method=method, transID=transID)
# We may need to get new replicas
self.__clearCacheForTrans(transID)
else:
# If the cache needs to be cleaned
self.__cleanCache(transID)
startTime = time.time()
dataReplicas = {}
nLfns = len(lfns)
self._logVerbose("Getting replicas for %d files" % nLfns, method=method, transID=transID)
cachedReplicaSets = self.replicaCache.get(transID, {})
cachedReplicas = {}
# Merge all sets of replicas
for replicas in cachedReplicaSets.values():
cachedReplicas.update(replicas)
self._logInfo("Number of cached replicas: %d" % len(cachedReplicas), method=method, transID=transID)
setCached = set(cachedReplicas)
setLfns = set(lfns)
for lfn in setLfns & setCached:
dataReplicas[lfn] = cachedReplicas[lfn]
newLFNs = setLfns - setCached
self._logInfo("ReplicaCache hit for %d out of %d LFNs" % (len(dataReplicas), nLfns),
method=method, transID=transID)
if newLFNs:
startTime = time.time()
self._logInfo("Getting replicas for %d files from catalog" % len(newLFNs),
method=method, transID=transID)
newReplicas = {}
for chunk in breakListIntoChunks(newLFNs, 10000):
res = self._getDataReplicasDM(transID, chunk, clients, forJobs=forJobs)
if res['OK']:
reps = dict((lfn, ses) for lfn, ses in res['Value'].items() if ses)
newReplicas.update(reps)
self.__updateCache(transID, reps)
else:
self._logWarn("Failed to get replicas for %d files" % len(chunk), res['Message'],
method=method, transID=transID)
self._logInfo("Obtained %d replicas from catalog in %.1f seconds"
% (len(newReplicas), time.time() - startTime),
method=method, transID=transID)
dataReplicas.update(newReplicas)
noReplicas = newLFNs - set(dataReplicas)
self.__writeCache(transID)
if noReplicas:
self._logWarn("Found %d files without replicas (or only in Failover)" % len(noReplicas),
method=method, transID=transID)
return S_OK(dataReplicas)
def _getDataReplicasDM(self, transID, lfns, clients, forJobs=True, ignoreMissing=False):
""" Get the replicas for the LFNs and check their statuses, using the replica manager
"""
method = '_getDataReplicasDM'
startTime = time.time()
self._logVerbose("Getting replicas%s from catalog for %d files" % (' for jobs' if forJobs else '', len(lfns)),
method=method, transID=transID)
if forJobs:
# Get only replicas eligible for jobs
res = clients['DataManager'].getReplicasForJobs(lfns, getUrl=False)
else:
# Get all replicas
res = clients['DataManager'].getReplicas(lfns, getUrl=False)
if not res['OK']:
return res
replicas = res['Value']
# Prepare a dictionary for all LFNs
dataReplicas = {}
self._logVerbose("Replica results for %d files obtained in %.2f seconds" %
(len(lfns), time.time() - startTime),
method=method, transID=transID)
# If files are neither Successful nor Failed, they are set problematic in the FC
problematicLfns = [lfn for lfn in lfns if lfn not in replicas['Successful'] and lfn not in replicas['Failed']]
if problematicLfns:
self._logInfo("%d files found problematic in the catalog, set ProbInFC" % len(problematicLfns))
res = clients['TransformationClient'].setFileStatusForTransformation(
transID, TransformationFilesStatus.PROB_IN_FC, problematicLfns
)
if not res['OK']:
self._logError("Failed to update status of problematic files:", res['Message'],
method=method, transID=transID)
# Create a dictionary containing all the file replicas
failoverLfns = []
for lfn, replicaDict in replicas['Successful'].items():
for se in replicaDict:
# This remains here for backward compatibility in case VOs have not defined SEs not to be used for jobs
if forJobs and 'failover' in se.lower():
self._logVerbose("Ignoring failover replica for %s." % lfn, method=method, transID=transID)
else:
dataReplicas.setdefault(lfn, []).append(se)
if not dataReplicas.get(lfn):
failoverLfns.append(lfn)
if failoverLfns:
self._logVerbose("%d files have no replica but possibly in Failover SE" % len(failoverLfns))
# Make sure that file missing from the catalog are marked in the transformation DB.
missingLfns = []
for lfn, reason in replicas['Failed'].items():
if "No such file or directory" in reason:
self._logVerbose("%s not found in the catalog." % lfn, method=method, transID=transID)
missingLfns.append(lfn)
if missingLfns:
self._logInfo("%d files not found in the catalog" % len(missingLfns))
if ignoreMissing:
dataReplicas.update(dict.fromkeys(missingLfns, []))
else:
res = clients['TransformationClient'].setFileStatusForTransformation(
transID,
TransformationFilesStatus.MISSING_IN_FC, missingLfns
)
if not res['OK']:
self._logError("Failed to update status of missing files:", res['Message'],
method=method, transID=transID)
return S_OK(dataReplicas)
def __updateCache(self, transID, newReplicas):
""" Add replicas to the cache
"""
self.replicaCache.setdefault(transID, {})[datetime.datetime.utcnow()] = newReplicas
def __clearCacheForTrans(self, transID):
""" Remove all replicas for a transformation
"""
self.replicaCache.pop(transID, None)
def __cleanReplicas(self, transID, lfns):
""" Remove cached replicas that are not in a list
"""
cachedReplicas = set()
for replicas in self.replicaCache.get(transID, {}).values():
cachedReplicas.update(replicas)
toRemove = cachedReplicas - set(lfns)
if toRemove:
self._logInfo("Remove %d files from cache" % len(toRemove), method='__cleanReplicas', transID=transID)
self.__removeFromCache(transID, toRemove)
def __cleanCache(self, transID):
""" Cleans the cache
"""
try:
if transID in self.replicaCache:
timeLimit = datetime.datetime.utcnow() - datetime.timedelta(days=self.replicaCacheValidity)
for updateTime in set(self.replicaCache[transID]):
nCache = len(self.replicaCache[transID][updateTime])
if updateTime < timeLimit or not nCache:
self._logInfo("Clear %s replicas for transformation %s, time %s" %
('%d cached' % nCache if nCache else 'empty cache', str(transID), str(updateTime)),
transID=transID, method='__cleanCache')
del self.replicaCache[transID][updateTime]
# Remove empty transformations
if not self.replicaCache[transID]:
del self.replicaCache[transID]
except Exception as x:
self._logException("Exception when cleaning replica cache:", lException=x)
def __removeFilesFromCache(self, transID, lfns):
removed = self.__removeFromCache(transID, lfns)
if removed:
self._logInfo("Removed %d replicas from cache" % removed, method='__removeFilesFromCache', transID=transID)
self.__writeCache(transID)
def __removeFromCache(self, transID, lfns):
if transID not in self.replicaCache:
return
removed = 0
if self.replicaCache[transID] and lfns:
for lfn in lfns:
for timeKey in self.replicaCache[transID]:
if self.replicaCache[transID][timeKey].pop(lfn, None):
removed += 1
return removed
def __cacheFile(self, transID):
return self.cacheFile.replace('.pkl', '_%s.pkl' % str(transID))
@gSynchro
def __readCache(self, transID):
""" Reads from the cache
"""
if transID in self.replicaCache:
return
try:
method = '__readCache'
fileName = self.__cacheFile(transID)
if not os.path.exists(fileName):
self.replicaCache[transID] = {}
else:
with open(fileName, 'rb') as cacheFile:
self.replicaCache[transID] = pickle.load(cacheFile)
self._logInfo("Successfully loaded replica cache from file %s (%d files)" %
(fileName, self.__filesInCache(transID)),
method=method, transID=transID)
except Exception as x:
self._logException("Failed to load replica cache from file %s" % fileName, lException=x,
method=method, transID=transID)
self.replicaCache[transID] = {}
def __filesInCache(self, transID):
cache = self.replicaCache.get(transID, {})
return sum(len(lfns) for lfns in cache.values())
@gSynchro
def __writeCache(self, transID=None):
""" Writes the cache
"""
method = '__writeCache'
try:
startTime = time.time()
transList = [transID] if transID else set(self.replicaCache)
filesInCache = 0
nCache = 0
for t_id in transList:
# Protect the copy of the cache
filesInCache += self.__filesInCache(t_id)
# write to a temporary file in order to avoid corrupted files
cacheFile = self.__cacheFile(t_id)
tmpFile = cacheFile + '.tmp'
with open(tmpFile, 'wb') as fd:
pickle.dump(self.replicaCache.get(t_id, {}), fd)
# Now rename the file as it shold
os.rename(tmpFile, cacheFile)
nCache += 1
self._logInfo("Successfully wrote %d replica cache file(s) (%d files) in %.1f seconds"
% (nCache, filesInCache, time.time() - startTime),
method=method, transID=transID if transID else None)
except Exception as x:
self._logException("Could not write replica cache file %s" % cacheFile, lException=x,
method=method, transID=t_id)
def __generatePluginObject(self, plugin, clients):
""" This simply instantiates the TransformationPlugin class with the relevant plugin name
"""
try:
plugModule = __import__(self.pluginLocation, globals(), locals(), ['TransformationPlugin'])
except ImportError as e:
self._logException("Failed to import 'TransformationPlugin' %s" % plugin, lException=e,
method="__generatePluginObject")
return S_ERROR()
try:
plugin_o = getattr(plugModule, 'TransformationPlugin')('%s' % plugin,
transClient=clients['TransformationClient'],
dataManager=clients['DataManager'])
return S_OK(plugin_o)
except AttributeError as e:
self._logException("Failed to create %s()" % plugin, lException=e, method="__generatePluginObject")
return S_ERROR()
plugin_o.setDirectory(self.workDirectory)
plugin_o.setCallback(self.pluginCallback)
def pluginCallback(self, transID, invalidateCache=False):
""" Standard plugin callback
"""
if invalidateCache:
try:
if transID in self.replicaCache:
self._logInfo("Removed cached replicas for transformation", method='pluginCallBack', transID=transID)
self.replicaCache.pop(transID)
self.__writeCache(transID)
except Exception:
pass