Jaculabilis/intake
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Compare commits

base: Jaculabilis:e014b9f9b29ea92ac2bb524a961f5d365e5640fd
Branches Tags
Jaculabilis:master
Jaculabilis:go
Jaculabilis:logging
Jaculabilis:develop
Jaculabilis:react
Jaculabilis:defunct-python-sources
..
compare: Jaculabilis:834eb8ae6277724c145f66d3f00bc6828130eaea
Branches Tags
Jaculabilis:go
Jaculabilis:logging
Jaculabilis:develop
Jaculabilis:master
Jaculabilis:react
Jaculabilis:defunct-python-sources

No commits in common. "e014b9f9b29ea92ac2bb524a961f5d365e5640fd" and "834eb8ae6277724c145f66d3f00bc6828130eaea" have entirely different histories.
e014b9f9b2 ... 834eb8ae62

3 changed files with 181 additions and 214 deletions
Show Stats Expand all files Collapse all files

130
core/execute.go
View File

@ -14,19 +14,50 @@ import (
"time" "time"
) )
func readPipe( func readStdout(
f io.ReadCloser, stdout io.ReadCloser,
send chan []byte, source string,
done chan int, postProcess func(item Item) Item,
items chan Item,
cparse chan bool,
) { ) {
defer func() { var item Item
done <- 0 parseError := false
}() scanout := bufio.NewScanner(stdout)
scanner := bufio.NewScanner(f) for scanout.Scan() {
for scanner.Scan() { data := scanout.Bytes()
data := scanner.Bytes() err := json.Unmarshal(data, &item)
send <- data if err != nil || item.Id == "" {
log.Printf("[%s: stdout] %s\n", source, strings.TrimSpace(string(data)))
parseError = true
} else {
if postProcess != nil {
item = postProcess(item)
}
item.Active = true // These fields aren't up to
item.Created = 0 // the action to set and
item.Source = source // shouldn't be overrideable
log.Printf("[%s: item] %s\n", source, item.Id)
items <- item
}
} }
// Only send the parsing result at the end, to block main until stdout is drained
cparse <- parseError
close(items)
}
func readStderr(stderr io.ReadCloser, source string, done chan bool) {
scanerr := bufio.NewScanner(stderr)
for scanerr.Scan() {
text := strings.TrimSpace(scanerr.Text())
log.Printf("[%s: stderr] %s\n", source, text)
}
done <- true
}
func writeStdin(stdin io.WriteCloser, text string) {
defer stdin.Close()
io.WriteString(stdin, text)
} }
func Execute( func Execute(
@ -37,11 +68,7 @@ func Execute(
input string, input string,
timeout time.Duration, timeout time.Duration,
postProcess func(item Item) Item, postProcess func(item Item) Item,
) ( ) ([]Item, []byte, error) {
items []Item,
newState []byte,
err error,
) {
log.Printf("executing %v", argv) log.Printf("executing %v", argv)
if len(argv) == 0 { if len(argv) == 0 {
@ -88,14 +115,22 @@ func Execute(
return nil, nil, err return nil, nil, err
} }
cout := make(chan []byte) cout := make(chan Item)
cerr := make(chan []byte) cparse := make(chan bool)
coutDone := make(chan int) cerr := make(chan bool)
cerrDone := make(chan int)
// Sink routine for items produced
var items []Item
go func() {
for item := range cout {
items = append(items, item)
}
}()
// Routines handling the process i/o // Routines handling the process i/o
go readPipe(stdout, cout, coutDone) go writeStdin(stdin, input)
go readPipe(stderr, cerr, cerrDone) go readStdout(stdout, source, postProcess, cout, cparse)
go readStderr(stderr, source, cerr)
// Kick off the command // Kick off the command
err = cmd.Start() err = cmd.Start()
@ -103,50 +138,9 @@ func Execute(
return nil, nil, err return nil, nil, err
} }
// Write any input to stdin and close it // Block until std{out,err} close
io.WriteString(stdin, input) <-cerr
stdin.Close() parseError := <-cparse
// Collect outputs until std{out,err} close
parseError := false
stdoutDone := false
stderrDone := false
monitor:
for {
select {
case data := <-cout:
var item Item
err := json.Unmarshal(data, &item)
if err != nil || item.Id == "" {
log.Printf("[%s: stdout] %s", source, strings.TrimSpace(string(data)))
parseError = true
} else {
if postProcess != nil {
item = postProcess(item)
}
item.Active = true // These fields aren't up to
item.Created = 0 // the action to set and
item.Source = source // shouldn't be overrideable
log.Printf("[%s: item] %s\n", source, item.Id)
items = append(items, item)
}
case data := <-cerr:
log.Printf("[%s: stderr] %s\n", source, data)
case <-coutDone:
stdoutDone = true
if stdoutDone && stderrDone {
break monitor
}
case <-cerrDone:
stderrDone = true
if stdoutDone && stderrDone {
break monitor
}
}
}
err = cmd.Wait() err = cmd.Wait()
if ctx.Err() == context.DeadlineExceeded { if ctx.Err() == context.DeadlineExceeded {
@ -165,12 +159,12 @@ monitor:
return nil, nil, errors.New("invalid JSON") return nil, nil, errors.New("invalid JSON")
} }
newState, err = os.ReadFile(stateFile.Name()) newState, err := os.ReadFile(stateFile.Name())
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("error: failed to read state file: %v", err) return nil, nil, fmt.Errorf("error: failed to read state file: %v", err)
} }
return return items, newState, nil
} }
// Execute an action that takes an item as input and returns the item modified. // Execute an action that takes an item as input and returns the item modified.

263
core/execute_test.go
View File

@ -6,19 +6,19 @@ import (
) )
func TestExecute(t *testing.T) { func TestExecute(t *testing.T) {
assertLen := func(t *testing.T, items []Item, length int) { assertLen := func(items []Item, length int) {
t.Helper() t.Helper()
if len(items) != length { if len(items) != length {
t.Fatalf("Expected %d items, got %d", length, len(items)) t.Fatalf("Expected %d items, got %d", length, len(items))
} }
} }
assertNil := func(t *testing.T, err error) { assertNil := func(err error) {
t.Helper() t.Helper()
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
} }
assertNotNil := func(t *testing.T, err error) { assertNotNil := func(err error) {
t.Helper() t.Helper()
if err == nil { if err == nil {
t.Fatal("expected err") t.Fatal("expected err")
@ -29,165 +29,138 @@ func TestExecute(t *testing.T) {
return item, err return item, err
} }
t.Run("Noop", func(t *testing.T) { res, err := execute([]string{"true"})
res, err := execute([]string{"true"}) assertNil(err)
assertNil(t, err) assertLen(res, 0)
assertLen(t, res, 0)
})
t.Run("ExitWithErrorCode", func(t *testing.T) { // Exit with error code
res, err := execute([]string{"false"}) res, err = execute([]string{"false"})
assertNotNil(t, err) assertNotNil(err)
assertLen(t, res, 0) assertLen(res, 0)
})
t.Run("ExitWithSpecificErrorCode", func(t *testing.T) { res, err = execute([]string{"sh", "-c", "exit 22"})
res, err := execute([]string{"sh", "-c", "exit 22"}) assertNotNil(err)
assertNotNil(t, err) assertLen(res, 0)
assertLen(t, res, 0)
})
t.Run("Timeout", func(t *testing.T) { // Timeout
res, _, err := Execute("_", []string{"sleep", "10"}, nil, nil, "", time.Millisecond, nil) res, _, err = Execute("_", []string{"sleep", "10"}, nil, nil, "", time.Millisecond, nil)
assertNotNil(t, err) assertNotNil(err)
assertLen(t, res, 0) assertLen(res, 0)
})
t.Run("ReturningItems", func(t *testing.T) { // Returning items
res, err := execute([]string{"jq", "-cn", `{id: "foo"}`}) res, err = execute([]string{"jq", "-cn", `{id: "foo"}`})
assertNil(t, err) assertNil(err)
assertLen(t, res, 1) assertLen(res, 1)
if res[0].Id != "foo" { if res[0].Id != "foo" {
t.Fatal("jq -cn test failed") t.Fatal("jq -cn test failed")
} }
})
t.Run("ReadFromStdin", func(t *testing.T) { // Read from stdin
res, _, err := Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, nil, "bar", time.Minute, nil) res, _, err = Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, nil, "bar", time.Minute, nil)
assertNil(t, err) assertNil(err)
assertLen(t, res, 1) assertLen(res, 1)
if res[0].Id != "bar" { if res[0].Id != "bar" {
t.Fatal("jq -cR test failed") t.Fatal("jq -cR test failed")
} }
})
t.Run("SetEnv", func(t *testing.T) { // Set env
res, _, err := Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, nil, "", time.Minute, nil) res, _, err = Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, nil, "", time.Minute, nil)
assertNil(t, err) assertNil(err)
assertLen(t, res, 1) assertLen(res, 1)
if res[0].Id != "baz" { if res[0].Id != "baz" {
t.Fatal("jq -cn env test failed") t.Fatal("jq -cn env test failed")
} }
})
t.Run("StderrLogging", func(t *testing.T) {
res, err := execute([]string{"sh", "-c", `echo 1>&2 Hello; jq -cn '{id: "box"}'; echo 1>&2 World`})
assertNil(t, err)
assertLen(t, res, 1)
if res[0].Id != "box" {
t.Fatal("stderr test failed")
}
})
t.Run("UnknownFieldIgnored", func(t *testing.T) { // With logging on stderr
res, err := execute([]string{"jq", "-cn", `{id: "test", unknownField: "what is this"}`}) res, err = execute([]string{"sh", "-c", `echo 1>&2 Hello; jq -cn '{id: "box"}'; echo 1>&2 World`})
assertNil(t, err) assertNil(err)
assertLen(t, res, 1) assertLen(res, 1)
}) if res[0].Id != "box" {
t.Fatal("stderr test failed")
}
t.Run("IncorrectIdType", func(t *testing.T) { // Unsupported item field is silently discarded
res, err := execute([]string{"jq", "-cn", `{id: ["list"]}`}) res, err = execute([]string{"jq", "-cn", `{id: "test", unknownField: "what is this"}`})
assertNotNil(t, err) assertNil(err)
assertLen(t, res, 0) assertLen(res, 1)
})
t.Run("IncorrectTimeType", func(t *testing.T) { // Field with incorrect type fails
res, err := execute([]string{"jq", "-cn", `{id: "test", time: "0"}`}) res, err = execute([]string{"jq", "-cn", `{id: ["list"]}`})
assertNotNil(t, err) assertNotNil(err)
assertLen(t, res, 0) assertLen(res, 0)
})
t.Run("NullId", func(t *testing.T) { res, err = execute([]string{"jq", "-cn", `{id: "test", time: "0"}`})
res, err := execute([]string{"jq", "-cn", `{id: null}`}) assertNotNil(err)
assertNotNil(t, err) assertLen(res, 0)
assertLen(t, res, 0)
})
// TODO maybe this *should* be an Execute error, via a map[string]bool res, err = execute([]string{"jq", "-cn", `{id: null}`})
t.Run("DuplicateItemIds", func(t *testing.T) { assertNotNil(err)
res, err := execute([]string{"jq", "-cn", `["a", "a"] | .[] | {id: .}`}) assertLen(res, 0)
assertNil(t, err)
assertLen(t, res, 2)
})
t.Run("ActionNullValueOk", func(t *testing.T) { // Items with duplicate ids is not a fetch error, but it will fail to update
res, err := execute([]string{"jq", "-cn", `{id: "test", action: {"hello": null}}`}) res, err = execute([]string{"jq", "-cn", `["a", "a"] | .[] | {id: .}`})
assertNil(t, err) assertNil(err)
assertLen(t, res, 1) assertLen(res, 2)
if res[0].Action["hello"] == nil {
t.Fatal("missing hello action")
}
if res[0].Action["goodbye"] != nil {
t.Fatal("nonexistent action should key to nil in Action")
}
})
t.Run("ActionEmptyStringOk", func(t *testing.T) { // Action keys are detected even with empty values
res, err := execute([]string{"jq", "-cn", `{id: "test", action: {"hello": ""}}`}) res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": null}}`})
assertNil(t, err) assertNil(err)
assertLen(t, res, 1) assertLen(res, 1)
if res[0].Action["hello"] == nil { if res[0].Action["hello"] == nil {
t.Fatal("missing hello action") t.Fatal("missing hello action")
} }
}) if res[0].Action["goodbye"] != nil {
t.Fatal("nonexistent action should key to nil in Action")
}
t.Run("ActionEmptyArrayOk", func(t *testing.T) { res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": ""}}`})
res, err := execute([]string{"jq", "-cn", `{id: "test", action: {"hello": []}}`}) assertNil(err)
assertNil(t, err) assertLen(res, 1)
assertLen(t, res, 1) if res[0].Action["hello"] == nil {
if res[0].Action["hello"] == nil { t.Fatal("missing hello action")
t.Fatal("missing hello action") }
}
})
t.Run("ActionEmptyObjectOk", func(t *testing.T) { res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": []}}`})
res, err := execute([]string{"jq", "-cn", `{id: "test", action: {"hello": {}}}`}) assertNil(err)
assertNil(t, err) assertLen(res, 1)
assertLen(t, res, 1) if res[0].Action["hello"] == nil {
if res[0].Action["hello"] == nil { t.Fatal("missing hello action")
t.Fatal("missing hello action") }
}
})
t.Run("ReadState", func(t *testing.T) { res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": {}}}`})
argv := []string{"sh", "-c", `cat $STATE_PATH | jq -cR '{id: "greeting", title: .} | .title = "Hello " + .title'`} assertNil(err)
res, _, err := Execute("_", argv, nil, []byte("world"), "", time.Minute, nil) assertLen(res, 1)
assertNil(t, err) if res[0].Action["hello"] == nil {
assertLen(t, res, 1) t.Fatal("missing hello action")
if res[0].Title != "Hello world" { }
t.Fatalf("expected 'Hello world' from read state, got '%s'", res[0].Title)
}
})
t.Run("WriteState", func(t *testing.T) { // Read state
argv := []string{"sh", "-c", `printf "Hello world" > $STATE_PATH; jq -cn '{id: "test"}'`} argv := []string{"sh", "-c", `cat $STATE_PATH | jq -cR '{id: "greeting", title: .} | .title = "Hello " + .title'`}
res, newState, err := Execute("_", argv, nil, nil, "", time.Minute, nil) res, _, err = Execute("_", argv, nil, []byte("world"), "", time.Minute, nil)
assertNil(t, err) assertNil(err)
assertLen(t, res, 1) assertLen(res, 1)
if string(newState) != "Hello world" { if res[0].Title != "Hello world" {
t.Fatalf("expected 'Hello world' from write state, got %s", string(newState)) t.Fatalf("expected 'Hello world' from read state, got '%s'", res[0].Title)
} }
})
t.Run("PostprocessSetTtl", func(t *testing.T) { // Write state
argv := []string{"jq", "-cn", `{id: "foo"}`} argv = []string{"sh", "-c", `printf "Hello world" > $STATE_PATH; jq -cn '{id: "test"}'`}
res, _, err := Execute("_", argv, nil, nil, "", time.Minute, func(item Item) Item { res, newState, err := Execute("_", argv, nil, nil, "", time.Minute, nil)
item.Ttl = 123456 assertNil(err)
return item assertLen(res, 1)
}) if string(newState) != "Hello world" {
assertNil(t, err) t.Fatalf("expected 'Hello world' from write state, got %s", string(newState))
assertLen(t, res, 1) }
if res[0].Ttl != 123456 {
t.Fatalf("expected ttl to be set to 123456, got %d", res[0].Ttl) // Postprocessing function
} argv = []string{"jq", "-cn", `{id: "foo"}`}
res, _, err = Execute("_", argv, nil, nil, "", time.Minute, func(item Item) Item {
item.Ttl = 123456
return item
}) })
assertNil(err)
assertLen(res, 1)
if res[0].Ttl != 123456 {
t.Fatalf("expected ttl to be set to 123456, got %d", res[0].Ttl)
}
} }

2
core/source_test.go
View File

@ -183,7 +183,7 @@ func TestOnCreateAction(t *testing.T) {
items := execute([]string{"jq", "-cn", `{id: "one"}`}) items := execute([]string{"jq", "-cn", `{id: "one"}`})
add, _, err := update(items) add, _, err := update(items)
if add != 1 || err != nil { if add != 1 || err != nil {
t.Fatalf("failed update with noop oncreate: %v", err) t.Fatal("failed update with noop oncreate")
} }
updated := getItem("one") updated := getItem("one")
updated.Created = 0 // zero out for comparison with pre-insert item updated.Created = 0 // zero out for comparison with pre-insert item