Commit 8f28a57b authored by Cristiano Urban's avatar Cristiano Urban
Browse files

Modified nodes db paths structure. 



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

transfer_service/store_preprocessor.py

+38 −14
Original line number Diff line number Diff line
@@ -3,6 +3,7 @@ 
# - data and permissions cleanup

#



import json

import os

import shutil

import sys
@@ -31,6 +32,12 @@ class StorePreprocessor(Preprocessor): 
                                  self.params["host"], 

                                  self.params.getint("port"), 

                                  self.params["db"])

        self.storageBasePaths = dict()

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

        self.storageBasePaths["tape"] = self.params["base_path"]

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

        self.storageBasePaths["server"] = self.params["base_path"]

        self.jobObj = None        

        self.dbConn.connect()

        super(StorePreprocessor, self).__init__()
@@ -54,6 +61,7 @@ class StorePreprocessor(Preprocessor): 
        fileList = []

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

            cwd = os.path.basename(folder)

            if folder != self.path:

                parent = os.path.dirname(folder)

                dirList.append(parent + '/' + cwd)

                i = 0
@@ -126,35 +134,51 @@ class StorePreprocessor(Preprocessor): 
        [ dirs, files ] = self.scanRecursive()

        

        # File catalog update

        out = open("log.txt", "w")

        for dir in dirs:

            print(f"DIR: {dir}")

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

            if self.jobObj["jobInfo"]["requestType"] == "HSTORE":

                pathPrefix = self.storageBasePaths["tape"].replace("{username}", self.jobObj["jobInfo"]["userName"])

            elif self.jobObj["jobInfo"]["requestType"] == "CSTORE":

                pathPrefix = self.storageBasePaths["server"].replace("{username}", self.jobObj["jobInfo"]["userName"])

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

            basePath = os.path.dirname(dir).replace(pathPrefix, "/" + self.jobObj["jobInfo"]["userName"])

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

            cnode = Node(os.path.basename(dir), "container");            

            cnode.setParentPath(os.path.dirname(dir))

            cnode.setParentPath(basePath)

            cnode.setOwnerID("3354")

            cnode.setCreatorID("3354")

            self.dbConn.insertNode(cnode, cnode.parentPath)

            

        for flist in files:

            for file in flist:

                print(f"FILE: {file}")

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

                if self.md5calc.fileIsValid(file):

                    if self.jobObj["jobInfo"]["requestType"] == "HSTORE":

                        pathPrefix = self.storageBasePaths["tape"].replace("{username}", self.jobObj["jobInfo"]["userName"])

                    elif self.jobObj["jobInfo"]["requestType"] == "CSTORE":

                        pathPrefix = self.storageBasePaths["server"].replace("{username}", self.jobObj["jobInfo"]["userName"])            

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

                    basePath = os.path.dirname(file).replace(pathPrefix, "/" + self.jobObj["jobInfo"]["userName"])

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

                    dnode = Node(os.path.basename(file), "data")

                    dnode.setParentPath(os.path.dirname(file))

                    dnode.setParentPath(basePath)

                    dnode.setOwnerID("3354")

                    dnode.setCreatorID("3354")

                    dnode.setContentMD5(self.md5calc.getMD5(file))                

                    self.dbConn.insertNode(dnode, dnode.parentPath)



        out.close()

        self.dbConn.disconnect()

        

    def run(self):

        while True:

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

                jobObj = self.pendingQueue.getJob()

                self.prepare(jobObj["jobInfo"]["userName"])

                self.jobObj = self.pendingQueue.getJob()

                self.prepare(self.jobObj["jobInfo"]["userName"])

                self.execute()

                self.pendingQueue.moveJobTo("ready")

                print("Job MOVED:")

                print("Job MOVED!")

            time.sleep(1)



# Test