DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Improve workflow stacktrace logging to reduce pressure for iterators

Browse Source
This commit is contained in:
Tomaž Jerman
2024-02-01 15:17:11 +01:00
parent 280ae83376
commit 74a54da6f1
2 changed files with 195 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/automation/service/session_test.go
+148 -1
View File
@@ -40,5 +40,152 @@ func TestSession_Start(t *testing.T) {
g.AddStep(wfexec.NewGenericStep(nil), s)
_, _, err = ses.Start(ctx, g, types.SessionStartParams{StepID: 42, Invoker: auth.Anonymous()})
req.EqualError(err, "cannot start workflow on a step with parents")
}
func TestSessionStackTraces(t *testing.T) {
stepID := uint64(0)
getStepID := func() uint64 {
stepID++
return uint64(stepID)
}
iter1 := &wfexec.Frame{StepID: getStepID(), ParentID: 0, Action: "iterator initialized"}
iter1stepExpr := &wfexec.Frame{StepID: getStepID(), ParentID: iter1.StepID, Action: ""}
iter1stepCnt := &wfexec.Frame{StepID: getStepID(), ParentID: iter1stepExpr.StepID, Action: "loop continue"}
iter2 := &wfexec.Frame{StepID: getStepID(), ParentID: iter1.StepID, Action: "iterator initialized"}
iter2stepExpr := &wfexec.Frame{StepID: getStepID(), ParentID: iter2.StepID, Action: ""}
iter2stepCnt := &wfexec.Frame{StepID: getStepID(), ParentID: iter2stepExpr.StepID, Action: "loop continue"}
rando1 := &wfexec.Frame{StepID: getStepID(), ParentID: iter2.StepID, Action: ""}
rando2 := &wfexec.Frame{StepID: getStepID(), ParentID: iter2.StepID, Action: ""}
trace := []*wfexec.Frame{
// first loop
iter1,
iter1stepExpr,
iter1stepCnt,
iter1,
iter1stepExpr,
iter1stepCnt,
iter1,
// second loop
iter2,
iter2stepExpr,
iter2stepCnt,
iter2,
iter2stepExpr,
iter2stepCnt,
iter2,
rando1,
rando2,
}
ses := &types.Session{}
for _, f := range trace {
ses.AppendRuntimeStacktrace(f)
}
require.Len(t, ses.RuntimeStacktrace, 10)
require.Equal(t, iter1.StepID, ses.RuntimeStacktrace[0].StepID)
require.Equal(t, iter1stepExpr.StepID, ses.RuntimeStacktrace[1].StepID)
require.Equal(t, iter1stepCnt.StepID, ses.RuntimeStacktrace[2].StepID)
require.Equal(t, iter1.StepID, ses.RuntimeStacktrace[3].StepID)
require.Equal(t, iter2.StepID, ses.RuntimeStacktrace[4].StepID)
require.Equal(t, iter2stepExpr.StepID, ses.RuntimeStacktrace[5].StepID)
require.Equal(t, iter2stepCnt.StepID, ses.RuntimeStacktrace[6].StepID)
require.Equal(t, iter2.StepID, ses.RuntimeStacktrace[7].StepID)
require.Equal(t, rando1.StepID, ses.RuntimeStacktrace[8].StepID)
require.Equal(t, rando2.StepID, ses.RuntimeStacktrace[9].StepID)
}
// Before:
// goos: darwin
// goarch: arm64
// pkg: github.com/cortezaproject/corteza/server/automation/service
// BenchmarkSessionStackTraces_1000-12 42758 28594 ns/op 87288 B/op 15 allocs/op
// BenchmarkSessionStackTraces_10000-12 3430 348308 ns/op 1160442 B/op 23 allocs/op
// BenchmarkSessionStackTraces_100000-12 229 5299188 ns/op 13317368 B/op 33 allocs/op
// BenchmarkSessionStackTraces_1000000-12 19 52896750 ns/op 128431352 B/op 43 allocs/op
// BenchmarkSessionStackTraces_10000000-12 3 436311292 ns/op 1202337018 B/op 53 allocs/op
// After:
// goos: darwin
// goarch: arm64
// pkg: github.com/cortezaproject/corteza/server/automation/service
// BenchmarkSessionStackTraces_1000-12 40142 29860 ns/op 120 B/op 4 allocs/op
// BenchmarkSessionStackTraces_10000-12 4387 264930 ns/op 120 B/op 4 allocs/op
// BenchmarkSessionStackTraces_100000-12 447 2735284 ns/op 120 B/op 4 allocs/op
// BenchmarkSessionStackTraces_1000000-12 42 26596636 ns/op 120 B/op 4 allocs/op
// BenchmarkSessionStackTraces_10000000-12 4 265553667 ns/op 120 B/op 4 allocs/op
func benchmarkSessionStackTraces(b *testing.B, iters int) {
stepID := uint64(0)
getStepID := func() uint64 {
stepID++
return uint64(stepID)
}
iter := &wfexec.Frame{
StepID: getStepID(),
Action: "iterator initialized",
}
loop := []*wfexec.Frame{iter}
loop = append(loop, &wfexec.Frame{
StepID: getStepID(),
ParentID: iter.ParentID,
Action: "",
})
loop = append(loop, &wfexec.Frame{
StepID: getStepID(),
ParentID: loop[len(loop)-1].StepID,
Action: "loop continue",
})
frames := []*wfexec.Frame{}
for x := 0; x < iters; x++ {
frames = append(frames, loop...)
}
frames = append(frames, iter)
frames = append(frames, &wfexec.Frame{
StepID: getStepID(),
ParentID: iter.ParentID,
Action: "termination",
})
b.StopTimer()
b.ResetTimer()
for n := 0; n < b.N; n++ {
ses := &types.Session{}
b.StartTimer()
for _, f := range frames {
ses.AppendRuntimeStacktrace(f)
}
b.StopTimer()
}
}
func BenchmarkSessionStackTraces_1000(b *testing.B) {
benchmarkSessionStackTraces(b, 1000)
}
func BenchmarkSessionStackTraces_10000(b *testing.B) {
benchmarkSessionStackTraces(b, 10000)
}
func BenchmarkSessionStackTraces_100000(b *testing.B) {
benchmarkSessionStackTraces(b, 100000)
}
func BenchmarkSessionStackTraces_1000000(b *testing.B) {
benchmarkSessionStackTraces(b, 1000000)
}
func BenchmarkSessionStackTraces_10000000(b *testing.B) {
benchmarkSessionStackTraces(b, 10000000)
}
server/automation/types/session.go
+47 -5
View File
@@ -184,13 +184,55 @@ func (s *Session) Apply(ssp SessionStartParams) {
}
}
// AppendRuntimeStacktrace adds a new frame to the runtime stacktrace
//
// This does have some smartness
// If the workflow uses longer iterators, the memory pressure got too high.
// To counter this, only the frames of the last iteration are preserved to
// better match what programming languages do.
func (s *Session) AppendRuntimeStacktrace(frame *wfexec.Frame) {
if !s.runtimeOpts.disableStacktrace {
s.l.RLock()
defer s.l.RUnlock()
s.RuntimeStacktrace = append(s.RuntimeStacktrace, frame)
if s.runtimeOpts.disableStacktrace {
return
}
s.l.RLock()
defer s.l.RUnlock()
// The only way to get to the same stepID is when we're in a loop
// Find where the first frame of the iterator is and slice the stack.
if frame.Action != "iterator initialized" && s.hasDuplicate(frame.StepID) {
var (
i = 0
f *wfexec.Frame
)
for i, f = range s.RuntimeStacktrace {
if f.StepID == frame.StepID {
break
}
}
// @todo this might cause a memory leak; investigate further
s.RuntimeStacktrace = s.RuntimeStacktrace[0:i]
}
// Push to the newly done trace
s.RuntimeStacktrace = append(s.RuntimeStacktrace, frame)
}
// @todo potentially optimize this; it'll probably be fine for now but
// might degrade performance for larger workflows.
// To investigate and potentially optimize it.
func (s *Session) hasDuplicate(stepID uint64) bool {
s.l.RLock()
defer s.l.RUnlock()
for _, f := range s.RuntimeStacktrace {
if f.StepID == stepID {
return true
}
}
return false
}
func (s *Session) CopyRuntimeStacktrace() {