kjuulh/dagger
Archived
1
0
Fork 0
Code Issues 1 Pull Requests 1 Projects Releases Wiki Activity

merge Route into Env, rebase dagger compute to use the new API

Browse Source 
Signed-off-by: Andrea Luzzardi <aluzzardi@gmail.com>
This commit is contained in:
Andrea Luzzardi 2021-03-24 17:55:21 -07:00 committed by Solomon Hykes
parent 43b3af6fff
commit cba524eb0f
13 changed files with 578 additions and 772 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

132
cmd/dagger/cmd/compute.go
View File

@ -1,20 +1,22 @@
package cmd
import (
"encoding/json"
"errors"
"fmt"
"os"
"strings"
"dagger.io/go/cmd/dagger/logger"
"dagger.io/go/dagger"
"go.mozilla.org/sops"
"go.mozilla.org/sops/decrypt"
"github.com/google/uuid"
"github.com/spf13/cobra"
"github.com/spf13/viper"
)
var (
input *dagger.InputValue
updater *dagger.InputValue
)
var computeCmd = &cobra.Command{
Use: "compute CONFIG",
Short: "Compute a configuration",
@ -30,25 +32,100 @@ var computeCmd = &cobra.Command{
lg := logger.New()
ctx := lg.WithContext(cmd.Context())
env, err := dagger.NewEnv()
if err != nil {
lg.Fatal().Err(err).Msg("unable to initialize environment")
}
if err := updater.SourceFlag().Set(args[0]); err != nil {
lg.Fatal().Err(err).Msg("invalid local source")
name := getRouteName(lg, cmd)
st := &dagger.RouteState{
ID: uuid.New().String(),
Name: name,
LayoutSource: dagger.DirInput(args[0], []string{"*.cue", "cue.mod"}),
}
if err := env.SetUpdater(updater.Value()); err != nil {
lg.Fatal().Err(err).Msg("invalid updater script")
for _, input := range viper.GetStringSlice("input-string") {
parts := strings.SplitN(input, "=", 2)
k, v := parts[0], parts[1]
err := st.AddInput(ctx, k, dagger.TextInput(v))
if err != nil {
lg.Fatal().Err(err).Str("input", k).Msg("failed to add input")
}
}
if err := env.SetInput(input.Value()); err != nil {
lg.Fatal().Err(err).Msg("invalid input")
for _, input := range viper.GetStringSlice("input-dir") {
parts := strings.SplitN(input, "=", 2)
k, v := parts[0], parts[1]
err := st.AddInput(ctx, k, dagger.DirInput(v, []string{}))
if err != nil {
lg.Fatal().Err(err).Str("input", k).Msg("failed to add input")
}
}
for _, input := range viper.GetStringSlice("input-git") {
parts := strings.SplitN(input, "=", 2)
k, v := parts[0], parts[1]
err := st.AddInput(ctx, k, dagger.GitInput(v, "", ""))
if err != nil {
lg.Fatal().Err(err).Str("input", k).Msg("failed to add input")
}
}
if f := viper.GetString("input-json"); f != "" {
lg := lg.With().Str("path", f).Logger()
content, err := os.ReadFile(f)
if err != nil {
lg.Fatal().Err(err).Msg("failed to read file")
}
plaintext, err := decrypt.Data(content, "json")
if err != nil && !errors.Is(err, sops.MetadataNotFound) {
lg.Fatal().Err(err).Msg("unable to decrypt")
}
if len(plaintext) > 0 {
content = plaintext
}
if !json.Valid(content) {
lg.Fatal().Msg("invalid json")
}
err = st.AddInput(ctx, "", dagger.JSONInput(string(content)))
if err != nil {
lg.Fatal().Err(err).Msg("failed to add input")
}
}
if f := viper.GetString("input-yaml"); f != "" {
lg := lg.With().Str("path", f).Logger()
content, err := os.ReadFile(f)
if err != nil {
lg.Fatal().Err(err).Msg("failed to read file")
}
plaintext, err := decrypt.Data(content, "yaml")
if err != nil && !errors.Is(err, sops.MetadataNotFound) {
lg.Fatal().Err(err).Msg("unable to decrypt")
}
if len(plaintext) > 0 {
content = plaintext
}
err = st.AddInput(ctx, "", dagger.YAMLInput(string(content)))
if err != nil {
lg.Fatal().Err(err).Msg("failed to add input")
}
}
route, err := dagger.NewRoute(st)
if err != nil {
lg.Fatal().Err(err).Msg("unable to initialize route")
}
c, err := dagger.NewClient(ctx, "")
if err != nil {
lg.Fatal().Err(err).Msg("unable to create client")
}
output, err := c.Compute(ctx, env)
output, err := c.Up(ctx, route)
if err != nil {
lg.Fatal().Err(err).Msg("failed to compute")
}
@ -57,24 +134,11 @@ var computeCmd = &cobra.Command{
}
func init() {
var err error
// Setup --input-* flags
input, err = dagger.NewInputValue("{}")
if err != nil {
panic(err)
}
computeCmd.Flags().Var(input.StringFlag(), "input-string", "TARGET=STRING")
computeCmd.Flags().Var(input.DirFlag(), "input-dir", "TARGET=PATH")
computeCmd.Flags().Var(input.GitFlag(), "input-git", "TARGET=REMOTE#REF")
computeCmd.Flags().Var(input.CueFlag(), "input-cue", "CUE")
computeCmd.Flags().Var(input.JSONFlag(), "input-json", "JSON")
computeCmd.Flags().Var(input.YAMLFlag(), "input-yaml", "YAML")
// Setup (future) --from-* flags
updater, err = dagger.NewInputValue("[...{do:string, ...}]")
if err != nil {
panic(err)
}
computeCmd.Flags().StringSlice("input-string", []string{}, "TARGET=STRING")
computeCmd.Flags().StringSlice("input-dir", []string{}, "TARGET=PATH")
computeCmd.Flags().StringSlice("input-git", []string{}, "TARGET=REMOTE#REF")
computeCmd.Flags().String("input-json", "", "JSON")
computeCmd.Flags().String("input-yaml", "", "YAML")
if err := viper.BindPFlags(computeCmd.Flags()); err != nil {
panic(err)

3
cmd/dagger/cmd/new.go
View File

@ -41,7 +41,8 @@ var newCmd = &cobra.Command{
if upRoute {
lg.Info().Str("route-id", route.ID()).Msg("bringing route online")
if err := route.Up(ctx, nil); err != nil {
// FIXME
if err := route.FIXME(ctx); err != nil {
lg.Fatal().Err(err).Str("route-id", route.ID()).Msg("failed to create route")
}
}

3
cmd/dagger/cmd/up.go
View File

@ -32,7 +32,8 @@ var upCmd = &cobra.Command{
}
// TODO: Implement options: --no-cache
if err := route.Up(ctx, nil); err != nil {
// FIXME
if err := route.FIXME(ctx); err != nil {
lg.Fatal().Err(err).Str("route-name", routeName).Str("route-id", route.ID()).Msg("failed to up the route")
}
},

6
dagger/client.go
View File

@ -61,7 +61,7 @@ func NewClient(ctx context.Context, host string) (*Client, error) {
}
// FIXME: return completed *Env, instead of *compiler.Value
func (c *Client) Compute(ctx context.Context, env *Env) (*compiler.Value, error) {
func (c *Client) Up(ctx context.Context, env *Route) (*compiler.Value, error) {
lg := log.Ctx(ctx)
eg, gctx := errgroup.WithContext(ctx)
@ -95,7 +95,7 @@ func (c *Client) Compute(ctx context.Context, env *Env) (*compiler.Value, error)
return out, compiler.Err(eg.Wait())
}
func (c *Client) buildfn(ctx context.Context, env *Env, ch chan *bk.SolveStatus, w io.WriteCloser) error {
func (c *Client) buildfn(ctx context.Context, env *Route, ch chan *bk.SolveStatus, w io.WriteCloser) error {
lg := log.Ctx(ctx)
// Scan local dirs to grant access
@ -138,7 +138,7 @@ func (c *Client) buildfn(ctx context.Context, env *Env, ch chan *bk.SolveStatus,
// Compute output overlay
lg.Debug().Msg("computing env")
if err := env.Compute(ctx, s); err != nil {
if err := env.Up(ctx, s, nil); err != nil {
return nil, err
}

4
dagger/dagger_test.go
View File

@ -33,8 +33,8 @@ func TestLocalDirs(t *testing.T) {
}
}
func mkEnv(t *testing.T, updater, input string) *Env {
env, err := NewEnv()
func mkEnv(t *testing.T, updater, input string) *Route {
env, err := NewRoute()
if err != nil {
t.Fatal(err)
}

298
dagger/deprecated_input.go
View File

@ -1,298 +0,0 @@
package dagger
import (
"encoding/json"
"errors"
"fmt"
"net/url"
"os"
"strings"
"cuelang.org/go/cue"
"github.com/spf13/pflag"
"dagger.io/go/dagger/compiler"
"go.mozilla.org/sops"
"go.mozilla.org/sops/decrypt"
)
// A mutable cue value with an API suitable for user inputs,
// such as command-line flag parsing.
type InputValue struct {
root *compiler.Value
}
func (iv *InputValue) Value() *compiler.Value {
return iv.root
}
func (iv *InputValue) String() string {
s, _ := iv.root.SourceString()
return s
}
func NewInputValue(base interface{}) (*InputValue, error) {
root, err := compiler.Compile("base", base)
if err != nil {
return nil, err
}
return &InputValue{
root: root,
}, nil
}
func (iv *InputValue) Set(s string, enc func(string) (interface{}, error)) error {
// Split from eg. 'foo.bar={bla:"bla"}`
k, vRaw := splitkv(s)
v, err := enc(vRaw)
if err != nil {
return err
}
root, err := iv.root.MergePath(v, k)
if err != nil {
return err
}
iv.root = root
return nil
}
// Adapter to receive string values from pflag
func (iv *InputValue) StringFlag() pflag.Value {
return stringFlag{
iv: iv,
}
}
type stringFlag struct {
iv *InputValue
}
func (sf stringFlag) Set(s string) error {
return sf.iv.Set(s, func(s string) (interface{}, error) {
return s, nil
})
}
func (sf stringFlag) String() string {
return sf.iv.String()
}
func (sf stringFlag) Type() string {
return "STRING"
}
// DIR FLAG
// Receive a local directory path and translate it into a component
func (iv *InputValue) DirFlag(include ...string) pflag.Value {
if include == nil {
include = []string{}
}
return dirFlag{
iv: iv,
include: include,
}
}
type dirFlag struct {
iv *InputValue
include []string
}
func (f dirFlag) Set(s string) error {
return f.iv.Set(s, func(s string) (interface{}, error) {
// FIXME: this is a hack because cue API can't merge into a list
include, err := json.Marshal(f.include)
if err != nil {
return nil, err
}
return compiler.Compile("", fmt.Sprintf(
`#compute: [{do:"local",dir:"%s", include:%s}]`,
s,
include,
))
})
}
func (f dirFlag) String() string {
return f.iv.String()
}
func (f dirFlag) Type() string {
return "PATH"
}
// GIT FLAG
// Receive a git repository reference and translate it into a component
func (iv *InputValue) GitFlag() pflag.Value {
return gitFlag{
iv: iv,
}
}
type gitFlag struct {
iv *InputValue
}
func (f gitFlag) Set(s string) error {
return f.iv.Set(s, func(s string) (interface{}, error) {
u, err := url.Parse(s)
if err != nil {
return nil, fmt.Errorf("invalid git url")
}
ref := u.Fragment // eg. #main
u.Fragment = ""
remote := u.String()
return compiler.Compile("", fmt.Sprintf(
`#compute: [{do:"fetch-git", remote:"%s", ref:"%s"}]`,
remote,
ref,
))
})
}
func (f gitFlag) String() string {
return f.iv.String()
}
func (f gitFlag) Type() string {
return "REMOTE,REF"
}
// SOURCE FLAG
// Adapter to receive a simple source description and translate it to a loader script.
// For example 'git+https://github.com/cuelang/cue#master` -> [{do:"git",remote:"https://github.com/cuelang/cue",ref:"master"}]
func (iv *InputValue) SourceFlag() pflag.Value {
return sourceFlag{
iv: iv,
}
}
type sourceFlag struct {
iv *InputValue
}
func (f sourceFlag) Set(s string) error {
return f.iv.Set(s, func(s string) (interface{}, error) {
u, err := url.Parse(s)
if err != nil {
return nil, err
}
switch u.Scheme {
case "", "file":
return compiler.Compile(
"source",
// FIXME: include only cue files as a shortcut. Make this configurable somehow.
fmt.Sprintf(`[{do:"local",dir:"%s",include:["*.cue","cue.mod"]}]`, u.Host+u.Path),
)
default:
return nil, fmt.Errorf("unsupported source scheme: %q", u.Scheme)
}
})
}
func (f sourceFlag) String() string {
return f.iv.String()
}
func (f sourceFlag) Type() string {
return "PATH | file://PATH | git+ssh://HOST/PATH | git+https://HOST/PATH"
}
// RAW CUE FLAG
// Adapter to receive raw cue values from pflag
func (iv *InputValue) CueFlag() pflag.Value {
return cueFlag{
iv: iv,
}
}
type cueFlag struct {
iv *InputValue
}
func (f cueFlag) Set(s string) error {
return f.iv.Set(s, func(s string) (interface{}, error) {
return compiler.Compile("cue input", s)
})
}
func (f cueFlag) String() string {
return f.iv.String()
}
func (f cueFlag) Type() string {
return "CUE"
}
func (iv *InputValue) YAMLFlag() pflag.Value {
return fileFlag{
iv: iv,
format: "yaml",
}
}
func (iv *InputValue) JSONFlag() pflag.Value {
return fileFlag{
iv: iv,
format: "json",
}
}
type fileFlag struct {
format string
iv *InputValue
}
func (f fileFlag) Set(s string) error {
return f.iv.Set(s, func(s string) (interface{}, error) {
content, err := os.ReadFile(s)
if err != nil {
return nil, err
}
plaintext, err := decrypt.Data(content, f.format)
if err != nil && !errors.Is(err, sops.MetadataNotFound) {
return nil, fmt.Errorf("unable to decrypt %q: %w", s, err)
}
if len(plaintext) > 0 {
content = plaintext
}
switch f.format {
case "json":
return compiler.DecodeJSON(s, content)
case "yaml":
return compiler.DecodeYAML(s, content)
default:
panic("unsupported file format")
}
})
}
func (f fileFlag) String() string {
return f.iv.String()
}
func (f fileFlag) Type() string {
return strings.ToUpper(f.format)
}
// UTILITIES
func splitkv(kv string) (cue.Path, string) {
parts := strings.SplitN(kv, "=", 2)
if len(parts) == 2 {
if parts[0] == "." || parts[0] == "" {
return cue.MakePath(), parts[1]
}
return cue.ParsePath(parts[0]), parts[1]
}
if len(parts) == 1 {
return cue.MakePath(), parts[0]
}
return cue.MakePath(), ""
}

323
dagger/env.go
View File

@ -1,323 +0,0 @@
package dagger
import (
"context"
"fmt"
"io/fs"
"strings"
"time"
"cuelang.org/go/cue"
cueflow "cuelang.org/go/tools/flow"
"dagger.io/go/dagger/compiler"
"dagger.io/go/stdlib"
"github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go/ext"
otlog "github.com/opentracing/opentracing-go/log"
"github.com/rs/zerolog/log"
)
type Env struct {
// Env boot script, eg. `[{do:"local",dir:"."}]`
// FIXME: rename to 'update' (script to update the env config)
// FIXME: embed update script in base as '#update' ?
// FIXME: simplify Env by making it single layer? Each layer is one env.
// How to update the base configuration
updater *compiler.Value
// Layer 1: base configuration
base *compiler.Value
// Layer 2: user inputs
input *compiler.Value
// Layer 3: computed values
output *compiler.Value
// All layers merged together: base + input + output
state *compiler.Value
}
func (env *Env) Updater() *compiler.Value {
return env.updater
}
// Set the updater script for this environment.
func (env *Env) SetUpdater(v *compiler.Value) error {
if v == nil {
var err error
v, err = compiler.Compile("", "[]")
if err != nil {
return err
}
}
env.updater = v
return nil
}
func NewEnv() (*Env, error) {
empty := compiler.EmptyStruct()
env := &Env{
base: empty,
input: empty,
output: empty,
}
if err := env.mergeState(); err != nil {
return nil, err
}
if err := env.SetUpdater(nil); err != nil {
return nil, err
}
return env, nil
}
func (env *Env) State() *compiler.Value {
return env.state
}
func (env *Env) Input() *compiler.Value {
return env.input
}
func (env *Env) SetInput(i *compiler.Value) error {
if i == nil {
i = compiler.EmptyStruct()
}
env.input = i
return env.mergeState()
}
// Update the base configuration
func (env *Env) Update(ctx context.Context, s Solver) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "Env.Update")
defer span.Finish()
p := NewPipeline("[internal] source", s, nil)
// execute updater script
if err := p.Do(ctx, env.updater); err != nil {
return err
}
// Build a Cue config by overlaying the source with the stdlib
sources := map[string]fs.FS{
stdlib.Path: stdlib.FS,
"/": p.FS(),
}
base, err := compiler.Build(sources)
if err != nil {
return fmt.Errorf("base config: %w", err)
}
env.base = base
// Commit
return env.mergeState()
}
func (env *Env) Base() *compiler.Value {
return env.base
}
func (env *Env) Output() *compiler.Value {
return env.output
}
// Scan all scripts in the environment for references to local directories (do:"local"),
// and return all referenced directory names.
// This is used by clients to grant access to local directories when they are referenced
// by user-specified scripts.
func (env *Env) LocalDirs() map[string]string {
dirs := map[string]string{}
localdirs := func(code ...*compiler.Value) {
Analyze(
func(op *compiler.Value) error {
do, err := op.Get("do").String()
if err != nil {
return err
}
// nolint:goconst
// FIXME: merge Env into Route, or fix the linter error
if do != "local" {
return nil
}
dir, err := op.Get("dir").String()
if err != nil {
return err
}
dirs[dir] = dir
return nil
},
code...,
)
}
// 1. Scan the environment state
// FIXME: use a common `flow` instance to avoid rescanning the tree.
inst := env.state.CueInst()
flow := cueflow.New(&cueflow.Config{}, inst, newTaskFunc(inst, noOpRunner))
for _, t := range flow.Tasks() {
v := compiler.Wrap(t.Value(), inst)
localdirs(v.Get("#compute"))
}
// 2. Scan the environment updater
localdirs(env.Updater())
return dirs
}
// FIXME: this is just a 3-way merge. Add var args to compiler.Value.Merge.
func (env *Env) mergeState() error {
// FIXME: make this cleaner in *compiler.Value by keeping intermediary instances
// FIXME: state.CueInst() must return an instance with the same
// contents as state.v, for the purposes of cueflow.
// That is not currently how *compiler.Value works, so we prepare the cue
// instance manually.
// --> refactor the compiler.Value API to do this for us.
var (
state = compiler.EmptyStruct()
stateInst = state.CueInst()
err error
)
stateInst, err = stateInst.Fill(env.base.Cue())
if err != nil {
return fmt.Errorf("merge base & input: %w", err)
}
stateInst, err = stateInst.Fill(env.input.Cue())
if err != nil {
return fmt.Errorf("merge base & input: %w", err)
}
stateInst, err = stateInst.Fill(env.output.Cue())
if err != nil {
return fmt.Errorf("merge output with base & input: %w", err)
}
state = compiler.Wrap(stateInst.Value(), stateInst)
// commit
env.state = state
return nil
}
// Compute missing values in env configuration, and write them to state.
func (env *Env) Compute(ctx context.Context, s Solver) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "Env.Compute")
defer span.Finish()
lg := log.Ctx(ctx)
// Cueflow cue instance
inst := env.state.CueInst()
// Reset the output
env.output = compiler.EmptyStruct()
// Cueflow config
flowCfg := &cueflow.Config{
UpdateFunc: func(c *cueflow.Controller, t *cueflow.Task) error {
if t == nil {
return nil
}
lg := lg.
With().
Str("component", t.Path().String()).
Str("state", t.State().String()).
Logger()
if t.State() != cueflow.Terminated {
return nil
}
// Merge task value into output
var err error
env.output, err = env.output.MergePath(t.Value(), t.Path())
if err != nil {
lg.
Error().
Err(err).
Msg("failed to fill task result")
return err
}
return nil
},
}
// Orchestrate execution with cueflow
flow := cueflow.New(flowCfg, inst, newTaskFunc(inst, newPipelineRunner(inst, s)))
if err := flow.Run(ctx); err != nil {
return err
}
{
span, _ := opentracing.StartSpanFromContext(ctx, "Env.Compute: merge state")
defer span.Finish()
return env.mergeState()
}
}
func newTaskFunc(inst *cue.Instance, runner cueflow.RunnerFunc) cueflow.TaskFunc {
return func(flowVal cue.Value) (cueflow.Runner, error) {
v := compiler.Wrap(flowVal, inst)
if !isComponent(v) {
// No compute script
return nil, nil
}
return runner, nil
}
}
func noOpRunner(t *cueflow.Task) error {
return nil
}
func newPipelineRunner(inst *cue.Instance, s Solver) cueflow.RunnerFunc {
return cueflow.RunnerFunc(func(t *cueflow.Task) error {
ctx := t.Context()
lg := log.
Ctx(ctx).
With().
Str("component", t.Path().String()).
Logger()
ctx = lg.WithContext(ctx)
span, ctx := opentracing.StartSpanFromContext(ctx,
fmt.Sprintf("compute: %s", t.Path().String()),
)
defer span.Finish()
start := time.Now()
lg.
Info().
Msg("computing")
for _, dep := range t.Dependencies() {
lg.
Debug().
Str("dependency", dep.Path().String()).
Msg("dependency detected")
}
v := compiler.Wrap(t.Value(), inst)
p := NewPipeline(t.Path().String(), s, NewFillable(t))
err := p.Do(ctx, v)
if err != nil {
span.LogFields(otlog.String("error", err.Error()))
ext.Error.Set(span, true)
// FIXME: this should use errdefs.IsCanceled(err)
if strings.Contains(err.Error(), "context canceled") {
lg.
Error().
Dur("duration", time.Since(start)).
Msg("canceled")
return err
}
lg.
Error().
Dur("duration", time.Since(start)).
Err(err).
Msg("failed")
return err
}
lg.
Info().
Dur("duration", time.Since(start)).
Msg("completed")
return nil
})
}

38
dagger/input.go
View File

@ -34,9 +34,9 @@ func DirInput(path string, include []string) Input {
}
type dirInput struct {
Type string
Path string
Include []string
Type string `json:"type,omitempty"`
Path string `json:"path,omitempty"`
Include []string `json:"include,omitempty"`
}
func (dir dirInput) Compile() (*compiler.Value, error) {
@ -46,7 +46,7 @@ func (dir dirInput) Compile() (*compiler.Value, error) {
return nil, err
}
llb := fmt.Sprintf(
`[{do:"local",dir:"%s",include:%s}]`,
`#compute: [{do:"local",dir:"%s", include:%s}]`,
dir.Path,
includeLLB,
)
@ -55,10 +55,10 @@ func (dir dirInput) Compile() (*compiler.Value, error) {
// An input artifact loaded from a git repository
type gitInput struct {
Type string
Remote string
Ref string
Dir string
Type string `json:"type,omitempty"`
Remote string `json:"remote,omitempty"`
Ref string `json:"ref,omitempty"`
Dir string `json:"dir,omitempty"`
}
func GitInput(remote, ref, dir string) Input {
@ -83,8 +83,8 @@ func DockerInput(ref string) Input {
}
type dockerInput struct {
Type string
Ref string
Type string `json:"type,omitempty"`
Ref string `json:"ref,omitempty"`
}
func (i dockerInput) Compile() (*compiler.Value, error) {
@ -100,12 +100,12 @@ func TextInput(data string) Input {
}
type textInput struct {
Type string
Data string
Type string `json:"type,omitempty"`
Data string `json:"data,omitempty"`
}
func (i textInput) Compile() (*compiler.Value, error) {
panic("NOT IMPLEMENTED")
return compiler.Compile("", fmt.Sprintf("%q", i.Data))
}
// An input value encoded as JSON
@ -117,13 +117,13 @@ func JSONInput(data string) Input {
}
type jsonInput struct {
Type string
Type string `json:"type,omitempty"`
// Marshalled JSON data
Data string
Data string `json:"data,omitempty"`
}
func (i jsonInput) Compile() (*compiler.Value, error) {
panic("NOT IMPLEMENTED")
return compiler.DecodeJSON("", []byte(i.Data))
}
// An input value encoded as YAML
@ -135,11 +135,11 @@ func YAMLInput(data string) Input {
}
type yamlInput struct {
Type string
Type string `json:"type,omitempty"`
// Marshalled YAML data
Data string
Data string `json:"data,omitempty"`
}
func (i yamlInput) Compile() (*compiler.Value, error) {
panic("NOT IMPLEMENTED")
return compiler.DecodeYAML("", []byte(i.Data))
}

2
dagger/input_test.go
View File

@ -5,7 +5,7 @@ import (
)
func TestEnvInputFlag(t *testing.T) {
env, err := NewEnv()
env, err := NewRoute()
if err != nil {
t.Fatal(err)
}

401
dagger/route.go
View File

@ -2,27 +2,45 @@ package dagger
import (
"context"
"os"
"errors"
"fmt"
"io/fs"
"strings"
"time"
"cuelang.org/go/cue"
cueflow "cuelang.org/go/tools/flow"
"dagger.io/go/dagger/compiler"
"dagger.io/go/stdlib"
"github.com/google/uuid"
"github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go/ext"
otlog "github.com/opentracing/opentracing-go/log"
"github.com/rs/zerolog/log"
)
const (
routeLocation = "$HOME/.config/dagger/routes"
)
func init() {
f := os.ExpandEnv(routeLocation)
if err := os.MkdirAll(f, 0755); err != nil {
panic(err)
}
}
// A deployment route
type Route struct {
st routeState
st *RouteState
// Env boot script, eg. `[{do:"local",dir:"."}]`
// FIXME: rename to 'update' (script to update the env config)
// FIXME: embed update script in base as '#update' ?
// FIXME: simplify Env by making it single layer? Each layer is one r.
// How to update the base configuration
updater *compiler.Value
// Layer 1: layout configuration
layout *compiler.Value
// Layer 2: user inputs
input *compiler.Value
// Layer 3: computed values
output *compiler.Value
// All layers merged together: layout + input + output
state *compiler.Value
}
func (r *Route) ID() string {
@ -37,88 +55,299 @@ func (r *Route) LayoutSource() Input {
return r.st.LayoutSource
}
func (r *Route) SetLayoutSource(ctx context.Context, src Input) error {
r.st.LayoutSource = src
func NewRoute(st *RouteState) (*Route, error) {
empty := compiler.EmptyStruct()
r := &Route{
st: st,
layout: empty,
input: empty,
output: empty,
}
// Prepare inputs
for _, input := range st.Inputs {
v, err := input.Value.Compile()
if err != nil {
return nil, err
}
if input.Key == "" {
r.input, err = r.input.Merge(v)
} else {
r.input, err = r.input.MergeTarget(v, input.Key)
}
if err != nil {
return nil, err
}
}
if err := r.mergeState(); err != nil {
return nil, err
}
return r, nil
}
func (r *Route) State() *compiler.Value {
return r.state
}
// Update the base configuration
func (r *Route) Update(ctx context.Context, s Solver) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "r.Update")
defer span.Finish()
layout, err := r.st.LayoutSource.Compile()
if err != nil {
return err
}
p := NewPipeline("[internal] source", s, nil)
// execute updater script
if err := p.Do(ctx, layout); err != nil {
return err
}
// Build a Cue config by overlaying the source with the stdlib
sources := map[string]fs.FS{
stdlib.Path: stdlib.FS,
"/": p.FS(),
}
base, err := compiler.Build(sources)
if err != nil {
return fmt.Errorf("base config: %w", err)
}
r.layout = base
// Commit
return r.mergeState()
}
func (r *Route) Base() *compiler.Value {
return r.layout
}
func (r *Route) Output() *compiler.Value {
return r.output
}
// Scan all scripts in the environment for references to local directories (do:"local"),
// and return all referenced directory names.
// This is used by clients to grant access to local directories when they are referenced
// by user-specified scripts.
func (r *Route) LocalDirs() map[string]string {
dirs := map[string]string{}
localdirs := func(code ...*compiler.Value) {
Analyze(
func(op *compiler.Value) error {
do, err := op.Get("do").String()
if err != nil {
return err
}
// nolint:goconst
// FIXME: merge Env into Route, or fix the linter error
if do != "local" {
return nil
}
dir, err := op.Get("dir").String()
if err != nil {
return err
}
dirs[dir] = dir
return nil
},
code...,
)
}
// 1. Scan the environment state
// FIXME: use a common `flow` instance to avoid rescanning the tree.
inst := r.state.CueInst()
flow := cueflow.New(&cueflow.Config{}, inst, newTaskFunc(inst, noOpRunner))
for _, t := range flow.Tasks() {
v := compiler.Wrap(t.Value(), inst)
localdirs(v.Get("#compute"))
}
// 2. Scan the layout
layout, err := r.st.LayoutSource.Compile()
if err != nil {
panic(err)
}
localdirs(layout)
return dirs
}
// FIXME: this is just a 3-way merge. Add var args to compiler.Value.Merge.
func (r *Route) mergeState() error {
// FIXME: make this cleaner in *compiler.Value by keeping intermediary instances
// FIXME: state.CueInst() must return an instance with the same
// contents as state.v, for the purposes of cueflow.
// That is not currently how *compiler.Value works, so we prepare the cue
// instance manually.
// --> refactor the compiler.Value API to do this for us.
var (
state = compiler.EmptyStruct()
stateInst = state.CueInst()
err error
)
stateInst, err = stateInst.Fill(r.layout.Cue())
if err != nil {
return fmt.Errorf("merge base & input: %w", err)
}
stateInst, err = stateInst.Fill(r.input.Cue())
if err != nil {
return fmt.Errorf("merge base & input: %w", err)
}
stateInst, err = stateInst.Fill(r.output.Cue())
if err != nil {
return fmt.Errorf("merge output with base & input: %w", err)
}
state = compiler.Wrap(stateInst.Value(), stateInst)
// commit
r.state = state
return nil
}
func (r *Route) AddInput(ctx context.Context, key string, value Input) error {
r.st.Inputs = append(r.st.Inputs, inputKV{Key: key, Value: value})
return nil
}
// Remove all inputs at the given key, including sub-keys.
// For example RemoveInputs("foo.bar") will remove all inputs
// at foo.bar, foo.bar.baz, etc.
func (r *Route) RemoveInputs(ctx context.Context, key string) error {
panic("NOT IMPLEMENTED")
}
// Contents of a route serialized to a file
type routeState struct {
// Globally unique route ID
ID string
// Human-friendly route name.
// A route may have more than one name.
// FIXME: store multiple names?
Name string
// Cue module containing the route layout
// The input's top-level artifact is used as a module directory.
LayoutSource Input
Inputs []inputKV
}
type inputKV struct {
Key string
Value Input
}
func CreateRoute(ctx context.Context, name string, o *CreateOpts) (*Route, error) {
return &Route{
st: routeState{
ID: uuid.New().String(),
Name: name,
},
}, nil
}
type CreateOpts struct{}
func DeleteRoute(ctx context.Context, o *DeleteOpts) (*Route, error) {
panic("NOT IMPLEMENTED")
}
type DeleteOpts struct{}
func LookupRoute(name string, o *LookupOpts) (*Route, error) {
panic("NOT IMPLEMENTED")
}
type LookupOpts struct{}
func LoadRoute(ctx context.Context, id string, o *LoadOpts) (*Route, error) {
panic("NOT IMPLEMENTED")
}
type LoadOpts struct{}
func (r *Route) Up(ctx context.Context, o *UpOpts) error {
panic("NOT IMPLEMENTED")
}
type UpOpts struct{}
func (r *Route) Down(ctx context.Context, o *DownOpts) error {
panic("NOT IMPLEMENTED")
// Up missing values in env configuration, and write them to state.
func (r *Route) Up(ctx context.Context, s Solver, _ *UpOpts) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "r.Compute")
defer span.Finish()
lg := log.Ctx(ctx)
// Cueflow cue instance
inst := r.state.CueInst()
// Reset the output
r.output = compiler.EmptyStruct()
// Cueflow config
flowCfg := &cueflow.Config{
UpdateFunc: func(c *cueflow.Controller, t *cueflow.Task) error {
if t == nil {
return nil
}
lg := lg.
With().
Str("component", t.Path().String()).
Str("state", t.State().String()).
Logger()
if t.State() != cueflow.Terminated {
return nil
}
// Merge task value into output
var err error
r.output, err = r.output.MergePath(t.Value(), t.Path())
if err != nil {
lg.
Error().
Err(err).
Msg("failed to fill task result")
return err
}
return nil
},
}
// Orchestrate execution with cueflow
flow := cueflow.New(flowCfg, inst, newTaskFunc(inst, newPipelineRunner(inst, s)))
if err := flow.Run(ctx); err != nil {
return err
}
{
span, _ := opentracing.StartSpanFromContext(ctx, "r.Compute: merge state")
defer span.Finish()
return r.mergeState()
}
}
type DownOpts struct{}
func (r *Route) Down(ctx context.Context, _ *DownOpts) error {
panic("NOT IMPLEMENTED")
}
func (r *Route) Query(ctx context.Context, expr interface{}, o *QueryOpts) (*compiler.Value, error) {
panic("NOT IMPLEMENTED")
}
func (r *Route) FIXME(ctx context.Context) error {
return errors.New("FIXME")
}
type QueryOpts struct{}
func newTaskFunc(inst *cue.Instance, runner cueflow.RunnerFunc) cueflow.TaskFunc {
return func(flowVal cue.Value) (cueflow.Runner, error) {
v := compiler.Wrap(flowVal, inst)
if !isComponent(v) {
// No compute script
return nil, nil
}
return runner, nil
}
}
func noOpRunner(t *cueflow.Task) error {
return nil
}
func newPipelineRunner(inst *cue.Instance, s Solver) cueflow.RunnerFunc {
return cueflow.RunnerFunc(func(t *cueflow.Task) error {
ctx := t.Context()
lg := log.
Ctx(ctx).
With().
Str("component", t.Path().String()).
Logger()
ctx = lg.WithContext(ctx)
span, ctx := opentracing.StartSpanFromContext(ctx,
fmt.Sprintf("compute: %s", t.Path().String()),
)
defer span.Finish()
start := time.Now()
lg.
Info().
Msg("computing")
for _, dep := range t.Dependencies() {
lg.
Debug().
Str("dependency", dep.Path().String()).
Msg("dependency detected")
}
v := compiler.Wrap(t.Value(), inst)
p := NewPipeline(t.Path().String(), s, NewFillable(t))
err := p.Do(ctx, v)
if err != nil {
span.LogFields(otlog.String("error", err.Error()))
ext.Error.Set(span, true)
// FIXME: this should use errdefs.IsCanceled(err)
if strings.Contains(err.Error(), "context canceled") {
lg.
Error().
Dur("duration", time.Since(start)).
Msg("canceled")
return err
}
lg.
Error().
Dur("duration", time.Since(start)).
Err(err).
Msg("failed")
return err
}
lg.
Info().
Dur("duration", time.Since(start)).
Msg("completed")
return nil
})
}

132
dagger/state.go Normal file
View File

@ -0,0 +1,132 @@
package dagger
import (
"context"
"encoding/json"
"errors"
"os"
"path"
"github.com/google/uuid"
)
const (
routeLocation = "$HOME/.config/dagger/routes"
)
// Contents of a route serialized to a file
type RouteState struct {
// Globally unique route ID
ID string `json:"id,omitempty"`
// Human-friendly route name.
// A route may have more than one name.
// FIXME: store multiple names?
Name string `json:"name,omitempty"`
// Cue module containing the route layout
// The input's top-level artifact is used as a module directory.
LayoutSource Input `json:"layout,omitempty"`
Inputs []inputKV `json:"inputs,omitempty"`
}
type inputKV struct {
Key string `json:"key,omitempty"`
Value Input `json:"value,omitempty"`
}
func (r *RouteState) SetLayoutSource(ctx context.Context, src Input) error {
r.LayoutSource = src
return nil
}
func (r *RouteState) AddInput(ctx context.Context, key string, value Input) error {
r.Inputs = append(r.Inputs, inputKV{Key: key, Value: value})
return nil
}
// Remove all inputs at the given key, including sub-keys.
// For example RemoveInputs("foo.bar") will remove all inputs
// at foo.bar, foo.bar.baz, etc.
func (r *RouteState) RemoveInputs(ctx context.Context, key string) error {
panic("NOT IMPLEMENTED")
}
func routePath(name string) string {
return path.Join(os.ExpandEnv(routeLocation), name+".json")
}
func syncRoute(r *Route) error {
p := routePath(r.st.Name)
if err := os.MkdirAll(path.Dir(p), 0755); err != nil {
return err
}
data, err := json.MarshalIndent(r.st, "", " ")
if err != nil {
return err
}
return os.WriteFile(p, data, 0644)
}
func loadRoute(name string) (*RouteState, error) {
data, err := os.ReadFile(routePath(name))
if err != nil {
return nil, err
}
var st *RouteState
if err := json.Unmarshal(data, st); err != nil {
return nil, err
}
return st, nil
}
func CreateRoute(ctx context.Context, name string, o *CreateOpts) (*Route, error) {
r, err := LookupRoute(name, &LookupOpts{})
if err != nil && !errors.Is(err, os.ErrNotExist) {
return nil, err
}
if r != nil {
return nil, os.ErrExist
}
r, err = NewRoute(
&RouteState{
ID: uuid.New().String(),
Name: name,
},
)
if err != nil {
return nil, err
}
return r, syncRoute(r)
}
type CreateOpts struct{}
func DeleteRoute(ctx context.Context, o *DeleteOpts) (*Route, error) {
panic("NOT IMPLEMENTED")
}
type DeleteOpts struct{}
func LookupRoute(name string, o *LookupOpts) (*Route, error) {
st, err := loadRoute(name)
if err != nil {
return nil, err
}
return &Route{
st: st,
}, nil
}
type LookupOpts struct{}
func LoadRoute(ctx context.Context, id string, o *LoadOpts) (*Route, error) {
panic("NOT IMPLEMENTED")
}
type LoadOpts struct{}

1
examples/react-netlify/main.cue
View File

@ -25,5 +25,6 @@ todoApp: netlify.#Site & {
contents: yarn.#Script & {
source: repository
run: "build"
env: "xx" :"bar"
}
}

7
tests/test.sh
View File

@ -129,7 +129,7 @@ test::exec(){
test::one "Exec: env valid" --exit=0 --stdout={} \
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute "$d"/exec/env/valid
test::one "Exec: env with overlay" --exit=0 \
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute --input-cue 'bar: "overlay environment"' "$d"/exec/env/overlay
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute --input-string 'bar=overlay environment' "$d"/exec/env/overlay
test::one "Exec: non existent dir" --exit=0 --stdout={} \
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute "$d"/exec/dir/doesnotexist
@ -230,16 +230,15 @@ test::input() {
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute "$d"/input/simple
test::one "Input: simple input" --exit=0 --stdout='{"in":"foobar","test":"received: foobar"}' \
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute --input-cue 'in: "foobar"' "$d"/input/simple
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute --input-string 'in=foobar' "$d"/input/simple
test::one "Input: default values" --exit=0 --stdout='{"in":"default input","test":"received: default input"}' \
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute "$d"/input/default
test::one "Input: override default value" --exit=0 --stdout='{"in":"foobar","test":"received: foobar"}' \
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute --input-cue 'in: "foobar"' "$d"/input/default
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute --input-string 'in=foobar' "$d"/input/default
}
test::subdir() {
test::one "Subdir: simple usage" --exit=0 --stdout='{"hello":"world"}' \
"$dagger" "${DAGGER_BINARY_ARGS[@]}" compute "$d"/subdir/simple