NETOBSERV-258: add TimeReceived in Kafka transformer #235
Conversation
Codecov Report
@@ Coverage Diff @@
## main #235 +/- ##
==========================================
+ Coverage 61.71% 61.74% +0.02%
==========================================
Files 67 67
Lines 3897 3900 +3
==========================================
+ Hits 2405 2408 +3
Misses 1346 1346
Partials 146 146
Flags with carried forward coverage won't be shown. Click here to find out more.
Continue to review full report at Codecov.
|
| @@ -42,6 +43,9 @@ func (c *DecodeJson) Decode(in []interface{}) []config.GenericMap { | |||
| continue | |||
| } | |||
| decodedLine2 := make(config.GenericMap, len(decodedLine)) | |||
| // flows directly ingested by flp-transformer won't have this field, so we need to add it | |||
| // here. If the received line already contains the field, it will be overridden later | |||
| decodedLine2["TimeReceived"] = time.Now().Unix() | |||
There was a problem hiding this comment.
May be we should add a check to ensure there is no TimeReceived field?
There was a problem hiding this comment.
In that case, if there is no TimeReceived field, it will be added to the current payload. If there is an actual TimeReceived field, it will be added after this, overriding the value that we set here.
There was a problem hiding this comment.
Is it what we want? To override TimeReceived if the field is already here?
There was a problem hiding this comment.
it's ok as a quick fix, but we can think about it and iterate from there.
Maybe it would make more sense to have it in the ingester stage rather than in the json decoder ? it looks more like an ingester responsibility to tell when something is received.
And I agree with Olivier that, if the field is already present, it shouldn't be overwritten. E.g. if we have:
(ovs) -> ipfix-ingester -> kafka-write -> kafka-read -> loki
the received time should be the one of ipfix-ingester .. or maybe at the opposite, on loki side?
As we use it to measure the collection latency (by comparing to FlowEndTime), maybe it makes actually more sense to deal with it directly in the loki write, after all.
There was a problem hiding this comment.
I explained it wrong. The timereceived field is never overridden if its already there (please check the tests)
@jotak currently we set the value into the ipfix or protobuf ingesters but we added a new scenario: the eBPF agent can send it directly to the flowlogs-pipeline-transformer instance directly as a JSON via kafka.
In that case, we are bridging the ingesters and the receive time needs to be set in the transformer instance.
But if we agree that the actual ingest time should be now right before Loki writing, this could be removed and we wouldn't need to consider this special case.
If a flow is directly ingested by the kafka transformer instance (e.g. the eBPF agent submits there directly), it needs to add the
TimeReceivedfield as soon as it unmarshalls the flow data.