LibreNMS2Kafka2Prometheus

This project is to describe how to transport data exported by LibreNMS into Prometheus through a Apche Kafka Broker.

To achieve this, we use the data pipeline software, Vector.

Disclaimer

For now, this is a Proof Of Concept (no resiliency is provied). Especially, the completeness of all the metric types isn't guaranteed.

Schema to illustrate

Settings : I use 3 servers (VMs running Debian Linux 13) :

• One server for LibreNMS
• One server for Kafka and Vector
• One server for Prometheus (and Grafana not represented)

---
config:
  theme: 'neutral'
---
flowchart TD
    
    L(LibreNMS) e1@== Export Data with Kafka Exporter ==> K@{ shape: das, label: "Kafka Topic" }
    e1@{ animate: true }
    K e2@== Getting Data ==> V1(Vector Kafka Sink)
    e2@{ animate: true }
    V1 e3@== Initiate Connection ==> K(Vector Kafka Sink)
    e3@{ animate: true }
    subgraph Vector        
        V1 e4@--> V2(Vector Transforms)
        e4@{ animate: true }
        V2 e5@--> V3(Vector Prometheus Exporter Sink)
        e5@{ animate: true }        
    end
    V3 e6@== Scraped Data ==> P1(Prometheus Scraper)
    e6@{ animate : true }
    P1 e7@== Initiate Scraping ==> V3
    e7@{ animate : true }
    subgraph Prometheus
        P1 e8@--> P2(Prometheus Database)
        e8@{ animate : true }
    end

Loading

So LibreNMS is pushing data into a Kafka Topic.

Vector Kafka Source is pulling data from this topic and passes all these to a bunch of transform and routes components of Vector to handle all the type of messages/events/metrics LibreNMS produces. These transforms also handle the modification/correction so thaht metrics can be read by Prometheus Scraping (Upercase metrics name or name with hyphen - are forbidden). All these metrics are concentrated into a single prometheus exporter sink.

Then Prometheus scraping process will then get these metrics and store them.

Technical implementations and configurations

Basic LibreNMS configuration for Kafka

This is a basic configuration of LibreNMS to export Data into kafka topic

{
    "enable": true,
    "debug": false,
    "security": {
        "debug": ""
    },
    "broker": {
        "list": "<YOUR BROKER HERE>"
    },
    "idempotence": false,
    "topic": "<YOUR TOPIC HERE>",
    "ssl": {
        "enable": false,
        "protocol": null,
        "ca": {
            "location": ""
        },
        "certificate": {
            "location": ""
        },
        "key": {
            "location": "",
            "password": ""
        },
        "keystore": {
            "location": "",
            "password": ""
        }
    },
    "flush": {
        "timeout": 1
    },
    "buffer": {
        "max": {
            "message": 100
        }
    },
    "batch": {
        "max": {
            "message": 10
        }
    },
    "linger": {
        "ms": 100
    },
    "request": {
        "required": {
            "acks": "0"
        }
    }
}

Vector Configuration

Complete configuration of Vector :

[

LibreNMS2Kafka2Prometheus/vector-Kafka2Prometheus4LibreNMS.yaml

Lines 1 to 11 in 665d45d

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

]

Here is a representation of the Vector configuration :

---
config:
  theme: 'neutral'
---
flowchart TD
    %% --- Styles ---
    classDef source fill:#e6f7ff,stroke:#4682b4,stroke-width:2px;
    classDef remap fill:#fff0e6,stroke:#ff8c00,stroke-width:2px;
    classDef route fill:#f0fff0,stroke:#2e8b57,stroke-width:2px;
    classDef log_to_metric fill:#ffe6e6,stroke:#ff1493,stroke-width:2px;
    classDef sink fill:#f0f8ff,stroke:#1e90ff,stroke-width:2px;

    %% --- Sources ---
    kafka_metrics_id["📡 kafka_metrics_id\n(type: kafka)"]:::source
    vector_metrics_id["📊 vector_metrics_id\n(type: internal_metrics)"]:::source

    %% --- Transforms: remap ---
    my_parse_metrics_id["🔧 my_parse_metrics_id\n(type: remap)"]:::remap
    my_rewrite_measurement_type_id["🔧 my_rewrite_measurement_type_id\n(type: remap)"]:::remap
    my_metrics_non_null_ports_id["🔧 my_metrics_non_null_ports_id\n(type: remap)"]:::remap
    my_metrics_downcase_poller_perf_id["🔧 my_metrics_downcase_poller_perf_id\n(type: remap)"]:::remap

    %% --- Transforms: route ---
    my_route_metrics_id["🛣️ my_route_metrics_id\n(type: route)"]:::route
    my_ucd_cpu_route_id["🛣️ my_ucd_cpu_route_id\n(type: route)"]:::route

    %% --- Transforms: log_to_metric ---
    my_metrics_ports_id["📈 my_metrics_ports_id\n(type: log_to_metric)"]:::log_to_metric
    my_metrics_poller_perf_id["📈 my_metrics_poller_perf_id\n(type: log_to_metric)"]:::log_to_metric
    my_icmp_metrics_id["📈 my_icmp_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_uptime_metrics_id["📈 my_uptime_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_availability_metrics_id["📈 my_availability_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_mempool_metrics_id["📈 my_mempool_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_storage_metrics_id["📈 my_storage_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_netstats_icmp_metrics_id["📈 my_netstats_icmp_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_netstats_ip_metrics_id["📈 my_netstats_ip_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_netstats_tcp_metrics_id["📈 my_netstats_tcp_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_netstats_udp_metrics_id["📈 my_netstats_udp_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_netstats_snmp_metrics_id["📈 my_netstats_snmp_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_processors_metrics_id["📈 my_processors_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_hr_processes_metrics_id["📈 my_hr_processes_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_hr_users_metrics_id["📈 my_hr_users_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_ipSystemStats_metrics_id["📈 my_ipSystemStats_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_ucd_diskio_metrics_id["📈 my_ucd_diskio_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_ucd_load_metrics_id["📈 my_ucd_load_metrics_id\n(type: log_to_metric)"]:::log_to_metric
    my_ucd_cpu_oid_id["📈 my_ucd_cpu_oid_id\n(type: log_to_metric)"]:::log_to_metric
    my_ucd_cpu_non_oid_id["📈 my_ucd_cpu_non_oid_id\n(type: log_to_metric)"]:::log_to_metric

    %% --- Sinks ---
    my_sink_id["💾 my_sink_id\n(type: prometheus_exporter)"]:::sink
    my_vector_prometheus_sink_id["💾 my_vector_prometheus_sink_id\n(type: prometheus_exporter)"]:::sink
    my_sink_metric_ports_file_id["📄 my_sink_metric_ports_file_id\n(type: file)"]:::sink
    my_sink_metric_ports_non_null_file_id["📄 my_sink_metric_ports_non_null_file_id\n(type: file)"]:::sink
    my_sink_metric_poller-perf_file_id["📄 my_sink_metric_poller-perf_file_id\n(type: file)"]:::sink
    my_sink_unmatched_file_id["📄 my_sink_unmatched_file_id\n(type: file)"]:::sink
    my_metric_vector_sink_console_id["📄 my_metric_vector_sink_console_id\n(type: file)"]:::sink

    %% --- Connexions ---
    %% Sources -> Transforms
    kafka_metrics_id --> my_parse_metrics_id
    my_parse_metrics_id --> my_rewrite_measurement_type_id
    my_rewrite_measurement_type_id --> my_route_metrics_id
    vector_metrics_id --> my_vector_prometheus_sink_id

    %% Route -> remap/log_to_metric
    my_route_metrics_id -.->|ports| my_metrics_non_null_ports_id
    my_route_metrics_id -.->|poller_perf| my_metrics_downcase_poller_perf_id
    my_route_metrics_id -.->|icmp_route| my_icmp_metrics_id
    my_route_metrics_id -.->|uptime_route| my_uptime_metrics_id
    my_route_metrics_id -.->|availability_route| my_availability_metrics_id
    my_route_metrics_id -.->|mempool_route| my_mempool_metrics_id
    my_route_metrics_id -.->|storage_route| my_storage_metrics_id
    my_route_metrics_id -.->|netstats_icmp_route| my_netstats_icmp_metrics_id
    my_route_metrics_id -.->|netstats_ip_route| my_netstats_ip_metrics_id
    my_route_metrics_id -.->|netstats_tcp_route| my_netstats_tcp_metrics_id
    my_route_metrics_id -.->|netstats_udp_route| my_netstats_udp_metrics_id
    my_route_metrics_id -.->|netstats_snmp_route| my_netstats_snmp_metrics_id
    my_route_metrics_id -.->|processors_route| my_processors_metrics_id
    my_route_metrics_id -.->|hr_processes_route| my_hr_processes_metrics_id
    my_route_metrics_id -.->|hr_users_route| my_hr_users_metrics_id
    my_route_metrics_id -.->|ipSystemStats_route| my_ipSystemStats_metrics_id
    my_route_metrics_id -.->|ucd_diskio_route| my_ucd_diskio_metrics_id
    my_route_metrics_id -.->|ucd_load_route| my_ucd_load_metrics_id
    my_route_metrics_id -.->|cpu_route| my_ucd_cpu_route_id

    %% remap -> log_to_metric
    my_metrics_non_null_ports_id --> my_metrics_ports_id
    my_metrics_downcase_poller_perf_id --> my_metrics_poller_perf_id

    %% ucd_cpu_route -> log_to_metric
    my_ucd_cpu_route_id -.->|ucd_route_oid_route| my_ucd_cpu_oid_id
    my_ucd_cpu_route_id -.->|ucd_route_non_oid_route| my_ucd_cpu_non_oid_id

    %% log_to_metric -> sinks
    my_metrics_ports_id --> my_sink_id
    my_metrics_poller_perf_id --> my_sink_id
    my_icmp_metrics_id --> my_sink_id
    my_uptime_metrics_id --> my_sink_id
    my_availability_metrics_id --> my_sink_id
    my_mempool_metrics_id --> my_sink_id
    my_storage_metrics_id --> my_sink_id
    my_netstats_icmp_metrics_id --> my_sink_id
    my_netstats_ip_metrics_id --> my_sink_id
    my_netstats_tcp_metrics_id --> my_sink_id
    my_netstats_udp_metrics_id --> my_sink_id
    my_netstats_snmp_metrics_id --> my_sink_id
    my_processors_metrics_id --> my_sink_id
    my_hr_processes_metrics_id --> my_sink_id
    my_hr_users_metrics_id --> my_sink_id
    my_ipSystemStats_metrics_id --> my_sink_id
    my_ucd_diskio_metrics_id --> my_sink_id
    my_ucd_load_metrics_id --> my_sink_id
    my_ucd_cpu_oid_id --> my_sink_id
    my_ucd_cpu_non_oid_id --> my_sink_id

    %% Sinks de debug
    my_route_metrics_id -.->|ports| my_sink_metric_ports_file_id
    my_metrics_non_null_ports_id --> my_sink_metric_ports_non_null_file_id
    my_metrics_downcase_poller_perf_id --> my_sink_metric_poller-perf_file_id
    my_route_metrics_id -.->|_unmatched| my_sink_unmatched_file_id
    my_metrics_non_null_ports_id -.->|dropped| my_metric_vector_sink_console_id

Loading

Prometheus configuration

[

LibreNMS2Kafka2Prometheus/prometheus.yaml

Lines 1 to 9 in 5c93b4c

	# my global config
	global:
	scrape_interval: 120s # Set the scrape interval to every 15 seconds. Default is every 1 minute.
	# evaluation_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute.
	evaluation_interval: 240s # Evaluate rules every 15 seconds. The default is every 1 minute.
	# scrape_timeout is set to the global default (10s).
	# Alertmanager configuration
	alerting:

]