Added skeleton for 'RetrieveExecutor' class. Not yet working. (032650eb) · Commits · VOSpace INAF / vospace-transfer-service

transfer_service/job_scheduler.py

+4 −0

Original line number	Diff line number	Diff line
		from retrieve_preprocessor import RetrievePreprocessor
		from store_preprocessor import StorePreprocessor
		from retrieve_executor import RetrieveExecutor
		from store_executor import StoreExecutor



		class JobScheduler(object):

		def __init__(self):
		self.retrievePreprocessor = RetrievePreprocessor()
		self.storePreprocessor = StorePreprocessor()
		self.retrieveExecutor = RetrieveExecutor()
		self.storeExecutor = StoreExecutor()

		def start(self):
		self.retrievePreprocessor.start()
		self.storePreprocessor.start()
		self.retrieveExecutor.start()
		self.storeExecutor.start()

transfer_service/retrieve_executor.py

0 → 100644

+81 −0

Original line number	Diff line number	Diff line
		import subprocess
		import sys

		from config import Config
		from db_connector import DbConnector
		from tape_client import TapeClient
		from task_executor import TaskExecutor

		class RetrieveExecutor(TaskExecutor):

		def __init__(self):
		config = Config("vos_ts.conf")
		self.params = config.loadSection("spectrum_archive")
		self.tapeClient = TapeClient(self.params["host"],
		self.params.getint("port"),
		self.params["user"],
		self.params["password"])
		self.params = config.loadSection("transfer_node")
		self.storageRetrievePath = self.params["retrieve_path"]
		self.params = config.loadSection("tape_library")
		self.tapeStorageBasePath = self.params["base_path"]
		self.params = config.loadSection("ia2_server")
		self.serverStorageBasePath = self.params["base_path"]
		self.params = config.loadSection("file_catalog")
		self.dbConn = DbConnector(self.params["user"],
		self.params["password"],
		self.params["host"],
		self.params.getint("port"),
		self.params["db"])
		self.jobObj = None
		super(RetrieveExecutor, self).__init__()

		def retrieveData(self):
		self.dbConn.connect()
		self.username = self.dbConn.getUserName()
		destPathPrefix = self.storageRetrievePath.replace("{username}", self.username)
		srcData = os.listdir(srcPathPrefix)
		if self.requestType == "CSTORE": # TO BE CHANGED!!!
		srcPathPrefix = self.tapeStorageBasePath.replace("{username}", self.username)
		# TO BE DONE
		else:
		srcPathPrefix = self.serverStorageBasePath.replace("{username}", self.username)
		for vospacePath in self.nodeList:
		osRelPath = self.dbConn.getOSPath(vospacePath)
		sp = subprocess.run(["rsync", "-av", srcPathPrefix + osRelPath, destPathPrefix + '/'], capture_output = True)
		if(sp.returncode or sp.stderr):
		self.dbConn.disconnect()
		return False
		self.dbConn.disconnect()
		return True

		def updateDb(self):
		self.dbConn.connect()
		self.dbConn.setPhase(self.jobId, "COMPLETED")
		for vospacePath in self.nodeList:
		osRelPath = self.dbConn.getOSPath(vospacePath)
		self.dbConn.setAsyncTrans(osRelPath, False);
		self.dbConn.disconnect()

		def cleanup(self):
		pass

		def run(self):
		print("Starting retrieve executor...")
		self.setSourceQueueName("read_ready")
		self.setDestinationQueueName("read_terminated")
		while True:
		self.wait()
		if self.srcQueue.len() > 0:
		self.jobObj = self.srcQueue.getJob()
		self.jobId = self.jobObj.jobId
		self.requestType = self.jobObj.jobInfo["requestType"]
		self.nodeList = self.jobObj.jobInfo["nodeList"]
		result = self.retrieveData()
		if result:
		self.updateDb()
		else:
		sys.exit("Failed to retrieve data!")
		self.srcQueue.moveJobTo(self.destQueue.name())
		print(f"Job {self.jobObj.jobId} MOVED from {self.srcQueue.name()} to {self.destQueue.name()}")

		No newline at end of file