Elasticsearch sink data_stream @timestamp field overwritten or not recognized ?

[Djeezus]

'lo all,

my vector config somehow doesn't transfer the "@timestamp" field from the source into the actual "@timestamp" field in opensearch document. Vector seems to insert/replace(?) the original Kafka message @timestamp with a timestamp that reflects the time of consumption.

I'm using following config to stream incoming Kafka messages towards an Opensearch 2.2.x., I originally didn't have a transform-section, but with and without result stays the same.

VECTOR configuration:

  sources:
    kafka:
...
  transforms:
    parse_timestamp:
      type: remap
      inputs: [kafka]
      source: |-
        .@timestamp = to_timestamp!(.@timestamp)

  sinks:
    opensearch:
      type: elasticsearch
      inputs: [parse_timestamp]
      endpoints:
        - http://opensearch-test-cluster-coordinator.opensearch.svc:9200
      data_stream:
        auto_routing: false
        type: vectorstream
        dataset: test
      mode: data_stream
      auth:
        strategy: basic
        user: xxxxxx
        password: xxxxxx

INPUT message example:

  "userid": "EQ7T90RXXAFLZC1Q",
  "name": "Kaila Schenk",
  "@timestamp": "2022-12-22T05:56:43.735Z",
  "address": {
    "street": "4447 Home Avenue",
    "town": "Dalbeattie",
    "postode": "GU82 8IB"
  },
...
}

OUTPUT opensearch through Vector:

{
  "_index": ".ds-vectorstream-test-default-000004",
  "_id": "wIQ1WIUBqxDTR_vq1cg6",
  "_version": 1,
  "_score": null,
  "_source": {
    "@timestamp": "2022-12-28T10:10:00.136Z",
    "address": {
      "postode": "GU82 8IB",
      "street": "4447 Home Avenue",
      "town": "Dalbeattie"
    },
    "data_stream": {
      "dataset": "test",
      "namespace": "default",
      "type": "vectorstream"
    },
    "description": "wonderful kent stream comprehensive troops mw resumes let congressional fin clearly chen secrets appearing ep gage know meetings substances brush chances episodes",
    "email": "kristina_saxon@yahoo.com",
    "headers": {},
    "logtype": "json",
    "message_key": null,
    "name": "Kaila Schenk",
    "offset": 7700542,
    "partition": 1,
    "pets": [
      "Lola",
      "Riley"
    ],
    "salary": 16384,
    "score": 6.5,
    "source_type": "kafka",
    "telephone": "+505-5900-068-364",
    "topic": "test-benchmark",
    "url": "http://www.shadow.com",
    "userid": "EQ7T90RXXAFLZC1Q",
    "verified": true
  },
  "fields": {
    "@timestamp": [
      "2022-12-28T10:10:00.136Z"
    ]
  },
  "sort": [
    1672222200136
  ]
}

Notice the @timestamp is not the one from the Kafka source message.

Am I missing something in my configuration ?

grtz,
Gert

[jszwedko]

Hi @Djeezus !

The issue here is that Vector has an internal log schema that determines where to find the timestamp of the log in sinks. By default this field is called timestamp so, in the Elasticsearch sink, it looks for a timestamp field, to send it as @timestamp. This can be changed using the global log_schema.timestamp_key option (note this applies to the entire configuration). Alternatively, you could change your remap transform to be:

  transforms:
    parse_timestamp:
      type: remap
      inputs: [kafka]
      source: |-
        .timestamp = to_timestamp!(del(.@timestamp))

This should put the timestamp where the elasticsearch sink expects to find it.

[Djeezus]

'lo jszwedko,

thank you, that was it !
Both the remapping and the global log_schema.timestamp_key work as expected.

I'm running a comparison test logstash vs vector with datastreams sinks, will keep you posted if something interesting pops up

grtz,
Gert