Monitoring generating insane timestamp

[chaen]

We regularly have entries for all types of monitoring that fails in being inserted in ElasticSearch with errors like

BulkIndexError: ('259 document(s) failed to index.', [{u'index': {u'status': 400, u'_type': u'_doc', u'_index': u'lhcb-certification_wmshistory_index-2021-08-10', u'error': {u'caused_by': {u'caused_by': {u'reason': u'date_time_parse_exception: Failed to parse with all enclosed parsers', u'type': u'date_time_parse_exception'}, u'reason': u'failed to parse date field [1627194654000000000000000000000000000000000000000000000000000000000000000000000000000000] with format [strict_date_optional_time||epoch_millis]', 

What happens is the following:

1. An entry is added to the MonitoringReporter

   DIRAC/src/DIRAC/MonitoringSystem/Client/MonitoringReporter.py

   Line 99 in cbce4f0

   def addRecord(self, rec):

2. At some point, the records are commited with

   DIRAC/src/DIRAC/MonitoringSystem/Client/MonitoringReporter.py

   Line 127 in cbce4f0

   def commit(self):

3. In this method, we take a slice of all the records to be committed, and pass it to MonitoringDB for treatment

   DIRAC/src/DIRAC/MonitoringSystem/Client/MonitoringReporter.py

   Lines 147 to 148 in cbce4f0

   	recordsToSend = documents[:self.__maxRecordsInABundle]
   	retVal = monitoringDB.put(recordsToSend, self.__monitoringType)

4. What is eventually called is generateDocs with this slice of records

   DIRAC/src/DIRAC/Core/Utilities/ElasticSearchDB.py

   Line 47 in cbce4f0

   def generateDocs(data, withTimeStamp=True):

5. In this method, we manipulate the timestamp. Either create it, or convert it, or, if it is Epoch, multiply it by 1000

   DIRAC/src/DIRAC/Core/Utilities/ElasticSearchDB.py

   Lines 64 to 75 in cbce4f0

   	try:
   	if isinstance(timestamp, datetime):
   	doc['timestamp'] = int(timestamp.strftime('%s')) * 1000
   	elif isinstance(timestamp, six.string_types):
   	timeobj = datetime.strptime(timestamp, '%Y-%m-%d %H:%M:%S.%f')
   	doc['timestamp'] = int(timeobj.strftime('%s')) * 1000
   	else: # we assume the timestamp is an unix epoch time (integer).
   	doc['timestamp'] = timestamp * 1000
   	except (TypeError, ValueError) as e:
   	# in case we are not able to convert the timestamp to epoch time....
   	sLog.error("Wrong timestamp", e)
   	doc['timestamp'] = int(Time.toEpoch()) * 1000

6. If ever, the DB insertion fails, for a reason or another, we try to send them to the MessageQueue, and if that fails, keep them for the next attempt

   DIRAC/src/DIRAC/MonitoringSystem/Client/MonitoringReporter.py

   Lines 149 to 161 in cbce4f0

   	if retVal['OK']:
   	recordSent += len(recordsToSend)
   	del documents[:self.__maxRecordsInABundle]
   	gLogger.verbose("%d records inserted to MonitoringDB" % (recordSent))
   	else:
   	if mqProducer is not None:
   	res = self.publishRecords(recordsToSend, mqProducer)
   	# if we managed to publish the records we can delete from the list
   	if res['OK']:
   	recordSent += len(recordsToSend)
   	del documents[:self.__maxRecordsInABundle]
   	else:
   	return res # in case of MQ problem

7. But at the next attempt, you will go through the same logic, in particular step 5. So you end up multiplying the timestamp again, and that's how you end up with crazy records like the one above

The reason this happens is that the original dictionary is updated. It is illustrated by the following example

record1 = {'timestamp':1}
record2 = {'timestamp':2}
records = [record1, record2]

# prints Before commit  [{'timestamp': 1}, {'timestamp': 2}]
print("Before commit ", records)


# equivalent of MonitoringReporter.commit

recordsToSend = records[:1]
# equivalent of generateDocs
for doc  in recordsToSend:
    doc['timestamp'] = doc['timestamp']*1000

# After commit  [{'timestamp': 1000}, {'timestamp': 2}]
print("After commit ", records)

While there is maybe a good reason for multiplying epoch by 1000 (is there really ?), we should:

• Not update the document in place
• Not systematically multiply by 1000, but only if it is needed

I did not yet investigate which solution is the best, nor which are the consequences of doing so. Not sure I am the best person to do it though.

But frankly, this monitoring is again a piece of art when it comes to over designing and complexifying things

[chaen]

Giving this more thoughts.

First of all, adding the timestamp when committing is wrong. It should be done when adding the record, not when trying to send it to ES, since there can be a relatively large delay of the order of minutes.

Second, the reason for all these multiplication/division is because epoch_millis is the default to store epoch in ES. I think it would make way more sense to store just seconds, so epoch_seconds.
We could bother with a compatibility layer (like, divide if it is a multiple of 1000 or something like that), BUT I think that we should rather properly fix it in the next major release, remove all those multiplication and division, and fix the existing entries in ES.

Luckily, this seems doable relatively easily. From the CERN ES team:

IIUC you essentially want to go on all documents in an index and divide the timestamp field with 1000. Sounds like a job for update API. I ran a quick test for 1 doc and it seemed to work, however I'd suggest to test it first with your data. This is what I did:

PUT _template/mytest
{
  "order" : 1,
  "index_patterns" : [
    "tes*"
  ],
  "settings" : { },
  "mappings" : {
    "properties" : {
      "counter" : {  "type" : "long" },
      "tags" : {  "type" : "text"  },
      "timestamp": { "type": "date"  }
    }
  },
  "aliases" : { }
}


PUT test/_doc/1
{
  "counter" : 1,
  "tags" : ["red"],
  "timestamp": 1628682087372
}

POST test/_update/1
{
  "script" : {
    "source": "ctx._source.timestamp /= params.count",
    "lang": "painless",
    "params" : {
      "count" : 1000
    }
  }
}

GET test/_doc/1
{
  "_index" : "test",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 2,
  "_seq_no" : 1,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "counter" : 1,
    "tags" : [
      "red"
    ],
    "timestamp" : 1628682087
  }
}

And we should do it at the same time as #5275

datetime field: https://www.elastic.co/guide/en/elasticsearch/reference/current/date.html
update API: https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update.html

[fstagni]

Please check #5328 for a possible solution to the bug.

Regarding the other issue here (milliseconds vs seconds), it would imply coordinating the deploy with an intervention which is of course delicate. To be discussed.

[chaen]

That's not a good fix. It solves that issue but can largely falsify the data. As I said a bit higher:

First of all, adding the timestamp when committing is wrong. It should be done when adding the record, not when trying to send it to ES, since there can be a relatively large delay of the order of minutes.

With this fix, we will be recreating the timestamp everytime we are trying to send the record. I think this fix should be reverted

[fstagni]

I am aware that the fix in #5328 is "not a full solution to the issue" (this is what I wrote in the PR): what is coded is a quick solution for the error mentioned in the description, but clearly it was not for solving it fully. Just reverting it without providing a better solution is not making things better, IMHO.

[rupozzi]

I have been looking into how to solve this directly on the indices but there are some things that we should keep in mind after running some tests:

• If we want to divide the timestamp that we have in each document in ES by 1000, the new timestamp will not be be "detected" by Kibana as a timestamp in seconds, since the mapping of the indices themselves is set to milliseconds, so it will show up on the first of January 1970, that is, the earliest possible date.
• In order to solve this we should change the mapping of the indices and set the timestamp format in seconds, but this is not possible as this error shows:
  "Mapper for [timestamp] conflicts with existing mapper:\n\tCannot update parameter [format] from [strict_date_optional_time||epoch_millis] to [strict_date_optional_time||epoch_second]"
• There is though one way to "update" the mapping but that is to create a new index for each existing one and re-index all of them with a new mapping.

The other option would be to go without touching the indices and instead just change the way we add and commit records, so directly add records with a timestamps in milliseconds, and removing the part where the timestamp is being multiplied by 1000.