feat(transform): Add runID to transform. Publish some events.

dustmop · dustmop · commit 5ceac77387b0 · 2021-01-25T22:38:24.000-05:00
diff --git a/cmd/apply.go b/cmd/apply.go
@@ -102,6 +102,7 @@ func (o *ApplyOptions) Run() error {
 		Refstr:       o.Refs.Ref(),
 		Transform:    &tf,
 		ScriptOutput: o.Out,
+		Wait:         true,
 	}
 	res := lib.ApplyResult{}
 	if err = o.TransformMethods.Apply(&params, &res); err != nil {
diff --git a/event/event.go b/event/event.go
@@ -143,7 +143,7 @@ func (b *bus) Publish(ctx context.Context, typ Type, payload interface{}) error
 	return b.publish(ctx, typ, "", payload)
 }
 
-// Publish sends an event with a given sessionID to the bus
+// PublishID sends an event with a given sessionID to the bus
 func (b *bus) PublishID(ctx context.Context, typ Type, sessionID string, payload interface{}) error {
 	return b.publish(ctx, typ, sessionID, payload)
 }
@@ -209,7 +209,7 @@ func (b *bus) SubscribeID(handler Handler, sessionID string) {
 	b.idSubs[sessionID] = append(b.idSubs[sessionID], handler)
 }
 
-// SubscribeAll requets all events from the bus
+// SubscribeAll requests all events from the bus
 func (b *bus) SubscribeAll(handler Handler) {
 	b.lk.Lock()
 	defer b.lk.Unlock()
diff --git a/event/transform.go b/event/transform.go
@@ -0,0 +1,10 @@
+package event
+
+const (
+	// ETTransformStart signals the start a transform execution
+	ETTransformStart = Type("transform:Start")
+	// ETTransformComplete signals the successful completion of a transform execution
+	ETTransformComplete = Type("transform:Complete")
+	// ETTransformFailure signals the failure of a transform execution
+	ETTransformFailure = Type("transform:Failure")
+)
diff --git a/go.mod b/go.mod
@@ -14,6 +14,7 @@ require (
 	github.com/gofrs/flock v0.7.1 // indirect
 	github.com/google/flatbuffers v1.12.1-0.20200706154056-969d0f7a6317
 	github.com/google/go-cmp v0.5.3
+	github.com/google/uuid v1.1.1
 	github.com/ipfs/go-cid v0.0.7
 	github.com/ipfs/go-datastore v0.4.4
 	github.com/ipfs/go-ipfs v0.6.0
@@ -38,9 +39,8 @@ require (
 	github.com/multiformats/go-multihash v0.0.14
 	github.com/olekukonko/tablewriter v0.0.4
 	github.com/pkg/errors v0.9.1
-	github.com/qri-io/dataset v0.2.1-0.20201201155506-9b4fc79ffde8
 	github.com/qri-io/dag v0.2.2-0.20201208212257-ae00241c4b48
-	github.com/qri-io/dataset v0.2.1-0.20201124144731-82162a0f76e6
+	github.com/qri-io/dataset v0.2.1-0.20210119191645-a8a3aa840464
 	github.com/qri-io/deepdiff v0.2.1-0.20200807143746-d02d9f531f5b
 	github.com/qri-io/didmod v0.0.0-20201123165422-8b2e224c993a
 	github.com/qri-io/doggos v0.1.0
diff --git a/go.sum b/go.sum
@@ -1132,6 +1132,10 @@ github.com/qri-io/dataset v0.2.1-0.20201124144731-82162a0f76e6 h1:a9CYZQ+DCzwqg8
 github.com/qri-io/dataset v0.2.1-0.20201124144731-82162a0f76e6/go.mod h1:HtwGskdCECbOON0iVQHEEm6fykwDqharlqabc1ssj3Y=
 github.com/qri-io/dataset v0.2.1-0.20201201155506-9b4fc79ffde8 h1:/9pbWabRT9BbjFp1AjdAsXKT2NQp+mGmyvnTylPyEHY=
 github.com/qri-io/dataset v0.2.1-0.20201201155506-9b4fc79ffde8/go.mod h1:HtwGskdCECbOON0iVQHEEm6fykwDqharlqabc1ssj3Y=
+github.com/qri-io/dataset v0.2.1-0.20210118153421-4b6176e58a47 h1:8VYob4qYMMqvaGuJ54Js+8VrUOF8O8TrOJ/aJOeihbA=
+github.com/qri-io/dataset v0.2.1-0.20210118153421-4b6176e58a47/go.mod h1:HtwGskdCECbOON0iVQHEEm6fykwDqharlqabc1ssj3Y=
+github.com/qri-io/dataset v0.2.1-0.20210119191645-a8a3aa840464 h1:EzaPWYsCUEJnx6cniZpeF5g3JZJ2yOhsICjRVP9z7Yk=
+github.com/qri-io/dataset v0.2.1-0.20210119191645-a8a3aa840464/go.mod h1:HtwGskdCECbOON0iVQHEEm6fykwDqharlqabc1ssj3Y=
 github.com/qri-io/deepdiff v0.2.1-0.20200807143746-d02d9f531f5b h1:T8qEIv+qLi5mVWvSS329wJ+HbN7cfMwCWjRVzh/+upo=
 github.com/qri-io/deepdiff v0.2.1-0.20200807143746-d02d9f531f5b/go.mod h1:NrL/b7YvexgpGb4HEO3Rlx5RrMLDfxuKDf/XDAq5ac0=
 github.com/qri-io/didmod v0.0.0-20201123165422-8b2e224c993a h1:40BIa59lae2xZ7iieb3UU4/X57jZsWZ6QgqwdjDQhig=
diff --git a/lib/datasets.go b/lib/datasets.go
@@ -26,6 +26,7 @@ import (
 	"github.com/qri-io/qri/dscache/build"
 	"github.com/qri-io/qri/dsref"
 	qrierr "github.com/qri-io/qri/errors"
+	"github.com/qri-io/qri/event"
 	"github.com/qri-io/qri/fsi"
 	"github.com/qri-io/qri/fsi/linkfile"
 	"github.com/qri-io/qri/repo"
@@ -622,8 +623,17 @@ func (m *DatasetMethods) Save(p *SaveParams, res *dataset.Dataset) error {
 		// string and control how transform functions
 		loader := NewParseResolveLoadFunc("", m.inst.defaultResolver(), m.inst)
 
+		// allocate an ID for the transform, for now just log the events it produces
+		runID := transform.NewRunID()
+		m.inst.bus.SubscribeID(func(ctx context.Context, e event.Event) error {
+			when := time.Unix(e.Timestamp/1000000000, e.Timestamp%1000000000)
+			log.Infof("[%s] event %s: %s", when, e.Type, e.Payload)
+			return nil
+		}, runID)
+
 		// apply the transform
-		err := transform.Apply(ctx, ds, loader, str, scriptOut, secrets)
+		shouldWait := true
+		err := transform.Apply(ctx, ds, loader, runID, m.inst.bus, shouldWait, str, scriptOut, secrets)
 		if err != nil {
 			return err
 		}
diff --git a/lib/transform.go b/lib/transform.go
@@ -4,10 +4,12 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"time"
 
 	"github.com/qri-io/dataset"
 	"github.com/qri-io/qri/base"
 	"github.com/qri-io/qri/dsref"
+	"github.com/qri-io/qri/event"
 	"github.com/qri-io/qri/transform"
 )
 
@@ -30,12 +32,10 @@ func NewTransformMethods(inst *Instance) *TransformMethods {
 type ApplyParams struct {
 	Refstr    string
 	Transform *dataset.Transform
-	Source    string
 	Secrets   map[string]string
+	Wait      bool
 
-	// TODO(dustmop): add `Wait bool`, if false, run the save asynchronously
-	// and return events on the bus that provide the progress of the save operation
-
+	Source       string
 	ScriptOutput io.Writer
 }
 
@@ -49,9 +49,8 @@ func (p *ApplyParams) Valid() error {
 
 // ApplyResult is the result of an apply command
 type ApplyResult struct {
-	Data *dataset.Dataset
-	// TODO(dustmop): Make Apply asynchronous, running the transform in a go-routine.
-	// Return a channel that will send progress on the execution.
+	Data  *dataset.Dataset
+	RunID string `json:"runID"`
 }
 
 // Apply runs a transform script
@@ -87,15 +86,26 @@ func (m *TransformMethods) Apply(p *ApplyParams, res *ApplyResult) error {
 	str := m.inst.node.LocalStreams
 	loader := NewParseResolveLoadFunc("", m.inst.defaultResolver(), m.inst)
 
+	// allocate an ID for the transform, for now just log the events it produces
+	runID := transform.NewRunID()
+	m.inst.bus.SubscribeID(func(ctx context.Context, e event.Event) error {
+		when := time.Unix(e.Timestamp/1000000000, e.Timestamp%1000000000)
+		log.Infof("[%s] event %s: %s", when, e.Type, e.Payload)
+		return nil
+	}, runID)
+
 	scriptOut := p.ScriptOutput
-	err = transform.Apply(ctx, ds, loader, str, scriptOut, p.Secrets)
+	err = transform.Apply(ctx, ds, loader, runID, m.inst.bus, p.Wait, str, scriptOut, p.Secrets)
 	if err != nil {
 		return err
 	}
 
-	if err = base.InlineJSONBody(ds); err != nil && err != base.ErrNoBodyToInline {
-		return err
+	if p.Wait {
+		if err = base.InlineJSONBody(ds); err != nil && err != base.ErrNoBodyToInline {
+			return err
+		}
+		res.Data = ds
 	}
-	res.Data = ds
+	res.RunID = runID
 	return nil
 }
diff --git a/lib/transform_test.go b/lib/transform_test.go
@@ -27,6 +27,7 @@ func TestApplyTransform(t *testing.T) {
 		Transform: &dataset.Transform{
 			ScriptPath: "testdata/cities_2/add_city.star",
 		},
+		Wait: true,
 	})
 	if err != nil {
 		t.Error(err)
diff --git a/lib/websocket.go b/lib/websocket.go
@@ -64,8 +64,10 @@ func (h *wsHandler) WSConnectionHandler(w http.ResponseWriter, r *http.Request)
 func (h *wsHandler) wsMessageHandler(_ context.Context, e event.Event) error {
 	ctx := context.Background()
 	evt := map[string]interface{}{
-		"type": string(e.Type),
-		"data": e.Payload,
+		"type":      string(e.Type),
+		"ts":        e.Timestamp,
+		"sessionID": e.SessionID,
+		"data":      e.Payload,
 	}
 
 	log.Debugf("sending event %q to %d websocket conns", e.Type, len(h.conns))
diff --git a/transform/apply.go b/transform/apply.go
@@ -6,31 +6,41 @@ import (
 	"fmt"
 	"io"
 
+	"github.com/google/uuid"
+	golog "github.com/ipfs/go-log"
 	"github.com/qri-io/dataset"
 	"github.com/qri-io/ioes"
 	"github.com/qri-io/qri/dsref"
+	"github.com/qri-io/qri/event"
 	"github.com/qri-io/qri/transform/startf"
 )
 
-// TODO(dustmop): Tests. Especially once the `apply` command exists.
+var log = golog.Logger("transform")
 
 // Apply applies the transform script to order to modify the changing dataset
 func Apply(
 	ctx context.Context,
 	ds *dataset.Dataset,
 	loader dsref.ParseResolveLoad,
+	runID string,
+	pub event.Publisher,
+	wait bool,
 	str ioes.IOStreams,
 	scriptOut io.Writer,
 	secrets map[string]string,
-) (err error) {
+) error {
 	var (
 		target = ds
 		head   *dataset.Dataset
+		err    error
 	)
 
 	if target.Transform == nil || target.Transform.ScriptFile() == nil {
 		return errors.New("apply requires a transform component with a script file")
 	}
+	if runID == "" {
+		return errors.New("apply requires a runID")
+	}
 
 	if ds.Name != "" {
 		head, err = loader(ctx, fmt.Sprintf("%s/%s", ds.Peername, ds.Name))
@@ -54,11 +64,18 @@ func Apply(
 		startf.AddDatasetLoader(loader),
 	}
 
+	pub.PublishID(ctx, event.ETTransformStart, runID, "")
+
 	if err = startf.ExecScript(ctx, target, head, opts...); err != nil {
+		pub.PublishID(ctx, event.ETTransformFailure, runID, "")
 		return err
 	}
 
-	str.PrintErr("✅ transform complete\n")
-
+	pub.PublishID(ctx, event.ETTransformComplete, runID, "")
 	return nil
 }
+
+// NewRunID creates a run identifier
+func NewRunID() string {
+	return uuid.New().String()
+}
diff --git a/transform/startf/entry_reader_test.go b/transform/startf/entry_reader_test.go
@@ -2,7 +2,6 @@ package startf
 
 import (
 	"fmt"
-	"log"
 	"testing"
 
 	"github.com/qri-io/dataset"
@@ -37,7 +36,7 @@ func TestEntryReaderSimpleList(t *testing.T) {
 	for i, e := range expect {
 		ent, err := r.ReadEntry()
 		if err != nil {
-			log.Fatal(err)
+			t.Fatal(err)
 		}
 
 		if e.index != ent.Index {
@@ -77,7 +76,7 @@ func TestEntryReaderSimpleDict(t *testing.T) {
 	for i, e := range expect {
 		ent, err := r.ReadEntry()
 		if err != nil {
-			log.Fatal(err)
+			t.Fatal(err)
 		}
 
 		if e.index != ent.Index {
diff --git a/transform/startf/transform.go b/transform/startf/transform.go
@@ -9,6 +9,7 @@ import (
 	"io"
 	"io/ioutil"
 
+	golog "github.com/ipfs/go-log"
 	"github.com/qri-io/dataset"
 	"github.com/qri-io/qfs"
 	"github.com/qri-io/qri/dsref"
@@ -22,6 +23,8 @@ import (
 	"go.starlark.net/starlark"
 )
 
+var log = golog.Logger("startf")
+
 // Version is the version of qri that this transform was run with
 var Version = version.Version
 
@@ -283,7 +286,6 @@ type specialFunc func(t *transform, thread *starlark.Thread, ctx *skyctx.Context
 func callDownloadFunc(t *transform, thread *starlark.Thread, ctx *skyctx.Context) (result starlark.Value, err error) {
 	httpGuard.EnableNtwk()
 	defer httpGuard.DisableNtwk()
-	t.print("📡 running download...\n")
 
 	var download *starlark.Function
 	if download, err = t.globalFunc("download"); err != nil {
@@ -304,7 +306,6 @@ func callTransformFunc(t *transform, thread *starlark.Thread, ctx *skyctx.Contex
 		}
 		return err
 	}
-	t.print("🤖  running transform...\n")
 
 	d := skyds.NewDataset(t.prev, t.checkFunc)
 	d.SetMutable(t.next)
@@ -314,14 +315,10 @@ func callTransformFunc(t *transform, thread *starlark.Thread, ctx *skyctx.Contex
 	return nil
 }
 
-// print writes output only if a node is specified
-func (t *transform) print(msg string) {
-	t.stderr.Write([]byte(msg))
-}
-
 func (t *transform) locals() starlark.StringDict {
 	return starlark.StringDict{
 		"load_dataset": starlark.NewBuiltin("load_dataset", t.LoadDataset),
+		"print":        starlark.NewBuiltin("print", t.print),
 	}
 }
 
@@ -369,6 +366,16 @@ func (t *transform) LoadDataset(thread *starlark.Thread, _ *starlark.Builtin, ar
 	return skyds.NewDataset(ds, nil).Methods(), nil
 }
 
+func (t *transform) print(thread *starlark.Thread, _ *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
+	var message starlark.String
+	if err := starlark.UnpackArgs("print", args, kwargs, "message", &message); err != nil {
+		return starlark.None, err
+	}
+	t.stderr.Write([]byte(message))
+	io.WriteString(t.stderr, "\n")
+	return starlark.None, nil
+}
+
 // MutatedComponentsFunc returns a function for checking if a field has been
 // modified. it's a kind of data structure mutual exclusion lock
 // TODO (b5) - this should be refactored & expanded
@@ -398,7 +405,7 @@ func MutatedComponentsFunc(dsp *dataset.Dataset) func(path ...string) error {
 			return fmt.Errorf(`transform script and user-supplied dataset are both trying to set:
   %s
 
-please adjust either the transform script or remove the supplied '%s'`, path[0], path[0])
+please adjust either the transform script or remove the supplied %q`, path[0], path[0])
 		}
 		return nil
 	}
diff --git a/transform/startf/transform_test.go b/transform/startf/transform_test.go
@@ -71,8 +71,7 @@ func TestExecScript(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	expect := `🤖  running transform...
-hello world!`
+	expect := "hello world!\n"
 	if string(output) != expect {
 		t.Errorf("stderr mismatch. expected: '%s', got: '%s'", expect, string(output))
 	}

Original file line number	Diff line number	Diff line change
`@@ -102,6 +102,7 @@ func (o *ApplyOptions) Run() error {`
`102`	`102`	`Refstr: o.Refs.Ref(),`
`103`	`103`	`Transform: &tf,`
`104`	`104`	`ScriptOutput: o.Out,`
	`105`	`+ Wait: true,`
`105`	`106`	`}`
`106`	`107`	`res := lib.ApplyResult{}`
`107`	`108`	`if err = o.TransformMethods.Apply(&params, &res); err != nil {`
Original file line number	Diff line number	Diff line change
`@@ -143,7 +143,7 @@ func (b *bus) Publish(ctx context.Context, typ Type, payload interface{}) error`
`143`	`143`	`return b.publish(ctx, typ, "", payload)`
`144`	`144`	`}`
`145`	`145`
`146`		`-// Publish sends an event with a given sessionID to the bus`
	`146`	`+// PublishID sends an event with a given sessionID to the bus`
`147`	`147`	`func (b *bus) PublishID(ctx context.Context, typ Type, sessionID string, payload interface{}) error {`
`148`	`148`	`return b.publish(ctx, typ, sessionID, payload)`
`149`	`149`	`}`
`@@ -209,7 +209,7 @@ func (b *bus) SubscribeID(handler Handler, sessionID string) {`
`209`	`209`	`b.idSubs[sessionID] = append(b.idSubs[sessionID], handler)`
`210`	`210`	`}`
`211`	`211`
`212`		`-// SubscribeAll requets all events from the bus`
	`212`	`+// SubscribeAll requests all events from the bus`
`213`	`213`	`func (b *bus) SubscribeAll(handler Handler) {`
`214`	`214`	`b.lk.Lock()`
`215`	`215`	`defer b.lk.Unlock()`