Prometheus Remote Write Sink Timestamp

[BinjaFan]

Hello,

I’m currently testing a setup where:

• Vector A sends metrics to a Kafka topic.
• Vector B consumes those metrics from Kafka.
• Vector B then writes them remotely to Prometheus.

Here’s the issue:

• When Prometheus is unavailable for a while, Vector doesnt acknowledges messages in kafka (which what i expect with acknowledgements set to true)
• Vector acknowledges metrics in Kafka as soon as Prometheus becomes available again.
• Although it looks like Vector is sending the data, I see gaps in Prometheus for the period when it was down.
• I’m not sure if Vector is sending the original timestamps to Prometheus or not or it is something on prometheus side.

I believe Vector should handle it since i tested the same thing using prometheus agent and it works without any issue.

Could someone please help me figure out how to preserve these timestamps so I don’t have gaps?

Below is my Vector B configuration:

---
sources:
  metrics:
    type: kafka
    bootstrap_servers: localhost:19092
    topics:
      - metrics
    group_id: metrics
    decoding:
      codec: native
    acknowledgements:
      enabled: true

sinks:
  rw:
    type: prometheus_remote_write
    inputs:
      - metrics
    endpoint: http://localhost:9090/api/v1/write
    batch:
      timeout_secs: 30 ## send data every 30 seconds
    healthcheck:
      enabled: false
    acknowledgements:
      enabled: true

[BinjaFan]

I made a simple comparison between how Vector is behaving vs Prometheus agent:

Prometheus agent is sending data points in order like this:

TimeSeries #0:
  Labels:
    __name__ = logs_lines_total
    filename = TODO
    from_agent = true
    instance = 192.168.1.162:9001
    job = prometheus
  Samples:
    Value: 757.000000, Timestamp: 1735502689373 - 1735502689 - 2024-12-29 23:34:49 +0330 +0330

TimeSeries #6:
  Labels:
    __name__ = logs_lines_total
    filename = TODO
    from_agent = true
    instance = 192.168.1.162:9001
    job = prometheus
  Samples:
    Value: 771.000000, Timestamp: 1735502704373 - 1735502704 - 2024-12-29 23:35:04 +0330 +0330

but vector is sending in wrong order, i believe thats why it is like that:

TimeSeries #0:
  Labels:
    __name__ = log_lines
    filename = TODO
  Samples:
    Value: 390.000000, Timestamp: 1735501023616 - 1735501023 - 2024-12-29 23:07:03 +0330 +0330
TimeSeries #0:
  Labels:
    __name__ = log_lines
    filename = TODO
  Samples:
    Value: 268.000000, Timestamp: 1735500779488 - 1735500779 - 2024-12-29 23:02:59 +0330 +0330
TimeSeries #0:
  Labels:
    __name__ = log_lines
    filename = TODO
  Samples:
    Value: 391.000000, Timestamp: 1735501025678 - 1735501025 - 2024-12-29 23:07:05 +0330 +0330
TimeSeries #0:
  Labels:
    __name__ = log_lines
    filename = TODO
  Samples:
    Value: 265.000000, Timestamp: 1735500773305 - 1735500773 - 2024-12-29 23:02:53 +0330 +0330
TimeSeries #0:
  Labels:
    __name__ = log_lines
    filename = TODO
  Samples:
    Value: 393.000000, Timestamp: 1735501029799 - 1735501029 - 2024-12-29 23:07:09 +0330 +0330

it is a simple test i wrote to check how they are sending data.

server which mocks prometheus remote write api:

package main

import (
        "fmt"
        "log"
        "net/http"
        "time"

        "github.com/prometheus/prometheus/storage/remote"
)

func main() {
        http.HandleFunc("/receive", func(w http.ResponseWriter, r *http.Request) {
                fmt.Println()
                log.Println("new request for mock endpoint")
                log.Println(">>>")
                defer r.Body.Close()
                writeReq, err := remote.DecodeWriteRequest(r.Body)
                if err != nil {
                        fmt.Println("errrrrrrrr", err)
                }

                // Print out the data
                for i, ts := range writeReq.Timeseries {
                        fmt.Printf("TimeSeries #%d:\n", i)
                        fmt.Println("  Labels:")
                        for _, lbl := range ts.Labels {
                                fmt.Printf("    %s = %s\n", lbl.Name, lbl.Value)
                        }
                        fmt.Println("  Samples:")
                        for _, smp := range ts.Samples {
                                TA := time.Unix(smp.Timestamp/1000, 0)
                                fmt.Printf("    Value: %f, Timestamp: %d - %d - %v\n", smp.Value, smp.Timestamp, smp.Timestamp/1000, TA)
                        }
                }
                fmt.Println("<<<")
                fmt.Println()

                w.WriteHeader(http.StatusOK)
                _, _ = w.Write([]byte("OK\n"))
        })

        log.Println("Listening on :8081")
        log.Fatal(http.ListenAndServe(":8081", nil))
}

[BinjaFan]

@jszwedko

could you please take a look at this?

[jszwedko]

The prometheus_remote_write sink should be using the timestamp encoded in the metric. You can see this here:

vector/src/sinks/prometheus/collector.rs

Line 41 in 029a2ff

let timestamp = metric.timestamp().map(|t| t.timestamp_millis());

Regarding ordering, batches can be sent concurrently in the prometheus_remote_write sink, but the ordering in each batch, I believe, should be the same order that the events were consumed from kafka so it is odd that you are seeing events out-of-order. I'd suggest using vector tap (will require enabling the API if you haven't already) to view the events coming out of the kafka source and see if they are in the expected order.

[BinjaFan]

okay, thank you very much!

i'll check and share the results with you.

[BinjaFan]

@jszwedko

i can not get kafka producer and prometheus remote write tap data.

when trying to tap to prometheus remote write

[tap] Warning: sink outputs cannot be tapped. Output pattern '*' matches sinks ["prometheusRemoteWrite"]

when trying to tap to kafka producer:

[tap] Warning: sink outputs cannot be tapped. Output pattern '*' matches sinks ["kafkaMetricsProducer"]

but from kafka consumer which is reading generated metrics from kafka i can see all the timestamps are correct:

component_id=kafkaMetricsConsumer component_kind=source component_type=kafka counter.value=2 kind=incremental name=log_lines tags.filename=TODO timestamp=2025-01-04T15:42:17.229Z
component_id=kafkaMetricsConsumer component_kind=source component_type=kafka counter.value=2 kind=incremental name=log_lines tags.filename=TODO timestamp=2025-01-04T15:42:21.344Z
component_id=kafkaMetricsConsumer component_kind=source component_type=kafka counter.value=3 kind=incremental name=log_lines tags.filename=TODO timestamp=2025-01-04T15:42:27.527Z
component_id=kafkaMetricsConsumer component_kind=source component_type=kafka counter.value=2 kind=incremental name=log_lines tags.filename=TODO timestamp=2025-01-04T15:42:31.647Z
component_id=kafkaMetricsConsumer component_kind=source component_type=kafka counter.value=3 kind=incremental name=log_lines tags.filename=TODO timestamp=2025-01-04T15:42:36.799Z

i believe issue is inside of prometheus remote write.

is there any chance for prometheus remote write to buffer the data and when remote write endpoint becomes available again, a race condition is happening between the buffered data and the ones which are getting read from the input and the ones which are getting retried(not sure if retry logic is implemented)?

[jszwedko]

You can tap the inputs to the sink with vector tap --inputs-of. That may be helpful to validate that it is receiving the events in order.

I am noticing that your receiver is handling requests concurrently. I'd suggest adding a prefix to each line of its output with a generated request ID. I'm curious if points are in-order within a request, but that you are receiving concurrent requests which will result in interleaved output.

is there any chance for prometheus remote write to buffer the data and when remote write endpoint becomes available again, a race condition is happening between the buffered data and the ones which are getting read from the input and the ones which are getting retried(not sure if retry logic is implemented)?

Buffered data should be handled in-order, but again it may be handled concurrently such that you'll have multiple requests in-flight at once.

If concurrency is the issue, you can limit it in the sink by setting request.concurrency = 1.