mirror of
https://github.com/project-zot/zot.git
synced 2024-12-16 21:56:37 -05:00
444 lines
13 KiB
Go
444 lines
13 KiB
Go
package scheduler_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"runtime"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
. "github.com/smartystreets/goconvey/convey"
|
|
|
|
"zotregistry.dev/zot/pkg/api/config"
|
|
"zotregistry.dev/zot/pkg/extensions/monitoring"
|
|
"zotregistry.dev/zot/pkg/log"
|
|
"zotregistry.dev/zot/pkg/scheduler"
|
|
)
|
|
|
|
type task struct {
|
|
log log.Logger
|
|
msg string
|
|
err bool
|
|
delay time.Duration
|
|
}
|
|
|
|
var errInternal = errors.New("task: internal error")
|
|
|
|
func (t *task) DoWork(ctx context.Context) error {
|
|
if t.err {
|
|
return errInternal
|
|
}
|
|
|
|
for idx := 0; idx < 5; idx++ {
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
|
|
time.Sleep(t.delay)
|
|
}
|
|
|
|
t.log.Info().Msg(t.msg)
|
|
|
|
return nil
|
|
}
|
|
|
|
func (t *task) String() string {
|
|
return t.Name()
|
|
}
|
|
|
|
func (t *task) Name() string {
|
|
return "TestTask"
|
|
}
|
|
|
|
type generator struct {
|
|
log log.Logger
|
|
priority string
|
|
done bool
|
|
index int
|
|
step int
|
|
limit int
|
|
taskDelay time.Duration
|
|
}
|
|
|
|
func (g *generator) Next() (scheduler.Task, error) {
|
|
if g.step > g.limit {
|
|
g.done = true
|
|
}
|
|
g.step++
|
|
g.index++
|
|
|
|
if g.step%11 == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
if g.step%13 == 0 {
|
|
return nil, errInternal
|
|
}
|
|
|
|
return &task{
|
|
log: g.log,
|
|
msg: fmt.Sprintf("executing %s task; index: %d", g.priority, g.index),
|
|
err: false,
|
|
delay: g.taskDelay,
|
|
}, nil
|
|
}
|
|
|
|
func (g *generator) IsDone() bool {
|
|
return g.done
|
|
}
|
|
|
|
func (g *generator) IsReady() bool {
|
|
return true
|
|
}
|
|
|
|
func (g *generator) Reset() {
|
|
g.done = false
|
|
g.step = 0
|
|
}
|
|
|
|
type shortGenerator struct {
|
|
log log.Logger
|
|
priority string
|
|
done bool
|
|
index int
|
|
step int
|
|
}
|
|
|
|
func (g *shortGenerator) Next() (scheduler.Task, error) {
|
|
g.done = true
|
|
|
|
return &task{log: g.log, msg: fmt.Sprintf("executing %s task; index: %d", g.priority, g.index), err: false}, nil
|
|
}
|
|
|
|
func (g *shortGenerator) IsDone() bool {
|
|
return g.done
|
|
}
|
|
|
|
func (g *shortGenerator) IsReady() bool {
|
|
return true
|
|
}
|
|
|
|
func (g *shortGenerator) Reset() {
|
|
g.done = true
|
|
g.step = 0
|
|
}
|
|
|
|
func TestScheduler(t *testing.T) {
|
|
Convey("Test active to waiting periodic generator", t, func() {
|
|
logFile, err := os.CreateTemp("", "zot-log*.txt")
|
|
So(err, ShouldBeNil)
|
|
|
|
defer os.Remove(logFile.Name()) // clean up
|
|
|
|
logger := log.NewLogger("debug", logFile.Name())
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(config.New(), metrics, logger)
|
|
|
|
genH := &shortGenerator{log: logger, priority: "high priority"}
|
|
// interval has to be higher than throttle value to simulate
|
|
sch.SubmitGenerator(genH, 1*time.Second, scheduler.HighPriority)
|
|
|
|
sch.RunScheduler()
|
|
time.Sleep(2 * time.Second)
|
|
sch.Shutdown()
|
|
|
|
data, err := os.ReadFile(logFile.Name())
|
|
So(err, ShouldBeNil)
|
|
So(string(data), ShouldContainSubstring, "waiting generator is ready, pushing to ready generators")
|
|
})
|
|
|
|
Convey("Test order of generators in queue", t, func() {
|
|
logFile, err := os.CreateTemp("", "zot-log*.txt")
|
|
So(err, ShouldBeNil)
|
|
|
|
defer os.Remove(logFile.Name()) // clean up
|
|
|
|
logger := log.NewLogger("debug", logFile.Name())
|
|
cfg := config.New()
|
|
cfg.Scheduler = &config.SchedulerConfig{NumWorkers: 3}
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(cfg, metrics, logger)
|
|
sch.RateLimit = 5 * time.Second
|
|
|
|
genL := &generator{log: logger, priority: "low priority", limit: 100, taskDelay: 100 * time.Millisecond}
|
|
sch.SubmitGenerator(genL, time.Duration(0), scheduler.LowPriority)
|
|
|
|
genM := &generator{log: logger, priority: "medium priority", limit: 100, taskDelay: 100 * time.Millisecond}
|
|
sch.SubmitGenerator(genM, time.Duration(0), scheduler.MediumPriority)
|
|
|
|
genH := &generator{log: logger, priority: "high priority", limit: 100, taskDelay: 100 * time.Millisecond}
|
|
sch.SubmitGenerator(genH, time.Duration(0), scheduler.HighPriority)
|
|
|
|
sch.RunScheduler()
|
|
time.Sleep(4 * time.Second)
|
|
sch.Shutdown()
|
|
|
|
data, err := os.ReadFile(logFile.Name())
|
|
So(err, ShouldBeNil)
|
|
|
|
So(string(data), ShouldContainSubstring, "executing high priority task; index: 1")
|
|
So(string(data), ShouldContainSubstring, "executing high priority task; index: 2")
|
|
So(string(data), ShouldNotContainSubstring, "executing medium priority task; index: 1")
|
|
So(string(data), ShouldNotContainSubstring, "failed to execute task")
|
|
})
|
|
|
|
Convey("Test reordering of generators in queue", t, func() {
|
|
logFile, err := os.CreateTemp("", "zot-log*.txt")
|
|
So(err, ShouldBeNil)
|
|
|
|
defer os.Remove(logFile.Name()) // clean up
|
|
|
|
logger := log.NewLogger("debug", logFile.Name())
|
|
cfg := config.New()
|
|
cfg.Scheduler = &config.SchedulerConfig{NumWorkers: 3}
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(cfg, metrics, logger)
|
|
sch.RateLimit = 1 * time.Nanosecond
|
|
|
|
// Testing repordering of generators using the same medium priority, as well as reordering with
|
|
// a low priority generator
|
|
|
|
genL := &generator{log: logger, priority: "low priority", limit: 110, taskDelay: time.Nanosecond}
|
|
sch.SubmitGenerator(genL, time.Duration(0), scheduler.LowPriority)
|
|
|
|
genM := &generator{log: logger, priority: "medium 1 priority", limit: 110, taskDelay: time.Nanosecond}
|
|
sch.SubmitGenerator(genM, time.Duration(0), scheduler.MediumPriority)
|
|
|
|
genH := &generator{log: logger, priority: "medium 2 priority", limit: 110, taskDelay: time.Nanosecond}
|
|
sch.SubmitGenerator(genH, time.Duration(0), scheduler.MediumPriority)
|
|
|
|
sch.RunScheduler()
|
|
time.Sleep(1 * time.Second)
|
|
sch.Shutdown()
|
|
|
|
data, err := os.ReadFile(logFile.Name())
|
|
So(err, ShouldBeNil)
|
|
|
|
// Check all tasks show up in the logs
|
|
for i := 1; i < 110; i++ {
|
|
if i%11 == 0 || i%13 == 0 {
|
|
continue
|
|
}
|
|
|
|
So(string(data), ShouldContainSubstring, fmt.Sprintf("executing medium 1 priority task; index: %d", i))
|
|
So(string(data), ShouldContainSubstring, fmt.Sprintf("executing medium 2 priority task; index: %d", i))
|
|
So(string(data), ShouldContainSubstring, fmt.Sprintf("executing low priority task; index: %d", i))
|
|
}
|
|
|
|
taskCounter := 0
|
|
priorityFlippedCounter := 0
|
|
samePriorityFlippedCounter := 0
|
|
lastPriority := "medium"
|
|
lastMediumGenerator := "1"
|
|
|
|
for _, line := range strings.Split(strings.TrimSuffix(string(data), "\n"), "\n") {
|
|
if !strings.Contains(line, "priority task; index: ") {
|
|
continue
|
|
}
|
|
|
|
taskCounter++
|
|
|
|
// low priority tasks start executing later
|
|
// medium priority generators are prioritized until the rank 100/9 (8 generated tasks)
|
|
// starting with 100/10, a low priority generator could potentially be prioritized instead
|
|
// there will be at least 8 * 2 medium priority tasks executed before low priority tasks are pushed
|
|
if taskCounter < 17 {
|
|
So(line, ShouldContainSubstring, "executing medium")
|
|
}
|
|
|
|
// medium priority 2*110 medium priority tasks should have been generated,
|
|
// medium priority generators should be done
|
|
// add around 10 low priority tasks to the counter
|
|
// and an additional margin of 5 to make sure the test is stable
|
|
if taskCounter > 225 {
|
|
So(line, ShouldContainSubstring, "executing low priority")
|
|
}
|
|
|
|
if strings.Contains(line, "executing medium") {
|
|
if !strings.Contains(line, fmt.Sprintf("executing medium %s", lastMediumGenerator)) {
|
|
samePriorityFlippedCounter++
|
|
if lastMediumGenerator == "1" {
|
|
lastMediumGenerator = "2"
|
|
} else {
|
|
lastMediumGenerator = "1"
|
|
}
|
|
}
|
|
}
|
|
|
|
if !strings.Contains(line, fmt.Sprintf("executing %s", lastPriority)) {
|
|
priorityFlippedCounter++
|
|
if lastPriority == "low" {
|
|
lastPriority = "medium"
|
|
} else {
|
|
lastPriority = "low"
|
|
}
|
|
}
|
|
}
|
|
|
|
// fairness: make sure none of the medium priority generators is favored by the algorithm
|
|
So(samePriorityFlippedCounter, ShouldBeGreaterThanOrEqualTo, 60)
|
|
t.Logf("Switched between medium priority generators %d times", samePriorityFlippedCounter)
|
|
// fairness: make sure the algorithm alternates between generator priorities
|
|
So(priorityFlippedCounter, ShouldBeGreaterThanOrEqualTo, 10)
|
|
t.Logf("Switched between generator priorities %d times", priorityFlippedCounter)
|
|
})
|
|
|
|
Convey("Test task returning an error", t, func() {
|
|
logFile, err := os.CreateTemp("", "zot-log*.txt")
|
|
So(err, ShouldBeNil)
|
|
|
|
defer os.Remove(logFile.Name()) // clean up
|
|
|
|
logger := log.NewLogger("debug", logFile.Name())
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(config.New(), metrics, logger)
|
|
|
|
t := &task{log: logger, msg: "", err: true}
|
|
sch.SubmitTask(t, scheduler.MediumPriority)
|
|
|
|
sch.RunScheduler()
|
|
time.Sleep(500 * time.Millisecond)
|
|
sch.Shutdown()
|
|
|
|
data, err := os.ReadFile(logFile.Name())
|
|
So(err, ShouldBeNil)
|
|
So(string(data), ShouldContainSubstring, "adding a new task")
|
|
So(string(data), ShouldContainSubstring, "failed to execute task")
|
|
})
|
|
|
|
Convey("Test resubmit generator", t, func() {
|
|
logFile, err := os.CreateTemp("", "zot-log*.txt")
|
|
So(err, ShouldBeNil)
|
|
|
|
defer os.Remove(logFile.Name()) // clean up
|
|
|
|
logger := log.NewLogger("debug", logFile.Name())
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(config.New(), metrics, logger)
|
|
|
|
genL := &generator{log: logger, priority: "low priority", limit: 100, taskDelay: 100 * time.Millisecond}
|
|
sch.SubmitGenerator(genL, 20*time.Millisecond, scheduler.LowPriority)
|
|
|
|
sch.RunScheduler()
|
|
time.Sleep(1 * time.Second)
|
|
sch.Shutdown()
|
|
|
|
data, err := os.ReadFile(logFile.Name())
|
|
So(err, ShouldBeNil)
|
|
So(string(data), ShouldContainSubstring, "executing low priority task; index: 1")
|
|
So(string(data), ShouldContainSubstring, "executing low priority task; index: 2")
|
|
})
|
|
|
|
Convey("Try to add a task with wrong priority", t, func() {
|
|
logFile, err := os.CreateTemp("", "zot-log*.txt")
|
|
So(err, ShouldBeNil)
|
|
|
|
defer os.Remove(logFile.Name()) // clean up
|
|
|
|
logger := log.NewLogger("debug", logFile.Name())
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(config.New(), metrics, logger)
|
|
|
|
t := &task{log: logger, msg: "", err: false}
|
|
sch.SubmitTask(t, -1)
|
|
|
|
data, err := os.ReadFile(logFile.Name())
|
|
So(err, ShouldBeNil)
|
|
So(string(data), ShouldNotContainSubstring, "adding a new task")
|
|
})
|
|
|
|
Convey("Test adding a new task when context is done", t, func() {
|
|
logFile, err := os.CreateTemp("", "zot-log*.txt")
|
|
So(err, ShouldBeNil)
|
|
|
|
defer os.Remove(logFile.Name()) // clean up
|
|
|
|
logger := log.NewLogger("debug", logFile.Name())
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(config.New(), metrics, logger)
|
|
|
|
sch.RunScheduler()
|
|
sch.Shutdown()
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
t := &task{log: logger, msg: "", err: false}
|
|
sch.SubmitTask(t, scheduler.LowPriority)
|
|
|
|
data, err := os.ReadFile(logFile.Name())
|
|
So(err, ShouldBeNil)
|
|
So(string(data), ShouldNotContainSubstring, "adding a new task")
|
|
})
|
|
|
|
Convey("Test stopping scheduler by calling Shutdown()", t, func() {
|
|
logFile, err := os.CreateTemp("", "zot-log*.txt")
|
|
So(err, ShouldBeNil)
|
|
|
|
defer os.Remove(logFile.Name()) // clean up
|
|
|
|
logger := log.NewLogger("debug", logFile.Name())
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(config.New(), metrics, logger)
|
|
|
|
genL := &generator{log: logger, priority: "medium priority", limit: 100, taskDelay: 100 * time.Millisecond}
|
|
sch.SubmitGenerator(genL, 20*time.Millisecond, scheduler.MediumPriority)
|
|
|
|
sch.RunScheduler()
|
|
time.Sleep(1 * time.Second)
|
|
sch.Shutdown()
|
|
|
|
data, err := os.ReadFile(logFile.Name())
|
|
So(err, ShouldBeNil)
|
|
So(string(data), ShouldContainSubstring, "executing medium priority task; index: 1")
|
|
So(string(data), ShouldContainSubstring, "executing medium priority task; index: 2")
|
|
So(string(data), ShouldContainSubstring, "received stop signal, gracefully shutting down...")
|
|
})
|
|
|
|
Convey("Test scheduler Priority.String() method", t, func() {
|
|
var p scheduler.Priority //nolint: varnamelen
|
|
// test invalid priority
|
|
p = 6238734
|
|
So(p.String(), ShouldEqual, "invalid")
|
|
p = scheduler.LowPriority
|
|
So(p.String(), ShouldEqual, "low")
|
|
p = scheduler.MediumPriority
|
|
So(p.String(), ShouldEqual, "medium")
|
|
p = scheduler.HighPriority
|
|
So(p.String(), ShouldEqual, "high")
|
|
})
|
|
|
|
Convey("Test scheduler State.String() method", t, func() {
|
|
var s scheduler.State //nolint: varnamelen
|
|
// test invalid state
|
|
s = -67
|
|
So(s.String(), ShouldEqual, "invalid")
|
|
s = scheduler.Ready
|
|
So(s.String(), ShouldEqual, "ready")
|
|
s = scheduler.Waiting
|
|
So(s.String(), ShouldEqual, "waiting")
|
|
s = scheduler.Done
|
|
So(s.String(), ShouldEqual, "done")
|
|
})
|
|
}
|
|
|
|
func TestGetNumWorkers(t *testing.T) {
|
|
Convey("Test setting the number of workers - default value", t, func() {
|
|
logger := log.NewLogger("debug", "logFile")
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(config.New(), metrics, logger)
|
|
defer os.Remove("logFile")
|
|
So(sch.NumWorkers, ShouldEqual, runtime.NumCPU()*4)
|
|
})
|
|
|
|
Convey("Test setting the number of workers - getting the value from config", t, func() {
|
|
cfg := config.New()
|
|
cfg.Scheduler = &config.SchedulerConfig{NumWorkers: 3}
|
|
logger := log.NewLogger("debug", "logFile")
|
|
metrics := monitoring.NewMetricsServer(true, logger)
|
|
sch := scheduler.NewScheduler(cfg, metrics, logger)
|
|
defer os.Remove("logFile")
|
|
So(sch.NumWorkers, ShouldEqual, 3)
|
|
})
|
|
}
|