chore: tailnet debug logging #7260
Add logging to coordinator
Signed-off-by: Spike Curtis <spike@coder.com>
- Loading branch information
commit 1b9f3285f429128672de815d4c57eb5b7028394e
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -390,7 +390,7 @@ func (api *API) updateEntitlements(ctx context.Context) error { | |
| } | ||
|
|
||
| if changed, enabled := featureChanged(codersdk.FeatureHighAvailability); changed { | ||
| coordinator := agpltailnet.NewCoordinator(api.Logger) | ||
| if enabled { | ||
| haCoordinator, err := tailnet.NewCoordinator(api.Logger, api.Pubsub) | ||
| if err != nil { | ||
|
C958
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -13,6 +13,8 @@ import ( | |
| "sync/atomic" | ||
| "time" | ||
|
|
||
| "cdr.dev/slog" | ||
|
|
||
| "github.com/google/uuid" | ||
| lru "github.com/hashicorp/golang-lru/v2" | ||
| "golang.org/x/exp/slices" | ||
|
|
@@ -111,16 +113,19 @@ func ServeCoordinator(conn net.Conn, updateNodes func(node []*Node) error) (func | |
| }, errChan | ||
| } | ||
|
|
||
| const LoggerName = "coord" | ||
spikecurtis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| // NewCoordinator constructs a new in-memory connection coordinator. This | ||
| // coordinator is incompatible with multiple Coder replicas as all node data is | ||
| // in-memory. | ||
| func NewCoordinator(logger slog.Logger) Coordinator { | ||
| nameCache, err := lru.New[uuid.UUID, string](512) | ||
| if err != nil { | ||
| panic("make lru cache: " + err.Error()) | ||
| } | ||
|
|
||
| return &coordinator{ | ||
| logger: logger.Named(LoggerName), | ||
| closed: false, | ||
| nodes: map[uuid.UUID]*Node{}, | ||
| agentSockets: map[uuid.UUID]*TrackedConn{}, | ||
|
|
@@ -137,6 +142,7 @@ func NewCoordinator() Coordinator { | |
| // This coordinator is incompatible with multiple Coder | ||
| // replicas as all node data is in-memory. | ||
| type coordinator struct { | ||
| logger slog.Logger | ||
| mutex sync.RWMutex | ||
| closed bool | ||
|
|
||
|
|
@@ -194,6 +200,8 @@ func (c *coordinator) AgentCount() int { | |
| // ServeClient accepts a WebSocket connection that wants to connect to an agent | ||
| // with the specified ID. | ||
| func (c *coordinator) ServeClient(conn net.Conn, id uuid.UUID, agent uuid.UUID) error { | ||
| logger := c.logger.With(slog.F("client_id", id), slog.F("agent_id", agent)) | ||
| logger.Debug(context.TODO(), "coordinating client") | ||
spikecurtis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| c.mutex.Lock() | ||
| if c.closed { | ||
| c.mutex.Unlock() | ||
|
|
@@ -210,6 +218,7 @@ func (c *coordinator) ServeClient(conn net.Conn, id uuid.UUID, agent uuid.UUID) | |
| return xerrors.Errorf("marshal node: %w", err) | ||
| } | ||
| _, err = conn.Write(data) | ||
| logger.Debug(context.TODO(), "wrote initial node") | ||
| if err != nil { | ||
| return xerrors.Errorf("write nodes: %w", err) | ||
| } | ||
|
|
@@ -230,7 +239,9 @@ func (c *coordinator) ServeClient(conn net.Conn, id uuid.UUID, agent uuid.UUID) | |
| LastWrite: now, | ||
| } | ||
| c.mutex.Unlock() | ||
| logger.Debug(context.TODO(), "added tracked connection") | ||
| defer func() { | ||
| logger.Debug(context.TODO(), "deleting tracked connection") | ||
spikecurtis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| c.mutex.Lock() | ||
| defer c.mutex.Unlock() | ||
| // Clean all traces of this connection from the map. | ||
|
|
@@ -259,11 +270,13 @@ func (c *coordinator) ServeClient(conn net.Conn, id uuid.UUID, agent uuid.UUID) | |
| } | ||
|
|
||
| func (c *coordinator) handleNextClientMessage(id, agent uuid.UUID, decoder *json.Decoder) error { | ||
| logger := c.logger.With(slog.F("client_id", id), slog.F("agent_id", agent)) | ||
| var node Node | ||
| err := decoder.Decode(&node) | ||
| if err != nil { | ||
| return xerrors.Errorf("read json: %w", err) | ||
| } | ||
| logger.Debug(context.TODO(), "got client node update", slog.F("node", node)) | ||
|
|
||
| c.mutex.Lock() | ||
| // Update the node of this client in our in-memory map. If an agent entirely | ||
|
|
@@ -274,6 +287,7 @@ func (c *coordinator) handleNextClientMessage(id, agent uuid.UUID, decoder *json | |
| agentSocket, ok := c.agentSockets[agent] | ||
| if !ok { | ||
| c.mutex.Unlock() | ||
| logger.Debug(context.TODO(), "no agent socket") | ||
spikecurtis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| return nil | ||
| } | ||
| c.mutex.Unlock() | ||
|
|
@@ -291,13 +305,16 @@ func (c *coordinator) handleNextClientMessage(id, agent uuid.UUID, decoder *json | |
| } | ||
| return xerrors.Errorf("write json: %w", err) | ||
| } | ||
| logger.Debug(context.TODO(), "sent client node to agent") | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // ServeAgent accepts a WebSocket connection to an agent that | ||
| // listens to incoming connections and publishes node updates. | ||
| func (c *coordinator) ServeAgent(conn net.Conn, id uuid.UUID, name string) error { | ||
| logger := c.logger.With(slog.F("agent_id", id)) | ||
| logger.Debug(context.TODO(), "coordinating agent") | ||
| c.mutex.Lock() | ||
| if c.closed { | ||
| c.mutex.Unlock() | ||
|
|
@@ -324,6 +341,7 @@ func (c *coordinator) ServeAgent(conn net.Conn, id uuid.UUID, name string) error | |
| return xerrors.Errorf("marshal json: %w", err) | ||
| } | ||
| _, err = conn.Write(data) | ||
| logger.Debug(context.TODO(), "wrote initial client(s) to agent", slog.F("nodes", nodes)) | ||
| if err != nil { | ||
| return xerrors.Errorf("write nodes: %w", err) | ||
| } | ||
|
|
@@ -356,6 +374,7 @@ func (c *coordinator) ServeAgent(conn net.Conn, id uuid.UUID, name string) error | |
| } | ||
|
|
||
| c.mutex.Unlock() | ||
| logger.Debug(context.TODO(), "added agent socket") | ||
| defer func() { | ||
| c.mutex.Lock() | ||
| defer c.mutex.Unlock() | ||
|
|
@@ -365,6 +384,7 @@ func (c *coordinator) ServeAgent(conn net.Conn, id uuid.UUID, name string) error | |
| if idConn, ok := c.agentSockets[id]; ok && idConn.ID == unique { | ||
| delete(c.agentSockets, id) | ||
| delete(c.nodes, id) | ||
| logger.Debug(context.TODO(), "deleted agent socket") | ||
| } | ||
| }() | ||
|
|
||
|
|
@@ -381,17 +401,20 @@ func (c *coordinator) ServeAgent(conn net.Conn, id uuid.UUID, name string) error | |
| } | ||
|
|
||
| func (c *coordinator) handleNextAgentMessage(id uuid.UUID, decoder *json.Decoder) error { | ||
| logger := c.logger.With(slog.F("agent_id", id)) | ||
| var node Node | ||
| err := decoder.Decode(&node) | ||
| if err != nil { | ||
| return xerrors.Errorf("read json: %w", err) | ||
| } | ||
| logger.Debug(context.TODO(), "decoded agent node", slog.F("node", node)) | ||
|
|
||
| c.mutex.Lock() | ||
| c.nodes[id] = &node | ||
| connectionSockets, ok := c.agentToConnectionSockets[id] | ||
| if !ok { | ||
| c.mutex.Unlock() | ||
| logger.Debug(context.TODO(), "no client sockets") | ||
spikecurtis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| return nil | ||
| } | ||
| data, err := json.Marshal([]*Node{&node}) | ||
|
|
@@ -403,11 +426,14 @@ func (c *coordinator) handleNextAgentMessage(id uuid.UUID, decoder *json.Decoder | |
| // Publish the new node to every listening socket. | ||
| var wg sync.WaitGroup | ||
| wg.Add(len(connectionSockets)) | ||
| for clientID, connectionSocket := range connectionSockets { | ||
| clientID := clientID | ||
| connectionSocket := connectionSocket | ||
| go func() { | ||
| _ = connectionSocket.SetWriteDeadline(time.Now().Add(5 * time.Second)) | ||
| _, err := connectionSocket.Write(data) | ||
| logger.Debug(context.TODO(), "sent agent node to client", | ||
| slog.F("client_id", clientID), slog.Error(err)) | ||
| wg.Done() | ||
| }() | ||
| } | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.