107 lines
2.5 KiB
Go
107 lines
2.5 KiB
Go
package ingest
|
|
|
|
import (
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// WorkerStatus 描述日志后台 worker 的当前运行状态。
|
|
type WorkerStatus struct {
|
|
Name string `json:"name"`
|
|
Running bool `json:"running"`
|
|
LastStartedAt time.Time `json:"last_started_at,omitempty"`
|
|
LastSucceededAt time.Time `json:"last_succeeded_at,omitempty"`
|
|
LastErrorAt time.Time `json:"last_error_at,omitempty"`
|
|
LastError string `json:"last_error,omitempty"`
|
|
QueueDepth *int64 `json:"queue_depth,omitempty"`
|
|
}
|
|
|
|
func markWorkerQueueDepth(name string, depth int64) {
|
|
lifecycle.Lock()
|
|
status := lifecycle.workers[name]
|
|
status.Name = name
|
|
status.QueueDepth = new(int64)
|
|
*status.QueueDepth = depth
|
|
lifecycle.workers[name] = status
|
|
lifecycle.Unlock()
|
|
}
|
|
|
|
var lifecycle = struct {
|
|
sync.RWMutex
|
|
wg sync.WaitGroup
|
|
workers map[string]WorkerStatus
|
|
}{workers: make(map[string]WorkerStatus)}
|
|
|
|
func startWorker(name string, run func() error) {
|
|
lifecycle.Lock()
|
|
lifecycle.workers[name] = WorkerStatus{Name: name, Running: true, LastStartedAt: time.Now().UTC()}
|
|
lifecycle.wg.Add(1)
|
|
lifecycle.Unlock()
|
|
go func() {
|
|
defer lifecycle.wg.Done()
|
|
err := run()
|
|
lifecycle.Lock()
|
|
status := lifecycle.workers[name]
|
|
status.Running = false
|
|
if err != nil {
|
|
status.LastErrorAt = time.Now().UTC()
|
|
status.LastError = err.Error()
|
|
}
|
|
lifecycle.workers[name] = status
|
|
lifecycle.Unlock()
|
|
}()
|
|
}
|
|
|
|
func markWorkerSucceeded(name string) {
|
|
lifecycle.Lock()
|
|
status := lifecycle.workers[name]
|
|
status.Name = name
|
|
status.LastSucceededAt = time.Now().UTC()
|
|
status.LastError = ""
|
|
lifecycle.workers[name] = status
|
|
lifecycle.Unlock()
|
|
}
|
|
|
|
func markWorkerError(name string, err error) {
|
|
if err == nil {
|
|
return
|
|
}
|
|
lifecycle.Lock()
|
|
status := lifecycle.workers[name]
|
|
status.Name = name
|
|
status.LastErrorAt = time.Now().UTC()
|
|
status.LastError = err.Error()
|
|
lifecycle.workers[name] = status
|
|
lifecycle.Unlock()
|
|
}
|
|
|
|
// WorkerStatuses 返回状态快照。
|
|
func WorkerStatuses() []WorkerStatus {
|
|
lifecycle.RLock()
|
|
defer lifecycle.RUnlock()
|
|
result := make([]WorkerStatus, 0, len(lifecycle.workers))
|
|
for _, status := range lifecycle.workers {
|
|
if status.QueueDepth != nil {
|
|
depth := *status.QueueDepth
|
|
status.QueueDepth = &depth
|
|
}
|
|
result = append(result, status)
|
|
}
|
|
return result
|
|
}
|
|
|
|
// Wait 等待后台 worker 退出，返回是否在超时内完成。
|
|
func Wait(timeout time.Duration) bool {
|
|
done := make(chan struct{})
|
|
go func() {
|
|
lifecycle.wg.Wait()
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
return true
|
|
case <-time.After(timeout):
|
|
return false
|
|
}
|
|
}
|