CTAStorage

class DIRAC.Resources.Storage.CTAStorage.CTAStorage(storageName, parameters)

Bases: GFAL2_XROOTStorage

Plugin to interact with CERN CTA.

It basically is XROOT with added tape capabilities. Since CTA supports ONLY xroot, do not forget to add xroot in your Operations/<Setup>/DataManagement/RegistrationProtocols list

Configuration example:

StorageElements
{
  CTA-PPS
  {
    BackendType = Cta
    AccessProtocols = root
    WriteProtocols = root
    # This is very important if you have to stage with this protocol,  but might transfer
    # using a different protocol, like https
    StageProtocols = root
    SEType = T1D0
    SpaceReservation = LHCb-Tape
    OccupancyLFN = /eos/ctalhcbpps/proc/accounting
    OccupancyPlugin = WLCGAccountingJson
    # Config for this plugin is below
    ###################################
    CTA
    {
      Host = eosctalhcbpps.cern.ch
      Protocol = root
      Path = /eos/ctalhcbpps/archivetest/
      Access = remote
    }
    ###################################
    GFAL2_HTTPS
    {
      Host = eosctalhcbpps.cern.ch
      Protocol = https
      Path = /eos/ctalhcbpps/archivetest/
      Access = remote
    }
  }
}

DYNAMIC_OPTIONS = {'SvcClass': 'svcClass'}

PROTOCOL_PARAMETERS = ['Protocol', 'Host', 'Path', 'Port', 'SpaceToken', 'WSUrl', 'SvcClass']

__init__(storageName, parameters)

c’tor

Parameters:

self – self reference
storageName (str) – SE name
parameters (dict) – passed to parent’s class

changeDirectory(directory): Change the directory to the supplied directory

constructURLFromLFN(lfn, withWSUrl=False): Overwrite to add the double slash

createDirectory(path) → DOKReturnType[T] | DErrorReturnType

Create directory on the storage

Parameters:: path (str) – path to be created on the storage (pfn : srm://…)
Returns:: Successful dict {path : True } Failed dict {path : error message } S_ERROR in case of argument problems

exists(path) → DOKReturnType[T] | DErrorReturnType

Check if the path exists on the storage

Parameters:: path (str) – path or list of paths to be checked
Returns:: Failed dictionary: {pfn : error message} Successful dictionary: {pfn : bool} S_ERROR in case of argument problems

getCurrentDirectory(): Get the current directory

getCurrentURL(fileName): Overwrite to add the double slash

getDirectory(path, localPath: str | None = None) → DOKReturnType[T] | DErrorReturnType

get a directory from the SE to a local path with all its files and subdirectories

Parameters:

path (str) – path (or list of paths) on the storage (srm://…)
localPath (str) – local path where the content of the remote directory will be saved, if not defined it takes current working directory.

Returns:

successful and failed dictionaries. The keys are the paths, the values are dictionary {‘Files’: amount of files downloaded, ‘Size’ : amount of data downloaded} S_ERROR in case of argument problems

getDirectoryMetadata(path) → DOKReturnType[T] | DErrorReturnType

Get metadata for the directory(ies) provided

Parameters:: path (str) – path (or list of paths) on storage (srm://…)
Returns:: Successful dict {path : metadata} Failed dict {path : errStr} S_ERROR in case of argument problems

getDirectorySize(path) → DOKReturnType[T] | DErrorReturnType

Get the size of the directory on the storage

Warning

it is not recursive

Parameters:

path (str) – path or list of paths on storage (srm://…)

Returns:

list of successful and failed dictionaries, both indexed by the path

In the failed, the value is the error message
In the successful the values are dictionaries:
- Files : amount of files in the dir
- Size : summed up size of all files
- subDirs : amount of sub dirs
S_ERROR in case of argument problems

getEndpoint()

This will get endpoint of the storage. It basically is the same as getURLBase(): but without the basePath

Returns:: ‘proto://hostname<:port>’

getFile(path, localPath=False) → DOKReturnType[T] | DErrorReturnType

Make a local copy of storage :path:

Parameters:

path (str) – path (or list of paths) on storage (srm://…)
localPath – destination folder. Default is from current directory

Returns:

Successful dict: {path : size} Failed dict: {path : errorMessage} S_ERROR in case of argument problems

getFileMetadata(path) → DOKReturnType[T] | DErrorReturnType

Get metadata associated to the file(s)

Parameters:: path (str) – path (or list of paths) on the storage (srm://…)
Returns:: successful dict { path : metadata } failed dict { path : error message } S_ERROR in case of argument problems

getFileSize(path) → DOKReturnType[T] | DErrorReturnType

Get the physical size of the given file

Parameters:: path – path (or list of path) on storage (pfn : srm://…)
Returns:: Successful dict {path : size} Failed dict {path : error message } S_ERROR in case of argument problems

getName(): The name with which the storage was instantiated

getOccupancy(**kwargs)

Get the StorageElement occupancy info in MB.

This generic implementation download a json file supposed to contain the necessary info.

Parameters:: occupancyLFN – (mandatory named argument) LFN of the json file.
Returns:: S_OK/S_ERROR dictionary. The S_OK value should contain a dictionary with Total and Free space in MB

getParameters(): Get the parameters with which the storage was instantiated

getTransportURL(pathDict, protocols)

Get a transport URL for a given URL. For a simple storage plugin it is just returning input URL if the plugin protocol is one of the requested protocols

Parameters:

pathDict (dict) – URL obtained from File Catalog or constructed according to convention
protocols (python:list) – a list of acceptable transport protocols in priority order

getURLBase(withWSUrl=False): Overwrite to add the double slash

getWLCGTokenPath(lfn: str, wlcgTokenBasePath: str) → str: Returns the path expected to be in a WLCG token It basically consists of basepath - tokenBasePath + LFN The tokenBasePath is a configuration on the storage side.

isDirectory(path) → DOKReturnType[T] | DErrorReturnType

check if the path provided is a directory or not

Parameters:: str – path or list of paths to be checked ( ‘srm://…’)
Returns:: dict ‘Failed’ : failed, ‘Successful’ : succesful S_ERROR in case of argument problems

isFile(path) → DOKReturnType[T] | DErrorReturnType

Check if the path provided is a file or not

Parameters:: str – path or list of paths to be checked ( ‘srm://…’)
Returns:: Failed dict: {path : error message} Successful dict: {path : bool} S_ERROR in case of argument problems

isNativeURL(url)

Check if URL :url: is valid for :self.protocol:

Parameters:

self – self reference
url (str) – URL

isURL(path)

Guess if the path looks like a URL

Parameters:

self – self reference
path (string) – input file LFN or URL

Returns boolean:

True if URL, False otherwise

listDirectory(path) → DOKReturnType[T] | DErrorReturnType

List the content of the path provided

TODO: add an option if we want or not the metadata, as it is an expensive operation ?

Parameters:: path (str) – single or list of paths (srm://…)
Returns:: failed dict {path : message } successful dict { path : {‘SubDirs’ : subDirs, ‘Files’ : files} }. They keys are the paths, the values are the dictionary ‘SubDirs’ and ‘Files’. Each are dictionaries with path as key and metadata as values S_ERROR in case of argument problems

prestageFile(path, lifetime: int = 86400) → DOKReturnType[T] | DErrorReturnType

Issue prestage request for file(s)

Parameters:

path (str) – path or list of paths to be prestaged
lifetime (int) – prestage lifetime in seconds (default 24h)

Returns:

succesful dict { url : token } failed dict { url : message } S_ERROR in case of argument problems

prestageFileStatus(path) → DOKReturnType[T] | DErrorReturnType

Checking the staging status of file(s) on the storage

Parameters:: path (dict) – dict { url : token }
Returns:: succesful dict { url : bool } failed dict { url : message } S_ERROR in case of argument problems

putDirectory(path) → DOKReturnType[T] | DErrorReturnType

Puts one or more local directories to the physical storage together with all its files

Parameters:: path (str) – dictionary { srm://… (destination) : localdir (source dir) }
Returns:: successful and failed dictionaries. The keys are the paths, the values are dictionary {‘Files’ : amount of files uploaded, ‘Size’ : amount of data upload } S_ERROR in case of argument problems

putFile(path, sourceSize: int = 0) → DOKReturnType[T] | DErrorReturnType

Put a copy of a local file or a file on another srm storage to a directory on the physical storage.

Parameters:

path – dictionary { lfn (srm://…) : localFile }
sourceSize – size of the file in byte. Mandatory for third party copy (WHY ???) Also, this parameter makes it essentially a non bulk operation for third party copy, unless all files have the same size…

Returns:

Successful dict: { path : size } Failed dict: { path : error message } S_ERROR in case of argument problems

releaseFile(path) → DOKReturnType[T] | DErrorReturnType

Release a pinned file

Parameters:: path (str) – PFN path { pfn : token } - pfn can be an empty string, then all files that have that same token get released. Just as you can pass an empty token string and a directory as pfn which then releases all the files in the directory an its subdirectories
Returns:: successful dict {url : token}, failed dict {url : message} S_ERROR in case of argument problems

removeDirectory(path, recursive: bool = False) → DOKReturnType[T] | DErrorReturnType

Remove a directory on the physical storage together with all its files and subdirectories.

Parameters:

path – single or list of path (srm://..)
recursive – if True, we recursively delete the subdir

Returns:

successful and failed dictionaries. The keys are the pathes, the values are dictionary {‘Files’: amount of files deleted, ‘Size’: amount of data deleted} S_ERROR in case of argument problems

removeFile(path) → DOKReturnType[T] | DErrorReturnType

Physically remove the file specified by path

A non existing file will be considered as successfully removed

Parameters:: path (str) – path (or list of paths) on storage (srm://…)
Returns:: Successful dict {path : True} Failed dict {path : error message} S_ERROR in case of argument problems

resetCurrentDirectory(): Reset the working directory to the base dir

setParameters(parameterDict): Set standard parameters, method can be overriden in subclasses to process specific parameters

setStorageElement(se)

updateURL(url, withWSUrl=False): Update the URL according to the current SE parameters