133728e barerepo 1mo
1
package store
133728e barerepo 1mo
3
import (
133728e barerepo 1mo
4
"context"
133728e barerepo 1mo
5
"database/sql"
133728e barerepo 1mo
6
"errors"
133728e barerepo 1mo
7
"strings"
133728e barerepo 1mo
8
"time"
133728e barerepo 1mo
11
// Runner is one machine on one repository, because barerepo dispatches and records but never executes.
133728e barerepo 1mo
12
type Runner struct {
133728e barerepo 1mo
13
ID int64
133728e barerepo 1mo
14
Repo string
133728e barerepo 1mo
15
Hostname string
133728e barerepo 1mo
16
OS string
133728e barerepo 1mo
17
Arch string
133728e barerepo 1mo
18
Labels []string
133728e barerepo 1mo
19
Attached time.Time
133728e barerepo 1mo
20
LastSeen time.Time
133728e barerepo 1mo
23
// Offline reports a runner that has missed three polls. Chapter 15.
133728e barerepo 1mo
24
func (r Runner) Offline(interval time.Duration) bool {
133728e barerepo 1mo
25
return time.Since(r.LastSeen) > 3*interval
133728e barerepo 1mo
28
// AttachRunner keys on token and hostname, so restarting a machine leaves no second row.
133728e barerepo 1mo
29
func (db *DB) AttachRunner(ctx context.Context, tokenID int64, repo, hostname, os, arch string, labels []string) (*Runner, error) {
133728e barerepo 1mo
30
now := now()
133728e barerepo 1mo
31
joined := strings.Join(labels, ",")
133728e barerepo 1mo
33
var id int64
133728e barerepo 1mo
34
err := db.QueryRowContext(ctx,
133728e barerepo 1mo
35
`SELECT id FROM runners WHERE token_id = ? AND hostname = ?`, tokenID, hostname).Scan(&id)
133728e barerepo 1mo
36
switch {
133728e barerepo 1mo
37
case errors.Is(err, sql.ErrNoRows):
133728e barerepo 1mo
38
err = db.QueryRowContext(ctx,
133728e barerepo 1mo
39
`INSERT INTO runners (token_id, repo, hostname, os, arch, labels, attached_at, last_seen)
133728e barerepo 1mo
40
VALUES (?, ?, ?, ?, ?, ?, ?, ?) RETURNING id`,
133728e barerepo 1mo
41
tokenID, repo, hostname, os, arch, joined, now.Unix(), now.Unix()).Scan(&id)
133728e barerepo 1mo
42
if err != nil {
133728e barerepo 1mo
43
return nil, err
133728e barerepo 1mo
45
case err != nil:
133728e barerepo 1mo
46
return nil, err
133728e barerepo 1mo
47
default:
133728e barerepo 1mo
48
if _, err := db.ExecContext(ctx,
133728e barerepo 1mo
49
`UPDATE runners SET os = ?, arch = ?, labels = ?, last_seen = ? WHERE id = ?`,
133728e barerepo 1mo
50
os, arch, joined, now.Unix(), id); err != nil {
133728e barerepo 1mo
51
return nil, err
133728e barerepo 1mo
54
return &Runner{ID: id, Repo: repo, Hostname: hostname, OS: os, Arch: arch,
133728e barerepo 1mo
55
Labels: labels, Attached: now, LastSeen: now}, nil
133728e barerepo 1mo
58
// SeeRunner marks a runner as still there.
133728e barerepo 1mo
59
func (db *DB) SeeRunner(ctx context.Context, id int64) error {
133728e barerepo 1mo
60
_, err := db.ExecContext(ctx, `UPDATE runners SET last_seen = ? WHERE id = ?`, now().Unix(), id)
133728e barerepo 1mo
61
return err
133728e barerepo 1mo
64
func (db *DB) Runner(ctx context.Context, id int64) (*Runner, error) {
133728e barerepo 1mo
65
var r Runner
133728e barerepo 1mo
66
var labels string
133728e barerepo 1mo
67
var attached, seen int64
133728e barerepo 1mo
68
err := db.QueryRowContext(ctx,
133728e barerepo 1mo
69
`SELECT id, repo, hostname, os, arch, labels, attached_at, last_seen FROM runners WHERE id = ?`, id).
133728e barerepo 1mo
70
Scan(&r.ID, &r.Repo, &r.Hostname, &r.OS, &r.Arch, &labels, &attached, &seen)
133728e barerepo 1mo
71
if errors.Is(err, sql.ErrNoRows) {
133728e barerepo 1mo
72
return nil, ErrNotFound
133728e barerepo 1mo
74
if err != nil {
133728e barerepo 1mo
75
return nil, err
133728e barerepo 1mo
77
r.Labels = splitLabels(labels)
133728e barerepo 1mo
78
r.Attached = time.Unix(attached, 0)
133728e barerepo 1mo
79
r.LastSeen = time.Unix(seen, 0)
133728e barerepo 1mo
80
return &r, nil
133728e barerepo 1mo
83
// RunnersOf lists the machines attached to a repository.
133728e barerepo 1mo
84
func (db *DB) RunnersOf(ctx context.Context, repo string) ([]Runner, error) {
133728e barerepo 1mo
85
rows, err := db.QueryContext(ctx,
133728e barerepo 1mo
86
`SELECT id, repo, hostname, os, arch, labels, attached_at, last_seen
133728e barerepo 1mo
87
FROM runners WHERE repo = ? ORDER BY last_seen DESC`, repo)
133728e barerepo 1mo
88
if err != nil {
133728e barerepo 1mo
89
return nil, err
133728e barerepo 1mo
91
defer rows.Close()
133728e barerepo 1mo
92
var out []Runner
133728e barerepo 1mo
93
for rows.Next() {
133728e barerepo 1mo
94
var r Runner
133728e barerepo 1mo
95
var labels string
133728e barerepo 1mo
96
var attached, seen int64
133728e barerepo 1mo
97
if err := rows.Scan(&r.ID, &r.Repo, &r.Hostname, &r.OS, &r.Arch, &labels, &attached, &seen); err != nil {
133728e barerepo 1mo
98
return nil, err
133728e barerepo 1mo
100
r.Labels = splitLabels(labels)
133728e barerepo 1mo
101
r.Attached = time.Unix(attached, 0)
133728e barerepo 1mo
102
r.LastSeen = time.Unix(seen, 0)
133728e barerepo 1mo
103
out = append(out, r)
133728e barerepo 1mo
104
}
133728e barerepo 1mo
105
return out, rows.Err()
133728e barerepo 1mo
108
// ForgetRunner drops a machine. Chapter 24: offline machines can be forgotten.
133728e barerepo 1mo
109
func (db *DB) ForgetRunner(ctx context.Context, repo string, id int64) error {
133728e barerepo 1mo
110
res, err := db.ExecContext(ctx, `DELETE FROM runners WHERE repo = ? AND id = ?`, repo, id)
133728e barerepo 1mo
111
if err != nil {
133728e barerepo 1mo
112
return err
133728e barerepo 1mo
113
}
133728e barerepo 1mo
114
if n, _ := res.RowsAffected(); n == 0 {
133728e barerepo 1mo
115
return ErrNotFound
133728e barerepo 1mo
116
}
133728e barerepo 1mo
117
return nil
133728e barerepo 1mo
120
func splitLabels(s string) []string {
133728e barerepo 1mo
121
if strings.TrimSpace(s) == "" {
133728e barerepo 1mo
122
return nil
133728e barerepo 1mo
123
}
133728e barerepo 1mo
124
return strings.Split(s, ",")
133728e barerepo 1mo
127
// RunnersByToken maps one account's runner tokens to the machines that attached with them.
133728e barerepo 1mo
128
func (db *DB) RunnersByToken(ctx context.Context, account string) (map[int64][]Runner, error) {
133728e barerepo 1mo
129
rows, err := db.QueryContext(ctx,
133728e barerepo 1mo
130
`SELECT r.token_id, r.id, r.repo, r.hostname, r.os, r.arch, r.labels, r.attached_at, r.last_seen
133728e barerepo 1mo
131
FROM runners r JOIN tokens t ON t.id = r.token_id
133728e barerepo 1mo
132
WHERE t.account = ? ORDER BY r.last_seen DESC`, account)
133728e barerepo 1mo
133
if err != nil {
133728e barerepo 1mo
134
return nil, err
133728e barerepo 1mo
135
}
133728e barerepo 1mo
136
defer rows.Close()
133728e barerepo 1mo
137
out := map[int64][]Runner{}
133728e barerepo 1mo
138
for rows.Next() {
133728e barerepo 1mo
139
var r Runner
133728e barerepo 1mo
140
var tokenID int64
133728e barerepo 1mo
141
var labels string
133728e barerepo 1mo
142
var attached, seen int64
133728e barerepo 1mo
143
if err := rows.Scan(&tokenID, &r.ID, &r.Repo, &r.Hostname, &r.OS, &r.Arch,
133728e barerepo 1mo
144
&labels, &attached, &seen); err != nil {
133728e barerepo 1mo
145
return nil, err
133728e barerepo 1mo
146
}
133728e barerepo 1mo
147
r.Labels = splitLabels(labels)
133728e barerepo 1mo
148
r.Attached = time.Unix(attached, 0)
133728e barerepo 1mo
149
r.LastSeen = time.Unix(seen, 0)
133728e barerepo 1mo
150
out[tokenID] = append(out[tokenID], r)
133728e barerepo 1mo
151
}
133728e barerepo 1mo
152
return out, rows.Err()
133728e barerepo 1mo
155
// RunnerWork is what a runner has done and whether it is doing something now. Chapter 24.
133728e barerepo 1mo
156
type RunnerWork struct {
133728e barerepo 1mo
157
Runs int
133728e barerepo 1mo
158
Busy bool
133728e barerepo 1mo
161
// WorkOf counts the jobs each runner of a repository has taken, and finds the ones running now.
133728e barerepo 1mo
162
func (db *DB) WorkOf(ctx context.Context, repo string) (map[int64]RunnerWork, error) {
133728e barerepo 1mo
163
rows, err := db.QueryContext(ctx,
133728e barerepo 1mo
164
`SELECT runner_id, COUNT(*), SUM(CASE WHEN state = ? THEN 1 ELSE 0 END) FROM jobs
133728e barerepo 1mo
165
WHERE repo = ? AND runner_id IS NOT NULL GROUP BY runner_id`, JobRunning, repo)
133728e barerepo 1mo
166
if err != nil {
133728e barerepo 1mo
167
return nil, err
133728e barerepo 1mo
168
}
133728e barerepo 1mo
169
defer rows.Close()
133728e barerepo 1mo
170
out := map[int64]RunnerWork{}
133728e barerepo 1mo
171
for rows.Next() {
133728e barerepo 1mo
172
var id int64
133728e barerepo 1mo
173
var runs, running int
133728e barerepo 1mo
174
if err := rows.Scan(&id, &runs, &running); err != nil {
133728e barerepo 1mo
175
return nil, err
133728e barerepo 1mo
176
}
133728e barerepo 1mo
177
out[id] = RunnerWork{Runs: runs, Busy: running > 0}
133728e barerepo 1mo
178
}
133728e barerepo 1mo
179
return out, rows.Err()