headscale/hscontrol/poll.go

726 lines
19 KiB
Go
Raw Normal View History

package hscontrol
import (
"context"
"fmt"
"net/http"
"time"
"github.com/juanfont/headscale/hscontrol/mapper"
"github.com/juanfont/headscale/hscontrol/types"
"github.com/juanfont/headscale/hscontrol/util"
"github.com/rs/zerolog/log"
"tailscale.com/tailcfg"
)
const (
keepAliveInterval = 60 * time.Second
)
2022-05-16 06:59:46 -06:00
type contextKey string
const machineNameContextKey = contextKey("machineName")
// handlePoll is the common code for the legacy and Noise protocols to
2022-08-15 02:43:39 -06:00
// managed the poll loop.
func (h *Headscale) handlePoll(
2022-06-26 04:06:25 -06:00
writer http.ResponseWriter,
ctx context.Context,
machine *types.Machine,
mapRequest tailcfg.MapRequest,
isNoise bool,
2022-06-20 04:30:51 -06:00
) {
mapp := mapper.NewMapper(
h.db,
h.privateKey2019,
isNoise,
h.DERPMap,
h.cfg.BaseDomain,
h.cfg.DNSConfig,
h.cfg.LogTail.Enabled,
h.cfg.RandomizeClientPort,
)
2022-06-26 04:06:25 -06:00
machine.Hostname = mapRequest.Hostinfo.Hostname
machine.HostInfo = types.HostInfo(*mapRequest.Hostinfo)
machine.DiscoKey = util.DiscoPublicKeyStripPrefix(mapRequest.DiscoKey)
now := time.Now().UTC()
err := h.db.ProcessMachineRoutes(machine)
if err != nil {
log.Error().
Caller().
Err(err).
Str("machine", machine.Hostname).
Msg("Error processing machine routes")
}
// update ACLRules with peer informations (to update server tags if necessary)
if h.ACLPolicy != nil {
// update routes with peer information
err = h.db.EnableAutoApprovedRoutes(h.ACLPolicy, machine)
if err != nil {
log.Error().
Caller().
Bool("noise", isNoise).
Str("machine", machine.Hostname).
Err(err).
Msg("Error running auto approved routes")
}
}
// From Tailscale client:
//
// ReadOnly is whether the client just wants to fetch the MapResponse,
// without updating their Endpoints. The Endpoints field will be ignored and
// LastSeen will not be updated and peers will not be notified of changes.
//
// The intended use is for clients to discover the DERP map at start-up
// before their first real endpoint update.
2022-06-26 04:06:25 -06:00
if !mapRequest.ReadOnly {
machine.Endpoints = mapRequest.Endpoints
2021-11-15 09:15:50 -07:00
machine.LastSeen = &now
}
2022-05-30 07:39:24 -06:00
if err := h.db.MachineSave(machine); err != nil {
log.Error().
Str("handler", "PollNetMap").
Bool("noise", isNoise).
Str("node_key", machine.NodeKey).
Str("machine", machine.Hostname).
Err(err).
Msg("Failed to persist/update machine in the database")
http.Error(writer, "", http.StatusInternalServerError)
2022-05-30 07:39:24 -06:00
return
2022-05-30 07:39:24 -06:00
}
mapResp, err := mapp.CreateMapResponse(mapRequest, machine, h.ACLPolicy)
if err != nil {
log.Error().
Str("handler", "PollNetMap").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
Str("node_key", machine.NodeKey).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Err(err).
Msg("Failed to get Map response")
2022-06-26 04:06:25 -06:00
http.Error(writer, "", http.StatusInternalServerError)
2021-11-14 08:46:09 -07:00
return
}
// We update our peers if the client is not sending ReadOnly in the MapRequest
// so we don't distribute its initial request (it comes with
// empty endpoints to peers)
// Details on the protocol can be found in https://github.com/tailscale/tailscale/blob/main/tailcfg/tailcfg.go#L696
log.Debug().
Str("handler", "PollNetMap").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
2022-06-26 04:06:25 -06:00
Bool("readOnly", mapRequest.ReadOnly).
Bool("omitPeers", mapRequest.OmitPeers).
Bool("stream", mapRequest.Stream).
Msg("Client map request processed")
2022-06-26 04:06:25 -06:00
if mapRequest.ReadOnly {
log.Info().
Str("handler", "PollNetMap").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Msg("Client is starting up. Probably interested in a DERP map")
2022-06-20 04:30:51 -06:00
2022-06-26 04:06:25 -06:00
writer.Header().Set("Content-Type", "application/json; charset=utf-8")
writer.WriteHeader(http.StatusOK)
_, err := writer.Write(mapResp)
2022-06-26 04:21:35 -06:00
if err != nil {
log.Error().
Caller().
Err(err).
Msg("Failed to write response")
}
2021-11-14 08:46:09 -07:00
if f, ok := writer.(http.Flusher); ok {
f.Flush()
}
return
}
// There has been an update to _any_ of the nodes that the other nodes would
// need to know about
2022-08-16 05:39:15 -06:00
h.setLastStateChangeToNow()
// The request is not ReadOnly, so we need to set up channels for updating
// peers via longpoll
// Only create update channel if it has not been created
log.Trace().
Caller().
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Msg("Loading or creating update channel")
const chanSize = 8
updateChan := make(chan struct{}, chanSize)
pollDataChan := make(chan []byte, chanSize)
2022-04-24 13:55:54 -06:00
defer closeChanWithLog(pollDataChan, machine.Hostname, "pollDataChan")
keepAliveChan := make(chan []byte)
2022-06-26 04:06:25 -06:00
if mapRequest.OmitPeers && !mapRequest.Stream {
log.Info().
Str("handler", "PollNetMap").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Msg("Client sent endpoint update and is ok with a response without peer list")
2022-06-26 04:06:25 -06:00
writer.Header().Set("Content-Type", "application/json; charset=utf-8")
writer.WriteHeader(http.StatusOK)
_, err := writer.Write(mapResp)
2022-06-26 04:21:35 -06:00
if err != nil {
log.Error().
Caller().
Err(err).
Msg("Failed to write response")
}
// It sounds like we should update the nodes when we have received a endpoint update
// even tho the comments in the tailscale code dont explicitly say so.
updateRequestsFromNode.WithLabelValues(machine.User.Name, machine.Hostname, "endpoint-update").
2021-11-13 01:36:45 -07:00
Inc()
updateChan <- struct{}{}
2021-11-14 08:46:09 -07:00
return
2022-06-26 04:06:25 -06:00
} else if mapRequest.OmitPeers && mapRequest.Stream {
log.Warn().
Str("handler", "PollNetMap").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Msg("Ignoring request, don't know how to handle it")
2022-06-26 04:06:25 -06:00
http.Error(writer, "", http.StatusBadRequest)
2021-11-14 08:46:09 -07:00
return
}
log.Info().
Str("handler", "PollNetMap").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Msg("Client is ready to access the tailnet")
log.Info().
Str("handler", "PollNetMap").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Msg("Sending initial map")
pollDataChan <- mapResp
log.Info().
Str("handler", "PollNetMap").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Msg("Notifying peers")
updateRequestsFromNode.WithLabelValues(machine.User.Name, machine.Hostname, "full-update").
2021-11-13 01:36:45 -07:00
Inc()
updateChan <- struct{}{}
h.pollNetMapStream(
2022-06-26 04:06:25 -06:00
writer,
ctx,
2022-06-26 04:06:25 -06:00
machine,
mapRequest,
2021-11-13 01:36:45 -07:00
pollDataChan,
keepAliveChan,
updateChan,
isNoise,
2021-11-13 01:36:45 -07:00
)
log.Trace().
Str("handler", "PollNetMap").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Msg("Finished stream, closing PollNetMap session")
}
// pollNetMapStream stream logic for /machine/map,
// ensuring we communicate updates and data to the connected clients.
func (h *Headscale) pollNetMapStream(
2022-06-26 04:06:25 -06:00
writer http.ResponseWriter,
ctxReq context.Context,
machine *types.Machine,
2021-11-15 09:15:50 -07:00
mapRequest tailcfg.MapRequest,
pollDataChan chan []byte,
keepAliveChan chan []byte,
updateChan chan struct{},
isNoise bool,
) {
// TODO(kradalby): This is a stepping stone, mapper should be initiated once
// per client or something similar
mapp := mapper.NewMapper(h.db,
h.privateKey2019,
isNoise,
h.DERPMap,
h.cfg.BaseDomain,
h.cfg.DNSConfig,
h.cfg.LogTail.Enabled,
h.cfg.RandomizeClientPort,
)
2022-07-11 12:33:24 -06:00
h.pollNetMapStreamWG.Add(1)
defer h.pollNetMapStreamWG.Done()
2022-06-30 15:35:22 -06:00
ctx := context.WithValue(ctxReq, machineNameContextKey, machine.Hostname)
2022-06-20 04:30:51 -06:00
ctx, cancel := context.WithCancel(ctx)
defer cancel()
2022-06-20 04:30:51 -06:00
go h.scheduledPollWorker(
ctx,
updateChan,
keepAliveChan,
mapRequest,
machine,
isNoise,
2022-06-20 04:30:51 -06:00
)
2022-06-20 13:40:28 -06:00
log.Trace().
2022-08-15 02:43:39 -06:00
Str("handler", "pollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-06-20 13:40:28 -06:00
Str("machine", machine.Hostname).
Msg("Waiting for data to stream...")
2022-06-20 13:40:28 -06:00
log.Trace().
2022-08-15 02:43:39 -06:00
Str("handler", "pollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-06-20 13:40:28 -06:00
Str("machine", machine.Hostname).
Msgf("pollData is %#v, keepAliveChan is %#v, updateChan is %#v", pollDataChan, keepAliveChan, updateChan)
2022-06-20 13:40:28 -06:00
for {
select {
case data := <-pollDataChan:
log.Trace().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "pollData").
Int("bytes", len(data)).
Msg("Sending data received via pollData channel")
2022-06-26 04:06:25 -06:00
_, err := writer.Write(data)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "pollData").
Err(err).
Msg("Cannot write data")
2021-11-14 08:46:09 -07:00
2022-06-20 13:40:28 -06:00
return
}
2022-06-26 04:25:26 -06:00
flusher, ok := writer.(http.Flusher)
if !ok {
log.Error().
Caller().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-06-26 04:25:26 -06:00
Str("machine", machine.Hostname).
Str("channel", "pollData").
Msg("Cannot cast writer to http.Flusher")
} else {
flusher.Flush()
}
2022-06-20 13:40:28 -06:00
log.Trace().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "pollData").
Int("bytes", len(data)).
Msg("Data from pollData channel written successfully")
// TODO(kradalby): Abstract away all the database calls, this can cause race conditions
// when an outdated machine object is kept alive, e.g. db is update from
// command line, but then overwritten.
err = h.db.UpdateMachineFromDatabase(machine)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "pollData").
Err(err).
Msg("Cannot update machine from database")
// client has been removed from database
// since the stream opened, terminate connection.
2022-06-20 13:40:28 -06:00
return
}
now := time.Now().UTC()
2021-11-15 09:15:50 -07:00
machine.LastSeen = &now
lastStateUpdate.WithLabelValues(machine.User.Name, machine.Hostname).
2021-11-13 01:36:45 -07:00
Set(float64(now.Unix()))
2021-11-15 09:15:50 -07:00
machine.LastSuccessfulUpdate = &now
err = h.db.TouchMachine(machine)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "pollData").
Err(err).
Msg("Cannot update machine LastSuccessfulUpdate")
2022-06-20 13:40:28 -06:00
return
}
2021-11-14 08:46:09 -07:00
2022-06-20 13:40:28 -06:00
log.Trace().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-06-20 13:40:28 -06:00
Str("machine", machine.Hostname).
Str("channel", "pollData").
Int("bytes", len(data)).
Msg("Machine entry in database updated successfully after sending data")
case data := <-keepAliveChan:
log.Trace().
Str("handler", "PollNetMapStream").
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "keepAlive").
Int("bytes", len(data)).
Msg("Sending keep alive message")
2022-06-26 04:06:25 -06:00
_, err := writer.Write(data)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "keepAlive").
Err(err).
Msg("Cannot write keep alive message")
2021-11-14 08:46:09 -07:00
2022-06-20 13:40:28 -06:00
return
}
2022-06-26 04:25:26 -06:00
flusher, ok := writer.(http.Flusher)
if !ok {
log.Error().
Caller().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-06-26 04:25:26 -06:00
Str("machine", machine.Hostname).
Str("channel", "keepAlive").
Msg("Cannot cast writer to http.Flusher")
} else {
flusher.Flush()
}
2022-06-20 13:40:28 -06:00
log.Trace().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "keepAlive").
Int("bytes", len(data)).
Msg("Keep alive sent successfully")
// TODO(kradalby): Abstract away all the database calls, this can cause race conditions
// when an outdated machine object is kept alive, e.g. db is update from
// command line, but then overwritten.
err = h.db.UpdateMachineFromDatabase(machine)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "keepAlive").
Err(err).
Msg("Cannot update machine from database")
// client has been removed from database
// since the stream opened, terminate connection.
2022-06-20 13:40:28 -06:00
return
}
now := time.Now().UTC()
2021-11-15 09:15:50 -07:00
machine.LastSeen = &now
err = h.db.TouchMachine(machine)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "keepAlive").
Err(err).
Msg("Cannot update machine LastSeen")
2022-06-20 13:40:28 -06:00
return
}
2021-11-14 08:46:09 -07:00
2022-06-20 13:40:28 -06:00
log.Trace().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-06-20 13:40:28 -06:00
Str("machine", machine.Hostname).
Str("channel", "keepAlive").
Int("bytes", len(data)).
Msg("Machine updated successfully after sending keep alive")
case <-updateChan:
log.Trace().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "update").
Msg("Received a request for update")
updateRequestsReceivedOnChannel.WithLabelValues(machine.User.Name, machine.Hostname).
2021-11-13 01:36:45 -07:00
Inc()
2022-06-20 13:40:28 -06:00
if h.db.IsOutdated(machine, h.getLastStateChange()) {
var lastUpdate time.Time
if machine.LastSuccessfulUpdate != nil {
lastUpdate = *machine.LastSuccessfulUpdate
}
log.Debug().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Time("last_successful_update", lastUpdate).
Time("last_state_change", h.getLastStateChange(machine.User)).
2022-04-24 13:55:54 -06:00
Msgf("There has been updates since the last successful update to %s", machine.Hostname)
data, err := mapp.CreateMapResponse(mapRequest, machine, h.ACLPolicy)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "update").
Err(err).
Msg("Could not get the map update")
2022-06-20 13:40:28 -06:00
return
}
2022-06-26 04:06:25 -06:00
_, err = writer.Write(data)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "update").
Err(err).
Msg("Could not write the map response")
updateRequestsSentToNode.WithLabelValues(machine.User.Name, machine.Hostname, "failed").
2021-11-13 01:36:45 -07:00
Inc()
2021-11-14 08:46:09 -07:00
2022-06-20 04:30:51 -06:00
return
}
2022-06-26 04:25:26 -06:00
flusher, ok := writer.(http.Flusher)
if !ok {
log.Error().
Caller().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-06-26 04:25:26 -06:00
Str("machine", machine.Hostname).
Str("channel", "update").
Msg("Cannot cast writer to http.Flusher")
} else {
flusher.Flush()
}
2022-06-25 12:47:42 -06:00
log.Trace().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "update").
Msg("Updated Map has been sent")
updateRequestsSentToNode.WithLabelValues(machine.User.Name, machine.Hostname, "success").
2021-11-13 01:36:45 -07:00
Inc()
2021-10-05 15:59:15 -06:00
// Keep track of the last successful update,
// we sometimes end in a state were the update
// is not picked up by a client and we use this
// to determine if we should "force" an update.
// TODO(kradalby): Abstract away all the database calls, this can cause race conditions
// when an outdated machine object is kept alive, e.g. db is update from
// command line, but then overwritten.
err = h.db.UpdateMachineFromDatabase(machine)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "update").
Err(err).
Msg("Cannot update machine from database")
// client has been removed from database
// since the stream opened, terminate connection.
2022-06-20 04:30:51 -06:00
return
}
now := time.Now().UTC()
lastStateUpdate.WithLabelValues(machine.User.Name, machine.Hostname).
2021-11-13 01:36:45 -07:00
Set(float64(now.Unix()))
2021-11-15 09:15:50 -07:00
machine.LastSuccessfulUpdate = &now
err = h.db.TouchMachine(machine)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "update").
Err(err).
Msg("Cannot update machine LastSuccessfulUpdate")
2022-06-20 13:40:28 -06:00
return
}
} else {
var lastUpdate time.Time
if machine.LastSuccessfulUpdate != nil {
lastUpdate = *machine.LastSuccessfulUpdate
}
log.Trace().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Time("last_successful_update", lastUpdate).
Time("last_state_change", h.getLastStateChange(machine.User)).
2022-04-24 13:55:54 -06:00
Msgf("%s is up to date", machine.Hostname)
}
2021-11-14 08:46:09 -07:00
2022-06-20 04:30:51 -06:00
case <-ctx.Done():
log.Info().
Str("handler", "PollNetMapStream").
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Msg("The client has closed the connection")
// TODO: Abstract away all the database calls, this can cause race conditions
// when an outdated machine object is kept alive, e.g. db is update from
// command line, but then overwritten.
err := h.db.UpdateMachineFromDatabase(machine)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "Done").
Err(err).
Msg("Cannot update machine from database")
// client has been removed from database
// since the stream opened, terminate connection.
2022-06-20 13:40:28 -06:00
return
}
now := time.Now().UTC()
2021-11-15 09:15:50 -07:00
machine.LastSeen = &now
err = h.db.TouchMachine(machine)
if err != nil {
log.Error().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
Str("channel", "Done").
Err(err).
Msg("Cannot update machine LastSeen")
}
2022-06-20 13:40:28 -06:00
// The connection has been closed, so we can stop polling.
return
case <-h.shutdownChan:
log.Info().
Str("handler", "PollNetMapStream").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
Str("machine", machine.Hostname).
Msg("The long-poll handler is shutting down")
2022-06-26 04:06:25 -06:00
return
}
2022-06-20 04:30:51 -06:00
}
}
func (h *Headscale) scheduledPollWorker(
ctx context.Context,
updateChan chan struct{},
keepAliveChan chan []byte,
2021-11-15 09:15:50 -07:00
mapRequest tailcfg.MapRequest,
machine *types.Machine,
isNoise bool,
) {
// TODO(kradalby): This is a stepping stone, mapper should be initiated once
// per client or something similar
mapp := mapper.NewMapper(h.db,
h.privateKey2019,
isNoise,
h.DERPMap,
h.cfg.BaseDomain,
h.cfg.DNSConfig,
h.cfg.LogTail.Enabled,
h.cfg.RandomizeClientPort,
)
keepAliveTicker := time.NewTicker(keepAliveInterval)
updateCheckerTicker := time.NewTicker(h.cfg.NodeUpdateCheckInterval)
defer closeChanWithLog(
updateChan,
2022-05-16 06:59:46 -06:00
fmt.Sprint(ctx.Value(machineNameContextKey)),
"updateChan",
)
defer closeChanWithLog(
keepAliveChan,
2022-05-16 06:59:46 -06:00
fmt.Sprint(ctx.Value(machineNameContextKey)),
"keepAliveChan",
)
for {
select {
case <-ctx.Done():
return
case <-keepAliveTicker.C:
data, err := mapp.CreateKeepAliveResponse(mapRequest, machine)
if err != nil {
log.Error().
Str("func", "keepAlive").
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
Err(err).
Msg("Error generating the keep alive msg")
2021-11-14 08:46:09 -07:00
return
}
log.Debug().
Str("func", "keepAlive").
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
Msg("Sending keepalive")
2022-12-25 04:30:52 -07:00
select {
case keepAliveChan <- data:
case <-ctx.Done():
return
}
case <-updateCheckerTicker.C:
log.Debug().
Str("func", "scheduledPollWorker").
2022-04-24 13:55:54 -06:00
Str("machine", machine.Hostname).
2022-08-14 15:11:33 -06:00
Bool("noise", isNoise).
Msg("Sending update request")
updateRequestsFromNode.WithLabelValues(machine.User.Name, machine.Hostname, "scheduled-update").
2021-11-13 01:36:45 -07:00
Inc()
2022-12-25 04:30:52 -07:00
select {
case updateChan <- struct{}{}:
case <-ctx.Done():
return
}
}
}
}
2022-04-25 14:33:53 -06:00
func closeChanWithLog[C chan []byte | chan struct{}](channel C, machine, name string) {
log.Trace().
Str("handler", "PollNetMap").
Str("machine", machine).
Str("channel", "Done").
Msg(fmt.Sprintf("Closing %s channel", name))
close(channel)
}