Skip to content

fix!: remove startup logs eof for streaming #8528

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Jul 18, 2023
Merged

fix!: remove startup logs eof for streaming #8528

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c1e6aa0
fix: remove startup logs eof for streaming
kylecarbs Jul 14, 2023
506dce5
fix(cli/cliui): handle never ending startup log stream in Agent
mafredri Jul 17, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 44 additions & 97 deletions coderd/workspaceagents.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -280,81 +280,61 @@ func (api *API) patchWorkspaceAgentStartupLogs(rw http.ResponseWriter, r *http.R
level = append(level, parsedLevel)
}

var logs []database.WorkspaceAgentStartupLog
// Ensure logs are not written after script ended.
scriptEndedError := xerrors.New("startup script has ended")
err := api.Database.InTx(func(db database.Store) error {
state, err := db.GetWorkspaceAgentLifecycleStateByID(ctx, workspaceAgent.ID)
if err != nil {
return xerrors.Errorf("workspace agent startup script status: %w", err)
logs, err := api.Database.InsertWorkspaceAgentStartupLogs(ctx, database.InsertWorkspaceAgentStartupLogsParams{
AgentID: workspaceAgent.ID,
CreatedAt: createdAt,
Output: output,
Level: level,
OutputLength: int32(outputLength),
})
if err != nil {
if !database.IsStartupLogsLimitError(err) {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Failed to upload startup logs",
Detail: err.Error(),
})
return
}

if state.ReadyAt.Valid {
// The agent startup script has already ended, so we don't want to
// process any more logs.
return scriptEndedError
if workspaceAgent.StartupLogsOverflowed {
httpapi.Write(ctx, rw, http.StatusRequestEntityTooLarge, codersdk.Response{
Message: "Startup logs limit exceeded",
Detail: err.Error(),
})
return
}

logs, err = db.InsertWorkspaceAgentStartupLogs(ctx, database.InsertWorkspaceAgentStartupLogsParams{
AgentID: workspaceAgent.ID,
CreatedAt: createdAt,
Output: output,
Level: level,
OutputLength: int32(outputLength),
err := api.Database.UpdateWorkspaceAgentStartupLogOverflowByID(ctx, database.UpdateWorkspaceAgentStartupLogOverflowByIDParams{
ID: workspaceAgent.ID,
StartupLogsOverflowed: true,
})
return err
}, nil)
if err != nil {
if errors.Is(err, scriptEndedError) {
httpapi.Write(ctx, rw, http.StatusConflict, codersdk.Response{
Message: "Failed to upload logs, startup script has already ended.",
if err != nil {
// We don't want to return here, because the agent will retry
// on failure and this isn't a huge deal. The overflow state
// is just a hint to the user that the logs are incomplete.
api.Logger.Warn(ctx, "failed to update workspace agent startup log overflow", slog.Error(err))
}

resource, err := api.Database.GetWorkspaceResourceByID(ctx, workspaceAgent.ResourceID)
if err != nil {
httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
Message: "Failed to get workspace resource.",
Detail: err.Error(),
})
return
}
if database.IsStartupLogsLimitError(err) {
if !workspaceAgent.StartupLogsOverflowed {
err := api.Database.UpdateWorkspaceAgentStartupLogOverflowByID(ctx, database.UpdateWorkspaceAgentStartupLogOverflowByIDParams{
ID: workspaceAgent.ID,
StartupLogsOverflowed: true,
})
if err != nil {
// We don't want to return here, because the agent will retry
// on failure and this isn't a huge deal. The overflow state
// is just a hint to the user that the logs are incomplete.
api.Logger.Warn(ctx, "failed to update workspace agent startup log overflow", slog.Error(err))
}

resource, err := api.Database.GetWorkspaceResourceByID(ctx, workspaceAgent.ResourceID)
if err != nil {
httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
Message: "Failed to get workspace resource.",
Detail: err.Error(),
})
return
}

build, err := api.Database.GetWorkspaceBuildByJobID(ctx, resource.JobID)
if err != nil {
httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
Message: "Internal error fetching workspace build job.",
Detail: err.Error(),
})
return
}

api.publishWorkspaceUpdate(ctx, build.WorkspaceID)
}

httpapi.Write(ctx, rw, http.StatusRequestEntityTooLarge, codersdk.Response{
Message: "Startup logs limit exceeded",
build, err := api.Database.GetWorkspaceBuildByJobID(ctx, resource.JobID)
if err != nil {
httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
Message: "Internal error fetching workspace build job.",
Detail: err.Error(),
})
return
}
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Failed to upload startup logs",
Detail: err.Error(),

api.publishWorkspaceUpdate(ctx, build.WorkspaceID)

httpapi.Write(ctx, rw, http.StatusRequestEntityTooLarge, codersdk.Response{
Message: "Startup logs limit exceeded",
})
return
}
Expand Down Expand Up @@ -497,18 +477,6 @@ func (api *API) workspaceAgentStartupLogs(rw http.ResponseWriter, r *http.Reques
return
}

if workspaceAgent.ReadyAt.Valid {
// Fast path, the startup script has finished running, so we can close
// the connection.
return
}
if !codersdk.WorkspaceAgentLifecycle(workspaceAgent.LifecycleState).Starting() {
// Backwards compatibility: Avoid waiting forever in case this agent is
// older than the current release and has already reported the ready
// state.
return
}

lastSentLogID := after
if len(logs) > 0 {
lastSentLogID = logs[len(logs)-1].ID
Expand Down Expand Up @@ -543,11 +511,9 @@ func (api *API) workspaceAgentStartupLogs(rw http.ResponseWriter, r *http.Reques
t := time.NewTicker(recheckInterval)
defer t.Stop()

var state database.GetWorkspaceAgentLifecycleStateByIDRow
go func() {
defer close(bufferedLogs)

var err error
for {
select {
case <-ctx.Done():
Expand All @@ -557,17 +523,6 @@ func (api *API) workspaceAgentStartupLogs(rw http.ResponseWriter, r *http.Reques
t.Reset(recheckInterval)
}

if !state.ReadyAt.Valid {
state, err = api.Database.GetWorkspaceAgentLifecycleStateByID(ctx, workspaceAgent.ID)
if err != nil {
if xerrors.Is(err, context.Canceled) {
return
}
logger.Warn(ctx, "failed to get workspace agent lifecycle state", slog.Error(err))
continue
}
}

logs, err := api.Database.GetWorkspaceAgentStartupLogsAfter(ctx, database.GetWorkspaceAgentStartupLogsAfterParams{
AgentID: workspaceAgent.ID,
CreatedAfter: lastSentLogID,
Expand All @@ -580,9 +535,7 @@ func (api *API) workspaceAgentStartupLogs(rw http.ResponseWriter, r *http.Reques
continue
}
if len(logs) == 0 {
if state.ReadyAt.Valid {
return
}
// Just keep listening - more logs might come in the future!
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can remove the ready state stuff higher up too.

continue
}

Expand Down Expand Up @@ -1689,12 +1642,6 @@ func (api *API) workspaceAgentReportLifecycle(rw http.ResponseWriter, r *http.Re
return
}

if readyAt.Valid {
api.publishWorkspaceAgentStartupLogsUpdate(ctx, workspaceAgent.ID, agentsdk.StartupLogsNotifyMessage{
EndOfLogs: true,
})
}

api.publishWorkspaceUpdate(ctx, workspace.ID)

httpapi.Write(ctx, rw, http.StatusNoContent, nil)
Expand Down
124 changes: 0 additions & 124 deletions coderd/workspaceagents_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -301,130 +301,6 @@ func TestWorkspaceAgentStartupLogs(t *testing.T) {
}
}
})
t.Run("CloseAfterLifecycleStateIsNotRunning", func(t *testing.T) {
t.Parallel()
ctx := testutil.Context(t, testutil.WaitMedium)
client := coderdtest.New(t, &coderdtest.Options{
IncludeProvisionerDaemon: true,
})
user := coderdtest.CreateFirstUser(t, client)
authToken := uuid.NewString()
version := coderdtest.CreateTemplateVersion(t, client, user.OrganizationID, &echo.Responses{
Parse: echo.ParseComplete,
ProvisionPlan: echo.ProvisionComplete,
ProvisionApply: []*proto.Provision_Response{{
Type: &proto.Provision_Response_Complete{
Complete: &proto.Provision_Complete{
Resources: []*proto.Resource{{
Name: "example",
Type: "aws_instance",
Agents: []*proto.Agent{{
Id: uuid.NewString(),
Auth: &proto.Agent_Token{
Token: authToken,
},
}},
}},
},
},
}},
})
template := coderdtest.CreateTemplate(t, client, user.OrganizationID, version.ID)
coderdtest.AwaitTemplateVersionJob(t, client, version.ID)
workspace := coderdtest.CreateWorkspace(t, client, user.OrganizationID, template.ID)
build := coderdtest.AwaitWorkspaceBuildJob(t, client, workspace.LatestBuild.ID)

agentClient := agentsdk.New(client.URL)
agentClient.SetSessionToken(authToken)

logs, closer, err := client.WorkspaceAgentStartupLogsAfter(ctx, build.Resources[0].Agents[0].ID, 0, true)
require.NoError(t, err)
defer func() {
_ = closer.Close()
}()

err = agentClient.PatchStartupLogs(ctx, agentsdk.PatchStartupLogs{
Logs: []agentsdk.StartupLog{
{
CreatedAt: database.Now(),
Output: "testing",
},
},
})
require.NoError(t, err)

err = agentClient.PostLifecycle(ctx, agentsdk.PostLifecycleRequest{
State: codersdk.WorkspaceAgentLifecycleReady,
ChangedAt: time.Now(),
})
require.NoError(t, err)

var gotLogs []codersdk.WorkspaceAgentStartupLog
for {
select {
case <-ctx.Done():
require.Fail(t, "timed out waiting for logs to end")
case l, ok := <-logs:
gotLogs = append(gotLogs, l...)
if !ok {
require.Len(t, gotLogs, 1, "expected one log")
return // Success.
}
}
}
})
t.Run("NoLogAfterScriptEnded", func(t *testing.T) {
t.Parallel()
ctx := testutil.Context(t, testutil.WaitMedium)
client := coderdtest.New(t, &coderdtest.Options{
IncludeProvisionerDaemon: true,
})
user := coderdtest.CreateFirstUser(t, client)
authToken := uuid.NewString()
version := coderdtest.CreateTemplateVersion(t, client, user.OrganizationID, &echo.Responses{
Parse: echo.ParseComplete,
ProvisionPlan: echo.ProvisionComplete,
ProvisionApply: []*proto.Provision_Response{{
Type: &proto.Provision_Response_Complete{
Complete: &proto.Provision_Complete{
Resources: []*proto.Resource{{
Name: "example",
Type: "aws_instance",
Agents: []*proto.Agent{{
Id: uuid.NewString(),
Auth: &proto.Agent_Token{
Token: authToken,
},
}},
}},
},
},
}},
})
template := coderdtest.CreateTemplate(t, client, user.OrganizationID, version.ID)
coderdtest.AwaitTemplateVersionJob(t, client, version.ID)
workspace := coderdtest.CreateWorkspace(t, client, user.OrganizationID, template.ID)
_ = coderdtest.AwaitWorkspaceBuildJob(t, client, workspace.LatestBuild.ID)

agentClient := agentsdk.New(client.URL)
agentClient.SetSessionToken(authToken)

err := agentClient.PostLifecycle(ctx, agentsdk.PostLifecycleRequest{
State: codersdk.WorkspaceAgentLifecycleReady,
ChangedAt: time.Now(),
})
require.NoError(t, err)

err = agentClient.PatchStartupLogs(ctx, agentsdk.PatchStartupLogs{
Logs: []agentsdk.StartupLog{
{
CreatedAt: database.Now(),
Output: "testing",
},
},
})
require.Error(t, err, "insert after script ended should not succeed")
})
}

func TestWorkspaceAgentListen(t *testing.T) {
Expand Down
1 change: 0 additions & 1 deletion codersdk/agentsdk/agentsdk.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -694,7 +694,6 @@ func StartupLogsNotifyChannel(agentID uuid.UUID) string {

type StartupLogsNotifyMessage struct {
CreatedAfter int64 `json:"created_after"`
EndOfLogs bool `json:"end_of_logs"`
}

type closeNetConn struct {
Expand Down