lewis.moe
forked from
tangled.org/core
Monorepo for Tangled
spindle/{db,engine}: rework db to use rkey
Signed-off-by: Anirudh Oppiliappan <anirudh@tangled.sh>
anirudh.fi
+1
spindle/config/config.go
+1
spindle/config/config.go
+4
-3
spindle/db/db.go
+4
-3
spindle/db/db.go
···
30
did text primary key
31
);
32
33
-
create table if not exists pipelines (
34
-
at_uri text not null,
35
status text not null,
36
37
-- only set if status is 'failed'
···
42
updated_at timestamp not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
43
finished_at timestamp,
44
45
-
primary key (at_uri)
46
);
47
`)
48
if err != nil {
···
30
did text primary key
31
);
32
33
+
create table if not exists pipeline_status (
34
+
rkey text not null,
35
+
pipeline text not null,
36
status text not null,
37
38
-- only set if status is 'failed'
···
43
updated_at timestamp not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
44
finished_at timestamp,
45
46
+
primary key (rkey)
47
);
48
`)
49
if err != nil {
+48
-42
spindle/db/pipelines.go
+48
-42
spindle/db/pipelines.go
···
8
"tangled.sh/tangled.sh/core/knotserver/notifier"
9
)
10
11
-
type PipelineStatus string
12
13
var (
14
-
PipelinePending PipelineStatus = "pending"
15
-
PipelineRunning PipelineStatus = "running"
16
-
PipelineFailed PipelineStatus = "failed"
17
-
PipelineTimeout PipelineStatus = "timeout"
18
-
PipelineCancelled PipelineStatus = "cancelled"
19
-
PipelineSuccess PipelineStatus = "success"
20
)
21
22
-
type Pipeline struct {
23
-
Rkey string `json:"rkey"`
24
-
Knot string `json:"knot"`
25
-
Status PipelineStatus `json:"status"`
26
27
// only if Failed
28
Error string `json:"error"`
···
33
FinishedAt time.Time `json:"finished_at"`
34
}
35
36
-
func (p Pipeline) AsRecord() *tangled.PipelineStatus {
37
exitCode64 := int64(p.ExitCode)
38
finishedAt := p.FinishedAt.String()
39
40
return &tangled.PipelineStatus{
41
-
Pipeline: fmt.Sprintf("at://%s/%s", p.Knot, p.Rkey),
42
-
Status: string(p.Status),
43
44
ExitCode: &exitCode64,
45
Error: &p.Error,
···
54
return fmt.Sprintf("at://%s/did:web:%s/%s", tangled.PipelineStatusNSID, knot, rkey)
55
}
56
57
-
func (db *DB) CreatePipeline(rkey, knot string, n *notifier.Notifier) error {
58
_, err := db.Exec(`
59
-
insert into pipelines (at_uri, status)
60
-
values (?, ?)
61
-
`, pipelineAtUri(rkey, knot), PipelinePending)
62
63
if err != nil {
64
return err
···
67
return nil
68
}
69
70
-
func (db *DB) MarkPipelineRunning(rkey, knot string, n *notifier.Notifier) error {
71
_, err := db.Exec(`
72
-
update pipelines
73
set status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
74
-
where at_uri = ?
75
-
`, PipelineRunning, pipelineAtUri(rkey, knot))
76
77
if err != nil {
78
return err
···
81
return nil
82
}
83
84
-
func (db *DB) MarkPipelineFailed(rkey, knot string, exitCode int, errorMsg string, n *notifier.Notifier) error {
85
_, err := db.Exec(`
86
-
update pipelines
87
set status = ?,
88
exit_code = ?,
89
error = ?,
90
updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now'),
91
finished_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
92
-
where at_uri = ?
93
-
`, PipelineFailed, exitCode, errorMsg, pipelineAtUri(rkey, knot))
94
if err != nil {
95
return err
96
}
···
98
return nil
99
}
100
101
-
func (db *DB) MarkPipelineTimeout(rkey, knot string, n *notifier.Notifier) error {
102
_, err := db.Exec(`
103
-
update pipelines
104
set status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
105
-
where at_uri = ?
106
-
`, PipelineTimeout, pipelineAtUri(rkey, knot))
107
if err != nil {
108
return err
109
}
···
111
return nil
112
}
113
114
-
func (db *DB) MarkPipelineSuccess(rkey, knot string, n *notifier.Notifier) error {
115
_, err := db.Exec(`
116
-
update pipelines
117
set status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now'),
118
finished_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
119
-
where at_uri = ?
120
-
`, PipelineSuccess, pipelineAtUri(rkey, knot))
121
122
if err != nil {
123
return err
···
126
return nil
127
}
128
129
-
func (db *DB) GetPipeline(rkey, knot string) (Pipeline, error) {
130
-
var p Pipeline
131
err := db.QueryRow(`
132
select rkey, status, error, exit_code, started_at, updated_at, finished_at
133
from pipelines
134
-
where at_uri = ?
135
-
`, pipelineAtUri(rkey, knot)).Scan(&p.Rkey, &p.Status, &p.Error, &p.ExitCode, &p.StartedAt, &p.UpdatedAt, &p.FinishedAt)
136
return p, err
137
}
138
139
-
func (db *DB) GetPipelines(cursor string) ([]Pipeline, error) {
140
whereClause := ""
141
args := []any{}
142
if cursor != "" {
···
146
147
query := fmt.Sprintf(`
148
select rkey, status, error, exit_code, started_at, updated_at, finished_at
149
-
from pipelines
150
%s
151
order by rkey asc
152
limit 100
···
158
}
159
defer rows.Close()
160
161
-
var pipelines []Pipeline
162
for rows.Next() {
163
-
var p Pipeline
164
rows.Scan(&p.Rkey, &p.Status, &p.Error, &p.ExitCode, &p.StartedAt, &p.UpdatedAt, &p.FinishedAt)
165
pipelines = append(pipelines, p)
166
}
167
168
if err := rows.Err(); err != nil {
169
return nil, err
170
}
171
172
return pipelines, nil
···
8
"tangled.sh/tangled.sh/core/knotserver/notifier"
9
)
10
11
+
type PipelineRunStatus string
12
13
var (
14
+
PipelinePending PipelineRunStatus = "pending"
15
+
PipelineRunning PipelineRunStatus = "running"
16
+
PipelineFailed PipelineRunStatus = "failed"
17
+
PipelineTimeout PipelineRunStatus = "timeout"
18
+
PipelineCancelled PipelineRunStatus = "cancelled"
19
+
PipelineSuccess PipelineRunStatus = "success"
20
)
21
22
+
type PipelineStatus struct {
23
+
Rkey string `json:"rkey"`
24
+
Pipeline string `json:"pipeline"`
25
+
Status PipelineRunStatus `json:"status"`
26
27
// only if Failed
28
Error string `json:"error"`
···
33
FinishedAt time.Time `json:"finished_at"`
34
}
35
36
+
func (p PipelineStatus) AsRecord() *tangled.PipelineStatus {
37
exitCode64 := int64(p.ExitCode)
38
finishedAt := p.FinishedAt.String()
39
40
return &tangled.PipelineStatus{
41
+
LexiconTypeID: tangled.PipelineStatusNSID,
42
+
Pipeline: p.Pipeline,
43
+
Status: string(p.Status),
44
45
ExitCode: &exitCode64,
46
Error: &p.Error,
···
55
return fmt.Sprintf("at://%s/did:web:%s/%s", tangled.PipelineStatusNSID, knot, rkey)
56
}
57
58
+
func (db *DB) CreatePipeline(rkey, pipeline string, n *notifier.Notifier) error {
59
_, err := db.Exec(`
60
+
insert into pipeline_status (rkey, status, pipeline)
61
+
values (?, ?, ?)
62
+
`, rkey, PipelinePending, pipeline)
63
64
if err != nil {
65
return err
···
68
return nil
69
}
70
71
+
func (db *DB) MarkPipelineRunning(rkey string, n *notifier.Notifier) error {
72
_, err := db.Exec(`
73
+
update pipeline_status
74
set status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
75
+
where rkey = ?
76
+
`, PipelineRunning, rkey)
77
78
if err != nil {
79
return err
···
82
return nil
83
}
84
85
+
func (db *DB) MarkPipelineFailed(rkey string, exitCode int, errorMsg string, n *notifier.Notifier) error {
86
_, err := db.Exec(`
87
+
update pipeline_status
88
set status = ?,
89
exit_code = ?,
90
error = ?,
91
updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now'),
92
finished_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
93
+
where rkey = ?
94
+
`, PipelineFailed, exitCode, errorMsg, rkey)
95
if err != nil {
96
return err
97
}
···
99
return nil
100
}
101
102
+
func (db *DB) MarkPipelineTimeout(rkey string, n *notifier.Notifier) error {
103
_, err := db.Exec(`
104
+
update pipeline_status
105
set status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
106
+
where rkey = ?
107
+
`, PipelineTimeout, rkey)
108
if err != nil {
109
return err
110
}
···
112
return nil
113
}
114
115
+
func (db *DB) MarkPipelineSuccess(rkey string, n *notifier.Notifier) error {
116
_, err := db.Exec(`
117
+
update pipeline_status
118
set status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now'),
119
finished_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
120
+
where rkey = ?
121
+
`, PipelineSuccess, rkey)
122
123
if err != nil {
124
return err
···
127
return nil
128
}
129
130
+
func (db *DB) GetPipelineStatus(rkey string) (PipelineStatus, error) {
131
+
var p PipelineStatus
132
err := db.QueryRow(`
133
select rkey, status, error, exit_code, started_at, updated_at, finished_at
134
from pipelines
135
+
where rkey = ?
136
+
`, rkey).Scan(&p.Rkey, &p.Status, &p.Error, &p.ExitCode, &p.StartedAt, &p.UpdatedAt, &p.FinishedAt)
137
return p, err
138
}
139
140
+
func (db *DB) GetPipelineStatusAsRecords(cursor string) ([]PipelineStatus, error) {
141
whereClause := ""
142
args := []any{}
143
if cursor != "" {
···
147
148
query := fmt.Sprintf(`
149
select rkey, status, error, exit_code, started_at, updated_at, finished_at
150
+
from pipeline_status
151
%s
152
order by rkey asc
153
limit 100
···
159
}
160
defer rows.Close()
161
162
+
var pipelines []PipelineStatus
163
for rows.Next() {
164
+
var p PipelineStatus
165
rows.Scan(&p.Rkey, &p.Status, &p.Error, &p.ExitCode, &p.StartedAt, &p.UpdatedAt, &p.FinishedAt)
166
pipelines = append(pipelines, p)
167
}
168
169
if err := rows.Err(); err != nil {
170
return nil, err
171
+
}
172
+
173
+
records := []*tangled.PipelineStatus{}
174
+
for _, p := range pipelines {
175
+
records = append(records, p.AsRecord())
176
}
177
178
return pipelines, nil
+2
-2
spindle/engine/engine.go
+2
-2
spindle/engine/engine.go
···
47
48
// SetupPipeline sets up a new network for the pipeline, and possibly volumes etc.
49
// in the future. In here also goes other setup steps.
50
-
func (e *Engine) SetupPipeline(ctx context.Context, pipeline *tangled.Pipeline, id string) error {
51
e.l.Info("setting up pipeline", "pipeline", id)
52
53
_, err := e.docker.VolumeCreate(ctx, volume.CreateOptions{
···
73
return err
74
}
75
76
-
err = e.db.CreatePipeline(id, e.n)
77
return err
78
}
79
···
47
48
// SetupPipeline sets up a new network for the pipeline, and possibly volumes etc.
49
// in the future. In here also goes other setup steps.
50
+
func (e *Engine) SetupPipeline(ctx context.Context, pipeline *tangled.Pipeline, atUri, id string) error {
51
e.l.Info("setting up pipeline", "pipeline", id)
52
53
_, err := e.docker.VolumeCreate(ctx, volume.CreateOptions{
···
73
return err
74
}
75
76
+
err = e.db.CreatePipeline(id, atUri, e.n)
77
return err
78
}
79
+19
-14
spindle/server.go
+19
-14
spindle/server.go
···
70
go func() {
71
logger.Info("starting event consumer")
72
knotEventSource := knotclient.NewEventSource("localhost:5555")
73
-
ccfg := knotclient.ConsumerConfig{
74
-
Logger: logger,
75
-
ProcessFunc: spindle.exec,
76
-
}
77
ccfg.AddEventSource(knotEventSource)
78
79
-
ec := knotclient.NewEventConsumer(ccfg)
80
81
ec.Start(ctx)
82
}()
···
95
}
96
97
func (s *Spindle) exec(ctx context.Context, src knotclient.EventSource, msg knotclient.Message) error {
98
-
pipeline := tangled.Pipeline{}
99
-
err := json.Unmarshal(msg.EventJson, &pipeline)
100
-
if err != nil {
101
-
fmt.Println("error unmarshalling", err)
102
-
return err
103
-
}
104
105
-
if msg.Nsid == tangled.PipelineNSID {
106
-
err = s.eng.SetupPipeline(ctx, &pipeline, msg.Rkey)
107
if err != nil {
108
return err
109
}
110
-
err = s.eng.StartWorkflows(ctx, &pipeline, msg.Rkey)
111
if err != nil {
112
return err
113
}
···
70
go func() {
71
logger.Info("starting event consumer")
72
knotEventSource := knotclient.NewEventSource("localhost:5555")
73
+
74
+
ccfg := knotclient.NewConsumerConfig()
75
+
ccfg.Logger = logger
76
+
ccfg.Dev = cfg.Server.Dev
77
+
ccfg.ProcessFunc = spindle.exec
78
ccfg.AddEventSource(knotEventSource)
79
80
+
ec := knotclient.NewEventConsumer(*ccfg)
81
82
ec.Start(ctx)
83
}()
···
96
}
97
98
func (s *Spindle) exec(ctx context.Context, src knotclient.EventSource, msg knotclient.Message) error {
99
+
if msg.Nsid == tangled.PipelineNSID {
100
+
pipeline := tangled.Pipeline{}
101
+
err := json.Unmarshal(msg.EventJson, &pipeline)
102
+
if err != nil {
103
+
fmt.Println("error unmarshalling", err)
104
+
return err
105
+
}
106
107
+
// this is a "fake" at uri for now
108
+
pipelineAtUri := fmt.Sprintf("at://%s/did:web:%s/%s", tangled.PipelineNSID, pipeline.TriggerMetadata.Repo.Knot, msg.Rkey)
109
+
110
+
rkey := TID()
111
+
err = s.eng.SetupPipeline(ctx, &pipeline, pipelineAtUri, rkey)
112
if err != nil {
113
return err
114
}
115
+
err = s.eng.StartWorkflows(ctx, &pipeline, rkey)
116
if err != nil {
117
return err
118
}
+3
-2
spindle/stream.go
+3
-2
spindle/stream.go
···
4
"net/http"
5
"time"
6
7
"github.com/gorilla/websocket"
8
-
"golang.org/x/net/context"
9
)
10
11
var upgrader = websocket.Upgrader{
···
74
}
75
76
func (s *Spindle) streamPipelines(conn *websocket.Conn, cursor *string) error {
77
-
ops, err := s.db.GetPipelines(*cursor)
78
if err != nil {
79
s.l.Debug("err", "err", err)
80
return err
···
4
"net/http"
5
"time"
6
7
+
"context"
8
+
9
"github.com/gorilla/websocket"
10
)
11
12
var upgrader = websocket.Upgrader{
···
75
}
76
77
func (s *Spindle) streamPipelines(conn *websocket.Conn, cursor *string) error {
78
+
ops, err := s.db.GetPipelineStatusAsRecords(*cursor)
79
if err != nil {
80
s.l.Debug("err", "err", err)
81
return err