monitor: refactor for architecture-specific connection implementations

Signed-off-by: wardn <wardn@users.noreply.github.com>

wgengine/monitor/monitor.go

@@ -0,0 +1,106 @@
+// Copyright (c) 2020 Tailscale Inc & AUTHORS All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package monitor provides facilities for monitoring network
+// interface changes.
+package monitor
+
+import (
+	"time"
+
+	"tailscale.com/logger"
+)
+
+// Message represents a message returned from a connection.
+// TODO(]|[): currently messages are being discarded, so the
+// properties of the message haven't been defined.
+type Message interface{}
+
+// Conn represents the connection that is being monitored.
+type Conn interface {
+	Close() error
+	Receive() (Message, error)
+}
+
+// ChangeFunc is a callback function that's called when
+// an interface status changes.
+type ChangeFunc func()
+
+// Mon represents a monitoring instance.
+type Mon struct {
+	logf   logger.Logf
+	cb     ChangeFunc
+	conn   Conn
+	change chan struct{}
+	stop   chan struct{}
+}
+
+// New instantiates and starts a monitoring instance. Change notifications
+// are propagated to the callback function.
+func New(logf logger.Logf, callback ChangeFunc) (*Mon, error) {
+	conn, err := NewConn()
+	if err != nil {
+		return nil, err
+	}
+	ret := &Mon{
+		logf:   logf,
+		cb:     callback,
+		conn:   conn,
+		change: make(chan struct{}, 1),
+		stop:   make(chan struct{}),
+	}
+	go ret.pump()
+	go ret.debounce()
+	return ret, nil
+}
+
+// Close is used to close the underlying connection.
+func (m *Mon) Close() error {
+	close(m.stop)
+	return m.conn.Close()
+}
+
+// pump continuously retrieves messages from the connection, notifying
+// the change channel of changes, and stopping when a stop is issued.
+func (m *Mon) pump() {
+	for {
+		_, err := m.conn.Receive()
+		if err != nil {
+			select {
+			case <-m.stop:
+				return
+			default:
+			}
+			// Keep retrying while we're not closed.
+			m.logf("Error receiving from connection: %v", err)
+			time.Sleep(time.Second)
+			continue
+		}
+
+		select {
+		case m.change <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// debounce calls the callback function with a delay between events
+// and exits when a stop is issued.
+func (m *Mon) debounce() {
+	for {
+		select {
+		case <-m.stop:
+			return
+		case <-m.change:
+		}
+
+		m.cb()
+
+		select {
+		case <-m.stop:
+			return
+		case <-time.After(100 * time.Millisecond):
+		}
+	}
+}

wgengine/monitor/monitor_freebsd.go

@@ -0,0 +1,47 @@
+// Copyright (c) 2020 Tailscale Inc & AUTHORS All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package monitor
+
+import (
+	"bufio"
+	"fmt"
+	"net"
+	"strings"
+)
+
+type devdConn struct {
+	conn net.Conn
+}
+
+func NewConn() (Conn, error) {
+	conn, err := net.Dial("unixpacket", "/var/run/devd.seqpacket.pipe")
+	if err != nil {
+		return nil, fmt.Errorf("devd dial error: %v", err)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("dialing devd socket: %v", err)
+	}
+	return &devdConn{conn}, nil
+}
+
+func (c *devdConn) Close() error {
+	return c.conn.Close()
+}
+
+func (c *devdConn) Receive() (Message, error) {
+	for {
+		msg, err := bufio.NewReader(c.conn).ReadString('\n')
+		if err != nil {
+			return nil, fmt.Errorf("reading devd socket: %v", err)
+		}
+		// Only return messages related to the network subsystem.
+		if !strings.Contains(msg, "system=IFNET") {
+			continue
+		}
+		// TODO(]|[): this is where the devd-specific message would
+		// get converted into a "standard" event message and returned.
+		return nil, nil
+	}
+}

wgengine/monitor/monitor_linux.go

@@ -0,0 +1,60 @@
+// Copyright (c) 2020 Tailscale Inc & AUTHORS All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package monitor
+
+import (
+	"fmt"
+
+	"github.com/mdlayher/netlink"
+	"golang.org/x/sys/unix"
+)
+
+const (
+	RTMGRP_IPV4_IFADDR = 0x10
+	RTMGRP_IPV4_ROUTE  = 0x40
+)
+
+// nlConn wraps a *netlink.Conn and returns a monitor.Message
+// instead of a netlink.Message. Currently, messages are discarded,
+// but down the line, when messages trigger different logic depending
+// on the type of event, this provides the capability of handling
+// each architecture-specific message in a generic fashion.
+type nlConn struct {
+	conn *netlink.Conn
+}
+
+func NewConn() (Conn, error) {
+	conn, err := netlink.Dial(unix.NETLINK_ROUTE, &netlink.Config{
+		// IPv4 address and route changes. Routes get us most of the
+		// events of interest, but we need address as well to cover
+		// things like DHCP deciding to give us a new address upon
+		// renewal - routing wouldn't change, but all reachability
+		// would.
+		//
+		// Why magic numbers? These aren't exposed in x/sys/unix
+		// yet. The values come from rtnetlink.h, RTMGRP_IPV4_IFADDR
+		// and RTMGRP_IPV4_ROUTE.
+		Groups: RTMGRP_IPV4_IFADDR | RTMGRP_IPV4_ROUTE,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("dialing netlink socket: %v", err)
+	}
+	return &nlConn{conn}, nil
+}
+
+func (c *nlConn) Close() error {
+	return c.conn.Close()
+}
+
+func (c *nlConn) Receive() (Message, error) {
+	// currently ignoring the message
+	_, err := c.conn.Receive()
+	if err != nil {
+		return nil, err
+	}
+	// TODO(]|[): this is where the NetLink-specific message would
+	// get converted into a "standard" event message and returned.
+	return nil, nil
+}

wgengine/router_linux.go

@@ -20,20 +20,20 @@ import (
 	"github.com/tailscale/wireguard-go/wgcfg"
 	"tailscale.com/atomicfile"
 	"tailscale.com/logger"
-	"tailscale.com/wgengine/rtnlmon"
+	"tailscale.com/wgengine/monitor"
 )
 
 type linuxRouter struct {
 	logf       func(fmt string, args ...interface{})
 	tunname    string
-	mon        *rtnlmon.Mon
+	mon        *monitor.Mon
 	netChanged func()
 	local      wgcfg.CIDR
 	routes     map[wgcfg.CIDR]struct{}
 }
 
 func NewUserspaceRouter(logf logger.Logf, tunname string, dev *device.Device, tuntap tun.Device, netChanged func()) Router {
-	mon, err := rtnlmon.New(logf, netChanged)
+	mon, err := monitor.New(logf, netChanged)
 	if err != nil {
 		log.Fatalf("rtnlmon.New() failed: %v", err)
 	}

wgengine/rtnlmon/mon.go

@@ -1,116 +0,0 @@
-// Copyright (c) 2020 Tailscale Inc & AUTHORS All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-// +build !windows
-
-// Package rtnlmon watches for "interesting" changes to the network
-// stack and fires a callback.
-package rtnlmon
-
-import (
-	"fmt"
-	"time"
-
-	"github.com/mdlayher/netlink"
-	"golang.org/x/sys/unix"
-	"tailscale.com/logger"
-)
-
-// Netlink is not a great protocol for *knowing* things. The protocol
-// design makes it impossible to track changes precisely. You can see
-// this by looking at things like Quagga or Bird, which all include
-// keeping a local impression of what they think is in the kernel, and
-// periodically doing a full state dump to find errors. They do use
-// events, but explicitly only as an optimization, because they can't
-// be trusted.
-//
-// Fortunately, we don't really need to know what exactly changed. We
-// just want to know that network conditions may have changed, and we
-// should re-explore connectivity. This is why we subscribe to events,
-// and then blindly fire our callback without looking at the content
-// of the notifications.
-
-type ChangeFunc func()
-
-type Mon struct {
-	logf   logger.Logf
-	cb     ChangeFunc
-	nl     *netlink.Conn
-	change chan struct{}
-	stop   chan struct{}
-}
-
-func New(logf logger.Logf, callback ChangeFunc) (*Mon, error) {
-	conn, err := netlink.Dial(unix.NETLINK_ROUTE, &netlink.Config{
-		// IPv4 address and route changes. Routes get us most of the
-		// events of interest, but we need address as well to cover
-		// things like DHCP deciding to give us a new address upon
-		// renewal - routing wouldn't change, but all reachability
-		// would.
-		//
-		// Why magic numbers? These aren't exposed in x/sys/unix
-		// yet. The values come from rtnetlink.h, RTMGRP_IPV4_IFADDR
-		// and RTMGRP_IPV4_ROUTE.
-		Groups: 0x10 | 0x40,
-	})
-	if err != nil {
-		return nil, fmt.Errorf("dialing netlink socket: %v", err)
-	}
-
-	ret := &Mon{
-		logf:   logf,
-		cb:     callback,
-		nl:     conn,
-		change: make(chan struct{}, 1),
-		stop:   make(chan struct{}),
-	}
-	go ret.pump()
-	go ret.debounce()
-	return ret, nil
-}
-
-func (m *Mon) Close() error {
-	close(m.stop)
-	return m.nl.Close()
-}
-
-func (m *Mon) pump() {
-	for {
-		_, err := m.nl.Receive()
-		if err != nil {
-			select {
-			case <-m.stop:
-				return
-			default:
-			}
-			// Keep retrying while we're not closed.
-			m.logf("Error receiving from netlink: %v", err)
-			time.Sleep(time.Second)
-			continue
-		}
-
-		select {
-		case m.change <- struct{}{}:
-		default:
-		}
-	}
-}
-
-func (m *Mon) debounce() {
-	for {
-		select {
-		case <-m.stop:
-			return
-		case <-m.change:
-		}
-
-		m.cb()
-
-		select {
-		case <-m.stop:
-			return
-		case <-time.After(100 * time.Millisecond):
-		}
-	}
-}