feat: add observability endpoints

core/nylon.go

@@ -58,10 +58,11 @@ type Nylon struct {
 	ConfigPath      string
 
 	// resources
-	Tun       tun.Device
-	wgUapi    net.Listener
-	Interface string
-	Device    *device.Device
+	Tun           tun.Device
+	wgUapi        net.Listener
+	Interface     string
+	Device        *device.Device
+	observability *observabilityServer
 
 	// only used for debugging & tests
 	AuxConfig map[string]any
@@ -252,6 +253,10 @@ func (n *Nylon) Init() error {
 func (n *Nylon) Start() error {
 	n.Log.Info("init modules complete")
 
+	if err := n.startObservability(); err != nil {
+		return err
+	}
+
 	n.Log.Info("Nylon has been initialized. To gracefully exit, send SIGINT or Ctrl+C.")
 
 	c := make(chan os.Signal, 1)
@@ -322,6 +327,9 @@ endLoop:
 }
 
 func (n *Nylon) Cleanup() error {
+	if n.observability != nil {
+		n.observability.close()
+	}
 	n.PingBuf.Stop()
 	for _, health := range n.prefixHealth {
 		health.monitor.Stop()

core/observability.go

@@ -0,0 +1,229 @@
+package core
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"sort"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/encodeous/nylon/protocol"
+)
+
+const observabilityTimeout = time.Second
+
+type observabilityServer struct {
+	server    *http.Server
+	listener  net.Listener
+	closeOnce sync.Once
+}
+
+type discoveryGroup struct {
+	Targets []string          `json:"targets"`
+	Labels  map[string]string `json:"labels"`
+}
+
+func (n *Nylon) startObservability() error {
+	if n.LocalCfg.ObservabilityAddr == "" {
+		return nil
+	}
+
+	listener, err := net.Listen("tcp", n.LocalCfg.ObservabilityAddr)
+	if err != nil {
+		return fmt.Errorf("listen on observability address %q: %w", n.LocalCfg.ObservabilityAddr, err)
+	}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/healthz", n.handleHealth)
+	mux.HandleFunc("/readyz", n.handleReady)
+	mux.HandleFunc("/metrics", n.handleMetrics)
+	mux.HandleFunc("/discovery", n.handleDiscovery)
+
+	obs := &observabilityServer{
+		listener: listener,
+		server: &http.Server{
+			Handler:           mux,
+			ReadHeaderTimeout: 5 * time.Second,
+		},
+	}
+	n.observability = obs
+	n.Log.Info("observability server started", "address", listener.Addr())
+
+	go func() {
+		if err := obs.server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			n.Log.Error("observability server failed", "error", err)
+			n.Cancel(fmt.Errorf("observability server failed: %w", err))
+		}
+	}()
+	go func() {
+		<-n.Context.Done()
+		obs.close()
+	}()
+	return nil
+}
+
+func (o *observabilityServer) close() {
+	o.closeOnce.Do(func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = o.server.Shutdown(ctx)
+	})
+}
+
+func (n *Nylon) handleHealth(w http.ResponseWriter, _ *http.Request) {
+	if n.Context.Err() != nil {
+		http.Error(w, "shutting down", http.StatusServiceUnavailable)
+		return
+	}
+	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+	_, _ = io.WriteString(w, "ok\n")
+}
+
+func (n *Nylon) statusSnapshot(ctx context.Context) (*protocol.StatusResponse, error) {
+	result := make(chan *protocol.StatusResponse, 1)
+	select {
+	case n.DispatchChannel <- func() error {
+		result <- handleStatus(n, &protocol.StatusRequest{}).GetStatus()
+		return nil
+	}:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-n.Context.Done():
+		return nil, context.Cause(n.Context)
+	}
+
+	select {
+	case status := <-result:
+		return status, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-n.Context.Done():
+		return nil, context.Cause(n.Context)
+	}
+}
+
+func (n *Nylon) handleReady(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithTimeout(r.Context(), observabilityTimeout)
+	defer cancel()
+	if _, err := n.statusSnapshot(ctx); err != nil {
+		http.Error(w, "not ready", http.StatusServiceUnavailable)
+		return
+	}
+	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+	_, _ = io.WriteString(w, "ok\n")
+}
+
+func (n *Nylon) handleMetrics(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithTimeout(r.Context(), observabilityTimeout)
+	defer cancel()
+	status, err := n.statusSnapshot(ctx)
+	if err != nil {
+		http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
+		return
+	}
+	w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
+	writePrometheusMetrics(w, status)
+}
+
+func (n *Nylon) handleDiscovery(w http.ResponseWriter, _ *http.Request) {
+	_, port, err := net.SplitHostPort(n.LocalCfg.ObservabilityAddr)
+	if err != nil {
+		http.Error(w, "service discovery unavailable", http.StatusInternalServerError)
+		return
+	}
+	groups := make([]discoveryGroup, 0)
+	for _, node := range n.CentralCfg.GetNodes() {
+		targets := make([]string, 0, len(node.Addresses))
+		for _, addr := range node.Addresses {
+			targets = append(targets, net.JoinHostPort(addr.String(), port))
+		}
+		if len(targets) != 0 {
+			groups = append(groups, discoveryGroup{
+				Targets: targets,
+				Labels:  map[string]string{"nylon_node": string(node.Id)},
+			})
+		}
+	}
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(groups)
+}
+
+func writePrometheusMetrics(w io.Writer, status *protocol.StatusResponse) {
+	metrics := metricWriter{w: w, seen: make(map[string]struct{})}
+	node := status.GetNode()
+	stats := node.GetStats()
+	metrics.metric("nylon_up", "Whether the nylon daemon is ready.", "gauge", nil, 1)
+	metrics.metric("nylon_config_timestamp_seconds", "Unix timestamp of the active central configuration.", "gauge", nil, float64(node.ConfigTimestamp)/float64(time.Second))
+	metrics.metric("nylon_neighbours", "Number of configured neighbours.", "gauge", nil, float64(stats.NeighbourCount))
+	metrics.metric("nylon_active_endpoints", "Number of active peer endpoints.", "gauge", nil, float64(stats.ActiveEndpointCount))
+	metrics.metric("nylon_selected_routes", "Number of selected Babel routes.", "gauge", nil, float64(stats.SelectedRouteCount))
+	metrics.metric("nylon_advertised_prefixes", "Number of locally advertised prefixes.", "gauge", nil, float64(stats.AdvertisedPrefixCount))
+	metrics.metric("nylon_wireguard_transmit_bytes_total", "WireGuard bytes transmitted by this node.", "counter", nil, float64(stats.TxBytes))
+	metrics.metric("nylon_wireguard_receive_bytes_total", "WireGuard bytes received by this node.", "counter", nil, float64(stats.RxBytes))
+
+	for _, neigh := range status.Neighbours {
+		labels := map[string]string{"peer": neigh.PeerId}
+		wg := neigh.GetWireguard()
+		metrics.metric("nylon_wireguard_peer_transmit_bytes_total", "WireGuard bytes transmitted to a peer.", "counter", labels, float64(wg.TxBytes))
+		metrics.metric("nylon_wireguard_peer_receive_bytes_total", "WireGuard bytes received from a peer.", "counter", labels, float64(wg.RxBytes))
+		handshake := float64(0)
+		if wg.LatestHandshakeUnix > 0 {
+			handshake = float64(wg.LatestHandshakeUnix) / float64(time.Second)
+		}
+		metrics.metric("nylon_wireguard_peer_latest_handshake_seconds", "Unix time of the latest WireGuard handshake.", "gauge", labels, handshake)
+		for _, endpoint := range neigh.Endpoints {
+			epLabels := map[string]string{"peer": neigh.PeerId, "endpoint": endpoint.Address}
+			active := float64(0)
+			if endpoint.Active {
+				active = 1
+			}
+			metrics.metric("nylon_endpoint_active", "Whether a peer endpoint is active.", "gauge", epLabels, active)
+			metrics.metric("nylon_endpoint_metric", "Current Babel endpoint metric.", "gauge", epLabels, float64(endpoint.Metric))
+			metrics.metric("nylon_endpoint_rtt_seconds", "Filtered endpoint round-trip time.", "gauge", epLabels, float64(endpoint.FilteredRttNs)/float64(time.Second))
+		}
+	}
+	for _, route := range status.GetRoutes().GetSelected() {
+		pub := route.GetPubRoute()
+		labels := map[string]string{
+			"prefix":   pub.GetSource().GetPrefix(),
+			"router":   pub.GetSource().GetNodeId(),
+			"next_hop": route.GetNh(),
+		}
+		metrics.metric("nylon_route_metric", "Metric of a selected Babel route.", "gauge", labels, float64(pub.GetFd().GetMetric()))
+	}
+}
+
+type metricWriter struct {
+	w    io.Writer
+	seen map[string]struct{}
+}
+
+func (m *metricWriter) metric(name, help, metricType string, labels map[string]string, value float64) {
+	if _, ok := m.seen[name]; !ok {
+		_, _ = fmt.Fprintf(m.w, "# HELP %s %s\n# TYPE %s %s\n", name, help, name, metricType)
+		m.seen[name] = struct{}{}
+	}
+	_, _ = io.WriteString(m.w, name)
+	if len(labels) != 0 {
+		keys := make([]string, 0, len(labels))
+		for key := range labels {
+			keys = append(keys, key)
+		}
+		sort.Strings(keys)
+		_, _ = io.WriteString(m.w, "{")
+		for i, key := range keys {
+			if i != 0 {
+				_, _ = io.WriteString(m.w, ",")
+			}
+			_, _ = fmt.Fprintf(m.w, `%s=%s`, key, strconv.Quote(labels[key]))
+		}
+		_, _ = io.WriteString(m.w, "}")
+	}
+	_, _ = fmt.Fprintf(m.w, " %g\n", value)
+}

core/observability_test.go

@@ -0,0 +1,62 @@
+package core
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"strings"
+	"testing"
+
+	"github.com/encodeous/nylon/protocol"
+	"github.com/encodeous/nylon/state"
+	"github.com/stretchr/testify/require"
+)
+
+func TestObservabilityHealth(t *testing.T) {
+	ctx, cancel := context.WithCancelCause(context.Background())
+	n := &Nylon{Context: ctx}
+
+	rec := httptest.NewRecorder()
+	n.handleHealth(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil))
+	require.Equal(t, http.StatusOK, rec.Code)
+	require.Equal(t, "ok\n", rec.Body.String())
+
+	cancel(context.Canceled)
+	rec = httptest.NewRecorder()
+	n.handleHealth(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil))
+	require.Equal(t, http.StatusServiceUnavailable, rec.Code)
+}
+
+func TestObservabilityDiscovery(t *testing.T) {
+	n := &Nylon{ConfigState: state.ConfigState{
+		LocalCfg: state.LocalCfg{ObservabilityAddr: "0.0.0.0:9090"},
+		CentralCfg: state.CentralCfg{Routers: []state.RouterCfg{{
+			NodeCfg: state.NodeCfg{
+				Id:        "alice",
+				Addresses: []netip.Addr{netip.MustParseAddr("10.0.0.1"), netip.MustParseAddr("fd00::1")},
+			},
+		}}},
+	}}
+	rec := httptest.NewRecorder()
+	n.handleDiscovery(rec, httptest.NewRequest(http.MethodGet, "/discovery", nil))
+	require.Equal(t, http.StatusOK, rec.Code)
+	require.JSONEq(t, `[{"targets":["10.0.0.1:9090","[fd00::1]:9090"],"labels":{"nylon_node":"alice"}}]`, rec.Body.String())
+}
+
+func TestPrometheusMetrics(t *testing.T) {
+	status := &protocol.StatusResponse{
+		Node: &protocol.NodeStatus{Stats: &protocol.NodeStats{TxBytes: 12}},
+		Neighbours: []*protocol.NeighbourInfo{
+			{PeerId: "bob", Wireguard: &protocol.WireGuardPeerStats{TxBytes: 7}},
+			{PeerId: "eve", Wireguard: &protocol.WireGuardPeerStats{TxBytes: 5}},
+		},
+	}
+	var buf bytes.Buffer
+	writePrometheusMetrics(&buf, status)
+	output := buf.String()
+	require.Contains(t, output, "# TYPE nylon_wireguard_transmit_bytes_total counter")
+	require.Contains(t, output, `nylon_wireguard_peer_transmit_bytes_total{peer="bob"} 7`)
+	require.Equal(t, 1, strings.Count(output, "# HELP nylon_wireguard_peer_transmit_bytes_total "))
+}

docs/reference/config.mdx

@@ -21,6 +21,7 @@ no_net_configure: false # if true, nylon won't touch system routes or interfaces
 log_path: "" # write logs to this file (empty = stderr only)
 interface_name: "" # override the interface name (default: "nylon", or utunX on macOS)
 dns_resolvers: [] # DNS servers for nylon's own lookups, e.g. ["1.1.1.1:53"]
+observability_addr: "" # e.g. "0.0.0.0:9090"; enables /metrics, /healthz, /readyz, and /discovery
 
 # Bootstrap: fetch central.yaml from a remote bundle on first start
 dist:

example/sample-node.yaml

@@ -8,6 +8,7 @@ no_net_configure: false # Default: false - do not configure system networking at
 log_path: "" # Default: "" - If set, Nylon will log to this file
 interface_name: "" # Default: "" - If set, Nylon will use this interface name instead of the default "nylon" or utunx on macOS
 dns_resolvers: [] # Default: [] - If set (e.g ["1.1.1.1:53"]), nylon will use these DNS resolvers for its own queries
+observability_addr: "" # e.g. "0.0.0.0:9090" - enables /metrics, /healthz, /readyz, and /discovery
 dist: # Optional: If set, Nylon will bootstrap central.yaml from this URL if it does not exist already
   url: https://static.example.com/network1.nybundle
   key: 7PaN6DmAayz4KnDnsXSXJH+Oy0TFGeoM4FEbQfLriVY=
@@ -20,4 +21,4 @@ pre_up: []
 pre_down: []
 post_up:
   - iptables -t nat -A POSTROUTING -s 10.0.0.0/24 -d 192.168.0.0/24 -j MASQUERADE # can be used in conjunction with an advertised prefix
-post_down: []
+post_down: []

state/config.go

@@ -49,21 +49,22 @@ type CentralCfg struct {
 // LocalCfg represents local node-level configuration
 type LocalCfg struct {
 	// Node Private Key
-	Key              NyPrivateKey
-	Id               NodeId                // unique id for this node
-	Port             uint16                // Address that the data plane can be accessed by
-	Dist             *LocalDistributionCfg `yaml:",omitempty"`                   // distribution configuration
-	UseSystemRouting bool                  `yaml:"use_system_routing,omitempty"` // all packets from peers will come out of the TUN interface
-	NoNetConfigure   bool                  `yaml:"no_net_configure,omitempty"`   // do not configure system networking at all
-	DnsResolvers     []string              `yaml:"dns_resolvers,omitempty"`      // DNS resolvers used for endpoints and config repositories
-	InterfaceName    string                `yaml:"interface_name,omitempty"`     // the name of the nylon interface
-	LogPath          string                `yaml:"log_path,omitempty"`           // if not empty, nylon will write to this file
-	UnexcludeIPs     []netip.Prefix        `yaml:"unexclude_ips,omitempty"`      // split tunnel, subtracts from centrally excluded ip ranges
-	ExcludeIPs       []netip.Prefix        `yaml:"exclude_ips,omitempty"`        // split tunnel, adds to the centrally excluded ip ranges
-	PreUp            []string              `yaml:"pre_up,omitempty"`             // a list of commands executed in order before the nylon interface is brought up
-	PreDown          []string              `yaml:"pre_down,omitempty"`           // a list of commands executed in order before the nylon interface is brought down
-	PostUp           []string              `yaml:"post_up,omitempty"`            // a list of commands executed in order after the nylon interface is brought up
-	PostDown         []string              `yaml:"post_down,omitempty"`          // a list of commands executed in order after the nylon interface is brought down
+	Key               NyPrivateKey
+	Id                NodeId                // unique id for this node
+	Port              uint16                // Address that the data plane can be accessed by
+	Dist              *LocalDistributionCfg `yaml:",omitempty"`                   // distribution configuration
+	UseSystemRouting  bool                  `yaml:"use_system_routing,omitempty"` // all packets from peers will come out of the TUN interface
+	NoNetConfigure    bool                  `yaml:"no_net_configure,omitempty"`   // do not configure system networking at all
+	DnsResolvers      []string              `yaml:"dns_resolvers,omitempty"`      // DNS resolvers used for endpoints and config repositories
+	InterfaceName     string                `yaml:"interface_name,omitempty"`     // the name of the nylon interface
+	LogPath           string                `yaml:"log_path,omitempty"`           // if not empty, nylon will write to this file
+	ObservabilityAddr string                `yaml:"observability_addr,omitempty"` // HTTP address for metrics, health, readiness, and service discovery
+	UnexcludeIPs      []netip.Prefix        `yaml:"unexclude_ips,omitempty"`      // split tunnel, subtracts from centrally excluded ip ranges
+	ExcludeIPs        []netip.Prefix        `yaml:"exclude_ips,omitempty"`        // split tunnel, adds to the centrally excluded ip ranges
+	PreUp             []string              `yaml:"pre_up,omitempty"`             // a list of commands executed in order before the nylon interface is brought up
+	PreDown           []string              `yaml:"pre_down,omitempty"`           // a list of commands executed in order before the nylon interface is brought down
+	PostUp            []string              `yaml:"post_up,omitempty"`            // a list of commands executed in order after the nylon interface is brought up
+	PostDown          []string              `yaml:"post_down,omitempty"`          // a list of commands executed in order after the nylon interface is brought down
 }
 
 func (c *CentralCfg) Clone() (error, *CentralCfg) {