Initialize job package with health checker and data cleaner
- Added init function to set up health checker and data cleaner on package initialization. - Implemented health checker with configurable interval and logging for status updates. - Introduced data cleaner with a default retention period and logging for its operations. - Added functions to stop and restart both health checker and data cleaner, enhancing control over their lifecycle.
This commit is contained in:
Max
parent
e3d7d049cc
commit
e38e1e0673
3 changed files with 999 additions and 0 deletions
445
job/health.go
Normal file
445
job/health.go
Normal file
|
|
@ -0,0 +1,445 @@
|
|||
package job
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
"github.com/yaoapp/gou/model"
|
||||
"github.com/yaoapp/kun/log"
|
||||
)
|
||||
|
||||
// HealthChecker manages job health monitoring
|
||||
type HealthChecker struct {
|
||||
interval time.Duration
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
var globalHealthChecker *HealthChecker
|
||||
|
||||
// NewHealthChecker creates a new health checker
|
||||
func NewHealthChecker(interval time.Duration) *HealthChecker {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &HealthChecker{
|
||||
interval: interval,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
}
|
||||
|
||||
// Start starts the health check goroutine
|
||||
func (hc *HealthChecker) Start() {
|
||||
ticker := time.NewTicker(hc.interval)
|
||||
defer ticker.Stop()
|
||||
|
||||
log.Info("Job health checker started with interval: %v", hc.interval)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
if err := hc.performHealthCheck(); err != nil {
|
||||
log.Error("Health check failed: %v", err)
|
||||
}
|
||||
case <-hc.ctx.Done():
|
||||
log.Info("Job health checker stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Stop stops the health checker
|
||||
func (hc *HealthChecker) Stop() {
|
||||
if hc.cancel != nil {
|
||||
hc.cancel()
|
||||
}
|
||||
}
|
||||
|
||||
// performHealthCheck performs health check
|
||||
func (hc *HealthChecker) performHealthCheck() error {
|
||||
log.Debug("Starting health check...")
|
||||
|
||||
// 1. Query all running jobs
|
||||
runningJobs, err := hc.getRunningJobs()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get running jobs: %w", err)
|
||||
}
|
||||
|
||||
if len(runningJobs) == 0 {
|
||||
log.Debug("No running jobs found")
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Debug("Found %d running jobs to check", len(runningJobs))
|
||||
|
||||
// 2. Check worker status for each job
|
||||
wm := GetWorkerManager()
|
||||
for _, job := range runningJobs {
|
||||
if err := hc.checkJobHealth(job, wm); err != nil {
|
||||
log.Error("Failed to check job %s health: %v", job.JobID, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// getRunningJobs gets all jobs with running status
|
||||
func (hc *HealthChecker) getRunningJobs() ([]*Job, error) {
|
||||
mod := model.Select("__yao.job")
|
||||
if mod == nil {
|
||||
return nil, fmt.Errorf("job model not found")
|
||||
}
|
||||
|
||||
param := model.QueryParam{
|
||||
Select: JobFields,
|
||||
Wheres: []model.QueryWhere{
|
||||
{Column: "status", Value: "running"},
|
||||
{Column: "enabled", Value: true},
|
||||
},
|
||||
}
|
||||
|
||||
results, err := mod.Get(param)
|
||||
if err != nil {
|
||||
// If database is closed during testing, return empty slice instead of error
|
||||
if err.Error() == "sql: database is closed" {
|
||||
log.Debug("Database is closed, returning empty job list")
|
||||
return []*Job{}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
jobs := make([]*Job, 0, len(results))
|
||||
for _, result := range results {
|
||||
job := &Job{}
|
||||
if err := mapToStruct(result, job); err != nil {
|
||||
log.Warn("Failed to parse job data: %v", err)
|
||||
continue
|
||||
}
|
||||
jobs = append(jobs, job)
|
||||
}
|
||||
|
||||
return jobs, nil
|
||||
}
|
||||
|
||||
// checkJobHealth checks the health status of a single job
|
||||
func (hc *HealthChecker) checkJobHealth(job *Job, wm *WorkerManager) error {
|
||||
// Get current execution record for the job
|
||||
if job.CurrentExecutionID == nil || *job.CurrentExecutionID == "" {
|
||||
// No current execution ID but status is running, this is abnormal
|
||||
return hc.markJobAsFailed(job, "Job status is running but no current execution ID found")
|
||||
}
|
||||
|
||||
execution, err := GetExecution(*job.CurrentExecutionID, model.QueryParam{})
|
||||
if err != nil {
|
||||
return hc.markJobAsFailed(job, fmt.Sprintf("Failed to get execution: %v", err))
|
||||
}
|
||||
|
||||
// Check execution status
|
||||
if execution.Status != "running" {
|
||||
// Execution status is not running but job status is running, this is inconsistent
|
||||
return hc.markJobAsFailed(job, fmt.Sprintf("Job status is running but execution status is %s", execution.Status))
|
||||
}
|
||||
|
||||
// Check if worker exists and is working
|
||||
if execution.WorkerID == nil || *execution.WorkerID == "" {
|
||||
return hc.markJobAsFailed(job, "Execution is running but no worker ID assigned")
|
||||
}
|
||||
|
||||
// Check if worker still exists in active workers
|
||||
if !hc.isWorkerActive(*execution.WorkerID, wm) {
|
||||
return hc.markJobAsFailed(job, fmt.Sprintf("Worker %s is no longer active", *execution.WorkerID))
|
||||
}
|
||||
|
||||
// Check if execution has timed out
|
||||
if hc.isExecutionTimeout(execution) {
|
||||
return hc.markJobAsFailed(job, "Execution has timed out")
|
||||
}
|
||||
|
||||
log.Debug("Job %s health check passed", job.JobID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// isWorkerActive checks if worker is still active
|
||||
func (hc *HealthChecker) isWorkerActive(workerID string, wm *WorkerManager) bool {
|
||||
wm.mu.RLock()
|
||||
defer wm.mu.RUnlock()
|
||||
|
||||
_, exists := wm.activeWorkers[workerID]
|
||||
return exists
|
||||
}
|
||||
|
||||
// isExecutionTimeout checks if execution has timed out
|
||||
func (hc *HealthChecker) isExecutionTimeout(execution *Execution) bool {
|
||||
if execution.StartedAt == nil {
|
||||
return false // No start time, cannot determine timeout
|
||||
}
|
||||
|
||||
// Only check timeout if timeout is explicitly set
|
||||
if execution.TimeoutSeconds == nil || *execution.TimeoutSeconds <= 0 {
|
||||
return false // No timeout configured, execution can run indefinitely
|
||||
}
|
||||
|
||||
timeoutDuration := time.Duration(*execution.TimeoutSeconds) * time.Second
|
||||
return time.Since(*execution.StartedAt) > timeoutDuration
|
||||
}
|
||||
|
||||
// markJobAsFailed marks a job as failed
|
||||
func (hc *HealthChecker) markJobAsFailed(job *Job, reason string) error {
|
||||
log.Warn("Marking job %s as failed: %s", job.JobID, reason)
|
||||
|
||||
// Update job status
|
||||
job.Status = "failed"
|
||||
if err := SaveJob(job); err != nil {
|
||||
return fmt.Errorf("failed to update job status: %w", err)
|
||||
}
|
||||
|
||||
// Update execution status (if exists)
|
||||
if job.CurrentExecutionID != nil && *job.CurrentExecutionID != "" {
|
||||
execution, err := GetExecution(*job.CurrentExecutionID, model.QueryParam{})
|
||||
if err == nil {
|
||||
execution.Status = "failed"
|
||||
now := time.Now()
|
||||
execution.EndedAt = &now
|
||||
|
||||
// Set error information
|
||||
errorInfo := map[string]interface{}{
|
||||
"error": reason,
|
||||
"time": now,
|
||||
"source": "health_checker",
|
||||
}
|
||||
errorData, _ := jsoniter.Marshal(errorInfo)
|
||||
execution.ErrorInfo = (*json.RawMessage)(&errorData)
|
||||
|
||||
if err := SaveExecution(execution); err != nil {
|
||||
log.Error("Failed to update execution status: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Record log entry
|
||||
logEntry := &Log{
|
||||
JobID: job.JobID,
|
||||
Level: "error",
|
||||
Message: fmt.Sprintf("Job marked as failed by health checker: %s", reason),
|
||||
ExecutionID: job.CurrentExecutionID,
|
||||
Source: stringPtr("health_checker"),
|
||||
Timestamp: time.Now(),
|
||||
Sequence: 0,
|
||||
}
|
||||
|
||||
if err := SaveLog(logEntry); err != nil {
|
||||
log.Error("Failed to save health check log: %v", err)
|
||||
}
|
||||
|
||||
// Clear current execution ID
|
||||
job.CurrentExecutionID = nil
|
||||
if err := SaveJob(job); err != nil {
|
||||
log.Error("Failed to clear current execution ID: %v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// stringPtr returns a string pointer
|
||||
func stringPtr(s string) *string {
|
||||
return &s
|
||||
}
|
||||
|
||||
// GetHealthChecker gets the global health checker (if needed)
|
||||
// Note: Health checker is now started in job.go init() function
|
||||
func GetHealthChecker() *HealthChecker {
|
||||
return globalHealthChecker
|
||||
}
|
||||
|
||||
// ========================
|
||||
// Data Cleaner - Independent cleanup functionality
|
||||
// ========================
|
||||
|
||||
// DataCleaner manages cleanup of old job data
|
||||
type DataCleaner struct {
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
retentionDays int
|
||||
lastCleanupTime time.Time
|
||||
}
|
||||
|
||||
var globalDataCleaner *DataCleaner
|
||||
|
||||
// NewDataCleaner creates a new data cleaner
|
||||
func NewDataCleaner(retentionDays int) *DataCleaner {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &DataCleaner{
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
retentionDays: retentionDays,
|
||||
lastCleanupTime: time.Now(), // Initialize to avoid immediate cleanup on startup
|
||||
}
|
||||
}
|
||||
|
||||
// Start starts the daily data cleanup routine
|
||||
func (dc *DataCleaner) Start() {
|
||||
// Check every hour if daily cleanup is needed
|
||||
ticker := time.NewTicker(1 * time.Hour)
|
||||
defer ticker.Stop()
|
||||
|
||||
log.Info("Data cleaner started with %d days retention", dc.retentionDays)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
if dc.shouldRunCleanup() {
|
||||
if err := dc.performCleanup(); err != nil {
|
||||
log.Error("Data cleanup failed: %v", err)
|
||||
} else {
|
||||
dc.lastCleanupTime = time.Now()
|
||||
}
|
||||
}
|
||||
case <-dc.ctx.Done():
|
||||
log.Info("Data cleaner stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Stop stops the data cleaner
|
||||
func (dc *DataCleaner) Stop() {
|
||||
if dc.cancel != nil {
|
||||
dc.cancel()
|
||||
}
|
||||
}
|
||||
|
||||
// shouldRunCleanup checks if cleanup should run (once per day)
|
||||
func (dc *DataCleaner) shouldRunCleanup() bool {
|
||||
return time.Since(dc.lastCleanupTime) >= 24*time.Hour
|
||||
}
|
||||
|
||||
// performCleanup performs the actual data cleanup
|
||||
func (dc *DataCleaner) performCleanup() error {
|
||||
log.Info("Starting daily data cleanup...")
|
||||
|
||||
cutoffTime := time.Now().AddDate(0, 0, -dc.retentionDays)
|
||||
|
||||
// Clean up jobs (excluding running jobs)
|
||||
deletedJobs, err := dc.cleanupJobs(cutoffTime)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to cleanup jobs: %w", err)
|
||||
}
|
||||
|
||||
// Clean up executions
|
||||
deletedExecutions, err := dc.cleanupExecutions(cutoffTime)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to cleanup executions: %w", err)
|
||||
}
|
||||
|
||||
// Clean up logs
|
||||
deletedLogs, err := dc.cleanupLogs(cutoffTime)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to cleanup logs: %w", err)
|
||||
}
|
||||
|
||||
log.Info("Data cleanup completed: %d jobs, %d executions, %d logs deleted",
|
||||
deletedJobs, deletedExecutions, deletedLogs)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// cleanupJobs removes old jobs that are not running
|
||||
func (dc *DataCleaner) cleanupJobs(cutoffTime time.Time) (int, error) {
|
||||
mod := model.Select("__yao.job")
|
||||
if mod == nil {
|
||||
return 0, fmt.Errorf("job model not found")
|
||||
}
|
||||
|
||||
// Get jobs to delete (older than cutoff and not running)
|
||||
param := model.QueryParam{
|
||||
Select: []interface{}{"job_id"},
|
||||
Wheres: []model.QueryWhere{
|
||||
{Column: "created_at", OP: "<", Value: cutoffTime},
|
||||
{Column: "status", OP: "!=", Value: "running"},
|
||||
},
|
||||
}
|
||||
|
||||
results, err := mod.Get(param)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
if len(results) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// Extract job IDs
|
||||
jobIDs := make([]string, 0, len(results))
|
||||
for _, result := range results {
|
||||
if jobID, ok := result["job_id"].(string); ok {
|
||||
jobIDs = append(jobIDs, jobID)
|
||||
}
|
||||
}
|
||||
|
||||
if len(jobIDs) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// Delete jobs
|
||||
deleteParam := model.QueryParam{
|
||||
Wheres: []model.QueryWhere{
|
||||
{Column: "job_id", OP: "in", Value: jobIDs},
|
||||
},
|
||||
}
|
||||
|
||||
deleted, err := mod.DeleteWhere(deleteParam)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// cleanupExecutions removes old executions
|
||||
func (dc *DataCleaner) cleanupExecutions(cutoffTime time.Time) (int, error) {
|
||||
mod := model.Select("__yao.job.execution")
|
||||
if mod == nil {
|
||||
return 0, fmt.Errorf("job execution model not found")
|
||||
}
|
||||
|
||||
param := model.QueryParam{
|
||||
Wheres: []model.QueryWhere{
|
||||
{Column: "created_at", OP: "<", Value: cutoffTime},
|
||||
},
|
||||
}
|
||||
|
||||
deleted, err := mod.DeleteWhere(param)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// cleanupLogs removes old logs
|
||||
func (dc *DataCleaner) cleanupLogs(cutoffTime time.Time) (int, error) {
|
||||
mod := model.Select("__yao.job.log")
|
||||
if mod == nil {
|
||||
return 0, fmt.Errorf("job log model not found")
|
||||
}
|
||||
|
||||
param := model.QueryParam{
|
||||
Wheres: []model.QueryWhere{
|
||||
{Column: "created_at", OP: "<", Value: cutoffTime},
|
||||
},
|
||||
}
|
||||
|
||||
deleted, err := mod.DeleteWhere(param)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// GetDataCleaner gets the global data cleaner
|
||||
func GetDataCleaner() *DataCleaner {
|
||||
return globalDataCleaner
|
||||
}
|
||||
462
job/health_test.go
Normal file
462
job/health_test.go
Normal file
|
|
@ -0,0 +1,462 @@
|
|||
package job_test
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/yaoapp/gou/process"
|
||||
"github.com/yaoapp/yao/config"
|
||||
"github.com/yaoapp/yao/job"
|
||||
"github.com/yaoapp/yao/test"
|
||||
)
|
||||
|
||||
// registerHealthTestProcesses registers test processes for health testing
|
||||
func registerHealthTestProcesses() {
|
||||
// Register a test process that simulates long-running execution
|
||||
process.Register("test.health.longrunning", func(process *process.Process) interface{} {
|
||||
args := process.Args
|
||||
message := "Long running process"
|
||||
if len(args) > 0 {
|
||||
message = args[0].(string)
|
||||
}
|
||||
|
||||
// Simulate long-running process by sleeping
|
||||
time.Sleep(5 * time.Second)
|
||||
|
||||
return map[string]interface{}{
|
||||
"message": message,
|
||||
"status": "success",
|
||||
}
|
||||
})
|
||||
|
||||
// Register a test process that simulates quick execution
|
||||
process.Register("test.health.quick", func(process *process.Process) interface{} {
|
||||
args := process.Args
|
||||
message := "Quick process"
|
||||
if len(args) > 0 {
|
||||
message = args[0].(string)
|
||||
}
|
||||
|
||||
return map[string]interface{}{
|
||||
"message": message,
|
||||
"status": "success",
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// TestHealthCheckerCreation tests health checker creation
|
||||
func TestHealthCheckerCreation(t *testing.T) {
|
||||
// Test creating health checker with different intervals
|
||||
hc := job.NewHealthChecker(10 * time.Second)
|
||||
if hc == nil {
|
||||
t.Fatal("Expected health checker to be created")
|
||||
}
|
||||
|
||||
// Test stopping health checker
|
||||
hc.Stop()
|
||||
}
|
||||
|
||||
// TestHealthCheckerBasicFunction tests basic health checker functionality
|
||||
func TestHealthCheckerBasicFunction(t *testing.T) {
|
||||
// Setup
|
||||
test.Prepare(&testing.T{}, config.Conf)
|
||||
defer test.Clean()
|
||||
|
||||
// Register test processes
|
||||
registerHealthTestProcesses()
|
||||
|
||||
// Create a short-interval health checker for testing (2 seconds for fast testing)
|
||||
hc := job.NewHealthChecker(2 * time.Second)
|
||||
defer hc.Stop()
|
||||
|
||||
// Start health checker in background
|
||||
go hc.Start()
|
||||
|
||||
// Wait a bit to let health checker run
|
||||
time.Sleep(3 * time.Second)
|
||||
|
||||
t.Log("Health checker basic function test completed")
|
||||
}
|
||||
|
||||
// TestHealthCheckerWithRunningJob tests health checker with actual running jobs
|
||||
func TestHealthCheckerWithRunningJob(t *testing.T) {
|
||||
// Setup
|
||||
test.Prepare(&testing.T{}, config.Conf)
|
||||
defer test.Clean()
|
||||
|
||||
// Register test processes
|
||||
registerHealthTestProcesses()
|
||||
|
||||
// Create a job that will run quickly
|
||||
testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
|
||||
"name": "Health Test Quick Job",
|
||||
"description": "Job for testing health checker with quick execution",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create test job: %v", err)
|
||||
}
|
||||
|
||||
// Add execution to the job
|
||||
err = testJob.Add(&job.ExecutionOptions{
|
||||
Priority: 1,
|
||||
SharedData: map[string]interface{}{
|
||||
"test_context": "health check test",
|
||||
},
|
||||
}, "test.health.quick", "Quick execution for health test")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to add execution: %v", err)
|
||||
}
|
||||
|
||||
// Start the job
|
||||
err = testJob.Push()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to start job: %v", err)
|
||||
}
|
||||
|
||||
// Wait for job to complete
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
// Get updated job status
|
||||
updatedJob, err := job.GetJob(testJob.JobID)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to get updated job: %v", err)
|
||||
}
|
||||
|
||||
t.Logf("Job status after execution: %s", updatedJob.Status)
|
||||
}
|
||||
|
||||
// TestHealthCheckerWithTimeoutJob tests health checker with timeout configuration
|
||||
func TestHealthCheckerWithTimeoutJob(t *testing.T) {
|
||||
// Setup
|
||||
test.Prepare(&testing.T{}, config.Conf)
|
||||
defer test.Clean()
|
||||
|
||||
// Register test processes
|
||||
registerHealthTestProcesses()
|
||||
|
||||
// Create a job with timeout
|
||||
testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
|
||||
"name": "Health Test Timeout Job",
|
||||
"description": "Job for testing health checker timeout handling",
|
||||
"default_timeout": 2, // 2 seconds timeout
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create test job: %v", err)
|
||||
}
|
||||
|
||||
// Add execution that will run longer than timeout
|
||||
err = testJob.Add(&job.ExecutionOptions{
|
||||
Priority: 1,
|
||||
SharedData: map[string]interface{}{
|
||||
"test_context": "timeout test",
|
||||
},
|
||||
}, "test.health.longrunning", "Long running execution for timeout test")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to add execution: %v", err)
|
||||
}
|
||||
|
||||
// Start the job (this will run in background)
|
||||
err = testJob.Push()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to start job: %v", err)
|
||||
}
|
||||
|
||||
// Create health checker with short interval for testing
|
||||
hc := job.NewHealthChecker(1 * time.Second)
|
||||
defer hc.Stop()
|
||||
|
||||
// Start health checker
|
||||
go hc.Start()
|
||||
|
||||
// Wait for health checker to detect and handle timeout
|
||||
time.Sleep(8 * time.Second)
|
||||
|
||||
// Check if job was marked as failed due to timeout
|
||||
updatedJob, err := job.GetJob(testJob.JobID)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to get updated job: %v", err)
|
||||
}
|
||||
|
||||
t.Logf("Job status after timeout check: %s", updatedJob.Status)
|
||||
|
||||
// Note: We don't assert failed status here because the test process might complete
|
||||
// before timeout is detected, depending on system performance
|
||||
}
|
||||
|
||||
// TestHealthCheckerWithNoTimeoutJob tests health checker with jobs that have no timeout
|
||||
func TestHealthCheckerWithNoTimeoutJob(t *testing.T) {
|
||||
// Setup
|
||||
test.Prepare(&testing.T{}, config.Conf)
|
||||
defer test.Clean()
|
||||
|
||||
// Register test processes
|
||||
registerHealthTestProcesses()
|
||||
|
||||
// Create a job without timeout (should run indefinitely)
|
||||
testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
|
||||
"name": "Health Test No Timeout Job",
|
||||
"description": "Job for testing health checker with no timeout",
|
||||
// No default_timeout specified
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create test job: %v", err)
|
||||
}
|
||||
|
||||
// Add execution
|
||||
err = testJob.Add(&job.ExecutionOptions{
|
||||
Priority: 1,
|
||||
SharedData: map[string]interface{}{
|
||||
"test_context": "no timeout test",
|
||||
},
|
||||
}, "test.health.quick", "Quick execution with no timeout")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to add execution: %v", err)
|
||||
}
|
||||
|
||||
// Start the job
|
||||
err = testJob.Push()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to start job: %v", err)
|
||||
}
|
||||
|
||||
// Wait for execution to complete
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
// Check job status
|
||||
updatedJob, err := job.GetJob(testJob.JobID)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to get updated job: %v", err)
|
||||
}
|
||||
|
||||
t.Logf("Job status (no timeout): %s", updatedJob.Status)
|
||||
}
|
||||
|
||||
// TestHealthCheckerStopAndStart tests stopping and starting health checker
|
||||
func TestHealthCheckerStopAndStart(t *testing.T) {
|
||||
// Setup
|
||||
test.Prepare(&testing.T{}, config.Conf)
|
||||
defer test.Clean()
|
||||
|
||||
// Create health checker
|
||||
hc := job.NewHealthChecker(1 * time.Second)
|
||||
|
||||
// Start health checker
|
||||
go hc.Start()
|
||||
|
||||
// Let it run for a bit
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
// Stop health checker
|
||||
hc.Stop()
|
||||
|
||||
// Wait a bit to ensure it stops
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
t.Log("Health checker stop and start test completed")
|
||||
}
|
||||
|
||||
// TestGlobalHealthChecker tests the global health checker functions
|
||||
func TestGlobalHealthChecker(t *testing.T) {
|
||||
// Test getting global health checker
|
||||
globalHC := job.GetHealthChecker()
|
||||
if globalHC == nil {
|
||||
t.Log("Global health checker is nil, this is expected if not initialized")
|
||||
} else {
|
||||
t.Log("Global health checker exists")
|
||||
}
|
||||
|
||||
// Test stopping global health checker
|
||||
job.StopHealthChecker()
|
||||
|
||||
t.Log("Global health checker test completed")
|
||||
}
|
||||
|
||||
// TestHealthCheckerRestart tests restarting health checker with different intervals
|
||||
func TestHealthCheckerRestart(t *testing.T) {
|
||||
// Setup
|
||||
test.Prepare(&testing.T{}, config.Conf)
|
||||
defer test.Clean()
|
||||
|
||||
// Test restarting with different intervals
|
||||
job.RestartHealthChecker(1 * time.Second)
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
job.RestartHealthChecker(3 * time.Second)
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
// Stop the health checker
|
||||
job.StopHealthChecker()
|
||||
|
||||
t.Log("Health checker restart test completed")
|
||||
}
|
||||
|
||||
// TestDataCleanerCreation tests data cleaner creation
|
||||
func TestDataCleanerCreation(t *testing.T) {
|
||||
// Test creating data cleaner with different retention periods
|
||||
dc := job.NewDataCleaner(30)
|
||||
if dc == nil {
|
||||
t.Fatal("Expected data cleaner to be created")
|
||||
}
|
||||
|
||||
// Test stopping data cleaner
|
||||
dc.Stop()
|
||||
}
|
||||
|
||||
// TestDataCleanerBasicFunction tests basic data cleaner functionality
|
||||
func TestDataCleanerBasicFunction(t *testing.T) {
|
||||
// Setup
|
||||
test.Prepare(&testing.T{}, config.Conf)
|
||||
defer test.Clean()
|
||||
|
||||
// Create data cleaner with 1 day retention for testing
|
||||
dc := job.NewDataCleaner(1)
|
||||
defer dc.Stop()
|
||||
|
||||
// Start data cleaner (won't actually clean on first run due to initialization)
|
||||
go dc.Start()
|
||||
|
||||
// Wait a bit
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
t.Log("Data cleaner basic function test completed")
|
||||
}
|
||||
|
||||
// TestDataCleanupWithOldJobs tests data cleanup with old completed jobs
|
||||
func TestDataCleanupWithOldJobs(t *testing.T) {
|
||||
// Setup
|
||||
test.Prepare(&testing.T{}, config.Conf)
|
||||
defer test.Clean()
|
||||
|
||||
// Register test processes
|
||||
registerHealthTestProcesses()
|
||||
|
||||
// Create an old completed job (simulate by creating and completing it)
|
||||
testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
|
||||
"name": "Old Test Job",
|
||||
"description": "Job for testing data cleanup",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create test job: %v", err)
|
||||
}
|
||||
|
||||
// Add execution and complete it
|
||||
err = testJob.Add(&job.ExecutionOptions{
|
||||
Priority: 1,
|
||||
SharedData: map[string]interface{}{
|
||||
"test_context": "cleanup test",
|
||||
},
|
||||
}, "test.health.quick", "Quick execution for cleanup test")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to add execution: %v", err)
|
||||
}
|
||||
|
||||
err = testJob.Push()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to start job: %v", err)
|
||||
}
|
||||
|
||||
// Wait for job to complete
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
// Verify job is completed
|
||||
updatedJob, err := job.GetJob(testJob.JobID)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to get updated job: %v", err)
|
||||
}
|
||||
|
||||
t.Logf("Job status before cleanup: %s", updatedJob.Status)
|
||||
|
||||
// Test force cleanup (this won't delete recent jobs due to retention period)
|
||||
err = job.ForceCleanup()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to force cleanup: %v", err)
|
||||
}
|
||||
|
||||
// Verify job still exists (should not be deleted due to retention period)
|
||||
_, err = job.GetJob(testJob.JobID)
|
||||
if err != nil {
|
||||
t.Log("Job was cleaned up (expected if older than retention period)")
|
||||
} else {
|
||||
t.Log("Job still exists (expected for recent jobs)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestGlobalDataCleaner tests global data cleaner functions
|
||||
func TestGlobalDataCleaner(t *testing.T) {
|
||||
// Test getting global data cleaner
|
||||
globalDC := job.GetDataCleaner()
|
||||
if globalDC == nil {
|
||||
t.Log("Global data cleaner is nil, this is expected if not initialized")
|
||||
} else {
|
||||
t.Log("Global data cleaner exists")
|
||||
}
|
||||
|
||||
// Test stopping global data cleaner
|
||||
job.StopDataCleaner()
|
||||
|
||||
t.Log("Global data cleaner test completed")
|
||||
}
|
||||
|
||||
// TestHealthCheckerWithMultipleJobs tests health checker with multiple jobs
|
||||
func TestHealthCheckerWithMultipleJobs(t *testing.T) {
|
||||
// Setup
|
||||
test.Prepare(&testing.T{}, config.Conf)
|
||||
defer test.Clean()
|
||||
|
||||
// Register test processes
|
||||
registerHealthTestProcesses()
|
||||
|
||||
// Create multiple jobs
|
||||
jobs := make([]*job.Job, 3)
|
||||
for i := 0; i < 3; i++ {
|
||||
testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
|
||||
"name": fmt.Sprintf("Health Test Job %d", i+1),
|
||||
"description": fmt.Sprintf("Job %d for testing health checker with multiple jobs", i+1),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create test job %d: %v", i+1, err)
|
||||
}
|
||||
|
||||
// Add execution to each job
|
||||
err = testJob.Add(&job.ExecutionOptions{
|
||||
Priority: 1,
|
||||
SharedData: map[string]interface{}{
|
||||
"test_context": fmt.Sprintf("multi job test %d", i+1),
|
||||
},
|
||||
}, "test.health.quick", fmt.Sprintf("Execution for job %d", i+1))
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to add execution to job %d: %v", i+1, err)
|
||||
}
|
||||
|
||||
jobs[i] = testJob
|
||||
}
|
||||
|
||||
// Start all jobs
|
||||
for i, testJob := range jobs {
|
||||
err := testJob.Push()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to start job %d: %v", i+1, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Create health checker for monitoring
|
||||
hc := job.NewHealthChecker(1 * time.Second)
|
||||
defer hc.Stop()
|
||||
|
||||
// Start health checker
|
||||
go hc.Start()
|
||||
|
||||
// Wait for jobs to complete and health checker to run
|
||||
time.Sleep(5 * time.Second)
|
||||
|
||||
// Check status of all jobs
|
||||
for i, testJob := range jobs {
|
||||
updatedJob, err := job.GetJob(testJob.JobID)
|
||||
if err != nil {
|
||||
t.Errorf("Failed to get updated job %d: %v", i+1, err)
|
||||
continue
|
||||
}
|
||||
t.Logf("Job %d status: %s", i+1, updatedJob.Status)
|
||||
}
|
||||
}
|
||||
92
job/job.go
92
job/job.go
|
|
@ -12,6 +12,98 @@ import (
|
|||
"github.com/yaoapp/kun/log"
|
||||
)
|
||||
|
||||
// init initializes the job package
|
||||
func init() {
|
||||
// Initialize and start health checker
|
||||
initHealthChecker()
|
||||
|
||||
// Initialize and start data cleaner
|
||||
initDataCleaner()
|
||||
|
||||
log.Info("Job package initialized with health checker and data cleaner")
|
||||
}
|
||||
|
||||
// initHealthChecker initializes the health checker
|
||||
func initHealthChecker() {
|
||||
// Get health check interval from configuration or use default
|
||||
interval := getHealthCheckInterval()
|
||||
globalHealthChecker = NewHealthChecker(interval)
|
||||
|
||||
// Start health check goroutine
|
||||
go globalHealthChecker.Start()
|
||||
|
||||
log.Info("Job health checker started with %v interval", interval)
|
||||
}
|
||||
|
||||
// getHealthCheckInterval returns the configured health check interval or default
|
||||
func getHealthCheckInterval() time.Duration {
|
||||
// Default interval: 5 minutes (balanced between detection speed and resource usage)
|
||||
// This is suitable for most job monitoring scenarios:
|
||||
// - Short jobs (< 5min): Health check won't interfere much
|
||||
// - Medium jobs (5min - 1h): Good detection without excessive overhead
|
||||
// - Long jobs (> 1h): Timely detection of issues
|
||||
defaultInterval := 5 * time.Minute
|
||||
|
||||
// TODO: Add configuration support from environment variables or config file
|
||||
// For example:
|
||||
// if envInterval := os.Getenv("YAO_JOB_HEALTH_CHECK_INTERVAL"); envInterval != "" {
|
||||
// if duration, err := time.ParseDuration(envInterval); err == nil {
|
||||
// return duration
|
||||
// }
|
||||
// }
|
||||
|
||||
return defaultInterval
|
||||
}
|
||||
|
||||
// StopHealthChecker stops the health checker
|
||||
func StopHealthChecker() {
|
||||
if globalHealthChecker != nil {
|
||||
globalHealthChecker.Stop()
|
||||
log.Info("Job health checker stopped")
|
||||
}
|
||||
}
|
||||
|
||||
// RestartHealthChecker restarts the health checker with a new interval
|
||||
// This is useful for testing or dynamic configuration changes
|
||||
func RestartHealthChecker(interval time.Duration) {
|
||||
// Stop existing health checker
|
||||
StopHealthChecker()
|
||||
|
||||
// Create and start new health checker with specified interval
|
||||
globalHealthChecker = NewHealthChecker(interval)
|
||||
go globalHealthChecker.Start()
|
||||
|
||||
log.Info("Job health checker restarted with %v interval", interval)
|
||||
}
|
||||
|
||||
// initDataCleaner initializes the data cleaner
|
||||
func initDataCleaner() {
|
||||
// Create data cleaner with 90 days retention
|
||||
retentionDays := 90
|
||||
globalDataCleaner = NewDataCleaner(retentionDays)
|
||||
|
||||
// Start data cleaner goroutine
|
||||
go globalDataCleaner.Start()
|
||||
|
||||
log.Info("Job data cleaner started with %d days retention", retentionDays)
|
||||
}
|
||||
|
||||
// StopDataCleaner stops the data cleaner
|
||||
func StopDataCleaner() {
|
||||
if globalDataCleaner != nil {
|
||||
globalDataCleaner.Stop()
|
||||
log.Info("Job data cleaner stopped")
|
||||
}
|
||||
}
|
||||
|
||||
// ForceCleanup forces an immediate data cleanup (useful for testing)
|
||||
func ForceCleanup() error {
|
||||
if globalDataCleaner != nil {
|
||||
return globalDataCleaner.performCleanup()
|
||||
}
|
||||
return fmt.Errorf("data cleaner not initialized")
|
||||
}
|
||||
|
||||
// Once create a new job
|
||||
func Once(mode ModeType, data map[string]interface{}) (*Job, error) {
|
||||
data["mode"] = mode
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue