Add qdisc collector for Linux (#580)

* Add qdisc collector for Linux

This collector gathers basic queueing discipline metrics via netlink,
similarly to what `tc -s qdisc show` does.

* qdisc collector: nl-specific code moved, names fixed

- netlink-specific parts moved to github.com/ema/qdisc
- avoid using shortened names
- counters renamed into XXX_total

* Get rid of parseMessage error checking leftover

* Add github.com/ema/qdisc to vendored packages

* Update help texts and comments

* Add qdisc collector to README file

* qdisc collector end-to-end testing

* Update qdisc dependency to latest version

Update github.com/ema/qdisc dependency to revision 2c7e72d, which
includes unit testing.

* qdisc collector: rename "iface" label into "device"

README.md

@@ -62,6 +62,7 @@ logind | Exposes session counts from [logind](http://www.freedesktop.org/wiki/So
 meminfo\_numa | Exposes memory statistics from `/proc/meminfo_numa`. | Linux
 mountstats | Exposes filesystem statistics from `/proc/self/mountstats`. Exposes detailed NFS client statistics. | Linux
 nfs | Exposes NFS client statistics from `/proc/net/rpc/nfs`. This is the same information as `nfsstat -c`. | Linux
+qdisc | Exposes [queuing discipline](https://en.wikipedia.org/wiki/Network_scheduler#Linux_kernel) statistics | Linux
 runit | Exposes service status from [runit](http://smarden.org/runit/). | _any_
 supervisord | Exposes service status from [supervisord](http://supervisord.org/). | _any_
 systemd | Exposes service and system status from [systemd](http://www.freedesktop.org/wiki/Software/systemd/). | Linux

collector/fixtures/e2e-output.txt

@@ -2114,6 +2114,26 @@ node_procs_blocked 0
 # HELP node_procs_running Number of processes in runnable state.
 # TYPE node_procs_running gauge
 node_procs_running 2
+# HELP node_qdisc_bytes_total Number of bytes sent.
+# TYPE node_qdisc_bytes_total counter
+node_qdisc_bytes_total{device="eth0",kind="pfifo_fast"} 83
+node_qdisc_bytes_total{device="wlan0",kind="fq"} 42
+# HELP node_qdisc_drops_total Number of packets dropped.
+# TYPE node_qdisc_drops_total counter
+node_qdisc_drops_total{device="eth0",kind="pfifo_fast"} 0
+node_qdisc_drops_total{device="wlan0",kind="fq"} 1
+# HELP node_qdisc_overlimits_total Number of overlimit packets.
+# TYPE node_qdisc_overlimits_total counter
+node_qdisc_overlimits_total{device="eth0",kind="pfifo_fast"} 0
+node_qdisc_overlimits_total{device="wlan0",kind="fq"} 0
+# HELP node_qdisc_packets_total Number of packets sent.
+# TYPE node_qdisc_packets_total counter
+node_qdisc_packets_total{device="eth0",kind="pfifo_fast"} 83
+node_qdisc_packets_total{device="wlan0",kind="fq"} 42
+# HELP node_qdisc_requeues_total Number of packets dequeued, not transmitted, and requeued.
+# TYPE node_qdisc_requeues_total counter
+node_qdisc_requeues_total{device="eth0",kind="pfifo_fast"} 2
+node_qdisc_requeues_total{device="wlan0",kind="fq"} 1
 # HELP node_scrape_collector_duration_seconds node_exporter: Duration of a collector scrape.
 # TYPE node_scrape_collector_duration_seconds gauge
 # HELP node_scrape_collector_success node_exporter: Whether a collector succeeded.
@@ -2139,6 +2159,7 @@ node_scrape_collector_success{collector="mountstats"} 1
 node_scrape_collector_success{collector="netdev"} 1
 node_scrape_collector_success{collector="netstat"} 1
 node_scrape_collector_success{collector="nfs"} 1
+node_scrape_collector_success{collector="qdisc"} 1
 node_scrape_collector_success{collector="sockstat"} 1
 node_scrape_collector_success{collector="stat"} 1
 node_scrape_collector_success{collector="textfile"} 1

collector/fixtures/qdisc/results.json

@@ -0,0 +1,17 @@
+[
+    {
+        "IfaceName": "wlan0",
+        "Bytes": 42,
+        "Packets": 42,
+        "Requeues": 1,
+        "Kind": "fq",
+        "Drops": 1
+    },
+    {
+        "IfaceName": "eth0",
+        "Bytes": 83,
+        "Packets": 83,
+        "Requeues": 2,
+        "Kind": "pfifo_fast"
+    }
+]

collector/qdisc_linux.go

@@ -0,0 +1,116 @@
+// Copyright 2017 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build !noqdisc
+
+package collector
+
+import (
+	"encoding/json"
+	"flag"
+	"io/ioutil"
+	"path/filepath"
+
+	"github.com/ema/qdisc"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type qdiscStatCollector struct {
+	bytes      typedDesc
+	packets    typedDesc
+	drops      typedDesc
+	requeues   typedDesc
+	overlimits typedDesc
+}
+
+var (
+	collectorQdisc = flag.String("collector.qdisc", "", "test fixtures to use for qdisc collector end-to-end testing")
+)
+
+func init() {
+	Factories["qdisc"] = NewQdiscStatCollector
+}
+
+func NewQdiscStatCollector() (Collector, error) {
+	return &qdiscStatCollector{
+		bytes: typedDesc{prometheus.NewDesc(
+			prometheus.BuildFQName(Namespace, "qdisc", "bytes_total"),
+			"Number of bytes sent.",
+			[]string{"device", "kind"}, nil,
+		), prometheus.CounterValue},
+		packets: typedDesc{prometheus.NewDesc(
+			prometheus.BuildFQName(Namespace, "qdisc", "packets_total"),
+			"Number of packets sent.",
+			[]string{"device", "kind"}, nil,
+		), prometheus.CounterValue},
+		drops: typedDesc{prometheus.NewDesc(
+			prometheus.BuildFQName(Namespace, "qdisc", "drops_total"),
+			"Number of packets dropped.",
+			[]string{"device", "kind"}, nil,
+		), prometheus.CounterValue},
+		requeues: typedDesc{prometheus.NewDesc(
+			prometheus.BuildFQName(Namespace, "qdisc", "requeues_total"),
+			"Number of packets dequeued, not transmitted, and requeued.",
+			[]string{"device", "kind"}, nil,
+		), prometheus.CounterValue},
+		overlimits: typedDesc{prometheus.NewDesc(
+			prometheus.BuildFQName(Namespace, "qdisc", "overlimits_total"),
+			"Number of overlimit packets.",
+			[]string{"device", "kind"}, nil,
+		), prometheus.CounterValue},
+	}, nil
+}
+
+func testQdiscGet(fixtures string) ([]qdisc.QdiscInfo, error) {
+	var res []qdisc.QdiscInfo
+
+	b, err := ioutil.ReadFile(filepath.Join(fixtures, "results.json"))
+	if err != nil {
+		return res, err
+	}
+
+	err = json.Unmarshal(b, &res)
+	return res, err
+}
+
+func (c *qdiscStatCollector) Update(ch chan<- prometheus.Metric) error {
+	var msgs []qdisc.QdiscInfo
+	var err error
+
+	fixtures := *collectorQdisc
+
+	if fixtures == "" {
+		msgs, err = qdisc.Get()
+	} else {
+		msgs, err = testQdiscGet(fixtures)
+	}
+
+	if err != nil {
+		return err
+	}
+
+	for _, msg := range msgs {
+		// Only report root qdisc information.
+		if msg.Parent != 0 {
+			continue
+		}
+
+		ch <- c.bytes.mustNewConstMetric(float64(msg.Bytes), msg.IfaceName, msg.Kind)
+		ch <- c.packets.mustNewConstMetric(float64(msg.Packets), msg.IfaceName, msg.Kind)
+		ch <- c.drops.mustNewConstMetric(float64(msg.Drops), msg.IfaceName, msg.Kind)
+		ch <- c.requeues.mustNewConstMetric(float64(msg.Requeues), msg.IfaceName, msg.Kind)
+		ch <- c.overlimits.mustNewConstMetric(float64(msg.Overlimits), msg.IfaceName, msg.Kind)
+	}
+
+	return nil
+}

end-to-end-test.sh

@@ -22,6 +22,7 @@ collectors=$(cat << COLLECTORS
   netdev
   netstat
   nfs
+  qdisc
   sockstat
   stat
   textfile
@@ -76,6 +77,7 @@ fi
   -collector.textfile.directory="collector/fixtures/textfile/two_metric_files/" \
   -collector.megacli.command="collector/fixtures/megacli" \
   -collector.wifi="collector/fixtures/wifi" \
+  -collector.qdisc="collector/fixtures/qdisc/" \
   -web.listen-address "127.0.0.1:${port}" \
   -log.level="debug" > "${tmpdir}/node_exporter.log" 2>&1 &
 

vendor/github.com/ema/qdisc/LICENSE.md

@@ -0,0 +1,10 @@
+MIT License
+===========
+
+Copyright (C) 2017 Emanuele Rocca
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

vendor/github.com/ema/qdisc/Makefile

@@ -0,0 +1,11 @@
+build:
+	go fmt
+	go build
+	go vet
+	staticcheck
+	#golint -set_exit_status
+	go test -v -race -tags=integration
+
+cover:
+	go test -coverprofile=coverage.out
+	go tool cover -html=coverage.out

vendor/github.com/ema/qdisc/README.md

@@ -0,0 +1,26 @@
+qdisc [![Build Status](https://travis-ci.org/ema/qdisc.svg?branch=master)](https://travis-ci.org/ema/qdisc)
+=====
+
+Package `qdisc` allows to get queuing discipline information via netlink,
+similarly to what `tc -s qdisc show` does.
+
+Example usage
+-------------
+
+    package main
+
+    import (
+        "fmt"
+
+        "github.com/ema/qdisc"
+    )
+
+    func main() {
+        info, err := qdisc.Get()
+
+        if err == nil {
+            for _, msg := range info {
+                fmt.Printf("%+v\n", msg)
+            }
+        }
+    }

vendor/github.com/ema/qdisc/get.go

@@ -0,0 +1,230 @@
+package qdisc
+
+import (
+	"fmt"
+	"math"
+	"net"
+
+	"github.com/mdlayher/netlink"
+	"github.com/mdlayher/netlink/nlenc"
+)
+
+const (
+	TCA_UNSPEC = iota
+	TCA_KIND
+	TCA_OPTIONS
+	TCA_STATS
+	TCA_XSTATS
+	TCA_RATE
+	TCA_FCNT
+	TCA_STATS2
+	TCA_STAB
+	__TCA_MAX
+)
+
+const (
+	TCA_STATS_UNSPEC = iota
+	TCA_STATS_BASIC
+	TCA_STATS_RATE_EST
+	TCA_STATS_QUEUE
+	TCA_STATS_APP
+	TCA_STATS_RATE_EST64
+	__TCA_STATS_MAX
+)
+
+// See struct tc_stats in /usr/include/linux/pkt_sched.h
+type TC_Stats struct {
+	Bytes      uint64
+	Packets    uint32
+	Drops      uint32
+	Overlimits uint32
+	Bps        uint32
+	Pps        uint32
+	Qlen       uint32
+	Backlog    uint32
+}
+
+// See /usr/include/linux/gen_stats.h
+type TC_Stats2 struct {
+	// struct gnet_stats_basic
+	Bytes   uint64
+	Packets uint32
+	// struct gnet_stats_queue
+	Qlen       uint32
+	Backlog    uint32
+	Drops      uint32
+	Requeues   uint32
+	Overlimits uint32
+}
+
+type QdiscInfo struct {
+	IfaceName  string
+	Parent     uint32
+	Handle     uint32
+	Kind       string
+	Bytes      uint64
+	Packets    uint32
+	Drops      uint32
+	Requeues   uint32
+	Overlimits uint32
+}
+
+func parseTCAStats(attr netlink.Attribute) TC_Stats {
+	var stats TC_Stats
+	stats.Bytes = nlenc.Uint64(attr.Data[0:8])
+	stats.Packets = nlenc.Uint32(attr.Data[8:12])
+	stats.Drops = nlenc.Uint32(attr.Data[12:16])
+	stats.Overlimits = nlenc.Uint32(attr.Data[16:20])
+	stats.Bps = nlenc.Uint32(attr.Data[20:24])
+	stats.Pps = nlenc.Uint32(attr.Data[24:28])
+	stats.Qlen = nlenc.Uint32(attr.Data[28:32])
+	stats.Backlog = nlenc.Uint32(attr.Data[32:36])
+	return stats
+}
+
+func parseTCAStats2(attr netlink.Attribute) TC_Stats2 {
+	var stats TC_Stats2
+
+	nested, _ := netlink.UnmarshalAttributes(attr.Data)
+
+	for _, a := range nested {
+		switch a.Type {
+		case TCA_STATS_BASIC:
+			stats.Bytes = nlenc.Uint64(a.Data[0:8])
+			stats.Packets = nlenc.Uint32(a.Data[8:12])
+		case TCA_STATS_QUEUE:
+			stats.Qlen = nlenc.Uint32(a.Data[0:4])
+			stats.Backlog = nlenc.Uint32(a.Data[4:8])
+			stats.Drops = nlenc.Uint32(a.Data[8:12])
+			stats.Requeues = nlenc.Uint32(a.Data[12:16])
+			stats.Overlimits = nlenc.Uint32(a.Data[16:20])
+		default:
+			// TODO: TCA_STATS_APP
+		}
+	}
+
+	return stats
+}
+
+func getQdiscMsgs(c *netlink.Conn) ([]netlink.Message, error) {
+	req := netlink.Message{
+		Header: netlink.Header{
+			Flags: netlink.HeaderFlagsRequest | netlink.HeaderFlagsDump,
+			Type:  38, // RTM_GETQDISC
+		},
+		Data: []byte{0},
+	}
+
+	// Perform a request, receive replies, and validate the replies
+	msgs, err := c.Execute(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute request: %v", err)
+	}
+
+	return msgs, nil
+}
+
+// See https://tools.ietf.org/html/rfc3549#section-3.1.3
+func parseMessage(msg netlink.Message) (QdiscInfo, error) {
+	var m QdiscInfo
+	var s TC_Stats
+	var s2 TC_Stats2
+
+	/*
+	   struct tcmsg {
+	       unsigned char   tcm_family;
+	       unsigned char   tcm__pad1;
+	       unsigned short  tcm__pad2;
+	       int     tcm_ifindex;
+	       __u32       tcm_handle;
+	       __u32       tcm_parent;
+	       __u32       tcm_info;
+	   };
+	*/
+
+	if len(msg.Data) < 20 {
+		return m, fmt.Errorf("Short message, len=%d", len(msg.Data))
+	}
+
+	ifaceIdx := nlenc.Uint32(msg.Data[4:8])
+
+	m.Handle = nlenc.Uint32(msg.Data[8:12])
+	m.Parent = nlenc.Uint32(msg.Data[12:16])
+
+	if m.Parent == math.MaxUint32 {
+		m.Parent = 0
+	}
+
+	// The first 20 bytes are taken by tcmsg
+	attrs, err := netlink.UnmarshalAttributes(msg.Data[20:])
+
+	if err != nil {
+		return m, fmt.Errorf("failed to unmarshal attributes: %v", err)
+	}
+
+	for _, attr := range attrs {
+		switch attr.Type {
+		case TCA_KIND:
+			m.Kind = nlenc.String(attr.Data)
+		case TCA_STATS2:
+			s2 = parseTCAStats2(attr)
+			m.Bytes = s2.Bytes
+			m.Packets = s2.Packets
+			m.Drops = s2.Drops
+			// requeues only available in TCA_STATS2, not in TCA_STATS
+			m.Requeues = s2.Requeues
+			m.Overlimits = s2.Overlimits
+		case TCA_STATS:
+			// Legacy
+			s = parseTCAStats(attr)
+			m.Bytes = s.Bytes
+			m.Packets = s.Packets
+			m.Drops = s.Drops
+			m.Overlimits = s.Overlimits
+		default:
+			// TODO: TCA_OPTIONS and TCA_XSTATS
+		}
+	}
+
+	iface, err := net.InterfaceByIndex(int(ifaceIdx))
+
+	if err == nil {
+		m.IfaceName = iface.Name
+	}
+
+	return m, err
+}
+
+func getAndParse(c *netlink.Conn) ([]QdiscInfo, error) {
+	var res []QdiscInfo
+
+	msgs, err := getQdiscMsgs(c)
+
+	if err != nil {
+		return nil, err
+	}
+
+	for _, msg := range msgs {
+		m, err := parseMessage(msg)
+
+		if err != nil {
+			return nil, err
+		}
+
+		res = append(res, m)
+	}
+
+	return res, nil
+}
+
+func Get() ([]QdiscInfo, error) {
+	const familyRoute = 0
+
+	c, err := netlink.Dial(familyRoute, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to dial netlink: %v", err)
+	}
+	defer c.Close()
+
+	return getAndParse(c)
+}

vendor/vendor.json

@@ -26,6 +26,12 @@
 			"revision": "e97b35f834b17eaa82afe3d44715c34736bfa12b",
 			"revisionTime": "2017-02-01T10:47:36Z"
 		},
+		{
+			"checksumSHA1": "5tGizHomeMFHcyROMD2ntrKFR1Q=",
+			"path": "github.com/ema/qdisc",
+			"revision": "2c7e72d2f110bdddf17026872f6c396a9e1e4809",
+			"revisionTime": "2017-05-16T10:52:09Z"
+		},
 		{
 			"checksumSHA1": "Yim1r9qkEFFLLH8JngB4kZTLsYI=",
 			"path": "github.com/godbus/dbus",