Commit f348abca authored by Cristiano Urban's avatar Cristiano Urban
Browse files

Added 'StoreExecutor' class. 



Signed-off-by: default avatarCristiano Urban <cristiano.urban@inaf.it>
parent 8aa9531a

transfer_service/store_executor.py

0 → 100644
+57 −0
Original line number Diff line number Diff line 
import os



from config import Config

from tape_client import TapeClient

from task_executor import TaskExecutor



class StoreExecutor(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.storageStorePath = self.params["store_path"]

        self.params = config.loadSection("tape_library")

        self.storageBasePath = self.params["base_path"]

        self.username = None

        super(StoreExecutor, self).__init__()

        

    def copyData(self, username):

        self.username = username

        srcPathPrefix = self.storageStorePath.replace("{username}", self.username)

        destPathPrefix = self.storageBasePath.replace("{username}", self.username)

        srcData = os.listdir(srcPathPrefix)

        self.tapeClient.connect()

        for el in srcData:

            self.tapeClient.copy(srcPathPrefix + '/' + el, destPathPrefix + '/' + el)

        self.tapeClient.disconnect()



    def cleanup(self):

        pass



    def run(self):

        print("Starting storage executor...")

        self.setSourceQueueName("write_ready")

        self.setDestinationQueueName("write_terminated")

        while True:

            self.wait()

            if self.srcQueue.len() > 0:

                self.jobObj = self.srcQueue.getJob()

                self.username = self.jobObj["jobInfo"]["userName"]

                # 1) Controlla il tipo di destinazione: hot (server) o cold (tape)

                # *) HOT

                # 1) Controlla di avere sufficiente spazio su disco sul server

                # 2) Ottieni la cartella o la lista delle cartelle sulla 'store' sul transf. node

                # 3) Avvia la copia delle cartelle con rsync da transf. node a frontend tape 

                # *) COLD

                # 1) Controlla che altri non stiano migrando in questo preciso momento (con tape client)

                # 2) Ottieni la cartella o la lista delle cartelle sulla 'store' sul transf. node

                # 3) Avvia la copia delle cartelle con rsync da transf. node a frontend tape 

                # 4) A copia finita, se tutto ok, rimuovi i dati sul tn e setta async_trans a true sul db

                self.copyData(self.username)

                self.srcQueue.moveJobTo(self.destQueue.name())

                print(f"Job {self.jobObj['jobId']} MOVED from {self.srcQueue.name()} to {self.destQueue.name()}")

transfer_service/transfer_service.py

+3 −0
Original line number Diff line number Diff line
@@ -3,6 +3,7 @@ import os 


from job_handler import JobHandler

from store_preprocessor import StorePreprocessor

from store_executor import StoreExecutor

from config import Config
@@ -13,6 +14,7 @@ class TransferService(object): 
        self.params = config.loadSection("amqp")

        self.jobHandler = JobHandler(self.params["host"], self.params.getint("port"))

        self.storePreprocessor = StorePreprocessor()

        self.storeExecutor = StoreExecutor()



        # PullFromVOSpace (via REST API)

        self.jobHandler.addAMQPServer('start', 'start_job_queue')
@@ -24,6 +26,7 @@ class TransferService(object): 


    def start(self):

        self.storePreprocessor.start()

        self.storeExecutor.start()

        self.jobHandler.start()



ts = TransferService()