mirror of
https://gitea.com/gitea/runner.git
synced 2026-11-06 23:30:29 +00:00
fix: keep dry runs off the docker daemon and speed up the tests (#1349)
Without a docker daemon six `act/runner` tests failed instead of passing. A dry run still created and removed job networks, the job teardown dropped the dry-run flag along with every other context value, and the container mocks asked the daemon for the runner arch. This also speeds up the tests and CI without dropping assertions. Most of the time went to git and docker round trips, which are now pinned, batched or coalesced, and the dind and e2e steps cache or overlap more of their setup. On CI, `act/runner` went from 77s to 55s and `test-dind` from 108s to 55s. Locally, `act/runner` without docker went from 53s to 7s. Split out of https://gitea.com/gitea/runner/pulls/1348, which adds Windows CI on top. The go cache changes moved to https://gitea.com/gitea/runner/pulls/1350. Written by Claude Code. --------- Co-authored-by: bircni <[email protected]> Reviewed-on: https://gitea.com/gitea/runner/pulls/1349 Reviewed-by: bircni <[email protected]> Co-authored-by: silverwind <[email protected]>
This commit is contained in:
silverwindandbircni authored and bircni committed
1 parent
8ae48ae033
commit
dafe8d8d0c
50 files changed
+1066
-600
No files matched your search
+143
-134
@@ -9,6 +9,7 @@ import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"gitea.com/gitea/runner/act/common/logging/logtest"
|
||||
@@ -216,33 +217,35 @@ func TestPollerReady(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestPollerPausesAndResumesForLocalAvailability(t *testing.T) {
|
||||
var fetches atomic.Int64
|
||||
cli := mocks.NewClient(t)
|
||||
cli.On("FetchTask", mock.Anything, mock.Anything).Maybe().Run(func(mock.Arguments) {
|
||||
fetches.Add(1)
|
||||
}).Return(&runnerv1.FetchTaskResponse{}, nil)
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
var fetches atomic.Int64
|
||||
cli := mocks.NewClient(t)
|
||||
cli.On("FetchTask", mock.Anything, mock.Anything).Maybe().Run(func(mock.Arguments) {
|
||||
fetches.Add(1)
|
||||
}).Return(&runnerv1.FetchTaskResponse{}, nil)
|
||||
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
cfg.Runner.FetchInterval = 10 * time.Millisecond
|
||||
cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
|
||||
runner := &availabilityRunner{reason: "low disk space"}
|
||||
poller := New(cfg, cli, runner)
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
cfg.Runner.FetchInterval = 10 * time.Millisecond
|
||||
cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
|
||||
runner := &availabilityRunner{reason: "low disk space"}
|
||||
poller := New(cfg, cli, runner)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Go(poller.Poll)
|
||||
time.Sleep(40 * time.Millisecond)
|
||||
assert.Zero(t, fetches.Load(), "an unavailable runner must not fetch a task")
|
||||
var wg sync.WaitGroup
|
||||
wg.Go(poller.Poll)
|
||||
<-time.After(40 * time.Millisecond)
|
||||
assert.Zero(t, fetches.Load(), "an unavailable runner must not fetch a task")
|
||||
|
||||
runner.ready.Store(true)
|
||||
require.Eventually(t, func() bool {
|
||||
return fetches.Load() > 0
|
||||
}, time.Second, 10*time.Millisecond, "polling should resume after local recovery")
|
||||
runner.ready.Store(true)
|
||||
require.Eventually(t, func() bool {
|
||||
return fetches.Load() > 0
|
||||
}, time.Second, 10*time.Millisecond, "polling should resume after local recovery")
|
||||
|
||||
shutdownCtx, cancel := context.WithTimeout(t.Context(), time.Second)
|
||||
defer cancel()
|
||||
require.NoError(t, poller.Shutdown(shutdownCtx))
|
||||
wg.Wait()
|
||||
shutdownCtx, cancel := context.WithTimeout(t.Context(), time.Second)
|
||||
defer cancel()
|
||||
require.NoError(t, poller.Shutdown(shutdownCtx))
|
||||
wg.Wait()
|
||||
})
|
||||
}
|
||||
|
||||
func (m *mockRunner) Run(ctx context.Context, _ *runnerv1.Task) error {
|
||||
@@ -276,36 +279,38 @@ func TestPollerCallsOnIdle(t *testing.T) {
|
||||
name = "PollOnce"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
cli := mocks.NewClient(t)
|
||||
cli.On("FetchTask", mock.Anything, mock.Anything).Return(&runnerv1.FetchTaskResponse{}, nil)
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
cli := mocks.NewClient(t)
|
||||
cli.On("FetchTask", mock.Anything, mock.Anything).Return(&runnerv1.FetchTaskResponse{}, nil)
|
||||
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
if !pollOnce {
|
||||
cfg.Runner.Capacity = 1
|
||||
}
|
||||
cfg.Runner.FetchInterval = 10 * time.Millisecond
|
||||
cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
|
||||
|
||||
runner := &idleAwareRunner{}
|
||||
poller := New(cfg, cli, runner)
|
||||
var wg sync.WaitGroup
|
||||
wg.Go(func() {
|
||||
if pollOnce {
|
||||
poller.PollOnce(func() {})
|
||||
} else {
|
||||
poller.Poll()
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
if !pollOnce {
|
||||
cfg.Runner.Capacity = 1
|
||||
}
|
||||
cfg.Runner.FetchInterval = 10 * time.Millisecond
|
||||
cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
|
||||
|
||||
runner := &idleAwareRunner{}
|
||||
poller := New(cfg, cli, runner)
|
||||
var wg sync.WaitGroup
|
||||
wg.Go(func() {
|
||||
if pollOnce {
|
||||
poller.PollOnce(func() {})
|
||||
} else {
|
||||
poller.Poll()
|
||||
}
|
||||
})
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
return runner.idleCalls.Load() > 0
|
||||
}, time.Second, 10*time.Millisecond)
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Second)
|
||||
defer cancel()
|
||||
require.NoError(t, poller.Shutdown(ctx))
|
||||
wg.Wait()
|
||||
})
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
return runner.idleCalls.Load() > 0
|
||||
}, time.Second, 10*time.Millisecond)
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Second)
|
||||
defer cancel()
|
||||
require.NoError(t, poller.Shutdown(ctx))
|
||||
wg.Wait()
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -323,109 +328,113 @@ func TestPollerPollOnceCallsOnTaskFetchedBeforeRunningIt(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestPoller_ConcurrencyLimitedByCapacity(t *testing.T) {
|
||||
const (
|
||||
capacity = 3
|
||||
totalTasks = 6
|
||||
taskDelay = 50 * time.Millisecond
|
||||
)
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
const (
|
||||
capacity = 3
|
||||
totalTasks = 6
|
||||
taskDelay = 50 * time.Millisecond
|
||||
)
|
||||
|
||||
var (
|
||||
tasksReturned atomic.Int64
|
||||
fetchConcur atomic.Int64
|
||||
maxFetchConcur atomic.Int64
|
||||
)
|
||||
var (
|
||||
tasksReturned atomic.Int64
|
||||
fetchConcur atomic.Int64
|
||||
maxFetchConcur atomic.Int64
|
||||
)
|
||||
|
||||
cli := mocks.NewClient(t)
|
||||
cli.On("FetchTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) {
|
||||
atomicMax(&maxFetchConcur, fetchConcur.Add(1))
|
||||
defer fetchConcur.Add(-1)
|
||||
cli := mocks.NewClient(t)
|
||||
cli.On("FetchTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) {
|
||||
atomicMax(&maxFetchConcur, fetchConcur.Add(1))
|
||||
defer fetchConcur.Add(-1)
|
||||
|
||||
n := tasksReturned.Add(1)
|
||||
if n <= totalTasks {
|
||||
return &runnerv1.FetchTaskResponse{
|
||||
Task: &runnerv1.Task{Id: n},
|
||||
}, nil
|
||||
}
|
||||
return &runnerv1.FetchTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
n := tasksReturned.Add(1)
|
||||
if n <= totalTasks {
|
||||
return &runnerv1.FetchTaskResponse{
|
||||
Task: &runnerv1.Task{Id: n},
|
||||
}, nil
|
||||
}
|
||||
return &runnerv1.FetchTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
|
||||
runner := &mockRunner{delay: taskDelay}
|
||||
runner := &mockRunner{delay: taskDelay}
|
||||
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
cfg.Runner.Capacity = capacity
|
||||
cfg.Runner.FetchInterval = 10 * time.Millisecond
|
||||
cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
cfg.Runner.Capacity = capacity
|
||||
cfg.Runner.FetchInterval = 10 * time.Millisecond
|
||||
cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
|
||||
|
||||
poller := New(cfg, cli, runner)
|
||||
poller := New(cfg, cli, runner)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Go(poller.Poll)
|
||||
var wg sync.WaitGroup
|
||||
wg.Go(poller.Poll)
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
return runner.totalCompleted.Load() >= totalTasks
|
||||
}, 2*time.Second, 10*time.Millisecond, "all tasks should complete")
|
||||
require.Eventually(t, func() bool {
|
||||
return runner.totalCompleted.Load() >= totalTasks
|
||||
}, 2*time.Second, 10*time.Millisecond, "all tasks should complete")
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Second)
|
||||
defer cancel()
|
||||
err = poller.Shutdown(ctx)
|
||||
require.NoError(t, err)
|
||||
wg.Wait()
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Second)
|
||||
defer cancel()
|
||||
err = poller.Shutdown(ctx)
|
||||
require.NoError(t, err)
|
||||
wg.Wait()
|
||||
|
||||
assert.LessOrEqual(t, runner.maxConcurrent.Load(), int64(capacity),
|
||||
"concurrent running tasks must not exceed capacity")
|
||||
assert.GreaterOrEqual(t, runner.maxConcurrent.Load(), int64(2),
|
||||
"with 6 tasks and capacity 3, at least 2 should overlap")
|
||||
assert.Equal(t, int64(1), maxFetchConcur.Load(),
|
||||
"FetchTask must never be called concurrently (single poller)")
|
||||
assert.Equal(t, int64(totalTasks), runner.totalCompleted.Load(),
|
||||
"all tasks should have been executed")
|
||||
assert.LessOrEqual(t, runner.maxConcurrent.Load(), int64(capacity),
|
||||
"concurrent running tasks must not exceed capacity")
|
||||
assert.GreaterOrEqual(t, runner.maxConcurrent.Load(), int64(2),
|
||||
"with 6 tasks and capacity 3, at least 2 should overlap")
|
||||
assert.Equal(t, int64(1), maxFetchConcur.Load(),
|
||||
"FetchTask must never be called concurrently (single poller)")
|
||||
assert.Equal(t, int64(totalTasks), runner.totalCompleted.Load(),
|
||||
"all tasks should have been executed")
|
||||
})
|
||||
}
|
||||
|
||||
func TestPoller_ShutdownForcesJobsOnTimeout(t *testing.T) {
|
||||
var served atomic.Bool
|
||||
cli := mocks.NewClient(t)
|
||||
cli.On("FetchTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) {
|
||||
if served.CompareAndSwap(false, true) {
|
||||
return &runnerv1.FetchTaskResponse{
|
||||
Task: &runnerv1.Task{Id: 1},
|
||||
}, nil
|
||||
}
|
||||
return &runnerv1.FetchTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
var served atomic.Bool
|
||||
cli := mocks.NewClient(t)
|
||||
cli.On("FetchTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) {
|
||||
if served.CompareAndSwap(false, true) {
|
||||
return &runnerv1.FetchTaskResponse{
|
||||
Task: &runnerv1.Task{Id: 1},
|
||||
}, nil
|
||||
}
|
||||
return &runnerv1.FetchTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
|
||||
runner := &mockRunner{delay: 30 * time.Second}
|
||||
runner := &mockRunner{delay: 30 * time.Second}
|
||||
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
cfg.Runner.Capacity = 1
|
||||
cfg.Runner.FetchInterval = 10 * time.Millisecond
|
||||
cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
cfg.Runner.Capacity = 1
|
||||
cfg.Runner.FetchInterval = 10 * time.Millisecond
|
||||
cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
|
||||
|
||||
poller := New(cfg, cli, runner)
|
||||
poller := New(cfg, cli, runner)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Go(poller.Poll)
|
||||
var wg sync.WaitGroup
|
||||
wg.Go(poller.Poll)
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
return runner.running.Load() == 1
|
||||
}, time.Second, 10*time.Millisecond, "task should start running")
|
||||
require.Eventually(t, func() bool {
|
||||
return runner.running.Load() == 1
|
||||
}, time.Second, 10*time.Millisecond, "task should start running")
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
|
||||
defer cancel()
|
||||
start := time.Now()
|
||||
err = poller.Shutdown(ctx)
|
||||
elapsed := time.Since(start)
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
|
||||
defer cancel()
|
||||
start := time.Now()
|
||||
err = poller.Shutdown(ctx)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
require.ErrorIs(t, err, context.DeadlineExceeded)
|
||||
assert.Less(t, elapsed, 5*time.Second,
|
||||
"Shutdown must not block on the parked task; shutdownJobs() must run on timeout")
|
||||
require.ErrorIs(t, err, context.DeadlineExceeded)
|
||||
assert.Less(t, elapsed, 5*time.Second,
|
||||
"Shutdown must not block on the parked task; shutdownJobs() must run on timeout")
|
||||
|
||||
wg.Wait()
|
||||
assert.Equal(t, int64(1), runner.totalCompleted.Load(),
|
||||
"the parked task must be cancelled and unwound")
|
||||
wg.Wait()
|
||||
assert.Equal(t, int64(1), runner.totalCompleted.Load(),
|
||||
"the parked task must be cancelled and unwound")
|
||||
})
|
||||
}
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"gitea.com/gitea/runner/act/common"
|
||||
"gitea.com/gitea/runner/act/common/logging"
|
||||
"gitea.com/gitea/runner/act/common/logging/logtest"
|
||||
"gitea.com/gitea/runner/act/runner"
|
||||
@@ -198,34 +199,36 @@ func TestReporter_LogFlush(t *testing.T) {
|
||||
{"batch size", 10 * time.Second, 5, []string{"log line 0", "log line 1", "log line 2", "log line 3", "log line 4"}},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var updateLogCalls atomic.Int64
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateLog", mock.Anything, mock.Anything).Return(
|
||||
func(_ctx context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
|
||||
updateLogCalls.Add(1)
|
||||
return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil
|
||||
},
|
||||
)
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(&runnerv1.UpdateTaskResponse{}, nil)
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
var updateLogCalls atomic.Int64
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateLog", mock.Anything, mock.Anything).Return(
|
||||
func(_ctx context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
|
||||
updateLogCalls.Add(1)
|
||||
return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil
|
||||
},
|
||||
)
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(&runnerv1.UpdateTaskResponse{}, nil)
|
||||
|
||||
cfg, _ := config.LoadDefault("")
|
||||
cfg.Runner.LogReportInterval = 10 * time.Second
|
||||
cfg.Runner.LogReportMaxLatency = tc.maxLatency
|
||||
cfg.Runner.LogReportBatchSize = tc.batchSize
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
reporter.RunDaemon()
|
||||
defer func() { require.NoError(t, reporter.Close("")) }()
|
||||
cfg, _ := config.LoadDefault("")
|
||||
cfg.Runner.LogReportInterval = 10 * time.Second
|
||||
cfg.Runner.LogReportMaxLatency = tc.maxLatency
|
||||
cfg.Runner.LogReportBatchSize = tc.batchSize
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
reporter.RunDaemon()
|
||||
defer func() { require.NoError(t, reporter.Close("")) }()
|
||||
|
||||
for _, message := range tc.messages {
|
||||
require.NoError(t, fire(reporter, logEntry{
|
||||
Message: message,
|
||||
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
|
||||
}))
|
||||
}
|
||||
assert.Eventually(t, func() bool {
|
||||
return updateLogCalls.Load() > 0
|
||||
}, 500*time.Millisecond, 10*time.Millisecond, "logs must flush before the periodic timers")
|
||||
for _, message := range tc.messages {
|
||||
require.NoError(t, fire(reporter, logEntry{
|
||||
Message: message,
|
||||
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
|
||||
}))
|
||||
}
|
||||
assert.Eventually(t, func() bool {
|
||||
return updateLogCalls.Load() > 0
|
||||
}, 500*time.Millisecond, 10*time.Millisecond, "logs must flush before the periodic timers")
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -313,43 +316,45 @@ func TestReporter_StateChangedRestoredOnError(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestReporter_StateNotifyFlush(t *testing.T) {
|
||||
var updateTaskCalls atomic.Int64
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
var updateTaskCalls atomic.Int64
|
||||
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return(
|
||||
func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
|
||||
return &runnerv1.UpdateLogResponse{
|
||||
AckIndex: req.Index + int64(len(req.Rows)),
|
||||
}, nil
|
||||
},
|
||||
)
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
|
||||
updateTaskCalls.Add(1)
|
||||
return &runnerv1.UpdateTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return(
|
||||
func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
|
||||
return &runnerv1.UpdateLogResponse{
|
||||
AckIndex: req.Index + int64(len(req.Rows)),
|
||||
}, nil
|
||||
},
|
||||
)
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
|
||||
updateTaskCalls.Add(1)
|
||||
return &runnerv1.UpdateTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
|
||||
cfg, _ := config.LoadDefault("")
|
||||
cfg.Runner.StateReportInterval = 10 * time.Second
|
||||
cfg.Runner.LogReportInterval = 10 * time.Second
|
||||
cfg, _ := config.LoadDefault("")
|
||||
cfg.Runner.StateReportInterval = 10 * time.Second
|
||||
cfg.Runner.LogReportInterval = 10 * time.Second
|
||||
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
reporter.RunDaemon()
|
||||
defer func() {
|
||||
_ = reporter.Close("")
|
||||
}()
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
reporter.RunDaemon()
|
||||
defer func() {
|
||||
_ = reporter.Close("")
|
||||
}()
|
||||
|
||||
require.NoError(t, fire(reporter, logEntry{
|
||||
Message: "step starting",
|
||||
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
|
||||
}))
|
||||
require.NoError(t, fire(reporter, logEntry{
|
||||
Message: "step starting",
|
||||
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
|
||||
}))
|
||||
|
||||
assert.Eventually(t, func() bool {
|
||||
return updateTaskCalls.Load() > 0
|
||||
}, 500*time.Millisecond, 10*time.Millisecond,
|
||||
"step transition should have triggered immediate state flush via stateNotify")
|
||||
assert.Eventually(t, func() bool {
|
||||
return updateTaskCalls.Load() > 0
|
||||
}, 500*time.Millisecond, 10*time.Millisecond,
|
||||
"step transition should have triggered immediate state flush via stateNotify")
|
||||
})
|
||||
}
|
||||
|
||||
func TestReporter_CloseAlwaysSendsRowsWithNoMore(t *testing.T) {
|
||||
@@ -401,33 +406,35 @@ func TestReporter_CloseAlwaysSendsRowsWithNoMore(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestReporter_StateHeartbeat(t *testing.T) {
|
||||
var updateTaskCalls atomic.Int64
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
var updateTaskCalls atomic.Int64
|
||||
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
|
||||
updateTaskCalls.Add(1)
|
||||
return &runnerv1.UpdateTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
|
||||
updateTaskCalls.Add(1)
|
||||
return &runnerv1.UpdateTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
|
||||
cfg, _ := config.LoadDefault("")
|
||||
cfg.Runner.StateReportInterval = 50 * time.Millisecond
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
cfg, _ := config.LoadDefault("")
|
||||
cfg.Runner.StateReportInterval = 50 * time.Millisecond
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
|
||||
reporter.stateMu.Lock()
|
||||
reporter.stateChanged = true
|
||||
reporter.stateMu.Unlock()
|
||||
require.NoError(t, reporter.ReportState(false))
|
||||
require.Equal(t, int64(1), updateTaskCalls.Load())
|
||||
reporter.stateMu.Lock()
|
||||
reporter.stateChanged = true
|
||||
reporter.stateMu.Unlock()
|
||||
require.NoError(t, reporter.ReportState(false))
|
||||
require.Equal(t, int64(1), updateTaskCalls.Load())
|
||||
|
||||
require.NoError(t, reporter.ReportState(false))
|
||||
assert.Equal(t, int64(1), updateTaskCalls.Load(), "no-op ReportState within stateReportInterval must skip")
|
||||
require.NoError(t, reporter.ReportState(false))
|
||||
assert.Equal(t, int64(1), updateTaskCalls.Load(), "no-op ReportState within stateReportInterval must skip")
|
||||
|
||||
time.Sleep(2 * cfg.Runner.StateReportInterval)
|
||||
require.NoError(t, reporter.ReportState(false))
|
||||
assert.Equal(t, int64(2), updateTaskCalls.Load(), "ReportState must heartbeat after stateReportInterval even with no state change")
|
||||
<-time.After(2 * cfg.Runner.StateReportInterval)
|
||||
require.NoError(t, reporter.ReportState(false))
|
||||
assert.Equal(t, int64(2), updateTaskCalls.Load(), "ReportState must heartbeat after stateReportInterval even with no state change")
|
||||
})
|
||||
}
|
||||
|
||||
func TestReporter_ServerCancelStillFlushesFinal(t *testing.T) {
|
||||
@@ -556,58 +563,60 @@ func TestReporter_CloseReportsCancelledOnCanceledCtx(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestReporter_StopHeartbeats(t *testing.T) {
|
||||
var updateTaskCalls atomic.Int64
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
var updateTaskCalls atomic.Int64
|
||||
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return(
|
||||
func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
|
||||
return &runnerv1.UpdateLogResponse{
|
||||
AckIndex: req.Index + int64(len(req.Rows)),
|
||||
}, nil
|
||||
},
|
||||
)
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
|
||||
updateTaskCalls.Add(1)
|
||||
return &runnerv1.UpdateTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return(
|
||||
func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
|
||||
return &runnerv1.UpdateLogResponse{
|
||||
AckIndex: req.Index + int64(len(req.Rows)),
|
||||
}, nil
|
||||
},
|
||||
)
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
|
||||
updateTaskCalls.Add(1)
|
||||
return &runnerv1.UpdateTaskResponse{}, nil
|
||||
},
|
||||
)
|
||||
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
cfg.Runner.StateReportInterval = 20 * time.Millisecond
|
||||
cfg.Runner.LogReportInterval = time.Hour
|
||||
cfg, err := config.LoadDefault("")
|
||||
require.NoError(t, err)
|
||||
cfg.Runner.StateReportInterval = 20 * time.Millisecond
|
||||
cfg.Runner.LogReportInterval = time.Hour
|
||||
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
reporter.RunDaemon()
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
reporter.RunDaemon()
|
||||
|
||||
reporter.stateMu.Lock()
|
||||
reporter.stateChanged = true
|
||||
reporter.state.Result = runnerv1.Result_RESULT_SUCCESS
|
||||
reporter.state.StoppedAt = timestamppb.Now()
|
||||
reporter.stateMu.Unlock()
|
||||
reporter.stateMu.Lock()
|
||||
reporter.stateChanged = true
|
||||
reporter.state.Result = runnerv1.Result_RESULT_SUCCESS
|
||||
reporter.state.StoppedAt = timestamppb.Now()
|
||||
reporter.stateMu.Unlock()
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
return updateTaskCalls.Load() >= 1
|
||||
}, time.Second, 5*time.Millisecond, "daemon must send at least one UpdateTask before StopHeartbeats")
|
||||
require.Eventually(t, func() bool {
|
||||
return updateTaskCalls.Load() >= 1
|
||||
}, time.Second, 5*time.Millisecond, "daemon must send at least one UpdateTask before StopHeartbeats")
|
||||
|
||||
beforeStop := updateTaskCalls.Load()
|
||||
reporter.StopHeartbeats()
|
||||
beforeStop := updateTaskCalls.Load()
|
||||
reporter.StopHeartbeats()
|
||||
|
||||
select {
|
||||
case <-reporter.daemon:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("StopHeartbeats must stop the daemon loop")
|
||||
}
|
||||
select {
|
||||
case <-reporter.daemon:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("StopHeartbeats must stop the daemon loop")
|
||||
}
|
||||
|
||||
time.Sleep(3 * cfg.Runner.StateReportInterval)
|
||||
assert.Equal(t, beforeStop, updateTaskCalls.Load(),
|
||||
"UpdateTask must not be called after StopHeartbeats")
|
||||
<-time.After(3 * cfg.Runner.StateReportInterval)
|
||||
assert.Equal(t, beforeStop, updateTaskCalls.Load(),
|
||||
"UpdateTask must not be called after StopHeartbeats")
|
||||
|
||||
require.NoError(t, reporter.Close(""))
|
||||
assert.Greater(t, updateTaskCalls.Load(), beforeStop,
|
||||
"Close() must still send a final UpdateTask after StopHeartbeats")
|
||||
require.NoError(t, reporter.Close(""))
|
||||
assert.Greater(t, updateTaskCalls.Load(), beforeStop,
|
||||
"Close() must still send a final UpdateTask after StopHeartbeats")
|
||||
})
|
||||
}
|
||||
|
||||
func TestAppendLogRow(t *testing.T) {
|
||||
@@ -786,48 +795,53 @@ func TestReporter_AckIndexBeyondBuffer(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestReporter_CloseWithStuckDaemon(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
entered := make(chan struct{})
|
||||
var once sync.Once
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
entered := make(chan struct{})
|
||||
var once sync.Once
|
||||
output := logging.Output()
|
||||
logging.SetOutput(common.NewLineWriter(func(line string) bool {
|
||||
if strings.Contains(line, "No Response from RunDaemon") {
|
||||
close(release)
|
||||
}
|
||||
return true
|
||||
}))
|
||||
t.Cleanup(func() { logging.SetOutput(output) })
|
||||
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
|
||||
once.Do(func() {
|
||||
close(entered)
|
||||
<-release
|
||||
client := mocks.NewClient(t)
|
||||
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
|
||||
once.Do(func() {
|
||||
close(entered)
|
||||
<-release
|
||||
})
|
||||
return &runnerv1.UpdateLogResponse{
|
||||
AckIndex: req.Index + int64(len(req.Rows)),
|
||||
}, nil
|
||||
})
|
||||
return &runnerv1.UpdateLogResponse{
|
||||
AckIndex: req.Index + int64(len(req.Rows)),
|
||||
}, nil
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
|
||||
return &runnerv1.UpdateTaskResponse{}, nil
|
||||
}).Maybe()
|
||||
|
||||
cfg, _ := config.LoadDefault("")
|
||||
cfg.Runner.LogReportInterval = 10 * time.Millisecond
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.daemonWait = time.Millisecond
|
||||
reporter.RunDaemon()
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
|
||||
require.NoError(t, fire(reporter, logEntry{
|
||||
Message: "hello",
|
||||
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
|
||||
Level: logging.LevelInfo,
|
||||
}))
|
||||
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("daemon never reached UpdateLog")
|
||||
}
|
||||
require.NoError(t, reporter.Close(""))
|
||||
})
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
|
||||
return &runnerv1.UpdateTaskResponse{}, nil
|
||||
}).Maybe()
|
||||
|
||||
cfg, _ := config.LoadDefault("")
|
||||
cfg.Runner.LogReportInterval = 10 * time.Millisecond
|
||||
reporter := newTestReporter(t, client, cfg)
|
||||
reporter.daemonWait = time.Millisecond
|
||||
reporter.RunDaemon()
|
||||
reporter.ResetSteps(newTestSteps(1))
|
||||
|
||||
require.NoError(t, fire(reporter, logEntry{
|
||||
Message: "hello",
|
||||
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
|
||||
Level: logging.LevelInfo,
|
||||
}))
|
||||
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("daemon never reached UpdateLog")
|
||||
}
|
||||
go func() {
|
||||
time.Sleep(10 * reporter.daemonWait)
|
||||
close(release)
|
||||
}()
|
||||
|
||||
require.NoError(t, reporter.Close(""))
|
||||
}
|
||||
|
||||
func TestReporter_ServerFailureResultIsReportedAsFailure(t *testing.T) {
|
||||
|
||||
Reference in new issue
Block a user