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.