Repository: go-co-op/gocron
Pull request #561 (2023-09-05 07:47:02)

feat: add election leader
## summary

Sometimes, we need gocron to support active-standby mode. Only the active instance can run jobs, and other instances are just backup instances. Only when there is a problem with the master instance can other backup instances run jobs.

we can use  libs to implement election interface, e.g. `etcd election`、`k8s leaderelection`、`consul`、 `zk` and `redis lock`.  😁

It is recommended that gocron can give users more choices, allowing users to choose distributed locks and distributed election solutions according to their needs. 

### election vs lock

Of course, the distributed lock solution is also very good, each gocron instances try to acquire lock, each gocron instaces can run jobs with probability.

The advantage of election is that all tasks are executed on one instance, and more business optimization operations can be done, such as simpler code, local cache, connection pool, etc. 😁

![image](https://github.com/go-co-op/gocron/assets/3785409/4e1485ad-a8ee-4054-a8ed-192ad40f8fe9)

Aftger the server-1 crashes and exits abnormally，the server-2 elected as leader. the server-2 can schedule all jobs.

![image](https://github.com/go-co-op/gocron/assets/3785409/1dbf8a50-af31-481b-8c3c-1053140d2004)


rfyiamcool: ### example

```
package main

import (
	"context"
	"errors"
	"fmt"
	"math/rand"
	"os"
	"os/signal"
	"sync"
	"time"

	"github.com/go-co-op/gocron"
	"go.etcd.io/etcd/clientv3"
	"go.etcd.io/etcd/clientv3/concurrency"
)

var (
	serverName = fmt.Sprintf("gocron-%03d", rand.Intn(100))
	endpoints  = []string{"http://127.0.0.1:2379"}
)

type Election struct {
	mu       sync.RWMutex
	ctx      context.Context
	cancel   context.CancelFunc
	isLeader bool
}

func newElection(ctx context.Context) *Election {
	cctx, cancel := context.WithCancel(ctx)
	return &Election{
		ctx:    cctx,
		cancel: cancel,
	}
}

func (e *Election) Close() error {
	e.cancel()
	return nil
}

func (e *Election) IsLeader(_ context.Context) error {
	e.mu.RLock()
	defer e.mu.RUnlock()

	if e.isLeader {
		return nil
	}

	return errors.New("non leader")
}

func (e *Election) start() error {
	cli, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
	if err != nil {
		return err
	}
	defer cli.Close()

	session, err := concurrency.NewSession(cli, concurrency.WithTTL(10))
	if err != nil {
		return err
	}
	defer session.Close()

	electionHandler := concurrency.NewElection(session, "/my-election")
	fmt.Println(123)

	go func() {
		for e.ctx.Err() == nil {
			if err := electionHandler.Campaign(context.Background(), serverName); err != nil {
				fmt.Println("campaign", err)
			}

			time.Sleep(200 * time.Millisecond)
		}
	}()

	cctx, cancel := context.WithCancel(context.TODO())
	defer cancel()

	for e.ctx.Err() == nil {
		resp := <-electionHandler.Observe(cctx)
		if len(resp.Kvs) == 0 {
			continue
		}

		e.mu.Lock()
		if string(resp.Kvs[0].Value) == serverName {
			e.isLeader = true
		} else {
			e.isLeader = false
			fmt.Println("current leader is", string(resp.Kvs[0].Value))
		}
		e.mu.Unlock()

		time.Sleep(1e9)
	}

	return nil
}

func main() {
	el := newElection(context.Background())
	go el.start()

	fmt.Println("srart", serverName)

	s := gocron.NewScheduler(time.UTC)
	s.WithDistributedElection(el)

	s.Every("1s").Do(func() {
		fmt.Println("call 1s", serverName)
	})

	s.Every("2s").Do(func() {
		fmt.Println("call 2s", serverName)
	})

	s.Every("3s").Do(func() {
		fmt.Println("call 3s", serverName)
	})

	s.StartAsync()

	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt)
	<-c

	fmt.Println("exit")
}
```
rfyiamcool: > do you have an implementation in mind that you'd like to add? I can create a repo for you in go-co-op for gocron-elector-etcd? or whichever tech

good，i can. 😀

Patch:
diff --git a/executor.go b/executor.go
index f1a30ac..9e86664 100644
--- a/executor.go
+++ b/executor.go
@@ -54,7 +54,8 @@ type executor struct {
 	limitModeRunningJobs    *atomic.Int64    // tracks the count of running jobs to check against the max
 	stopped                 *atomic.Bool     // allow workers to drain the buffered limitModeQueue
 
-	distributedLocker Locker // support running jobs across multiple instances
+	distributedLocker  Locker  // support running jobs across multiple instances
+	distributedElector Elector // support running jobs across multiple instances
 }
 
 func newExecutor() executor {
@@ -160,6 +161,14 @@ func (e *executor) runJob(f jobFunction) {
 		if lockKey == "" {
 			lockKey = f.funcName
 		}
+		if e.distributedElector != nil {
+			err := e.distributedElector.IsLeader(e.ctx)
+			if err != nil {
+				return
+			}
+			runJob(f)
+			return
+		}
 		if e.distributedLocker != nil {
 			l, err := e.distributedLocker.Lock(f.ctx, lockKey)
 			if err != nil || l == nil {
@@ -190,6 +199,8 @@ func (e *executor) runJob(f jobFunction) {
 				}
 				_ = l.Unlock(f.ctx)
 			}()
+			runJob(f)
+			return
 		}
 		runJob(f)
 	case singletonMode:
diff --git a/locker.go b/locker.go
index dc713f9..193b20c 100644
--- a/locker.go
+++ b/locker.go
@@ -21,3 +21,10 @@ type Locker interface {
 type Lock interface {
 	Unlock(ctx context.Context) error
 }
+
+// Elector determines the leader from instances asking to be the leader. Only
+// the leader runs jobs. If the leader goes down, a new leader will be elected.
+type Elector interface {
+	// IsLeader should return an error if the job should not be scheduled and nil if the job should be scheduled.
+	IsLeader(ctx context.Context) error
+}
diff --git a/scheduler.go b/scheduler.go
index a8ddc29..6c60db5 100644
--- a/scheduler.go
+++ b/scheduler.go
@@ -1493,6 +1493,18 @@ func (s *Scheduler) WithDistributedLocker(l Locker) {
 	s.executor.distributedLocker = l
 }
 
+// WithDistributedElector prevents the same job from being run more than once
+// when multiple schedulers are trying to schedule the same job, by allowing only
+// the leader to run jobs. Non-leaders wait until the leader instance goes down
+// and then a new leader is elected.
+//
+// Compared with the distributed lock, the election is the same as leader/follower framework.
+// All jobs are only scheduled and execute on the leader scheduler instance. Only when the leader scheduler goes down
+// and one of the scheduler instances is successfully elected, then the new leader scheduler instance can schedule jobs.
+func (s *Scheduler) WithDistributedElector(e Elector) {
+	s.executor.distributedElector = e
+}
+
 // RegisterEventListeners accepts EventListeners and registers them for all jobs
 // in the scheduler at the time this function is called.
 // The event listeners are then called at the times described by each listener.


Test patch:
diff --git a/scheduler_test.go b/scheduler_test.go
index 22d63f6..91b5be5 100644
--- a/scheduler_test.go
+++ b/scheduler_test.go
@@ -2,6 +2,7 @@ package gocron
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sort"
 	"sync"
@@ -2851,3 +2852,68 @@ func TestDataRace(t *testing.T) {
 	time.Sleep(1 * time.Second)
 	sut.Stop()
 }
+
+var _ Elector = (*elector)(nil)
+
+type elector struct {
+	isLeader bool
+}
+
+func (e *elector) IsLeader(_ context.Context) error {
+	if e.isLeader {
+		return nil
+	}
+	return errors.New("is not leader")
+}
+
+func (e *elector) setLeader() {
+	e.isLeader = true
+}
+
+func TestScheduler_EnableDistributedElector(t *testing.T) {
+	runTestWithDistributedElector(t, 0)
+}
+
+func TestScheduler_EnableDistributedElectorWithMaxConcurrent(t *testing.T) {
+	runTestWithDistributedElector(t, 1)
+}
+
+func runTestWithDistributedElector(t *testing.T, maxConcurrentJobs int) {
+	resultChan := make(chan int, 20)
+	f := func(schedulerInstance int) {
+		resultChan <- schedulerInstance
+	}
+
+	leaderIndex := 0
+	schedulers := make([]*Scheduler, 0)
+	for i := 0; i < 3; i++ {
+		el := &elector{}
+		if i == leaderIndex {
+			el.setLeader()
+		}
+
+		s := NewScheduler(time.UTC)
+		s.WithDistributedElector(el)
+		if maxConcurrentJobs > 0 {
+			s.SetMaxConcurrentJobs(maxConcurrentJobs, WaitMode)
+		}
+		_, err := s.Every("50ms").Do(f, i)
+		require.NoError(t, err)
+		schedulers = append(schedulers, s)
+	}
+	for i := range schedulers {
+		schedulers[i].StartAsync()
+	}
+	time.Sleep(530 * time.Millisecond)
+	for i := range schedulers {
+		schedulers[i].Stop()
+	}
+	close(resultChan)
+
+	// 10 <- len <- 12
+	assert.Greater(t, len(resultChan), 10)
+	assert.Less(t, len(resultChan), 12)
+	for r := range resultChan {
+		assert.Equal(t, leaderIndex, r)
+	}
+}