pipeline

package
v0.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 21, 2026 License: Apache-2.0 Imports: 47 Imported by: 0

Documentation

Overview

Package pipeline is a user-defined CI/CD layer on top of the existing build and deploy primitives. A pipeline is a YAML file describing triggers, jobs, and steps; the Engine drives runs level-triggered from persisted state and never touches the reconciler directly.

Index

Constants

View Source
const (
	KindRun              = "run"
	KindTest             = "test"
	KindBuild            = "build"
	KindDeploy           = "deploy"
	KindPromote          = "promote"
	KindRollback         = "rollback"
	KindNotify           = "notify"
	KindApproval         = "approval"
	KindArtifactUpload   = "artifact-upload"
	KindArtifactDownload = "artifact-download"
	TemplatePrefix       = "template/"
)

Step kinds, the values of a step's `uses` field. A step with a `run` script is kind "run".

View Source
const (
	TriggerPush        = "push"
	TriggerPullRequest = "pull_request"
	TriggerTag         = "tag"
	TriggerMergeGroup  = "merge_group"
	TriggerManual      = "manual"
	TriggerSchedule    = "schedule"
	TriggerAPI         = "api"
	TriggerWebhook     = "webhook"
)

Trigger kinds recorded on a run.

View Source
const (
	ForksBlock   = "block"
	ForksApprove = "approve"
	ForksAllow   = "allow"
)

Fork policies for pull requests whose source is another repository.

View Source
const (
	FixRerun        = "rerun"
	FixAddRetries   = "add_retries"
	FixRaiseTimeout = "raise_timeout"
	FixRaiseMemory  = "raise_memory"
	FixEditPipeline = "edit_pipeline"
	FixSetSecret    = "set_secret"
	FixFreeDisk     = "free_disk"
)

Fix kinds. Auto kinds can be applied by the control plane, guided kinds point the operator at the one place the fix lives.

View Source
const (
	KindSetup           = "setup"
	KindArtifactRestore = "artifact-restore"
	KindArtifactCollect = "artifact-collect"
)

Implicit step kinds. Setup prepares the workspace and container, restore unpacks needed jobs' artifacts, and collect stores this job's own.

View Source
const (
	ParamString  = "string"
	ParamChoice  = "choice"
	ParamBoolean = "boolean"
)

Parameter types.

View Source
const (
	ReportPending = "pending"
	ReportSuccess = "success"
	ReportFailure = "failure"
	ReportError   = "error"
)

Commit status states a run is reported as.

View Source
const (
	SyncCreated   = "created"
	SyncUpdated   = "updated"
	SyncUnchanged = "unchanged"
	// SyncDiverged means the stored definition was edited after the last
	// sync (or was never synced) and repo_is_truth is off, so it was kept.
	SyncDiverged = "diverged"
	SyncInvalid  = "invalid"
	SyncRefused  = "refused"

	// SourceRepo marks a pipeline whose definition comes from the repository.
	SourceRepo = "repo"
)

Sync outcomes reported per pipeline file.

View Source
const FixEditSource = "edit_source"

FixEditSource is a guided fix that lives in the app's source or build settings.

View Source
const MaxMatrixCombinations = 256

MaxMatrixCombinations bounds one job's expansion.

View Source
const SchemaVersion = 1

SchemaVersion is the only pipeline file version this package accepts.

Variables

View Source
var (
	ErrArtifactTooLarge = errors.New("pipeline: artifact exceeds the size limit")
	ErrArtifactCorrupt  = errors.New("pipeline: artifact failed its sha256 check")
	ErrArtifactPath     = errors.New("pipeline: artifact path is outside the store")
)

Artifact errors callers can match.

View Source
var ErrInvalidInput = errors.New("pipeline: invalid input")

ErrInvalidInput marks a run request the pipeline's declared inputs reject.

View Source
var ErrNoPathTrigger = errors.New("pipeline: path filters need an on.push or on.pull_request trigger")

ErrNoPathTrigger is returned by ApplyFilters when the pipeline has neither a push nor a pull_request trigger to attach path filters to.

View Source
var ErrNoRepo = errors.New("pipeline: app has no connected repository")

ErrNoRepo is returned by Source when an app has no connected repository.

View Source
var ErrNoReportTarget = errors.New("pipeline: no git forge to report to")

ErrNoReportTarget is returned by a StatusReporter when the run's app has no git forge to report to. The engine treats it as "nothing to do".

View Source
var ErrSHAUnavailable = errors.New("pipeline sync: commit unavailable")

ErrSHAUnavailable means the requested commit could not be fetched, so the definitions at that commit are unknown.

View Source
var PRTypes = []string{prActionOpened, prActionReopened, prActionSynchronize}

PRTypes lists the values on.pull_request.types accepts.

Functions

func ApplyFilters

func ApplyFilters(data []byte, paths, pathsIgnore []string, reportStatus *bool) ([]byte, error)

ApplyFilters rewrites a pipeline's YAML so its push and pull_request triggers carry paths and paths_ignore, and sets report_status. A nil list or flag leaves that setting as it is; an empty non-nil list removes it.

func ApplyFix

func ApplyFix(src []byte, job, kind string) ([]byte, error)

ApplyFix rewrites a pipeline file to apply an auto fix to one job. It keeps comments and ordering, and refuses kinds that are not auto-applicable.

func ComboKey

func ComboKey(job string, c Combo) string

ComboKey is the stable job key for a matrix instance, e.g. "test[go=1.22]".

func Discover

func Discover(repoDir, brandedName string) ([]string, error)

Discover returns the pipeline files under repoDir (*.yaml and *.yml in the first candidate directory that exists), sorted by name.

func DiscoverDirs

func DiscoverDirs(brandedName string) []string

DiscoverDirs returns the candidate pipeline directories relative to a repository root: a generic name first, then the branded one, so a product rename never strands existing repositories. brandedName is the caller's brand short name; pass "" to skip it.

func Diverged

func Diverged(p store.Pipeline) bool

Diverged reports whether a repo-sourced definition was edited after its last sync.

func EvalCondition

func EvalCondition(expr string, sc Scope) (bool, error)

EvalCondition evaluates an `if:` expression. An empty expression is true only when the scope is in a success state.

func HashYAML

func HashYAML(y string) string

HashYAML is the fingerprint stored as a definition's synced hash.

func Interpolate

func Interpolate(s string, sc Scope) (string, error)

Interpolate replaces every ${{ path }} in s. A missing "secrets." path is an error so a typo never silently runs a step without its credential; any other missing path resolves to the empty string.

func InterpolateScript

func InterpolateScript(s string, sc Scope) (body string, env map[string]string, err error)

InterpolateScript is Interpolate for shell script text. Values that could be attacker-controlled are not spliced into the script: each is replaced by a ${PIPELINE_EXPR_*} reference and returned in env, which the caller exports single-quoted, so the shell never parses them as code.

func ResolveParameters

func ResolveParameters(def *Definition, given map[string]string, maxValueBytes int) (map[string]string, error)

ResolveParameters validates the supplied values against the declared parameters and returns the values the run records, defaults filled in.

func SchemaJSON

func SchemaJSON() []byte

SchemaJSON returns the JSON Schema pipeline files are validated against.

func TargetApps

func TargetApps(def *Definition, ownApp string) []string

TargetApps returns the literal app or service names the definition's deploy, promote, rollback, and notify steps act on, excluding the pipeline's own app. Validation guarantees these are literals.

func Validate

func Validate(data []byte) (*Definition, []Issue)

Validate parses data, checks it against the JSON Schema, then runs the semantic checks (needs graph, conditions, step arguments, cron expressions). The definition is non-nil only when there are no issues.

func ValidateCondition

func ValidateCondition(expr string) error

ValidateCondition checks that expr parses, without evaluating it.

Types

type Actions

type Actions interface {
	Build(ctx context.Context, req BuildRequest, log func(line string)) (image string, err error)
	Deploy(ctx context.Context, req DeployRequest, log func(line string)) error
	Promote(ctx context.Context, from, to string, log func(line string)) (image string, err error)
	Rollback(ctx context.Context, service string, log func(line string)) (image string, err error)
	Notify(ctx context.Context, app string, succeeded bool, message string) error
}

Actions are the control-plane primitives non-container steps call. They wrap the existing build and deploy code; the Engine never reconciles.

type ArtifactIndex

type ArtifactIndex interface {
	SavePipelineArtifact(ctx context.Context, a store.PipelineArtifact) error
	ListPipelineArtifacts(ctx context.Context, runID string) ([]store.PipelineArtifact, error)
	ListExpiredPipelineArtifacts(ctx context.Context, now time.Time) ([]store.PipelineArtifact, error)
	DeletePipelineArtifact(ctx context.Context, id string) error
}

ArtifactIndex is the metadata persistence the artifact store needs. *store.DB satisfies it.

type ArtifactLimits

type ArtifactLimits struct {
	// MaxBytes caps one stored archive. MaxUnpackedBytes caps what it
	// expands to, so a small archive cannot fill the restore workspace.
	MaxBytes         int64
	MaxUnpackedBytes int64
	// MaxEntries caps the files, directories and links in one archive.
	MaxEntries int
	// MaxRunBytes caps all archives of one run together.
	MaxRunBytes int64
	Retention   time.Duration
}

ArtifactLimits are the artifact size and retention caps, set from the APP_PIPELINE_ARTIFACT_* environment variables by the caller.

type ArtifactStore

type ArtifactStore struct {
	// contains filtered or unexported fields
}

ArtifactStore keeps artifact archives on local disk under one directory, one subdirectory per run. File names derive from IDs, never from user input.

func NewArtifactStore

func NewArtifactStore(dir string, lim ArtifactLimits) (*ArtifactStore, error)

NewArtifactStore creates dir if needed.

func (*ArtifactStore) Begin

func (s *ArtifactStore) Begin(runID, id string, runUsed int64) (*ArtifactWriter, error)

Begin opens a writer for a job's archive. runUsed is the bytes the run's earlier archives already hold.

func (*ArtifactStore) Limits

func (s *ArtifactStore) Limits() ArtifactLimits

Limits returns the configured caps.

func (*ArtifactStore) Open

func (s *ArtifactStore) Open(meta store.PipelineArtifact) (*os.File, error)

Open verifies the stored archive against meta.SHA256 and returns it positioned at the start. A mismatch returns ErrArtifactCorrupt.

func (*ArtifactStore) OpenVerified

func (s *ArtifactStore) OpenVerified(meta store.PipelineArtifact) (*VerifiedReader, error)

OpenVerified opens meta's archive for streaming with the sha256 checked as the bytes are read, not before.

func (*ArtifactStore) Remove

func (s *ArtifactStore) Remove(rel string) error

Remove deletes a stored archive; a missing file is not an error.

func (*ArtifactStore) Sweep

func (s *ArtifactStore) Sweep(ctx context.Context, idx ArtifactIndex, runExists func(ctx context.Context, runID string) bool) (int, error)

Sweep deletes archives past retention and run directories whose run no longer exists. It returns how many archives it removed.

type ArtifactWriter

type ArtifactWriter struct {
	// contains filtered or unexported fields
}

ArtifactWriter receives one archive's bytes, enforcing the size cap and hashing as it goes. Abort or Commit must be called exactly once.

func (*ArtifactWriter) Abort

func (a *ArtifactWriter) Abort()

Abort discards the partial file.

func (*ArtifactWriter) Commit

func (a *ArtifactWriter) Commit() (rel string, size int64, sum string, err error)

Commit closes the file and moves it into place, returning its relative path, size, and sha256.

func (*ArtifactWriter) Size

func (a *ArtifactWriter) Size() int64

Size is the bytes written so far.

func (*ArtifactWriter) Write

func (a *ArtifactWriter) Write(p []byte) (int, error)

Write stores p, failing with ErrArtifactTooLarge once the cap is passed.

type BuildRequest

type BuildRequest struct {
	App        string
	Ref        string
	SHA        string
	Type       string
	Context    string
	Dockerfile string
	Image      string
	Tag        string
}

BuildRequest is one image build for the build step.

type CacheSpec

type CacheSpec struct {
	Key  string `yaml:"key" json:"key"`
	Path string `yaml:"path" json:"path"`
}

CacheSpec mounts a persistent named volume at Path. Runs sharing a Key share the volume.

type CatalogTemplate

type CatalogTemplate struct {
	ID          string   `json:"id"`
	Name        string   `json:"name"`
	Language    string   `json:"language"`
	Description string   `json:"description"`
	Tags        []string `json:"tags"`
	YAML        string   `json:"yaml,omitempty"`
}

CatalogTemplate is a ready-to-save starter pipeline file.

func Catalog

func Catalog() []CatalogTemplate

Catalog returns the built-in starter pipelines sorted by language then name.

func CatalogByID

func CatalogByID(id string) (CatalogTemplate, error)

CatalogByID looks up one starter pipeline.

type Combo

type Combo map[string]string

Combo is one matrix combination.

type Concurrency

type Concurrency struct {
	Group            string `yaml:"group" json:"group"`
	CancelInProgress bool   `yaml:"cancel_in_progress,omitempty" json:"cancel_in_progress,omitempty"`
}

Concurrency serializes runs sharing a group; CancelInProgress cancels the older active runs in the group instead of queueing behind them.

type Config

type Config struct {
	Store   Store
	Runtime RuntimeResolver
	Actions Actions
	Secrets SecretResolver
	Source  Source
	// RunEnv returns extra environment variables for a run's steps, for
	// example the preview environment a pull request run belongs to. Optional.
	RunEnv func(ctx context.Context, run store.PipelineRun) map[string]string
	// OIDCIssuer mints a job's oidc token; nil means opting a job into
	// oidc fails with a clear error rather than silently getting none.
	OIDCIssuer OIDCIssuer
	// OIDCRequestURL is the base URL for the runtime "request a token
	// for this audience" endpoint (Engine.OIDCTokenRequestHandler),
	// reachable from inside a job's own container, never externally.
	// Empty means a job that lists more than one oidc.audiences entry
	// fails with a clear error instead of getting an unreachable URL.
	OIDCRequestURL string
	Logger         *slog.Logger
	Now            func() time.Time
	NewID          func() string

	// NamePrefix namespaces containers and volumes (the brand short name).
	NamePrefix string
	// GitImage runs the checkout; it needs a POSIX shell and git.
	GitImage        string
	MaxParallelJobs int
	MaxLogLines     int
	StepTimeout     time.Duration
	JobTimeout      time.Duration
	ApprovalTimeout time.Duration
	KeepRuns        int
	// MaxInputBytes caps one run parameter value.
	MaxInputBytes int

	// Artifacts stores job artifact archives on disk. Nil means a job that
	// declares artifacts fails with a clear error instead of dropping them.
	Artifacts *ArtifactStore

	// Reporter, when set, posts run state to the git forge as a commit
	// status. ReportTimeout bounds each post.
	Reporter      StatusReporter
	ReportTimeout time.Duration
}

Config wires an Engine. Zero limits fall back to the defaults below; cmd wiring overrides them from environment variables.

type Definition

type Definition struct {
	Version int      `yaml:"version" json:"version"`
	Name    string   `yaml:"name" json:"name"`
	On      Triggers `yaml:"on" json:"on"`
	Stages  []string `yaml:"stages,omitempty" json:"stages,omitempty"`
	// Parameters declares the values a manual run, API call, or webhook may set.
	Parameters  []Parameter       `yaml:"parameters,omitempty" json:"parameters,omitempty"`
	Env         map[string]string `yaml:"env,omitempty" json:"env,omitempty"`
	Concurrency *Concurrency      `yaml:"concurrency,omitempty" json:"concurrency,omitempty"`
	// ReportStatus posts run state back to the forge as a commit status.
	// Nil means on, when the app has provider credentials.
	ReportStatus *bool               `yaml:"report_status,omitempty" json:"report_status,omitempty"`
	Timeout      Duration            `yaml:"timeout,omitempty" json:"timeout,omitempty"`
	Templates    map[string]Template `yaml:"templates,omitempty" json:"templates,omitempty"`
	Jobs         map[string]*Job     `yaml:"jobs" json:"jobs"`
	// JobOrder is the order jobs appear in the file, kept for stable display.
	JobOrder []string `yaml:"-" json:"job_order"`
}

Definition is a parsed pipeline file.

func Parse

func Parse(data []byte) (*Definition, error)

Parse decodes YAML into a Definition and expands step templates. It does not validate; call Validate for structural and semantic checks.

func (*Definition) DeclaredParameters

func (d *Definition) DeclaredParameters() []Parameter

DeclaredParameters lists every parameter the pipeline declares: the top-level `parameters` list plus the older `on.manual.inputs` map, by name.

func (*Definition) ReportsStatus

func (d *Definition) ReportsStatus() bool

ReportsStatus reports whether run state is posted back to the forge.

type DeployRequest

type DeployRequest struct {
	Service  string
	Image    string
	Strategy string
	Wait     time.Duration
}

DeployRequest is one deploy of an existing service to an image.

type DirFetcher

type DirFetcher interface {
	FetchFiles(ctx context.Context, url, token, branch, sha string, dirs []string) (RepoFiles, error)
}

DirFetcher reads the first existing pipeline directory of a repository at sha, or at the branch tip when sha is empty.

type Duration

type Duration time.Duration

Duration is a time.Duration that unmarshals from a Go duration string.

func (Duration) MarshalYAML

func (d Duration) MarshalYAML() (any, error)

MarshalYAML renders the duration in Go duration syntax.

func (*Duration) UnmarshalYAML

func (d *Duration) UnmarshalYAML(n *yaml.Node) error

UnmarshalYAML parses values like "30s" or "1h30m".

type Engine

type Engine struct {
	// contains filtered or unexported fields
}

Engine drives pipeline runs from persisted state. Every Tick re-derives what to do from the database, so a restart resumes in-flight runs: jobs recorded as running with no live goroutine are picked up again.

func New

func New(cfg Config) *Engine

New builds an Engine. Call Run to drive it, or Tick from tests.

func (*Engine) Cancel

func (e *Engine) Cancel(ctx context.Context, runID string) error

Cancel flags a run for cancellation and wakes the engine.

func (*Engine) Close

func (e *Engine) Close()

Close interrupts running jobs and waits for them to stop.

func (*Engine) DecideHold

func (e *Engine) DecideHold(ctx context.Context, runID string, approved bool, by string) (bool, error)

DecideHold releases or rejects a run held for approval. It reports false when the run is not pending a decision.

func (*Engine) Nudge

func (e *Engine) Nudge()

Nudge asks the Run loop to tick soon.

func (*Engine) OIDCTokenRequestHandler

func (e *Engine) OIDCTokenRequestHandler() http.Handler

OIDCTokenRequestHandler serves the runtime token request endpoint a job script calls as PIPELINE_OIDC_REQUEST_URL: GET ?audience=X with "Authorization: Bearer <PIPELINE_OIDC_REQUEST_TOKEN>" mints a token for that one audience, only if the job's own `oidc` config allows it. Binding this to a container-reachable, never externally reachable address is the caller's job (see cmd/levelrail's wiring), not this package's: the bearer token is the actual access control either way.

func (*Engine) Rerun

func (e *Engine) Rerun(ctx context.Context, runID, actor string) (store.PipelineRun, error)

Rerun starts a new run from an earlier one's trigger and inputs, using the pipeline's current definition.

func (*Engine) Run

func (e *Engine) Run(ctx context.Context, interval, idleInterval time.Duration) error

Run ticks every interval while runs are active, and backs off to idleInterval when there are none. A Nudge always ticks at once. Interrupted jobs stay recorded as running and are resumed by the next process.

func (*Engine) Start

Start queues a run of a saved pipeline. The definition is re-validated and snapshotted onto the run so later edits never change it.

func (*Engine) Tick

func (e *Engine) Tick(ctx context.Context) error

Tick advances every active run once. Errors from one run never stop the rest.

func (*Engine) TriggerEvent

func (e *Engine) TriggerEvent(ctx context.Context, app string, ev Event, ref, sha, actor string) []store.PipelineRun

TriggerEvent starts a run of every enabled pipeline of app that accepts ev and records why each pipeline did or did not start. A pull request from a fork follows the pipeline's forks policy: blocked, held for approval, or allowed. Failures to start one pipeline are logged and never block the rest.

func (*Engine) Wait

func (e *Engine) Wait()

Wait blocks until no job goroutines are running (for tests).

type Event

type Event struct {
	Kind   string
	Branch string
	Tag    string
	// Fork marks a pull request whose source is another repository.
	Fork bool
	// HeadRepo is that source repository's full name, for the trigger log.
	HeadRepo string
	// Action is a pull request's normalized action (opened, synchronize).
	Action string
	// Changed lists the files the event touched. ChangedFn, when set, is
	// called at most once to fetch the list when Changed is empty. An
	// empty result means the list is unknown, so path filters pass.
	Changed   []string
	ChangedFn func(ctx context.Context) ([]string, error)
}

Event is what a trigger is matched against.

type FilterView

type FilterView struct {
	Paths        []string `json:"paths"`
	PathsIgnore  []string `json:"paths_ignore"`
	ReportStatus bool     `json:"report_status"`
	// Split is true when push and pull_request carry different filters. The
	// shown Paths and PathsIgnore are the push trigger's, and writing them
	// through ApplyFilters would overwrite the pull request ones.
	Split bool `json:"split"`
}

FilterView is the path filters and status reporting flag of a definition, as the editor and CLI show them.

func FiltersOf

func FiltersOf(def *Definition) FilterView

FiltersOf reads the filters from the push trigger, falling back to the pull_request trigger when there is no push trigger.

type Fix

type Fix struct {
	Code   string `json:"code"`
	Kind   string `json:"kind"`
	Job    string `json:"job"`
	Target string `json:"target,omitempty"`
	Hint   string `json:"hint"`
	Match  string `json:"match"`
}

Fix is a typed suggestion for a failed job. Code names the matched pattern, Kind says how it is applied, Target is where a guided fix lives.

func SuggestBuildFixes

func SuggestBuildFixes(step, output string) []Fix

SuggestBuildFixes matches a failed deploy build's log against the pipeline error patterns and returns deploy-specific typed fixes.

func SuggestFixes

func SuggestFixes(job, output string) []Fix

SuggestFixes matches the tail of a failed job's output against known error patterns, one fix per kind per pattern, strongest pattern first.

type GitFetcher

type GitFetcher struct{}

GitFetcher reads pipeline files with a shallow, in-memory clone, so nothing is written to disk and no checkout is needed.

func (GitFetcher) FetchFiles

func (GitFetcher) FetchFiles(ctx context.Context, url, token, branch, sha string, dirs []string) (RepoFiles, error)

FetchFiles returns the *.yaml and *.yml files of the first directory in dirs that exists. With sha empty it reads the tip of branch (the remote default when empty); otherwise it reads that exact commit and returns ErrSHAUnavailable when the commit cannot be fetched.

type Input

type Input struct {
	Description string `yaml:"description,omitempty" json:"description,omitempty"`
	Default     string `yaml:"default,omitempty" json:"default,omitempty"`
	Required    bool   `yaml:"required,omitempty" json:"required,omitempty"`
	Type        string `yaml:"type,omitempty" json:"type,omitempty"`
	// Choices lists the allowed values of a choice input; Options is the
	// older spelling of the same field.
	Choices []string `yaml:"choices,omitempty" json:"choices,omitempty"`
	Options []string `yaml:"options,omitempty" json:"options,omitempty"`
}

Input is one manual-run parameter.

func (Input) ChoiceList

func (in Input) ChoiceList() []string

ChoiceList returns the allowed values, from choices or the older options.

type Insights

type Insights struct {
	Window     int          `json:"window"`
	Jobs       []JobInsight `json:"jobs"`
	DiffRunID  string       `json:"diff_run_id,omitempty"`
	DiffBaseID string       `json:"diff_base_run_id,omitempty"`
	Diff       []JobDelta   `json:"diff"`
}

Insights is the run-over-run view of a pipeline.

func ComputeInsights

func ComputeInsights(runs []RunSample) Insights

ComputeInsights builds per-job stats and flaky flags over runs (newest first), plus a duration diff of the newest run against the latest earlier run in which every shared job succeeded.

type Issue

type Issue struct {
	Path    string `json:"path"`
	Line    int    `json:"line"`
	Message string `json:"message"`
}

Issue is one validation problem, located by dotted path and source line.

func (Issue) String

func (i Issue) String() string

type Job

type Job struct {
	Name            string            `yaml:"name,omitempty" json:"name,omitempty"`
	Stage           string            `yaml:"stage,omitempty" json:"stage,omitempty"`
	Needs           StringList        `yaml:"needs,omitempty" json:"needs,omitempty"`
	If              string            `yaml:"if,omitempty" json:"if,omitempty"`
	Image           string            `yaml:"image,omitempty" json:"image,omitempty"`
	Node            string            `yaml:"node,omitempty" json:"node,omitempty"`
	Env             map[string]string `yaml:"env,omitempty" json:"env,omitempty"`
	Matrix          *Matrix           `yaml:"matrix,omitempty" json:"matrix,omitempty"`
	Resources       *Resources        `yaml:"resources,omitempty" json:"resources,omitempty"`
	Timeout         Duration          `yaml:"timeout,omitempty" json:"timeout,omitempty"`
	Retries         int               `yaml:"retries,omitempty" json:"retries,omitempty"`
	ContinueOnError bool              `yaml:"continue_on_error,omitempty" json:"continue_on_error,omitempty"`
	Checkout        *bool             `yaml:"checkout,omitempty" json:"checkout,omitempty"`
	Cache           []CacheSpec       `yaml:"cache,omitempty" json:"cache,omitempty"`
	OIDC            *OIDCRequest      `yaml:"oidc,omitempty" json:"oidc,omitempty"`
	// Artifacts lists workspace paths stored when the job succeeds and
	// restored into the workspace of every job that needs this one.
	Artifacts StringList `yaml:"artifacts,omitempty" json:"artifacts,omitempty"`
	Steps     []Step     `yaml:"steps" json:"steps"`
}

Job is a group of steps run in one container on one node.

type JobDelta

type JobDelta struct {
	Job             string  `json:"job"`
	BaselineSeconds int64   `json:"baseline_seconds"`
	CurrentSeconds  int64   `json:"current_seconds"`
	DeltaSeconds    int64   `json:"delta_seconds"`
	DeltaPercent    float64 `json:"delta_percent"`
}

JobDelta is how a job's duration moved between two runs.

type JobInsight

type JobInsight struct {
	Job         string  `json:"job"`
	Runs        int     `json:"runs"`
	Failures    int     `json:"failures"`
	FailureRate float64 `json:"failure_rate"`
	P50Seconds  int64   `json:"p50_seconds"`
	LastSeconds int64   `json:"last_seconds"`
	Flaky       bool    `json:"flaky"`
	FlakyReason string  `json:"flaky_reason,omitempty"`
}

JobInsight summarises one job over a window of runs.

type JobSample

type JobSample struct {
	Key             string
	Status          string
	DurationSeconds int64
	Attempt         int
}

JobSample is one job's outcome in one run. Key may carry a matrix suffix.

type ManualTrigger

type ManualTrigger struct {
	Inputs map[string]Input `yaml:"inputs,omitempty" json:"inputs,omitempty"`
}

ManualTrigger declares the inputs a manual run prompts for.

type Matrix

type Matrix struct {
	Vars    map[string]StringList `yaml:",inline" json:"vars,omitempty"`
	Include []map[string]string   `yaml:"include,omitempty" json:"include,omitempty"`
	Exclude []map[string]string   `yaml:"exclude,omitempty" json:"exclude,omitempty"`
}

Matrix expands a job into one instance per combination of Vars.

func (*Matrix) Expand

func (m *Matrix) Expand() ([]Combo, error)

Expand returns the job's combinations. A nil or empty matrix yields one empty combination so callers treat plain and matrix jobs alike.

type MergeGroupTrigger

type MergeGroupTrigger struct {
	Branches StringList `yaml:"branches,omitempty" json:"branches,omitempty"`
}

MergeGroupTrigger filters merge queue events by the branch they merge into.

type OIDCIssuer

type OIDCIssuer func(ctx context.Context, req OIDCTokenRequest) (string, error)

OIDCIssuer mints a short-lived signed token for a job that opts in via `oidc: {audience: ...}` (Job.OIDC), for cloud OIDC federation (AWS IAM, GCP workload identity, Vault JWT auth) without a long-lived credential. Optional: nil means a job requesting oidc fails with a clear per-job error rather than silently getting no token. internal/oidc.Manager's IssueToken satisfies this once its request/response types are adapted by the caller (cmd/levelrail/pipelines.go), keeping this package unaware of internal/oidc entirely, the same "translation is the caller's job" shape RunEnv above already establishes.

type OIDCRequest

type OIDCRequest struct {
	// Audience is the token's "aud" claim, the value the cloud
	// provider's trust policy expects (e.g. "sts.amazonaws.com").
	// Pre-minted at job start and injected as PIPELINE_OIDC_TOKEN.
	Audience string `yaml:"audience,omitempty" json:"audience,omitempty"`
	// Audiences lists every additional audience this job may request a
	// token for at runtime, one at a time, via PIPELINE_OIDC_REQUEST_URL
	// (Audience is always implicitly allowed too). A job that needs
	// tokens for more than one audience in the same run (e.g. one step
	// calling AWS, another GCP) uses this instead of pre-minting every
	// combination up front.
	Audiences []string `yaml:"audiences,omitempty" json:"audiences,omitempty"`
}

OIDCRequest opts a job into minted OIDC tokens for cloud provider federation (AWS IAM, GCP workload identity, Vault JWT auth) without a long-lived credential. See docs/pipelines-oidc.md.

type OIDCTokenRequest

type OIDCTokenRequest struct {
	Audience   string
	Subject    string
	Repo       string
	Ref        string
	PipelineID string
}

OIDCTokenRequest is what a job's `oidc: {audience: ...}` config asks OIDCIssuer to mint a token for.

type PRTrigger

type PRTrigger struct {
	Branches StringList `yaml:"branches,omitempty" json:"branches,omitempty"`
	// Types limits which pull request actions start a run: opened,
	// reopened, synchronize. Empty means all three.
	Types       StringList `yaml:"types,omitempty" json:"types,omitempty"`
	Paths       StringList `yaml:"paths,omitempty" json:"paths,omitempty"`
	PathsIgnore StringList `yaml:"paths_ignore,omitempty" json:"paths_ignore,omitempty"`
	// Forks is block (the default), approve (hold the run until an
	// approver releases it), or allow.
	Forks string `yaml:"forks,omitempty" json:"forks,omitempty"`
}

PRTrigger filters pull request events by target branch glob and sets what happens to a pull request from a fork.

func (*PRTrigger) ForkPolicy

func (t *PRTrigger) ForkPolicy() string

ForkPolicy returns the effective fork policy, defaulting to block.

type Parameter

type Parameter struct {
	Name        string   `yaml:"name" json:"name"`
	Type        string   `yaml:"type,omitempty" json:"type,omitempty"`
	Default     string   `yaml:"default,omitempty" json:"default,omitempty"`
	Choices     []string `yaml:"choices,omitempty" json:"choices,omitempty"`
	Required    bool     `yaml:"required,omitempty" json:"required,omitempty"`
	Description string   `yaml:"description,omitempty" json:"description,omitempty"`
}

Parameter is one declared run parameter, set by a manual run, an API call, or a webhook query string.

func (Parameter) EffectiveType

func (p Parameter) EffectiveType() string

EffectiveType returns the declared type, inferred when left blank.

type RefTrigger

type RefTrigger struct {
	Branches    StringList `yaml:"branches,omitempty" json:"branches,omitempty"`
	Paths       StringList `yaml:"paths,omitempty" json:"paths,omitempty"`
	PathsIgnore StringList `yaml:"paths_ignore,omitempty" json:"paths_ignore,omitempty"`
}

RefTrigger filters push events by branch glob and by the paths the commits changed.

type RepoFiles

type RepoFiles struct {
	SHA   string
	Dir   string
	Files map[string][]byte
	// Oversize lists files skipped for exceeding maxSyncFileSize, by size.
	Oversize map[string]int64
}

RepoFiles is the pipeline directory of a repository at one commit.

type ReportReceipt

type ReportReceipt struct {
	Provider string
	// URL is the dashboard page the status links to, empty when the
	// control plane has no dashboard URL configured.
	URL string
}

ReportReceipt says where a status went.

type ReportRequest

type ReportRequest struct {
	App         string
	Pipeline    string
	RunID       string
	RunNumber   int
	SHA         string
	Ref         string
	State       string
	Description string
	// Context labels the status so several pipelines stay distinct.
	Context string
}

ReportRequest is one commit status to post for a run.

type Resources

type Resources struct {
	Memory string  `yaml:"memory,omitempty" json:"memory,omitempty"`
	CPU    float64 `yaml:"cpu,omitempty" json:"cpu,omitempty"`
}

Resources caps a job container. Memory uses Kubernetes-style suffixes.

type RunSample

type RunSample struct {
	RunID     string
	Number    int
	CommitSHA string
	Jobs      []JobSample
}

RunSample is one finished run, as handed to Insights newest first.

type Runtime

type Runtime interface {
	Create(ctx context.Context, spec docker.ContainerSpec) (string, error)
	Start(ctx context.Context, id string) error
	Remove(ctx context.Context, id string, force bool) error
	ListByPrefix(ctx context.Context, prefix string) ([]docker.ContainerState, error)
	EnsureVolume(ctx context.Context, name string) error
	ExecWithInput(ctx context.Context, containerID string, cmd []string, stdin io.Reader) (io.ReadCloser, error)
	docker.ArchiveRuntime
}

Runtime is the slice of docker.Runtime a job needs, so tests can fake it and remote nodes work through the same interface as the local daemon.

type RuntimeResolver

type RuntimeResolver func(nodeID string) (Runtime, error)

RuntimeResolver returns the runtime for a node ID ("" means the local node).

type Scheduler

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler starts pipelines whose `on.schedule` cron expressions come due. It mirrors internal/scheduledtask: a schedule arms on first sight and never catches up runs missed while the control plane was down.

func NewScheduler

func NewScheduler(engine *Engine) *Scheduler

NewScheduler builds a Scheduler over engine.

func (*Scheduler) Run

func (s *Scheduler) Run(ctx context.Context, interval time.Duration) error

Run ticks every interval until ctx is done.

func (*Scheduler) Tick

func (s *Scheduler) Tick(ctx context.Context) error

Tick checks every enabled pipeline once.

type Scope

type Scope struct {
	Vars      map[string]string
	Success   bool
	Failure   bool
	Cancelled bool
}

Scope is what expressions and ${{ }} references resolve against. Vars holds dotted paths such as "ref", "matrix.go", "needs.build.result".

type SecretResolver

type SecretResolver interface {
	Resolve(ctx context.Context, app, key string) (string, error)
}

SecretResolver reads an app's secret value. *secrets.Manager satisfies it.

type Source

type Source interface {
	RepoInfo(ctx context.Context, app string) (url, token string, err error)
}

Source resolves the repository a job checks out.

type StartOptions

type StartOptions struct {
	Trigger string
	Actor   string
	Ref     string
	SHA     string
	Inputs  map[string]string
	// BaseBranch is a pull request's target branch, which trigger filters
	// match against instead of the source branch in Ref.
	BaseBranch string
	// Hold, when set, creates the run held for approval with this reason:
	// no job starts and no secret is read until an approver releases it.
	Hold string
	// Payload and PayloadContentType carry an inbound webhook's body. Jobs
	// read it from the environment; it is never interpolated into scripts.
	Payload            []byte
	PayloadContentType string
}

StartOptions describes one run to start.

type StatusReporter

type StatusReporter interface {
	ReportRun(ctx context.Context, req ReportRequest) (ReportReceipt, error)
}

StatusReporter posts a run's state to the git forge that hosts its commit. Implementations return ErrNoReportTarget when there is no forge.

type Step

type Step struct {
	ID              string            `yaml:"id,omitempty" json:"id,omitempty"`
	Name            string            `yaml:"name,omitempty" json:"name,omitempty"`
	Run             string            `yaml:"run,omitempty" json:"run,omitempty"`
	Uses            string            `yaml:"uses,omitempty" json:"uses,omitempty"`
	With            map[string]string `yaml:"with,omitempty" json:"with,omitempty"`
	Env             map[string]string `yaml:"env,omitempty" json:"env,omitempty"`
	Secrets         []string          `yaml:"secrets,omitempty" json:"secrets,omitempty"`
	If              string            `yaml:"if,omitempty" json:"if,omitempty"`
	Timeout         Duration          `yaml:"timeout,omitempty" json:"timeout,omitempty"`
	Retries         int               `yaml:"retries,omitempty" json:"retries,omitempty"`
	ContinueOnError bool              `yaml:"continue_on_error,omitempty" json:"continue_on_error,omitempty"`
}

Step is one unit of work. Exactly one of Run or Uses is set.

func (Step) DisplayName

func (s Step) DisplayName() string

DisplayName is the step's name, falling back to its kind or script.

func (Step) Kind

func (s Step) Kind() string

Kind returns the step's kind, KindRun for a script step.

func (Step) NeedsContainer

func (s Step) NeedsContainer() bool

NeedsContainer reports whether the step executes inside the job container.

type Store

type Store interface {
	GetPipeline(ctx context.Context, id string) (store.Pipeline, error)
	ListPipelines(ctx context.Context, app string) ([]store.Pipeline, error)
	CreatePipelineRun(ctx context.Context, r store.PipelineRun) (store.PipelineRun, error)
	GetPipelineRun(ctx context.Context, id string) (store.PipelineRun, error)
	ListActivePipelineRuns(ctx context.Context) ([]store.PipelineRun, error)
	ListPipelineRunsInGroup(ctx context.Context, group string) ([]store.PipelineRun, error)
	SetPipelineRunStatus(ctx context.Context, id, status, reason string, started, finished *time.Time) error
	RequestPipelineRunCancel(ctx context.Context, id string) error
	CreatePipelineJobs(ctx context.Context, jobs []store.PipelineJob) error
	ListPipelineJobs(ctx context.Context, runID string) ([]store.PipelineJob, error)
	SetPipelineJobStatus(ctx context.Context, id, status, reason, nodeID string, attempt int, started, finished *time.Time) error
	SetPipelineJobOutputs(ctx context.Context, id, outputsJSON string) error
	SetPipelineStepStatus(ctx context.Context, jobID string, idx int, status, reason string, exitCode *int, attempt int, started, finished *time.Time) error
	AppendPipelineLogs(ctx context.Context, lines []store.PipelineLogLine) error
	CreatePipelineApproval(ctx context.Context, a store.PipelineApproval) error
	ListPipelineApprovals(ctx context.Context, runID string) ([]store.PipelineApproval, error)
	PrunePipelineRuns(ctx context.Context, keep int) (int64, error)
	AddPipelineTriggerLog(ctx context.Context, e store.PipelineTriggerLog) error
	DecidePipelineRunHold(ctx context.Context, id string, approved bool, by string, now time.Time) (bool, error)
}

Store is the persistence surface the Engine needs. *store.DB satisfies it.

type StringList

type StringList []string

StringList unmarshals either a single string or a list of strings.

func (*StringList) UnmarshalYAML

func (l *StringList) UnmarshalYAML(n *yaml.Node) error

UnmarshalYAML accepts a scalar or a sequence.

type SyncConfig

type SyncConfig struct {
	Store   SyncStore
	Source  Source
	Fetcher DirFetcher
	// BrandName is the brand short name for the branded pipeline directory.
	BrandName string
	Logger    *slog.Logger
	Now       func() time.Time
	NewID     func() string
}

SyncConfig wires a Syncer.

type SyncItem

type SyncItem struct {
	File    string `json:"file"`
	Name    string `json:"name"`
	Outcome string `json:"outcome"`
	Message string `json:"message,omitempty"`
}

SyncItem is the outcome for one pipeline file.

type SyncResult

type SyncResult struct {
	SHA   string     `json:"sha"`
	Dir   string     `json:"dir,omitempty"`
	Items []SyncItem `json:"items"`
}

SyncResult is the outcome of one repository sync.

type SyncStore

type SyncStore interface {
	ListPipelines(ctx context.Context, app string) ([]store.Pipeline, error)
	SavePipeline(ctx context.Context, p store.Pipeline) (store.Pipeline, error)
	GetPipelineSync(ctx context.Context, app string) (store.PipelineSync, error)
	SavePipelineSync(ctx context.Context, s store.PipelineSync) error
	GetGitSource(ctx context.Context, serviceName string) (*store.GitSource, error)
}

SyncStore is the persistence surface a Syncer needs. *store.DB satisfies it.

type Syncer

type Syncer struct {
	// contains filtered or unexported fields
}

Syncer copies an app's repository pipeline directory into its saved pipeline definitions.

func NewSyncer

func NewSyncer(cfg SyncConfig) *Syncer

NewSyncer builds a Syncer, defaulting the logger, clock, and ID source.

func (*Syncer) Sync

func (s *Syncer) Sync(ctx context.Context, app string) (SyncResult, error)

Sync reads the repository at the app's tracked branch and upserts the definitions it finds. The outcome, or the error, is recorded so the dashboard can show it.

func (*Syncer) SyncOnPush

func (s *Syncer) SyncOnPush(ctx context.Context, app, ref, sha string) (res SyncResult, ran bool, err error)

SyncOnPush syncs the definitions at the pushed commit sha when ref is a push to the app's tracked branch. ran is false for any other ref or an app with no repository.

type TagTrigger

type TagTrigger struct {
	Patterns StringList `yaml:"patterns,omitempty" json:"patterns,omitempty"`
}

TagTrigger filters tag pushes by glob.

type Template

type Template struct {
	Params []string `yaml:"params,omitempty" json:"params,omitempty"`
	Steps  []Step   `yaml:"steps" json:"steps"`
}

Template is a reusable list of steps invoked with `uses: template/<name>`. Parameters are referenced as ${{ inputs.<param> }} inside the steps.

type Triggers

type Triggers struct {
	Push        *RefTrigger        `yaml:"push,omitempty" json:"push,omitempty"`
	PullRequest *PRTrigger         `yaml:"pull_request,omitempty" json:"pull_request,omitempty"`
	Tag         *TagTrigger        `yaml:"tag,omitempty" json:"tag,omitempty"`
	MergeGroup  *MergeGroupTrigger `yaml:"merge_group,omitempty" json:"merge_group,omitempty"`
	Manual      *ManualTrigger     `yaml:"manual,omitempty" json:"manual,omitempty"`
	Schedule    []string           `yaml:"schedule,omitempty" json:"schedule,omitempty"`
	API         bool               `yaml:"api,omitempty" json:"api,omitempty"`
	// Webhook accepts runs from the pipeline's inbound webhook trigger.
	Webhook *WebhookTrigger `yaml:"webhook,omitempty" json:"webhook,omitempty"`
}

Triggers lists what starts a run. A nil field means that trigger is off.

func (Triggers) IsEmpty

func (t Triggers) IsEmpty() bool

IsEmpty reports whether no trigger is configured.

func (Triggers) Matches

func (t Triggers) Matches(ev Event) bool

Matches reports whether the definition should run for ev.

func (*Triggers) UnmarshalYAML

func (t *Triggers) UnmarshalYAML(n *yaml.Node) error

UnmarshalYAML decodes the `on` block. A trigger key with no value (`push:`) means "on, with defaults", which a plain pointer field would decode as off.

type VerifiedReader

type VerifiedReader struct {
	// contains filtered or unexported fields
}

VerifiedReader streams a stored archive while hashing it, so the sha256 is checked without reading the file twice. Verify must run after the consumer is done: it drains whatever the consumer left unread and reports ErrArtifactCorrupt on a size or hash mismatch.

func (*VerifiedReader) Close

func (v *VerifiedReader) Close() error

Close releases the file.

func (*VerifiedReader) Read

func (v *VerifiedReader) Read(p []byte) (int, error)

func (*VerifiedReader) Verify

func (v *VerifiedReader) Verify() error

Verify finishes the hash and returns ErrArtifactCorrupt on a mismatch.

type VolumeRemover

type VolumeRemover interface {
	RemoveVolume(ctx context.Context, name string) error
}

VolumeRemover is an optional Runtime capability used for best-effort workspace cleanup; runtimes without it leave volumes for the volume prune.

type WebhookTrigger

type WebhookTrigger struct{}

WebhookTrigger opts the pipeline into inbound webhook runs. The URL token and signature mode are managed through the API, not the file.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL