tangled.org / core
Monorepo for Tangled tangled.org
2
Fork 30

Configure Feed

Select the types of activity you want to include in your feed.

spindle: add artifact store

Signed-off-by: dawn <dawn@tangled.org>

author
dawn date (Jul 27, 2026, 12:54 PM +0300) commit 5d361cfc parent f02cd6df change-id kxzqyrzk
build.yml
fmt.yml
test.yml
+606 -169
-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
+21 -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 + pipelines = { 145 160 workflowTimeout = mkOption { 146 161 type = types.str; 147 162 default = "5m"; ··· 388 403 "SPINDLE_NIX_CACHE_READ_URLS=${concatStringsSep "," cfg.pipelines.nixCache.readUrls}" 389 404 "SPINDLE_NIX_CACHE_TRUSTED_PUBLIC_KEYS=${concatStringsSep "," cfg.pipelines.nixCache.trustedPublicKeys}" 390 405 "SPINDLE_NIX_CACHE_UPLOAD_URL=${cfg.pipelines.nixCache.uploadUrl}" 391 - "SPINDLE_S3_LOG_BUCKET=${cfg.pipelines.logBucket}" 406 + "SPINDLE_ARTIFACT_STORES_DISK_DIR=${cfg.artifactStores.disk.dir}" 407 + "SPINDLE_ARTIFACT_STORES_S3_BUCKET=${cfg.artifactStores.s3.bucket}" 408 + "SPINDLE_ARTIFACT_STORES_S3_REGION=${cfg.artifactStores.s3.region}" 392 409 ]; 393 410 ExecStart = "${cfg.package}/bin/spindle"; 394 411 Restart = "always";
+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
··· 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
··· 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
··· 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
··· 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
··· 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
··· 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 ··· 82 87 secretValues := make([]string, len(allSecrets)) 83 88 for i, s := range allSecrets { 84 89 secretValues[i] = s.Value 85 - } 86 - 87 - s3, err := NewS3(cfg.S3.LogBucket) 88 - if err != nil { 89 - l.Error("error creating s3 client", "err", err) 90 90 } 91 91 92 92 // wid.String() is lossy so two different names can map to the same key ··· 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
··· 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
··· 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
··· 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
··· 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. ··· 163 166 motd: defaultMotd, 164 167 rootCtx: ctx, 165 168 jobWake: make(chan struct{}, 1), 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") 166 182 } 167 183 168 184 err = e.AddSpindle(rbacDomain) ··· 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
··· 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
··· 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
··· 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 {