first commit

README.md

@@ -0,0 +1,86 @@
+# Nfdump Exporter
+
+This is a prototype exporter for nfdump. It exposes metrics processed by the Proetheus monitoring system.
+
+It's purpose is to to play and experiment with nfdump netflow data and Promtheus/Grafana to build a new graphical UI as a repacement for aging NfSen.
+
+This experimental exporter exposes counters for flows/packets and bytes per protocol (tcp/udp/icmp/other) and the source identifier from the nfcapd collector. (currently hardwired "live")
+
+## Metrics:
+
+```
+  namespace = "nfsen"
+	uptime = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, "collector", "uptime"),
+		"nfsen uptime.",
+		[]string{"version"}, nil,
+	)
+	flowsReceived = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, "collector", "flows"),
+		"How many flows have been received (per ident and protocol (tcp/udp/icmp/other)).",
+		[]string{"ident", "proto"}, nil,
+	)
+	packetsReceived = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, "collector", "packets"),
+		"How many packets have been received (per ident and protocol) (tcp/udp/icmp/other).",
+		[]string{"ident", "proto"}, nil,
+	)
+	bytesReceived = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, "collector", "bytes"),
+		"How many bytes have been received (per ident and protocol) (tcp/udp/icmp/other).",
+		[]string{"ident", "proto"}, nil,
+	)
+```
+
+
+
+## Usage:
+
+```
+Usage of ./nfsen_exporter:
+  -UNIX socket string
+    	Path for nfcapd collectors to connect (default "/tmp/nfsen.sock")
+  -listen string
+    	Address to listen on for telemetry (default ":9141")
+  -metrics URI string
+    	Path under which to expose metrics (default "/metrics")
+
+```
+
+The nfsen_exporter listens on a UNIX socket for statistics sent by the nfcapd collector. 
+
+Add this to prometheus.yml:
+
+```
+  - job_name: "nfsen"
+
+    # metrics_path defaults to '/metrics'
+    # scheme defaults to 'http'.
+
+    static_configs:
+      - targets: ["localhost:9141"]
+```
+
+
+
+## Nfdump
+
+The metric export is integrated in nfdump 1.7-beta
+
+In order not to pollute an existing nfdump netflow installation, forward the traffic from an existing collector. Add: `-R 127.0.0.1/9999` to the argument list and setup the new collector. You may also send it to another host, which runs also Prometheus for example. 
+
+Build nfdump 1.7-beta:
+
+`git clone -b unicorn https://github.com/phaag/nfdump.git nfdump.unicorn` 
+
+Build nfdump with `sh bootstrap.sh; ./configure` but do not run make install, as it would replace your existing installation. Create a tmp flow dir and run the collector from the src directory. For example:
+
+`./nfcapd -l <tmpflows> -S2 -y -p 9999 -m <metric socket>`  
+
+When adding `-m <metric socket>` nfcapd exports the internal statistics every 5s the the exporter. 
+
+
+
+## Note:
+
+Only the statistics is exposed and not the netflow recods itself.

dataSocket.go

@@ -0,0 +1,197 @@
+/*  
+ *  Copyright (c) 2021, Peter Haag
+ *  All rights reserved.
+ *  
+ *  Redistribution and use in source and binary forms, with or without 
+ *  modification, are permitted provided that the following conditions are met:
+ *  
+ *   * Redistributions of source code must retain the above copyright notice, 
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright notice, 
+ *     this list of conditions and the following disclaimer in the documentation 
+ *     and/or other materials provided with the distribution.
+ *   * Neither the name of the author nor the names of its contributors may be 
+ *     used to endorse or promote products derived from this software without 
+ *     specific prior written permission.
+ *  
+ *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
+ *  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
+ *  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
+ *  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
+ *  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
+ *  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
+ *  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
+ *  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
+ *  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 
+ *  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
+ *  POSSIBILITY OF SUCH DAMAGE.
+ *  
+ */
+
+/*
+ * dataSocket implements a UNIX socket server to receive data from nfcapd
+ * Up to now the exporter implements flows/packets/bytes counters per
+ * protocol(tcp/udp/icmp/other and the source identifier from the collector
+ *
+ */
+
+package main
+
+/*
+
+#include <stdint.h>
+
+typedef struct metric_record_s {
+	// Ident
+	char		ident[128];
+	//  uptime
+	uint64_t	uptime;
+    // flow stat
+    uint64_t    numflows_tcp;
+    uint64_t    numflows_udp;
+    uint64_t    numflows_icmp;
+    uint64_t    numflows_other;
+    // bytes stat
+    uint64_t    numbytes_tcp;
+    uint64_t    numbytes_udp;
+    uint64_t    numbytes_icmp;
+    uint64_t    numbytes_other;
+    // packet stat
+    uint64_t    numpackets_tcp;
+    uint64_t    numpackets_udp;
+    uint64_t    numpackets_icmp;
+    uint64_t    numpackets_other;
+} metric_record_t;
+*/
+import "C"
+
+import (
+	// "encoding/binary"
+	"fmt"
+	"os"
+	"log"
+	"net"
+	"unsafe"
+)
+
+const packetPrefix byte = '@'
+
+type nfsenMetric struct {
+	// Ident
+	ident string
+	//  uptime
+	uptime uint64
+    // flow stat
+    numFlows_tcp uint64
+    numFlows_udp uint64
+    numFlows_icmp uint64
+    numFlows_other uint64
+    // bytes stat
+    numBytes_tcp uint64
+    numBytes_udp uint64
+    numBytes_icmp uint64
+    numBytes_other uint64
+    // packet stat
+    numPackets_tcp uint64
+    numPackets_udp uint64
+    numPackets_icmp uint64
+    numPackets_other uint64
+}
+
+var metric nfsenMetric
+
+type socketConf struct {
+	socketPath	string
+	listener	net.Listener
+}
+
+func New(socketPath string) *socketConf {
+    conf := new(socketConf)
+    conf.socketPath = socketPath
+    return conf
+}
+
+func (socket *socketConf) Open() error {
+
+	if err := os.RemoveAll(socket.socketPath); err != nil {
+		return err
+	}
+	listener, err := net.Listen("unix", socket.socketPath)
+    if err != nil {
+        return err
+    }
+	socket.listener = listener
+	return nil
+
+} // End of Open
+
+func (socket *socketConf) Close() error {
+
+	return socket.listener.Close()
+
+} // End of Close
+
+func processStat(conn net.Conn) {
+
+    defer conn.Close()
+
+	// storage for reading from socket.
+    readBuf := make([]byte, 10240)
+
+	dataLen, err := conn.Read(readBuf)
+	if err != nil || dataLen == 0{
+		fmt.Printf("Socket read error: %v\n", err)
+		return
+	}
+	if readBuf[0] != packetPrefix {
+		fmt.Printf("Message prefix error - got %u\n", readBuf[0])
+		return
+	}
+
+	/*
+	version := readBuf[1]
+	payloadSize := int(binary.LittleEndian.Uint16(readBuf[2:4]))
+
+	fmt.Printf("Message size: %d, payload size: %d version: %d\n",
+		dataLen, payloadSize, version);
+	*/
+
+	var s *C.metric_record_t = (*C.metric_record_t)(unsafe.Pointer(&readBuf[4]))
+	mutex.Lock()
+	metric.ident = C.GoString(&s.ident[0])
+	metric.uptime = uint64(s.uptime)
+	metric.numFlows_tcp   = uint64(s.numflows_tcp)
+	metric.numFlows_udp   = uint64(s.numflows_udp)
+	metric.numFlows_icmp  = uint64(s.numflows_icmp)
+	metric.numFlows_other = uint64(s.numflows_other)
+
+	metric.numBytes_tcp   = uint64(s.numbytes_tcp)
+	metric.numBytes_udp   = uint64(s.numbytes_udp)
+	metric.numBytes_icmp  = uint64(s.numbytes_icmp)
+	metric.numBytes_other = uint64(s.numbytes_other)
+
+	metric.numPackets_tcp   = uint64(s.numpackets_tcp)
+	metric.numPackets_udp   = uint64(s.numpackets_udp)
+	metric.numPackets_icmp  = uint64(s.numpackets_icmp)
+	metric.numPackets_other = uint64(s.numpackets_other)
+	mutex.Unlock()
+
+
+} // end of processStat
+
+func (socket *socketConf) Run() {
+
+	go func() {
+		for {
+			// Accept new connections from nfcapd collectors and
+			// dispatching them to goroutine processStat
+			conn, err := socket.listener.Accept()
+			if err != nil {
+				log.Fatal("accept error:", err)
+			}
+			// fmt.Printf("New connection\n")
+			go processStat(conn)
+		}
+	}()
+
+} // End of Run

go.mod

@@ -0,0 +1,5 @@
+module nfsen_exporter
+
+go 1.14
+
+require github.com/prometheus/client_golang v1.11.0