258 lines
8.5 KiB
Go
258 lines
8.5 KiB
Go
package e2e
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"errors"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/coder/websocket"
|
|
"github.com/coder/websocket/wsjson"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/shcizo/package-updater/internal/agent/conn"
|
|
"github.com/shcizo/package-updater/internal/agent/updater"
|
|
"github.com/shcizo/package-updater/internal/agenttoken"
|
|
"github.com/shcizo/package-updater/internal/master/jobs"
|
|
"github.com/shcizo/package-updater/internal/master/model"
|
|
"github.com/shcizo/package-updater/internal/protocol"
|
|
)
|
|
|
|
const (
|
|
slusk = "gitea.shcizo.se/shcizo/slusk"
|
|
agentID = int64(2)
|
|
waitFor = 3 * time.Second
|
|
pollEvery = 10 * time.Millisecond
|
|
lostMessage = "lost: the agent did not know this job when it reconnected"
|
|
)
|
|
|
|
// gateExecutor writes a line of output, then blocks until release is closed
|
|
// or the job is cancelled.
|
|
type gateExecutor struct {
|
|
started chan string // receives the job's Ref when it begins
|
|
release chan struct{}
|
|
}
|
|
|
|
func newGateExecutor() *gateExecutor {
|
|
return &gateExecutor{started: make(chan string, 8), release: make(chan struct{})}
|
|
}
|
|
|
|
func (g *gateExecutor) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) {
|
|
_, _ = io.WriteString(out, "pulling "+job.Ref+"\n")
|
|
g.started <- job.Ref
|
|
select {
|
|
case <-g.release:
|
|
case <-ctx.Done():
|
|
return "", ctx.Err()
|
|
}
|
|
_, _ = io.WriteString(out, "recreated\n")
|
|
return job.Ref + "@sha256:e2e", nil
|
|
}
|
|
|
|
// jobAgent is a connected agent with a real runner and its two lifetimes: the
|
|
// connection and the runner can be stopped separately.
|
|
type jobAgent struct {
|
|
runner *updater.Runner
|
|
stopConn func()
|
|
stopRun func()
|
|
}
|
|
|
|
func startJobAgent(t *testing.T, m testMaster, token string, exec updater.Executor) *jobAgent {
|
|
t.Helper()
|
|
src := &staticSource{}
|
|
src.set(web)
|
|
runner := updater.NewRunner(exec, time.Minute)
|
|
runCtx, stopRun := context.WithCancel(context.Background())
|
|
runDone := make(chan struct{})
|
|
go func() { runner.Run(runCtx); close(runDone) }()
|
|
|
|
a := conn.New(conn.Config{URL: m.url, Token: token, Fingerprint: m.fingerprint, Version: "e2e",
|
|
MinBackoff: 10 * time.Millisecond, MaxBackoff: 50 * time.Millisecond, ProgressInterval: 10 * time.Millisecond,
|
|
}, src, runner, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
connCtx, cancelConn := context.WithCancel(context.Background())
|
|
connDone := make(chan struct{})
|
|
go func() { _ = a.Run(connCtx); close(connDone) }()
|
|
|
|
ja := &jobAgent{
|
|
runner: runner,
|
|
stopConn: func() { cancelConn(); <-connDone },
|
|
stopRun: func() { stopRun(); <-runDone },
|
|
}
|
|
t.Cleanup(func() { ja.stopConn(); ja.stopRun() })
|
|
return ja
|
|
}
|
|
|
|
// jobEnv is a master with one agent (ID 2, token "pua_ok") that owns a Compose
|
|
// service, and a jobs service wired to the master's current hub.
|
|
type jobEnv struct {
|
|
store *memStore
|
|
m testMaster
|
|
jobs *jobs.Service
|
|
exec *gateExecutor
|
|
agent *jobAgent
|
|
}
|
|
|
|
func newJobEnv(t *testing.T) *jobEnv {
|
|
t.Helper()
|
|
store := &memStore{agents: map[string]model.Agent{agenttoken.Hash("pua_ok"): {ID: agentID, Name: "node-2"}}}
|
|
store.seedService(model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
|
|
AgentID: ptr(agentID), ImageRepo: slusk, ActualRef: slusk + ":1.4.2@sha256:old"})
|
|
m := startMaster(t, store)
|
|
exec := newGateExecutor()
|
|
env := &jobEnv{store: store, m: m, exec: exec,
|
|
jobs: jobs.New(store, m.hub, m.broker, "gitea.shcizo.se/shcizo/package-updater:latest", slog.New(slog.NewTextHandler(io.Discard, nil))),
|
|
agent: startJobAgent(t, m, "pua_ok", exec)}
|
|
env.waitRoutable(t)
|
|
return env
|
|
}
|
|
|
|
func ptr[T any](v T) *T { return &v }
|
|
|
|
// waitRoutable waits until the master can send jobs to the agent.
|
|
func (e *jobEnv) waitRoutable(t *testing.T) {
|
|
t.Helper()
|
|
svc, err := e.store.GetService(context.Background(), 1)
|
|
require.NoError(t, err)
|
|
require.Eventually(t, func() bool { _, err := e.m.hub.Route(svc); return err == nil }, waitFor, pollEvery)
|
|
}
|
|
|
|
func (e *jobEnv) job(t *testing.T, id int64) model.Job {
|
|
t.Helper()
|
|
j, err := e.store.GetJob(context.Background(), id)
|
|
require.NoError(t, err)
|
|
return j
|
|
}
|
|
|
|
func (e *jobEnv) waitStatus(t *testing.T, id int64, want model.JobStatus) model.Job {
|
|
t.Helper()
|
|
require.Eventually(t, func() bool { return e.job(t, id).Status == want }, waitFor, pollEvery,
|
|
"job %d should become %s, is %s (%s)", id, want, e.job(t, id).Status, e.job(t, id).Error)
|
|
return e.job(t, id)
|
|
}
|
|
|
|
func (e *jobEnv) waitStarted(t *testing.T) {
|
|
t.Helper()
|
|
select {
|
|
case <-e.exec.started:
|
|
case <-time.After(waitFor):
|
|
t.Fatal("the agent never started the job")
|
|
}
|
|
}
|
|
|
|
func (e *jobEnv) trigger(t *testing.T) model.Job {
|
|
t.Helper()
|
|
j, err := e.jobs.Trigger(context.Background(), 1, "sub-1")
|
|
require.NoError(t, err)
|
|
return j
|
|
}
|
|
|
|
func TestManualJobFlowsToAgentAndBack(t *testing.T) {
|
|
e := newJobEnv(t)
|
|
close(e.exec.release)
|
|
|
|
j := e.trigger(t)
|
|
done := e.waitStatus(t, j.ID, model.JobSucceeded)
|
|
|
|
assert.Equal(t, slusk+":1.4.2@sha256:e2e", done.ToRef)
|
|
assert.Contains(t, done.Output, "pulling "+slusk+":1.4.2")
|
|
assert.Contains(t, done.Output, "recreated")
|
|
svc, err := e.store.GetService(context.Background(), 1)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, slusk+":1.4.2@sha256:e2e", svc.DesiredRef)
|
|
require.Eventually(t, func() bool { return len(e.agent.runner.Known()) == 0 }, waitFor, pollEvery, "the result must be acked")
|
|
}
|
|
|
|
func TestJobFinishedWhileDisconnectedIsRecordedOnce(t *testing.T) {
|
|
e := newJobEnv(t)
|
|
j := e.trigger(t)
|
|
e.waitStarted(t)
|
|
|
|
e.m.restart() // drops the agent's connection while the executor blocks
|
|
close(e.exec.release)
|
|
|
|
done := e.waitStatus(t, j.ID, model.JobSucceeded)
|
|
assert.NotEqual(t, lostMessage, done.Error)
|
|
assert.GreaterOrEqual(t, e.m.attempts.Load(), int64(2), "the agent must have reconnected")
|
|
require.Eventually(t, func() bool { return len(e.agent.runner.Known()) == 0 }, waitFor, pollEvery)
|
|
e.store.mu.Lock()
|
|
defer e.store.mu.Unlock()
|
|
assert.Equal(t, 1, e.store.finishApplied)
|
|
}
|
|
|
|
func TestMasterRestartKeepsJobTheAgentKnows(t *testing.T) {
|
|
e := newJobEnv(t)
|
|
j := e.trigger(t)
|
|
e.waitStarted(t)
|
|
|
|
e.m.restart()
|
|
e.waitRoutable(t) // the agent reconnected and said hello with its known jobs
|
|
assert.True(t, e.job(t, j.ID).Active(), "the agent still has the job, so it is not lost")
|
|
close(e.exec.release)
|
|
|
|
done := e.waitStatus(t, j.ID, model.JobSucceeded)
|
|
assert.Empty(t, done.Error)
|
|
}
|
|
|
|
func TestAgentRestartMarksJobLost(t *testing.T) {
|
|
e := newJobEnv(t)
|
|
j := e.trigger(t)
|
|
e.waitStarted(t)
|
|
|
|
// The agent process dies: connection first, so the cancelled job cannot
|
|
// report a result, then the runner.
|
|
e.agent.stopConn()
|
|
e.agent.stopRun()
|
|
fresh := startJobAgent(t, e.m, "pua_ok", newGateExecutor())
|
|
_ = fresh
|
|
|
|
lost := e.waitStatus(t, j.ID, model.JobFailed)
|
|
assert.Equal(t, lostMessage, lost.Error)
|
|
}
|
|
|
|
func TestPhaseOneAgentGetsNoJobs(t *testing.T) {
|
|
store := &memStore{agents: map[string]model.Agent{agenttoken.Hash("pua_old"): {ID: agentID, Name: "node-2"}}}
|
|
svcRow := model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
|
|
AgentID: ptr(agentID), ActualRef: slusk + ":1.4.2@sha256:old"}
|
|
store.seedService(svcRow)
|
|
m := startMaster(t, store)
|
|
js := jobs.New(store, m.hub, m.broker, "gitea.shcizo.se/shcizo/package-updater:latest", slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
|
|
// A raw client speaking the phase-1 protocol: its hello has no
|
|
// protocol_version.
|
|
ctx := context.Background()
|
|
c, _, err := websocket.Dial(ctx, m.url, &websocket.DialOptions{
|
|
HTTPHeader: http.Header{"Authorization": {"Bearer pua_old"}},
|
|
HTTPClient: &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}},
|
|
})
|
|
require.NoError(t, err)
|
|
defer c.CloseNow()
|
|
hello, err := protocol.Encode(protocol.TypeHello, map[string]any{"agent_version": "1", "hostname": "node-2", "swarm_manager": false})
|
|
require.NoError(t, err)
|
|
require.NoError(t, wsjson.Write(ctx, c, hello))
|
|
snap, err := protocol.Encode(protocol.TypeSnapshot, protocol.Snapshot{Services: []protocol.Service{web}})
|
|
require.NoError(t, err)
|
|
require.NoError(t, wsjson.Write(ctx, c, snap))
|
|
|
|
require.Eventually(t, func() bool {
|
|
_, err := m.hub.Route(svcRow)
|
|
return errors.Is(err, model.ErrAgentTooOld)
|
|
}, waitFor, pollEvery)
|
|
|
|
j, err := js.Trigger(ctx, 1, "sub-1")
|
|
require.NoError(t, err)
|
|
assert.Equal(t, model.JobFailed, j.Status)
|
|
assert.Equal(t, "agent too old for updates", j.Error)
|
|
|
|
rctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
|
|
defer cancel()
|
|
var env protocol.Envelope
|
|
err = wsjson.Read(rctx, c, &env)
|
|
assert.True(t, err != nil && strings.Contains(err.Error(), "deadline"), "no job.start expected, got %q (err %v)", env.Type, err)
|
|
}
|