feat: add scheduler lifecycle monitoring <> #785 (#889)

* git commit -m "feat: add SchedulerStarted and SchedulerStopped monitoring"

* fix lint issue

* Update errors.go

Co-authored-by: John Roesler <[email protected]>

* feat() updated with remaning metrics & events  (JobRegistered/JobUnregistered, JobStarted/JobRunning/JobFailed/JobCompleted)

* feat: enhance scheduler and job observability by adding new monitor events for lifecycle, performance, and concurrency limits.

* docs: expand metrics section to include scheduler lifecycle events and `SchedulerMonitor` details with Prometheus example

* refactor: conditionally send scheduler notifications only when a scheduler monitor is configured.

---------

Co-authored-by: John Roesler <[email protected]>

Author: iyashjayesh

README.md

@@ -169,12 +169,52 @@ The Logger interface can be implemented with your desired logging library.
 The provided NewLogger uses the standard library's log package.
 
 ### Metrics
-Metrics may be collected from the execution of each job.
+Metrics may be collected from the execution of each job and scheduler lifecycle events.
 - [**Monitor**](https://pkg.go.dev/github.com/go-co-op/gocron/v2#Monitor):
 - [**MonitorStatus**](https://pkg.go.dev/github.com/go-co-op/gocron/v2#MonitorStatus) (includes status and error (if any) of the Job)
 A monitor can be used to collect metrics for each job from a scheduler.
   - Implementations: [go-co-op monitors](https://github.com/go-co-op?q=-monitor&type=all&language=&sort=)
     (don't see what you need? request on slack to get a repo created to contribute it!)
+- [**SchedulerMonitor**](https://pkg.go.dev/github.com/go-co-op/gocron/v2#SchedulerMonitor):
+A scheduler monitor provides comprehensive observability into scheduler and job lifecycle events.
+
+  **Available Metrics:**
+  - **Scheduler Lifecycle**: `SchedulerStarted`, `SchedulerStopped`, `SchedulerShutdown`
+  - **Job Management**: `JobRegistered`, `JobUnregistered` - track jobs added/removed from scheduler
+  - **Job Execution**: `JobStarted`, `JobRunning`, `JobCompleted`, `JobFailed` - monitor job execution flow
+  - **Performance**: `JobExecutionTime`, `JobSchedulingDelay` - measure job duration and scheduling lag
+  - **Concurrency**: `ConcurrencyLimitReached` - detect when singleton or limit mode constraints are hit
+
+  **Derived Metrics** (calculable from events):
+  - Error rate: `JobFailed / (JobCompleted + JobFailed)`
+  - Average execution time: from `JobExecutionTime` events
+  - Active jobs: `JobRegistered - JobUnregistered`
+  - Current queue depth: `JobStarted - (JobCompleted + JobFailed)`
+
+  **Example - Prometheus Integration:**
+  ```go
+  type PrometheusMonitor struct {
+      jobsCompleted   prometheus.Counter
+      jobsFailed      prometheus.Counter
+      executionTime   prometheus.Histogram
+      schedulingDelay prometheus.Histogram
+  }
+
+  func (p *PrometheusMonitor) JobExecutionTime(job gocron.Job, duration time.Duration) {
+      p.executionTime.Observe(duration.Seconds())
+  }
+
+  func (p *PrometheusMonitor) JobSchedulingDelay(job gocron.Job, scheduled, actual time.Time) {
+      if delay := actual.Sub(scheduled); delay > 0 {
+          p.schedulingDelay.Observe(delay.Seconds())
+      }
+  }
+
+  // Initialize scheduler with monitor
+  s, _ := gocron.NewScheduler(gocron.WithSchedulerMonitor(monitor))
+  ```
+
+  **Use Cases:** Prometheus metrics, custom dashboards, alerting systems, performance monitoring
 
 ### Testing
 The gocron library is set up to enable testing.

errors.go

@@ -48,6 +48,7 @@ var (
 	ErrWithDistributedLockerNil      = errors.New("gocron: WithDistributedLocker: locker must not be nil")
 	ErrWithDistributedJobLockerNil   = errors.New("gocron: WithDistributedJobLocker: locker must not be nil")
 	ErrWithIdentifierNil             = errors.New("gocron: WithIdentifier: identifier must not be nil")
+	ErrSchedulerMonitorNil           = errors.New("gocron: WithSchedulerMonitor: monitor must not be nil")
 	ErrWithLimitConcurrentJobsZero   = errors.New("gocron: WithLimitConcurrentJobs: limit must be greater than 0")
 	ErrWithLocationNil               = errors.New("gocron: WithLocation: location must not be nil")
 	ErrWithLoggerNil                 = errors.New("gocron: WithLogger: logger must not be nil")
@@ -59,6 +60,7 @@ var (
 	ErrStartTimeLaterThanEndTime     = errors.New("gocron: WithStartDateTime: start must not be later than end")
 	ErrStopTimeEarlierThanStartTime  = errors.New("gocron: WithStopDateTime: end must not be earlier than start")
 	ErrWithStopTimeoutZeroOrNegative = errors.New("gocron: WithStopTimeout: timeout must be greater than 0")
+	ErrWithSchedulerMonitorNil       = errors.New("gocron: WithSchedulerMonitor: scheduler monitor cannot be nil")
 	ErrWithLimitedRunsZero           = errors.New("gocron: WithLimitedRuns: limit must be greater than 0")
 )
 

executor.go

@@ -56,6 +56,8 @@ type executor struct {
 	monitor Monitor
 	// monitorStatus for reporting metrics
 	monitorStatus MonitorStatus
+	// reference to parent scheduler for lifecycle notifications
+	scheduler *scheduler
 }
 
 type jobIn struct {
@@ -155,6 +157,15 @@ func (e *executor) start() {
 							// all runners are busy, reschedule the work for later
 							// which means we just skip it here and do nothing
 							// TODO when metrics are added, this should increment a rescheduled metric
+							// Notify concurrency limit reached if monitor is configured
+							if e.scheduler != nil && e.scheduler.schedulerMonitor != nil {
+								ctx2, cancel2 := context.WithCancel(executorCtx)
+								job := requestJobCtx(ctx2, jIn.id, e.jobOutRequest)
+								cancel2()
+								if job != nil {
+									e.scheduler.notifyConcurrencyLimitReached("limit", e.scheduler.jobFromInternalJob(*job))
+								}
+							}
 							e.sendOutForRescheduling(&jIn)
 						}
 					} else {
@@ -209,6 +220,10 @@ func (e *executor) start() {
 								// which means we just skip it here and do nothing
 								e.incrementJobCounter(*j, SingletonRescheduled)
 								e.sendOutForRescheduling(&jIn)
+								// Notify concurrency limit reached if monitor is configured
+								if e.scheduler != nil && e.scheduler.schedulerMonitor != nil {
+									e.scheduler.notifyConcurrencyLimitReached("singleton", e.scheduler.jobFromInternalJob(*j))
+								}
 							}
 						} else {
 							// wait mode, fill up that queue (buffered channel, so it's ok)
@@ -416,18 +431,36 @@ func (e *executor) runJob(j internalJob, jIn jobIn) {
 
 	_ = callJobFuncWithParams(j.beforeJobRuns, j.id, j.name)
 
+	//  Notify job started
+	actualStartTime := time.Now()
+	if e.scheduler != nil && e.scheduler.schedulerMonitor != nil {
+		jobObj := e.scheduler.jobFromInternalJob(j)
+		e.scheduler.notifyJobStarted(jobObj)
+		// Notify scheduling delay if job had a scheduled time
+		if len(j.nextScheduled) > 0 {
+			e.scheduler.notifyJobSchedulingDelay(jobObj, j.nextScheduled[0], actualStartTime)
+		}
+	}
+
 	err := callJobFuncWithParams(j.beforeJobRunsSkipIfBeforeFuncErrors, j.id, j.name)
 	if err != nil {
 		e.sendOutForRescheduling(&jIn)
-
 		select {
 		case e.jobsOutCompleted <- j.id:
 		case <-e.ctx.Done():
 		}
-
+		// Notify job failed (before actual run)
+		if e.scheduler != nil && e.scheduler.schedulerMonitor != nil {
+			e.scheduler.notifyJobFailed(e.scheduler.jobFromInternalJob(j), err)
+		}
 		return
 	}
 
+	// Notify job running
+	if e.scheduler != nil && e.scheduler.schedulerMonitor != nil {
+		e.scheduler.notifyJobRunning(e.scheduler.jobFromInternalJob(j))
+	}
+
 	// For intervalFromCompletion, we need to reschedule AFTER the job completes,
 	// not before. For regular jobs, we reschedule before execution (existing behavior).
 	if !j.intervalFromCompletion {
@@ -448,11 +481,25 @@ func (e *executor) runJob(j internalJob, jIn jobIn) {
 	if err != nil {
 		_ = callJobFuncWithParams(j.afterJobRunsWithError, j.id, j.name, err)
 		e.incrementJobCounter(j, Fail)
-		e.recordJobTimingWithStatus(startTime, time.Now(), j, Fail, err)
+		endTime := time.Now()
+		e.recordJobTimingWithStatus(startTime, endTime, j, Fail, err)
+		// Notify job failed
+		if e.scheduler != nil && e.scheduler.schedulerMonitor != nil {
+			jobObj := e.scheduler.jobFromInternalJob(j)
+			e.scheduler.notifyJobFailed(jobObj, err)
+			e.scheduler.notifyJobExecutionTime(jobObj, endTime.Sub(startTime))
+		}
 	} else {
 		_ = callJobFuncWithParams(j.afterJobRuns, j.id, j.name)
 		e.incrementJobCounter(j, Success)
-		e.recordJobTimingWithStatus(startTime, time.Now(), j, Success, nil)
+		endTime := time.Now()
+		e.recordJobTimingWithStatus(startTime, endTime, j, Success, nil)
+		// Notify job completed
+		if e.scheduler != nil && e.scheduler.schedulerMonitor != nil {
+			jobObj := e.scheduler.jobFromInternalJob(j)
+			e.scheduler.notifyJobCompleted(jobObj)
+			e.scheduler.notifyJobExecutionTime(jobObj, endTime.Sub(startTime))
+		}
 	}
 
 	// For intervalFromCompletion, reschedule AFTER the job completes

go.mod

@@ -1,6 +1,6 @@
 module github.com/go-co-op/gocron/v2
 
-go 1.24.0
+go 1.21.4
 
 require (
 	github.com/google/uuid v1.6.0

gocron-monitor-test/debug_restart.go

@@ -0,0 +1,150 @@
+package main
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/go-co-op/gocron/v2"
+)
+
+type DebugMonitor struct {
+	startCount      int
+	stopCount       int
+	jobRegCount     int
+	jobUnregCount   int
+	jobStartCount   int
+	jobRunningCount int
+	jobCompletCount int
+	jobFailCount    int
+}
+
+func (m *DebugMonitor) SchedulerStarted() {
+	m.startCount++
+	fmt.Printf("✓ SchedulerStarted() called (total: %d)\n", m.startCount)
+}
+
+func (m *DebugMonitor) SchedulerShutdown() {
+	m.stopCount++
+	fmt.Printf("✓ SchedulerShutdown() called (total: %d)\n", m.stopCount)
+}
+
+func (m *DebugMonitor) JobRegistered(job *gocron.Job) {
+	m.jobRegCount++
+	fmt.Printf("✓ JobRegistered() called (total: %d) - Job ID: %s\n", m.jobRegCount, (*job).ID())
+}
+
+func (m *DebugMonitor) JobUnregistered(job *gocron.Job) {
+	m.jobUnregCount++
+	fmt.Printf("✓ JobUnregistered() called (total: %d) - Job ID: %s\n", m.jobUnregCount, (*job).ID())
+}
+
+func (m *DebugMonitor) JobStarted(job *gocron.Job) {
+	m.jobStartCount++
+	fmt.Printf("✓ JobStarted() called (total: %d) - Job ID: %s\n", m.jobStartCount, (*job).ID())
+}
+
+func (m *DebugMonitor) JobRunning(job *gocron.Job) {
+	m.jobRunningCount++
+	fmt.Printf("✓ JobRunning() called (total: %d) - Job ID: %s\n", m.jobRunningCount, (*job).ID())
+}
+
+func (m *DebugMonitor) JobCompleted(job *gocron.Job) {
+	m.jobCompletCount++
+	fmt.Printf("✓ JobCompleted() called (total: %d) - Job ID: %s\n", m.jobCompletCount, (*job).ID())
+}
+
+func (m *DebugMonitor) JobFailed(job *gocron.Job, err error) {
+	m.jobFailCount++
+	fmt.Printf("✓ JobFailed() called (total: %d) - Job ID: %s, Error: %v\n", m.jobFailCount, (*job).ID(), err)
+}
+
+func main() {
+	// ONE monitor, multiple scheduler instances
+	monitor := &DebugMonitor{}
+
+	fmt.Println("=== Cycle 1 (Scheduler Instance 1) ===")
+	s1, err := gocron.NewScheduler(
+		gocron.WithSchedulerMonitor(monitor),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	// Create and register some test jobs
+	fmt.Println("Creating jobs...")
+	_, err = s1.NewJob(
+		gocron.DurationJob(1*time.Second),
+		gocron.NewTask(func() { fmt.Println("Job 1 running") }),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	_, err = s1.NewJob(
+		gocron.DurationJob(2*time.Second),
+		gocron.NewTask(func() error {
+			fmt.Println("Job 2 executing and returning error")
+			return fmt.Errorf("simulated job failure")
+		}), // This job will fail with error
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Println("Calling Start()...")
+	s1.Start()
+	time.Sleep(3 * time.Second) // Wait for jobs to execute
+
+	fmt.Println("Calling Shutdown()...")
+	err = s1.Shutdown()
+	if err != nil {
+		fmt.Printf("Shutdown error: %v\n", err)
+	}
+
+	fmt.Println("\n=== Cycle 2 (Job Updates) ===")
+	s2, err := gocron.NewScheduler(
+		gocron.WithSchedulerMonitor(monitor),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Println("Creating and updating jobs...")
+	job3, err := s2.NewJob(
+		gocron.DurationJob(1*time.Second),
+		gocron.NewTask(func() { fmt.Println("Job 3 running") }),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	// Update the job
+	_, err = s2.Update(
+		job3.ID(),
+		gocron.DurationJob(2*time.Second),
+		gocron.NewTask(func() { fmt.Println("Job 3 updated") }),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Println("Calling Start()...")
+	s2.Start()
+	time.Sleep(3 * time.Second)
+
+	fmt.Println("Calling Shutdown()...")
+	err = s2.Shutdown()
+	if err != nil {
+		fmt.Printf("Shutdown error: %v\n", err)
+	}
+
+	fmt.Println("\n=== Summary ===")
+	fmt.Printf("Total Scheduler Starts: %d\n", monitor.startCount)
+	fmt.Printf("Total Scheduler Stops: %d\n", monitor.stopCount)
+	fmt.Printf("Total Jobs Registered: %d\n", monitor.jobRegCount)
+	fmt.Printf("Total Jobs Unregistered: %d\n", monitor.jobUnregCount)
+	fmt.Printf("Total Jobs Started: %d\n", monitor.jobStartCount)
+	fmt.Printf("Total Jobs Running: %d\n", monitor.jobRunningCount)
+	fmt.Printf("Total Jobs Completed: %d\n", monitor.jobCompletCount)
+	fmt.Printf("Total Jobs Failed: %d\n", monitor.jobFailCount)
+}

gocron-monitor-test/go.mod

@@ -0,0 +1,13 @@
+module test
+
+go 1.21.4
+
+require github.com/go-co-op/gocron/v2 v2.17.0
+
+require (
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/jonboulle/clockwork v0.5.0 // indirect
+	github.com/robfig/cron/v3 v3.0.1 // indirect
+)
+
+replace github.com/go-co-op/gocron/v2 => ../

gocron-monitor-test/go.sum

@@ -0,0 +1,16 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/jonboulle/clockwork v0.5.0 h1:Hyh9A8u51kptdkR+cqRpT1EebBwTn1oK9YfGYbdFz6I=
+github.com/jonboulle/clockwork v0.5.0/go.mod h1:3mZlmanh0g2NDKO5TWZVJAfofYk64M7XN3SzBPjZF60=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=