diff --git a/internal/master/store/agents.go b/internal/master/store/agents.go index c477665..1d363f3 100644 --- a/internal/master/store/agents.go +++ b/internal/master/store/agents.go @@ -64,7 +64,7 @@ func (s *Store) AgentByTokenHash(ctx context.Context, hash string) (model.Agent, } // RevokeAgent marks the agent revoked and its Compose and plain-container -// services removed, in one transaction. Swarm rows are shared with other +// services removed and its queued and running jobs failed, in one transaction. Swarm rows are shared with other // managers and left alone. Its token stops working immediately; closing a // live connection is the hub's job. func (s *Store) RevokeAgent(ctx context.Context, id int64, at time.Time) error { @@ -83,6 +83,9 @@ func (s *Store) RevokeAgent(ctx context.Context, id int64, at time.Time) error { WHERE agent_id = $1 AND kind IN ('compose', 'container') AND removed_at IS NULL`, id, at); err != nil { return fmt.Errorf("revoke agent: remove services: %w", err) } + if _, err := tx.Exec(ctx, failActiveJobsSQL, id, "agent revoked", at); err != nil { + return fmt.Errorf("revoke agent: fail jobs: %w", err) + } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("revoke agent: commit: %w", err) } diff --git a/internal/master/store/agents_integration_test.go b/internal/master/store/agents_integration_test.go index 568ffc1..7227d93 100644 --- a/internal/master/store/agents_integration_test.go +++ b/internal/master/store/agents_integration_test.go @@ -138,3 +138,24 @@ func TestRevokedAgent_UpsertDoesNotResurrectRemovedRow(t *testing.T) { require.NoError(t, err) assert.Empty(t, got, "the revoked agent's removed row must stay removed") } + +func TestRevokeAgent_FailsActiveJobs(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + a, svcA := jobFixture(t, s, "a") + b, svcB := jobFixture(t, s, "b") + ja, err := s.CreateJob(ctx, newJob(svcA, a)) + require.NoError(t, err) + jb, err := s.CreateJob(ctx, newJob(svcB, b)) + require.NoError(t, err) + + require.NoError(t, s.RevokeAgent(ctx, a, t0)) + + got, err := s.GetJob(ctx, ja.ID) + require.NoError(t, err) + assert.Equal(t, model.JobFailed, got.Status) + assert.Equal(t, "agent revoked", got.Error) + got, err = s.GetJob(ctx, jb.ID) + require.NoError(t, err) + assert.Equal(t, model.JobQueued, got.Status) +} diff --git a/internal/master/store/jobs.go b/internal/master/store/jobs.go new file mode 100644 index 0000000..4f708c6 --- /dev/null +++ b/internal/master/store/jobs.go @@ -0,0 +1,204 @@ +package store + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +const lostJobMessage = "lost: the agent did not know this job when it reconnected" + +// failActiveJobsSQL fails every queued or running job of agent $1 with +// message $2 at time $3. +const failActiveJobsSQL = ` +UPDATE update_jobs SET status = 'failed', error = $2, finished_at = $3 +WHERE agent_id = $1 AND status IN ('queued', 'running')` + +const jobSelectSQL = ` +SELECT j.id, j.service_id, s.kind, s.compose_project, s.compose_service, s.swarm_service, s.container_name, + j.agent_id, COALESCE(a.name, ''), j.from_ref, j.to_ref, j.trigger, j.requested_by, j.status, + %s, j.output_truncated, j.error, j.created_at, j.started_at, j.finished_at +FROM update_jobs j +JOIN services s ON s.id = j.service_id +LEFT JOIN agents a ON a.id = j.agent_id` + +func scanJob(row pgx.Row) (model.Job, error) { + var j model.Job + var svc model.Service + var kind, trigger, status string + err := row.Scan(&j.ID, &j.ServiceID, &kind, &svc.ComposeProject, &svc.ComposeService, &svc.SwarmService, &svc.ContainerName, + &j.AgentID, &j.AgentName, &j.FromRef, &j.ToRef, &trigger, &j.RequestedBy, &status, + &j.Output, &j.OutputTruncated, &j.Error, &j.CreatedAt, &j.StartedAt, &j.FinishedAt) + svc.Kind = protocol.Kind(kind) + j.ServiceName = svc.DisplayName() + j.Trigger = model.JobTrigger(trigger) + j.Status = model.JobStatus(status) + return j, err +} + +func (s *Store) getJob(ctx context.Context, id int64) (model.Job, error) { + j, err := scanJob(s.pool.QueryRow(ctx, fmt.Sprintf(jobSelectSQL, "j.output")+` WHERE j.id = $1`, id)) + if errors.Is(err, pgx.ErrNoRows) { + return model.Job{}, model.ErrNotFound + } + if err != nil { + return model.Job{}, fmt.Errorf("get job: %w", err) + } + return j, nil +} + +// CreateJob inserts a queued job. If the service already has a queued or +// running job, it returns that job and model.ErrJobActive (detected via the +// unique index, so it holds under concurrent calls). +func (s *Store) CreateJob(ctx context.Context, j model.NewJob) (model.Job, error) { + var id int64 + err := s.pool.QueryRow(ctx, ` + INSERT INTO update_jobs (service_id, agent_id, from_ref, to_ref, trigger, requested_by, status, created_at) + VALUES ($1, $2, $3, $4, $5, $6, 'queued', $7) + RETURNING id`, + j.ServiceID, j.AgentID, j.FromRef, j.ToRef, string(j.Trigger), j.RequestedBy, j.CreatedAt).Scan(&id) + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == "23505" && pgErr.ConstraintName == "update_jobs_one_active" { + active, err := scanJob(s.pool.QueryRow(ctx, fmt.Sprintf(jobSelectSQL, "j.output")+ + ` WHERE j.service_id = $1 AND j.status IN ('queued', 'running')`, j.ServiceID)) + if err != nil { + return model.Job{}, fmt.Errorf("find active job: %w", err) + } + return active, model.ErrJobActive + } + if err != nil { + return model.Job{}, fmt.Errorf("create job: %w", err) + } + return s.getJob(ctx, id) +} + +// FailJob sets a queued or running job to failed with msg and finished_at = +// at. Returns false (no error) if the job is no longer active. +func (s *Store) FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error) { + tag, err := s.pool.Exec(ctx, ` + UPDATE update_jobs SET status = 'failed', error = $2, finished_at = $3 + WHERE id = $1 AND status IN ('queued', 'running')`, id, msg, at) + if err != nil { + return false, fmt.Errorf("fail job: %w", err) + } + return tag.RowsAffected() == 1, nil +} + +// GetJob returns one job with its output. Returns model.ErrNotFound if missing. +func (s *Store) GetJob(ctx context.Context, id int64) (model.Job, error) { + return s.getJob(ctx, id) +} + +// ListJobs returns jobs newest first, without Output. +func (s *Store) ListJobs(ctx context.Context, f model.JobFilter) ([]model.Job, error) { + limit := f.Limit + if limit <= 0 { + limit = 100 + } + rows, err := s.pool.Query(ctx, fmt.Sprintf(jobSelectSQL, "''")+` + WHERE ($1 = 0 OR j.service_id = $1) + ORDER BY j.id DESC + LIMIT $2`, f.ServiceID, limit) + if err != nil { + return nil, fmt.Errorf("list jobs: %w", err) + } + defer rows.Close() + var out []model.Job + for rows.Next() { + j, err := scanJob(rows) + if err != nil { + return nil, fmt.Errorf("scan job: %w", err) + } + out = append(out, j) + } + return out, rows.Err() +} + +// FailLostJobs fails every queued or running job sent to agentID whose ID is +// not in known, with a "lost" error. Returns the IDs it failed. +func (s *Store) FailLostJobs(ctx context.Context, agentID int64, known []int64, at time.Time) ([]int64, error) { + if known == nil { + known = []int64{} + } + rows, err := s.pool.Query(ctx, ` + UPDATE update_jobs SET status = 'failed', error = $3, finished_at = $4 + WHERE agent_id = $1 AND status IN ('queued', 'running') AND NOT (id = ANY($2)) + RETURNING id`, agentID, known, lostJobMessage, at) + if err != nil { + return nil, fmt.Errorf("fail lost jobs: %w", err) + } + defer rows.Close() + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("scan lost job: %w", err) + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +// JobProgress appends chunk to the output of a queued or running job sent to +// agentID, keeping the last protocol.MaxJobOutput characters, and moves a +// queued job to running (started_at = at if unset). Returns false if no such +// active job belongs to agentID. +func (s *Store) JobProgress(ctx context.Context, agentID, jobID int64, chunk string, at time.Time) (bool, error) { + tag, err := s.pool.Exec(ctx, ` + UPDATE update_jobs SET + status = 'running', + started_at = COALESCE(started_at, $4), + output_truncated = output_truncated OR length(output) + length($3::text) > $5, + output = right(output || $3::text, $5) + WHERE id = $1 AND agent_id = $2 AND status IN ('queued', 'running')`, + jobID, agentID, chunk, at, protocol.MaxJobOutput) + if err != nil { + return false, fmt.Errorf("job progress: %w", err) + } + return tag.RowsAffected() == 1, nil +} + +// FinishJob records the agent's result for a queued or running job sent to +// agentID: status, error, output (replacing what progress assembled), +// started_at (kept if already set), finished_at; to_ref = f.Ref when set. On +// success it also sets the service's desired_ref = f.Ref, in the same +// transaction. Returns false if no such active job belongs to agentID. +func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.JobFinish) (bool, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return false, fmt.Errorf("finish job: begin: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + var serviceID int64 + err = tx.QueryRow(ctx, ` + UPDATE update_jobs SET + status = $3, error = $4, output = $5, output_truncated = $6, + to_ref = CASE WHEN $7::text <> '' THEN $7::text ELSE to_ref END, + started_at = COALESCE(started_at, $8), finished_at = $9 + WHERE id = $1 AND agent_id = $2 AND status IN ('queued', 'running') + RETURNING service_id`, + jobID, agentID, string(f.Status), f.Error, f.Output, f.OutputTruncated, f.Ref, f.StartedAt, f.FinishedAt).Scan(&serviceID) + if errors.Is(err, pgx.ErrNoRows) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("finish job: %w", err) + } + if f.Status == model.JobSucceeded { + if _, err := tx.Exec(ctx, `UPDATE services SET desired_ref = $2 WHERE id = $1`, serviceID, f.Ref); err != nil { + return false, fmt.Errorf("finish job: desired ref: %w", err) + } + } + if err := tx.Commit(ctx); err != nil { + return false, fmt.Errorf("finish job: commit: %w", err) + } + return true, nil +} diff --git a/internal/master/store/jobs_integration_test.go b/internal/master/store/jobs_integration_test.go new file mode 100644 index 0000000..2e7a54d --- /dev/null +++ b/internal/master/store/jobs_integration_test.go @@ -0,0 +1,305 @@ +//go:build integration + +package store + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +// jobFixture seeds one agent with one Compose service and returns their IDs. +func jobFixture(t *testing.T, s *Store, agentName string) (agentID, serviceID int64) { + t.Helper() + ctx := context.Background() + agentID = newAgent(t, s, agentName) + require.NoError(t, s.ApplySnapshot(ctx, agentID, worker, []protocol.Service{slusk}, t0)) + svcs, err := s.ListServices(ctx) + require.NoError(t, err) + for _, svc := range svcs { + if svc.AgentID != nil && *svc.AgentID == agentID { + return agentID, svc.ID + } + } + t.Fatal("service not found") + return +} + +func newJob(serviceID, agentID int64) model.NewJob { + return model.NewJob{ + ServiceID: serviceID, AgentID: &agentID, + FromRef: "r:1@sha256:a", ToRef: "r:1", Trigger: model.TriggerManual, + RequestedBy: "sub", CreatedAt: t0, + } +} + +func TestCreateJob_OneActivePerService(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agentID, svcID := jobFixture(t, s, "a") + + var wg sync.WaitGroup + jobs := make([]model.Job, 2) + errs := make([]error, 2) + for i := range jobs { + wg.Add(1) + go func() { + defer wg.Done() + jobs[i], errs[i] = s.CreateJob(ctx, newJob(svcID, agentID)) + }() + } + wg.Wait() + + assert.Equal(t, jobs[0].ID, jobs[1].ID) + var failures int + for _, err := range errs { + if err != nil { + assert.ErrorIs(t, err, model.ErrJobActive) + failures++ + } + } + assert.Equal(t, 1, failures) + + ok, err := s.FailJob(ctx, jobs[0].ID, "x", t0) + require.NoError(t, err) + assert.True(t, ok) + ok, err = s.FailJob(ctx, jobs[0].ID, "x", t0) + require.NoError(t, err) + assert.False(t, ok) + + next, err := s.CreateJob(ctx, newJob(svcID, agentID)) + require.NoError(t, err) + assert.NotEqual(t, jobs[0].ID, next.ID) +} + +func TestFailLostJobs(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + a, svcA := jobFixture(t, s, "a") + b := newAgent(t, s, "b") + require.NoError(t, s.ApplySnapshot(ctx, b, worker, []protocol.Service{adhoc}, t0)) + all, err := s.ListServices(ctx) + require.NoError(t, err) + var svcB int64 + for _, svc := range all { + if svc.ID != svcA { + svcB = svc.ID + } + } + + // Only one active job per service, so A's jobs go through the lifecycle. + j3, err := s.CreateJob(ctx, newJob(svcA, a)) + require.NoError(t, err) + ok, err := s.FinishJob(ctx, a, j3.ID, model.JobFinish{Status: model.JobSucceeded, Ref: "r:1@sha256:d", StartedAt: t0, FinishedAt: t0}) + require.NoError(t, err) + require.True(t, ok) + j1, err := s.CreateJob(ctx, newJob(svcA, a)) + require.NoError(t, err) + j4, err := s.CreateJob(ctx, newJob(svcB, b)) + require.NoError(t, err) + _, err = s.JobProgress(ctx, b, j4.ID, "x", t0) + require.NoError(t, err) + + ids, err := s.FailLostJobs(ctx, a, []int64{j3.ID}, t0) + require.NoError(t, err) + assert.Equal(t, []int64{j1.ID}, ids) + + got, err := s.GetJob(ctx, j1.ID) + require.NoError(t, err) + assert.Equal(t, model.JobFailed, got.Status) + assert.Contains(t, got.Error, "lost") + got, err = s.GetJob(ctx, j3.ID) + require.NoError(t, err) + assert.Equal(t, model.JobSucceeded, got.Status) + got, err = s.GetJob(ctx, j4.ID) + require.NoError(t, err) + assert.Equal(t, model.JobRunning, got.Status) + + // A running job the agent still knows is kept. + j5, err := s.CreateJob(ctx, newJob(svcA, a)) + require.NoError(t, err) + _, err = s.JobProgress(ctx, a, j5.ID, "", t0) + require.NoError(t, err) + ids, err = s.FailLostJobs(ctx, a, []int64{j5.ID}, t0) + require.NoError(t, err) + assert.Empty(t, ids) +} + +func TestJobProgress_OwnerOnly(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + a, svc := jobFixture(t, s, "a") + b := newAgent(t, s, "b") + j, err := s.CreateJob(ctx, newJob(svc, a)) + require.NoError(t, err) + + started := t0.Add(time.Second) + ok, err := s.JobProgress(ctx, a, j.ID, "hello ", started) + require.NoError(t, err) + assert.True(t, ok) + got, err := s.GetJob(ctx, j.ID) + require.NoError(t, err) + assert.Equal(t, model.JobRunning, got.Status) + require.NotNil(t, got.StartedAt) + assert.True(t, started.Equal(*got.StartedAt)) + + ok, err = s.JobProgress(ctx, b, j.ID, "evil", t0.Add(time.Minute)) + require.NoError(t, err) + assert.False(t, ok) + got, err = s.GetJob(ctx, j.ID) + require.NoError(t, err) + assert.Equal(t, "hello ", got.Output) + + ok, err = s.JobProgress(ctx, a, j.ID, "world", t0.Add(time.Minute)) + require.NoError(t, err) + assert.True(t, ok) + got, err = s.GetJob(ctx, j.ID) + require.NoError(t, err) + assert.Equal(t, "hello world", got.Output) + assert.True(t, started.Equal(*got.StartedAt)) + + _, err = s.FailJob(ctx, j.ID, "x", t0) + require.NoError(t, err) + ok, err = s.JobProgress(ctx, a, j.ID, "late", t0) + require.NoError(t, err) + assert.False(t, ok) +} + +func TestJobProgress_KeepsTail(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + a, svc := jobFixture(t, s, "a") + j, err := s.CreateJob(ctx, newJob(svc, a)) + require.NoError(t, err) + + _, err = s.JobProgress(ctx, a, j.ID, strings.Repeat("a", protocol.MaxJobOutput), t0) + require.NoError(t, err) + got, err := s.GetJob(ctx, j.ID) + require.NoError(t, err) + assert.False(t, got.OutputTruncated) + + _, err = s.JobProgress(ctx, a, j.ID, "TAIL", t0) + require.NoError(t, err) + got, err = s.GetJob(ctx, j.ID) + require.NoError(t, err) + assert.True(t, got.OutputTruncated) + assert.Len(t, got.Output, protocol.MaxJobOutput) + assert.True(t, strings.HasSuffix(got.Output, "aaTAIL")) +} + +func TestFinishJob_OnlyOnceAndOnlyByOwner(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + a, svc := jobFixture(t, s, "a") + b := newAgent(t, s, "b") + j, err := s.CreateJob(ctx, newJob(svc, a)) + require.NoError(t, err) + _, err = s.JobProgress(ctx, a, j.ID, "partial", t0) + require.NoError(t, err) + + fin := model.JobFinish{Status: model.JobSucceeded, Ref: "r:1@sha256:d", Output: "full", StartedAt: t0.Add(time.Hour), FinishedAt: t0.Add(2 * time.Hour)} + ok, err := s.FinishJob(ctx, b, j.ID, fin) + require.NoError(t, err) + assert.False(t, ok) + + ok, err = s.FinishJob(ctx, a, j.ID, fin) + require.NoError(t, err) + assert.True(t, ok) + got, err := s.GetJob(ctx, j.ID) + require.NoError(t, err) + assert.Equal(t, model.JobSucceeded, got.Status) + assert.Equal(t, "r:1@sha256:d", got.ToRef) + assert.Equal(t, "full", got.Output) + assert.True(t, t0.Equal(*got.StartedAt), "started_at set by the first progress is kept") + assert.True(t, fin.FinishedAt.Equal(*got.FinishedAt)) + service, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, "r:1@sha256:d", service.DesiredRef) + + again := fin + again.Ref = "r:2@sha256:e" + again.Output = "other" + ok, err = s.FinishJob(ctx, a, j.ID, again) + require.NoError(t, err) + assert.False(t, ok) + got, err = s.GetJob(ctx, j.ID) + require.NoError(t, err) + assert.Equal(t, "full", got.Output) + service, err = s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, "r:1@sha256:d", service.DesiredRef) + + j2, err := s.CreateJob(ctx, newJob(svc, a)) + require.NoError(t, err) + ok, err = s.FinishJob(ctx, a, j2.ID, model.JobFinish{Status: model.JobFailed, Error: "boom", Output: "o", StartedAt: t0, FinishedAt: t0}) + require.NoError(t, err) + assert.True(t, ok) + got, err = s.GetJob(ctx, j2.ID) + require.NoError(t, err) + assert.Equal(t, model.JobFailed, got.Status) + assert.Equal(t, "boom", got.Error) + assert.Equal(t, "r:1", got.ToRef) + service, err = s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, "r:1@sha256:d", service.DesiredRef) +} + +func TestGetJobAndListJobs(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + a, svcA := jobFixture(t, s, "a") + b := newAgent(t, s, "b") + require.NoError(t, s.ApplySnapshot(ctx, b, worker, []protocol.Service{adhoc}, t0)) + all, err := s.ListServices(ctx) + require.NoError(t, err) + var svcB int64 + for _, svc := range all { + if svc.ID != svcA { + svcB = svc.ID + } + } + + j1, err := s.CreateJob(ctx, newJob(svcA, a)) + require.NoError(t, err) + _, err = s.JobProgress(ctx, a, j1.ID, "out", t0) + require.NoError(t, err) + _, err = s.FailJob(ctx, j1.ID, "x", t0) + require.NoError(t, err) + j2, err := s.CreateJob(ctx, newJob(svcA, a)) + require.NoError(t, err) + j3, err := s.CreateJob(ctx, newJob(svcB, b)) + require.NoError(t, err) + + got, err := s.GetJob(ctx, j1.ID) + require.NoError(t, err) + assert.Equal(t, "slusk/web", got.ServiceName) + assert.Equal(t, "a", got.AgentName) + assert.Equal(t, "out", got.Output) + + _, err = s.GetJob(ctx, 99999) + assert.ErrorIs(t, err, model.ErrNotFound) + + list, err := s.ListJobs(ctx, model.JobFilter{}) + require.NoError(t, err) + require.Len(t, list, 3) + assert.Equal(t, []int64{j3.ID, j2.ID, j1.ID}, []int64{list[0].ID, list[1].ID, list[2].ID}) + for _, j := range list { + assert.Empty(t, j.Output) + } + + list, err = s.ListJobs(ctx, model.JobFilter{ServiceID: svcA}) + require.NoError(t, err) + assert.Len(t, list, 2) + + list, err = s.ListJobs(ctx, model.JobFilter{Limit: 1}) + require.NoError(t, err) + assert.Len(t, list, 1) +} diff --git a/internal/master/store/services.go b/internal/master/store/services.go index 6f858eb..a36ceee 100644 --- a/internal/master/store/services.go +++ b/internal/master/store/services.go @@ -39,13 +39,14 @@ func lockActiveAgent(ctx context.Context, tx pgx.Tx, agentID int64) error { const upsertServiceSQL = ` INSERT INTO services ( - kind, location_key, agent_id, compose_project, compose_service, working_dir, + kind, location_key, agent_id, compose_project, compose_service, working_dir, config_files, swarm_cluster_id, swarm_service, container_name, image_repo, actual_ref, actual_digest, status, self, first_seen_at, updated_at) -VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $15) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $16) ON CONFLICT (location_key) DO UPDATE SET agent_id = EXCLUDED.agent_id, working_dir = EXCLUDED.working_dir, + config_files = EXCLUDED.config_files, image_repo = EXCLUDED.image_repo, actual_ref = EXCLUDED.actual_ref, actual_digest = EXCLUDED.actual_digest, @@ -61,7 +62,7 @@ func upsertService(ctx context.Context, db execer, agentID int64, clusterID stri } _, err := db.Exec(ctx, upsertServiceSQL, string(svc.Kind), model.LocationKey(agentID, clusterID, svc.Key()), agentID, - svc.ComposeProject, svc.ComposeService, svc.WorkingDir, + svc.ComposeProject, svc.ComposeService, svc.WorkingDir, svc.ConfigFiles, clusterCol, svc.SwarmService, svc.ContainerName, imageref.Parse(svc.Image).Repo, svc.Image, svc.Digest, svc.Status, svc.Self, at) @@ -144,14 +145,27 @@ func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, age return nil } +const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''), + s.compose_project, s.compose_service, s.working_dir, s.config_files, s.swarm_cluster_id, + s.swarm_service, s.container_name, s.image_repo, s.desired_ref, s.actual_ref, s.actual_digest, + s.status, s.self, s.update_mode, s.first_seen_at, s.updated_at, s.removed_at` + +func scanService(row pgx.Row) (model.Service, error) { + var m model.Service + var kind string + err := row.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName, + &m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.ConfigFiles, &m.SwarmClusterID, + &m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.DesiredRef, &m.ActualRef, &m.ActualDigest, + &m.Status, &m.Self, &m.UpdateMode, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) + m.Kind = protocol.Kind(kind) + return m, err +} + // ListServices returns all services not marked removed: machine-bound // services grouped by agent name first, Swarm services last. func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) { rows, err := s.pool.Query(ctx, ` - SELECT s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''), - s.compose_project, s.compose_service, s.working_dir, s.swarm_cluster_id, - s.swarm_service, s.container_name, s.image_repo, s.actual_ref, s.actual_digest, - s.status, s.self, s.update_mode, s.first_seen_at, s.updated_at + SELECT `+serviceColumns+` FROM services s LEFT JOIN agents a ON a.id = s.agent_id WHERE s.removed_at IS NULL ORDER BY s.kind = 'swarm', COALESCE(a.name, ''), s.location_key`) @@ -161,16 +175,27 @@ func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) { defer rows.Close() var out []model.Service for rows.Next() { - var m model.Service - var kind string - if err := rows.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName, - &m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.SwarmClusterID, - &m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.ActualRef, &m.ActualDigest, - &m.Status, &m.Self, &m.UpdateMode, &m.FirstSeenAt, &m.UpdatedAt); err != nil { + m, err := scanService(rows) + if err != nil { return nil, fmt.Errorf("scan service: %w", err) } - m.Kind = protocol.Kind(kind) out = append(out, m) } return out, rows.Err() } + +// GetService returns one service, including a removed one (RemovedAt set). +// Returns model.ErrNotFound if the ID does not exist. +func (s *Store) GetService(ctx context.Context, id int64) (model.Service, error) { + m, err := scanService(s.pool.QueryRow(ctx, ` + SELECT `+serviceColumns+` + FROM services s LEFT JOIN agents a ON a.id = s.agent_id + WHERE s.id = $1`, id)) + if errors.Is(err, pgx.ErrNoRows) { + return model.Service{}, model.ErrNotFound + } + if err != nil { + return model.Service{}, fmt.Errorf("get service: %w", err) + } + return m, nil +} diff --git a/internal/master/store/services_integration_test.go b/internal/master/store/services_integration_test.go index 87d8013..728c8bc 100644 --- a/internal/master/store/services_integration_test.go +++ b/internal/master/store/services_integration_test.go @@ -160,3 +160,36 @@ func TestListServices_SwarmLast(t *testing.T) { } func itoa(n int64) string { return strconv.FormatInt(n, 10) } + +func TestGetService_IncludesRemoved(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "node-2") + require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk}, t0)) + svcs, err := s.ListServices(ctx) + require.NoError(t, err) + require.Len(t, svcs, 1) + + require.NoError(t, s.RemoveService(ctx, id, "", slusk.Key(), t0)) + got, err := s.GetService(ctx, svcs[0].ID) + require.NoError(t, err) + assert.NotNil(t, got.RemovedAt) + assert.Equal(t, "node-2", got.AgentName) + + _, err = s.GetService(ctx, 99999) + assert.ErrorIs(t, err, model.ErrNotFound) +} + +func TestApplySnapshot_StoresConfigFiles(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "node-2") + withFiles := slusk + withFiles.ConfigFiles = "/srv/slusk/compose.yaml,/srv/slusk/compose.override.yaml" + require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{withFiles}, t0)) + + got, err := s.ListServices(ctx) + require.NoError(t, err) + require.Len(t, got, 1) + assert.Equal(t, withFiles.ConfigFiles, got[0].ConfigFiles) +}