mirror of
https://github.com/OliveTin/OliveTin
synced 2026-08-26 12:36:30 +00:00
587 lines
15 KiB
Go
587 lines
15 KiB
Go
package executor
|
|
|
|
import (
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/OliveTin/OliveTin/internal/auth"
|
|
config "github.com/OliveTin/OliveTin/internal/config"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
func testGroupExecutor(actions []*config.Action, groups map[string]*config.ActionGroup) (*Executor, *config.Config) {
|
|
cfg := config.DefaultConfig()
|
|
cfg.ActionGroups = groups
|
|
cfg.Actions = actions
|
|
cfg.Sanitize()
|
|
|
|
e := DefaultExecutor(cfg)
|
|
e.RebuildActionMap()
|
|
|
|
return e, cfg
|
|
}
|
|
|
|
func TestGroupConcurrencyQueuesSecondAction(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
slowAction := &config.Action{
|
|
Title: "Unity Job 1",
|
|
Shell: "sleep 2",
|
|
Groups: []string{"unity"},
|
|
}
|
|
fastAction := &config.Action{
|
|
Title: "Unity Job 2",
|
|
Shell: "echo queued-run",
|
|
Groups: []string{"unity"},
|
|
}
|
|
|
|
e, cfg := testGroupExecutor(
|
|
[]*config.Action{slowAction, fastAction},
|
|
map[string]*config.ActionGroup{
|
|
"unity": {MaxConcurrent: 1},
|
|
},
|
|
)
|
|
|
|
binding1 := e.FindBindingWithNoEntity(slowAction)
|
|
binding2 := e.FindBindingWithNoEntity(fastAction)
|
|
require.NotNil(t, binding1)
|
|
require.NotNil(t, binding2)
|
|
|
|
wg1, tracking1 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding1,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
waitUntilExecutionStarted(t, e, tracking1)
|
|
|
|
wg2, tracking2 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding2,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
require.Eventually(t, func() bool {
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
return ok && snapshot.Queued
|
|
}, time.Second, 10*time.Millisecond)
|
|
|
|
wg1.Wait()
|
|
wg2.Wait()
|
|
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
require.True(t, ok)
|
|
assert.False(t, snapshot.Queued)
|
|
assert.False(t, snapshot.Blocked)
|
|
assert.Equal(t, int32(0), snapshot.ExitCode)
|
|
assert.Contains(t, snapshot.Output, "queued-run")
|
|
}
|
|
|
|
func TestQueuedActionNotifiesWhenExecutionBegins(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
slowAction := &config.Action{
|
|
Title: "Hold group",
|
|
Shell: "sleep 1",
|
|
Groups: []string{"unity"},
|
|
}
|
|
queuedAction := &config.Action{
|
|
Title: "Queued job",
|
|
Shell: "echo queued-run",
|
|
Groups: []string{"unity"},
|
|
}
|
|
|
|
e, cfg := testGroupExecutor(
|
|
[]*config.Action{slowAction, queuedAction},
|
|
map[string]*config.ActionGroup{
|
|
"unity": {MaxConcurrent: 1},
|
|
},
|
|
)
|
|
|
|
notifications := make(chan startedNotification, 8)
|
|
e.AddListener(&executionStartedCollector{ch: notifications})
|
|
|
|
wg1, tracking1 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: e.FindBindingWithNoEntity(slowAction),
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
waitUntilExecutionStarted(t, e, tracking1)
|
|
|
|
wg2, tracking2 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: e.FindBindingWithNoEntity(queuedAction),
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
require.Eventually(t, func() bool {
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
return ok && snapshot.Queued
|
|
}, time.Second, 10*time.Millisecond)
|
|
|
|
wg1.Wait()
|
|
wg2.Wait()
|
|
|
|
sawQueuedStart, sawRunningStart := collectQueuedStartNotifications(notifications, tracking2)
|
|
|
|
assert.True(t, sawQueuedStart, "queued action should notify when queued")
|
|
assert.True(t, sawRunningStart, "queued action should notify again when execution begins")
|
|
}
|
|
|
|
func isQueuedStartNotification(notification startedNotification, trackingID string) bool {
|
|
return notification.trackingID == trackingID && notification.queued && !notification.started
|
|
}
|
|
|
|
func isRunningStartNotification(notification startedNotification, trackingID string) bool {
|
|
return notification.trackingID == trackingID && !notification.queued && notification.started
|
|
}
|
|
|
|
func collectQueuedStartNotifications(notifications <-chan startedNotification, trackingID string) (sawQueuedStart, sawRunningStart bool) {
|
|
for len(notifications) > 0 {
|
|
notification := <-notifications
|
|
if isQueuedStartNotification(notification, trackingID) {
|
|
sawQueuedStart = true
|
|
}
|
|
if isRunningStartNotification(notification, trackingID) {
|
|
sawRunningStart = true
|
|
}
|
|
}
|
|
return sawQueuedStart, sawRunningStart
|
|
}
|
|
|
|
func TestDifferentGroupsRunConcurrently(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
actionA := &config.Action{
|
|
Title: "Group A Job",
|
|
Shell: "sleep 1",
|
|
Groups: []string{"groupA"},
|
|
}
|
|
actionB := &config.Action{
|
|
Title: "Group B Job",
|
|
Shell: "echo group-b",
|
|
Groups: []string{"groupB"},
|
|
}
|
|
|
|
e, cfg := testGroupExecutor(
|
|
[]*config.Action{actionA, actionB},
|
|
map[string]*config.ActionGroup{
|
|
"groupA": {MaxConcurrent: 1},
|
|
"groupB": {MaxConcurrent: 1},
|
|
},
|
|
)
|
|
|
|
wg1, tracking1 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: e.FindBindingWithNoEntity(actionA),
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
waitUntilExecutionStarted(t, e, tracking1)
|
|
|
|
wg2, tracking2 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: e.FindBindingWithNoEntity(actionB),
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
require.Eventually(t, func() bool {
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
return ok && snapshot.ExecutionFinished && !snapshot.Queued
|
|
}, 2*time.Second, 20*time.Millisecond)
|
|
|
|
wg1.Wait()
|
|
wg2.Wait()
|
|
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
require.True(t, ok)
|
|
assert.Contains(t, snapshot.Output, "group-b")
|
|
}
|
|
|
|
func TestPerActionConcurrencyStillBlocksWithoutQueue(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
action := &config.Action{
|
|
Title: "Single binding",
|
|
Shell: "sleep 1",
|
|
MaxConcurrent: 1,
|
|
}
|
|
|
|
e, cfg := testGroupExecutor([]*config.Action{action}, nil)
|
|
binding := e.FindBindingWithNoEntity(action)
|
|
|
|
wg1, tracking1 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
waitUntilExecutionStarted(t, e, tracking1)
|
|
|
|
wg2, tracking2 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
wg1.Wait()
|
|
wg2.Wait()
|
|
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
require.True(t, ok)
|
|
assert.True(t, snapshot.Blocked)
|
|
assert.False(t, snapshot.Queued)
|
|
}
|
|
|
|
func TestGroupedSameBindingQueuesWhenGroupFull(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
action := &config.Action{
|
|
Title: "Single binding grouped",
|
|
Shell: "sleep 1",
|
|
Groups: []string{"unity"},
|
|
}
|
|
|
|
e, cfg := testGroupExecutor(
|
|
[]*config.Action{action},
|
|
map[string]*config.ActionGroup{
|
|
"unity": {MaxConcurrent: 1, QueueSize: 5},
|
|
},
|
|
)
|
|
binding := e.FindBindingWithNoEntity(action)
|
|
|
|
wg1, tracking1 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
waitUntilExecutionStarted(t, e, tracking1)
|
|
|
|
wg2, tracking2 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
require.Eventually(t, func() bool {
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
return ok && snapshot.Queued
|
|
}, time.Second, 10*time.Millisecond)
|
|
|
|
wg1.Wait()
|
|
wg2.Wait()
|
|
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
require.True(t, ok)
|
|
assert.False(t, snapshot.Blocked)
|
|
}
|
|
|
|
func TestGroupAllowsTwoConcurrentSameBinding(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
action := &config.Action{
|
|
Title: "Long running action",
|
|
Shell: "sleep 1",
|
|
Groups: []string{"con2queue10"},
|
|
}
|
|
|
|
e, cfg := testGroupExecutor(
|
|
[]*config.Action{action},
|
|
map[string]*config.ActionGroup{
|
|
"con2queue10": {MaxConcurrent: 2, QueueSize: 10},
|
|
},
|
|
)
|
|
binding := e.FindBindingWithNoEntity(action)
|
|
|
|
wg1, tracking1 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
waitUntilExecutionStarted(t, e, tracking1)
|
|
|
|
wg2, tracking2 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
require.Eventually(t, func() bool {
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
return ok && snapshot.ExecutionStarted && !snapshot.Queued && !snapshot.Blocked
|
|
}, 2*time.Second, 10*time.Millisecond)
|
|
|
|
wg1.Wait()
|
|
wg2.Wait()
|
|
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
require.True(t, ok)
|
|
assert.False(t, snapshot.Blocked)
|
|
assert.False(t, snapshot.Queued)
|
|
}
|
|
|
|
func TestGroupQueuesThirdAndBlocksWhenQueueFull(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
action := &config.Action{
|
|
Title: "Long running action",
|
|
Shell: "sleep 1",
|
|
Groups: []string{"con2queue10"},
|
|
}
|
|
|
|
e, cfg := testGroupExecutor(
|
|
[]*config.Action{action},
|
|
map[string]*config.ActionGroup{
|
|
"con2queue10": {MaxConcurrent: 2, QueueSize: 2},
|
|
},
|
|
)
|
|
binding := e.FindBindingWithNoEntity(action)
|
|
|
|
wg1, tracking1 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
waitUntilExecutionStarted(t, e, tracking1)
|
|
|
|
wg2, tracking2 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
waitUntilExecutionStarted(t, e, tracking2)
|
|
|
|
trackings := []string{tracking1, tracking2}
|
|
waitGroups := []*sync.WaitGroup{wg1, wg2}
|
|
|
|
for idx := 0; idx < 3; idx++ {
|
|
wg, tracking := e.ExecRequest(&ExecutionRequest{
|
|
Binding: binding,
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
trackings = append(trackings, tracking)
|
|
waitGroups = append(waitGroups, wg)
|
|
}
|
|
|
|
require.Eventually(t, func() bool {
|
|
return groupExecutionDistributionMatches(e, trackings, 2, 2, 1)
|
|
}, 2*time.Second, 20*time.Millisecond)
|
|
|
|
for _, wg := range waitGroups {
|
|
wg.Wait()
|
|
}
|
|
}
|
|
|
|
func waitUntilExecutionStarted(t *testing.T, e *Executor, trackingID string) {
|
|
t.Helper()
|
|
|
|
require.Eventually(t, func() bool {
|
|
snapshot, ok := e.SnapshotLog(trackingID)
|
|
return ok && snapshot.ExecutionStarted
|
|
}, 2*time.Second, 10*time.Millisecond)
|
|
}
|
|
|
|
type executionStartedCollector struct {
|
|
ch chan startedNotification
|
|
}
|
|
|
|
type startedNotification struct {
|
|
trackingID string
|
|
started bool
|
|
queued bool
|
|
}
|
|
|
|
func (c *executionStartedCollector) OnExecutionStarted(entry *InternalLogEntry) {
|
|
c.ch <- startedNotification{
|
|
trackingID: entry.ExecutionTrackingID,
|
|
started: entry.ExecutionStarted,
|
|
queued: entry.Queued,
|
|
}
|
|
}
|
|
|
|
func (c *executionStartedCollector) OnExecutionFinished(_ *InternalLogEntry) {}
|
|
|
|
func (c *executionStartedCollector) OnOutputChunk(_ []byte, _ string) {}
|
|
|
|
func (c *executionStartedCollector) OnActionMapRebuilt() {}
|
|
|
|
func assertWaitGroupPending(t *testing.T, wg *sync.WaitGroup) {
|
|
t.Helper()
|
|
|
|
done := make(chan struct{})
|
|
|
|
go func() {
|
|
wg.Wait()
|
|
close(done)
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
t.Fatal("wait group completed before queued execution finished")
|
|
case <-time.After(100 * time.Millisecond):
|
|
}
|
|
}
|
|
|
|
func assertWaitGroupCompletes(t *testing.T, wg *sync.WaitGroup) {
|
|
t.Helper()
|
|
|
|
done := make(chan struct{})
|
|
|
|
go func() {
|
|
wg.Wait()
|
|
close(done)
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("wait group did not complete after queue drained")
|
|
}
|
|
}
|
|
|
|
func TestStartActionAndWaitWaitsForQueuedExecution(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
first := &config.Action{
|
|
Title: "Hold group",
|
|
Shell: "sleep 1",
|
|
Groups: []string{"unity"},
|
|
}
|
|
second := &config.Action{
|
|
Title: "Wait in queue",
|
|
Shell: "echo waited",
|
|
Groups: []string{"unity"},
|
|
}
|
|
|
|
e, cfg := testGroupExecutor(
|
|
[]*config.Action{first, second},
|
|
map[string]*config.ActionGroup{
|
|
"unity": {MaxConcurrent: 1},
|
|
},
|
|
)
|
|
|
|
wg1, tracking1 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: e.FindBindingWithNoEntity(first),
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
waitUntilExecutionStarted(t, e, tracking1)
|
|
|
|
wg2, tracking2 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: e.FindBindingWithNoEntity(second),
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
assertWaitGroupPending(t, wg2)
|
|
|
|
wg1.Wait()
|
|
|
|
assertWaitGroupCompletes(t, wg2)
|
|
|
|
snapshot, ok := e.SnapshotLog(tracking2)
|
|
require.True(t, ok)
|
|
assert.Contains(t, snapshot.Output, "waited")
|
|
}
|
|
|
|
func TestGroupQueueBlocksWhenQueueFull(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
actions := []*config.Action{
|
|
{Title: "Hold 1", Shell: "sleep 1", Groups: []string{"unity"}},
|
|
{Title: "Hold 2", Shell: "sleep 1", Groups: []string{"unity"}},
|
|
{Title: "Hold 3", Shell: "sleep 1", Groups: []string{"unity"}},
|
|
{Title: "Hold 4", Shell: "sleep 1", Groups: []string{"unity"}},
|
|
}
|
|
|
|
e, cfg := testGroupExecutor(
|
|
actions,
|
|
map[string]*config.ActionGroup{
|
|
"unity": {MaxConcurrent: 1, QueueSize: 2},
|
|
},
|
|
)
|
|
|
|
wg1, tracking1 := e.ExecRequest(&ExecutionRequest{
|
|
Binding: e.FindBindingWithNoEntity(actions[0]),
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
waitUntilExecutionStarted(t, e, tracking1)
|
|
|
|
trackings := []string{tracking1}
|
|
waitGroups := []*sync.WaitGroup{wg1}
|
|
|
|
for _, action := range actions[1:] {
|
|
wg, tracking := e.ExecRequest(&ExecutionRequest{
|
|
Binding: e.FindBindingWithNoEntity(action),
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
trackings = append(trackings, tracking)
|
|
waitGroups = append(waitGroups, wg)
|
|
}
|
|
|
|
require.Eventually(t, func() bool {
|
|
return groupExecutionDistributionMatches(e, trackings, 1, 2, 1)
|
|
}, 2*time.Second, 20*time.Millisecond)
|
|
|
|
for _, wg := range waitGroups {
|
|
wg.Wait()
|
|
}
|
|
}
|
|
|
|
func groupExecutionDistributionMatches(e *Executor, trackings []string, wantRunning, wantQueued, wantBlocked int) bool {
|
|
running := countSnapshots(e, trackings, isRunningSnapshot)
|
|
queued := countSnapshots(e, trackings, func(snapshot LogEntrySnapshot) bool { return snapshot.Queued })
|
|
blocked := countSnapshots(e, trackings, func(snapshot LogEntrySnapshot) bool { return snapshot.Blocked })
|
|
return running == wantRunning && queued == wantQueued && blocked == wantBlocked
|
|
}
|
|
|
|
func countSnapshots(e *Executor, trackings []string, matches func(LogEntrySnapshot) bool) int {
|
|
count := 0
|
|
|
|
for _, tracking := range trackings {
|
|
snapshot, ok := e.SnapshotLog(tracking)
|
|
if ok && matches(snapshot) {
|
|
count++
|
|
}
|
|
}
|
|
|
|
return count
|
|
}
|
|
|
|
func isRunningSnapshot(snapshot LogEntrySnapshot) bool {
|
|
return snapshot.ExecutionStarted && !snapshot.ExecutionFinished
|
|
}
|
|
|
|
func TestUnknownActionGroupReferenceWarnsAndSkipsLimit(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
action := &config.Action{
|
|
Title: "Unknown group action",
|
|
Shell: "echo ok",
|
|
Groups: []string{"missing"},
|
|
}
|
|
|
|
e, cfg := testGroupExecutor([]*config.Action{action}, map[string]*config.ActionGroup{})
|
|
wg, tracking := e.ExecRequest(&ExecutionRequest{
|
|
Binding: e.FindBindingWithNoEntity(action),
|
|
Cfg: cfg,
|
|
AuthenticatedUser: auth.UserFromSystem(cfg, "testuser"),
|
|
})
|
|
|
|
wg.Wait()
|
|
|
|
snapshot, ok := e.SnapshotLog(tracking)
|
|
require.True(t, ok)
|
|
assert.False(t, snapshot.Queued)
|
|
assert.Equal(t, int32(0), snapshot.ExitCode)
|
|
}
|