RPKI cache-connection status metrics

README.md

@@ -65,6 +65,7 @@ Flags:
       --[no-]collector.pim       Enable the pim collector (default: disabled).
       --[no-]collector.route     Enable the route collector (default: enabled, to disable use
                                  --no-collector.route).
+      --[no-]collector.rpki      Enable the rpki collector (default: disabled).
       --[no-]collector.vrrp      Enable the vrrp collector (default: disabled).
       --web.telemetry-path="/metrics"
                                  Path under which to expose metrics.
@@ -136,6 +137,7 @@ Name | Description
 --- | ---
 BGP IPv6 | Per VRF and address family (currently support unicast only) BGP IPv6 metrics:<br> - RIB entries<br> - RIB memory usage<br> - Configured peer count<br> - Peer memory usage<br> - Configure peer group count<br> - Peer group memory usage<br> - Peer messages in<br> - Peer messages out<br> - Peer active prfixes<br> - Peer state (established/down)<br> - Peer uptime
 BGP L2VPN | Per VRF and address family (currently support EVPN only) BGP L2VPN EVPN metrics:<br> - RIB entries<br> - RIB memory usage<br> - Configured peer count<br> - Peer memory usage<br> - Configure peer group count<br> - Peer group memory usage<br> - Peer messages in<br> - Peer messages out<br> - Peer active prfixes<br> - Peer state (established/down)<br> - Peer uptime 
+RPKI | Per VRF RPKI cache-connection metrics (requires FRR compiled with `--enable-rpki`):<br> - Cache connection state (connected/disconnected)<br> - Cache connection preference
 VRRP | Per VRRP Interface, VrID and Protocol:<br> - Rx and TX statistics<br> - VRRP Status<br> - VRRP State Transitions<br>
 PIM | PIM metrics:<br> - Neighbor count<br> - Neighbor uptime
 

collector/collector.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"log/slog"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -155,3 +156,22 @@ func newCounter(ch chan<- prometheus.Metric, descName *prometheus.Desc, metric f
 func cmdOutputProcessError(cmd, output string, err error) error {
 	return fmt.Errorf("cannot process output of %s: %w: command output: %s", cmd, err, output)
 }
+
+func getVRFs() ([]string, error) {
+	output, err := executeZebraCommand("show vrf")
+	if err != nil {
+		return nil, err
+	}
+	return parseVRFs(output), nil
+}
+
+func parseVRFs(output []byte) []string {
+	vrfs := []string{"default"}
+	for _, line := range strings.Split(string(output), "\n") {
+		fields := strings.Fields(line)
+		if len(fields) >= 2 && fields[0] == "vrf" {
+			vrfs = append(vrfs, fields[1])
+		}
+	}
+	return vrfs
+}

collector/route.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
-	"strings"
 
 	"github.com/alecthomas/kingpin/v2"
 	"github.com/prometheus/client_golang/prometheus"
@@ -136,25 +135,6 @@ func emitRouteSummaryMetrics(ch chan<- prometheus.Metric, rs routeSummary, afi s
 	}
 }
 
-func getVRFs() ([]string, error) {
-	output, err := executeZebraCommand("show vrf")
-	if err != nil {
-		return nil, err
-	}
-	return parseVRFs(output), nil
-}
-
-func parseVRFs(output []byte) []string {
-	vrfs := []string{"default"}
-	for _, line := range strings.Split(string(output), "\n") {
-		fields := strings.Fields(line)
-		if len(fields) >= 2 && fields[0] == "vrf" {
-			vrfs = append(vrfs, fields[1])
-		}
-	}
-	return vrfs
-}
-
 type routeSummary struct {
 	Routes         []route `json:"routes"`
 	RoutesTotal    uint32  `json:"routesTotal"`

collector/rpki.go

@@ -0,0 +1,97 @@
+package collector
+
+import (
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"strconv"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var rpkiSubsystem = "rpki"
+
+func init() {
+	registerCollector(rpkiSubsystem, disabledByDefault, NewRPKICollector)
+}
+
+type rpkiCollector struct {
+	logger       *slog.Logger
+	descriptions map[string]*prometheus.Desc
+}
+
+// NewRPKICollector collects RPKI cache-connection metrics, implemented as per the Collector interface.
+func NewRPKICollector(logger *slog.Logger) (Collector, error) {
+	return &rpkiCollector{logger: logger, descriptions: getRPKIDesc()}, nil
+}
+
+func getRPKIDesc() map[string]*prometheus.Desc {
+	labels := []string{"vrf", "mode", "host", "port"}
+	return map[string]*prometheus.Desc{
+		"cacheState":      colPromDesc(rpkiSubsystem, "cache_state", "State of the RPKI cache connection (1 = connected, 0 = disconnected).", labels),
+		"cachePreference": colPromDesc(rpkiSubsystem, "cache_preference", "Preference value of the RPKI cache connection.", labels),
+	}
+}
+
+// Update implemented as per the Collector interface.
+func (c *rpkiCollector) Update(ch chan<- prometheus.Metric) error {
+	vrfs, err := getVRFs()
+	if err != nil {
+		return err
+	}
+
+	for _, vrf := range vrfs {
+		var cmd string
+		if vrf == "default" {
+			cmd = "show rpki cache-connection json"
+		} else {
+			cmd = fmt.Sprintf("show rpki cache-connection vrf %s json", vrf)
+		}
+
+		output, err := executeBGPCommand(cmd)
+		if err != nil {
+			return err
+		}
+		if len(output) == 0 {
+			continue
+		}
+
+		if err := processRPKICacheConnection(ch, output, vrf, c.descriptions); err != nil {
+			return cmdOutputProcessError(cmd, string(output), err)
+		}
+	}
+	return nil
+}
+
+func processRPKICacheConnection(ch chan<- prometheus.Metric, jsonRPKI []byte, vrf string, rpkiDesc map[string]*prometheus.Desc) error {
+	var cacheConn rpkiCacheConnection
+	if err := json.Unmarshal(jsonRPKI, &cacheConn); err != nil {
+		return err
+	}
+
+	for _, conn := range cacheConn.Connections {
+		labels := []string{vrf, conn.Mode, conn.Host, strconv.Itoa(conn.Port)}
+
+		state := 0.0
+		if conn.State == "connected" {
+			state = 1.0
+		}
+
+		newGauge(ch, rpkiDesc["cacheState"], state, labels...)
+		newGauge(ch, rpkiDesc["cachePreference"], float64(conn.Preference), labels...)
+	}
+	return nil
+}
+
+type rpkiCacheConnection struct {
+	ConnectedGroup int              `json:"connectedGroup"`
+	Connections    []rpkiConnection `json:"connections"`
+}
+
+type rpkiConnection struct {
+	Mode       string `json:"mode"`
+	Host       string `json:"host"`
+	Port       int    `json:"port,string"`
+	Preference int    `json:"preference"`
+	State      string `json:"state"`
+}

collector/rpki_test.go

@@ -0,0 +1,41 @@
+package collector
+
+import (
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var expectedRPKIMetrics = map[string]float64{
+	"frr_rpki_cache_state{host=172.20.15.59,mode=tcp,port=8082,vrf=default}":      1,
+	"frr_rpki_cache_preference{host=172.20.15.59,mode=tcp,port=8082,vrf=default}": 10,
+	"frr_rpki_cache_state{host=172.20.15.60,mode=tcp,port=8083,vrf=default}":      0,
+	"frr_rpki_cache_preference{host=172.20.15.60,mode=tcp,port=8083,vrf=default}": 20,
+}
+
+func TestProcessRPKICacheConnection(t *testing.T) {
+	ch := make(chan prometheus.Metric, 1024)
+	if err := processRPKICacheConnection(ch, readTestFixture(t, "show_rpki_cache_connection.json"), "default", getRPKIDesc()); err != nil {
+		t.Errorf("error calling processRPKICacheConnection: %s", err)
+	}
+	close(ch)
+
+	gotMetrics := collectMetrics(t, ch)
+	compareMetrics(t, gotMetrics, expectedRPKIMetrics)
+}
+
+var expectedRPKIVRFMetrics = map[string]float64{
+	"frr_rpki_cache_state{host=172.20.15.59,mode=tcp,port=8082,vrf=TEST}":      1,
+	"frr_rpki_cache_preference{host=172.20.15.59,mode=tcp,port=8082,vrf=TEST}": 10,
+}
+
+func TestProcessRPKICacheConnectionVRF(t *testing.T) {
+	ch := make(chan prometheus.Metric, 1024)
+	if err := processRPKICacheConnection(ch, readTestFixture(t, "show_rpki_cache_connection_vrf_TEST.json"), "TEST", getRPKIDesc()); err != nil {
+		t.Errorf("error calling processRPKICacheConnection VRF: %s", err)
+	}
+	close(ch)
+
+	gotMetrics := collectMetrics(t, ch)
+	compareMetrics(t, gotMetrics, expectedRPKIVRFMetrics)
+}

collector/testdata/show_rpki_cache_connection.json

@@ -0,0 +1,19 @@
+{
+  "connectedGroup":10,
+  "connections":[
+    {
+      "mode":"tcp",
+      "host":"172.20.15.59",
+      "port":"8082",
+      "preference":10,
+      "state":"connected"
+    },
+    {
+      "mode":"tcp",
+      "host":"172.20.15.60",
+      "port":"8083",
+      "preference":20,
+      "state":"disconnected"
+    }
+  ]
+}

collector/testdata/show_rpki_cache_connection_vrf_TEST.json

@@ -0,0 +1,12 @@
+{
+  "connectedGroup":10,
+  "connections":[
+    {
+      "mode":"tcp",
+      "host":"172.20.15.59",
+      "port":"8082",
+      "preference":10,
+      "state":"connected"
+    }
+  ]
+}