Minor changes. (bc75a1e1) · Commits · VOSpace INAF / vospace-transfer-service

transfer_service/data_amqp_server.py

+1 −1

Original line number	Original line	Diff line number	Diff line
	@@ -93,7 +93,7 @@ class DataAMQPServer(RedisRpcServer):
	username = requestBody["userName"]		username = requestBody["userName"]
	self.prepare(username)		self.prepare(username)
	job = Job()		job = Job()
	job.setType("other")		job.setType("vos_data")
	job.setInfo(requestBody)		job.setInfo(requestBody)
	job.setPhase("PENDING")		job.setPhase("PENDING")
	job.setOwnerId(self.dbConn.getUserId(username))		job.setOwnerId(self.dbConn.getUserId(username))

transfer_service/import_amqp_server.py

+23 −9

Original line number	Original line	Diff line number	Diff line
	@@ -4,7 +4,8 @@ import os

	from config import Config		from config import Config
	from db_connector import DbConnector		from db_connector import DbConnector
	from import_job_queue import ImportJobQueue		from job_queue import JobQueue
			from job import Job
	from redis_rpc_server import RedisRpcServer		from redis_rpc_server import RedisRpcServer
	from system_utils import SystemUtils		from system_utils import SystemUtils

	@@ -24,7 +25,7 @@ class ImportAMQPServer(RedisRpcServer):
	1)		1)
	self.params = config.loadSection("scheduling")		self.params = config.loadSection("scheduling")
	self.maxReadyJobs = self.params.getint("max_ready_jobs")		self.maxReadyJobs = self.params.getint("max_ready_jobs")
	self.importReadyQueue = ImportJobQueue("import_ready")		self.importReadyQueue = JobQueue("import_ready")
	self.systemUtils = SystemUtils()		self.systemUtils = SystemUtils()
	super(ImportAMQPServer, self).__init__(host, port, db, rpcQueue)		super(ImportAMQPServer, self).__init__(host, port, db, rpcQueue)

	@@ -80,13 +81,26 @@ class ImportAMQPServer(RedisRpcServer):
	"errorCode": 8,		"errorCode": 8,
	"errorMsg": "Import queue is full, please, retry later." }		"errorMsg": "Import queue is full, please, retry later." }
	else:		else:
	job = dict()		jobObj = Job()
	job["userId"] = userId		jobObj.setType("vos_import")
	job["path"] = path		jobInfo = requestBody.copy()
	job["pathPrefix"] = pathPrefix		#jobInfo["userId"] = userId
	job["storageId"] = storageId		jobInfo["pathPrefix"] = pathPrefix
	job["storageType"] = storageType		jobInfo["storageId"] = storageId
	self.importReadyQueue.insertJob(job)		jobInfo["storageType"] = storageType
			jobObj.setInfo(jobInfo)
			jobObj.setPhase("QUEUED")
			jobObj.setOwnerId(userId)
			self.dbConn.insertJob(jobObj)


			#job = dict()
			#job["userId"] = userId
			#job["path"] = path
			#job["pathPrefix"] = pathPrefix
			#job["storageId"] = storageId
			#job["storageType"] = storageType
			self.importReadyQueue.insertJob(jobObj)

	#p = Process(target = self.load,		#p = Process(target = self.load,
	# args = (self.tapeClient,		# args = (self.tapeClient,

transfer_service/import_executor.py

+22 −9

Original line number	Original line	Diff line number	Diff line
	@@ -7,15 +7,15 @@ from config import Config
	from checksum import Checksum		from checksum import Checksum
	from datetime import datetime as dt		from datetime import datetime as dt
	from db_connector import DbConnector		from db_connector import DbConnector
	from import_task_executor import ImportTaskExecutor
	from mailer import Mailer		from mailer import Mailer
	from node import Node		from node import Node
	from system_utils import SystemUtils		from system_utils import SystemUtils
	from tabulate import tabulate		from tabulate import tabulate
	from tape_client import TapeClient		from tape_client import TapeClient
			from task_executor import TaskExecutor


	class ImportExecutor(ImportTaskExecutor):		class ImportExecutor(TaskExecutor):

	def __init__(self):		def __init__(self):
	self.md5calc = Checksum()		self.md5calc = Checksum()
	@@ -34,7 +34,8 @@ class ImportExecutor(ImportTaskExecutor):
	self.params["user"],		self.params["user"],
	self.params["pkey_file_path"])		self.params["pkey_file_path"])
	self.systemUtils = SystemUtils()		self.systemUtils = SystemUtils()
	self.job = None		self.jobObj = None
			self.jobId = None
	self.userId = None		self.userId = None
	self.path = None		self.path = None
	self.pathPrefix = None		self.pathPrefix = None
	@@ -45,6 +46,9 @@ class ImportExecutor(ImportTaskExecutor):
	def importVOSpaceNodes(self):		def importVOSpaceNodes(self):
	"""This method performs the VOSpace import operation."""		"""This method performs the VOSpace import operation."""

			self.dbConn.setPhase(self.jobId, "EXECUTING")
			self.dbConn.setStartTime(self.jobId)

	start = dt.now()		start = dt.now()
	nodeList = []		nodeList = []
	timestamp = dt.now().strftime("%Y_%m_%d-%H_%M_%S")		timestamp = dt.now().strftime("%Y_%m_%d-%H_%M_%S")
	@@ -140,6 +144,13 @@ class ImportExecutor(ImportTaskExecutor):
	nlfp.close()		nlfp.close()
	end = dt.now()		end = dt.now()

			# Update job status (to be moved)
			results = [{"target": ""}]
			self.dbConn.setResults(self.jobId, results)
			self.jobObj.setPhase("COMPLETED")
			self.dbConn.setPhase(self.jobId, "COMPLETED")
			self.dbConn.setEndTime(self.jobId)

	m = Mailer()		m = Mailer()
	m.addRecipient("cristiano.urban@inaf.it")		m.addRecipient("cristiano.urban@inaf.it")
	msg = f"""		msg = f"""
	@@ -160,6 +171,7 @@ class ImportExecutor(ImportTaskExecutor):

	os.remove(nodeListFile)		os.remove(nodeListFile)


	def run(self):		def run(self):
	print("Starting import executor...")		print("Starting import executor...")
	self.setSourceQueueName("import_ready")		self.setSourceQueueName("import_ready")
	@@ -167,12 +179,13 @@ class ImportExecutor(ImportTaskExecutor):
	while True:		while True:
	self.wait()		self.wait()
	if self.srcQueue.len() > 0:		if self.srcQueue.len() > 0:
	self.job = self.srcQueue.getJob()		self.jobObj = self.srcQueue.getJob()
	self.userId = self.job["userId"]		self.jobId = self.jobObj.jobId
	self.path = self.job["path"]		self.userId = self.jobObj.ownerId
	self.pathPrefix = self.job["pathPrefix"]		self.path = self.jobObj.jobInfo["path"]
	self.storageId = self.job["storageId"]		self.pathPrefix = self.jobObj.jobInfo["pathPrefix"]
	self.storageType = self.job["storageType"]		self.storageId = self.jobObj.jobInfo["storageId"]
			self.storageType = self.jobObj.jobInfo["storageType"]
	self.importVOSpaceNodes()		self.importVOSpaceNodes()
	if self.destQueue.len() == self.maxTerminatedJobs:		if self.destQueue.len() == self.maxTerminatedJobs:
	self.destQueue.extractJob()		self.destQueue.extractJob()