MonitoringDB

Wrapper on top of ElasticDB. It is used to manage the DIRAC monitoring types.

Configuration Parameters:

The following option can be set in Systems/Monitoring/Databases/MonitoringDB

IndexPrefix: Prefix used to prepend to indexes created in the OpenSearch instance.

For each monitoring types managed, the Period (how often a new index is created) can be defined with:

MonitoringTypes
{
  RMSMonitoring
  {
    # Indexing strategy. Possible values: day, week, month, year, null
    Period = month
  }
  WMSHistory
  {
    # Indexing strategy. Possible values: day, week, month, year, null
    Period = day
  }
}

class DIRAC.MonitoringSystem.DB.MonitoringDB.MonitoringDB(name='Monitoring/MonitoringDB')

Bases: ElasticDB

Extension of ElasticDB for Monitoring system DB

RESULT_SIZE = 10000

__init__(name='Monitoring/MonitoringDB'): Standard constructor

addIndexTemplate(name: str, index_patterns: list, mapping: dict, priority: int = 1, settings: dict = None) → dict

Adds an index template.

Parameters:

self – self reference
name (str) – index name
index_patterns (list) – list of index patterns to match
mapping (dict) – it is the mapping of the index

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 Opensearch
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: str, docID: str)

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…

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: str, docID: str) → bool

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)

getDataForAGivenPeriod(typeName, condDict, initialDate='', endDate='')

Retrieves the history of logging entries for the given component during a given given time period

Parameters:

typeName (str) – name of the monitoring type
condDict (dict) –
conditions for the query
- key -> name of the field
- value -> list of possible values
initialDate (str) – Indicates the start of the time period in the format ‘DD/MM/YYYY hh:mm’
endDate (str) – Indicate the end of the time period in the format ‘DD/MM/YYYY hh:mm’

Returns:

Entries from the database for the given component recorded between the initial and the end dates

getDoc(index: str, docID: str) → dict

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

getDocs(indexFunc, docIDs: list[str], vo: str) → list[dict]

Efficiently retrieve many documents from an index.

Parameters:

index – name of the index
docIDs – document IDs

getIndexName(typeName)

Parameters:: typeName (str) – monitoring type

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

getKeyValues(monitoringType): Get all values for a given key field in a type

getLimitedData(typeName, condDict, size=10)

Returns a list of records for a given selection.

Parameters:

typeName (str) – name of the monitoring type
condDict (dict) –
-> conditions for the query
- key -> name of the field
- value -> list of possible values
size (int) – Indicates how many entries should be retrieved from the log

Returns:

Up to size entries for the given component from the database

getMapping(monitoringType)

It returns the mapping of a certain monitoring type

Parameters:: monitoringType (str) – the monitoring type for example WMSHistory
Returns:: an empty dictionary if there is no mapping defenied.

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)

put(records, monitoringType)

It is used to insert the data to ES. Calls bulk_index()

Parameters:

records (list) – it is a list of documents (dictionary)
monitoringType (str) – is the type of the monitoring

query(index: str, 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 OpenSearch DSL language

retrieveAggregatedData(typeName, startTime, endTime, interval, selectField, condDict, grouping, metainfo={})

Get data from the DB using simple aggregations. Note: this method is equivalent to retrieveBucketedData. The difference is in the dynamic bucketing. We do not perform dynamic bucketing on the raw data.

Parameters:

typeName (str) – name of the monitoring type
startTime (int) – start time, expressed as epoch
endtime (int) – end time, expressed as epoch
interval (str) – interval expressed in ES notation
selectField (str) – the field on which we are aggregating (the plot type)
condDict (dict) – optional additional conditions for the query
grouping (str) – what we are grouping on
metainfo (dict) – meta information

Returns:

S_OK/S_ERROR with dictionary of key/value pairs

retrieveBucketedData(typeName, startTime, endTime, interval, selectField, condDict, grouping, metainfo=None)

Get bucketed data from the DB. This is the standard method used.

Parameters:

typeName (str) – name of the monitoring type
startTime (int) – start time, expressed as epoch
endtime (int) – end time, expressed as epoch
interval (str) – interval expressed in ES notation
selectField (str) – the field on which we are aggregating (the plot type)
condDict (dict) – optional additional conditions for the query
grouping (str) – what we are grouping on
metainfo (dict) – meta information

Returns:

S_OK/S_ERROR with dictionary of key/value pairs

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

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

Parameters:

index – index name
query – It is the query in OpenSearch DSL language
updateByQuery – A bool to determine update by query or index values using index function.
docID – ID for the document to be created.

updateDoc(index: str, docID: str, body) → dict

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