Skip to content

Commit 01c52c7

Browse files
kylecarbskylecarbs
committed
Add coordinator
1 parent d36f61d commit 01c52c7

File tree

9 files changed

+338
-73
lines changed

9 files changed

+338
-73
lines changed

‎.vscode/settings.json

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,9 @@
8181
"Srcs",
8282
"stretchr",
8383
"stuntest",
84+
"tailbroker",
8485
"tailcfg",
86+
"tailexchange",
8587
"tailnet",
8688
"Tailscale",
8789
"TCGETS",
@@ -118,6 +120,7 @@
118120
"workspacebuilds",
119121
"workspacename",
120122
"wsconncache",
123+
"wsjson",
121124
"xerrors",
122125
"xstate",
123126
"yamux"

‎agent/agent.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,7 @@ type agent struct {
126126
sshServer *ssh.Server
127127

128128
enableTailnet bool
129-
network *tailnet.Server
129+
network *tailnet.Conn
130130
nodeDialer NodeDialer
131131
}
132132

@@ -180,7 +180,7 @@ func (a *agent) runTailnet(ctx context.Context, addresses []netip.Addr, derpMap
180180
ipRanges = append(ipRanges, netip.PrefixFrom(address, 128))
181181
}
182182
var err error
183-
a.network, err = tailnet.New(&tailnet.Options{
183+
a.network, err = tailnet.NewConn(&tailnet.Options{
184184
Addresses: ipRanges,
185185
DERPMap: derpMap,
186186
Logger: a.logger.Named("tailnet"),

‎agent/conn.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -136,7 +136,7 @@ func (c *WebRTCConn) Close() error {
136136

137137
type TailnetConn struct {
138138
Target netip.Addr
139-
*tailnet.Server
139+
*tailnet.Conn
140140
}
141141

142142
func (c *TailnetConn) Closed() <-chan struct{} {

‎coderd/coderd_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -609,14 +609,14 @@ func TestDERP(t *testing.T) {
609609
},
610610
}
611611
w1IP := tailnet.IP()
612-
w1, err := tailnet.New(&tailnet.Options{
612+
w1, err := tailnet.NewConn(&tailnet.Options{
613613
Addresses: []netip.Prefix{netip.PrefixFrom(w1IP, 128)},
614614
Logger: logger.Named("w1"),
615615
DERPMap: derpMap,
616616
})
617617
require.NoError(t, err)
618618

619-
w2, err := tailnet.New(&tailnet.Options{
619+
w2, err := tailnet.NewConn(&tailnet.Options{
620620
Addresses: []netip.Prefix{netip.PrefixFrom(tailnet.IP(), 128)},
621621
Logger: logger.Named("w2"),
622622
DERPMap: derpMap,

‎codersdk/workspaceagents.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -319,7 +319,7 @@ func (c *Client) DialWorkspaceAgentTailnet(ctx context.Context, agentID uuid.UUI
319319
}
320320

321321
ip := tailnet.IP()
322-
server, err := tailnet.New(&tailnet.Options{
322+
server, err := tailnet.NewConn(&tailnet.Options{
323323
Addresses: []netip.Prefix{netip.PrefixFrom(ip, 128)},
324324
DERPMap: &derpMap,
325325
Logger: logger,

‎tailnet/tailnet.go renamed to ‎tailnet/conn.go

Lines changed: 56 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -49,8 +49,8 @@ type Options struct {
4949
Logger slog.Logger
5050
}
5151

52-
// New constructs a new Wireguard server that will accept connections from the addresses provided.
53-
func New(options *Options) (*Server, error) {
52+
// NewConn constructs a new Wireguard server that will accept connections from the addresses provided.
53+
func NewConn(options *Options) (*Conn, error) {
5454
if options == nil {
5555
options = &Options{}
5656
}
@@ -186,7 +186,7 @@ func New(options *Options) (*Server, error) {
186186
logIPSet := netipx.IPSetBuilder{}
187187
logIPs, _ := logIPSet.IPSet()
188188
wireguardEngine.SetFilter(filter.New(netMap.PacketFilter, localIPs, logIPs, nil, Logger(options.Logger.Named("packet-filter"))))
189-
server := &Server{
189+
server := &Conn{
190190
logger: options.Logger,
191191
magicConn: magicConn,
192192
dialer: dialer,
@@ -218,8 +218,8 @@ func IP() netip.Addr {
218218
return netip.AddrFrom16(uid)
219219
}
220220

221-
// Server is an actively listening Wireguard connection.
222-
type Server struct {
221+
// Conn is an actively listening Wireguard connection.
222+
type Conn struct {
223223
mutex sync.Mutex
224224
logger slog.Logger
225225

@@ -237,15 +237,15 @@ type Server struct {
237237
// SetNodeCallback is triggered when a network change occurs and peer
238238
// renegotiation may be required. Clients should constantly be emitting
239239
// node changes.
240-
func (s *Server) SetNodeCallback(callback func(node *Node)) {
241-
s.magicConn.SetNetInfoCallback(func(ni *tailcfg.NetInfo) {
242-
s.logger.Info(context.Background(), "latency", slog.F("latency", ni.DERPLatency))
240+
func (c *Conn) SetNodeCallback(callback func(node *Node)) {
241+
c.magicConn.SetNetInfoCallback(func(ni *tailcfg.NetInfo) {
242+
c.logger.Info(context.Background(), "latency", slog.F("latency", ni.DERPLatency))
243243
callback(&Node{
244-
ID: s.netMap.SelfNode.ID,
245-
Key: s.netMap.SelfNode.Key,
246-
Addresses: s.netMap.SelfNode.Addresses,
247-
AllowedIPs: s.netMap.SelfNode.AllowedIPs,
248-
DiscoKey: s.magicConn.DiscoPublicKey(),
244+
ID: c.netMap.SelfNode.ID,
245+
Key: c.netMap.SelfNode.Key,
246+
Addresses: c.netMap.SelfNode.Addresses,
247+
AllowedIPs: c.netMap.SelfNode.AllowedIPs,
248+
DiscoKey: c.magicConn.DiscoPublicKey(),
249249
PreferredDERP: ni.PreferredDERP,
250250
DERPLatency: ni.DERPLatency,
251251
})
@@ -254,11 +254,11 @@ func (s *Server) SetNodeCallback(callback func(node *Node)) {
254254

255255
// UpdateNodes connects with a set of peers. This can be constantly updated,
256256
// and peers will continually be reconnected as necessary.
257-
func (s *Server) UpdateNodes(nodes []*Node) error {
258-
s.mutex.Lock()
259-
defer s.mutex.Unlock()
257+
func (c *Conn) UpdateNodes(nodes []*Node) error {
258+
c.mutex.Lock()
259+
defer c.mutex.Unlock()
260260
peerMap := map[tailcfg.NodeID]*tailcfg.Node{}
261-
for _, peer := range s.netMap.Peers {
261+
for _, peer := range c.netMap.Peers {
262262
peerMap[peer.ID] = peer
263263
}
264264
for _, node := range nodes {
@@ -272,41 +272,41 @@ func (s *Server) UpdateNodes(nodes []*Node) error {
272272
Hostinfo: hostinfo.New().View(),
273273
}
274274
}
275-
s.netMap.Peers = make([]*tailcfg.Node, 0, len(peerMap))
275+
c.netMap.Peers = make([]*tailcfg.Node, 0, len(peerMap))
276276
for _, peer := range peerMap {
277-
s.netMap.Peers = append(s.netMap.Peers, peer)
277+
c.netMap.Peers = append(c.netMap.Peers, peer)
278278
}
279-
cfg, err := nmcfg.WGCfg(s.netMap, Logger(s.logger.Named("wgconfig")), netmap.AllowSingleHosts, "")
279+
cfg, err := nmcfg.WGCfg(c.netMap, Logger(c.logger.Named("wgconfig")), netmap.AllowSingleHosts, "")
280280
if err != nil {
281281
return xerrors.Errorf("update wireguard config: %w", err)
282282
}
283-
err = s.wireguardEngine.Reconfig(cfg, s.wireguardRouter, &dns.Config{}, &tailcfg.Debug{})
283+
err = c.wireguardEngine.Reconfig(cfg, c.wireguardRouter, &dns.Config{}, &tailcfg.Debug{})
284284
if err != nil {
285285
return xerrors.Errorf("reconfig: %w", err)
286286
}
287-
netMapCopy := *s.netMap
288-
s.wireguardEngine.SetNetworkMap(&netMapCopy)
287+
netMapCopy := *c.netMap
288+
c.wireguardEngine.SetNetworkMap(&netMapCopy)
289289
return nil
290290
}
291291

292292
// Ping sends a ping to the Wireguard engine.
293-
func (s *Server) Ping(ip netip.Addr, pingType tailcfg.PingType, cb func(*ipnstate.PingResult)) {
294-
s.wireguardEngine.Ping(ip, pingType, cb)
293+
func (c *Conn) Ping(ip netip.Addr, pingType tailcfg.PingType, cb func(*ipnstate.PingResult)) {
294+
c.wireguardEngine.Ping(ip, pingType, cb)
295295
}
296296

297297
// Close shuts down the Wireguard connection.
298-
func (s *Server) Close() error {
299-
s.mutex.Lock()
300-
defer s.mutex.Unlock()
301-
for _, l := range s.listeners {
298+
func (c *Conn) Close() error {
299+
c.mutex.Lock()
300+
defer c.mutex.Unlock()
301+
for _, l := range c.listeners {
302302
_ = l.Close()
303303
}
304-
_ = s.dialer.Close()
305-
_ = s.magicConn.Close()
306-
_ = s.netStack.Close()
307-
_ = s.wireguardMonitor.Close()
308-
_ = s.tunDevice.Close()
309-
s.wireguardEngine.Close()
304+
_ = c.dialer.Close()
305+
_ = c.magicConn.Close()
306+
_ = c.netStack.Close()
307+
_ = c.wireguardMonitor.Close()
308+
_ = c.tunDevice.Close()
309+
c.wireguardEngine.Close()
310310
return nil
311311
}
312312

@@ -326,54 +326,54 @@ type Node struct {
326326

327327
// Listen announces only on the Tailscale network.
328328
// It will start the server if it has not been started yet.
329-
func (s *Server) Listen(network, addr string) (net.Listener, error) {
329+
func (c *Conn) Listen(network, addr string) (net.Listener, error) {
330330
host, port, err := net.SplitHostPort(addr)
331331
if err != nil {
332332
return nil, xerrors.Errorf("wgnet: %w", err)
333333
}
334334
lk := listenKey{network, host, port}
335335
ln := &listener{
336-
s: s,
336+
s: c,
337337
key: lk,
338338
addr: addr,
339339

340340
conn: make(chan net.Conn),
341341
}
342-
s.mutex.Lock()
343-
if s.listeners == nil {
344-
s.listeners = map[listenKey]*listener{}
342+
c.mutex.Lock()
343+
if c.listeners == nil {
344+
c.listeners = map[listenKey]*listener{}
345345
}
346-
if _, ok := s.listeners[lk]; ok {
347-
s.mutex.Unlock()
346+
if _, ok := c.listeners[lk]; ok {
347+
c.mutex.Unlock()
348348
return nil, xerrors.Errorf("wgnet: listener already open for %s, %s", network, addr)
349349
}
350-
s.listeners[lk] = ln
351-
s.mutex.Unlock()
350+
c.listeners[lk] = ln
351+
c.mutex.Unlock()
352352
return ln, nil
353353
}
354354

355-
func (s *Server) DialContextTCP(ctx context.Context, ipp netip.AddrPort) (*gonet.TCPConn, error) {
356-
return s.netStack.DialContextTCP(ctx, ipp)
355+
func (c *Conn) DialContextTCP(ctx context.Context, ipp netip.AddrPort) (*gonet.TCPConn, error) {
356+
return c.netStack.DialContextTCP(ctx, ipp)
357357
}
358358

359-
func (s *Server) DialContextUDP(ctx context.Context, ipp netip.AddrPort) (*gonet.UDPConn, error) {
360-
return s.netStack.DialContextUDP(ctx, ipp)
359+
func (c *Conn) DialContextUDP(ctx context.Context, ipp netip.AddrPort) (*gonet.UDPConn, error) {
360+
return c.netStack.DialContextUDP(ctx, ipp)
361361
}
362362

363-
func (s *Server) forwardTCP(c net.Conn, port uint16) {
364-
s.mutex.Lock()
365-
ln, ok := s.listeners[listenKey{"tcp", "", fmt.Sprint(port)}]
366-
s.mutex.Unlock()
363+
func (c *Conn) forwardTCP(conn net.Conn, port uint16) {
364+
c.mutex.Lock()
365+
ln, ok := c.listeners[listenKey{"tcp", "", fmt.Sprint(port)}]
366+
c.mutex.Unlock()
367367
if !ok {
368-
_ = c.Close()
368+
_ = conn.Close()
369369
return
370370
}
371371
t := time.NewTimer(time.Second)
372372
defer t.Stop()
373373
select {
374-
case ln.conn <- c:
374+
case ln.conn <- conn:
375375
case <-t.C:
376-
_ = c.Close()
376+
_ = conn.Close()
377377
}
378378
}
379379

@@ -384,7 +384,7 @@ type listenKey struct {
384384
}
385385

386386
type listener struct {
387-
s *Server
387+
s *Conn
388388
key listenKey
389389
addr string
390390
conn chan net.Conn
@@ -420,12 +420,3 @@ func Logger(logger slog.Logger) tslogger.Logf {
420420
logger.Debug(context.Background(), fmt.Sprintf(format, args...))
421421
})
422422
}
423-
424-
// The exchanger is entirely in-memory and works based on connected nodes.
425-
// It uses a PubSub system to dynamically add/remove nodes from the network
426-
// and build a netmap based on connection ID.
427-
//
428-
// Each node is allocated it's own internal connection ID.
429-
//
430-
// The connecting node *just* requires information about the other node.
431-
// The other node needs connection information of all the others.

‎tailnet/tailnet_test.go renamed to ‎tailnet/conn_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,14 +33,14 @@ func TestTailnet(t *testing.T) {
3333
derpMap := runDERPAndStun(t, tailnet.Logger(logger.Named("derp")))
3434

3535
w1IP := tailnet.IP()
36-
w1, err := tailnet.New(&tailnet.Options{
36+
w1, err := tailnet.NewConn(&tailnet.Options{
3737
Addresses: []netip.Prefix{netip.PrefixFrom(w1IP, 128)},
3838
Logger: logger.Named("w1"),
3939
DERPMap: derpMap,
4040
})
4141
require.NoError(t, err)
4242

43-
w2, err := tailnet.New(&tailnet.Options{
43+
w2, err := tailnet.NewConn(&tailnet.Options{
4444
Addresses: []netip.Prefix{netip.PrefixFrom(tailnet.IP(), 128)},
4545
Logger: logger.Named("w2"),
4646
DERPMap: derpMap,

0 commit comments

Comments
 (0)