May 14, 2025 · Apr 21, 2025 · Apr 21, 2025 · Apr 23, 2025 · Apr 24, 2025 · Apr 24, 2025
diff --git a/agent/agent.go b/agent/agent.go
 "tailscale.com/util/clientmetric"

 "cdr.dev/slog"

 "github.com/coder/retry"

 "github.com/coder/clistat"
 "github.com/coder/coder/v2/agent/agentcontainers"
 "github.com/coder/coder/v2/agent/agentexec"
 "github.com/coder/coder/v2/tailnet"
 tailnetproto "github.com/coder/coder/v2/tailnet/proto"
 "github.com/coder/quartz"
 "github.com/coder/retry"
 )

 const (
 if ctx.Err() != nil {
 // Context canceled errors may come from websocket pings, so we
 // don't want to use `errors.Is(err, context.Canceled)` here.
 a.logger.Warn(ctx, "runLoop exited with error", slog.Error(ctx.Err()))
 return
 }
 if a.isClosed() {
 a.logger.Warn(ctx, "runLoop exited because agent is closed")
 return
 }
 if errors.Is(err, io.EOF) {
 return a.statsReporter.reportLoop(ctx, aAPI)
 })

 return connMan.wait()
 err = connMan.wait()
 if err != nil {
 a.logger.Warn(context.Background(), "connection manager errored", slog.Error(err))
 }
 return err
 }

 // handleManifest returns a function that fetches and processes the manifest
diff --git a/agent/reaper/reaper_unix.go b/agent/reaper/reaper_unix.go
 package reaper

 import (
 "fmt"
 "os"
 "os/signal"
 "syscall"
 s := <-sc
 sig, ok := s.(syscall.Signal)
 if ok {
 // TODO:
 // Tried using a logger here but the I/O streams are already closed at this point...
 // Why is os.Stderr still working then?
 _, _ = fmt.Fprintf(os.Stderr, "reaper caught %q signal, killing process %v\n", sig.String(), pid)
 _ = syscall.Kill(pid, sig)
 }
 }
diff --git a/cli/agent.go b/cli/agent.go
 "golang.org/x/xerrors"
 "gopkg.in/natefinch/lumberjack.v2"

 "github.com/coder/retry"

 "github.com/prometheus/client_golang/prometheus"

 "cdr.dev/slog"
 "cdr.dev/slog/sloggers/sloghuman"
 "cdr.dev/slog/sloggers/slogjson"
 "cdr.dev/slog/sloggers/slogstackdriver"
 "github.com/coder/serpent"

 "github.com/coder/coder/v2/agent"
 "github.com/coder/coder/v2/agent/agentcontainers"
 "github.com/coder/coder/v2/agent/agentexec"
 "github.com/coder/coder/v2/cli/clilog"
 "github.com/coder/coder/v2/codersdk"
 "github.com/coder/coder/v2/codersdk/agentsdk"
 "github.com/coder/serpent"
 )

 func (r *RootCmd) workspaceAgent() *serpent.Command {
 // This command isn't useful to manually execute.
 Hidden: true,
 Handler: func(inv *serpent.Invocation) error {
 ctx, cancel := context.WithCancel(inv.Context())
 defer cancel()
 ctx, cancel := context.WithCancelCause(inv.Context())
 defer func() {
 cancel(xerrors.New("agent exited"))
 }()

 var (
 ignorePorts = map[int]string{}
 return xerrors.Errorf("add executable to $PATH: %w", err)
 }

 prometheusRegistry := prometheus.NewRegistry()
 subsystemsRaw := inv.Environ.Get(agent.EnvAgentSubsystem)
 subsystems := []codersdk.AgentSubsystem{}
 for _, s := range strings.Split(subsystemsRaw, ",") {
 containerLister = agentcontainers.NewDocker(execer)
 }

 agnt := agent.New(agent.Options{
 Client:        client,
 Logger:        logger,
 LogDir:        logDir,
 ScriptDataDir: scriptDataDir,
 // #nosec G115 - Safe conversion as tailnet listen port is within uint16 range (0-65535)
 TailnetListenPort: uint16(tailnetListenPort),
 ExchangeToken: func(ctx context.Context) (string, error) {
 if exchangeToken == nil {
 return client.SDK.SessionToken(), nil
 // TODO: timeout ok?
 reinitCtx, reinitCancel := context.WithTimeout(context.Background(), time.Hour*24)
 defer reinitCancel()
 reinitEvents := make(chan agentsdk.ReinitializationEvent)

 go func() {
 // Retry to wait for reinit, main context cancels the retrier.
 for retrier := retry.New(100*time.Millisecond, 10*time.Second); retrier.Wait(ctx); {
 select {
 case <-reinitCtx.Done():
 return
 default:
 }
 resp, err := exchangeToken(ctx)

 err := client.WaitForReinit(reinitCtx, reinitEvents)
 if err != nil {
 return "", err
 logger.Error(ctx, "failed to wait for reinit instructions, will retry",slog.Error(err))
 }
 client.SetSessionToken(resp.SessionToken)
 return resp.SessionToken, nil
 },
 EnvironmentVariables: environmentVariables,
 IgnorePorts:          ignorePorts,
 SSHMaxTimeout:        sshMaxTimeout,
 Subsystems:           subsystems,

 PrometheusRegistry: prometheusRegistry,
 BlockFileTransfer:  blockFileTransfer,
 Execer:             execer,
 ContainerLister:    containerLister,

 ExperimentalDevcontainersEnabled: experimentalDevcontainersEnabled,
 })

 promHandler := agent.PrometheusMetricsHandler(prometheusRegistry, logger)
 prometheusSrvClose := ServeHandler(ctx, logger, promHandler, prometheusAddress, "prometheus")
 defer prometheusSrvClose()

 debugSrvClose := ServeHandler(ctx, logger, agnt.HTTPDebug(), debugAddress, "debug")
 defer debugSrvClose()

 <-ctx.Done()
 return agnt.Close()
 }
 }()

 var (
 lastErr  error
 mustExit bool
 )
 for {
 prometheusRegistry := prometheus.NewRegistry()

 agnt := agent.New(agent.Options{
 Client:        client,
 Logger:        logger,
 LogDir:        logDir,
 ScriptDataDir: scriptDataDir,
 // #nosec G115 - Safe conversion as tailnet listen port is within uint16 range (0-65535)
 TailnetListenPort: uint16(tailnetListenPort),
 ExchangeToken: func(ctx context.Context) (string, error) {
 if exchangeToken == nil {
 return client.SDK.SessionToken(), nil
 }
 resp, err := exchangeToken(ctx)
 if err != nil {
 return "", err
 }
 client.SetSessionToken(resp.SessionToken)
 return resp.SessionToken, nil
 },
 EnvironmentVariables: environmentVariables,
 IgnorePorts:          ignorePorts,
 SSHMaxTimeout:        sshMaxTimeout,
 Subsystems:           subsystems,

 PrometheusRegistry:               prometheusRegistry,
 BlockFileTransfer:                blockFileTransfer,
 Execer:                           execer,
 ContainerLister:                  containerLister,
 ExperimentalDevcontainersEnabled: experimentalDevcontainersEnabled,
 })

 promHandler := agent.PrometheusMetricsHandler(prometheusRegistry, logger)
 prometheusSrvClose := ServeHandler(ctx, logger, promHandler, prometheusAddress, "prometheus")

 debugSrvClose := ServeHandler(ctx, logger, agnt.HTTPDebug(), debugAddress, "debug")

 select {
 case <-ctx.Done():
 logger.Warn(ctx, "agent shutting down", slog.Error(ctx.Err()), slog.F("cause", context.Cause(ctx)))
 mustExit = true
 case event := <-reinitEvents:
 logger.Warn(ctx, "agent received instruction to reinitialize",
 slog.F("message", event.Message), slog.F("reason", event.Reason))
 }

 lastErr = agnt.Close()
 debugSrvClose()
 prometheusSrvClose()

 if mustExit {
 reinitCancel()
 break
 }

 logger.Info(ctx, "agent reinitializing")
 }
 return lastErr
 },
 }

diff --git a/cli/agent_test.go b/cli/agent_test.go
 "github.com/coder/coder/v2/coderd/coderdtest"
 "github.com/coder/coder/v2/coderd/database"
 "github.com/coder/coder/v2/coderd/database/dbfake"
 "github.com/coder/coder/v2/coderd/database/dbtestutil"
 "github.com/coder/coder/v2/codersdk"
 "github.com/coder/coder/v2/codersdk/agentsdk"
 "github.com/coder/coder/v2/codersdk/workspacesdk"
 "github.com/coder/coder/v2/provisionersdk/proto"
 "github.com/coder/coder/v2/testutil"
 })
 }

 func TestAgent_Prebuild(t *testing.T) {
 t.Parallel()

 db, pubsub := dbtestutil.NewDB(t)
 client := coderdtest.New(t, &coderdtest.Options{
 Database: db,
 Pubsub:   pubsub,
 })
 user := coderdtest.CreateFirstUser(t, client)
 r := dbfake.WorkspaceBuild(t, db, database.WorkspaceTable{
 OrganizationID: user.OrganizationID,
 OwnerID:        user.UserID,
 }).WithAgent(func(a []*proto.Agent) []*proto.Agent {
 a[0].Scripts = []*proto.Script{
 {
 DisplayName: "Prebuild Test Script",
 Script:      "sleep 5", // Make reinitiazation take long enough to assert that it happened
 RunOnStart:  true,
 },
 }
 return a
 }).Do()

 // Spin up an agent
 logDir := t.TempDir()
 inv, _ := clitest.New(t,
 "agent",
 "--auth", "token",
 "--agent-token", r.AgentToken,
 "--agent-url", client.URL.String(),
 "--log-dir", logDir,
 )
 clitest.Start(t, inv)

 // Check that the agent is in a happy steady state
 waiter := coderdtest.NewWorkspaceAgentWaiter(t, client, r.Workspace.ID)
 waiter.WaitFor(coderdtest.AgentsReady)

 // Trigger reinitialization
 channel := agentsdk.PrebuildClaimedChannel(r.Workspace.ID)
 err := pubsub.Publish(channel, []byte(user.UserID.String()))
 require.NoError(t, err)

 // Check that the agent reinitializes
 waiter.WaitFor(coderdtest.AgentsNotReady)

 // Check that reinitialization completed
 waiter.WaitFor(coderdtest.AgentsReady)
 }

 func matchAgentWithVersion(rs []codersdk.WorkspaceResource) bool {
 if len(rs) < 1 {
 return false
diff --git a/coderd/apidoc/docs.go b/coderd/apidoc/docs.go
Original file line number	Diff line number	Diff line change
Expand Up		@@ -36,6 +36,9 @@ import (
		"tailscale.com/util/clientmetric"

		"cdr.dev/slog"

		"github.com/coder/retry"
SasSwart marked this conversation as resolved. OutdatedShow resolvedHide resolved

		"github.com/coder/clistat"
		"github.com/coder/coder/v2/agent/agentcontainers"
		"github.com/coder/coder/v2/agent/agentexec"
Expand All		@@ -53,7 +56,6 @@ import (
		"github.com/coder/coder/v2/tailnet"
		tailnetproto "github.com/coder/coder/v2/tailnet/proto"
		"github.com/coder/quartz"
		"github.com/coder/retry"
		)

		const (
Expand DownExpand Up		@@ -365,9 +367,11 @@ func (a *agent) runLoop() {
		if ctx.Err() != nil {
		// Context canceled errors may come from websocket pings, so we
		// don't want to use `errors.Is(err, context.Canceled)` here.
		a.logger.Warn(ctx, "runLoop exited with error", slog.Error(ctx.Err()))
		return
		}
		if a.isClosed() {
		a.logger.Warn(ctx, "runLoop exited because agent is closed")
		return
		}
		if errors.Is(err, io.EOF) {
Expand DownExpand Up		@@ -1048,7 +1052,11 @@ func (a *agent) run() (retErr error) {
		return a.statsReporter.reportLoop(ctx, aAPI)
		})

		return connMan.wait()
		err = connMan.wait()
		if err != nil {
		a.logger.Warn(context.Background(), "connection manager errored", slog.Error(err))
SasSwart marked this conversation as resolved. OutdatedShow resolvedHide resolved
		}
		return err
		}

		// handleManifest returns a function that fetches and processes the manifest
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -3,6 +3,7 @@
		package reaper

		import (
		"fmt"
		"os"
		"os/signal"
		"syscall"
Expand All		@@ -29,6 +30,10 @@ func catchSignals(pid int, sigs []os.Signal) {
		s := <-sc
		sig, ok := s.(syscall.Signal)
		if ok {
		// TODO:
		// Tried using a logger here but the I/O streams are already closed at this point...
		// Why is os.Stderr still working then?
		_, _ = fmt.Fprintf(os.Stderr, "reaper caught %q signal, killing process %v\n", sig.String(), pid)
		_ = syscall.Kill(pid, sig)
		}
		}
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -19,12 +19,16 @@ import (
		"golang.org/x/xerrors"
		"gopkg.in/natefinch/lumberjack.v2"

		"github.com/coder/retry"

		"github.com/prometheus/client_golang/prometheus"

		"cdr.dev/slog"
		"cdr.dev/slog/sloggers/sloghuman"
		"cdr.dev/slog/sloggers/slogjson"
		"cdr.dev/slog/sloggers/slogstackdriver"
		"github.com/coder/serpent"

		"github.com/coder/coder/v2/agent"
		"github.com/coder/coder/v2/agent/agentcontainers"
		"github.com/coder/coder/v2/agent/agentexec"
Expand All		@@ -34,7 +38,6 @@ import (
		"github.com/coder/coder/v2/cli/clilog"
		"github.com/coder/coder/v2/codersdk"
		"github.com/coder/coder/v2/codersdk/agentsdk"
		"github.com/coder/serpent"
		)

		func (r RootCmd) workspaceAgent() serpent.Command {
Expand DownExpand Up		@@ -63,8 +66,10 @@ func (r RootCmd) workspaceAgent() serpent.Command {
		// This command isn't useful to manually execute.
		Hidden: true,
		Handler: func(inv *serpent.Invocation) error {
		ctx, cancel := context.WithCancel(inv.Context())
		defer cancel()
		ctx, cancel := context.WithCancelCause(inv.Context())
		defer func() {
		cancel(xerrors.New("agent exited"))
		}()

		var (
		ignorePorts = map[int]string{}
Expand DownExpand Up		@@ -281,7 +286,6 @@ func (r RootCmd) workspaceAgent() serpent.Command {
		return xerrors.Errorf("add executable to $PATH: %w", err)
		}

		prometheusRegistry := prometheus.NewRegistry()
		subsystemsRaw := inv.Environ.Get(agent.EnvAgentSubsystem)
		subsystems := []codersdk.AgentSubsystem{}
		for _, s := range strings.Split(subsystemsRaw, ",") {
Expand DownExpand Up		@@ -328,46 +332,90 @@ func (r RootCmd) workspaceAgent() serpent.Command {
		containerLister = agentcontainers.NewDocker(execer)
		}

		agnt := agent.New(agent.Options{
		Client: client,
		Logger: logger,
		LogDir: logDir,
		ScriptDataDir: scriptDataDir,
		// #nosec G115 - Safe conversion as tailnet listen port is within uint16 range (0-65535)
		TailnetListenPort: uint16(tailnetListenPort),
		ExchangeToken: func(ctx context.Context) (string, error) {
		if exchangeToken == nil {
		return client.SDK.SessionToken(), nil
		// TODO: timeout ok?
		reinitCtx, reinitCancel := context.WithTimeout(context.Background(), time.Hour*24)
		defer reinitCancel()
		reinitEvents := make(chan agentsdk.ReinitializationEvent)

		go func() {
		// Retry to wait for reinit, main context cancels the retrier.
		for retrier := retry.New(100time.Millisecond, 10time.Second); retrier.Wait(ctx); {
		select {
		case <-reinitCtx.Done():
		return
		default:
		}
		resp, err := exchangeToken(ctx)

		err := client.WaitForReinit(reinitCtx, reinitEvents)
		if err != nil {
		return "", err
		logger.Error(ctx, "failed to wait for reinit instructions, will retry",slog.Error(err))
		}
		client.SetSessionToken(resp.SessionToken)
		return resp.SessionToken, nil
		},
		EnvironmentVariables: environmentVariables,
		IgnorePorts: ignorePorts,
		SSHMaxTimeout: sshMaxTimeout,
		Subsystems: subsystems,

		PrometheusRegistry: prometheusRegistry,
		BlockFileTransfer: blockFileTransfer,
		Execer: execer,
		ContainerLister: containerLister,

		ExperimentalDevcontainersEnabled: experimentalDevcontainersEnabled,
		})

		promHandler := agent.PrometheusMetricsHandler(prometheusRegistry, logger)
		prometheusSrvClose := ServeHandler(ctx, logger, promHandler, prometheusAddress, "prometheus")
		defer prometheusSrvClose()

		debugSrvClose := ServeHandler(ctx, logger, agnt.HTTPDebug(), debugAddress, "debug")
		defer debugSrvClose()

		<-ctx.Done()
		return agnt.Close()
		}
		}()

		var (
		lastErr error
		mustExit bool
		)
		for {
		prometheusRegistry := prometheus.NewRegistry()

		agnt := agent.New(agent.Options{
		Client: client,
		Logger: logger,
		LogDir: logDir,
		ScriptDataDir: scriptDataDir,
		// #nosec G115 - Safe conversion as tailnet listen port is within uint16 range (0-65535)
		TailnetListenPort: uint16(tailnetListenPort),
		ExchangeToken: func(ctx context.Context) (string, error) {
		if exchangeToken == nil {
		return client.SDK.SessionToken(), nil
		}
		resp, err := exchangeToken(ctx)
		if err != nil {
		return "", err
		}
		client.SetSessionToken(resp.SessionToken)
		return resp.SessionToken, nil
		},
		EnvironmentVariables: environmentVariables,
		IgnorePorts: ignorePorts,
		SSHMaxTimeout: sshMaxTimeout,
		Subsystems: subsystems,

		PrometheusRegistry: prometheusRegistry,
		BlockFileTransfer: blockFileTransfer,
		Execer: execer,
		ContainerLister: containerLister,
		ExperimentalDevcontainersEnabled: experimentalDevcontainersEnabled,
		})

		promHandler := agent.PrometheusMetricsHandler(prometheusRegistry, logger)
		prometheusSrvClose := ServeHandler(ctx, logger, promHandler, prometheusAddress, "prometheus")

		debugSrvClose := ServeHandler(ctx, logger, agnt.HTTPDebug(), debugAddress, "debug")

		select {
		case <-ctx.Done():
		logger.Warn(ctx, "agent shutting down", slog.Error(ctx.Err()), slog.F("cause", context.Cause(ctx)))
		mustExit = true
		case event := <-reinitEvents:
		logger.Warn(ctx, "agent received instruction to reinitialize",
		slog.F("message", event.Message), slog.F("reason", event.Reason))
		}

		lastErr = agnt.Close()
		debugSrvClose()
		prometheusSrvClose()

		if mustExit {
		reinitCancel()
		break
		}

		logger.Info(ctx, "agent reinitializing")
		}
		return lastErr
		},
		}

Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -21,7 +21,9 @@ import (
		"github.com/coder/coder/v2/coderd/coderdtest"
		"github.com/coder/coder/v2/coderd/database"
		"github.com/coder/coder/v2/coderd/database/dbfake"
		"github.com/coder/coder/v2/coderd/database/dbtestutil"
		"github.com/coder/coder/v2/codersdk"
		"github.com/coder/coder/v2/codersdk/agentsdk"
		"github.com/coder/coder/v2/codersdk/workspacesdk"
		"github.com/coder/coder/v2/provisionersdk/proto"
		"github.com/coder/coder/v2/testutil"
Expand DownExpand Up		@@ -321,6 +323,56 @@ func TestWorkspaceAgent(t *testing.T) {
		})
		}

		func TestAgent_Prebuild(t *testing.T) {
		t.Parallel()

		db, pubsub := dbtestutil.NewDB(t)
		client := coderdtest.New(t, &coderdtest.Options{
		Database: db,
		Pubsub: pubsub,
		})
		user := coderdtest.CreateFirstUser(t, client)
		r := dbfake.WorkspaceBuild(t, db, database.WorkspaceTable{
		OrganizationID: user.OrganizationID,
		OwnerID: user.UserID,
		}).WithAgent(func(a []proto.Agent) []proto.Agent {
		a[0].Scripts = []*proto.Script{
		{
		DisplayName: "Prebuild Test Script",
		Script: "sleep 5", // Make reinitiazation take long enough to assert that it happened
		RunOnStart: true,
		},
		}
		return a
		}).Do()

		// Spin up an agent
		logDir := t.TempDir()
		inv, _ := clitest.New(t,
		"agent",
		"--auth", "token",
		"--agent-token", r.AgentToken,
		"--agent-url", client.URL.String(),
		"--log-dir", logDir,
		)
		clitest.Start(t, inv)

		// Check that the agent is in a happy steady state
		waiter := coderdtest.NewWorkspaceAgentWaiter(t, client, r.Workspace.ID)
		waiter.WaitFor(coderdtest.AgentsReady)

		// Trigger reinitialization
		channel := agentsdk.PrebuildClaimedChannel(r.Workspace.ID)
		err := pubsub.Publish(channel, []byte(user.UserID.String()))
		require.NoError(t, err)

		// Check that the agent reinitializes
		waiter.WaitFor(coderdtest.AgentsNotReady)

		// Check that reinitialization completed
		waiter.WaitFor(coderdtest.AgentsReady)
		}

		func matchAgentWithVersion(rs []codersdk.WorkspaceResource) bool {
		if len(rs) < 1 {
		return false
Expand Down