tangled.org
Signed-off-by: dawn dawn@tangled.org
+607
-169
Diff
Round #7
-1
docker-compose.yml
-1
docker-compose.yml
···
181
181
SPINDLE_MICROVM_PIPELINES_IMAGE_DIR: /var/lib/spindle/images
182
182
SPINDLE_MICROVM_PIPELINES_OVERLAY_DIR: /var/lib/spindle/overlays
183
183
SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11240"
184
-
SPINDLE_S3_LOG_BUCKET: ""
185
184
SPINDLE_MICROVM_PIPELINES_ENABLE_CGROUPS: "false"
186
185
# route guest nix substitution + uploads through the local ncps cache.
187
186
# ncps re-signs on serve with cache.local's key, so the guest trusts the
+22
-4
nix/modules/spindle.nix
+22
-4
nix/modules/spindle.nix
···
136
136
};
137
137
};
138
138
139
-
pipelines = {
140
-
logBucket = mkOption {
139
+
artifactStores = {
140
+
disk.dir = mkOption {
141
+
type = types.path;
142
+
default = "/var/log/spindle";
143
+
description = "Root directory for disk artifacts";
144
+
};
145
+
146
+
s3.bucket = mkOption {
141
147
type = types.str;
142
148
default = "tangled-logs";
143
-
description = "S3 bucket for workflow logs";
149
+
description = "S3 bucket for artifacts";
144
150
};
151
+
152
+
s3.region = mkOption {
153
+
type = types.str;
154
+
default = "us-east-1";
155
+
description = "AWS region for the artifact bucket";
156
+
};
157
+
};
158
+
159
+
160
+
pipelines = {
145
161
workflowTimeout = mkOption {
146
162
type = types.str;
147
163
default = "5m";
···
388
404
"SPINDLE_NIX_CACHE_READ_URLS=${concatStringsSep "," cfg.pipelines.nixCache.readUrls}"
389
405
"SPINDLE_NIX_CACHE_TRUSTED_PUBLIC_KEYS=${concatStringsSep "," cfg.pipelines.nixCache.trustedPublicKeys}"
390
406
"SPINDLE_NIX_CACHE_UPLOAD_URL=${cfg.pipelines.nixCache.uploadUrl}"
391
-
"SPINDLE_S3_LOG_BUCKET=${cfg.pipelines.logBucket}"
407
+
"SPINDLE_ARTIFACT_STORES_DISK_DIR=${cfg.artifactStores.disk.dir}"
408
+
"SPINDLE_ARTIFACT_STORES_S3_BUCKET=${cfg.artifactStores.s3.bucket}"
409
+
"SPINDLE_ARTIFACT_STORES_S3_REGION=${cfg.artifactStores.s3.region}"
392
410
];
393
411
ExecStart = "${cfg.package}/bin/spindle";
394
412
Restart = "always";
+2
-1
nix/vm.nix
+2
-1
nix/vm.nix
···
162
162
};
163
163
};
164
164
165
+
artifactStores.s3.bucket = envVarOr "SPINDLE_ARTIFACT_STORES_S3_BUCKET" "tangled-logs";
166
+
165
167
pipelines = {
166
-
logBucket = envVarOr "SPINDLE_S3_LOG_BUCKET" "";
167
168
microvm.enableKVM = nestedVirt;
168
169
nixCache = {
169
170
readUrls = ["http://127.0.0.1:8501"];
+240
spindle/artifactstore/artifactstore.go
+240
spindle/artifactstore/artifactstore.go
···
1
+
package artifactstore
2
+
3
+
import (
4
+
"context"
5
+
"errors"
6
+
"fmt"
7
+
"io"
8
+
"os"
9
+
"path/filepath"
10
+
"strings"
11
+
12
+
"github.com/aws/aws-sdk-go-v2/aws"
13
+
"github.com/aws/aws-sdk-go-v2/config"
14
+
"github.com/aws/aws-sdk-go-v2/service/s3"
15
+
spindleconfig "tangled.org/core/spindle/config"
16
+
)
17
+
18
+
type Writer interface {
19
+
Put(ctx context.Context, ref string, r io.Reader) error
20
+
}
21
+
22
+
type Reader interface {
23
+
Open(ctx context.Context, ref string) (io.ReadCloser, error)
24
+
}
25
+
26
+
type Store interface {
27
+
Writer
28
+
Reader
29
+
}
30
+
31
+
type DiskStore struct {
32
+
root string
33
+
}
34
+
35
+
func NewDiskStore(root string) (*DiskStore, error) {
36
+
if root == "" {
37
+
return nil, fmt.Errorf("artifact disk directory is required")
38
+
}
39
+
return &DiskStore{root: filepath.Clean(root)}, nil
40
+
}
41
+
42
+
func (s *DiskStore) Put(_ context.Context, ref string, r io.Reader) error {
43
+
path, err := s.resolve(ref)
44
+
if err != nil {
45
+
return err
46
+
}
47
+
if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
48
+
return fmt.Errorf("mkdir for artifact %q: %w", path, err)
49
+
}
50
+
tmpFile, err := os.CreateTemp(filepath.Dir(path), ".tmp-artifact-*")
51
+
if err != nil {
52
+
return fmt.Errorf("create temp artifact: %w", err)
53
+
}
54
+
tmpPath := tmpFile.Name()
55
+
defer func() {
56
+
_ = tmpFile.Close()
57
+
_ = os.Remove(tmpPath)
58
+
}()
59
+
if _, err := io.Copy(tmpFile, r); err != nil {
60
+
return fmt.Errorf("write artifact content: %w", err)
61
+
}
62
+
if err := tmpFile.Sync(); err != nil {
63
+
return fmt.Errorf("sync artifact file: %w", err)
64
+
}
65
+
if err := tmpFile.Close(); err != nil {
66
+
return fmt.Errorf("close artifact file: %w", err)
67
+
}
68
+
if err := os.Rename(tmpPath, path); err != nil {
69
+
return fmt.Errorf("rename artifact file to target: %w", err)
70
+
}
71
+
return nil
72
+
}
73
+
74
+
func (s *DiskStore) Open(_ context.Context, ref string) (io.ReadCloser, error) {
75
+
path, err := s.resolve(ref)
76
+
if err != nil {
77
+
return nil, err
78
+
}
79
+
f, err := os.Open(path)
80
+
if err != nil {
81
+
return nil, fmt.Errorf("open disk artifact %q: %w", path, err)
82
+
}
83
+
return f, nil
84
+
}
85
+
86
+
func (s *DiskStore) resolve(ref string) (string, error) {
87
+
if ref == "" || filepath.IsAbs(ref) {
88
+
return "", fmt.Errorf("invalid disk artifact ref %q", ref)
89
+
}
90
+
path := filepath.Join(s.root, filepath.Clean(ref))
91
+
rel, err := filepath.Rel(s.root, path)
92
+
if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
93
+
return "", fmt.Errorf("artifact ref %q escapes disk root %q", ref, s.root)
94
+
}
95
+
return path, nil
96
+
}
97
+
98
+
type s3API interface {
99
+
PutObject(ctx context.Context, params *s3.PutObjectInput, optFns ...func(*s3.Options)) (*s3.PutObjectOutput, error)
100
+
GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error)
101
+
}
102
+
103
+
type S3Store struct {
104
+
client s3API
105
+
bucket string
106
+
}
107
+
108
+
func NewS3Store(client s3API, bucket string) (*S3Store, error) {
109
+
if client == nil {
110
+
return nil, fmt.Errorf("s3 client is required")
111
+
}
112
+
if bucket == "" {
113
+
return nil, fmt.Errorf("artifact S3 bucket is required")
114
+
}
115
+
return &S3Store{client: client, bucket: bucket}, nil
116
+
}
117
+
118
+
func (s *S3Store) Put(ctx context.Context, ref string, r io.Reader) error {
119
+
if err := validateObjectRef(ref); err != nil {
120
+
return err
121
+
}
122
+
_, err := s.client.PutObject(ctx, &s3.PutObjectInput{
123
+
Bucket: aws.String(s.bucket),
124
+
Key: aws.String(ref),
125
+
Body: r,
126
+
})
127
+
if err != nil {
128
+
return fmt.Errorf("s3 put object: %w", err)
129
+
}
130
+
return nil
131
+
}
132
+
133
+
func (s *S3Store) Open(ctx context.Context, ref string) (io.ReadCloser, error) {
134
+
if err := validateObjectRef(ref); err != nil {
135
+
return nil, err
136
+
}
137
+
res, err := s.client.GetObject(ctx, &s3.GetObjectInput{
138
+
Bucket: aws.String(s.bucket),
139
+
Key: aws.String(ref),
140
+
})
141
+
if err != nil {
142
+
return nil, fmt.Errorf("s3 get object: %w", err)
143
+
}
144
+
return res.Body, nil
145
+
}
146
+
147
+
func validateObjectRef(ref string) error {
148
+
if ref == "" || strings.HasPrefix(ref, "/") || strings.Contains(ref, "://") {
149
+
return fmt.Errorf("invalid artifact ref %q", ref)
150
+
}
151
+
return nil
152
+
}
153
+
154
+
type Stores struct {
155
+
order []string
156
+
stores map[string]Store
157
+
}
158
+
159
+
func NewStores(cfg spindleconfig.ArtifactStores, diskFallback, legacyS3Bucket string) (*Stores, error) {
160
+
stores := &Stores{stores: make(map[string]Store)}
161
+
diskDir := cfg.Disk.Dir
162
+
if diskDir == "" {
163
+
diskDir = diskFallback
164
+
}
165
+
if diskDir != "" {
166
+
disk, err := NewDiskStore(diskDir)
167
+
if err != nil {
168
+
return nil, err
169
+
}
170
+
stores.order = append(stores.order, "disk")
171
+
stores.stores["disk"] = disk
172
+
}
173
+
174
+
bucket := cfg.S3.Bucket
175
+
if bucket == "" {
176
+
bucket = legacyS3Bucket
177
+
}
178
+
if bucket != "" {
179
+
awsCfg, err := config.LoadDefaultConfig(context.Background(), config.WithRegion(cfg.S3.Region))
180
+
if err != nil {
181
+
return nil, fmt.Errorf("load aws config: %w", err)
182
+
}
183
+
s3Store, err := NewS3Store(s3.NewFromConfig(awsCfg), bucket)
184
+
if err != nil {
185
+
return nil, err
186
+
}
187
+
stores.order = append(stores.order, "s3")
188
+
stores.stores["s3"] = s3Store
189
+
}
190
+
return stores, nil
191
+
}
192
+
193
+
func (s *Stores) Names() []string {
194
+
return append([]string(nil), s.order...)
195
+
}
196
+
197
+
func (s *Stores) Store(name string) (Store, bool) {
198
+
store, ok := s.stores[name]
199
+
return store, ok
200
+
}
201
+
202
+
func (s *Stores) Open(ctx context.Context, ref string) (io.ReadCloser, error) {
203
+
var errs []error
204
+
for _, name := range s.order {
205
+
rc, err := s.stores[name].Open(ctx, ref)
206
+
if err == nil {
207
+
return rc, nil
208
+
}
209
+
errs = append(errs, fmt.Errorf("%s: %w", name, err))
210
+
}
211
+
return nil, fmt.Errorf("open artifact %q: %w", ref, errors.Join(errs...))
212
+
}
213
+
214
+
func (s *Stores) PutFile(ctx context.Context, ref, sourcePath string) []error {
215
+
var errs []error
216
+
for _, name := range s.order {
217
+
store := s.stores[name]
218
+
if disk, ok := store.(*DiskStore); ok {
219
+
target, err := disk.resolve(ref)
220
+
if err == nil {
221
+
source, sourceErr := filepath.Abs(sourcePath)
222
+
targetAbs, targetErr := filepath.Abs(target)
223
+
if sourceErr == nil && targetErr == nil && source == targetAbs {
224
+
continue
225
+
}
226
+
}
227
+
}
228
+
file, err := os.Open(sourcePath)
229
+
if err != nil {
230
+
errs = append(errs, fmt.Errorf("%s: open source: %w", name, err))
231
+
continue
232
+
}
233
+
err = store.Put(ctx, ref, file)
234
+
_ = file.Close()
235
+
if err != nil {
236
+
errs = append(errs, fmt.Errorf("%s: %w", name, err))
237
+
}
238
+
}
239
+
return errs
240
+
}
+130
spindle/artifactstore/artifactstore_test.go
+130
spindle/artifactstore/artifactstore_test.go
···
1
+
package artifactstore
2
+
3
+
import (
4
+
"bytes"
5
+
"context"
6
+
"io"
7
+
"os"
8
+
"strings"
9
+
"sync"
10
+
"testing"
11
+
12
+
"github.com/aws/aws-sdk-go-v2/service/s3"
13
+
)
14
+
15
+
type mockS3Client struct {
16
+
mu sync.Mutex
17
+
store map[string][]byte
18
+
}
19
+
20
+
func newMockS3Client() *mockS3Client {
21
+
return &mockS3Client{
22
+
store: make(map[string][]byte),
23
+
}
24
+
}
25
+
26
+
func (m *mockS3Client) PutObject(ctx context.Context, params *s3.PutObjectInput, optFns ...func(*s3.Options)) (*s3.PutObjectOutput, error) {
27
+
m.mu.Lock()
28
+
defer m.mu.Unlock()
29
+
b, err := io.ReadAll(params.Body)
30
+
if err != nil {
31
+
return nil, err
32
+
}
33
+
key := *params.Bucket + "/" + *params.Key
34
+
m.store[key] = b
35
+
return &s3.PutObjectOutput{}, nil
36
+
}
37
+
38
+
func (m *mockS3Client) GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
39
+
m.mu.Lock()
40
+
defer m.mu.Unlock()
41
+
key := *params.Bucket + "/" + *params.Key
42
+
data, ok := m.store[key]
43
+
if !ok {
44
+
return nil, os.ErrNotExist
45
+
}
46
+
return &s3.GetObjectOutput{
47
+
Body: io.NopCloser(bytes.NewReader(data)),
48
+
}, nil
49
+
}
50
+
51
+
func TestDiskStore(t *testing.T) {
52
+
tempDir := t.TempDir()
53
+
store, err := NewDiskStore(tempDir)
54
+
if err != nil {
55
+
t.Fatal(err)
56
+
}
57
+
58
+
ctx := context.Background()
59
+
ref := "logs/test.log"
60
+
content := "hello world log content"
61
+
62
+
if err := store.Put(ctx, ref, strings.NewReader(content)); err != nil {
63
+
t.Fatalf("Put failed: %v", err)
64
+
}
65
+
66
+
rc, err := store.Open(ctx, ref)
67
+
if err != nil {
68
+
t.Fatalf("Open failed: %v", err)
69
+
}
70
+
defer rc.Close()
71
+
72
+
got, err := io.ReadAll(rc)
73
+
if err != nil {
74
+
t.Fatalf("ReadAll failed: %v", err)
75
+
}
76
+
if string(got) != content {
77
+
t.Fatalf("got content %q, want %q", string(got), content)
78
+
}
79
+
}
80
+
81
+
func TestDiskStoreTraversalProtection(t *testing.T) {
82
+
tempDir := t.TempDir()
83
+
store, err := NewDiskStore(tempDir)
84
+
if err != nil {
85
+
t.Fatal(err)
86
+
}
87
+
88
+
ctx := context.Background()
89
+
badRef := "../outside"
90
+
91
+
err = store.Put(ctx, badRef, strings.NewReader("bad"))
92
+
if err == nil {
93
+
t.Fatal("expected error putting file outside diskDir, got nil")
94
+
}
95
+
96
+
_, err = store.Open(ctx, badRef)
97
+
if err == nil {
98
+
t.Fatal("expected error opening file outside diskDir, got nil")
99
+
}
100
+
}
101
+
102
+
func TestS3Store(t *testing.T) {
103
+
mock := newMockS3Client()
104
+
store, err := NewS3Store(mock, "mybucket")
105
+
if err != nil {
106
+
t.Fatal(err)
107
+
}
108
+
109
+
ctx := context.Background()
110
+
ref := "logs/run1.log"
111
+
content := "s3 log payload"
112
+
113
+
if err := store.Put(ctx, ref, strings.NewReader(content)); err != nil {
114
+
t.Fatalf("Put to S3 failed: %v", err)
115
+
}
116
+
117
+
rc, err := store.Open(ctx, ref)
118
+
if err != nil {
119
+
t.Fatalf("Open from S3 failed: %v", err)
120
+
}
121
+
defer rc.Close()
122
+
123
+
got, err := io.ReadAll(rc)
124
+
if err != nil {
125
+
t.Fatalf("ReadAll failed: %v", err)
126
+
}
127
+
if string(got) != content {
128
+
t.Fatalf("got content %q, want %q", string(got), content)
129
+
}
130
+
}
+17
-2
spindle/config/config.go
+17
-2
spindle/config/config.go
···
57
57
MaxConcurrentWorkflows int `env:"MAX_CONCURRENT_WORKFLOWS, default=8"` // max number of workflow containers running at once (memory cap)
58
58
}
59
59
60
-
type S3 struct {
60
+
type ArtifactStoreDisk struct {
61
+
Dir string `env:"DIR"`
62
+
}
63
+
64
+
type ArtifactStoreS3 struct {
65
+
Bucket string `env:"BUCKET"`
66
+
Region string `env:"REGION, default=us-east-1"`
67
+
}
68
+
69
+
type ArtifactStores struct {
70
+
Disk ArtifactStoreDisk `env:",prefix=DISK_"`
71
+
S3 ArtifactStoreS3 `env:",prefix=S3_"`
72
+
}
73
+
74
+
type LegacyS3 struct {
61
75
LogBucket string `env:"LOG_BUCKET"`
62
76
}
63
77
···
98
112
NixeryPipelines NixeryPipelines `env:",prefix=SPINDLE_NIXERY_PIPELINES_"`
99
113
MicroVMPipelines MicroVMPipelines `env:",prefix=SPINDLE_MICROVM_PIPELINES_"`
100
114
NixCache NixCache `env:",prefix=SPINDLE_NIX_CACHE_"`
101
-
S3 S3 `env:",prefix=SPINDLE_S3_"`
115
+
ArtifactStores ArtifactStores `env:",prefix=SPINDLE_ARTIFACT_STORES_"`
116
+
LegacyS3 LegacyS3 `env:",prefix=SPINDLE_S3_"`
102
117
}
103
118
104
119
func Load(ctx context.Context) (*Config, error) {
+32
spindle/db/artifacts.go
+32
spindle/db/artifacts.go
···
1
+
package db
2
+
3
+
type FinishedLog struct {
4
+
LeaseID string
5
+
Workflow string
6
+
Ref string
7
+
Hash string
8
+
}
9
+
10
+
func (d *DB) GetFinishedLog(workflow string) (*FinishedLog, error) {
11
+
var fl FinishedLog
12
+
err := d.QueryRow(
13
+
`select lease_id, workflow, ref, hash
14
+
from mill_artifacts
15
+
where workflow = ?
16
+
order by id desc limit 1`,
17
+
workflow,
18
+
).Scan(&fl.LeaseID, &fl.Workflow, &fl.Ref, &fl.Hash)
19
+
if err != nil {
20
+
return nil, err
21
+
}
22
+
return &fl, nil
23
+
}
24
+
25
+
func (d *DB) SaveArtifactRef(leaseID, workflow, ref, hash string) error {
26
+
_, err := d.Exec(
27
+
`insert into mill_artifacts (lease_id, workflow, ref, hash)
28
+
values (?, ?, ?, ?)`,
29
+
leaseID, workflow, ref, hash,
30
+
)
31
+
return err
32
+
}
+8
spindle/db/db.go
+8
spindle/db/db.go
···
132
132
foreign key (pipeline_id) references pipelines(id) on delete cascade
133
133
);
134
134
135
+
create table if not exists mill_artifacts (
136
+
id integer primary key autoincrement,
137
+
lease_id text not null,
138
+
workflow text not null,
139
+
ref text not null,
140
+
hash text not null
141
+
);
142
+
135
143
create table if not exists migrations (
136
144
id integer primary key autoincrement,
137
145
name text unique
+42
-16
spindle/engine/engine.go
+42
-16
spindle/engine/engine.go
···
2
2
3
3
import (
4
4
"context"
5
+
"crypto/sha256"
6
+
"encoding/hex"
5
7
"errors"
6
8
"fmt"
9
+
"io"
7
10
"log/slog"
8
-
"path/filepath"
11
+
"os"
9
12
"sync"
13
+
"time"
10
14
11
15
"tangled.org/core/notifier"
16
+
"tangled.org/core/spindle/artifactstore"
12
17
"tangled.org/core/spindle/config"
13
18
"tangled.org/core/spindle/db"
14
19
"tangled.org/core/spindle/models"
···
66
71
FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error
67
72
}
68
73
69
-
func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
74
+
func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, stores *artifactstore.Stores, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
70
75
l.Info("starting all workflows in parallel", "pipeline", pipelineId)
71
76
72
77
var allSecrets []secrets.UnlockedSecret
···
84
89
secretValues[i] = s.Value
85
90
}
86
91
87
-
s3, err := NewS3(cfg.S3.LogBucket)
88
-
if err != nil {
89
-
l.Error("error creating s3 client", "err", err)
90
-
}
91
-
92
92
// wid.String() is lossy so two different names can map to the same key
93
93
// eg. "foo bar" and "foo-bar"...
94
94
wfCounts := make(map[string]int)
···
128
128
l.Info("skipping finished workflow", "wid", wid, "status", st.Status)
129
129
return
130
130
}
131
-
defer func() {
132
-
if s3 != nil {
133
-
logFile := filepath.Join(cfg.Server.LogDir, fmt.Sprintf("%s.log", wid.String()))
134
-
if err := s3.WriteFile(ctx, logFile); err != nil {
135
-
l.Error("error uploading logs", "err", err)
136
-
}
137
-
}
138
-
}()
139
-
140
131
wfLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues)
141
132
if err != nil {
142
133
l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
143
134
wfLogger = models.NullLogger{}
144
135
} else {
145
136
l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
137
+
defer archiveWorkflowLog(l, stores, db, cfg.Server.LogDir, wid)
146
138
defer wfLogger.Close()
147
139
}
148
140
···
235
227
wg.Wait()
236
228
l.Info("all workflows completed")
237
229
}
230
+
231
+
func archiveWorkflowLog(l *slog.Logger, stores *artifactstore.Stores, database *db.DB, logDir string, wid models.WorkflowId) {
232
+
if stores == nil {
233
+
return
234
+
}
235
+
logPath := models.LogFilePath(logDir, wid)
236
+
file, err := os.Open(logPath)
237
+
if err != nil {
238
+
l.Error("open workflow log for archival", "wid", wid, "err", err)
239
+
return
240
+
}
241
+
hash := sha256.New()
242
+
if _, err := io.Copy(hash, file); err != nil {
243
+
_ = file.Close()
244
+
l.Error("hash workflow log", "wid", wid, "err", err)
245
+
return
246
+
}
247
+
_ = file.Close()
248
+
249
+
ref := wid.String() + ".log"
250
+
uploadCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
251
+
defer cancel()
252
+
errs := stores.PutFile(uploadCtx, ref, logPath)
253
+
for _, err := range errs {
254
+
l.Error("archive workflow log", "wid", wid, "err", err)
255
+
}
256
+
if len(errs) == len(stores.Names()) {
257
+
return
258
+
}
259
+
digest := "sha256:" + hex.EncodeToString(hash.Sum(nil))
260
+
if err := database.SaveArtifactRef(wid.String(), wid.Name, ref, digest); err != nil {
261
+
l.Error("save workflow log artifact", "wid", wid, "err", err)
262
+
}
263
+
}
+3
-3
spindle/engine/engine_test.go
+3
-3
spindle/engine/engine_test.go
···
114
114
}
115
115
116
116
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
117
-
StartWorkflows(logger, nil, cfg, testDB, nil, context.Background(), pipeline, pipelineId)
117
+
StartWorkflows(logger, nil, cfg, nil, testDB, nil, context.Background(), pipeline, pipelineId)
118
118
119
119
eng.mu.Lock()
120
120
setupCalls := append([]models.WorkflowId(nil), eng.setupCalls...)
···
195
195
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
196
196
doneChan := make(chan struct{})
197
197
go func() {
198
-
StartWorkflows(logger, nil, cfg, testDB, nil, context.Background(), pipeline, pipelineId)
198
+
StartWorkflows(logger, nil, cfg, nil, testDB, nil, context.Background(), pipeline, pipelineId)
199
199
close(doneChan)
200
200
}()
201
201
···
250
250
}
251
251
252
252
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
253
-
StartWorkflows(logger, nil, cfg, testDB, nil, context.Background(), pipeline, pipelineId)
253
+
StartWorkflows(logger, nil, cfg, nil, testDB, nil, context.Background(), pipeline, pipelineId)
254
254
255
255
st, err := testDB.GetStatus(wid)
256
256
if err != nil {
-75
spindle/engine/s3.go
-75
spindle/engine/s3.go
···
1
-
package engine
2
-
3
-
import (
4
-
"context"
5
-
"fmt"
6
-
"io"
7
-
"os"
8
-
"path/filepath"
9
-
10
-
"github.com/aws/aws-sdk-go-v2/aws"
11
-
"github.com/aws/aws-sdk-go-v2/config"
12
-
"github.com/aws/aws-sdk-go-v2/service/s3"
13
-
)
14
-
15
-
type S3 struct {
16
-
bucket string
17
-
client *s3.Client
18
-
}
19
-
20
-
const BaseS3Path = "spindle/workflows"
21
-
22
-
func NewS3(bucket string) (*S3, error) {
23
-
if bucket == "" {
24
-
return nil, fmt.Errorf("s3 bucket not provided")
25
-
}
26
-
27
-
ctx := context.Background()
28
-
sdkConfig, err := config.LoadDefaultConfig(ctx)
29
-
30
-
if err != nil {
31
-
return nil, fmt.Errorf("error loading s3 config: %w", err)
32
-
}
33
-
s3Client := s3.NewFromConfig(sdkConfig)
34
-
35
-
return &S3{
36
-
bucket: bucket,
37
-
client: s3Client,
38
-
}, nil
39
-
}
40
-
41
-
func (s *S3) WriteFile(ctx context.Context, path string) error {
42
-
s3Key := fmt.Sprintf("%s/%s", BaseS3Path, filepath.Base(path))
43
-
44
-
file, err := os.Open(path)
45
-
if err != nil {
46
-
return fmt.Errorf("error opening file %s: %w", path, err)
47
-
}
48
-
defer file.Close()
49
-
50
-
_, err = s.client.PutObject(ctx, &s3.PutObjectInput{
51
-
Bucket: &s.bucket,
52
-
Key: &s3Key,
53
-
Body: file,
54
-
})
55
-
56
-
if err != nil {
57
-
return fmt.Errorf("error writing to s3: %w", err)
58
-
}
59
-
60
-
return nil
61
-
}
62
-
63
-
func (s *S3) ReadFile(ctx context.Context, name string) ([]byte, error) {
64
-
res, err := s.client.GetObject(ctx, &s3.GetObjectInput{
65
-
Bucket: &s.bucket,
66
-
Key: aws.String(name),
67
-
})
68
-
69
-
if err != nil {
70
-
return nil, fmt.Errorf("error reading file %s: %w", name, err)
71
-
}
72
-
defer res.Body.Close()
73
-
74
-
return io.ReadAll(res.Body)
75
-
}
+52
spindle/logview/logview.go
+52
spindle/logview/logview.go
···
1
+
package logview
2
+
3
+
import (
4
+
"bufio"
5
+
"context"
6
+
"io"
7
+
"strings"
8
+
9
+
"github.com/hpcloud/tail"
10
+
"tangled.org/core/spindle/artifactstore"
11
+
"tangled.org/core/spindle/db"
12
+
"tangled.org/core/spindle/models"
13
+
)
14
+
15
+
// streams a workflow's log lines: tails the log file while running, reads
16
+
// the uploaded artifact once finished, same for every role. stop ends a
17
+
// live follow early, the channel closes when the source drains or ctx ends
18
+
func Follow(ctx context.Context, d *db.DB, reader artifactstore.Reader, logDir string, wid models.WorkflowId, finished bool) (<-chan *tail.Line, func(), error) {
19
+
if finished && reader != nil && d != nil {
20
+
if fl, err := d.GetFinishedLog(wid.Name); err == nil && fl.Ref != "" {
21
+
rc, err := reader.Open(ctx, fl.Ref)
22
+
if err == nil {
23
+
ch := make(chan *tail.Line, 64)
24
+
followCtx, cancel := context.WithCancel(ctx)
25
+
go func() {
26
+
defer close(ch)
27
+
defer rc.Close()
28
+
scanner := bufio.NewScanner(rc)
29
+
for scanner.Scan() {
30
+
select {
31
+
case <-followCtx.Done():
32
+
return
33
+
case ch <- &tail.Line{Text: strings.TrimSuffix(scanner.Text(), "\r")}:
34
+
}
35
+
}
36
+
}()
37
+
return ch, cancel, nil
38
+
}
39
+
}
40
+
}
41
+
42
+
t, err := tail.TailFile(models.LogFilePath(logDir, wid), tail.Config{
43
+
Follow: !finished,
44
+
ReOpen: !finished,
45
+
MustExist: false,
46
+
Location: &tail.SeekInfo{Offset: 0, Whence: io.SeekStart},
47
+
})
48
+
if err != nil {
49
+
return nil, nil, err
50
+
}
51
+
return t.Lines, func() { _ = t.Stop() }, nil
52
+
}
+28
-11
spindle/server.go
+28
-11
spindle/server.go
···
32
32
"tangled.org/core/rbac"
33
33
"tangled.org/core/repoident"
34
34
"tangled.org/core/repoverify"
35
+
"tangled.org/core/spindle/artifactstore"
35
36
"tangled.org/core/spindle/config"
36
37
"tangled.org/core/spindle/db"
37
38
"tangled.org/core/spindle/engine"
···
71
72
motdMu sync.RWMutex
72
73
rootCtx context.Context
73
74
jobWake chan struct{}
75
+
stores *artifactstore.Stores
76
+
reader artifactstore.Reader
74
77
}
75
78
76
79
// New creates a new Spindle server with the provided configuration and engines.
···
164
167
rootCtx: ctx,
165
168
jobWake: make(chan struct{}, 1),
166
169
}
170
+
diskFallback := cfg.Server.LogDir
171
+
if cfg.ArtifactStores.Disk.Dir == "" {
172
+
logger.Warn("using SPINDLE_SERVER_LOG_DIR as the implicit disk artifact store; configure SPINDLE_ARTIFACT_STORES_DISK_DIR explicitly")
173
+
}
174
+
stores, err := artifactstore.NewStores(cfg.ArtifactStores, diskFallback, cfg.LegacyS3.LogBucket)
175
+
if err != nil {
176
+
return nil, fmt.Errorf("failed to setup artifact stores: %w", err)
177
+
}
178
+
spindle.stores = stores
179
+
spindle.reader = stores
180
+
if cfg.LegacyS3.LogBucket != "" {
181
+
logger.Warn("SPINDLE_S3_LOG_BUCKET is deprecated; use SPINDLE_ARTIFACT_STORES_S3_BUCKET")
182
+
}
167
183
168
184
err = e.AddSpindle(rbacDomain)
169
185
if err != nil {
···
387
403
l := log.SubLogger(s.l, "xrpc")
388
404
389
405
x := xrpc.Xrpc{
390
-
Logger: l,
391
-
Db: s.db,
392
-
Enforcer: s.e,
393
-
Engines: s.engs,
394
-
Config: s.cfg,
395
-
Resolver: s.res,
396
-
Vault: s.vault,
397
-
Notifier: s.Notifier(),
398
-
ServiceAuth: serviceAuth,
399
-
Trigger: s,
406
+
Logger: l,
407
+
Db: s.db,
408
+
Enforcer: s.e,
409
+
Engines: s.engs,
410
+
Config: s.cfg,
411
+
ArtifactReader: s.reader,
412
+
Resolver: s.res,
413
+
Vault: s.vault,
414
+
Notifier: s.Notifier(),
415
+
ServiceAuth: serviceAuth,
416
+
Trigger: s,
400
417
}
401
418
402
419
return x.Router()
···
820
837
workflows[eng] = append(workflows[eng], *ewf)
821
838
}
822
839
823
-
engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.rootCtx, &models.Pipeline{
840
+
engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.stores, s.db, s.n, s.rootCtx, &models.Pipeline{
824
841
RepoDid: syntax.DID(job.RepoDid),
825
842
Workflows: workflows,
826
843
TrustedSource: trustedSource,
+12
-26
spindle/stream.go
+12
-26
spindle/stream.go
···
4
4
"context"
5
5
"errors"
6
6
"fmt"
7
-
"io"
8
7
"net/http"
9
8
"time"
10
9
11
10
"tangled.org/core/eventstream"
12
11
"tangled.org/core/log"
12
+
"tangled.org/core/spindle/logview"
13
13
"tangled.org/core/spindle/models"
14
14
15
15
"github.com/go-chi/chi/v5"
16
16
"github.com/gorilla/websocket"
17
-
"github.com/hpcloud/tail"
18
17
)
19
18
20
19
var upgrader = websocket.Upgrader{
···
89
88
}
90
89
isFinished := models.StatusKind(status.Status).IsFinish()
91
90
92
-
filePath := models.LogFilePath(s.cfg.Server.LogDir, wid)
93
-
94
-
config := tail.Config{
95
-
Follow: !isFinished,
96
-
ReOpen: !isFinished,
97
-
MustExist: false,
98
-
Location: &tail.SeekInfo{
99
-
Offset: 0,
100
-
Whence: io.SeekStart,
101
-
},
102
-
// Logger: tail.DiscardingLogger,
103
-
}
104
-
105
-
t, err := tail.TailFile(filePath, config)
91
+
lines, stop, err := logview.Follow(ctx, s.db, s.reader, s.cfg.Server.LogDir, wid, isFinished)
106
92
if err != nil {
107
-
return fmt.Errorf("failed to tail log file: %w", err)
93
+
return fmt.Errorf("failed to follow workflow log: %w", err)
108
94
}
109
-
defer t.Stop()
110
-
95
+
defer stop()
111
96
for {
112
97
select {
113
98
case <-ctx.Done():
114
99
return ctx.Err()
115
-
case line := <-t.Lines:
116
-
if line == nil && isFinished {
117
-
return fmt.Errorf("tail completed")
100
+
case line, ok := <-lines:
101
+
if !ok && isFinished {
102
+
return fmt.Errorf("log completed")
103
+
}
104
+
if !ok {
105
+
return fmt.Errorf("log channel closed unexpectedly")
118
106
}
119
-
120
107
if line == nil {
121
-
return fmt.Errorf("tail channel closed unexpectedly")
108
+
continue
122
109
}
123
-
124
110
if line.Err != nil {
125
-
return fmt.Errorf("error tailing log file: %w", line.Err)
111
+
return fmt.Errorf("error following workflow log: %w", line.Err)
126
112
}
127
113
128
114
if err := conn.WriteMessage(websocket.TextMessage, []byte(line.Text)); err != nil {
+7
-20
spindle/xrpc/ci_pipeline_subscribe_logs.go
+7
-20
spindle/xrpc/ci_pipeline_subscribe_logs.go
···
4
4
"context"
5
5
"encoding/json"
6
6
"fmt"
7
-
"io"
8
7
"net/http"
9
8
"sync"
10
9
"time"
···
12
11
"github.com/bluesky-social/indigo/atproto/atclient"
13
12
"github.com/bluesky-social/indigo/atproto/syntax"
14
13
"github.com/gorilla/websocket"
15
-
"github.com/hpcloud/tail"
16
14
"tangled.org/core/api/tangled"
15
+
"tangled.org/core/spindle/logview"
17
16
"tangled.org/core/spindle/models"
18
17
)
19
18
···
166
165
isFinished = models.StatusKind(status.Status).IsFinish()
167
166
}
168
167
169
-
filePath := models.LogFilePath(x.Config.Server.LogDir, wid)
170
-
171
-
tailConfig := tail.Config{
172
-
Follow: !isFinished,
173
-
ReOpen: !isFinished,
174
-
MustExist: false,
175
-
Location: &tail.SeekInfo{
176
-
Offset: 0,
177
-
Whence: io.SeekStart,
178
-
},
179
-
}
180
-
181
-
t, err := tail.TailFile(filePath, tailConfig)
168
+
lines, stop, err := logview.Follow(ctx, x.Db, x.ArtifactReader, x.Config.Server.LogDir, wid, isFinished)
182
169
if err != nil {
183
-
l.Error("failed to tail log file", "workflow", wfName, "err", err)
170
+
l.Error("failed to follow workflow log", "workflow", wfName, "err", err)
184
171
return
185
172
}
186
-
defer t.Stop()
173
+
defer stop()
187
174
188
-
// if we are following, poll status in database to stop tailing when finished
175
+
// if we are following, poll status in database to stop when finished
189
176
if !isFinished {
190
177
go func() {
191
178
ticker := time.NewTicker(2 * time.Second)
···
197
184
case <-ticker.C:
198
185
status, err := x.Db.GetStatus(wid)
199
186
if err == nil && models.StatusKind(status.Status).IsFinish() {
200
-
t.Stop()
187
+
stop()
201
188
return
202
189
}
203
190
}
···
209
196
select {
210
197
case <-ctx.Done():
211
198
return
212
-
case line, ok := <-t.Lines:
199
+
case line, ok := <-lines:
213
200
if !ok || line == nil {
214
201
return
215
202
}
+12
-10
spindle/xrpc/xrpc.go
+12
-10
spindle/xrpc/xrpc.go
···
15
15
"tangled.org/core/idresolver"
16
16
"tangled.org/core/notifier"
17
17
"tangled.org/core/rbac"
18
+
"tangled.org/core/spindle/artifactstore"
18
19
"tangled.org/core/spindle/config"
19
20
"tangled.org/core/spindle/db"
20
21
"tangled.org/core/spindle/models"
···
41
42
}
42
43
43
44
type Xrpc struct {
44
-
Logger *slog.Logger
45
-
Db *db.DB
46
-
Enforcer *rbac.Enforcer
47
-
Engines map[string]models.Engine
48
-
Config *config.Config
49
-
Resolver *idresolver.Resolver
50
-
Vault secrets.Manager
51
-
Notifier *notifier.Notifier
52
-
ServiceAuth *serviceauth.ServiceAuth
53
-
Trigger PipelineTrigger
45
+
Logger *slog.Logger
46
+
Db *db.DB
47
+
Enforcer *rbac.Enforcer
48
+
Engines map[string]models.Engine
49
+
Config *config.Config
50
+
ArtifactReader artifactstore.Reader
51
+
Resolver *idresolver.Resolver
52
+
Vault secrets.Manager
53
+
Notifier *notifier.Notifier
54
+
ServiceAuth *serviceauth.ServiceAuth
55
+
Trigger PipelineTrigger
54
56
}
55
57
56
58
func (x *Xrpc) Router() http.Handler {
History
9 rounds
0 comments
1 commit
Expand
Collapse
spindle: add artifact store
Signed-off-by: dawn <dawn@tangled.org>
3/3 success
Expand
Collapse
Expand 0 comments
1 commit
Expand
Collapse
spindle: add artifact store
Signed-off-by: dawn <dawn@tangled.org>
1/3 failed, 2/3 cancelled
Expand
Collapse
Expand 0 comments
1 commit
Expand
Collapse
spindle: add artifact store
Signed-off-by: dawn <dawn@tangled.org>
2/3 success, 1/3 failed
Expand
Collapse
Expand 0 comments
1 commit
Expand
Collapse
spindle: implement mill
Signed-off-by: dawn <dawn@tangled.org>
3/3 success
Expand
Collapse
Expand 0 comments
1 commit
Expand
Collapse
spindle: implement mill
Signed-off-by: dawn <dawn@tangled.org>
2/3 success, 1/3 failed
Expand
Collapse
Expand 0 comments
1 commit
Expand
Collapse
spindle: implement mill
Signed-off-by: dawn <dawn@tangled.org>
2/3 success, 1/3 failed
Expand
Collapse
Expand 0 comments
1 commit
Expand
Collapse
spindle: implement mill
Signed-off-by: dawn <dawn@tangled.org>
1/3 success, 2/3 failed
Expand
Collapse
Expand 0 comments
1 commit
Expand
Collapse
spindle: implement mill
Signed-off-by: dawn <dawn@tangled.org>
3/3 failed
Expand
Collapse
Expand 0 comments
1 commit
Expand
Collapse
spindle: implement mill
Signed-off-by: dawn <dawn@tangled.org>