opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-24 15:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: rename startup logs to agent logs (#8649)

Browse Source 
* chore: rename startup logs to agent logs

This also adds a `source` property to every agent log. It
should allow us to group logs and display them nicer in
the UI as they stream in.

* Fix migration order

* Fix naming

* Rename the frontend

* Fix tests

* Fix down migration

* Match enums for workspace agent logs

* Fix inserting log source

* Fix migration order

* Fix logs tests

* Fix psql insert
This commit is contained in:
Kyle Carberry
2023-07-28 15:57:23 +00:00
committed by GitHub
parent 78b06397a6
commit bd944e0d21
48 changed files with 1471 additions and 876 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codersdk/agentsdk/agentsdk.go
+15 -14
View File
@@ -621,20 +621,21 @@ func (c *Client) PostStartup(ctx context.Context, req PostStartupRequest) error
return nil
}
type StartupLog struct {
CreatedAt time.Time `json:"created_at"`
Output string `json:"output"`
Level codersdk.LogLevel `json:"level"`
type Log struct {
CreatedAt time.Time `json:"created_at"`
Output string `json:"output"`
Level codersdk.LogLevel `json:"level"`
Source codersdk.WorkspaceAgentLogSource `json:"source"`
}
type PatchStartupLogs struct {
Logs []StartupLog `json:"logs"`
type PatchLogs struct {
Logs []Log `json:"logs"`
}
// PatchStartupLogs writes log messages to the agent startup script.
// PatchLogs writes log messages to the agent startup script.
// Log messages are limited to 1MB in total.
func (c *Client) PatchStartupLogs(ctx context.Context, req PatchStartupLogs) error {
res, err := c.SDK.Request(ctx, http.MethodPatch, "/api/v2/workspaceagents/me/startup-logs", req)
func (c *Client) PatchLogs(ctx context.Context, req PatchLogs) error {
res, err := c.SDK.Request(ctx, http.MethodPatch, "/api/v2/workspaceagents/me/logs", req)
if err != nil {
return err
}
@@ -737,13 +738,13 @@ func websocketNetConn(ctx context.Context, conn *websocket.Conn, msgType websock
}
}
// StartupLogsNotifyChannel returns the channel name responsible for notifying
// of new startup logs.
func StartupLogsNotifyChannel(agentID uuid.UUID) string {
return fmt.Sprintf("startup-logs:%s", agentID)
// LogsNotifyChannel returns the channel name responsible for notifying
// of new logs.
func LogsNotifyChannel(agentID uuid.UUID) string {
return fmt.Sprintf("agent-logs:%s", agentID)
}
type StartupLogsNotifyMessage struct {
type LogsNotifyMessage struct {
CreatedAfter int64 `json:"created_after"`
}
codersdk/agentsdk/logs.go
+24 -17
View File
@@ -16,10 +16,11 @@ import (
)
type startupLogsWriter struct {
buf bytes.Buffer // Buffer to track partial lines.
ctx context.Context
send func(ctx context.Context, log ...StartupLog) error
level codersdk.LogLevel
buf bytes.Buffer // Buffer to track partial lines.
ctx context.Context
send func(ctx context.Context, log ...Log) error
level codersdk.LogLevel
source codersdk.WorkspaceAgentLogSource
}
func (w *startupLogsWriter) Write(p []byte) (int, error) {
@@ -39,10 +40,11 @@ func (w *startupLogsWriter) Write(p []byte) (int, error) {
partial = w.buf.Bytes()
w.buf.Reset()
}
err := w.send(w.ctx, StartupLog{
err := w.send(w.ctx, Log{
CreatedAt: time.Now().UTC(), // UTC, like database.Now().
Level: w.level,
Output: string(partial) + string(p[:nl-cr]),
Source: w.source,
})
if err != nil {
return n - len(p), err
@@ -61,10 +63,11 @@ func (w *startupLogsWriter) Write(p []byte) (int, error) {
func (w *startupLogsWriter) Close() error {
if w.buf.Len() > 0 {
defer w.buf.Reset()
return w.send(w.ctx, StartupLog{
return w.send(w.ctx, Log{
CreatedAt: time.Now().UTC(), // UTC, like database.Now().
Level: w.level,
Output: w.buf.String(),
Source: w.source,
})
}
return nil
@@ -78,20 +81,24 @@ func (w *startupLogsWriter) Close() error {
//
// Neither Write nor Close is safe for concurrent use and must be used
// by a single goroutine.
func StartupLogsWriter(ctx context.Context, sender func(ctx context.Context, log ...StartupLog) error, level codersdk.LogLevel) io.WriteCloser {
func StartupLogsWriter(ctx context.Context, sender func(ctx context.Context, log ...Log) error, source codersdk.WorkspaceAgentLogSource, level codersdk.LogLevel) io.WriteCloser {
if source == "" {
source = codersdk.WorkspaceAgentLogSourceExternal
}
return &startupLogsWriter{
ctx: ctx,
send: sender,
level: level,
ctx: ctx,
send: sender,
level: level,
source: source,
}
}
// SendStartupLogs will send agent startup logs to the server. Calls to
// LogsSender will send agent startup logs to the server. Calls to
// sendLog are non-blocking and will return an error if flushAndClose
// has been called. Calling sendLog concurrently is not supported. If
// the context passed to flushAndClose is canceled, any remaining logs
// will be discarded.
func StartupLogsSender(patchStartupLogs func(ctx context.Context, req PatchStartupLogs) error, logger slog.Logger) (sendLog func(ctx context.Context, log ...StartupLog) error, flushAndClose func(context.Context) error) {
func LogsSender(patchLogs func(ctx context.Context, req PatchLogs) error, logger slog.Logger) (sendLog func(ctx context.Context, log ...Log) error, flushAndClose func(context.Context) error) {
// The main context is used to close the sender goroutine and cancel
// any outbound requests to the API. The shutdown context is used to
// signal the sender goroutine to flush logs and then exit.
@@ -100,7 +107,7 @@ func StartupLogsSender(patchStartupLogs func(ctx context.Context, req PatchStart
// Synchronous sender, there can only be one outbound send at a time.
sendDone := make(chan struct{})
send := make(chan []StartupLog, 1)
send := make(chan []Log, 1)
go func() {
// Set flushTimeout and backlogLimit so that logs are uploaded
// once every 250ms or when 100 logs have been added to the
@@ -110,7 +117,7 @@ func StartupLogsSender(patchStartupLogs func(ctx context.Context, req PatchStart
flush := time.NewTicker(flushTimeout)
var backlog []StartupLog
var backlog []Log
defer func() {
flush.Stop()
if len(backlog) > 0 {
@@ -150,7 +157,7 @@ func StartupLogsSender(patchStartupLogs func(ctx context.Context, req PatchStart
// meaning these requests won't be interrupted by
// shutdown.
for r := retry.New(time.Second, 5*time.Second); r.Wait(ctx); {
err := patchStartupLogs(ctx, PatchStartupLogs{
err := patchLogs(ctx, PatchLogs{
Logs: backlog,
})
if err == nil {
@@ -185,8 +192,8 @@ func StartupLogsSender(patchStartupLogs func(ctx context.Context, req PatchStart
}
}()
var queue []StartupLog
sendLog = func(callCtx context.Context, log ...StartupLog) error {
var queue []Log
sendLog = func(callCtx context.Context, log ...Log) error {
select {
case <-shutdownCtx.Done():
return xerrors.Errorf("closed: %w", shutdownCtx.Err())
codersdk/agentsdk/logs_test.go
+44 -25
View File
@@ -28,8 +28,9 @@ func TestStartupLogsWriter_Write(t *testing.T) {
name string
ctx context.Context
level codersdk.LogLevel
source codersdk.WorkspaceAgentLogSource
writes []string
want []agentsdk.StartupLog
want []agentsdk.Log
wantErr bool
closeFirst bool
}{
@@ -38,10 +39,12 @@ func TestStartupLogsWriter_Write(t *testing.T) {
ctx: context.Background(),
level: codersdk.LogLevelInfo,
writes: []string{"hello world\n"},
want: []agentsdk.StartupLog{
source: codersdk.WorkspaceAgentLogSourceShutdownScript,
want: []agentsdk.Log{
{
Level: codersdk.LogLevelInfo,
Output: "hello world",
Source: codersdk.WorkspaceAgentLogSourceShutdownScript,
},
},
},
@@ -50,14 +53,16 @@ func TestStartupLogsWriter_Write(t *testing.T) {
ctx: context.Background(),
level: codersdk.LogLevelInfo,
writes: []string{"hello world\n", "goodbye world\n"},
want: []agentsdk.StartupLog{
want: []agentsdk.Log{
{
Level: codersdk.LogLevelInfo,
Output: "hello world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "goodbye world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
},
},
@@ -66,30 +71,36 @@ func TestStartupLogsWriter_Write(t *testing.T) {
ctx: context.Background(),
level: codersdk.LogLevelInfo,
writes: []string{"\n\n", "hello world\n\n\n", "goodbye world\n"},
want: []agentsdk.StartupLog{
want: []agentsdk.Log{
{
Level: codersdk.LogLevelInfo,
Output: "",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "hello world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "goodbye world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
},
},
@@ -98,10 +109,11 @@ func TestStartupLogsWriter_Write(t *testing.T) {
ctx: context.Background(),
level: codersdk.LogLevelInfo,
writes: []string{"hello world\n", "goodbye world"},
want: []agentsdk.StartupLog{
want: []agentsdk.Log{
{
Level: codersdk.LogLevelInfo,
Output: "hello world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
},
},
@@ -111,14 +123,16 @@ func TestStartupLogsWriter_Write(t *testing.T) {
level: codersdk.LogLevelInfo,
writes: []string{"hello world\n", "goodbye world"},
closeFirst: true,
want: []agentsdk.StartupLog{
want: []agentsdk.Log{
{
Level: codersdk.LogLevelInfo,
Output: "hello world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "goodbye world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
},
},
@@ -127,14 +141,16 @@ func TestStartupLogsWriter_Write(t *testing.T) {
ctx: context.Background(),
level: codersdk.LogLevelInfo,
writes: []string{"hello world\n", "goodbye", " world\n"},
want: []agentsdk.StartupLog{
want: []agentsdk.Log{
{
Level: codersdk.LogLevelInfo,
Output: "hello world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "goodbye world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
},
},
@@ -143,18 +159,21 @@ func TestStartupLogsWriter_Write(t *testing.T) {
ctx: context.Background(),
level: codersdk.LogLevelInfo,
writes: []string{"hello world\r\n", "\r\r\n", "goodbye world\n"},
want: []agentsdk.StartupLog{
want: []agentsdk.Log{
{
Level: codersdk.LogLevelInfo,
Output: "hello world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "\r",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
{
Level: codersdk.LogLevelInfo,
Output: "goodbye world",
Source: codersdk.WorkspaceAgentLogSourceExternal,
},
},
},
@@ -172,8 +191,8 @@ func TestStartupLogsWriter_Write(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
var got []agentsdk.StartupLog
send := func(ctx context.Context, log ...agentsdk.StartupLog) error {
var got []agentsdk.Log
send := func(ctx context.Context, log ...agentsdk.Log) error {
select {
case <-ctx.Done():
return ctx.Err()
@@ -182,7 +201,7 @@ func TestStartupLogsWriter_Write(t *testing.T) {
got = append(got, log...)
return nil
}
w := agentsdk.StartupLogsWriter(tt.ctx, send, tt.level)
w := agentsdk.StartupLogsWriter(tt.ctx, send, tt.source, tt.level)
for _, s := range tt.writes {
_, err := w.Write([]byte(s))
if err != nil {
@@ -233,7 +252,7 @@ func TestStartupLogsSender(t *testing.T) {
name string
sendCount int
discard []int
patchResp func(req agentsdk.PatchStartupLogs) error
patchResp func(req agentsdk.PatchLogs) error
}{
{
name: "single log",
@@ -247,7 +266,7 @@ func TestStartupLogsSender(t *testing.T) {
name: "too large",
sendCount: 1,
discard: []int{1},
patchResp: func(req agentsdk.PatchStartupLogs) error {
patchResp: func(req agentsdk.PatchLogs) error {
return statusError(http.StatusRequestEntityTooLarge)
},
},
@@ -260,8 +279,8 @@ func TestStartupLogsSender(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitMedium)
defer cancel()
got := []agentsdk.StartupLog{}
patchStartupLogs := func(_ context.Context, req agentsdk.PatchStartupLogs) error {
got := []agentsdk.Log{}
patchLogs := func(_ context.Context, req agentsdk.PatchLogs) error {
if tt.patchResp != nil {
err := tt.patchResp(req)
if err != nil {
@@ -272,15 +291,15 @@ func TestStartupLogsSender(t *testing.T) {
return nil
}
sendLog, flushAndClose := agentsdk.StartupLogsSender(patchStartupLogs, slogtest.Make(t, nil).Leveled(slog.LevelDebug))
sendLog, flushAndClose := agentsdk.LogsSender(patchLogs, slogtest.Make(t, nil).Leveled(slog.LevelDebug))
defer func() {
err := flushAndClose(ctx)
require.NoError(t, err)
}()
var want []agentsdk.StartupLog
var want []agentsdk.Log
for i := 0; i < tt.sendCount; i++ {
want = append(want, agentsdk.StartupLog{
want = append(want, agentsdk.Log{
CreatedAt: time.Now(),
Level: codersdk.LogLevelInfo,
Output: fmt.Sprintf("hello world %d", i),
@@ -306,18 +325,18 @@ func TestStartupLogsSender(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
defer cancel()
patchStartupLogs := func(_ context.Context, _ agentsdk.PatchStartupLogs) error {
patchLogs := func(_ context.Context, _ agentsdk.PatchLogs) error {
assert.Fail(t, "should not be called")
return nil
}
sendLog, flushAndClose := agentsdk.StartupLogsSender(patchStartupLogs, slogtest.Make(t, nil).Leveled(slog.LevelDebug))
sendLog, flushAndClose := agentsdk.LogsSender(patchLogs, slogtest.Make(t, nil).Leveled(slog.LevelDebug))
defer func() {
_ = flushAndClose(ctx)
}()
cancel()
err := sendLog(ctx, agentsdk.StartupLog{
err := sendLog(ctx, agentsdk.Log{
CreatedAt: time.Now(),
Level: codersdk.LogLevelInfo,
Output: "hello world",
@@ -336,18 +355,18 @@ func TestStartupLogsSender(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
defer cancel()
var want, got []agentsdk.StartupLog
patchStartupLogs := func(_ context.Context, req agentsdk.PatchStartupLogs) error {
var want, got []agentsdk.Log
patchLogs := func(_ context.Context, req agentsdk.PatchLogs) error {
got = append(got, req.Logs...)
return nil
}
sendLog, flushAndClose := agentsdk.StartupLogsSender(patchStartupLogs, slogtest.Make(t, nil).Leveled(slog.LevelDebug))
sendLog, flushAndClose := agentsdk.LogsSender(patchLogs, slogtest.Make(t, nil).Leveled(slog.LevelDebug))
defer func() {
_ = flushAndClose(ctx)
}()
err := sendLog(ctx, agentsdk.StartupLog{
err := sendLog(ctx, agentsdk.Log{
CreatedAt: time.Now(),
Level: codersdk.LogLevelInfo,
Output: "hello world",