959ee42bcb
Aggregate monitor history into daily rollups and add maintenance helpers for rollup refresh and history cleanup.
98 lines
2.5 KiB
Go
98 lines
2.5 KiB
Go
package service
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
var (
|
|
channelMonitorMaintenanceOnce sync.Once
|
|
channelMonitorMaintenanceStopMu sync.Mutex
|
|
channelMonitorMaintenanceStopCh chan struct{}
|
|
channelMonitorMaintenanceStopped bool
|
|
|
|
channelMonitorMaintenanceRunMu sync.Mutex
|
|
channelMonitorMaintenanceActive bool
|
|
)
|
|
|
|
func StartChannelMonitorMaintenanceTask() {
|
|
channelMonitorMaintenanceOnce.Do(func() {
|
|
channelMonitorMaintenanceStopMu.Lock()
|
|
defer channelMonitorMaintenanceStopMu.Unlock()
|
|
|
|
channelMonitorMaintenanceStopCh = make(chan struct{})
|
|
channelMonitorMaintenanceStopped = false
|
|
go runChannelMonitorMaintenanceLoop(channelMonitorMaintenanceStopCh)
|
|
})
|
|
}
|
|
|
|
func StopChannelMonitorMaintenanceTask() {
|
|
channelMonitorMaintenanceStopMu.Lock()
|
|
defer channelMonitorMaintenanceStopMu.Unlock()
|
|
|
|
if channelMonitorMaintenanceStopCh != nil && !channelMonitorMaintenanceStopped {
|
|
close(channelMonitorMaintenanceStopCh)
|
|
channelMonitorMaintenanceStopped = true
|
|
}
|
|
}
|
|
|
|
func RunChannelMonitorMaintenanceOnce(now time.Time) error {
|
|
if !markChannelMonitorMaintenanceRunning() {
|
|
return nil
|
|
}
|
|
defer unmarkChannelMonitorMaintenanceRunning()
|
|
|
|
var errs []error
|
|
if _, err := AggregateChannelMonitorDailyRollupsForDate(now.AddDate(0, 0, -1)); err != nil {
|
|
errs = append(errs, fmt.Errorf("aggregate yesterday rollup: %w", err))
|
|
}
|
|
if _, err := AggregateChannelMonitorDailyRollupsForDate(now); err != nil {
|
|
errs = append(errs, fmt.Errorf("aggregate today rollup: %w", err))
|
|
}
|
|
if _, err := CleanOldChannelMonitorHistory(ChannelMonitorHistoryRetention, now); err != nil {
|
|
errs = append(errs, fmt.Errorf("clean old history: %w", err))
|
|
}
|
|
return errors.Join(errs...)
|
|
}
|
|
|
|
func runChannelMonitorMaintenanceLoop(stopCh <-chan struct{}) {
|
|
if err := RunChannelMonitorMaintenanceOnce(time.Now()); err != nil {
|
|
log.Printf("channel monitor maintenance: %v", err)
|
|
}
|
|
|
|
ticker := time.NewTicker(time.Hour)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case now := <-ticker.C:
|
|
if err := RunChannelMonitorMaintenanceOnce(now); err != nil {
|
|
log.Printf("channel monitor maintenance: %v", err)
|
|
}
|
|
case <-stopCh:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func markChannelMonitorMaintenanceRunning() bool {
|
|
channelMonitorMaintenanceRunMu.Lock()
|
|
defer channelMonitorMaintenanceRunMu.Unlock()
|
|
|
|
if channelMonitorMaintenanceActive {
|
|
return false
|
|
}
|
|
channelMonitorMaintenanceActive = true
|
|
return true
|
|
}
|
|
|
|
func unmarkChannelMonitorMaintenanceRunning() {
|
|
channelMonitorMaintenanceRunMu.Lock()
|
|
defer channelMonitorMaintenanceRunMu.Unlock()
|
|
|
|
channelMonitorMaintenanceActive = false
|
|
}
|