Skip to content

Commit 8e69e5e

Browse files
b5b5
committed
feat(run): run package defines state of a transform run
run.State is a passable, cachable data structure that represents the state of a run. We build one up as events are emitted & can use them for storage upon completion.
1 parent e5579a0 commit 8e69e5e

File tree

3 files changed

+288
-0
lines changed

3 files changed

+288
-0
lines changed

‎event/event.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -164,6 +164,10 @@ func (b *bus) publish(ctx context.Context, typ Type, sessionID string, payload i
164164
Payload: payload,
165165
}
166166

167+
if b.closed {
168+
return e, ErrBusClosed
169+
}
170+
167171
// TODO(dustmop): Add instrumentation, perhaps to ctx, to make logging / tracing
168172
// a single event easier to do.
169173

‎transform/run/run.go‎

Lines changed: 168 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,168 @@
1+
// Package run defines metadata about transform script execution
2+
package run
3+
4+
import (
5+
"fmt"
6+
"time"
7+
8+
"github.com/google/uuid"
9+
"github.com/qri-io/qri/event"
10+
)
11+
12+
// NewID creates a run identifier
13+
func NewID() string {
14+
return uuid.New().String()
15+
}
16+
17+
// Status enumerates all possible execution states of a transform script or
18+
// step within a script, in relation to the current time.
19+
// Scripts & steps that have completed are broken into categories based on exit
20+
// state
21+
type Status string
22+
23+
const (
24+
// RSWaiting indicates a script/step that has yet to start
25+
RSWaiting = Status("waiting")
26+
// RSRunning indicates a script/step is currently executing
27+
RSRunning = Status("running")
28+
// RSSucceeded indicates a script/step has completed without error
29+
RSSucceeded = Status("succeeded")
30+
// RSFailed indicates a script/step completed & exited when an unexpected error
31+
// occured
32+
RSFailed = Status("failed")
33+
// RSUnchanged indicates a script completed but no changes were found
34+
// since the last version of the script succeeded
35+
RSUnchanged = Status("unchanged")
36+
// RSSkipped indicates a script/step was not executed
37+
RSSkipped = Status("skipped")
38+
)
39+
40+
// State is a passable, cachable data structure that describes the execution of
41+
// a transform. State structs can act as a sink of transform events, collapsing
42+
// the state transition of multiple transform events into a single structure
43+
type State struct {
44+
ID string `json:"id"`
45+
Number int `json:"number"`
46+
Status Status `json:"status"`
47+
Message string `json:"message"`
48+
StartTime *time.Time `json:"startTime"`
49+
StopTime *time.Time `json:"stopTime"`
50+
Duration int `json:"duration"`
51+
Steps []*StepState `json:"steps"`
52+
}
53+
54+
// NewState is a simple constructor to remind package consumers that state
55+
// structs must be initialized with an identifier to act as a sink of transform
56+
// events
57+
func NewState(id string) *State {
58+
return &State{
59+
ID: id,
60+
}
61+
}
62+
63+
// AddTransformEvent alters state based on a given event
64+
func (rs *State) AddTransformEvent(e event.Event) error {
65+
if rs.ID != e.SessionID {
66+
// silently ignore session ID mismatch
67+
return nil
68+
}
69+
70+
switch e.Type {
71+
case event.ETTransformStart:
72+
rs.Status = RSRunning
73+
rs.StartTime = toTimePointer(e.Timestamp)
74+
return nil
75+
case event.ETTransformStop:
76+
rs.StopTime = toTimePointer(e.Timestamp)
77+
if tl, ok := e.Payload.(event.TransformLifecycle); ok {
78+
rs.Status = Status(tl.Status)
79+
}
80+
if rs.StartTime != nil && rs.StopTime != nil {
81+
rs.Duration = int(rs.StopTime.Sub(*rs.StartTime))
82+
}
83+
return nil
84+
case event.ETTransformStepStart:
85+
s, err := NewStepStateFromEvent(e)
86+
if err != nil {
87+
return err
88+
}
89+
s.Status = RSRunning
90+
s.StartTime = toTimePointer(e.Timestamp)
91+
rs.Steps = append(rs.Steps, s)
92+
return nil
93+
case event.ETTransformStepStop:
94+
step, err := rs.lastStep()
95+
if err != nil {
96+
return err
97+
}
98+
step.StopTime = toTimePointer(e.Timestamp)
99+
if tsl, ok := e.Payload.(event.TransformStepLifecycle); ok {
100+
step.Status = Status(tsl.Status)
101+
} else {
102+
step.Status = RSFailed
103+
}
104+
if step.StartTime != nil && step.StopTime != nil {
105+
step.Duration = int(step.StopTime.Sub(*step.StartTime))
106+
}
107+
return nil
108+
case event.ETTransformStepSkip:
109+
s, err := NewStepStateFromEvent(e)
110+
if err != nil {
111+
return err
112+
}
113+
s.Status = RSSkipped
114+
rs.Steps = append(rs.Steps, s)
115+
return nil
116+
case event.ETTransformPrint,
117+
event.ETTransformError,
118+
event.ETTransformDatasetPreview:
119+
return rs.appendStepOutputLog(e)
120+
}
121+
return fmt.Errorf("unexpected event type: %q", e.Type)
122+
}
123+
124+
func (rs *State) lastStep() (*StepState, error) {
125+
if len(rs.Steps) > 0 {
126+
return rs.Steps[len(rs.Steps)-1], nil
127+
}
128+
return nil, fmt.Errorf("expected step to exist")
129+
}
130+
131+
func (rs *State) appendStepOutputLog(e event.Event) error {
132+
step, err := rs.lastStep()
133+
if err != nil {
134+
return err
135+
}
136+
137+
step.Output = append(step.Output, e)
138+
return nil
139+
}
140+
141+
// StepState describes the execution of a transform step
142+
type StepState struct {
143+
Name string `json:"name"`
144+
Category string `json:"category"`
145+
Status Status `json:"status"`
146+
StartTime *time.Time `json:"startTime"`
147+
StopTime *time.Time `json:"stopTime"`
148+
Duration int `json:"duration"`
149+
Output []event.Event `json:"output"`
150+
}
151+
152+
// NewStepStateFromEvent constructs StepState from an event
153+
func NewStepStateFromEvent(e event.Event) (*StepState, error) {
154+
if tsl, ok := e.Payload.(event.TransformStepLifecycle); ok {
155+
return &StepState{
156+
Name: tsl.Name,
157+
Category: tsl.Category,
158+
Status: Status(tsl.Status),
159+
}, nil
160+
}
161+
return nil, fmt.Errorf("run step event data must be a transform step lifecycle struct")
162+
}
163+
164+
func toTimePointer(unixnano int64) *time.Time {
165+
// TODO (b5) - we're dropping nanosecond precision here :/
166+
t := time.Unix(unixnano, 0)
167+
return &t
168+
}

‎transform/run/run_test.go‎

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
package run
2+
3+
import (
4+
"fmt"
5+
"testing"
6+
7+
"github.com/google/go-cmp/cmp"
8+
"github.com/qri-io/qri/event"
9+
)
10+
11+
func TestStateAddTransformEvent(t *testing.T) {
12+
runID := NewID()
13+
states := []struct {
14+
e event.Event
15+
r *State
16+
}{
17+
{
18+
event.Event{Type: event.ETTransformStart, Timestamp: 1609460600090, SessionID: runID, Payload: event.TransformLifecycle{StepCount: 4, Status: "running"}},
19+
&State{ID: runID, StartTime: toTimePointer(1609460600090), Status: RSRunning},
20+
},
21+
{
22+
event.Event{Type: event.ETTransformStepStart, Timestamp: 1609460700090, SessionID: runID, Payload: event.TransformStepLifecycle{Name: "setup"}},
23+
&State{ID: runID, StartTime: toTimePointer(1609460600090), Status: RSRunning, Steps: []*StepState{
24+
{Name: "setup", StartTime: toTimePointer(1609460700090), Status: RSRunning},
25+
}},
26+
},
27+
// {
28+
// event.Event{ Type: event.ETVersionPulled, Timestamp: 1609460800090, SessionID: runID, Payload: {"refstring": "rico/presidents@QmFoo", "remote": "https://registy.qri.cloud" }},
29+
// &State{},
30+
// },
31+
{
32+
event.Event{Type: event.ETTransformStepStop, Timestamp: 1609460900090, SessionID: runID, Payload: event.TransformStepLifecycle{Name: "setup", Status: "succeeded"}},
33+
&State{ID: runID, StartTime: toTimePointer(1609460600090), Status: RSRunning, Steps: []*StepState{
34+
{Name: "setup", StartTime: toTimePointer(1609460700090), StopTime: toTimePointer(1609460900090), Duration: 200000000000000, Status: RSSucceeded},
35+
}},
36+
},
37+
{
38+
event.Event{Type: event.ETTransformStepStart, Timestamp: 1609461000090, SessionID: runID, Payload: event.TransformStepLifecycle{Name: "download"}},
39+
&State{ID: runID, StartTime: toTimePointer(1609460600090), Status: RSRunning, Steps: []*StepState{
40+
{Name: "setup", StartTime: toTimePointer(1609460700090), StopTime: toTimePointer(1609460900090), Duration: 200000000000000, Status: RSSucceeded},
41+
{Name: "download", StartTime: toTimePointer(1609461000090), Status: RSRunning},
42+
}},
43+
},
44+
{
45+
event.Event{Type: event.ETTransformPrint, Timestamp: 1609461100090, SessionID: runID, Payload: event.TransformMessage{Msg: "oh hai there"}},
46+
&State{ID: runID, StartTime: toTimePointer(1609460600090), Status: RSRunning, Steps: []*StepState{
47+
{Name: "setup", StartTime: toTimePointer(1609460700090), StopTime: toTimePointer(1609460900090), Duration: 200000000000000, Status: RSSucceeded},
48+
{Name: "download", StartTime: toTimePointer(1609461000090), Status: RSRunning, Output: []event.Event{
49+
{Type: event.ETTransformPrint, Timestamp: 1609461100090, SessionID: runID, Payload: event.TransformMessage{Msg: "oh hai there"}},
50+
}},
51+
}},
52+
},
53+
// {
54+
// event.Event{ Type: event.ETHttpRequestStart, Timestamp: 1609461200090, SessionID: runID, Payload: {"id": runID, "downloadSize": 230409, "method": "Gevent.ET", "url": "https://registy.qri.cloud" }},
55+
// &State{},
56+
// {
57+
// {
58+
// event.Event{ Type: event.ETHttpRequestStop, Timestamp: 1609461300090, SessionID: runID, Payload: {"size": 230409, "method": "Gevent.ET", "url": "https://registy.qri.cloud" }},
59+
// &State{},
60+
// },
61+
{
62+
event.Event{Type: event.ETTransformStepStop, Timestamp: 1609461400090, SessionID: runID, Payload: event.TransformStepLifecycle{Name: "download", Status: "succeeded"}},
63+
&State{ID: runID, StartTime: toTimePointer(1609460600090), Status: RSRunning, Steps: []*StepState{
64+
{Name: "setup", StartTime: toTimePointer(1609460700090), StopTime: toTimePointer(1609460900090), Duration: 200000000000000, Status: RSSucceeded},
65+
{Name: "download", StartTime: toTimePointer(1609461000090), StopTime: toTimePointer(1609461400090), Duration: 400000000000000, Status: RSSucceeded, Output: []event.Event{
66+
{Type: event.ETTransformPrint, Timestamp: 1609461100090, SessionID: runID, Payload: event.TransformMessage{Msg: "oh hai there"}},
67+
}},
68+
}},
69+
},
70+
{
71+
event.Event{Type: event.ETTransformStepStart, Timestamp: 1609461500090, SessionID: runID, Payload: event.TransformStepLifecycle{Name: "transform"}},
72+
&State{ID: runID, StartTime: toTimePointer(1609460600090), Status: RSRunning, Steps: []*StepState{
73+
{Name: "setup", StartTime: toTimePointer(1609460700090), StopTime: toTimePointer(1609460900090), Duration: 200000000000000, Status: RSSucceeded},
74+
{Name: "download", StartTime: toTimePointer(1609461000090), StopTime: toTimePointer(1609461400090), Duration: 400000000000000, Status: RSSucceeded, Output: []event.Event{
75+
{Type: event.ETTransformPrint, Timestamp: 1609461100090, SessionID: runID, Payload: event.TransformMessage{Msg: "oh hai there"}},
76+
}},
77+
{Name: "transform", StartTime: toTimePointer(1609461500090), Status: RSRunning},
78+
}},
79+
},
80+
{
81+
event.Event{Type: event.ETTransformStepStop, Timestamp: 1609461600090, SessionID: runID, Payload: event.TransformStepLifecycle{Name: "transform", Status: "succeeded"}},
82+
&State{ID: runID, StartTime: toTimePointer(1609460600090), Status: RSRunning, Steps: []*StepState{
83+
{Name: "setup", StartTime: toTimePointer(1609460700090), StopTime: toTimePointer(1609460900090), Duration: 200000000000000, Status: RSSucceeded},
84+
{Name: "download", StartTime: toTimePointer(1609461000090), StopTime: toTimePointer(1609461400090), Duration: 400000000000000, Status: RSSucceeded, Output: []event.Event{
85+
{Type: event.ETTransformPrint, Timestamp: 1609461100090, SessionID: runID, Payload: event.TransformMessage{Msg: "oh hai there"}},
86+
}},
87+
{Name: "transform", StartTime: toTimePointer(1609461500090), StopTime: toTimePointer(1609461600090), Duration: 100000000000000, Status: RSSucceeded},
88+
}},
89+
},
90+
{
91+
event.Event{Type: event.ETTransformStop, Timestamp: 1609461900090, SessionID: runID, Payload: event.TransformLifecycle{Status: "failed"}},
92+
&State{ID: runID, StartTime: toTimePointer(1609460600090), StopTime: toTimePointer(1609461900090), Duration: 1300000000000000, Status: RSFailed, Steps: []*StepState{
93+
{Name: "setup", StartTime: toTimePointer(1609460700090), StopTime: toTimePointer(1609460900090), Duration: 200000000000000, Status: RSSucceeded},
94+
{Name: "download", StartTime: toTimePointer(1609461000090), StopTime: toTimePointer(1609461400090), Duration: 400000000000000, Status: RSSucceeded, Output: []event.Event{
95+
{Type: event.ETTransformPrint, Timestamp: 1609461100090, SessionID: runID, Payload: event.TransformMessage{Msg: "oh hai there"}},
96+
}},
97+
{Name: "transform", StartTime: toTimePointer(1609461500090), StopTime: toTimePointer(1609461600090), Duration: 100000000000000, Status: RSSucceeded},
98+
}},
99+
},
100+
}
101+
102+
for i, s := range states {
103+
t.Run(fmt.Sprintf("after_event_%d", i), func(t *testing.T) {
104+
got := NewState(runID)
105+
for j := 0; j <= i; j++ {
106+
if err := got.AddTransformEvent(states[j].e); err != nil {
107+
t.Fatal(err)
108+
}
109+
}
110+
111+
if diff := cmp.Diff(s.r, got); diff != "" {
112+
t.Errorf("result mismatch. (-want +got):\n%s", diff)
113+
}
114+
})
115+
}
116+
}

0 commit comments

Comments
 (0)