feat: add agent metadata (#6614)

This commit is contained in:
Ammar Bandukwala
2023-03-31 15:26:19 -05:00
committed by GitHub
parent c191692751
commit ca4fa81570
62 changed files with 3139 additions and 727 deletions
+200 -37
View File
@@ -2,12 +2,14 @@ package agent
import (
"bufio"
"bytes"
"context"
"crypto/rand"
"crypto/rsa"
"encoding/binary"
"encoding/json"
"errors"
"flag"
"fmt"
"io"
"net"
@@ -33,6 +35,7 @@ import (
"go.uber.org/atomic"
gossh "golang.org/x/crypto/ssh"
"golang.org/x/exp/slices"
"golang.org/x/sync/singleflight"
"golang.org/x/xerrors"
"tailscale.com/net/speedtest"
"tailscale.com/tailcfg"
@@ -83,12 +86,13 @@ type Options struct {
}
type Client interface {
Metadata(ctx context.Context) (agentsdk.Metadata, error)
Manifest(ctx context.Context) (agentsdk.Manifest, error)
Listen(ctx context.Context) (net.Conn, error)
ReportStats(ctx context.Context, log slog.Logger, statsChan <-chan *agentsdk.Stats, setInterval func(time.Duration)) (io.Closer, error)
PostLifecycle(ctx context.Context, state agentsdk.PostLifecycleRequest) error
PostAppHealth(ctx context.Context, req agentsdk.PostAppHealthsRequest) error
PostStartup(ctx context.Context, req agentsdk.PostStartupRequest) error
PostMetadata(ctx context.Context, key string, req agentsdk.PostMetadataRequest) error
PatchStartupLogs(ctx context.Context, req agentsdk.PatchStartupLogs) error
}
@@ -156,8 +160,8 @@ type agent struct {
closed chan struct{}
envVars map[string]string
// metadata is atomic because values can change after reconnection.
metadata atomic.Value
// manifest is atomic because values can change after reconnection.
manifest atomic.Pointer[agentsdk.Manifest]
sessionToken atomic.Pointer[string]
sshServer *ssh.Server
sshMaxTimeout time.Duration
@@ -183,6 +187,7 @@ type agent struct {
// failure, you'll want the agent to reconnect.
func (a *agent) runLoop(ctx context.Context) {
go a.reportLifecycleLoop(ctx)
go a.reportMetadataLoop(ctx)
for retrier := retry.New(100*time.Millisecond, 10*time.Second); retrier.Wait(ctx); {
a.logger.Info(ctx, "connecting to coderd")
@@ -205,6 +210,168 @@ func (a *agent) runLoop(ctx context.Context) {
}
}
func (a *agent) collectMetadata(ctx context.Context, md codersdk.WorkspaceAgentMetadataDescription) *codersdk.WorkspaceAgentMetadataResult {
var out bytes.Buffer
result := &codersdk.WorkspaceAgentMetadataResult{
// CollectedAt is set here for testing purposes and overrode by
// the server to the time the server received the result to protect
// against clock skew.
//
// In the future, the server may accept the timestamp from the agent
// if it is certain the clocks are in sync.
CollectedAt: time.Now(),
}
cmd, err := a.createCommand(ctx, md.Script, nil)
if err != nil {
result.Error = err.Error()
return result
}
cmd.Stdout = &out
cmd.Stderr = &out
// The error isn't mutually exclusive with useful output.
err = cmd.Run()
const bufLimit = 10 << 10
if out.Len() > bufLimit {
err = errors.Join(
err,
xerrors.Errorf("output truncated from %v to %v bytes", out.Len(), bufLimit),
)
out.Truncate(bufLimit)
}
if err != nil {
result.Error = err.Error()
}
result.Value = out.String()
return result
}
func adjustIntervalForTests(i int64) time.Duration {
// In tests we want to set shorter intervals because engineers are
// impatient.
base := time.Second
if flag.Lookup("test.v") != nil {
base = time.Millisecond * 100
}
return time.Duration(i) * base
}
type metadataResultAndKey struct {
result *codersdk.WorkspaceAgentMetadataResult
key string
}
func (a *agent) reportMetadataLoop(ctx context.Context) {
baseInterval := adjustIntervalForTests(1)
const metadataLimit = 128
var (
baseTicker = time.NewTicker(baseInterval)
lastCollectedAts = make(map[string]time.Time)
metadataResults = make(chan metadataResultAndKey, metadataLimit)
)
defer baseTicker.Stop()
var flight singleflight.Group
for {
select {
case <-ctx.Done():
return
case mr := <-metadataResults:
lastCollectedAts[mr.key] = mr.result.CollectedAt
err := a.client.PostMetadata(ctx, mr.key, *mr.result)
if err != nil {
a.logger.Error(ctx, "report metadata", slog.Error(err))
}
case <-baseTicker.C:
}
if len(metadataResults) > 0 {
// The inner collection loop expects the channel is empty before spinning up
// all the collection goroutines.
a.logger.Debug(
ctx, "metadata collection backpressured",
slog.F("queue_len", len(metadataResults)),
)
continue
}
manifest := a.manifest.Load()
if manifest == nil {
continue
}
if len(manifest.Metadata) > metadataLimit {
a.logger.Error(
ctx, "metadata limit exceeded",
slog.F("limit", metadataLimit), slog.F("got", len(manifest.Metadata)),
)
continue
}
// If the manifest changes (e.g. on agent reconnect) we need to
// purge old cache values to prevent lastCollectedAt from growing
// boundlessly.
for key := range lastCollectedAts {
if slices.IndexFunc(manifest.Metadata, func(md codersdk.WorkspaceAgentMetadataDescription) bool {
return md.Key == key
}) < 0 {
delete(lastCollectedAts, key)
}
}
// Spawn a goroutine for each metadata collection, and use a
// channel to synchronize the results and avoid both messy
// mutex logic and overloading the API.
for _, md := range manifest.Metadata {
collectedAt, ok := lastCollectedAts[md.Key]
if ok {
// If the interval is zero, we assume the user just wants
// a single collection at startup, not a spinning loop.
if md.Interval == 0 {
continue
}
// The last collected value isn't quite stale yet, so we skip it.
if collectedAt.Add(
adjustIntervalForTests(md.Interval),
).After(time.Now()) {
continue
}
}
md := md
// We send the result to the channel in the goroutine to avoid
// sending the same result multiple times. So, we don't care about
// the return values.
flight.DoChan(md.Key, func() (interface{}, error) {
timeout := md.Timeout
if timeout == 0 {
timeout = md.Interval
}
ctx, cancel := context.WithTimeout(ctx,
time.Duration(timeout)*time.Second,
)
defer cancel()
select {
case <-ctx.Done():
return 0, nil
case metadataResults <- metadataResultAndKey{
key: md.Key,
result: a.collectMetadata(ctx, md),
}:
}
return 0, nil
})
}
}
}
// reportLifecycleLoop reports the current lifecycle state once.
// Only the latest state is reported, intermediate states may be
// lost if the agent can't communicate with the API.
@@ -279,40 +446,40 @@ func (a *agent) run(ctx context.Context) error {
}
a.sessionToken.Store(&sessionToken)
metadata, err := a.client.Metadata(ctx)
manifest, err := a.client.Manifest(ctx)
if err != nil {
return xerrors.Errorf("fetch metadata: %w", err)
}
a.logger.Info(ctx, "fetched metadata", slog.F("metadata", metadata))
a.logger.Info(ctx, "fetched manifest", slog.F("manifest", manifest))
// Expand the directory and send it back to coderd so external
// applications that rely on the directory can use it.
//
// An example is VS Code Remote, which must know the directory
// before initializing a connection.
metadata.Directory, err = expandDirectory(metadata.Directory)
manifest.Directory, err = expandDirectory(manifest.Directory)
if err != nil {
return xerrors.Errorf("expand directory: %w", err)
}
err = a.client.PostStartup(ctx, agentsdk.PostStartupRequest{
Version: buildinfo.Version(),
ExpandedDirectory: metadata.Directory,
ExpandedDirectory: manifest.Directory,
})
if err != nil {
return xerrors.Errorf("update workspace agent version: %w", err)
}
oldMetadata := a.metadata.Swap(metadata)
oldManifest := a.manifest.Swap(&manifest)
// The startup script should only execute on the first run!
if oldMetadata == nil {
if oldManifest == nil {
a.setLifecycle(ctx, codersdk.WorkspaceAgentLifecycleStarting)
// Perform overrides early so that Git auth can work even if users
// connect to a workspace that is not yet ready. We don't run this
// concurrently with the startup script to avoid conflicts between
// them.
if metadata.GitAuthConfigs > 0 {
if manifest.GitAuthConfigs > 0 {
// If this fails, we should consider surfacing the error in the
// startup log and setting the lifecycle state to be "start_error"
// (after startup script completion), but for now we'll just log it.
@@ -327,7 +494,7 @@ func (a *agent) run(ctx context.Context) error {
scriptStart := time.Now()
err = a.trackConnGoroutine(func() {
defer close(scriptDone)
scriptDone <- a.runStartupScript(ctx, metadata.StartupScript)
scriptDone <- a.runStartupScript(ctx, manifest.StartupScript)
})
if err != nil {
return xerrors.Errorf("track startup script: %w", err)
@@ -336,8 +503,8 @@ func (a *agent) run(ctx context.Context) error {
var timeout <-chan time.Time
// If timeout is zero, an older version of the coder
// provider was used. Otherwise a timeout is always > 0.
if metadata.StartupScriptTimeout > 0 {
t := time.NewTimer(metadata.StartupScriptTimeout)
if manifest.StartupScriptTimeout > 0 {
t := time.NewTimer(manifest.StartupScriptTimeout)
defer t.Stop()
timeout = t.C
}
@@ -354,7 +521,7 @@ func (a *agent) run(ctx context.Context) error {
return
}
// Only log if there was a startup script.
if metadata.StartupScript != "" {
if manifest.StartupScript != "" {
execTime := time.Since(scriptStart)
if err != nil {
a.logger.Warn(ctx, "startup script failed", slog.F("execution_time", execTime), slog.Error(err))
@@ -371,13 +538,13 @@ func (a *agent) run(ctx context.Context) error {
appReporterCtx, appReporterCtxCancel := context.WithCancel(ctx)
defer appReporterCtxCancel()
go NewWorkspaceAppHealthReporter(
a.logger, metadata.Apps, a.client.PostAppHealth)(appReporterCtx)
a.logger, manifest.Apps, a.client.PostAppHealth)(appReporterCtx)
a.closeMutex.Lock()
network := a.network
a.closeMutex.Unlock()
if network == nil {
network, err = a.createTailnet(ctx, metadata.DERPMap)
network, err = a.createTailnet(ctx, manifest.DERPMap)
if err != nil {
return xerrors.Errorf("create tailnet: %w", err)
}
@@ -396,7 +563,7 @@ func (a *agent) run(ctx context.Context) error {
a.startReportingConnectionStats(ctx)
} else {
// Update the DERP map!
network.SetDERPMap(metadata.DERPMap)
network.SetDERPMap(manifest.DERPMap)
}
a.logger.Debug(ctx, "running tailnet connection coordinator")
@@ -926,9 +1093,9 @@ func (a *agent) init(ctx context.Context) {
}
// createCommand processes raw command input with OpenSSH-like behavior.
// If the rawCommand provided is empty, it will default to the users shell.
// If the script provided is empty, it will default to the users shell.
// This injects environment variables specified by the user at launch too.
func (a *agent) createCommand(ctx context.Context, rawCommand string, env []string) (*exec.Cmd, error) {
func (a *agent) createCommand(ctx context.Context, script string, env []string) (*exec.Cmd, error) {
currentUser, err := user.Current()
if err != nil {
return nil, xerrors.Errorf("get current user: %w", err)
@@ -940,14 +1107,10 @@ func (a *agent) createCommand(ctx context.Context, rawCommand string, env []stri
return nil, xerrors.Errorf("get user shell: %w", err)
}
rawMetadata := a.metadata.Load()
if rawMetadata == nil {
manifest := a.manifest.Load()
if manifest == nil {
return nil, xerrors.Errorf("no metadata was provided")
}
metadata, valid := rawMetadata.(agentsdk.Metadata)
if !valid {
return nil, xerrors.Errorf("metadata is the wrong type: %T", metadata)
}
// OpenSSH executes all commands with the users current shell.
// We replicate that behavior for IDE support.
@@ -955,11 +1118,11 @@ func (a *agent) createCommand(ctx context.Context, rawCommand string, env []stri
if runtime.GOOS == "windows" {
caller = "/c"
}
args := []string{caller, rawCommand}
args := []string{caller, script}
// gliderlabs/ssh returns a command slice of zero
// when a shell is requested.
if len(rawCommand) == 0 {
if len(script) == 0 {
args = []string{}
if runtime.GOOS != "windows" {
// On Linux and macOS, we should start a login
@@ -969,7 +1132,7 @@ func (a *agent) createCommand(ctx context.Context, rawCommand string, env []stri
}
cmd := exec.CommandContext(ctx, shell, args...)
cmd.Dir = metadata.Directory
cmd.Dir = manifest.Directory
// If the metadata directory doesn't exist, we run the command
// in the users home directory.
@@ -1010,14 +1173,14 @@ func (a *agent) createCommand(ctx context.Context, rawCommand string, env []stri
// This adds the ports dialog to code-server that enables
// proxying a port dynamically.
cmd.Env = append(cmd.Env, fmt.Sprintf("VSCODE_PROXY_URI=%s", metadata.VSCodePortProxyURI))
cmd.Env = append(cmd.Env, fmt.Sprintf("VSCODE_PROXY_URI=%s", manifest.VSCodePortProxyURI))
// Hide Coder message on code-server's "Getting Started" page
cmd.Env = append(cmd.Env, "CS_DISABLE_GETTING_STARTED_OVERRIDE=true")
// Load environment variables passed via the agent.
// These should override all variables we manually specify.
for envKey, value := range metadata.EnvironmentVariables {
for envKey, value := range manifest.EnvironmentVariables {
// Expanding environment variables allows for customization
// of the $PATH, among other variables. Customers can prepend
// or append to the $PATH, so allowing expand is required!
@@ -1080,9 +1243,9 @@ func (a *agent) handleSSHSession(session ssh.Session) (retErr error) {
session.DisablePTYEmulation()
if !isQuietLogin(session.RawCommand()) {
metadata, ok := a.metadata.Load().(agentsdk.Metadata)
if ok {
err = showMOTD(session, metadata.MOTDFile)
manifest := a.manifest.Load()
if manifest != nil {
err = showMOTD(session, manifest.MOTDFile)
if err != nil {
a.logger.Error(ctx, "show MOTD", slog.Error(err))
}
@@ -1512,19 +1675,19 @@ func (a *agent) Close() error {
a.setLifecycle(ctx, codersdk.WorkspaceAgentLifecycleShuttingDown)
lifecycleState := codersdk.WorkspaceAgentLifecycleOff
if metadata, ok := a.metadata.Load().(agentsdk.Metadata); ok && metadata.ShutdownScript != "" {
if manifest := a.manifest.Load(); manifest != nil && manifest.ShutdownScript != "" {
scriptDone := make(chan error, 1)
scriptStart := time.Now()
go func() {
defer close(scriptDone)
scriptDone <- a.runShutdownScript(ctx, metadata.ShutdownScript)
scriptDone <- a.runShutdownScript(ctx, manifest.ShutdownScript)
}()
var timeout <-chan time.Time
// If timeout is zero, an older version of the coder
// provider was used. Otherwise a timeout is always > 0.
if metadata.ShutdownScriptTimeout > 0 {
t := time.NewTimer(metadata.ShutdownScriptTimeout)
if manifest.ShutdownScriptTimeout > 0 {
t := time.NewTimer(manifest.ShutdownScriptTimeout)
defer t.Stop()
timeout = t.C
}
+172 -42
View File
@@ -33,6 +33,7 @@ import (
"github.com/stretchr/testify/require"
"go.uber.org/goleak"
"golang.org/x/crypto/ssh"
"golang.org/x/exp/maps"
"golang.org/x/xerrors"
"tailscale.com/net/speedtest"
"tailscale.com/tailcfg"
@@ -61,7 +62,7 @@ func TestAgent_Stats_SSH(t *testing.T) {
defer cancel()
//nolint:dogsled
conn, _, stats, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, stats, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
sshClient, err := conn.SSHClient(ctx)
require.NoError(t, err)
@@ -94,7 +95,7 @@ func TestAgent_Stats_ReconnectingPTY(t *testing.T) {
defer cancel()
//nolint:dogsled
conn, _, stats, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, stats, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
ptyConn, err := conn.ReconnectingPTY(ctx, uuid.New(), 128, 128, "/bin/bash")
require.NoError(t, err)
@@ -124,7 +125,7 @@ func TestAgent_Stats_Magic(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
defer cancel()
//nolint:dogsled
conn, _, _, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, _, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
sshClient, err := conn.SSHClient(ctx)
require.NoError(t, err)
defer sshClient.Close()
@@ -151,7 +152,7 @@ func TestAgent_Stats_Magic(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
defer cancel()
//nolint:dogsled
conn, _, stats, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, stats, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
sshClient, err := conn.SSHClient(ctx)
require.NoError(t, err)
defer sshClient.Close()
@@ -186,7 +187,7 @@ func TestAgent_Stats_Magic(t *testing.T) {
func TestAgent_SessionExec(t *testing.T) {
t.Parallel()
session := setupSSHSession(t, agentsdk.Metadata{})
session := setupSSHSession(t, agentsdk.Manifest{})
command := "echo test"
if runtime.GOOS == "windows" {
@@ -199,7 +200,7 @@ func TestAgent_SessionExec(t *testing.T) {
func TestAgent_GitSSH(t *testing.T) {
t.Parallel()
session := setupSSHSession(t, agentsdk.Metadata{})
session := setupSSHSession(t, agentsdk.Manifest{})
command := "sh -c 'echo $GIT_SSH_COMMAND'"
if runtime.GOOS == "windows" {
command = "cmd.exe /c echo %GIT_SSH_COMMAND%"
@@ -219,7 +220,7 @@ func TestAgent_SessionTTYShell(t *testing.T) {
// it seems like it could be either.
t.Skip("ConPTY appears to be inconsistent on Windows.")
}
session := setupSSHSession(t, agentsdk.Metadata{})
session := setupSSHSession(t, agentsdk.Manifest{})
command := "sh"
if runtime.GOOS == "windows" {
command = "cmd.exe"
@@ -242,7 +243,7 @@ func TestAgent_SessionTTYShell(t *testing.T) {
func TestAgent_SessionTTYExitCode(t *testing.T) {
t.Parallel()
session := setupSSHSession(t, agentsdk.Metadata{})
session := setupSSHSession(t, agentsdk.Manifest{})
command := "areallynotrealcommand"
err := session.RequestPty("xterm", 128, 128, ssh.TerminalModes{})
require.NoError(t, err)
@@ -281,7 +282,7 @@ func TestAgent_Session_TTY_MOTD(t *testing.T) {
// Set HOME so we can ensure no ~/.hushlogin is present.
t.Setenv("HOME", tmpdir)
session := setupSSHSession(t, agentsdk.Metadata{
session := setupSSHSession(t, agentsdk.Manifest{
MOTDFile: name,
})
err = session.RequestPty("xterm", 128, 128, ssh.TerminalModes{})
@@ -327,7 +328,7 @@ func TestAgent_Session_TTY_Hushlogin(t *testing.T) {
// Set HOME so we can ensure ~/.hushlogin is present.
t.Setenv("HOME", tmpdir)
session := setupSSHSession(t, agentsdk.Metadata{
session := setupSSHSession(t, agentsdk.Manifest{
MOTDFile: name,
})
err = session.RequestPty("xterm", 128, 128, ssh.TerminalModes{})
@@ -357,7 +358,7 @@ func TestAgent_Session_TTY_FastCommandHasOutput(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
defer cancel()
//nolint:dogsled
conn, _, _, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, _, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
sshClient, err := conn.SSHClient(ctx)
require.NoError(t, err)
defer sshClient.Close()
@@ -407,7 +408,7 @@ func TestAgent_Session_TTY_HugeOutputIsNotLost(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
defer cancel()
//nolint:dogsled
conn, _, _, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, _, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
sshClient, err := conn.SSHClient(ctx)
require.NoError(t, err)
defer sshClient.Close()
@@ -706,7 +707,7 @@ func TestAgent_SFTP(t *testing.T) {
home = "/" + strings.ReplaceAll(home, "\\", "/")
}
//nolint:dogsled
conn, _, _, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, _, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
sshClient, err := conn.SSHClient(ctx)
require.NoError(t, err)
defer sshClient.Close()
@@ -738,7 +739,7 @@ func TestAgent_SCP(t *testing.T) {
defer cancel()
//nolint:dogsled
conn, _, _, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, _, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
sshClient, err := conn.SSHClient(ctx)
require.NoError(t, err)
defer sshClient.Close()
@@ -757,7 +758,7 @@ func TestAgent_EnvironmentVariables(t *testing.T) {
t.Parallel()
key := "EXAMPLE"
value := "value"
session := setupSSHSession(t, agentsdk.Metadata{
session := setupSSHSession(t, agentsdk.Manifest{
EnvironmentVariables: map[string]string{
key: value,
},
@@ -774,7 +775,7 @@ func TestAgent_EnvironmentVariables(t *testing.T) {
func TestAgent_EnvironmentVariableExpansion(t *testing.T) {
t.Parallel()
key := "EXAMPLE"
session := setupSSHSession(t, agentsdk.Metadata{
session := setupSSHSession(t, agentsdk.Manifest{
EnvironmentVariables: map[string]string{
key: "$SOMETHINGNOTSET",
},
@@ -801,7 +802,7 @@ func TestAgent_CoderEnvVars(t *testing.T) {
t.Run(key, func(t *testing.T) {
t.Parallel()
session := setupSSHSession(t, agentsdk.Metadata{})
session := setupSSHSession(t, agentsdk.Manifest{})
command := "sh -c 'echo $" + key + "'"
if runtime.GOOS == "windows" {
command = "cmd.exe /c echo %" + key + "%"
@@ -824,7 +825,7 @@ func TestAgent_SSHConnectionEnvVars(t *testing.T) {
t.Run(key, func(t *testing.T) {
t.Parallel()
session := setupSSHSession(t, agentsdk.Metadata{})
session := setupSSHSession(t, agentsdk.Manifest{})
command := "sh -c 'echo $" + key + "'"
if runtime.GOOS == "windows" {
command = "cmd.exe /c echo %" + key + "%"
@@ -848,7 +849,7 @@ func TestAgent_StartupScript(t *testing.T) {
client := &client{
t: t,
agentID: uuid.New(),
metadata: agentsdk.Metadata{
manifest: agentsdk.Manifest{
StartupScript: command,
DERPMap: &tailcfg.DERPMap{},
},
@@ -879,7 +880,7 @@ func TestAgent_StartupScript(t *testing.T) {
client := &client{
t: t,
agentID: uuid.New(),
metadata: agentsdk.Metadata{
manifest: agentsdk.Manifest{
StartupScript: command,
DERPMap: &tailcfg.DERPMap{},
},
@@ -912,13 +913,125 @@ func TestAgent_StartupScript(t *testing.T) {
})
}
func TestAgent_Metadata(t *testing.T) {
t.Parallel()
t.Run("Once", func(t *testing.T) {
t.Parallel()
script := "echo -n hello"
if runtime.GOOS == "windows" {
script = "powershell " + script
}
//nolint:dogsled
_, client, _, _, _ := setupAgent(t, agentsdk.Manifest{
Metadata: []codersdk.WorkspaceAgentMetadataDescription{
{
Key: "greeting",
Interval: 0,
Script: script,
},
},
}, 0)
var gotMd map[string]agentsdk.PostMetadataRequest
require.Eventually(t, func() bool {
gotMd = client.getMetadata()
return len(gotMd) == 1
}, testutil.WaitShort, testutil.IntervalMedium)
collectedAt := gotMd["greeting"].CollectedAt
require.Never(t, func() bool {
gotMd = client.getMetadata()
if len(gotMd) != 1 {
panic("unexpected number of metadata")
}
return !gotMd["greeting"].CollectedAt.Equal(collectedAt)
}, testutil.WaitShort, testutil.IntervalMedium)
})
t.Run("Many", func(t *testing.T) {
if runtime.GOOS == "windows" {
// Shell scripting in Windows is a pain, and we have already tested
// that the OS logic works in the simpler "Once" test above.
t.Skip()
}
t.Parallel()
dir := t.TempDir()
const reportInterval = 2
const intervalUnit = 100 * time.Millisecond
var (
greetingPath = filepath.Join(dir, "greeting")
script = "echo hello | tee -a " + greetingPath
)
_, client, _, _, _ := setupAgent(t, agentsdk.Manifest{
Metadata: []codersdk.WorkspaceAgentMetadataDescription{
{
Key: "greeting",
Interval: reportInterval,
Script: script,
},
{
Key: "bad",
Interval: reportInterval,
Script: "exit 1",
},
},
}, 0)
require.Eventually(t, func() bool {
return len(client.getMetadata()) == 2
}, testutil.WaitShort, testutil.IntervalMedium)
for start := time.Now(); time.Since(start) < testutil.WaitMedium; time.Sleep(testutil.IntervalMedium) {
md := client.getMetadata()
if len(md) != 2 {
panic("unexpected number of metadata entries")
}
require.Equal(t, "hello\n", md["greeting"].Value)
require.Equal(t, "exit status 1", md["bad"].Error)
greetingByt, err := os.ReadFile(greetingPath)
require.NoError(t, err)
var (
numGreetings = bytes.Count(greetingByt, []byte("hello"))
idealNumGreetings = time.Since(start) / (reportInterval * intervalUnit)
// We allow a 50% error margin because the report loop may backlog
// in CI and other toasters. In production, there is no hard
// guarantee on timing either, and the frontend gives similar
// wiggle room to the staleness of the value.
upperBound = int(idealNumGreetings) + 1
lowerBound = (int(idealNumGreetings) / 2)
)
if idealNumGreetings < 50 {
// There is an insufficient sample size.
continue
}
t.Logf("numGreetings: %d, idealNumGreetings: %d", numGreetings, idealNumGreetings)
// The report loop may slow down on load, but it should never, ever
// speed up.
if numGreetings > upperBound {
t.Fatalf("too many greetings: %d > %d in %v", numGreetings, upperBound, time.Since(start))
} else if numGreetings < lowerBound {
t.Fatalf("too few greetings: %d < %d", numGreetings, lowerBound)
}
}
})
}
func TestAgent_Lifecycle(t *testing.T) {
t.Parallel()
t.Run("StartTimeout", func(t *testing.T) {
t.Parallel()
_, client, _, _, _ := setupAgent(t, agentsdk.Metadata{
_, client, _, _, _ := setupAgent(t, agentsdk.Manifest{
StartupScript: "sleep 5",
StartupScriptTimeout: time.Nanosecond,
}, 0)
@@ -947,7 +1060,7 @@ func TestAgent_Lifecycle(t *testing.T) {
t.Run("StartError", func(t *testing.T) {
t.Parallel()
_, client, _, _, _ := setupAgent(t, agentsdk.Metadata{
_, client, _, _, _ := setupAgent(t, agentsdk.Manifest{
StartupScript: "false",
StartupScriptTimeout: 30 * time.Second,
}, 0)
@@ -976,7 +1089,7 @@ func TestAgent_Lifecycle(t *testing.T) {
t.Run("Ready", func(t *testing.T) {
t.Parallel()
_, client, _, _, _ := setupAgent(t, agentsdk.Metadata{
_, client, _, _, _ := setupAgent(t, agentsdk.Manifest{
StartupScript: "true",
StartupScriptTimeout: 30 * time.Second,
}, 0)
@@ -1005,7 +1118,7 @@ func TestAgent_Lifecycle(t *testing.T) {
t.Run("ShuttingDown", func(t *testing.T) {
t.Parallel()
_, client, _, _, closer := setupAgent(t, agentsdk.Metadata{
_, client, _, _, closer := setupAgent(t, agentsdk.Manifest{
ShutdownScript: "sleep 5",
StartupScriptTimeout: 30 * time.Second,
}, 0)
@@ -1043,7 +1156,7 @@ func TestAgent_Lifecycle(t *testing.T) {
t.Run("ShutdownTimeout", func(t *testing.T) {
t.Parallel()
_, client, _, _, closer := setupAgent(t, agentsdk.Metadata{
_, client, _, _, closer := setupAgent(t, agentsdk.Manifest{
ShutdownScript: "sleep 5",
ShutdownScriptTimeout: time.Nanosecond,
}, 0)
@@ -1090,7 +1203,7 @@ func TestAgent_Lifecycle(t *testing.T) {
t.Run("ShutdownError", func(t *testing.T) {
t.Parallel()
_, client, _, _, closer := setupAgent(t, agentsdk.Metadata{
_, client, _, _, closer := setupAgent(t, agentsdk.Manifest{
ShutdownScript: "false",
ShutdownScriptTimeout: 30 * time.Second,
}, 0)
@@ -1141,7 +1254,7 @@ func TestAgent_Lifecycle(t *testing.T) {
client := &client{
t: t,
agentID: uuid.New(),
metadata: agentsdk.Metadata{
manifest: agentsdk.Manifest{
DERPMap: tailnettest.RunDERPAndSTUN(t),
StartupScript: "echo 1",
ShutdownScript: "echo " + expected,
@@ -1194,7 +1307,7 @@ func TestAgent_Startup(t *testing.T) {
t.Run("EmptyDirectory", func(t *testing.T) {
t.Parallel()
_, client, _, _, _ := setupAgent(t, agentsdk.Metadata{
_, client, _, _, _ := setupAgent(t, agentsdk.Manifest{
StartupScript: "true",
StartupScriptTimeout: 30 * time.Second,
Directory: "",
@@ -1208,7 +1321,7 @@ func TestAgent_Startup(t *testing.T) {
t.Run("HomeDirectory", func(t *testing.T) {
t.Parallel()
_, client, _, _, _ := setupAgent(t, agentsdk.Metadata{
_, client, _, _, _ := setupAgent(t, agentsdk.Manifest{
StartupScript: "true",
StartupScriptTimeout: 30 * time.Second,
Directory: "~",
@@ -1224,7 +1337,7 @@ func TestAgent_Startup(t *testing.T) {
t.Run("HomeEnvironmentVariable", func(t *testing.T) {
t.Parallel()
_, client, _, _, _ := setupAgent(t, agentsdk.Metadata{
_, client, _, _, _ := setupAgent(t, agentsdk.Manifest{
StartupScript: "true",
StartupScriptTimeout: 30 * time.Second,
Directory: "$HOME",
@@ -1251,7 +1364,7 @@ func TestAgent_ReconnectingPTY(t *testing.T) {
defer cancel()
//nolint:dogsled
conn, _, _, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, _, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
id := uuid.New()
netConn, err := conn.ReconnectingPTY(ctx, id, 100, 100, "/bin/bash")
require.NoError(t, err)
@@ -1353,7 +1466,7 @@ func TestAgent_Dial(t *testing.T) {
}()
//nolint:dogsled
conn, _, _, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
conn, _, _, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
require.True(t, conn.AwaitReachable(context.Background()))
conn1, err := conn.DialContext(context.Background(), l.Addr().Network(), l.Addr().String())
require.NoError(t, err)
@@ -1375,7 +1488,7 @@ func TestAgent_Speedtest(t *testing.T) {
defer cancel()
derpMap := tailnettest.RunDERPAndSTUN(t)
//nolint:dogsled
conn, _, _, _, _ := setupAgent(t, agentsdk.Metadata{
conn, _, _, _, _ := setupAgent(t, agentsdk.Manifest{
DERPMap: derpMap,
}, 0)
defer conn.Close()
@@ -1397,7 +1510,7 @@ func TestAgent_Reconnect(t *testing.T) {
client := &client{
t: t,
agentID: agentID,
metadata: agentsdk.Metadata{
manifest: agentsdk.Manifest{
DERPMap: derpMap,
},
statsChan: statsCh,
@@ -1432,7 +1545,7 @@ func TestAgent_WriteVSCodeConfigs(t *testing.T) {
client := &client{
t: t,
agentID: uuid.New(),
metadata: agentsdk.Metadata{
manifest: agentsdk.Manifest{
GitAuthConfigs: 1,
DERPMap: &tailcfg.DERPMap{},
},
@@ -1461,7 +1574,7 @@ func TestAgent_WriteVSCodeConfigs(t *testing.T) {
func setupSSHCommand(t *testing.T, beforeArgs []string, afterArgs []string) *exec.Cmd {
//nolint:dogsled
agentConn, _, _, _, _ := setupAgent(t, agentsdk.Metadata{}, 0)
agentConn, _, _, _, _ := setupAgent(t, agentsdk.Manifest{}, 0)
listener, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
waitGroup := sync.WaitGroup{}
@@ -1504,7 +1617,7 @@ func setupSSHCommand(t *testing.T, beforeArgs []string, afterArgs []string) *exe
return exec.Command("ssh", args...)
}
func setupSSHSession(t *testing.T, options agentsdk.Metadata) *ssh.Session {
func setupSSHSession(t *testing.T, options agentsdk.Manifest) *ssh.Session {
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
defer cancel()
//nolint:dogsled
@@ -1528,7 +1641,7 @@ func (c closeFunc) Close() error {
return c()
}
func setupAgent(t *testing.T, metadata agentsdk.Metadata, ptyTimeout time.Duration) (
func setupAgent(t *testing.T, metadata agentsdk.Manifest, ptyTimeout time.Duration) (
*codersdk.WorkspaceAgentConn,
*client,
<-chan *agentsdk.Stats,
@@ -1548,7 +1661,7 @@ func setupAgent(t *testing.T, metadata agentsdk.Metadata, ptyTimeout time.Durati
c := &client{
t: t,
agentID: agentID,
metadata: metadata,
manifest: metadata,
statsChan: statsCh,
coordinator: coordinator,
}
@@ -1631,7 +1744,8 @@ func assertWritePayload(t *testing.T, w io.Writer, payload []byte) {
type client struct {
t *testing.T
agentID uuid.UUID
metadata agentsdk.Metadata
manifest agentsdk.Manifest
metadata map[string]agentsdk.PostMetadataRequest
statsChan chan *agentsdk.Stats
coordinator tailnet.Coordinator
lastWorkspaceAgent func()
@@ -1643,8 +1757,8 @@ type client struct {
logs []agentsdk.StartupLog
}
func (c *client) Metadata(_ context.Context) (agentsdk.Metadata, error) {
return c.metadata, nil
func (c *client) Manifest(_ context.Context) (agentsdk.Manifest, error) {
return c.manifest, nil
}
func (c *client) Listen(_ context.Context) (net.Conn, error) {
@@ -1718,6 +1832,22 @@ func (c *client) getStartup() agentsdk.PostStartupRequest {
return c.startup
}
func (c *client) getMetadata() map[string]agentsdk.PostMetadataRequest {
c.mu.Lock()
defer c.mu.Unlock()
return maps.Clone(c.metadata)
}
func (c *client) PostMetadata(_ context.Context, key string, req agentsdk.PostMetadataRequest) error {
c.mu.Lock()
defer c.mu.Unlock()
if c.metadata == nil {
c.metadata = make(map[string]agentsdk.PostMetadataRequest)
}
c.metadata[key] = req
return nil
}
func (c *client) PostStartup(_ context.Context, startup agentsdk.PostStartupRequest) error {
c.mu.Lock()
defer c.mu.Unlock()