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

Added 'cleanup()' and 'update()' methods + minor changes. 



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

transfer_service/store_executor.py

+35 −6
Original line number Diff line number Diff line 
import os

import shutil

import sys



from config import Config

from db_connector import DbConnector

from tape_client import TapeClient

from task_executor import TaskExecutor
@@ -17,11 +20,18 @@ class StoreExecutor(TaskExecutor): 
        self.storageStorePath = self.params["store_path"]

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

        self.storageBasePath = 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

        self.username = None

        self.nodeList = []

        super(StoreExecutor, self).__init__()

        

    def copyData(self, username):

        self.username = username

    def copyData(self):

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

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

        srcData = os.listdir(srcPathPrefix)
@@ -31,7 +41,23 @@ class StoreExecutor(TaskExecutor): 
        self.tapeClient.disconnect()



    def cleanup(self):

        pass

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

        srcData = os.listdir(srcPathPrefix)

        for el in srcData:

            nodeOSPath = srcPathPrefix + '/' + el

            if os.path.isdir(nodeOSPath):

                shutil.rmtree(nodeOSPath)

            elif os.path.isfile(nodeOSPath):

                os.remove(nodeOSPath)

            else:

                sys.exit("Unable to remove file/dir on the transfer node!!!")

                

    def update(self):

        self.dbConn.connect()

        for node in self.nodeList:

            self.dbConn.setAsyncTrans(node, True);

        self.dbConn.disconnect()

        



    def run(self):

        print("Starting storage executor...")
@@ -41,7 +67,8 @@ class StoreExecutor(TaskExecutor): 
            self.wait()

            if self.srcQueue.len() > 0:

                self.jobObj = self.srcQueue.getJob()

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

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

                self.nodeList = self.jobObj.jobInfo["nodeList"]

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

                # *) HOT

                # 1) Controlla di avere sufficiente spazio su disco sul server
@@ -52,6 +79,8 @@ class StoreExecutor(TaskExecutor): 
                # 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.copyData()

                self.cleanup()

                self.update()

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

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