ElasticJobParametersDB

Module containing a front-end to the ElasticSearch-based ElasticJobParametersDB. This module interacts with one ES index: “ElasticJobParametersDB”, which is a drop-in replacement for MySQL-based table JobDB.JobParameters. While JobDB.JobParameters in MySQL is defined as:

CREATE TABLE `JobParameters` (
  `JobID` INT(11) UNSIGNED NOT NULL,
  `Name` VARCHAR(100) NOT NULL,
  `Value` BLOB NOT NULL,
  PRIMARY KEY (`JobID`,`Name`),
  FOREIGN KEY (`JobID`) REFERENCES `Jobs`(`JobID`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

Here we define a mapping as:

"JobID": {"type": "long"},
"Name": {"type": "keyword"},
"Value": {"type": "text"}

which is an “equivalent” representation.

The reason for switching to a ES-based JobParameters lies in the extended searching capabilities of ES (ES will analyze+index text fields, while MySQL won’t do that on BLOB types). This results in higher traceability for DIRAC jobs.

The following class methods are provided for public usage

getJobParameters()
setJobParameter()
deleteJobParameters()

class DIRAC.WorkloadManagementSystem.DB.ElasticJobParametersDB.ElasticJobParametersDB

Bases: DIRAC.Core.Base.ElasticDB.ElasticDB

RESULT_SIZE = 10000

__init__(): Standard Constructor

bulk_index(indexPrefix, data=None, mapping=None, period='day', withTimeStamp=True)

Parameters

indexPrefix (str) – index name.
data (list) – contains a list of dictionary
mapping (dict) – the mapping used by elasticsearch
period (str) – Accepts ‘day’ and ‘month’. We can specify which kind of indexes will be created.
withTimeStamp (bool) – add timestamp to data, if not there already.

Returns

S_OK/S_ERROR

clusterName = ''

createIndex(indexPrefix, mapping=None, period='day')

Parameters

indexPrefix (str) – it is the index name
mapping (dict) – the configuration of the index.
period (str) – We can specify, which kind of index will be created. Currently only daily and monthly indexes are supported.

deleteByQuery(indexName, query)

Delete data by query (careful!)

Parameters

indexName (str) – the name of the index
query (str) – the JSON-formatted query for which we want to issue the delete

deleteDoc(index, docID)

Deletes a document in an index.

Parameters

index – name of the index
docID – document ID

deleteIndex(indexName)

Parameters: indexName (str) – the name of the index to be deleted…

deleteJobParameters(jobID, paramList=None)

Deletes Job Parameters defined for jobID.: Returns a dictionary with the Job Parameters. If paramList is empty - all the parameters for the job are removed

Parameters

self – self reference
jobID – Job ID
paramList – list of parameters to be returned (also a string is treated)

Returns

S_OK()/S_ERROR()

existingIndex(indexName)

Checks the existance of an index, by its name

Parameters: indexName (str) – the name of the index
Returns: S_OK/S_ERROR if the request is successful

existsDoc(index, docID)

Returns information about whether a document exists in an index.

Parameters

index – name of the index
docID – document ID

static generateFullIndexName(indexName, period)

Given an index prefix we create the actual index name.

Parameters

indexName (str) – it is the name of the index
period (str) – We can specify which kind of indexes will be created (day, week, month, year, null).

Returns

string with full index name

getCSOption(optionName, defaultValue=None)

getDoc(index, docID)

Retrieves a document in an index.

Parameters

index – name of the index
docID – document ID

getDocTypes(indexName)

Returns mappings, by index.

Parameters: indexName (str) – is the name of the index…
Returns: S_OK or S_ERROR

getIndexPrefix(): It returns the DIRAC setup.

getIndexes(indexName=None): It returns the available indexes…

getJobParameters(jobID, paramList=None)

Get Job Parameters defined for jobID.: Returns a dictionary with the Job Parameters. If paramList is empty - all the parameters are returned.

Parameters

self – self reference
jobID – Job ID
paramList – list of parameters to be returned (also a string is treated)

Returns

dict with all Job Parameter values

getUniqueValue(indexName, key, orderBy=False)

Parameters

indexName (str) – the name of the index which will be used for the query
orderBy (dict) – it is a dictionary in case we want to order the result {key:’desc’} or {key:’asc’}

Returns

a list of unique value for a certain key from the dictionary.

index(indexName, body=None, docID=None, op_type='index')

Parameters

indexName (str) – the name of the index to be used
body (dict) – the data which will be indexed (basically the JSON)
id (int) – optional document id
op_type (str) – Explicit operation type. (options: ‘index’ (default) or ‘create’)

Returns

the index name in case of success.

pingDB()

Try to connect to the database

Returns: S_OK(TRUE/FALSE)

query(index, query)

Executes a query and returns its result (uses ES DSL language).

Parameters

self – self reference
index (str) – index name
query (dict) – It is the query in ElasticSearch DSL language

setJobParameter(jobID, key, value)

Inserts data into ElasticJobParametersDB index

Parameters

self – self reference
jobID – Job ID
key – parameter key
value – parameter value

Returns

S_OK/S_ERROR as result of indexing

setJobParameters(jobID, parameters)

Inserts data into ElasticJobParametersDB index using bulk indexing

Parameters

self – self reference
jobID – Job ID
parameters – list of tuples (name, value) pairs

Returns

S_OK/S_ERROR as result of indexing

update(index, query=None, updateByQuery=True, docID=None)

Executes an update of a document, and returns S_OK/S_ERROR

Parameters

self – self reference
index (str) – index name
query (dict) – It is the query in ElasticSearch DSL language
updateByQuery (bool) – A bool to determine update by query or index values using index function.
docID (int) – ID for the document to be created.

updateDoc(index, docID, body)

Update an existing document with a script or partial document

Parameters

index – name of the index
docID – document ID
body – The request definition requires either script or partial doc