Merge pull request #15 from williamchanrico/major_refactor_inventory

Add support for longest-prefix CIDR address inventory match

Author: williamchanrico

Makefile

@@ -3,7 +3,7 @@
 NAME          := planet-exporter
 BIN_DIRECTORY := ./bin
 REVISION      := $(shell git rev-parse --short HEAD 2>/dev/null)
-VERSION       := v0.2.0-dev
+VERSION       := v0.3.0-dev
 
 ifndef REVISION
 	override REVISION = none
@@ -66,6 +66,10 @@ go-build-linux: ## Build the app for linux platforms exclude: "arm64" CGO errors
 	done;
 	@rm -rf $(BIN_DIRECTORY)/$(APP_NAME)
 
+.PHONY: go-test
+go-test: ## Run go test
+	go test -v ./...
+
 # DOCKER TASKS
 # Build the container
 .PHONY: docker-build

README.md

@@ -137,11 +137,12 @@ planet-exporter \
 
 #### Inventory
 
-Query inventory data to map IP into `hostgroup` (an identifier based on ansible convention) and `domain`.
+Query inventory data that will be used to map `ip_address` into `hostgroup` (an identifier based on Ansible convention) and `domain`.
+The `ip_address` may use CIDR notation (e.g. "10.1.0.0/16") and Inventory task will use the longest-prefix match.
 
-Without this task enabled, those hostgroup and domain fields will be left empty.
+Without this task enabled, those hostgroup and domain fields will be empty.
 
-The flag `--task-inventory-addr` should contain an HTTP endpoint that returns inventory data in the supported format:
+The flag `--task-inventory-addr` accepts an HTTP endpoint that returns inventory data in the supported format:
 
 ##### --task-inventory-format=arrayjson
 
@@ -156,6 +157,11 @@ The flag `--task-inventory-addr` should contain an HTTP endpoint that returns in
     "ip_address": "10.2.3.4",
     "domain": "debugapp.service.consul",
     "hostgroup": "debugapp"
+  },
+  {
+    "ip_address": "10.3.0.0/16",
+    "domain": "",
+    "hostgroup": "unknown-but-its-network-xyz"
   }
 ]
 ```
@@ -165,6 +171,7 @@ The flag `--task-inventory-addr` should contain an HTTP endpoint that returns in
 ```json
 {"ip_address":"10.0.1.2","domain":"xyz.service.consul","hostgroup":"xyz"}
 {"ip_address":"172.16.1.2","domain":"abc.service.consul","hostgroup":"abc"}
+{"ip_address":"10.3.0.0/16","domain":"","hostgroup":"unknown-but-its-network-xyz"}
 ```
 
 #### Socketstat

cmd/planet-exporter/internal/internal.go

@@ -160,7 +160,7 @@ func (s Service) collect(ctx context.Context, interval time.Duration) {
 	log.Infof("Task Darkstat: %v", s.Config.TaskDarkstatEnabled)
 	taskdarkstat.InitTask(ctx, s.Config.TaskDarkstatEnabled, s.Config.TaskDarkstatAddr)
 
-	log.Infof("Task ebpf: %v", s.Config.TaskEbpfEnabled)
+	log.Infof("Task EBPF: %v", s.Config.TaskEbpfEnabled)
 	taskebpf.InitTask(ctx, s.Config.TaskEbpfEnabled, s.Config.TaskEbpfAddr)
 
 	log.Infof("Task Inventory: %v", s.Config.TaskInventoryEnabled)

collector/task/darkstat/darkstat.go

@@ -155,24 +155,25 @@ func toHostMetrics(darkstatHostBytesTotal *prom2json.Family) ([]Metric, error) {
 	// To label source traffic that we need to build dependency graph
 	localHostgroup := localAddr.String()
 	localDomain := localAddr.String()
-	localInventory, ok := inventoryHosts[localAddr.String()]
+	localInventory, ok := inventoryHosts.GetHost(localAddr.String())
 	if ok {
 		localHostgroup = localInventory.Hostgroup
 		localDomain = localInventory.Domain
+	} else {
+		log.Warnf("Local address don't exist in inventory: %v", localAddr.String())
 	}
-	log.Debugf("Local address don't exist in inventory: %v", localAddr.String())
 
 	for _, m := range darkstatHostBytesTotal.Metrics {
 		metric := m.(prom2json.Metric)
 
-		ip := net.ParseIP(metric.Labels["ip"])
-
-		// Skip its own IP as we don't need it
-		if ip.Equal(nil) || ip.Equal(localAddr) {
+		// Skip its own IP.
+		// We're not interested in traffic coming from and going to itself.
+		remoteIP := net.ParseIP(metric.Labels["ip"])
+		if remoteIP.Equal(nil) || remoteIP.Equal(localAddr) {
 			continue
 		}
 
-		inventoryHostInfo := inventoryHosts[metric.Labels["ip"]]
+		remoteInventoryHost, _ := inventoryHosts.GetHost(metric.Labels["ip"])
 
 		bandwidth, err := strconv.ParseFloat(metric.Value, 64)
 		if err != nil {
@@ -191,10 +192,10 @@ func toHostMetrics(darkstatHostBytesTotal *prom2json.Family) ([]Metric, error) {
 
 		hosts = append(hosts, Metric{
 			LocalHostgroup:  localHostgroup,
-			RemoteHostgroup: inventoryHostInfo.Hostgroup,
+			RemoteHostgroup: remoteInventoryHost.Hostgroup,
 			RemoteIPAddr:    metric.Labels["ip"],
 			LocalDomain:     localDomain,
-			RemoteDomain:    inventoryHostInfo.Domain,
+			RemoteDomain:    remoteInventoryHost.Domain,
 			Direction:       direction,
 			Bandwidth:       bandwidth,
 		})

collector/task/ebpf/tcptop.go

@@ -176,22 +176,25 @@ func toHostMetrics(bytesMetric *prom2json.Family, direction string) ([]Metric, e
 	// To label source traffic that we need to build dependency graph.
 	localHostgroup := localAddr.String()
 	localDomain := localAddr.String()
-	localInventory, ok := inventoryHosts[localAddr.String()]
+	localInventory, ok := inventoryHosts.GetHost(localAddr.String())
 	if ok {
 		localHostgroup = localInventory.Hostgroup
 		localDomain = localInventory.Domain
+	} else {
+		log.Warnf("Local address doesn't exist in the inventory: %v", localAddr.String())
 	}
-	log.Debugf("Local address doesn't exist in the inventory: %v", localAddr.String())
 
 	for _, m := range bytesMetric.Metrics {
 		metric := m.(prom2json.Metric)
-		destIP := net.ParseIP(metric.Labels["daddr"])
 
-		if destIP.Equal(nil) || destIP.Equal(localAddr) {
+		// Skip its own IP.
+		// We're not interested in traffic coming from and going to itself.
+		remoteIP := net.ParseIP(metric.Labels["daddr"])
+		if remoteIP.Equal(nil) || remoteIP.Equal(localAddr) {
 			continue
 		}
 
-		inventoryHostInfo := inventoryHosts[metric.Labels["daddr"]]
+		remoteInventoryHost, _ := inventoryHosts.GetHost(metric.Labels["daddr"])
 
 		bandwidth, err := strconv.ParseFloat(metric.Value, 64)
 		if err != nil {
@@ -201,10 +204,10 @@ func toHostMetrics(bytesMetric *prom2json.Family, direction string) ([]Metric, e
 
 		hosts = append(hosts, Metric{
 			LocalHostgroup:  localHostgroup,
-			RemoteHostgroup: inventoryHostInfo.Hostgroup,
+			RemoteHostgroup: remoteInventoryHost.Hostgroup,
 			RemoteIPAddr:    metric.Labels["daddr"],
 			LocalDomain:     localDomain,
-			RemoteDomain:    inventoryHostInfo.Domain,
+			RemoteDomain:    remoteInventoryHost.Domain,
 			Direction:       direction,
 			Bandwidth:       bandwidth,
 		})

collector/task/inventory/host.go

@@ -0,0 +1,69 @@
+package inventory
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"io/ioutil"
+	"net/http"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// Host contains an inventory entry
+type Host struct {
+	Domain    string `json:"domain"`
+	Hostgroup string `json:"hostgroup"`
+	IPAddress string `json:"ip_address"`
+}
+
+// requestHosts requests a new inventory host entries from upstream inventoryAddr
+func requestHosts(ctx context.Context, httpClient *http.Client, inventoryFormat, inventoryAddr string) ([]Host, error) {
+	request, err := http.NewRequestWithContext(ctx, http.MethodGet, inventoryAddr, nil)
+	if err != nil {
+		return nil, err
+	}
+	response, err := httpClient.Do(request)
+	if err != nil {
+		return nil, err
+	}
+	defer response.Body.Close()
+
+	return parseHosts(inventoryFormat, response.Body)
+}
+
+// parseHosts parses inventory data as a list of Host
+func parseHosts(format string, data io.Reader) ([]Host, error) {
+	var result []Host
+
+	decoder := json.NewDecoder(data)
+	decoder.DisallowUnknownFields()
+
+	switch format {
+	case "ndjson":
+		inventoryEntry := Host{}
+		for decoder.More() {
+			err := decoder.Decode(&inventoryEntry)
+			if err != nil {
+				log.Errorf("Skip an inventory host entry due to parser error: %v", err)
+				continue
+			}
+			result = append(result, inventoryEntry)
+		}
+
+	case "arrayjson":
+		err := decoder.Decode(&result)
+		if err != nil {
+			return nil, err
+		}
+
+		// We expect a single JSON array object here. Clear unexpected data that's left in the io.ReadCloser
+		if decoder.More() {
+			bytesCopied, _ := io.Copy(ioutil.Discard, data)
+			log.Warnf("Unexpected remaining data (%v Bytes) while parsing inventory hosts", bytesCopied)
+		}
+	}
+	log.Debugf("Parsed %v inventory hosts", len(result))
+
+	return result, nil
+}