Q&A : kafka-transported Metrics into Prometheus exporter

[htam-net]

Question

Hello

I would like to be able to pull metrics from a kafka topic source and expose them with prometheus exporter sink.

LibreNMS -> Kafka -> Vector -> Prometheus Exporter

With the supplied configuration (and VRL program), I am able to get and transform event into table of metrics. But As vector top cli command shows, all events are outputted from the remap with VRL transform but none gets in the prometheus exporter sink.

I added the reroute_dropped parameter (with the correspondingfile sink) but no message is output in this dropped sink.

I haven't found the expected format that is expected in input for the prometheus exporter sink.

NB : I was able to expose internal vector metrics with an other prometheus exporter sink (no problem)

Here a sample message from the kafka topic (in 1 line) :

{
    "measurement": "ports",
    "fields": {
        "INOCTETS": "38525827765",
        "OUTOCTETS": "138659900139",
        "INERRORS": "0",
        "OUTERRORS": "0",
        "INUCASTPKTS": "55716873",
        "OUTUCASTPKTS": "84610249",
        "INNUCASTPKTS": null,
        "OUTNUCASTPKTS": null,
        "INDISCARDS": "0",
        "OUTDISCARDS": "0",
        "INUNKNOWNPROTOS": null,
        "INBROADCASTPKTS": "1308652",
        "OUTBROADCASTPKTS": "11585320",
        "INMULTICASTPKTS": "2883812",
        "OUTMULTICASTPKTS": "230907583",
        "ifInUcastPkts_rate": 4,
        "ifOutUcastPkts_rate": 3,
        "ifInErrors_rate": 0,
        "ifOutErrors_rate": 0,
        "ifInOctets_rate": 1419,
        "ifOutOctets_rate": 16413,
        "ifInBits_rate": 7233,
        "ifOutBits_rate": 130241
    },
    "tags": {
        "ifName": "tw1",
        "ifAlias": "portTW1",
        "ifIndex": 1,
        "current_polled_time": 1781853005,
        "hostname": "1.2.3.4"
    }

Vector Config

data_dir: /tmp/vector
api: 
  enabled: true
  address: "0.0.0.0:8686"
schema:
  # enabled: true
  validation: true

sources:
  kafka_metrics_id:
    type: kafka
    bootstrap_servers: localhost:9092
    group_id: group-de-vector
    topics:
      - mylibrenms

transforms:
  my_parse_metrics_id:
    type: remap
    reroute_dropped: true
    inputs:
      - kafka_metrics_id
    file: ./vector-librenms2Kafka2Prometheus-transform.vrl
      
sinks:
  my_sink_id:
    type: prometheus_exporter
    inputs:
      - my_parse_metrics_id
    address: "0.0.0.0:9598"
  
dropped:
    type: file
    inputs:
      - my_parse_metrics_id.dropped
    encoding:
      codec: text
    path: /tmp/vector/dropped-%Y-%m-%d.log

my VRL transform program : ./vector-librenms2Kafka2Prometheus-transform.vrl

message = parse_json!(.message)
tstamp = .timestamp
topic = .topic

# 2. Extracting tags to reuse them for each metric
labels = {}
if exists(message.tags) {
    for_each(object!(message.tags)) -> |key, val| {
        label_obj = {}
        if is_null(val) {
            labels=set!(labels, path: [key], data: "null")
        } else {         
            labels=set!(labels, path: [key], data: val)
        }
    }
    # adding the kafka topic as label for further tracability
    labels=set!(labels, ["topic"], topic)
}

# defining the namespace from measurement
namespace = message.measurement

# Initialize final table of metrics
metrics = []
if exists(message.fields) {
    # parse all fields in the log event that must be transformed into metrics
    for_each(object!(message.fields)) -> |field_name, field_value| {
        
        # On ignore les valeurs nulles pour ne pas polluer Prometheus
        if !is_null(field_value) {
        
            # check for gauge or counter value type
            type_kind = "gauge"
            kind = "absolute"
            if ends_with(field_name, "OCTETS") || ends_with(field_name, "PKTS") || ends_with(field_name, "ERRORS") || ends_with(field_name, "DISCARDS") {
                type_kind = "counter"
                kind = "incremental"
            }
        
            # Conversion in float
            parsed_value = to_float(field_value) ?? 0.0
            
            # Construction of the metric object in Vector format
            metric_obj = {
                "name": field_name,
                "namespace" : namespace,
                "kind": kind,
                "tags": labels,
                "timestamps" : tstamp
            }
        
            # adding the value depending on type
            if type_kind == "counter" {
                metric_obj.counter = { "value": parsed_value }
            } else {
                metric_obj.gauge = { "value": parsed_value }
            }
            
            # Adding a "metric" header
            # from example : https://vector.dev/docs/reference/configuration/transforms/log_to_metric/#examples-counter
            metric_master_obj = {
                "metric": metric_obj
            }

            # Ajout au tableau global
            metrics = push(metrics, metric_master_obj)
        }
    }
}

# 4. Final replacement to output
. = metrics

.

### Vector Logs

Vector Logs display no error (event in trace mode).

[daveshenal]

Worth trying... restructure your pipeline to use log_to_metric instead of building metric objects in VRL. The prometheus_exporter sink expects true metric events, and VRL's remap transform may not be producing those even when the shape looks correct.

transforms:
  my_parse_metrics_id:
    type: remap
    inputs:
      - kafka_metrics_id
    source: |
      . = parse_json!(.message)

  my_metrics_id:
    type: log_to_metric
    inputs:
      - my_parse_metrics_id
    metrics:
      - type: gauge
        field: "fields.ifInOctets_rate"
        name: "ifInOctets_rate"
        namespace: "{{measurement}}"
        tags:
          ifName: "{{tags.ifName}}"
          hostname: "{{tags.hostname}}"
      - type: counter
        field: "fields.INOCTETS"
        name: "INOCTETS"
        namespace: "{{measurement}}"
        tags:
          ifName: "{{tags.ifName}}"
          hostname: "{{tags.hostname}}"
      # repeat for each field...

sinks:
  my_sink_id:
    type: prometheus_exporter
    inputs:
      - my_metrics_id
    address: "0.0.0.0:9598"

[htam-net]

Thank you very much... This is really helpful.
Also, I appreciate that my writing is supposed to be correct :-)

I have one last hiccup : some value are null (and not 0) and this seems to trouble
If I read correctly, getting a fallback value isn't implemented yet and I have to get back to remap transform with VRL... and that's where I get lost again.

(Also to further help future people interested in this architecture, I have introduce a lot of route transform to catch all the type of fields, value, etc...). I will post the result when completed.

[htam-net]

To continue, I added a remap transform in front of the problematic log_to_metric transform.
It is working in commandline vector vrl (null values are replaced with 0`) ... but not in a pipeline

Here is the VRL file :

map_values(object!(.fields)) -> |metrique_value| {
    if is_null(metrique_value) {
        metrique_value = 0
        # log("conversion null -> 0" , level: "trace", rate_limit_secs: 0)
    } else {
        metrique_value = metrique_value
    }
    metrique_value
}

Before this VRL file :

{"fields":{"INBROADCASTPKTS":"6363","INDISCARDS":"0","INERRORS":"0","INMULTICASTPKTS":"0","INNUCASTPKTS":null,"INOCTETS":"26911204","INUCASTPKTS":"383791","INUNKNOWNPROTOS":null,"OUTBROADCASTPKTS":"11549122","OUTDISCARDS":"0","OUTERRORS":"0","OUTMULTICASTPKTS":"139165922","OUTNUCASTPKTS":null,"OUTOCTETS":"27906147393","OUTUCASTPKTS":"374073","ifInBits_rate":41,"ifInErrors_rate":0,"ifInOctets_rate":5,"ifInUcastPkts_rate":0,"ifOutBits_rate":67014,"ifOutErrors_rate":0,"ifOutOctets_rate":8180,"ifOutUcastPkts_rate":0},"measurement":"ports","tags":{"current_polled_time":1782478205,"hostname":"192.168.1.4","ifAlias":"alarmeGrange","ifIndex":8,"ifName":"tw8"}}

After (in commandline) :

{ "INBROADCASTPKTS": "6363", "INDISCARDS": "0", "INERRORS": "0", "INMULTICASTPKTS": "0", "INNUCASTPKTS": 0, "INOCTETS": "26911204", "INUCASTPKTS": "383791", "INUNKNOWNPROTOS": 0, "OUTBROADCASTPKTS": "11549122", "OUTDISCARDS": "0", "OUTERRORS": "0", "OUTMULTICASTPKTS": "139165922", "OUTNUCASTPKTS": 0, "OUTOCTETS": "27906147393", "OUTUCASTPKTS": "374073", "ifInBits_rate": 41, "ifInErrors_rate": 0, "ifInOctets_rate": 5, "ifInUcastPkts_rate": 0, "ifOutBits_rate": 67014, "ifOutErrors_rate": 0, "ifOutOctets_rate": 8180, "ifOutUcastPkts_rate": 0 }

[htam-net]

Here is the right VRL file :

.fields = map_values(object!(.fields)) -> |metrique_value| {
    if is_null(metrique_value) {
        metrique_value = 0
        # log("conversion de null en 0" , level: "trace", rate_limit_secs: 0)
    } else {
        metrique_value = metrique_value
    }
    metrique_value
}

[htam-net]

If any one needs it, in the case of LibreNMS, I have written this configuration examples : https://github.com/htam-net/LibreNMS2Kafka2Prometheus