Commit 84990f2a authored by Cristiano Urban's avatar Cristiano Urban
Browse files

Minor adds. 



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

transfer_service/store_amqp_server.py

+7 −3
Original line number Diff line number Diff line
@@ -29,7 +29,9 @@ class StoreAMQPServer(AMQPServer): 
                                  self.params["host"], 

                                  self.params.getint("port"), 

                                  self.params["db"])

        self.pendingQueueWrite = JobQueue("pending")        

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

        self.storageStorePath = self.params["store_path"]

        self.pendingQueueWrite = JobQueue("write_pending")        

        self.job = None

        self.username = None

        self.path = None
@@ -45,7 +47,8 @@ class StoreAMQPServer(AMQPServer): 
            self.job.setType("other")            

            self.job.setInfo(requestBody)

            self.job.setPhase("PENDING")            

            folderPath = "/home/" + user + "/store"

            #folderPath = "/home/" + user + "/store"

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

            userInfo = self.userInfo(user)

            # Check if the user exists on the transfer node

            if not userInfo:
@@ -105,7 +108,8 @@ class StoreAMQPServer(AMQPServer): 
    # or simply add a chmod -x here, to be faster?

    def prepare(self, username):

        self.username = username

        self.path = "/home/" + username + "/store"

        #self.path = "/home/" + username + "/store"

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

        for folder, subfolders, files in os.walk(self.path):

            os.chown(folder, 0, 0)

            os.chmod(folder, 0o555)

transfer_service/store_preprocessor.py

+30 −30
Original line number Diff line number Diff line
@@ -15,11 +15,11 @@ from checksum import Checksum 
from file_grouper import FileGrouper

from db_connector import DbConnector

from node import Node

from preprocessor import Preprocessor

from task_executor import TaskExecutor

from config import Config





class StorePreprocessor(Preprocessor):

class StorePreprocessor(TaskExecutor):



    def __init__(self):

        self.md5calc = Checksum()
@@ -154,7 +154,6 @@ class StorePreprocessor(Preprocessor): 
                if tstampWrapperDirPattern.search(basePath):

                    tstampWrapperDir = tstampWrapperDirPattern.search(basePath).group(0).lstrip('/')

                    out.write(f"tstampWrapperDir: {tstampWrapperDir}\n")

                    #basePath = tstampWrapperDirPattern.sub("", basePath).rstrip('/')

                    basePath = tstampWrapperDirPattern.sub("", basePath)

                    out.write(f"newBasePath: {basePath}\n")

                    cnode.setWrapperDir(tstampWrapperDir)
@@ -175,7 +174,6 @@ class StorePreprocessor(Preprocessor): 
                    nodeName = os.path.basename(file)

                    out.write(f"nodeName: {nodeName}\n")

                    dnode = Node(nodeName, "data")

                    #if not tstampWrapperDirPattern.match(nodeName):

                    if tstampWrapperDirPattern.search(basePath):

                        tstampWrapperDir = tstampWrapperDirPattern.search(basePath).group(0).lstrip('/')

                        out.write(f"tstampWrapperDir: {tstampWrapperDir}\n")
@@ -193,14 +191,16 @@ class StorePreprocessor(Preprocessor): 
        self.dbConn.disconnect()



    def run(self):

        self.setSourceQueueName("write_pending")

        self.setDestinationQueueName("write_ready")

        while True:

            self.wait()

            if self.readyQueue.len() <= self.maxReadyJobs and self.pendingQueue.len() > 0:

                self.jobObj = self.pendingQueue.getJob()

            if self.destQueue.len() < self.maxReadyJobs and self.srcQueue.len() > 0:

                self.jobObj = self.srcQueue.getJob()

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

                self.prepare(self.username)

                self.execute()

                self.pendingQueue.moveJobTo("ready")

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

                print("Job MOVED!")



# Test