Override timestamp in log_to_metric or use native_json

[bmgante]

Hi all,
I am trying to use vector to generate metrics from logs (sent from nifi) using the log_to_metric. It works but i am not able to override the timestamp in the metric with a the .ts field comming from the log (need to send metrics in the past). Below my general vector config:

[sources.nifi]
type = "http"
address = "0.0.0.0:6000"
encoding = "json"

[transforms.remap]
type = "remap"
inputs = ["nifi"]
source = '''
del(.timestamp)
'''

[transforms.log_to_metric]
type = "log_to_metric"
inputs = ["remap"]

  [[transforms.log_to_metric.metrics]]
  type="gauge"
  name="{{name}}"
  field="value"

    [transforms.log_to_metric.metrics.tags]
    "label1" = "{{labels.label1}}"
    "label2" = "{{labels.label2}}"

[sinks.prometheus_remote]
type = "prometheus_remote_write"
inputs = ["log_to_metric"]
endpoint = "http://thanos-receiver.namespace.svc.cluster.local:10908/api/v1/receive"
healthcheck.enabled = false

Tried also to use in alternative the native_json approach but i am always getting error 500 parse error (even with the json well formatted).

[sources.nifi]
type = "http_server"
address = "0.0.0.0:6000"
decoding.codec = "native_json"

[sinks.prometheus_remote]
type = "prometheus_remote_write"
inputs = ["nifi"]
endpoint = "http://thanos-receiver.namespace.svc.cluster.local:10908/api/v1/receive"
healthcheck.enabled = false

Input json:

{
   "metric":{
      "name":"my_metric",
      "counter":{
         "value":17468
      },
      "kind":"incremental"
   }
}

Error from vector:
{"timestamp":"2025-04-30T12:04:23.324593Z","level":"ERROR","message":"unable to process request (internal error)","status":500,"error":"Some(Rejection(ErrorMessage { code: 400, message: \"Failed decoding body: ParsingError(Error parsing JSON: Error(\\\"EOF while parsing an object\\\", line: 1, column: 1))\" }))","target":"warp::filters::trace","span":{"component_id":"nifi","component_kind":"source","component_name":"nifi","component_type":"http_server","name":"source"},"spans":[{"component_id":"nifi","component_kind":"source","component_name":"nifi","component_type":"http_server","name":"source"}]}

Any idea out there?
Thanks

[pront]

Hi @bmgante, you are removing the timestamp with del(.timestamp) and you are not adding one later. Also, please take a look at the event model for metrics: https://vector.dev/docs/about/under-the-hood/architecture/data-model/metric/. It's stricter than logs.

[bmgante]

Hi @pront
Even without that remap got the same behaviour. I am wondering if log_to_metric does not allow timestamp override and considers always time of ingestion.

Wrt the json_native approach, any idea?

[xtrime-ru]

I have same request. Log_to_metric needs metrics.timestamp parameter with support of Vector's template syntax. To override timestamp with value from log contents.

[pront]

Even without that remap got the same behaviour. I am wondering if log_to_metric does not allow timestamp override and considers always time of ingestion.

Hello, I looked at the implementation. Code is here: https://github.com/vectordotdev/vector/blob/master/src/transforms/log_to_metric.rs#L786-L790

The log_to_metric transform should use the timestamp from the log. If a timestamp is not set, it will use now(). Maybe when you were testing this, the log didn't have a timestamp or it was not at the expected path. There are also unit tests for this but an easy way to verify is to override the timestamp and check the end result:

[transforms.remap]
type = "remap"
inputs = ["nifi"]
source = '''
	.timestamp = parse_timestamp!("2025-09-04T13:38:56.519Z", format: "%Y-%m-%dT%H:%M:%S.%fZ")
'''

All metrics should have the same timestamp value "2025-09-04T17:38:56.000000519Z".

[xtrime-ru]

I ended up not using log_to_metric at all. Mainly, because vector influx-sink plugin don't support writing to telegraf in influx line protocol.

Here is my pipeline to count kafka tasks and sending counter to telegraf over tcp socket:

[sources.queue_input_socket]
type = "socket"
mode = "unix_stream"
path = "/var/vector-run/vector.sock"
socket_file_mode = 666
decoding.codec = "json"

[transforms.queue_transform]
type = "remap"
inputs = ["queue_input_http", "queue_input_socket"]
source = '''
.data.topic = .topic
. = .data
'''

[transforms.influx_message]
type = "remap"
inputs = ["queue_transform"]
source = '''
.message = "queue,topic=" + to_string!(.topic) + " vector_received=1 " + to_string!(.headers.timestamp) + "000000000";
'''

[sinks.influx_sum]
type = "socket"
mode = "tcp"
address = "${TELEGRAF_SUM}"
inputs = ["influx_message"]
encoding.codec = "raw_message"
batch.max_bytes = 10485760
batch.max_events = 10000
batch.timeout_secs = 5.0