From 91e34c55708e6e05027d8dc51e720dce11652cfb Mon Sep 17 00:00:00 2001 From: Eric Wendland Date: Wed, 29 Jul 2026 09:11:23 +0200 Subject: [PATCH] feat: add observability endpoints --- core/nylon.go | 16 ++- core/observability.go | 229 +++++++++++++++++++++++++++++++++++++ core/observability_test.go | 62 ++++++++++ docs/reference/config.mdx | 1 + example/sample-node.yaml | 3 +- state/config.go | 31 ++--- state/validation.go | 6 + 7 files changed, 328 insertions(+), 20 deletions(-) create mode 100644 core/observability.go create mode 100644 core/observability_test.go diff --git a/core/nylon.go b/core/nylon.go index 47a24869..61553d23 100644 --- a/core/nylon.go +++ b/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() diff --git a/core/observability.go b/core/observability.go new file mode 100644 index 00000000..619130f9 --- /dev/null +++ b/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) +} diff --git a/core/observability_test.go b/core/observability_test.go new file mode 100644 index 00000000..66fb37c0 --- /dev/null +++ b/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 ")) +} diff --git a/docs/reference/config.mdx b/docs/reference/config.mdx index 566822ad..3b8ebe7f 100644 --- a/docs/reference/config.mdx +++ b/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: diff --git a/example/sample-node.yaml b/example/sample-node.yaml index e487fe4a..7af856e1 100644 --- a/example/sample-node.yaml +++ b/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: [] \ No newline at end of file +post_down: [] diff --git a/state/config.go b/state/config.go index 309b4d4c..7c963f69 100644 --- a/state/config.go +++ b/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) { diff --git a/state/validation.go b/state/validation.go index 6685a32c..267c3bde 100644 --- a/state/validation.go +++ b/state/validation.go @@ -2,6 +2,7 @@ package state import ( "fmt" + "net" "net/netip" "net/url" "regexp" @@ -37,6 +38,11 @@ func NodeConfigValidator(central *CentralCfg, node *LocalCfg) error { return fmt.Errorf("interface name is invalid: %v", err) } } + if node.ObservabilityAddr != "" { + if _, _, err := net.SplitHostPort(node.ObservabilityAddr); err != nil { + return fmt.Errorf("observability address must be a valid host:port: %v", err) + } + } if node.Dist != nil { _, err := url.Parse(node.Dist.Url) if err != nil {