Monorepo for Tangled
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513package spindle
import ( "context" "crypto/sha256" "database/sql" _ "embed" "encoding/binary" "encoding/json" "errors" "fmt" "io" "log/slog" "maps" "net/http" "path/filepath" "sort" "sync" "time"
"github.com/bluesky-social/indigo/atproto/syntax" indigoxrpc "github.com/bluesky-social/indigo/xrpc" "github.com/go-chi/chi/v5" "github.com/go-git/go-git/v5/plumbing" "github.com/go-git/go-git/v5/plumbing/object" "github.com/hashicorp/go-version" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "tangled.org/core/api/tangled" "tangled.org/core/eventconsumer" "tangled.org/core/eventconsumer/cursor" "tangled.org/core/eventstream" "tangled.org/core/gitutil" "tangled.org/core/idresolver" "tangled.org/core/jetstream" knotdb "tangled.org/core/knotserver/db" kgit "tangled.org/core/knotserver/git" "tangled.org/core/log" "tangled.org/core/notifier" "tangled.org/core/rbac" "tangled.org/core/repoident" "tangled.org/core/repoverify" "tangled.org/core/spindle/artifactstore" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" "tangled.org/core/spindle/engine" "tangled.org/core/spindle/engines/dummy" "tangled.org/core/spindle/engines/nixery" "tangled.org/core/spindle/mill" "tangled.org/core/spindle/mill/executor" "tangled.org/core/spindle/models" "tangled.org/core/spindle/observability" "tangled.org/core/spindle/quota" "tangled.org/core/spindle/secrets" "tangled.org/core/spindle/storage" "tangled.org/core/spindle/webhook" "tangled.org/core/spindle/xrpc" "tangled.org/core/tid" "tangled.org/core/workflow" "tangled.org/core/xrpc/serviceauth")
//go:embed motdvar defaultMotd []byte
const ( rbacDomain = "thisserver" executorShutdownTimeout = 8 * time.Minute serverShutdownTimeout = 10 * time.Second
// sparse-checkout wants a leading slash to anchor the pattern at the root sparseWorkflowDir = "/" + workflow.WorkflowDir)
type executorClient interface { Connect(context.Context) Drain(context.Context) error RegisterMetrics(*observability.Metrics) SetQuotaClient(*executor.QuotaClient)}
type Spindle struct { jc *jetstream.JetstreamClient tap *Tap embedTap *embeddedTap db *db.DB e *rbac.Enforcer l *slog.Logger n *notifier.Notifier wh *webhook.Service metrics *observability.Metrics engs map[string]models.Engine jobWake chan struct{} jobWorkers sync.WaitGroup scheduler *pipelineScheduler cfg *config.Config ks *eventconsumer.Consumer res *idresolver.Resolver verify repoverify.Verifier vault secrets.Manager cache storage.Storage motd []byte motdMu sync.RWMutex rootCtx context.Context rootCancel context.CancelFunc store artifactstore.Store stores *artifactstore.Stores reader artifactstore.Reader qm *quota.Manager // set only when this spindle hosts the mill or joins one as an executor mill *mill.Mill exec executorClient}
func newCacheStore(ctx context.Context, cfg *config.Config) (storage.Storage, error) { switch cfg.Cache.Backend { case "": return nil, nil case "disk": dir := cfg.Cache.DiskDir if dir == "" { dir = filepath.Join(filepath.Dir(cfg.Server.DBPath), "cache") } return storage.NewDisk(dir) case "s3": return storage.NewUnversionedS3(ctx, cfg.Cache.S3Bucket, cfg.Cache.S3Prefix) default: return nil, fmt.Errorf("storage: unknown backend %q", cfg.Cache.Backend) }}
// New creates a new Spindle server with the provided configuration and engines.func New(ctx context.Context, cfg *config.Config, d *db.DB, engines map[string]models.Engine, qm *quota.Manager) (*Spindle, error) { logger := log.FromContext(ctx) metrics := observability.GetMetrics(ctx) if metrics == nil { metrics = observability.NewMetrics() ctx = observability.WithMetrics(ctx, metrics) } n := notifier.New() lifecycleCtx, cancelLifecycle := context.WithCancel(context.WithoutCancel(ctx)) keepLifecycle := false defer func() { if !keepLifecycle { cancelLifecycle() } }()
if cfg.Role == config.RoleExecutor { if err := cleanupOrphanRepos(ctx, d, logger); err != nil { return nil, fmt.Errorf("failed to run startup cleanup: %w", err) } } else if err := runStartupMigrations(ctx, d, cfg.Server.Tap.Embed, cfg.Server.Tap.DBPath, logger); err != nil { return nil, fmt.Errorf("failed to run startup migrations: %w", err) }
cacheStore, err := newCacheStore(ctx, cfg) if err != nil { return nil, fmt.Errorf("failed to setup cache storage: %w", err) } if cacheStore != nil { logger.Info("cache storage enabled", "backend", cfg.Cache.Backend, "storeID", cfg.Cache.StoreID) }
spindle := &Spindle{ db: d, l: logger, n: &n, wh: webhook.New(d, cfg.Server.Dev), metrics: metrics, engs: engines, cfg: cfg, cache: cacheStore, motd: defaultMotd, rootCtx: lifecycleCtx, rootCancel: cancelLifecycle, jobWake: make(chan struct{}, 1), qm: qm, } metrics.AttachDB(ctx, d) diskFallback := "" if cfg.Role == config.RoleStandalone { diskFallback = cfg.Server.LogDir if cfg.ArtifactStores.Disk.Dir == "" { logger.Warn("using SPINDLE_SERVER_LOG_DIR as the implicit disk artifact store; configure SPINDLE_ARTIFACT_STORES_DISK_DIR explicitly") } } stores, err := artifactstore.NewStores(cfg.ArtifactStores, diskFallback, cfg.LegacyS3.LogBucket) if err != nil { return nil, fmt.Errorf("failed to setup artifact stores: %w", err) } spindle.stores = stores if cfg.LegacyS3.LogBucket != "" { logger.Warn("SPINDLE_S3_LOG_BUCKET is deprecated; use SPINDLE_ARTIFACT_STORES_S3_BUCKET") } if cfg.Role != config.RoleExecutor { engine.StartCachePruner(ctx, logger, d, cacheStore, cfg.Cache.Retention, cfg.Cache.PruneInterval) } if cfg.Role == config.RoleStandalone { spindle.reader = stores } else { name := cfg.Mill.ArtifactStore if name == "" { names := stores.Names() if len(names) != 1 { return nil, fmt.Errorf("%s requires SPINDLE_MILL_ARTIFACT_STORE when %d artifact stores are configured", cfg.Role, len(names)) } name = names[0] logger.Warn("SPINDLE_MILL_ARTIFACT_STORE is not set; inferred the only configured store", "store", name) } store, ok := stores.Store(name) if !ok { return nil, fmt.Errorf("SPINDLE_MILL_ARTIFACT_STORE=%q is not configured", name) } spindle.store = store spindle.reader = store } if cfg.Role == config.RoleExecutor { keepLifecycle = true return spindle, nil }
e, err := rbac.NewEnforcer(cfg.Server.DBPath) if err != nil { return nil, fmt.Errorf("failed to setup rbac enforcer: %w", err) } e.E.EnableAutoSave(true) spindle.e = e
switch cfg.Server.Secrets.Provider { case "openbao": if cfg.Server.Secrets.OpenBao.ProxyAddr == "" { return nil, fmt.Errorf("openbao proxy address is required when using openbao secrets provider") } spindle.vault, err = secrets.NewOpenBaoManager( cfg.Server.Secrets.OpenBao.ProxyAddr, logger, secrets.WithMountPath(cfg.Server.Secrets.OpenBao.Mount), ) if err != nil { return nil, fmt.Errorf("failed to setup openbao secrets provider: %w", err) } logger.Info("using openbao secrets provider", "proxy_address", cfg.Server.Secrets.OpenBao.ProxyAddr, "mount", cfg.Server.Secrets.OpenBao.Mount) case "sqlite", "": spindle.vault, err = secrets.NewSQLiteManager(cfg.Server.DBPath, secrets.WithTableName("secrets")) if err != nil { return nil, fmt.Errorf("failed to setup sqlite secrets provider: %w", err) } logger.Info("using sqlite secrets provider", "path", cfg.Server.DBPath) default: return nil, fmt.Errorf("unknown secrets provider: %s", cfg.Server.Secrets.Provider) }
collections := []string{ tangled.SpindleMemberNSID, tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID, tangled.RepoPullStatusNSID, } jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, "spindle", collections, nil, log.SubLogger(logger, "jetstream"), d, true, true) if err != nil { return nil, fmt.Errorf("failed to setup jetstream client: %w", err) } spindle.jc = jc jc.AddDid(cfg.Server.Owner) // pull (status) records are created by arbitrary users too, same hack as in tap jc.ExemptCollection(tangled.RepoPullNSID) jc.ExemptCollection(tangled.RepoPullStatusNSID)
// Check if the spindle knows about any Dids; dids, err := d.GetAllDids() if err != nil { return nil, fmt.Errorf("failed to get all dids: %w", err) } for _, d := range dids { jc.AddDid(d) }
knownRepos, err := d.AllRepos() if err != nil { return nil, fmt.Errorf("failed to get known repos: %w", err) } for _, r := range knownRepos { if r.Owner != "" { jc.AddDid(r.Owner.String()) } }
spindle.res = idresolver.DefaultResolver(cfg.Server.PlcUrl) spindle.verify = repoverify.New(spindle.res, cfg.Server.Dev) spindle.scheduler = newPipelineScheduler(spindle)
err = e.AddSpindle(rbacDomain) if err != nil { return nil, fmt.Errorf("failed to set rbac domain: %w", err) } err = spindle.configureOwner() if err != nil { return nil, err } logger.Info("owner set", "did", cfg.Server.Owner)
cursorStore, err := cursor.NewSQLiteStore(cfg.Server.DBPath) if err != nil { return nil, fmt.Errorf("failed to setup sqlite3 cursor store: %w", err) }
err = jc.StartJetstream(lifecycleCtx, spindle.ingest()) if err != nil { return nil, fmt.Errorf("failed to start jetstream consumer: %w", err) }
// spindle listen to knot stream for sh.tangled.git.refUpdate // which will sync the local workflow files in spindle and enqueues the // pipeline job for on-push workflows ccfg := eventconsumer.NewConsumerConfig() ccfg.Logger = log.SubLogger(logger, "eventconsumer") ccfg.ProcessFunc = spindle.processKnotStream ccfg.CursorStore = cursorStore if cfg.Server.Dev { ccfg.RetryInterval = 5 * time.Second ccfg.MaxRetryInterval = 10 * time.Second } else { ccfg.RetryInterval = 1 * time.Minute ccfg.MaxRetryInterval = 10 * time.Minute } knownKnots, err := d.Knots() if err != nil { return nil, err } for _, knot := range knownKnots { logger.Info("adding source start", "knot", knot) src := eventconsumer.NewKnotSource(knot) eventconsumer.MigrateLegacyCursor(cursorStore, src) ccfg.Sources[src] = struct{}{} } spindle.ks = eventconsumer.NewConsumer(*ccfg) if cfg.Server.Tap.Embed { pw, err := randomAdminPassword() if err != nil { return nil, err } cfg.Server.Tap.AdminPassword = pw logger.Info("embedded tap: using random admin password") }
spindle.tap = NewTapClient(spindle)
keepLifecycle = true return spindle, nil}func (s *Spindle) DB() *db.DB { return s.db}
func (s *Spindle) Engines() map[string]models.Engine { return s.engs}
func (s *Spindle) Vault() secrets.Manager { return s.vault}
func (s *Spindle) Notifier() *notifier.Notifier { return s.n}
func (s *Spindle) Enforcer() *rbac.Enforcer { return s.e}
func (s *Spindle) SetMotdContent(content []byte) { s.motdMu.Lock() defer s.motdMu.Unlock() s.motd = content}
func (s *Spindle) GetMotdContent() []byte { s.motdMu.RLock() defer s.motdMu.RUnlock() return s.motd}
// runs the server. blocksfunc (s *Spindle) Start(ctx context.Context) error { runCtx := s.rootCtx cancelRun := s.rootCancel if runCtx == nil || cancelRun == nil { runCtx, cancelRun = context.WithCancel(context.WithoutCancel(ctx)) s.rootCtx = runCtx s.rootCancel = cancelRun } defer cancelRun()
_, err := observability.StartMetricsServer(runCtx, s.cfg.Server.MetricsListenAddr, s.l, s.metrics.Registry()) if err != nil { return fmt.Errorf("starting metrics listener: %w", err) } if s.scheduler != nil { s.scheduler.Start(runCtx) }
// an executor dials out to its mill and takes work from it var execDone chan struct{} if s.exec != nil { execDone = make(chan struct{}) go func() { defer close(execDone) s.exec.Connect(runCtx) }() } waitForExecutor := func(waitCtx context.Context) error { if execDone == nil { return nil } select { case <-execDone: return nil case <-waitCtx.Done(): return waitCtx.Err() } }
if s.mill != nil && s.cfg.Mill.JumpListenAddr != "" { go s.mill.ServeJump(runCtx, s.cfg.Mill.JumpListenAddr, s.cfg.Mill.JumpHostKeyPath, s.cfg.Mill.DebugExecutorPort, s.cfg.Mill.MaxJumpConnections) }
if stopper, ok := s.vault.(secrets.Stopper); ok { defer stopper.Stop() }
if s.cfg.Role != config.RoleExecutor { tapCtx, tapCancel := context.WithCancel(runCtx)
if s.cfg.Server.Tap.Embed { emb, err := startEmbeddedTap(tapCtx, s.cfg, log.SubLogger(s.l, "embedtap")) if err != nil { tapCancel() return fmt.Errorf("starting embedded tap: %w", err) } s.embedTap = emb defer func() { tapCancel() s.embedTap.Shutdown() }()
go s.watchTapDrain(tapCtx, tapCancel) } else { defer tapCancel() }
// tap must be ready before owner wipes start s.resumeWipes(ctx)
go func() { s.l.Info("starting knot event consumer") s.ks.Start(runCtx) }()
s.l.Info("starting tap client", "url", s.cfg.Server.Tap.Url) s.tap.Start(tapCtx) }
// start workers after resuming wipes so banned jobs cannot run workersCtx, cancelWorkers := context.WithCancel(runCtx) s.StartJobWorkers(workersCtx) defer func() { cancelWorkers() done := make(chan struct{}) go func() { s.jobWorkers.Wait() close(done) }() select { case <-done: case <-time.After(5 * time.Second): s.l.Warn("timed out waiting for job workers to stop") } }()
server := &http.Server{ Addr: s.cfg.Server.ListenAddr, Handler: s.Router(), } serverErr := make(chan error, 1) go func() { s.l.Info("starting spindle server", "address", s.cfg.Server.ListenAddr) serverErr <- server.ListenAndServe() }()
var listenErr error select { case listenErr = <-serverErr: case <-ctx.Done(): }
if s.exec != nil { drainCtx, cancelDrain := context.WithTimeout(context.Background(), s.cfg.Mill.DrainTimeout) if err := s.exec.Drain(drainCtx); err != nil { s.l.Warn("executor drain did not complete", "err", err) } cancelDrain() }
if s.cfg.Role != config.RoleExecutor { webhookCtx, cancelWebhooks := context.WithTimeout(context.Background(), serverShutdownTimeout) s.wh.Close(webhookCtx) cancelWebhooks() }
cancelRun() waitCtx, cancelWait := context.WithTimeout(context.Background(), executorShutdownTimeout) if err := waitForExecutor(waitCtx); err != nil { s.l.Warn("executor shutdown did not complete", "err", err) } cancelWait()
shutdownCtx, cancelShutdown := context.WithTimeout(context.Background(), serverShutdownTimeout) shutdownErr := server.Shutdown(shutdownCtx) cancelShutdown() if listenErr == nil { listenErr = <-serverErr } if errors.Is(listenErr, http.ErrServerClosed) { listenErr = nil } if shutdownErr != nil { shutdownErr = fmt.Errorf("shutting down spindle server: %w", shutdownErr) } return errors.Join(listenErr, shutdownErr)}
func (s *Spindle) declareTapInterest(ctx context.Context) { repos, reposErr := s.db.AllRepos() if reposErr != nil { s.l.Warn("tap declare: failed to load known repos", "err", reposErr) } members, membersErr := s.db.GetAllDids() if membersErr != nil { s.l.Warn("tap declare: failed to load known members", "err", membersErr) }
seen := make(map[syntax.DID]struct{}, len(repos)+len(members)+1) dids := make([]syntax.DID, 0, len(repos)+len(members)+1) add := func(did syntax.DID) { if did == "" { return } if _, ok := seen[did]; ok { return } seen[did] = struct{}{} dids = append(dids, did) }
owner, err := syntax.ParseDID(s.cfg.Server.Owner) if err != nil { s.l.Warn("tap declare: invalid configured owner", "owner", s.cfg.Server.Owner, "err", err) } else { add(owner) } for _, member := range members { did, err := syntax.ParseDID(member) if err != nil { s.l.Warn("tap declare: invalid known member", "did", member, "err", err) continue } add(did) } for _, repo := range repos { add(repo.Owner) } sort.Slice(dids, func(i, j int) bool { return dids[i] < dids[j] })
if err := s.tap.AddOwnerDIDs(ctx, dids); err != nil { s.l.Warn("tap declare: AddRepos rejected", "count", len(dids), "err", err) return } s.l.Info("tap declare: owner DIDs registered", "count", len(dids))}
func Run(ctx context.Context) error { cfg, err := config.Load(ctx) if err != nil { return fmt.Errorf("failed to load config: %w", err) }
shutdownTracing, err := observability.InitTracing(ctx, cfg.Tracing) if err != nil { return fmt.Errorf("failed to initialize tracing: %w", err) } defer func() { shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := shutdownTracing(shutdownCtx); err != nil { log.FromContext(ctx).Error("failed to shut down tracing", "err", err) } }()
otelHandler, shutdownLogging, err := observability.InitLogging(ctx, cfg.Logging) if err != nil { return fmt.Errorf("failed to initialize logging: %w", err) } defer func() { shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := shutdownLogging(shutdownCtx); err != nil { log.FromContext(ctx).Error("failed to shut down logging", "err", err) } }()
currentLogger := log.FromContext(ctx) combinedHandler := log.NewFanoutHandler(currentLogger.Handler(), otelHandler) combinedLogger := slog.New(combinedHandler) slog.SetDefault(combinedLogger) ctx = log.IntoContext(ctx, combinedLogger)
metrics := observability.NewMetrics() ctx = observability.WithMetrics(ctx, metrics)
if err := ensureGitVersion(); err != nil { return fmt.Errorf("ensuring git version: %w", err) }
d, err := db.Make(ctx, cfg.Server.DBPath) if err != nil { return fmt.Errorf("failed to setup db: %w", err) }
var qs *db.QuotaStore var qm *quota.Manager if cfg.Role != config.RoleExecutor { defaults, err := cfg.ToDefaults() if err != nil { return fmt.Errorf("invalid quota configuration: %w", err) } for scope, resources := range defaults { for resource, limit := range resources { metrics.SetQuotaDefaultLimit(string(scope), string(resource), limit) } } qs = db.NewQuotaStore(d, defaults) qm = quota.NewManager(qs, 50*time.Millisecond, metrics.QuotaObserver()) defer qm.Close()
if cfg.Role != config.RoleMill { if err := qs.Recover(ctx, nil); err != nil { return fmt.Errorf("failed to recover quota reservations: %w", err) } } }
logger := log.FromContext(ctx)
var engines map[string]models.Engine var m *mill.Mill var quotaClient *executor.QuotaClient
if cfg.Role == config.RoleMill { // on a mill host, engines place jobs on executors instead of running // them. all names share one Mill m = mill.New(log.SubLogger(logger, "mill"), mill.Config{ LogDir: cfg.Server.LogDir, MaxPending: cfg.Mill.MaxPending, ReconnectGrace: cfg.Mill.ReconnectGrace, CancelAckTimeout: cfg.Mill.CancelAckTimeout, CancelTeardownTimeout: cfg.Mill.CancelTeardownTimeout, CacheStoreID: cfg.Cache.StoreID, CacheMaxBytesPerOwner: cfg.Cache.MaxBytesPerOwner, }) engines = map[string]models.Engine{ "nixery": mill.NewEngine("nixery", m), "microvm": mill.NewEngine("microvm", m), "dummy": mill.NewEngine("dummy", m), } } else { // standalone and executor both run real engines locally nixeryEng, err := nixery.New(ctx, cfg) if err != nil { return err } var engineQStore quota.ReservationStore if cfg.Role == config.RoleExecutor { quotaClient = executor.NewQuotaClient() engineQStore = nil } else { engineQStore = qs } microvmEng, err := newMicrovmEngine(ctx, cfg, d, engineQStore) if err != nil { return err } engines = map[string]models.Engine{ "nixery": nixeryEng, "microvm": microvmEng, "dummy": dummy.New(logger), } }
s, err := New(ctx, cfg, d, engines, qm) if err != nil { return err }
if cfg.Role == config.RoleMill || cfg.Role == config.RoleStandalone { // flattens the store snapshot into bounded label tuples, the collector // drops anything outside the bounded scope, resource and status sets metrics.SetQuotaLoader(func() (observability.QuotaSnapshot, error) { snap, err := qs.MetricsSnapshot(context.Background()) if err != nil { return observability.QuotaSnapshot{}, err } var out observability.QuotaSnapshot for scope, resources := range snap.Usage { for resource, used := range resources { out.Usage = append(out.Usage, observability.QuotaUsage{ Scope: string(scope), Resource: resource, Used: used, }) } } for scope, resources := range snap.Subjects { for resource, statuses := range resources { for status, count := range statuses { out.Subjects = append(out.Subjects, observability.QuotaSubjectCount{ Scope: string(scope), Resource: string(resource), Status: status, Count: count, }) } } } return out, nil }) } if m != nil { // the engines built above hold the mill, but the mill's db and // notifier only exist after New, so attach them here m.Attach(s.DB(), s.Notifier(), qm) m.AttachCache(s.cache) s.mill = m s.mill.RegisterMetrics(s.metrics) if err := m.RestoreState(); err != nil { return fmt.Errorf("restoring mill state: %w", err) } liveIDs := m.LiveQuotaReservationIDs() for i := range liveIDs { liveIDs[i] = quota.StorageReservationID(liveIDs[i]) } if err := qs.Recover(ctx, liveIDs); err != nil { return fmt.Errorf("failed to recover mill quota reservations: %w", err) } } if cfg.Role == config.RoleExecutor { s.exec, err = executor.New(cfg, engines, s.DB(), s.Notifier(), log.SubLogger(logger, "executor"), s.store, s.cache) if err != nil { return err } s.exec.SetQuotaClient(quotaClient) s.exec.RegisterMetrics(s.metrics) }
return s.Start(ctx)}
func (s *Spindle) Router() http.Handler { mux := chi.NewRouter() mux.Use(observability.HTTPMiddleware(s.metrics)) if s.cfg.Tracing.Endpoint != "" { mux.Use(observability.OTelRouteMiddleware) }
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { w.Write(s.GetMotdContent()) }) mux.Get("/_health", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") _, _ = io.WriteString(w, "{\"status\":\"ok\"}\n") }) if s.cfg.Role != config.RoleExecutor { mux.HandleFunc("/events", s.Events) mux.HandleFunc("/logs/{knot}/{rkey}/{name}", s.Logs)
if s.mill != nil { mux.HandleFunc("/mill", s.mill.HandleExecutorConn) }
mux.Mount("/xrpc", s.XrpcRouter()) }
if s.cfg.Tracing.Endpoint == "" { return mux } return otelhttp.NewHandler( mux, "HTTP", otelhttp.WithSpanNameFormatter(func(string, *http.Request) string { return "HTTP" }), )}
func (s *Spindle) XrpcRouter() http.Handler { serviceAuth := serviceauth.NewServiceAuth(s.l, s.res.Directory(), s.cfg.Server.Did().String())
l := log.SubLogger(s.l, "xrpc")
x := xrpc.Xrpc{ Logger: l, Db: s.db, Enforcer: s.e, Engines: s.engs, Config: s.cfg, ArtifactReader: s.reader, Resolver: s.res, Vault: s.vault, Notifier: s.Notifier(), Webhooks: s.wh, ServiceAuth: serviceAuth, Trigger: s, QuotaStore: db.NewQuotaStore(s.db, quota.Defaults{}), Wiper: s, }
return x.Router()}
func (s *Spindle) processKnotStream(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error { ctx, span := observability.Tracer().Start(ctx, "knot.ingest") defer span.End()
metrics := s.metrics err := s.processKnotStreamInner(ctx, src, msg) if err != nil { metrics.RecordEventIngestion("knot", "error") span.SetStatus(codes.Error, "failed to process knot stream event") } else { metrics.RecordEventIngestion("knot", "success") span.SetStatus(codes.Ok, "success") } return err}
func (s *Spindle) processKnotStreamInner(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error { l := log.FromContext(ctx).With("handler", "processKnotStream") l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey) if msg.Nsid == knotdb.RepoCollaboratorUpdateNSID { return s.ingestKnotCollaborator(ctx, l, src, msg) } if msg.Nsid == tangled.GitRefUpdateNSID { event := tangled.GitRefUpdate{} if err := json.Unmarshal(msg.EventJson, &event); err != nil { l.Error("error unmarshalling", "err", err) return err } l = l.With("repo", event.Repo, "ref", event.Ref, "newSha", event.NewSha) l.Debug("debug")
repoDid := syntax.DID(event.Repo) repo, err := s.db.GetRepoByDid(repoDid) if err != nil { return fmt.Errorf("unknown repoDid %s: %w", repoDid, err) }
if ban, err := s.db.IsBanned(repoDid, repo.Owner); err != nil { return fmt.Errorf("checking bans: %w", err) } else if ban != nil { l.Warn("dropping push for banned repo") return nil }
if src.Host != repo.Knot { return fmt.Errorf("repo knot does not match event source: %s != %s", src.Host, repo.Knot) }
// deliver push webhooks independently of CI; a skip-ci push option // should not suppress webhook notifications s.wh.FirePush(ctx, repo, event.CommitterDid, event.Ref, event.OldSha, event.NewSha)
skipPipeline := kgit.HasSkipCIPushOption(event.PushOptions)
repoCloneUri := s.newRepoCloneUrl(src.Host, repoDid) repoPath := s.newRepoPath(repoDid) // the default-branch schedule refresh below syncs on demand, so a // skip-ci push needs no eager checkout if !skipPipeline { if err := gitutil.SparseSync(ctx, repoCloneUri, repoPath, event.NewSha, sparseWorkflowDir); err != nil { return fmt.Errorf("sync git repo: %w", err) } }
triggerRepo, err := s.buildTriggerRepo(ctx, repo) if err != nil { return fmt.Errorf("building trigger repo: %w", err) } if s.scheduler != nil && event.Ref == plumbing.NewBranchReferenceName(triggerRepo.DefaultBranch).String() { if err := s.scheduler.RefreshRepo(ctx, *repo); err != nil { l.Warn("failed to refresh schedules after default-branch push", "err", err) } } if skipPipeline { l.Info("push event requested ci skip, skipping pipeline") return nil }
trigger := tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindPush), Push: &tangled.Pipeline_PushTriggerData{ Ref: event.Ref, OldSha: event.OldSha, NewSha: event.NewSha, }, Repo: triggerRepo, }
pipelineId, err := s.runPipeline(ctx, repoDid, trigger, event.ChangedFiles, repoCloneUri, repoPath, event.NewSha, nil, triggerRepo) if err != nil { return err } if pipelineId.Rkey == "" { l.Info("no workflow matched 'push' trigger, skipping the event") return nil } l.Info("pipeline triggered", "pipeline", pipelineId.AtUri()) }
return nil}
func (s *Spindle) ingestKnotCollaborator(ctx context.Context, l *slog.Logger, src eventconsumer.Source, msg eventstream.Event) error { var rec knotdb.RepoCollaboratorUpdate if err := json.Unmarshal(msg.EventJson, &rec); err != nil { l.Error("error unmarshalling collaboratorUpdate", "err", err) return err }
subject, err := syntax.ParseDID(rec.Subject) if err != nil { l.Info("skipping collaboratorUpdate with malformed subject", "subject", rec.Subject, "err", err) return nil } repoDid, err := syntax.ParseDID(rec.Repo) if err != nil { l.Info("skipping collaboratorUpdate with malformed repo", "repo", rec.Repo, "err", err) return nil }
repo, err := s.db.GetRepoByDid(repoDid) if errors.Is(err, sql.ErrNoRows) { l.Info("skipping collaboratorUpdate for unknown repo", "repo", repoDid) return nil } if err != nil { return fmt.Errorf("lookup repo %s: %w", repoDid, err) } if src.Host != repo.Knot { l.Warn("dropping collaboratorUpdate from non-owning knot", "src", src.Host, "repoKnot", repo.Knot) return nil }
switch rec.Op { case knotdb.AclOpAdd: if err := s.e.AddCollaborator(subject.String(), rbac.ThisServer, repoDid.String()); err != nil { return fmt.Errorf("add collaborator policy: %w", err) } if err := s.db.AddKnotCollaborator(repoDid, subject); err != nil { return fmt.Errorf("track collaborator: %w", err) } l.Info("added knot-managed collaborator", "subject", subject, "repo", repoDid) case knotdb.AclOpRemove: if err := s.e.RemoveCollaborator(subject.String(), rbac.ThisServer, repoDid.String()); err != nil { return fmt.Errorf("remove collaborator policy: %w", err) } if err := s.db.DeleteRepoCollaboratorBySubjectRepo(subject, repoDid); err != nil { return fmt.Errorf("delete collaborator row: %w", err) } l.Info("removed knot-managed collaborator", "subject", subject, "repo", repoDid) default: return fmt.Errorf("collaboratorUpdate unknown op %q", rec.Op) } return nil}
// buildTriggerRepo gathers trigger metadata, resolving default branch from the knotfunc (s *Spindle) buildTriggerRepo(ctx context.Context, repo *db.Repo) (*tangled.Pipeline_TriggerRepo, error) { rkey := string(repo.Rkey) repoDid := repo.RepoDid.String() return s.buildTriggerRepoFrom(ctx, repo.Knot, repo.Owner.String(), rkey, repoDid), nil}
func (s *Spindle) buildTriggerRepoFrom(ctx context.Context, knot, did, rkey, repoDid string) *tangled.Pipeline_TriggerRepo { scheme := "https" if s.cfg.Server.Dev { scheme = "http" } client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, knot)}
// this should maybe (?) be in the refUpdate event itself to save a roundtrip defaultBranch := "" if out, err := tangled.RepoGetDefaultBranch(ctx, client, repoDid); err == nil { defaultBranch = out.Name }
var rkeyPtr *string if rkey != "" { rkeyPtr = &rkey } return &tangled.Pipeline_TriggerRepo{ Did: did, Knot: knot, Repo: rkeyPtr, RepoDid: &repoDid, DefaultBranch: defaultBranch, }}
func (s *Spindle) resolvePipelineSourceRepo(ctx context.Context, trigger *tangled.Pipeline_TriggerMetadata) (*tangled.Pipeline_TriggerRepo, error) { if trigger == nil { return nil, nil } if trigger.SourceRepo == nil || *trigger.SourceRepo == "" { return trigger.Repo, nil } repoDid, err := syntax.ParseDID(*trigger.SourceRepo) if err != nil { return nil, fmt.Errorf("parse sourceRepo %s: %w", *trigger.SourceRepo, err) } return s.resolveSourceRepoInfo(ctx, repoDid)}
func (s *Spindle) resolveSourceRepoInfo(ctx context.Context, repoDid syntax.DID) (*tangled.Pipeline_TriggerRepo, error) { repo, err := s.db.GetRepoByDid(repoDid) if err == nil { return s.buildTriggerRepo(ctx, repo) }
// verify repo, we don't want git sync to point to arbitrary endpoints res, err := s.verify(ctx, repoident.RepoDid(repoDid)) if err != nil { return nil, fmt.Errorf("verify sourceRepo %s: %w", repoDid, err) } ownership, ok := res.Ownership() if !ok { return nil, fmt.Errorf("verify sourceRepo %s: knot %s answered %s", repoDid, res.KnotURL, res.Answer()) } return s.buildTriggerRepoFrom(ctx, res.KnotURL.Host(), ownership.OwnerDid.String(), ownership.Rkey.String(), repoDid.String()), nil}
func (s *Spindle) runPipeline(ctx context.Context, repoDid syntax.DID, trigger tangled.Pipeline_TriggerMetadata, changedFiles []string, repoCloneUri, repoPath, rev string, only []string, sourceRepo *tangled.Pipeline_TriggerRepo) (models.PipelineId, error) { l := log.FromContext(ctx)
compiler := workflow.Compiler{ ChangedFiles: changedFiles, Trigger: trigger, }
rawPipeline, err := s.loadPipeline(ctx, repoCloneUri, repoPath, rev) if err != nil { return models.PipelineId{}, fmt.Errorf("loading pipeline: %w", err) } if len(rawPipeline) == 0 { return models.PipelineId{}, nil }
tpl := compiler.Compile(compiler.Parse(rawPipeline)) // todo(dawn): pass compile error to workflow log for _, w := range compiler.Diagnostics.Errors { l.Error(w.String()) } for _, w := range compiler.Diagnostics.Warnings { l.Warn(w.String()) }
if len(only) > 0 { tpl.Workflows = filterWorkflows(tpl.Workflows, only) } if len(tpl.Workflows) == 0 { return models.PipelineId{}, nil }
pipelineId := models.PipelineId{ Knot: trigger.Repo.Knot, Rkey: tid.TID(), } if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { return models.PipelineId{}, fmt.Errorf("creating pipeline event: %w", err) } err = s.processPipeline(ctx, repoDid, tpl, pipelineId, sourceRepo) return pipelineId, err}
func filterWorkflows(workflows []*tangled.Pipeline_Workflow, only []string) []*tangled.Pipeline_Workflow { allowed := make(map[string]struct{}, len(only)) for _, n := range only { allowed[n] = struct{}{} } var filtered []*tangled.Pipeline_Workflow for _, w := range workflows { if w == nil { continue } if _, ok := allowed[w.Name]; ok { filtered = append(filtered, w) } } return filtered}
// TriggerManual dispatches a pipeline at sha, authorized against and recorded// under repoDid. sourceRepo, pull, and inputs are optional trigger payload.func (s *Spindle) TriggerManual(ctx context.Context, repoDid syntax.DID, sha, ref string, workflows []string, sourceRepo syntax.DID, pull xrpc.PullContext, inputs []*tangled.Pipeline_Pair) (syntax.ATURI, error) { repo, err := s.db.GetRepoByDid(repoDid) if err != nil { return "", fmt.Errorf("unknown repoDid %s: %w", repoDid, err) }
if ban, err := s.db.IsBanned(repoDid, repo.Owner); err != nil { return "", fmt.Errorf("checking bans: %w", err) } else if ban != nil { return "", fmt.Errorf("repo %s is banned", repoDid) }
triggerRepo, err := s.buildTriggerRepo(ctx, repo) if err != nil { return "", fmt.Errorf("building trigger repo: %w", err) }
trigger := tangled.Pipeline_TriggerMetadata{Repo: triggerRepo} if pull.IsPullRequest { var pullAt *string if pull.Pull != "" { pullAtStr := pull.Pull.String() pullAt = &pullAtStr } trigger.Kind = string(workflow.TriggerKindPullRequest) trigger.PullRequest = &tangled.Pipeline_PullRequestTriggerData{ SourceBranch: pull.SourceBranch, TargetBranch: pull.TargetBranch, SourceSha: sha, Pull: pullAt, } } else { var refPtr *string if ref != "" { refPtr = &ref } trigger.Kind = string(workflow.TriggerKindManual) trigger.Manual = &tangled.Pipeline_ManualTriggerData{ Sha: sha, Ref: refPtr, Inputs: inputs, } }
repoCloneUri, repoPath, sourceInfo, err := s.resolveCheckout(ctx, repoDid, sourceRepo) if err != nil { return "", err } if sourceInfo == nil { sourceInfo = triggerRepo } else { // reject a banned source before running its workflow sourceOwner, _ := syntax.ParseDID(sourceInfo.Did) if ban, err := s.db.IsBanned(sourceRepo, sourceOwner); err != nil { return "", fmt.Errorf("checking bans: %w", err) } else if ban != nil { return "", fmt.Errorf("source repo %s is banned", sourceRepo) } sourceRepoStr := sourceRepo.String() trigger.SourceRepo = &sourceRepoStr }
pipelineId, err := s.runPipeline(ctx, repoDid, trigger, nil, repoCloneUri, repoPath, sha, workflows, sourceInfo) if err != nil { return "", err } if pipelineId.Rkey == "" { return "", xrpc.ErrNoMatchingWorkflows } return pipelineId.AtUri(), nil}
// sourceInfo is nil when the checkout comes from the target repo.func (s *Spindle) resolveCheckout(ctx context.Context, repoDid syntax.DID, sourceRepo syntax.DID) (cloneUri, repoPath string, sourceInfo *tangled.Pipeline_TriggerRepo, err error) { repo, err := s.db.GetRepoByDid(repoDid) if err != nil { return "", "", nil, fmt.Errorf("unknown repoDid %s: %w", repoDid, err) }
cloneUri = s.newRepoCloneUrl(repo.Knot, repoDid) repoPath = s.newRepoPath(repoDid) if sourceRepo != "" && sourceRepo != repoDid { sourceInfo, err = s.resolveSourceRepoInfo(ctx, sourceRepo) if err != nil { return "", "", nil, err } cloneUri = models.BuildRepoURL(sourceInfo) repoPath = s.newRepoPath(sourceRepo) } return cloneUri, repoPath, sourceInfo, nil}
// resolves the workflow definition at sha without executing it// returns a deterministic fingerprint over the resolved files.func (s *Spindle) DescribeWorkflowDefinition(ctx context.Context, repoDid syntax.DID, sha string, sourceRepo syntax.DID) (*tangled.CiDescribeWorkflowDefinition_Output, error) { repoCloneUri, repoPath, _, err := s.resolveCheckout(ctx, repoDid, sourceRepo) if err != nil { return nil, err }
rawPipeline, err := s.loadPipeline(ctx, repoCloneUri, repoPath, sha) if err != nil { return nil, fmt.Errorf("loading pipeline: %w", err) }
hash := fingerprintWorkflowDefinition(rawPipeline) workflows := make([]string, 0, len(rawPipeline)) for _, w := range rawPipeline { workflows = append(workflows, w.Name) }
return &tangled.CiDescribeWorkflowDefinition_Output{ Derived: true, Hash: &hash, Workflows: workflows, }, nil}
func fingerprintWorkflowDefinition(rawPipeline workflow.RawPipeline) string { sorted := make([]workflow.RawWorkflow, len(rawPipeline)) copy(sorted, rawPipeline) sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name })
h := sha256.New() var lenBuf [8]byte for _, w := range sorted { binary.LittleEndian.PutUint64(lenBuf[:], uint64(len(w.Contents))) h.Write([]byte(w.Name)) // terminate name to avoid ["fo", "o"] == ["f", "oo"] h.Write([]byte{0}) h.Write(lenBuf[:]) h.Write(w.Contents) } return fmt.Sprintf("sha256:%x", h.Sum(nil))}
func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev string) (workflow.RawPipeline, error) { if err := gitutil.SparseSync(ctx, repoUri, repoPath, rev, sparseWorkflowDir); err != nil { return nil, fmt.Errorf("syncing git repo: %w", err) } gr, err := kgit.Open(repoPath, rev) if err != nil { return nil, fmt.Errorf("opening git repo: %w", err) }
workflowDir, err := gr.FileTree(ctx, workflow.WorkflowDir) if errors.Is(err, object.ErrDirectoryNotFound) { // return empty RawPipeline when directory doesn't exist return nil, nil } else if err != nil { return nil, fmt.Errorf("loading file tree: %w", err) }
var rawPipeline workflow.RawPipeline for _, e := range workflowDir { if !e.IsFile() { continue }
fpath := filepath.Join(workflow.WorkflowDir, e.Name) contents, err := gr.RawContent(fpath) if err != nil { return nil, fmt.Errorf("reading raw content of '%s': %w", fpath, err) }
rawPipeline = append(rawPipeline, workflow.RawWorkflow{ Name: e.Name, Contents: contents, }) }
return rawPipeline, nil}
func (s *Spindle) newRepoPath(repo syntax.DID) string { return filepath.Join(s.cfg.Server.RepoDir, repo.String())}
func (s *Spindle) newRepoCloneUrl(knot string, did syntax.DID) string { scheme := "https://" if s.cfg.Server.Dev { scheme = "http://" } return fmt.Sprintf("%s%s/%s", scheme, knot, did)}
const RequiredVersion = "2.49.0"
func ensureGitVersion() error { v, err := gitutil.Version() if err != nil { return fmt.Errorf("fetching git version: %w", err) } if v.LessThan(version.Must(version.NewVersion(RequiredVersion))) { return fmt.Errorf("installed git version %q is not supported, Spindle requires git version >= %q", v, RequiredVersion) } return nil}
func (s *Spindle) configureOwner() error { cfgOwner := s.cfg.Server.Owner
existing, err := s.e.GetSpindleUsersByRole("server:owner", rbacDomain) if err != nil { return err }
switch len(existing) { case 0: // no owner configured, continue case 1: // find existing owner existingOwner := existing[0]
// no ownership change, this is okay if existingOwner == s.cfg.Server.Owner { break }
// remove existing owner err = s.e.RemoveSpindleOwner(rbacDomain, existingOwner) if err != nil { return nil } default: return fmt.Errorf("more than one owner in DB, try deleting %q and starting over", s.cfg.Server.DBPath) }
return s.e.AddSpindleOwner(rbacDomain, cfgOwner)}
func (s *Spindle) StartJobWorkers(ctx context.Context) { for range s.cfg.Server.MaxJobCount { s.jobWorkers.Add(1) go func() { defer s.jobWorkers.Done() for { job, err := s.db.DequeueJob(ctx) if err != nil { s.l.Error("failed to dequeue job", "error", err) } if job == nil { // sleep until a new job wakes us select { case <-ctx.Done(): return case <-s.jobWake: } continue } s.metrics.RecordJobQueueActivity("dequeue") s.runJob(ctx, job) } }() }}
func (s *Spindle) runJob(ctx context.Context, job *db.JobRow) { pipelineId := models.PipelineId{ Knot: job.PipelineIdKnot, Rkey: job.PipelineIdRkey, } repoDID := job.RepoDid if job.SourceRepo != nil && job.SourceRepo.RepoDid != nil && *job.SourceRepo.RepoDid != "" { repoDID = *job.SourceRepo.RepoDid }
jobCtx := observability.ExtractFromTraceparentAndTracestate(ctx, job.Traceparent, job.Tracestate) if job.CreatedAtNs > 0 { queueDelay := time.Since(time.Unix(0, job.CreatedAtNs)) if queueDelay < 0 { queueDelay = 0 } s.metrics.RecordJobDequeueLatency(jobCtx, queueDelay) }
jobCtx, span := observability.Tracer().Start(jobCtx, "job.run") if span.IsRecording() { attrs := []attribute.KeyValue{ attribute.Int64(observability.JobIDKey, job.Id), attribute.String(observability.PipelineIDKey, pipelineId.AtUri().String()), } if repoDID != "" { attrs = append(attrs, attribute.String(observability.RepoDIDKey, repoDID)) } if job.RepoDid != "" && job.RepoDid != repoDID { attrs = append(attrs, attribute.String(observability.TargetRepoDIDKey, job.RepoDid)) } span.SetAttributes(attrs...) } failed := false defer func() { if failed { span.SetStatus(codes.Error, "job failed") } else { span.SetStatus(codes.Ok, "success") } span.End() }()
l := log.SubLogger(log.SubLogger(s.l, "job"), "engine").With( "job_id", job.Id, ) if job.RepoDid != "" && job.RepoDid != repoDID { l = l.With("target_repo_did", job.RepoDid) }
pipelineEnv := models.PipelineEnvVarsForSource(job.Tpl.TriggerMetadata, pipelineId, job.SourceRepo)
initTpl := job.Tpl if job.SourceRepo != nil && job.Tpl.TriggerMetadata != nil { tm := *job.Tpl.TriggerMetadata tm.Repo = job.SourceRepo initTpl.TriggerMetadata = &tm } trustedSource := models.TrustedPipelineSource(initTpl.TriggerMetadata, job.RepoDid)
workflows := make(map[models.Engine][]models.Workflow) for _, w := range job.Tpl.Workflows { if w == nil { continue } eng, ok := s.engs[w.Engine] if !ok { failed = true _ = s.db.StatusFailed(models.WorkflowId{ PipelineId: pipelineId, Name: w.Name, }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) s.metrics.RecordWorkflowTerminal( fmt.Sprint(w.Engine), "failure", string(engine.FailureClassUser), string(engine.FailureReasonWorkflowInvalid), ) continue }
ewf, err := eng.InitWorkflow(*w, initTpl) if err != nil { failed = true _ = s.db.StatusFailed(models.WorkflowId{ PipelineId: pipelineId, Name: w.Name, }, fmt.Sprintf("init workflow: %s", err), -1, s.n) failureClass, failureReason := engine.FailureAttribution("failure", err) s.metrics.RecordWorkflowTerminal( fmt.Sprint(w.Engine), "failure", failureClass, failureReason, ) continue } ewf.RunID = fmt.Sprintf("%d", job.Id)
if ewf.Environment == nil { ewf.Environment = make(map[string]string) } maps.Copy(ewf.Environment, pipelineEnv)
ewf.Engine = w.Engine ewf.RepoDID = job.RepoDid if !trustedSource { ewf.Caches = nil } workflows[eng] = append(workflows[eng], *ewf) }
engine.StartWorkflows(l, s.vault, s.cfg, s.qm, s.stores, s.db, s.n, s.cache, nil, jobCtx, &models.Pipeline{ RepoDid: syntax.DID(job.RepoDid), Workflows: workflows, TrustedSource: trustedSource, TriggerMetadata: job.Tpl.TriggerMetadata, }, pipelineId)}
func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId, sourceRepo *tangled.Pipeline_TriggerRepo) error { traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx) if err := s.db.EnqueueJobWithPending( s.rootCtx, repoDid.String(), pipelineId, sourceRepo, tpl, traceparent, tracestate, s.n, ); err != nil { return fmt.Errorf("failed to enqueue durable job: %w", err) } s.l.Info("pipeline enqueued successfully to db", "id", pipelineId)
select { case s.jobWake <- struct{}{}: default: }
return nil}