Honor configured PVE polling interval in scheduler
This commit is contained in:
parent
0adc4e7838
commit
5a6c6091bf
3 changed files with 128 additions and 6 deletions
|
|
@ -4183,6 +4183,19 @@ func (m *Monitor) getExecutor() PollExecutor {
|
||||||
return exec
|
return exec
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func clampInterval(value, min, max time.Duration) time.Duration {
|
||||||
|
if value <= 0 {
|
||||||
|
return min
|
||||||
|
}
|
||||||
|
if min > 0 && value < min {
|
||||||
|
return min
|
||||||
|
}
|
||||||
|
if max > 0 && value > max {
|
||||||
|
return max
|
||||||
|
}
|
||||||
|
return value
|
||||||
|
}
|
||||||
|
|
||||||
func (m *Monitor) effectivePVEPollingInterval() time.Duration {
|
func (m *Monitor) effectivePVEPollingInterval() time.Duration {
|
||||||
const minInterval = 10 * time.Second
|
const minInterval = 10 * time.Second
|
||||||
const maxInterval = time.Hour
|
const maxInterval = time.Hour
|
||||||
|
|
@ -4200,6 +4213,27 @@ func (m *Monitor) effectivePVEPollingInterval() time.Duration {
|
||||||
return interval
|
return interval
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (m *Monitor) baseIntervalForInstanceType(instanceType InstanceType) time.Duration {
|
||||||
|
if m == nil || m.config == nil {
|
||||||
|
return DefaultSchedulerConfig().BaseInterval
|
||||||
|
}
|
||||||
|
|
||||||
|
switch instanceType {
|
||||||
|
case InstanceTypePVE:
|
||||||
|
return m.effectivePVEPollingInterval()
|
||||||
|
case InstanceTypePBS:
|
||||||
|
return clampInterval(m.config.PBSPollingInterval, 10*time.Second, time.Hour)
|
||||||
|
case InstanceTypePMG:
|
||||||
|
return clampInterval(m.config.PMGPollingInterval, 10*time.Second, time.Hour)
|
||||||
|
default:
|
||||||
|
base := m.config.AdaptivePollingBaseInterval
|
||||||
|
if base <= 0 {
|
||||||
|
base = DefaultSchedulerConfig().BaseInterval
|
||||||
|
}
|
||||||
|
return clampInterval(base, time.Second, 0)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Start begins the monitoring loop
|
// Start begins the monitoring loop
|
||||||
func (m *Monitor) Start(ctx context.Context, wsHub *websocket.Hub) {
|
func (m *Monitor) Start(ctx context.Context, wsHub *websocket.Hub) {
|
||||||
pollingInterval := m.effectivePVEPollingInterval()
|
pollingInterval := m.effectivePVEPollingInterval()
|
||||||
|
|
@ -4883,9 +4917,10 @@ func (m *Monitor) rescheduleTask(task ScheduledTask) {
|
||||||
}
|
}
|
||||||
|
|
||||||
if m.scheduler == nil {
|
if m.scheduler == nil {
|
||||||
|
baseInterval := m.baseIntervalForInstanceType(task.InstanceType)
|
||||||
nextInterval := task.Interval
|
nextInterval := task.Interval
|
||||||
if nextInterval <= 0 && m.config != nil {
|
if nextInterval <= 0 {
|
||||||
nextInterval = m.config.AdaptivePollingBaseInterval
|
nextInterval = baseInterval
|
||||||
}
|
}
|
||||||
if nextInterval <= 0 {
|
if nextInterval <= 0 {
|
||||||
nextInterval = DefaultSchedulerConfig().BaseInterval
|
nextInterval = DefaultSchedulerConfig().BaseInterval
|
||||||
|
|
|
||||||
|
|
@ -127,11 +127,11 @@ func (m *Monitor) buildScheduledTasks(now time.Time) []ScheduledTask {
|
||||||
|
|
||||||
if m.scheduler == nil {
|
if m.scheduler == nil {
|
||||||
tasks := make([]ScheduledTask, 0, len(descriptors))
|
tasks := make([]ScheduledTask, 0, len(descriptors))
|
||||||
interval := m.config.AdaptivePollingBaseInterval
|
|
||||||
if interval <= 0 {
|
|
||||||
interval = DefaultSchedulerConfig().BaseInterval
|
|
||||||
}
|
|
||||||
for _, desc := range descriptors {
|
for _, desc := range descriptors {
|
||||||
|
interval := m.baseIntervalForInstanceType(desc.Type)
|
||||||
|
if interval <= 0 {
|
||||||
|
interval = DefaultSchedulerConfig().BaseInterval
|
||||||
|
}
|
||||||
tasks = append(tasks, ScheduledTask{
|
tasks = append(tasks, ScheduledTask{
|
||||||
InstanceName: desc.Name,
|
InstanceName: desc.Name,
|
||||||
InstanceType: desc.Type,
|
InstanceType: desc.Type,
|
||||||
|
|
|
||||||
87
internal/monitoring/monitor_polling_test.go
Normal file
87
internal/monitoring/monitor_polling_test.go
Normal file
|
|
@ -0,0 +1,87 @@
|
||||||
|
package monitoring
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/rcourtman/pulse-go-rewrite/internal/config"
|
||||||
|
"github.com/rcourtman/pulse-go-rewrite/pkg/pbs"
|
||||||
|
"github.com/rcourtman/pulse-go-rewrite/pkg/pmg"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestBuildScheduledTasksUsesConfiguredIntervals(t *testing.T) {
|
||||||
|
now := time.Now()
|
||||||
|
cfg := &config.Config{
|
||||||
|
PVEPollingInterval: 2 * time.Minute,
|
||||||
|
PBSPollingInterval: 45 * time.Second,
|
||||||
|
PMGPollingInterval: 90 * time.Second,
|
||||||
|
AdaptivePollingBaseInterval: 10 * time.Second,
|
||||||
|
}
|
||||||
|
|
||||||
|
monitor := &Monitor{
|
||||||
|
config: cfg,
|
||||||
|
pveClients: map[string]PVEClientInterface{"pve-1": nil},
|
||||||
|
pbsClients: map[string]*pbs.Client{"pbs-1": nil},
|
||||||
|
pmgClients: map[string]*pmg.Client{"pmg-1": nil},
|
||||||
|
}
|
||||||
|
|
||||||
|
tasks := monitor.buildScheduledTasks(now)
|
||||||
|
if len(tasks) != 3 {
|
||||||
|
t.Fatalf("expected 3 tasks, got %d", len(tasks))
|
||||||
|
}
|
||||||
|
|
||||||
|
got := map[InstanceType]time.Duration{}
|
||||||
|
for _, task := range tasks {
|
||||||
|
if !task.NextRun.Equal(now) {
|
||||||
|
t.Fatalf("expected NextRun to equal provided time, got %v", task.NextRun)
|
||||||
|
}
|
||||||
|
got[task.InstanceType] = task.Interval
|
||||||
|
}
|
||||||
|
|
||||||
|
if got[InstanceTypePVE] != cfg.PVEPollingInterval {
|
||||||
|
t.Fatalf("expected PVE interval %v, got %v", cfg.PVEPollingInterval, got[InstanceTypePVE])
|
||||||
|
}
|
||||||
|
if got[InstanceTypePBS] != cfg.PBSPollingInterval {
|
||||||
|
t.Fatalf("expected PBS interval %v, got %v", cfg.PBSPollingInterval, got[InstanceTypePBS])
|
||||||
|
}
|
||||||
|
if got[InstanceTypePMG] != cfg.PMGPollingInterval {
|
||||||
|
t.Fatalf("expected PMG interval %v, got %v", cfg.PMGPollingInterval, got[InstanceTypePMG])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRescheduleTaskUsesInstanceIntervalWhenSchedulerDisabled(t *testing.T) {
|
||||||
|
cfg := &config.Config{
|
||||||
|
PVEPollingInterval: 75 * time.Second,
|
||||||
|
AdaptivePollingBaseInterval: 10 * time.Second,
|
||||||
|
}
|
||||||
|
|
||||||
|
monitor := &Monitor{
|
||||||
|
config: cfg,
|
||||||
|
taskQueue: NewTaskQueue(),
|
||||||
|
}
|
||||||
|
|
||||||
|
task := ScheduledTask{
|
||||||
|
InstanceName: "pve-1",
|
||||||
|
InstanceType: InstanceTypePVE,
|
||||||
|
Interval: 0,
|
||||||
|
NextRun: time.Now(),
|
||||||
|
}
|
||||||
|
|
||||||
|
monitor.rescheduleTask(task)
|
||||||
|
|
||||||
|
monitor.taskQueue.mu.Lock()
|
||||||
|
entry, ok := monitor.taskQueue.entries[schedulerKey(task.InstanceType, task.InstanceName)]
|
||||||
|
monitor.taskQueue.mu.Unlock()
|
||||||
|
if !ok {
|
||||||
|
t.Fatalf("expected task to be rescheduled in queue")
|
||||||
|
}
|
||||||
|
|
||||||
|
if entry.task.Interval != cfg.PVEPollingInterval {
|
||||||
|
t.Fatalf("expected interval %v, got %v", cfg.PVEPollingInterval, entry.task.Interval)
|
||||||
|
}
|
||||||
|
|
||||||
|
remaining := time.Until(entry.task.NextRun)
|
||||||
|
if remaining < cfg.PVEPollingInterval-2*time.Second || remaining > cfg.PVEPollingInterval+time.Second {
|
||||||
|
t.Fatalf("expected NextRun about %v from now, got %v", cfg.PVEPollingInterval, remaining)
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in a new issue