From aee66d8a71876f647d83f15904bf132f61343197 Mon Sep 17 00:00:00 2001 From: Viktor Liu Date: Thu, 4 Jun 2026 17:37:26 +0200 Subject: [PATCH] Expose relay transport in status and label relay peer metrics by transport --- client/internal/peer/status.go | 23 ++++++------ client/internal/relay/relay.go | 3 ++ client/proto/daemon.pb.go | 13 +++++-- client/proto/daemon.proto | 1 + client/status/status.go | 4 +++ client/status/status_test.go | 18 ++++++++-- relay/metrics/realy.go | 9 ++--- relay/server/listener/conn.go | 2 ++ relay/server/listener/quic/conn.go | 5 +++ relay/server/listener/ws/conn.go | 5 +++ relay/server/relay.go | 5 +-- shared/relay/client/client.go | 20 +++++++++++ shared/relay/client/dialer/quic/conn.go | 5 +++ shared/relay/client/dialer/ws/conn.go | 5 +++ shared/relay/client/manager.go | 48 +++++++++++++++++++++++++ 15 files changed, 146 insertions(+), 20 deletions(-) diff --git a/client/internal/peer/status.go b/client/internal/peer/status.go index c5fa4e9f9..35ad423ca 100644 --- a/client/internal/peer/status.go +++ b/client/internal/peer/status.go @@ -1015,27 +1015,29 @@ func (d *Status) GetRelayStates() []relay.ProbeResult { return d.relayStates } - // extend the list of stun, turn servers with relay address + // extend the list of stun, turn servers with the relay server connections relayStates := slices.Clone(d.relayStates) - // if the server connection is not established then we will use the general address - // in case of connection we will use the instance specific address - instanceAddr, _, err := d.relayMgr.RelayInstanceAddress() - if err != nil { - // TODO add their status + states := d.relayMgr.RelayStates() + if len(states) == 0 { + // no relay connection tracked yet; surface configured servers as unavailable for _, r := range d.relayMgr.ServerURLs() { relayStates = append(relayStates, relay.ProbeResult{ URI: r, - Err: err, + Err: relayClient.ErrRelayClientNotConnected, }) } return relayStates } - relayState := relay.ProbeResult{ - URI: instanceAddr, + for _, rs := range states { + relayStates = append(relayStates, relay.ProbeResult{ + URI: rs.URL, + Err: rs.Err, + Transport: rs.Transport, + }) } - return append(relayStates, relayState) + return relayStates } func (d *Status) ForwardingRules() []firewall.ForwardRule { @@ -1395,6 +1397,7 @@ func (fs FullStatus) ToProto() *proto.FullStatus { pbRelayState := &proto.RelayState{ URI: relayState.URI, Available: relayState.Err == nil, + Transport: relayState.Transport, } if err := relayState.Err; err != nil { pbRelayState.Error = err.Error() diff --git a/client/internal/relay/relay.go b/client/internal/relay/relay.go index f00a8d93a..c7946cae7 100644 --- a/client/internal/relay/relay.go +++ b/client/internal/relay/relay.go @@ -32,6 +32,9 @@ type ProbeResult struct { URI string Err error Addr string + // Transport is the negotiated relay transport (e.g. "quic", "ws"), empty + // for stun/turn probes or when not connected. + Transport string } type StunTurnProbe struct { diff --git a/client/proto/daemon.pb.go b/client/proto/daemon.pb.go index 2c054c99a..437220ee9 100644 --- a/client/proto/daemon.pb.go +++ b/client/proto/daemon.pb.go @@ -1832,6 +1832,7 @@ type RelayState struct { URI string `protobuf:"bytes,1,opt,name=URI,proto3" json:"URI,omitempty"` Available bool `protobuf:"varint,2,opt,name=available,proto3" json:"available,omitempty"` Error string `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"` + Transport string `protobuf:"bytes,4,opt,name=transport,proto3" json:"transport,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1887,6 +1888,13 @@ func (x *RelayState) GetError() string { return "" } +func (x *RelayState) GetTransport() string { + if x != nil { + return x.Transport + } + return "" +} + type NSGroupState struct { state protoimpl.MessageState `protogen:"open.v1"` Servers []string `protobuf:"bytes,1,rep,name=servers,proto3" json:"servers,omitempty"` @@ -6406,12 +6414,13 @@ const file_daemon_proto_rawDesc = "" + "\x0fManagementState\x12\x10\n" + "\x03URL\x18\x01 \x01(\tR\x03URL\x12\x1c\n" + "\tconnected\x18\x02 \x01(\bR\tconnected\x12\x14\n" + - "\x05error\x18\x03 \x01(\tR\x05error\"R\n" + + "\x05error\x18\x03 \x01(\tR\x05error\"p\n" + "\n" + "RelayState\x12\x10\n" + "\x03URI\x18\x01 \x01(\tR\x03URI\x12\x1c\n" + "\tavailable\x18\x02 \x01(\bR\tavailable\x12\x14\n" + - "\x05error\x18\x03 \x01(\tR\x05error\"r\n" + + "\x05error\x18\x03 \x01(\tR\x05error\x12\x1c\n" + + "\ttransport\x18\x04 \x01(\tR\ttransport\"r\n" + "\fNSGroupState\x12\x18\n" + "\aservers\x18\x01 \x03(\tR\aservers\x12\x18\n" + "\adomains\x18\x02 \x03(\tR\adomains\x12\x18\n" + diff --git a/client/proto/daemon.proto b/client/proto/daemon.proto index dedff43e2..befd1f4f4 100644 --- a/client/proto/daemon.proto +++ b/client/proto/daemon.proto @@ -370,6 +370,7 @@ message RelayState { string URI = 1; bool available = 2; string error = 3; + string transport = 4; } message NSGroupState { diff --git a/client/status/status.go b/client/status/status.go index 11ed06c2d..1de129765 100644 --- a/client/status/status.go +++ b/client/status/status.go @@ -98,6 +98,7 @@ type RelayStateOutputDetail struct { URI string `json:"uri" yaml:"uri"` Available bool `json:"available" yaml:"available"` Error string `json:"error" yaml:"error"` + Transport string `json:"transport" yaml:"transport"` } type RelayStateOutput struct { @@ -218,6 +219,7 @@ func mapRelays(relays []*proto.RelayState) RelayStateOutput { URI: relay.URI, Available: available, Error: relay.GetError(), + Transport: relay.GetTransport(), }, ) @@ -439,6 +441,8 @@ func (o *OutputOverview) GeneralSummary(showURL bool, showRelays bool, showNameS available = "Unavailable" reason = fmt.Sprintf(", reason: %s", relay.Error) } + } else if relay.Transport != "" { + available = fmt.Sprintf("%s via %s", available, relay.Transport) } relaysString += fmt.Sprintf("\n [%s] is %s%s", relay.URI, available, reason) diff --git a/client/status/status_test.go b/client/status/status_test.go index 0986bf0cd..529ecd7d4 100644 --- a/client/status/status_test.go +++ b/client/status/status_test.go @@ -356,12 +356,14 @@ func TestParsingToJSON(t *testing.T) { { "uri": "stun:my-awesome-stun.com:3478", "available": true, - "error": "" + "error": "", + "transport": "" }, { "uri": "turns:my-awesome-turn.com:443?transport=tcp", "available": false, - "error": "context: deadline exceeded" + "error": "context: deadline exceeded", + "transport": "" } ] }, @@ -480,9 +482,11 @@ relays: - uri: stun:my-awesome-stun.com:3478 available: true error: "" + transport: "" - uri: turns:my-awesome-turn.com:443?transport=tcp available: false error: 'context: deadline exceeded' + transport: "" netbirdIp: 192.168.178.100/16 netbirdIpv6: fd00::100 publicKey: Some-Pub-Key @@ -641,3 +645,13 @@ func TestTimeAgo(t *testing.T) { }) } } + +func TestMapRelaysTransport(t *testing.T) { + out := mapRelays([]*proto.RelayState{ + {URI: "rels://relay.example:443", Available: true, Transport: "quic"}, + {URI: "rels://relay2.example:443", Available: true, Transport: "ws"}, + }) + require.Len(t, out.Details, 2) + assert.Equal(t, "quic", out.Details[0].Transport) + assert.Equal(t, "ws", out.Details[1].Transport) +} diff --git a/relay/metrics/realy.go b/relay/metrics/realy.go index efb597ff5..49a357557 100644 --- a/relay/metrics/realy.go +++ b/relay/metrics/realy.go @@ -6,6 +6,7 @@ import ( "time" log "github.com/sirupsen/logrus" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" ) @@ -119,8 +120,8 @@ func NewMetrics(ctx context.Context, meter metric.Meter) (*Metrics, error) { } // PeerConnected increments the number of connected peers and increments number of idle connections -func (m *Metrics) PeerConnected(id string) { - m.peers.Add(m.ctx, 1) +func (m *Metrics) PeerConnected(id, transport string) { + m.peers.Add(m.ctx, 1, metric.WithAttributes(attribute.String("transport", transport))) m.mutexActivity.Lock() defer m.mutexActivity.Unlock() @@ -138,8 +139,8 @@ func (m *Metrics) RecordPeerStoreTime(duration time.Duration) { } // PeerDisconnected decrements the number of connected peers and decrements number of idle or active connections -func (m *Metrics) PeerDisconnected(id string) { - m.peers.Add(m.ctx, -1) +func (m *Metrics) PeerDisconnected(id, transport string) { + m.peers.Add(m.ctx, -1, metric.WithAttributes(attribute.String("transport", transport))) m.mutexActivity.Lock() defer m.mutexActivity.Unlock() diff --git a/relay/server/listener/conn.go b/relay/server/listener/conn.go index ef0869594..c46cdf894 100644 --- a/relay/server/listener/conn.go +++ b/relay/server/listener/conn.go @@ -11,4 +11,6 @@ type Conn interface { Write(ctx context.Context, b []byte) (n int, err error) RemoteAddr() net.Addr Close() error + // Protocol returns the transport name ("quic", "ws"). + Protocol() string } diff --git a/relay/server/listener/quic/conn.go b/relay/server/listener/quic/conn.go index d8dafcd1f..da5e12d36 100644 --- a/relay/server/listener/quic/conn.go +++ b/relay/server/listener/quic/conn.go @@ -42,6 +42,11 @@ func (c *Conn) RemoteAddr() net.Addr { return c.session.RemoteAddr() } +// Protocol returns the transport name for this connection. +func (c *Conn) Protocol() string { + return "quic" +} + func (c *Conn) Close() error { c.closedMu.Lock() if c.closed { diff --git a/relay/server/listener/ws/conn.go b/relay/server/listener/ws/conn.go index c22b5719d..b1b64fe8e 100644 --- a/relay/server/listener/ws/conn.go +++ b/relay/server/listener/ws/conn.go @@ -64,6 +64,11 @@ func (c *Conn) RemoteAddr() net.Addr { return c.rAddr } +// Protocol returns the transport name for this connection. +func (c *Conn) Protocol() string { + return "ws" +} + func (c *Conn) Close() error { c.closedMu.Lock() c.closed = true diff --git a/relay/server/relay.go b/relay/server/relay.go index 56add8bea..84c424b8e 100644 --- a/relay/server/relay.go +++ b/relay/server/relay.go @@ -154,15 +154,16 @@ func (r *Relay) Accept(conn listener.Conn) { } r.notifier.PeerCameOnline(peer.ID()) + transport := conn.Protocol() r.metrics.RecordPeerStoreTime(time.Since(storeTime)) - r.metrics.PeerConnected(peer.String()) + r.metrics.PeerConnected(peer.String(), transport) go func() { peer.Work() if deleted := r.store.DeletePeer(peer); deleted { r.notifier.PeerWentOffline(peer.ID()) } peer.log.Debugf("relay connection closed") - r.metrics.PeerDisconnected(peer.String()) + r.metrics.PeerDisconnected(peer.String(), transport) }() if err := h.handshakeResponse(hsCtx); err != nil { diff --git a/shared/relay/client/client.go b/shared/relay/client/client.go index 002b8d134..68dd2ab42 100644 --- a/shared/relay/client/client.go +++ b/shared/relay/client/client.go @@ -182,6 +182,23 @@ type Client struct { // datagramFallbackTriggered guards a single fallback per connection so a // burst of oversized datagrams triggers one reconnect, not many. datagramFallbackTriggered atomic.Bool + + // transport is the negotiated relay transport (e.g. "quic", "ws") of the + // current connection, guarded by mu. + transport string +} + +// transportConn is implemented by relay connections that know their transport. +type transportConn interface { + Protocol() string +} + +// Transport returns the negotiated relay transport of the current connection, +// or an empty string when not connected. +func (c *Client) Transport() string { + c.mu.Lock() + defer c.mu.Unlock() + return c.transport } // SetTransportFallback wires the shared datagram-transport fallback tracker. @@ -402,6 +419,9 @@ func (c *Client) connect(ctx context.Context) (*RelayAddr, error) { } c.relayConn = conn c.datagramFallbackTriggered.Store(false) + if tc, ok := conn.(transportConn); ok { + c.transport = tc.Protocol() + } instanceURL, err := c.handShake(ctx) if err != nil { diff --git a/shared/relay/client/dialer/quic/conn.go b/shared/relay/client/dialer/quic/conn.go index a5c982551..e5ad77b29 100644 --- a/shared/relay/client/dialer/quic/conn.go +++ b/shared/relay/client/dialer/quic/conn.go @@ -57,6 +57,11 @@ func (c *Conn) Write(b []byte) (int, error) { return len(b), nil } +// Protocol returns the transport name for this connection. +func (c *Conn) Protocol() string { + return Network +} + func (c *Conn) RemoteAddr() net.Addr { return c.session.RemoteAddr() } diff --git a/shared/relay/client/dialer/ws/conn.go b/shared/relay/client/dialer/ws/conn.go index 9497fab89..eec417c50 100644 --- a/shared/relay/client/dialer/ws/conn.go +++ b/shared/relay/client/dialer/ws/conn.go @@ -33,6 +33,11 @@ func NewConn(wsConn *websocket.Conn, serverAddress string, underlying net.Conn) } } +// Protocol returns the transport name for this connection. +func (c *Conn) Protocol() string { + return Network +} + func (c *Conn) Read(b []byte) (n int, err error) { t, ioReader, err := c.Conn.Reader(c.ctx) if err != nil { diff --git a/shared/relay/client/manager.go b/shared/relay/client/manager.go index f87da15de..c4eba750b 100644 --- a/shared/relay/client/manager.go +++ b/shared/relay/client/manager.go @@ -242,6 +242,54 @@ func (m *Manager) ServerURLs() []string { return m.serverPicker.ServerURLs.Load().([]string) } +// RelayConnState is the connection state of a single relay server. +type RelayConnState struct { + // URL is the server's instance address when connected, otherwise the + // configured server URL. + URL string + // Transport is the negotiated transport ("quic", "ws"), empty if not connected. + Transport string + // Err is set when the relay is not connected. + Err error +} + +// RelayStates returns the connection state of the home relay and every foreign +// relay the manager currently tracks. +func (m *Manager) RelayStates() []RelayConnState { + var states []RelayConnState + + m.relayClientMu.RLock() + home := m.relayClient + m.relayClientMu.RUnlock() + if home != nil { + states = append(states, relayConnState(home)) + } + + m.relayClientsMutex.RLock() + for _, rt := range m.relayClients { + rt.RLock() + rc, rtErr := rt.relayClient, rt.err + rt.RUnlock() + switch { + case rc != nil: + states = append(states, relayConnState(rc)) + case rtErr != nil: + states = append(states, RelayConnState{Err: rtErr}) + } + } + m.relayClientsMutex.RUnlock() + + return states +} + +func relayConnState(c *Client) RelayConnState { + addr, err := c.ServerInstanceURL() + if err != nil { + return RelayConnState{URL: c.connectionURL, Err: err} + } + return RelayConnState{URL: addr, Transport: c.Transport()} +} + // HasRelayAddress returns true if the manager is serving. With this method can check if the peer can communicate with // Relay service. func (m *Manager) HasRelayAddress() bool {