derp, magicsock: track home (preferred) vs visiting connections for stats

derp/derp.go

@@ -67,12 +67,13 @@ Steady state:
 * server then sends frameRecvPacket to recipient
 */
 const (
-	frameServerKey  = frameType(0x01) // 8B magic + 32B public key + (0+ bytes future use)
+	frameServerKey     = frameType(0x01) // 8B magic + 32B public key + (0+ bytes future use)
-	frameClientInfo = frameType(0x02) // 32B pub key + 24B nonce + naclbox(json)
+	frameClientInfo    = frameType(0x02) // 32B pub key + 24B nonce + naclbox(json)
-	frameServerInfo = frameType(0x03) // 24B nonce + naclbox(json)
+	frameServerInfo    = frameType(0x03) // 24B nonce + naclbox(json)
-	frameSendPacket = frameType(0x04) // 32B dest pub key + packet bytes
+	frameSendPacket    = frameType(0x04) // 32B dest pub key + packet bytes
-	frameRecvPacket = frameType(0x05) // v0/1: packet bytes, v2: 32B src pub key + packet bytes
+	frameRecvPacket    = frameType(0x05) // v0/1: packet bytes, v2: 32B src pub key + packet bytes
-	frameKeepAlive  = frameType(0x06) // no payload, no-op (to be replaced with ping/pong)
+	frameKeepAlive     = frameType(0x06) // no payload, no-op (to be replaced with ping/pong)
+	frameNotePreferred = frameType(0x07) // 1 byte payload: 0x01 or 0x00 for whether this is client's home node
 )
 
 var bin = binary.BigEndian

derp/derp_client.go

@@ -12,6 +12,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"sync"
 	"time"
 
 	"golang.org/x/crypto/nacl/box"
@@ -27,8 +28,10 @@ type Client struct {
 	logf         logger.Logf
 	nc           net.Conn
 	br           *bufio.Reader
-	bw           *bufio.Writer
+
-	readErr      error // sticky read error
+	wmu     sync.Mutex // hold while writing to bw
+	bw      *bufio.Writer
+	readErr error // sticky read error
 }
 
 func NewClient(privateKey key.Private, nc net.Conn, brw *bufio.ReadWriter, logf logger.Logf) (*Client, error) {
@@ -143,6 +146,9 @@ func (c *Client) send(dstKey key.Public, pkt []byte) (ret error) {
 		return fmt.Errorf("packet too big: %d", len(pkt))
 	}
 
+	c.wmu.Lock()
+	defer c.wmu.Unlock()
+
 	if err := writeFrameHeader(c.bw, frameSendPacket, uint32(len(dstKey)+len(pkt))); err != nil {
 		return err
 	}
@@ -155,6 +161,32 @@ func (c *Client) send(dstKey key.Public, pkt []byte) (ret error) {
 	return c.bw.Flush()
 }
 
+// NotePreferred sends a packet that tells the server whether this
+// client is the user's preferred server. This is only used in the
+// server for stats.
+func (c *Client) NotePreferred(preferred bool) (err error) {
+	defer func() {
+		if err != nil {
+			err = fmt.Errorf("derp.NotePreferred: %v", err)
+		}
+	}()
+
+	c.wmu.Lock()
+	defer c.wmu.Unlock()
+
+	if err := writeFrameHeader(c.bw, frameNotePreferred, 1); err != nil {
+		return err
+	}
+	var b byte = 0x00
+	if preferred {
+		b = 0x01
+	}
+	if err := c.bw.WriteByte(b); err != nil {
+		return err
+	}
+	return c.bw.Flush()
+}
+
 // ReceivedMessage represents a type returned by Client.Recv. Unless
 // otherwise documented, the returned message aliases the byte slice
 // provided to Recv and thus the message is only as good as that

derp/derp_server.go

@@ -16,6 +16,7 @@ import (
 	"expvar"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math/big"
 	"net"
 	"os"
@@ -47,6 +48,9 @@ type Server struct {
 	packetsRecv, bytesRecv expvar.Int
 	packetsDropped         expvar.Int
 	accepts                expvar.Int
+	curClients             expvar.Int
+	curHomeClients         expvar.Int // ones with preferred
+	unknownFrames          expvar.Int
 
 	mu          sync.Mutex
 	closed      bool
@@ -137,13 +141,14 @@ func (s *Server) registerClient(c *sclient) {
 	defer s.mu.Unlock()
 	old := s.clients[c.key]
 	if old == nil {
-		s.logf("derp: %s: client %x: adding connection", c.nc.RemoteAddr(), c.key)
+		c.logf("adding connection")
 	} else {
 		old.nc.Close()
-		s.logf("derp: %s: client %x: adding connection, replacing %s", c.nc.RemoteAddr(), c.key, old.nc.RemoteAddr())
+		c.logf("adding connection, replacing %s", old.nc.RemoteAddr())
 	}
 	s.clients[c.key] = c
 	s.clientsEver[c.key] = true
+	s.curClients.Add(1)
 }
 
 // unregisterClient removes a client from the server.
@@ -152,9 +157,14 @@ func (s *Server) unregisterClient(c *sclient) {
 	defer s.mu.Unlock()
 	cur := s.clients[c.key]
 	if cur == c {
-		s.logf("derp: %s: client %x: removing connection", c.nc.RemoteAddr(), c.key)
+		c.logf("removing connection")
 		delete(s.clients, c.key)
 	}
+
+	s.curClients.Add(-1)
+	if c.preferred {
+		s.curHomeClients.Add(-1)
+	}
 }
 
 func (s *Server) accept(nc net.Conn, brw *bufio.ReadWriter) error {
@@ -172,14 +182,24 @@ func (s *Server) accept(nc net.Conn, brw *bufio.ReadWriter) error {
 		return fmt.Errorf("client %x rejected: %v", clientKey, err)
 	}
 
+	lim := rate.Inf
+	if s.BytesPerSecond != 0 {
+		lim = rate.Limit(s.BytesPerSecond)
+	}
+	const burstBytes = 1 << 20 // generous bandwidth delay product? must be over 64k max packet size.
+	limiter := rate.NewLimiter(lim, burstBytes)
+
 	// At this point we trust the client so we don't time out.
 	nc.SetDeadline(time.Time{})
 
 	c := &sclient{
-		key: clientKey,
+		s:       s,
-		nc:  nc,
+		key:     clientKey,
-		br:  br,
+		nc:      nc,
-		bw:  bw,
+		br:      br,
+		bw:      bw,
+		limiter: limiter,
+		logf:    logger.WithPrefix(s.logf, fmt.Sprintf("derp client %v/%x: ", nc.RemoteAddr(), clientKey)),
 	}
 	if clientInfo != nil {
 		c.info = *clientInfo
@@ -199,64 +219,96 @@ func (s *Server) accept(nc net.Conn, brw *bufio.ReadWriter) error {
 	}
 	defer s.unregisterClient(c)
 
+	return c.run()
+}
+
+func (c *sclient) run() error {
+	s := c.s
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	go s.sendClientKeepAlives(ctx, c)
 
-	lim := rate.Inf
+	go s.sendClientKeepAlives(ctx, c)
-	if s.BytesPerSecond != 0 {
-		lim = rate.Limit(s.BytesPerSecond)
-	}
-	const burstBytes = 1 << 20 // generous bandwidth delay product? must be over 64k max packet size.
-	limiter := rate.NewLimiter(lim, burstBytes)
 
 	for {
 		ft, fl, err := readFrameHeader(c.br)
 		if err != nil {
 			return fmt.Errorf("client %x: readFrameHeader: %v", c.key, err)
 		}
-		if ft != frameSendPacket {
+		switch ft {
-			// TODO: nothing else yet supported
+		case frameNotePreferred:
-			return fmt.Errorf("client %x: unsupported frame %v", c.key, ft)
+			err = c.handleFrameNotePreferred(ft, fl)
+		case frameSendPacket:
+			err = c.handleFrameSendPacket(ctx, ft, fl)
+		default:
+			err = c.handleUnknownFrame(ctx, ft, fl)
 		}
-		dstKey, contents, err := s.recvPacket(ctx, c.br, fl, limiter)
 		if err != nil {
-			return fmt.Errorf("client %x: recvPacket: %v", c.key, err)
+			return err
 		}
+	}
+}
 
-		s.mu.Lock()
+func (c *sclient) handleUnknownFrame(ctx context.Context, ft frameType, fl uint32) error {
-		dst := s.clients[dstKey]
+	if err := c.limiter.WaitN(ctx, int(fl)); err != nil {
-		s.mu.Unlock()
+		return fmt.Errorf("rate limit: %v", err)
+	}
+	_, err := io.CopyN(ioutil.Discard, c.br, int64(fl))
+	return err
+}
 
-		if dst == nil {
+func (c *sclient) handleFrameNotePreferred(ft frameType, fl uint32) error {
-			s.packetsDropped.Add(1)
+	if fl != 1 {
-			if debug {
+		return fmt.Errorf("frameNotePreferred wrong size")
-				s.logf("derp: %s: client %x: dropping packet for unknown %x", nc.RemoteAddr(), c.key, dstKey)
+	}
-			}
+	v, err := c.br.ReadByte()
-			continue
+	if err != nil {
+		return fmt.Errorf("frameNotePreferred ReadByte: %v", err)
+	}
+	c.setPreferred(v != 0)
+	return nil
+}
+
+func (c *sclient) handleFrameSendPacket(ctx context.Context, ft frameType, fl uint32) error {
+	s := c.s
+
+	dstKey, contents, err := s.recvPacket(ctx, c.br, fl, c.limiter)
+	if err != nil {
+		return fmt.Errorf("client %x: recvPacket: %v", c.key, err)
+	}
+
+	s.mu.Lock()
+	dst := s.clients[dstKey]
+	s.mu.Unlock()
+
+	if dst == nil {
+		s.packetsDropped.Add(1)
+		if debug {
+			c.logf("dropping packet for unknown %x", dstKey)
 		}
+		return nil
+	}
 
-		dst.mu.Lock()
+	dst.mu.Lock()
-		err = s.sendPacket(dst.bw, &dst.info, c.key, contents)
+	err = s.sendPacket(dst.bw, &dst.info, c.key, contents)
-		dst.mu.Unlock()
+	dst.mu.Unlock()
 
-		if err != nil {
+	if err != nil {
-			s.logf("derp: %s: client %x: dropping packet for %x: %v", nc.RemoteAddr(), c.key, dstKey, err)
+		c.logf("write error sending packet to %x: %v", dstKey, err)
 
-			// If we cannot send to a destination, shut it down.
+		// If we cannot send to a destination, shut it down.
-			// Let its receive loop do the cleanup.
+		// Let its receive loop do the cleanup.
-			s.mu.Lock()
+		s.mu.Lock()
-			if s.clients[dstKey] == dst {
+		if s.clients[dstKey] == dst {
-				s.clients[dstKey].nc.Close()
+			s.clients[dstKey].nc.Close()
-			}
-			s.mu.Unlock()
 		}
+		s.mu.Unlock()
 	}
+	return err
 }
 
 func (s *Server) sendClientKeepAlives(ctx context.Context, c *sclient) {
 	if err := c.keepAliveLoop(ctx); err != nil {
-		s.logf("derp: %s: client %x: keep alive failed: %v", c.nc.RemoteAddr(), c.key, err)
+		c.logf("keep alive failed: %v", err)
 	}
 }
 
@@ -392,18 +444,34 @@ func (s *Server) recvPacket(ctx context.Context, br *bufio.Reader, frameLen uint
 //
 // (The "s" prefix is to more explicitly distinguish it from Client in derp_client.go)
 type sclient struct {
-	nc   net.Conn
+	s       *Server
-	key  key.Public
+	nc      net.Conn
-	info clientInfo
+	key     key.Public
+	info    clientInfo
+	logf    logger.Logf
+	limiter *rate.Limiter
 
 	keepAliveTimer *time.Timer
 	keepAliveReset chan struct{}
 
+	preferred bool
+
 	mu sync.Mutex // mu guards writing to bw
 	br *bufio.Reader
 	bw *bufio.Writer
 }
 
+func (c *sclient) setPreferred(v bool) {
+	if c.preferred == v {
+		return
+	}
+	if v {
+		c.s.curHomeClients.Add(1)
+	} else {
+		c.s.curHomeClients.Add(-1)
+	}
+}
+
 func (c *sclient) keepAliveLoop(ctx context.Context) error {
 	jitterMs, err := crand.Int(crand.Reader, big.NewInt(5000))
 	if err != nil {
@@ -449,13 +517,15 @@ func (s *Server) expVarFunc(f func() interface{}) expvar.Func {
 // ExpVar returns an expvar variable suitable for registering with expvar.Publish.
 func (s *Server) ExpVar() expvar.Var {
 	m := new(metrics.Set)
-	m.Set("gauge_current_connnections", s.expVarFunc(func() interface{} { return len(s.netConns) }))
 	m.Set("counter_unique_clients_ever", s.expVarFunc(func() interface{} { return len(s.clientsEver) }))
+	m.Set("gauge_current_connnections", &s.curClients)
+	m.Set("gauge_current_home_connnections", &s.curHomeClients)
 	m.Set("accepts", &s.accepts)
 	m.Set("bytes_received", &s.bytesRecv)
 	m.Set("bytes_sent", &s.bytesSent)
 	m.Set("packets_dropped", &s.packetsDropped)
 	m.Set("packets_sent", &s.packetsSent)
 	m.Set("packets_received", &s.packetsRecv)
+	m.Set("unknown_frames", &s.unknownFrames)
 	return m
 }

derp/derphttp/derphttp_client.go

@@ -48,10 +48,11 @@ type Client struct {
 	ctx       context.Context // closed via cancelCtx in Client.Close
 	cancelCtx context.CancelFunc
 
-	mu      sync.Mutex
+	mu        sync.Mutex
-	closed  bool
+	preferred bool
-	netConn io.Closer
+	closed    bool
-	client  *derp.Client
+	netConn   io.Closer
+	client    *derp.Client
 }
 
 // NewClient returns a new DERP-over-HTTP client. It connects lazily.
@@ -223,6 +224,12 @@ func (c *Client) connect(ctx context.Context, caller string) (client *derp.Clien
 	if err != nil {
 		return nil, err
 	}
+	if c.preferred {
+		if err := derpClient.NotePreferred(true); err != nil {
+			go httpConn.Close()
+			return nil, err
+		}
+	}
 
 	c.client = derpClient
 	c.netConn = tcpConn
@@ -240,6 +247,25 @@ func (c *Client) Send(dstKey key.Public, b []byte) error {
 	return err
 }
 
+// NotePreferred notes whether this Client is the caller's preferred
+// (home) DERP node. It's only used for stats.
+func (c *Client) NotePreferred(v bool) {
+	c.mu.Lock()
+	if c.preferred == v {
+		c.mu.Unlock()
+		return
+	}
+	c.preferred = v
+	client := c.client
+	c.mu.Unlock()
+
+	if client != nil {
+		if err := client.NotePreferred(v); err != nil {
+			c.closeForReconnect()
+		}
+	}
+}
+
 func (c *Client) Recv(b []byte) (derp.ReceivedMessage, error) {
 	client, err := c.connect(context.TODO(), "derphttp.Client.Recv")
 	if err != nil {

wgengine/magicsock/magicsock.go

@@ -348,11 +348,18 @@ func (c *Conn) setNearestDERP(derpNum int) (wantDERP bool) {
 		c.myDerp = 0
 		return false
 	}
+	if derpNum == c.myDerp {
+		// No change.
+		return true
+	}
+	c.myDerp = derpNum
+	for i, ad := range c.activeDerp {
+		go ad.c.NotePreferred(i == c.myDerp)
+	}
 	if derpNum != 0 && derpNum != c.myDerp {
 		// On change, start connecting to it:
 		go c.derpWriteChanOfAddr(&net.UDPAddr{IP: derpMagicIP, Port: derpNum})
 	}
-	c.myDerp = derpNum
 	return true
 }
 
@@ -641,6 +648,7 @@ func (c *Conn) derpWriteChanOfAddr(addr *net.UDPAddr) chan<- derpWriteRequest {
 			c.logf("derphttp.NewClient: port %d, host %q invalid? err: %v", addr.Port, host, err)
 			return nil
 		}
+		dc.NotePreferred(c.myDerp == addr.Port)
 		dc.DNSCache = dnscache.Get()
 		dc.TLSConfig = c.derpTLSConfig