forked from
tangled.org/core
Monorepo for Tangled — https://tangled.org
start splitting out workflow
+195
-77
+50
-58
spindle/engines/nixery/engine.go
+50
-58
spindle/engines/nixery/engine.go
···
18
18
"github.com/docker/docker/api/types/network"
19
19
"github.com/docker/docker/client"
20
20
"github.com/docker/docker/pkg/stdcopy"
21
-
"gopkg.in/yaml.v3"
22
21
"tangled.org/core/api/tangled"
23
22
"tangled.org/core/log"
24
23
"tangled.org/core/spindle/config"
25
24
"tangled.org/core/spindle/engine"
26
25
"tangled.org/core/spindle/models"
27
26
"tangled.org/core/spindle/secrets"
27
+
"tangled.org/core/spindle/workflow"
28
28
)
29
29
30
30
const (
···
43
43
cleanup map[string][]cleanupFunc
44
44
}
45
45
46
-
type Step struct {
47
-
name string
48
-
kind models.StepKind
49
-
command string
50
-
environment map[string]string
51
-
}
52
-
53
-
func (s Step) Name() string {
54
-
return s.name
55
-
}
56
-
57
-
func (s Step) Command() string {
58
-
return s.command
59
-
}
60
-
61
-
func (s Step) Kind() models.StepKind {
62
-
return s.kind
63
-
}
64
-
65
46
// setupSteps get added to start of Steps
66
47
type setupSteps []models.Step
67
48
···
76
57
env map[string]string
77
58
}
78
59
79
-
func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
80
-
swf := &models.Workflow{}
81
-
addl := addlFields{}
60
+
// nixeryWorkflowSchema defines the nixery-specific YAML fields
61
+
// Common fields are embedded inline to avoid duplicate unmarshaling
62
+
type nixeryWorkflowSchema struct {
63
+
workflow.CommonWorkflowFields `yaml:",inline"`
64
+
Dependencies map[string][]string `yaml:"dependencies"`
65
+
}
82
66
83
-
dwf := &struct {
84
-
Steps []struct {
85
-
Command string `yaml:"command"`
86
-
Name string `yaml:"name"`
87
-
Environment map[string]string `yaml:"environment"`
88
-
} `yaml:"steps"`
89
-
Dependencies map[string][]string `yaml:"dependencies"`
90
-
Environment map[string]string `yaml:"environment"`
91
-
}{}
92
-
err := yaml.Unmarshal([]byte(twf.Raw), &dwf)
93
-
if err != nil {
94
-
return nil, err
67
+
// GetCommonFields returns a pointer to the embedded common fields
68
+
func (s *nixeryWorkflowSchema) GetCommonFields() *workflow.CommonWorkflowFields {
69
+
return &s.CommonWorkflowFields
70
+
}
71
+
72
+
// NewWorkflowSchema returns the schema struct for unmarshaling nixery workflow YAMLs
73
+
func (e *Engine) NewWorkflowSchema() interface{} {
74
+
return &nixeryWorkflowSchema{}
75
+
}
76
+
77
+
func (e *Engine) InitWorkflow(wf *models.Workflow, tpl tangled.Pipeline) error {
78
+
// Extract the workflow data wrapper
79
+
wfData, ok := wf.Data.(*workflow.WorkflowData)
80
+
if !ok {
81
+
return fmt.Errorf("invalid workflow data type for nixery engine")
95
82
}
96
83
97
-
for _, dstep := range dwf.Steps {
98
-
sstep := Step{}
99
-
sstep.environment = dstep.Environment
100
-
sstep.command = dstep.Command
101
-
sstep.name = dstep.Name
102
-
sstep.kind = models.StepKindUser
103
-
swf.Steps = append(swf.Steps, sstep)
84
+
// Extract the parsed schema
85
+
schema, ok := wfData.Schema.(*nixeryWorkflowSchema)
86
+
if !ok {
87
+
return fmt.Errorf("invalid workflow schema type for nixery engine")
104
88
}
105
-
swf.Name = twf.Name
106
-
addl.env = dwf.Environment
107
-
addl.image = workflowImage(dwf.Dependencies, e.cfg.NixeryPipelines.Nixery)
108
89
109
-
setup := &setupSteps{}
90
+
// Steps are already created by the parser with environment support - no need to replace!
110
91
92
+
// Build the setup steps that run before user steps
93
+
setup := &setupSteps{}
111
94
setup.addStep(nixConfStep())
112
-
setup.addStep(cloneStep(twf, *tpl.TriggerMetadata, e.cfg.Server.Dev))
113
-
// this step could be empty
114
-
if s := dependencyStep(dwf.Dependencies); s != nil {
115
-
setup.addStep(*s)
95
+
96
+
// Use the original workflow for clone step (it has the Clone field)
97
+
setup.addStep(cloneStep(wfData.OriginalWorkflow, *tpl.TriggerMetadata, e.cfg.Server.Dev))
98
+
99
+
// Add dependency installation step if dependencies are specified
100
+
if s := dependencyStep(schema.Dependencies); s != nil {
101
+
setup.addStep(s)
116
102
}
117
103
118
-
// append setup steps in order to the start of workflow steps
119
-
swf.Steps = append(*setup, swf.Steps...)
120
-
swf.Data = addl
104
+
// Prepend setup steps to the beginning of the workflow
105
+
wf.Steps = append(*setup, wf.Steps...)
121
106
122
-
return swf, nil
107
+
// Store engine-specific runtime metadata
108
+
wf.Data = addlFields{
109
+
image: workflowImage(schema.Dependencies, e.cfg.NixeryPipelines.Nixery),
110
+
env: wfData.CommonEnv, // Use common environment from parser
111
+
// container will be set later in SetupWorkflow
112
+
}
113
+
114
+
return nil
123
115
}
124
116
125
117
func (e *Engine) WorkflowTimeout() time.Duration {
···
295
287
workflowEnvs.AddEnv(s.Key, s.Value)
296
288
}
297
289
298
-
step := w.Steps[idx].(Step)
290
+
step := w.Steps[idx]
299
291
300
292
select {
301
293
case <-ctx.Done():
···
304
296
}
305
297
306
298
envs := append(EnvVars(nil), workflowEnvs...)
307
-
for k, v := range step.environment {
299
+
for k, v := range step.Environment() {
308
300
envs.AddEnv(k, v)
309
301
}
310
302
envs.AddEnv("HOME", homeDir)
311
303
312
304
mkExecResp, err := e.docker.ContainerExecCreate(ctx, addl.container, container.ExecOptions{
313
-
Cmd: []string{"bash", "-c", step.command},
305
+
Cmd: []string{"bash", "-c", step.Command()},
314
306
AttachStdout: true,
315
307
AttachStderr: true,
316
308
Env: envs,
···
333
325
// Docker doesn't provide an API to kill an exec run
334
326
// (sure, we could grab the PID and kill it ourselves,
335
327
// but that's wasted effort)
336
-
e.l.Warn("step timed out", "step", step.Name)
328
+
e.l.Warn("step timed out", "step", step.Name())
337
329
338
330
<-tailDone
339
331
+18
-17
spindle/engines/nixery/setup_steps.go
+18
-17
spindle/engines/nixery/setup_steps.go
···
6
6
7
7
"tangled.org/core/api/tangled"
8
8
"tangled.org/core/spindle/steps"
9
+
"tangled.org/core/spindle/models"
9
10
)
10
11
11
-
func nixConfStep() Step {
12
+
func nixConfStep() models.Step {
12
13
setupCmd := `mkdir -p /etc/nix
13
14
echo 'extra-experimental-features = nix-command flakes' >> /etc/nix/nix.conf
14
15
echo 'build-users-group = ' >> /etc/nix/nix.conf`
15
-
return Step{
16
-
command: setupCmd,
17
-
name: "Configure Nix",
16
+
return models.DefaultStep{
17
+
StepCommand: setupCmd,
18
+
StepName: "Configure Nix",
18
19
}
19
20
}
20
21
···
25
26
// - git fetch --depth=<d> --recurse-submodules=<yes|no> <sha>
26
27
// - git checkout FETCH_HEAD
27
28
// And supports all trigger types (push, PR, manual) and clone options.
28
-
func cloneStep(twf tangled.Pipeline_Workflow, tr tangled.Pipeline_TriggerMetadata, dev bool) Step {
29
+
func cloneStep(twf tangled.Pipeline_Workflow, tr tangled.Pipeline_TriggerMetadata, dev bool) models.Step {
29
30
// Use shared clone command builder
30
31
cmds, err := steps.BuildCloneCommands(steps.CloneConfig{
31
32
Workflow: twf,
···
35
36
})
36
37
if err != nil {
37
38
// Return error step - this will cause the workflow to fail with a clear message
38
-
return Step{
39
-
command: fmt.Sprintf("echo 'Failed to build clone commands: %s' && exit 1", err.Error()),
40
-
name: "Clone repository into workspace (error)",
39
+
return models.DefaultStep{
40
+
StepCommand: fmt.Sprintf("echo 'Failed to build clone commands: %s' && exit 1", err.Error()),
41
+
StepName: "Clone repository into workspace (error)",
41
42
}
42
43
}
43
44
44
45
// Check if cloning should be skipped
45
46
if cmds.Skip {
46
-
return Step{}
47
+
return models.DefaultStep{}
47
48
}
48
49
49
50
// Nixery-specific: join commands with newlines for shell execution
50
-
return Step{
51
-
command: strings.Join(cmds.All, "\n"),
52
-
name: "Clone repository into workspace",
51
+
return models.DefaultStep{
52
+
StepCommand: strings.Join(cmds.All, "\n"),
53
+
StepName: "Clone repository into workspace",
53
54
}
54
55
}
55
56
···
57
58
// For dependencies using a custom registry (i.e. not nixpkgs), it collects
58
59
// all packages and adds a single 'nix profile install' step to the
59
60
// beginning of the workflow's step list.
60
-
func dependencyStep(deps map[string][]string) *Step {
61
+
func dependencyStep(deps map[string][]string) models.Step {
61
62
var customPackages []string
62
63
63
64
for registry, packages := range deps {
···
77
78
if len(customPackages) > 0 {
78
79
installCmd := "nix --extra-experimental-features nix-command --extra-experimental-features flakes profile install"
79
80
cmd := fmt.Sprintf("%s %s", installCmd, strings.Join(customPackages, " "))
80
-
installStep := Step{
81
-
command: cmd,
82
-
name: "Install custom dependencies",
83
-
environment: map[string]string{
81
+
installStep := models.DefaultStep{
82
+
StepCommand: cmd,
83
+
StepName: "Install custom dependencies",
84
+
StepEnvironment: map[string]string{
84
85
"NIX_NO_COLOR": "1",
85
86
"NIX_SHOW_DOWNLOAD_PROGRESS": "0",
86
87
},
+13
-1
spindle/models/engine.go
+13
-1
spindle/models/engine.go
···
9
9
)
10
10
11
11
type Engine interface {
12
-
InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*Workflow, error)
12
+
// NewWorkflowSchema returns an empty schema struct that will be used to unmarshal
13
+
// the workflow YAML. This allows each engine to define its own YAML structure
14
+
// with engine-specific fields (e.g., dependencies for nixery, image/architecture for k8s).
15
+
NewWorkflowSchema() interface{}
16
+
17
+
// InitWorkflow is called after the workflow has been parsed and initial steps created.
18
+
// Engines should use this to:
19
+
// - Extract their schema from wf.Data
20
+
// - Prepend any setup steps (e.g., clone, dependency installation)
21
+
// - Transform or replace steps if needed
22
+
// - Store final metadata in wf.Data for use in SetupWorkflow
23
+
InitWorkflow(wf *Workflow, tpl tangled.Pipeline) error
24
+
13
25
SetupWorkflow(ctx context.Context, wid WorkflowId, wf *Workflow) error
14
26
WorkflowTimeout() time.Duration
15
27
DestroyWorkflow(ctx context.Context, wid WorkflowId) error
+25
spindle/models/pipeline.go
+25
spindle/models/pipeline.go
···
10
10
Name() string
11
11
Command() string
12
12
Kind() StepKind
13
+
Environment() map[string]string
14
+
}
15
+
16
+
// DefaultStep provides a basic implementation of the models.Step interface
17
+
type DefaultStep struct {
18
+
StepName string
19
+
StepCommand string
20
+
StepKind StepKind
21
+
StepEnvironment map[string]string
22
+
}
23
+
24
+
func (s DefaultStep) Name() string {
25
+
return s.StepName
26
+
}
27
+
28
+
func (s DefaultStep) Command() string {
29
+
return s.StepCommand
30
+
}
31
+
32
+
func (s DefaultStep) Kind() StepKind {
33
+
return s.StepKind
34
+
}
35
+
36
+
func (s DefaultStep) Environment() map[string]string {
37
+
return s.StepEnvironment
13
38
}
14
39
15
40
type StepKind int
+3
-1
spindle/server.go
+3
-1
spindle/server.go
···
24
24
"tangled.org/core/spindle/models"
25
25
"tangled.org/core/spindle/queue"
26
26
"tangled.org/core/spindle/secrets"
27
+
"tangled.org/core/spindle/workflow"
27
28
"tangled.org/core/spindle/xrpc"
28
29
"tangled.org/core/xrpc/serviceauth"
29
30
)
···
331
332
workflows[eng] = []models.Workflow{}
332
333
}
333
334
334
-
ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl)
335
+
// Use the workflow parser to parse and initialize the workflow
336
+
ewf, err := workflow.ParseWorkflow(*w, tpl, eng)
335
337
if err != nil {
336
338
return err
337
339
}
+86
spindle/workflow/parser.go
+86
spindle/workflow/parser.go
···
1
+
package workflow
2
+
3
+
import (
4
+
"fmt"
5
+
6
+
"gopkg.in/yaml.v3"
7
+
"tangled.org/core/api/tangled"
8
+
"tangled.org/core/spindle/models"
9
+
)
10
+
11
+
// WorkflowData wraps the parsed engine schema and original workflow
12
+
// for engines that need access to both during initialization
13
+
type WorkflowData struct {
14
+
Schema interface{}
15
+
OriginalWorkflow tangled.Pipeline_Workflow
16
+
CommonEnv map[string]string // Workflow-level environment variables (common to all engines)
17
+
}
18
+
19
+
// CommonWorkflowFields represents the standard fields present in all workflow YAMLs.
20
+
// Engines should embed this in their schema structs using `yaml:",inline"` to avoid
21
+
// duplicate YAML unmarshaling.
22
+
type CommonWorkflowFields struct {
23
+
Engine string `yaml:"engine"`
24
+
Environment map[string]string `yaml:"environment"` // Workflow-level environment variables
25
+
Steps []struct {
26
+
Name string `yaml:"name"`
27
+
Command string `yaml:"command"`
28
+
Environment map[string]string `yaml:"environment"` // Per-step environment (overrides/extends workflow-level)
29
+
} `yaml:"steps"`
30
+
}
31
+
32
+
// ParseWorkflow parses a workflow YAML using the specified engine's schema
33
+
// and returns a fully initialized workflow ready for execution.
34
+
func ParseWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline, engine models.Engine) (*models.Workflow, error) {
35
+
// 1. Get engine-specific schema struct (should embed CommonWorkflowFields)
36
+
schema := engine.NewWorkflowSchema()
37
+
38
+
// 2. Unmarshal the YAML into the engine's schema ONCE
39
+
// This captures both common fields (via inline embedding) and engine-specific fields
40
+
if err := yaml.Unmarshal([]byte(twf.Raw), schema); err != nil {
41
+
return nil, fmt.Errorf("failed to parse workflow into engine schema: %w", err)
42
+
}
43
+
44
+
// 3. Extract common fields from the schema (via type assertion to interface with embedded fields)
45
+
commonFields, ok := schema.(interface {
46
+
GetCommonFields() *CommonWorkflowFields
47
+
})
48
+
if !ok {
49
+
return nil, fmt.Errorf("engine schema must implement GetCommonFields() method")
50
+
}
51
+
common := commonFields.GetCommonFields()
52
+
53
+
// 4. Convert common steps to models.Step
54
+
steps := make([]models.Step, len(common.Steps))
55
+
for i, s := range common.Steps {
56
+
steps[i] = models.DefaultStep{
57
+
StepName: s.Name,
58
+
StepCommand: s.Command,
59
+
StepKind: models.StepKindUser,
60
+
StepEnvironment: s.Environment,
61
+
}
62
+
}
63
+
64
+
// 5. Create the workflow with parsed data
65
+
// Store both the schema and original workflow for engines that need access to
66
+
// fields like Clone options
67
+
workflowData := &WorkflowData{
68
+
Schema: schema,
69
+
OriginalWorkflow: twf,
70
+
CommonEnv: common.Environment,
71
+
}
72
+
73
+
wf := &models.Workflow{
74
+
Steps: steps,
75
+
Name: twf.Name,
76
+
Data: workflowData,
77
+
}
78
+
79
+
// 6. Let the engine initialize the workflow
80
+
// This is where engines can prepend setup steps, transform data, etc.
81
+
if err := engine.InitWorkflow(wf, tpl); err != nil {
82
+
return nil, fmt.Errorf("engine initialization failed: %w", err)
83
+
}
84
+
85
+
return wf, nil
86
+
}