feat(job): enhance health checker and data cleaner with panic recovery and graceful shutdown
- Added panic recovery to health checker and data cleaner to prevent crashes from unexpected errors. - Implemented graceful shutdown for both components, ensuring they complete ongoing tasks before stopping. - Updated the job import in the load module to manage health checker and data cleaner lifecycle. - Introduced new channels to signal completion of health checks and data cleanups.
This commit is contained in:
parent
1a5e1db2ec
commit
5c1664e882
5 changed files with 553 additions and 388 deletions
856
data/bindata.go
856
data/bindata.go
File diff suppressed because one or more lines are too long
|
|
@ -31,6 +31,7 @@ import (
|
||||||
"github.com/yaoapp/yao/flow"
|
"github.com/yaoapp/yao/flow"
|
||||||
"github.com/yaoapp/yao/fs"
|
"github.com/yaoapp/yao/fs"
|
||||||
"github.com/yaoapp/yao/i18n"
|
"github.com/yaoapp/yao/i18n"
|
||||||
|
"github.com/yaoapp/yao/job"
|
||||||
"github.com/yaoapp/yao/kb"
|
"github.com/yaoapp/yao/kb"
|
||||||
"github.com/yaoapp/yao/mcp"
|
"github.com/yaoapp/yao/mcp"
|
||||||
"github.com/yaoapp/yao/messenger"
|
"github.com/yaoapp/yao/messenger"
|
||||||
|
|
@ -477,6 +478,10 @@ func Unload() (err error) {
|
||||||
// Stop Runtime
|
// Stop Runtime
|
||||||
err = runtime.Stop()
|
err = runtime.Stop()
|
||||||
|
|
||||||
|
// Stop Job health checker and data cleaner before closing DB
|
||||||
|
job.StopHealthChecker()
|
||||||
|
job.StopDataCleaner()
|
||||||
|
|
||||||
// Close DB
|
// Close DB
|
||||||
err = share.DBClose()
|
err = share.DBClose()
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -4,6 +4,7 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
jsoniter "github.com/json-iterator/go"
|
jsoniter "github.com/json-iterator/go"
|
||||||
|
|
@ -16,6 +17,8 @@ type HealthChecker struct {
|
||||||
interval time.Duration
|
interval time.Duration
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
|
done chan struct{}
|
||||||
|
started atomic.Bool
|
||||||
}
|
}
|
||||||
|
|
||||||
var globalHealthChecker *HealthChecker
|
var globalHealthChecker *HealthChecker
|
||||||
|
|
@ -27,11 +30,20 @@ func NewHealthChecker(interval time.Duration) *HealthChecker {
|
||||||
interval: interval,
|
interval: interval,
|
||||||
ctx: ctx,
|
ctx: ctx,
|
||||||
cancel: cancel,
|
cancel: cancel,
|
||||||
|
done: make(chan struct{}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start starts the health check goroutine
|
// Start starts the health check goroutine
|
||||||
func (hc *HealthChecker) Start() {
|
func (hc *HealthChecker) Start() {
|
||||||
|
hc.started.Store(true)
|
||||||
|
defer close(hc.done)
|
||||||
|
defer func() {
|
||||||
|
if r := recover(); r != nil {
|
||||||
|
log.Error("Health checker recovered from panic: %v", r)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
ticker := time.NewTicker(hc.interval)
|
ticker := time.NewTicker(hc.interval)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
|
|
@ -40,7 +52,7 @@ func (hc *HealthChecker) Start() {
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
if err := hc.performHealthCheck(); err != nil {
|
if err := hc.safePerformHealthCheck(); err != nil {
|
||||||
log.Error("Health check failed: %v", err)
|
log.Error("Health check failed: %v", err)
|
||||||
}
|
}
|
||||||
case <-hc.ctx.Done():
|
case <-hc.ctx.Done():
|
||||||
|
|
@ -50,11 +62,28 @@ func (hc *HealthChecker) Start() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stop stops the health checker
|
// safePerformHealthCheck wraps performHealthCheck with panic recovery
|
||||||
|
func (hc *HealthChecker) safePerformHealthCheck() (err error) {
|
||||||
|
defer func() {
|
||||||
|
if r := recover(); r != nil {
|
||||||
|
err = fmt.Errorf("health check panic: %v", r)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
return hc.performHealthCheck()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stop stops the health checker and waits for the goroutine to exit
|
||||||
func (hc *HealthChecker) Stop() {
|
func (hc *HealthChecker) Stop() {
|
||||||
if hc.cancel != nil {
|
if hc.cancel != nil {
|
||||||
hc.cancel()
|
hc.cancel()
|
||||||
}
|
}
|
||||||
|
if hc.started.Load() {
|
||||||
|
select {
|
||||||
|
case <-hc.done:
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
log.Error("Health checker stop timed out")
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// performHealthCheck performs health check
|
// performHealthCheck performs health check
|
||||||
|
|
@ -261,6 +290,8 @@ func GetHealthChecker() *HealthChecker {
|
||||||
type DataCleaner struct {
|
type DataCleaner struct {
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
|
done chan struct{}
|
||||||
|
started atomic.Bool
|
||||||
retentionDays int
|
retentionDays int
|
||||||
lastCleanupTime time.Time
|
lastCleanupTime time.Time
|
||||||
}
|
}
|
||||||
|
|
@ -273,14 +304,22 @@ func NewDataCleaner(retentionDays int) *DataCleaner {
|
||||||
return &DataCleaner{
|
return &DataCleaner{
|
||||||
ctx: ctx,
|
ctx: ctx,
|
||||||
cancel: cancel,
|
cancel: cancel,
|
||||||
|
done: make(chan struct{}),
|
||||||
retentionDays: retentionDays,
|
retentionDays: retentionDays,
|
||||||
lastCleanupTime: time.Now(), // Initialize to avoid immediate cleanup on startup
|
lastCleanupTime: time.Now(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start starts the daily data cleanup routine
|
// Start starts the daily data cleanup routine
|
||||||
func (dc *DataCleaner) Start() {
|
func (dc *DataCleaner) Start() {
|
||||||
// Check every hour if daily cleanup is needed
|
dc.started.Store(true)
|
||||||
|
defer close(dc.done)
|
||||||
|
defer func() {
|
||||||
|
if r := recover(); r != nil {
|
||||||
|
log.Error("Data cleaner recovered from panic: %v", r)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
ticker := time.NewTicker(1 * time.Hour)
|
ticker := time.NewTicker(1 * time.Hour)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
|
|
@ -290,7 +329,7 @@ func (dc *DataCleaner) Start() {
|
||||||
select {
|
select {
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
if dc.shouldRunCleanup() {
|
if dc.shouldRunCleanup() {
|
||||||
if err := dc.performCleanup(); err != nil {
|
if err := dc.safePerformCleanup(); err != nil {
|
||||||
log.Error("Data cleanup failed: %v", err)
|
log.Error("Data cleanup failed: %v", err)
|
||||||
} else {
|
} else {
|
||||||
dc.lastCleanupTime = time.Now()
|
dc.lastCleanupTime = time.Now()
|
||||||
|
|
@ -303,11 +342,28 @@ func (dc *DataCleaner) Start() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stop stops the data cleaner
|
// safePerformCleanup wraps performCleanup with panic recovery
|
||||||
|
func (dc *DataCleaner) safePerformCleanup() (err error) {
|
||||||
|
defer func() {
|
||||||
|
if r := recover(); r != nil {
|
||||||
|
err = fmt.Errorf("data cleanup panic: %v", r)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
return dc.performCleanup()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stop stops the data cleaner and waits for the goroutine to exit
|
||||||
func (dc *DataCleaner) Stop() {
|
func (dc *DataCleaner) Stop() {
|
||||||
if dc.cancel != nil {
|
if dc.cancel != nil {
|
||||||
dc.cancel()
|
dc.cancel()
|
||||||
}
|
}
|
||||||
|
if dc.started.Load() {
|
||||||
|
select {
|
||||||
|
case <-dc.done:
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
log.Error("Data cleaner stop timed out")
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// shouldRunCleanup checks if cleanup should run (once per day)
|
// shouldRunCleanup checks if cleanup should run (once per day)
|
||||||
|
|
|
||||||
|
|
@ -863,6 +863,12 @@ func issueTokens(ctx context.Context, params *IssueTokensParams) (*LoginResponse
|
||||||
|
|
||||||
// prepareUserKBCollection prepares KB collection for user (called asynchronously after login)
|
// prepareUserKBCollection prepares KB collection for user (called asynchronously after login)
|
||||||
func prepareUserKBCollection(userID, teamID, locale string) {
|
func prepareUserKBCollection(userID, teamID, locale string) {
|
||||||
|
defer func() {
|
||||||
|
if r := recover(); r != nil {
|
||||||
|
log.Warn("prepareUserKBCollection recovered from panic: %v", r)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
// Get global KB setting
|
// Get global KB setting
|
||||||
kbSetting := assistant.GetGlobalKBSetting()
|
kbSetting := assistant.GetGlobalKBSetting()
|
||||||
if kbSetting == nil || kbSetting.Chat == nil {
|
if kbSetting == nil || kbSetting.Chat == nil {
|
||||||
|
|
|
||||||
|
|
@ -102,7 +102,7 @@
|
||||||
"comment": "User-specified complete file path",
|
"comment": "User-specified complete file path",
|
||||||
"length": 1000,
|
"length": 1000,
|
||||||
"nullable": true,
|
"nullable": true,
|
||||||
"index": true
|
"index": false
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
"name": "path",
|
"name": "path",
|
||||||
|
|
@ -111,7 +111,7 @@
|
||||||
"comment": "Actual storage path for the file",
|
"comment": "Actual storage path for the file",
|
||||||
"length": 1000,
|
"length": 1000,
|
||||||
"nullable": false,
|
"nullable": false,
|
||||||
"index": true
|
"index": false
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
"name": "groups",
|
"name": "groups",
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue