Skip to content

Commit

Permalink
Prepare notify channel before sending first update (#1730)
Browse files Browse the repository at this point in the history
* create channel before sending first update

Signed-off-by: Kristoffer Dalby <kristoffer@tailscale.com>

* do not notify on register, wait for connect

Signed-off-by: Kristoffer Dalby <kristoffer@tailscale.com>

---------

Signed-off-by: Kristoffer Dalby <kristoffer@tailscale.com>
  • Loading branch information
kradalby committed Feb 12, 2024
1 parent c3257e2 commit 68a8ece
Show file tree
Hide file tree
Showing 3 changed files with 30 additions and 45 deletions.
27 changes: 8 additions & 19 deletions hscontrol/auth.go
Original file line number Diff line number Diff line change
Expand Up @@ -311,9 +311,6 @@ func (h *Headscale) handleAuthKey(

nodeKey := registerRequest.NodeKey

var update types.StateUpdate
var mkey key.MachinePublic

// retrieve node information if it exist
// The error is not important, because if it does not
// exist, then this is a new node and we will move
Expand All @@ -338,9 +335,6 @@ func (h *Headscale) handleAuthKey(
return
}

mkey = node.MachineKey
update = types.StateUpdateExpire(node.ID, registerRequest.Expiry)

aclTags := pak.Proto().GetAclTags()
if len(aclTags) > 0 {
// This conditional preserves the existing behaviour, although SaaS would reset the tags on auth-key login
Expand All @@ -357,6 +351,14 @@ func (h *Headscale) handleAuthKey(
return
}
}

mkey := node.MachineKey
update := types.StateUpdateExpire(node.ID, registerRequest.Expiry)

if update.Valid() {
ctx := types.NotifyCtx(context.Background(), "handle-authkey", "na")
h.nodeNotifier.NotifyWithIgnore(ctx, update, mkey.String())
}
} else {
now := time.Now().UTC()

Expand Down Expand Up @@ -400,13 +402,6 @@ func (h *Headscale) handleAuthKey(

return
}

mkey = node.MachineKey
update = types.StateUpdate{
Type: types.StatePeerChanged,
ChangeNodes: types.Nodes{node},
Message: "called from auth.handleAuthKey",
}
}

err = h.db.DB.Transaction(func(tx *gorm.DB) error {
Expand Down Expand Up @@ -456,12 +451,6 @@ func (h *Headscale) handleAuthKey(
return
}

// TODO(kradalby): if notifying after register make sense.
if update.Valid() {
ctx := types.NotifyCtx(context.Background(), "handle-authkey", "na")
h.nodeNotifier.NotifyWithIgnore(ctx, update, mkey.String())
}

log.Info().
Str("node", registerRequest.Hostinfo.Hostname).
Str("ips", strings.Join(node.IPAddresses.StringSlice(), ", ")).
Expand Down
10 changes: 0 additions & 10 deletions hscontrol/grpcv1.go
Original file line number Diff line number Diff line change
Expand Up @@ -200,16 +200,6 @@ func (api headscaleV1APIServer) RegisterNode(
return nil, err
}

stateUpdate := types.StateUpdate{
Type: types.StatePeerChanged,
ChangeNodes: types.Nodes{node},
Message: "called from api.RegisterNode",
}
if stateUpdate.Valid() {
ctx := types.NotifyCtx(ctx, "cli-registernode", node.Hostname)
api.h.nodeNotifier.NotifyWithIgnore(ctx, stateUpdate, node.MachineKey.String())
}

return &v1.RegisterNodeResponse{Node: node.Proto()}, nil
}

Expand Down
38 changes: 22 additions & 16 deletions hscontrol/poll.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"github.com/rs/zerolog/log"
xslices "golang.org/x/exp/slices"
"gorm.io/gorm"
"tailscale.com/envknob"
"tailscale.com/tailcfg"
)

Expand Down Expand Up @@ -277,6 +278,25 @@ func (h *Headscale) handlePoll(
return
}

// Set up the client stream
h.pollNetMapStreamWG.Add(1)
defer h.pollNetMapStreamWG.Done()

// Use a buffered channel in case a node is not fully ready
// to receive a message to make sure we dont block the entire
// notifier.
// 12 is arbitrarily chosen.
chanSize := 3
if size, ok := envknob.LookupInt("HEADSCALE_TUNING_POLL_QUEUE_SIZE"); ok {
chanSize = size
}
updateChan := make(chan types.StateUpdate, chanSize)
defer closeChanWithLog(updateChan, node.Hostname, "updateChan")

// Register the node's update channel
h.nodeNotifier.AddNode(node.MachineKey, updateChan)
defer h.nodeNotifier.RemoveNode(node.MachineKey)

// When a node connects to control, list the peers it has at
// that given point, further updates are kept in memory in
// the Mapper, which lives for the duration of the polling
Expand All @@ -289,8 +309,9 @@ func (h *Headscale) handlePoll(
return
}

isConnected := h.nodeNotifier.ConnectedMap()
for _, peer := range peers {
online := h.nodeNotifier.IsConnected(peer.MachineKey)
online := isConnected[peer.MachineKey]
peer.IsOnline = &online
}

Expand Down Expand Up @@ -357,21 +378,6 @@ func (h *Headscale) handlePoll(
go h.pollFailoverRoutes(logErr, "new node", node)
}

// Set up the client stream
h.pollNetMapStreamWG.Add(1)
defer h.pollNetMapStreamWG.Done()

// Use a buffered channel in case a node is not fully ready
// to receive a message to make sure we dont block the entire
// notifier.
// 12 is arbitrarily chosen.
updateChan := make(chan types.StateUpdate, 12)
defer closeChanWithLog(updateChan, node.Hostname, "updateChan")

// Register the node's update channel
h.nodeNotifier.AddNode(node.MachineKey, updateChan)
defer h.nodeNotifier.RemoveNode(node.MachineKey)

keepAliveTicker := time.NewTicker(keepAliveInterval)

ctx, cancel := context.WithCancel(context.WithValue(ctx, nodeNameContextKey, node.Hostname))
Expand Down

0 comments on commit 68a8ece

Please sign in to comment.