Merge pull request #70 from libi/develop

millstone: new breaking change v0.5.1, from develop to master.

Author: libi

.github/workflows/test.yml

@@ -2,9 +2,9 @@ name: Testing
 
 on:
   push:
-    branches: [ "master" ]
+    branches: [ "master", "develop" ]
   pull_request:
-    branches: [ "master" ]
+    branches: [ "master", "develop" ]
 
 jobs:
 

dcron.go

@@ -17,11 +17,12 @@ import (
 const (
 	defaultReplicas = 50
 	defaultDuration = 3 * time.Second
-)
 
-const (
 	dcronRunning = 1
 	dcronStopped = 0
+
+	dcronStateSteady  = "dcronStateSteady"
+	dcronStateUpgrade = "dcronStateUpgrade"
 )
 
 type RecoverFuncType func(d *Dcron)
@@ -45,6 +46,9 @@ type Dcron struct {
 	crOptions []cron.Option
 
 	RecoverFunc RecoverFuncType
+
+	recentJobs IRecentJobPacker
+	state      atomic.Value
 }
 
 // NewDcron create a Dcron
@@ -66,7 +70,6 @@ func NewDcronWithOption(serverName string, driver driver.DriverV2, dcronOpts ...
 
 	dcron.cr = cron.New(dcron.crOptions...)
 	dcron.nodePool = NewNodePool(serverName, driver, dcron.nodeUpdateDuration, dcron.hashReplicas, dcron.logger)
-
 	return dcron
 }
 
@@ -137,8 +140,24 @@ func (d *Dcron) Remove(jobName string) {
 	}
 }
 
-func (d *Dcron) allowThisNodeRun(jobName string) bool {
-	return d.nodePool.CheckJobAvailable(jobName)
+func (d *Dcron) allowThisNodeRun(jobName string) (ok bool) {
+	ok, err := d.nodePool.CheckJobAvailable(jobName)
+	if err != nil {
+		d.logger.Errorf("allow this node run error, err=%v", err)
+		ok = false
+		d.state.Store(dcronStateUpgrade)
+	} else {
+		d.state.Store(dcronStateSteady)
+		if d.recentJobs != nil {
+			go d.reRunRecentJobs(d.recentJobs.PopAllJobs())
+		}
+	}
+	if d.recentJobs != nil {
+		if d.state.Load().(string) == dcronStateUpgrade {
+			d.recentJobs.AddJob(jobName, time.Now())
+		}
+	}
+	return
 }
 
 // Start job
@@ -147,14 +166,13 @@ func (d *Dcron) Start() {
 	if d.RecoverFunc != nil {
 		d.RecoverFunc(d)
 	}
-
 	if atomic.CompareAndSwapInt32(&d.running, dcronStopped, dcronRunning) {
 		if err := d.startNodePool(); err != nil {
 			atomic.StoreInt32(&d.running, dcronStopped)
 			return
 		}
 		d.cr.Start()
-		d.logger.Infof("dcron started , nodeID is %s", d.nodePool.GetNodeID())
+		d.logger.Infof("dcron started, nodeID is %s", d.nodePool.GetNodeID())
 	} else {
 		d.logger.Infof("dcron have started")
 	}
@@ -171,8 +189,7 @@ func (d *Dcron) Run() {
 			atomic.StoreInt32(&d.running, dcronStopped)
 			return
 		}
-
-		d.logger.Infof("dcron running nodeID is %s", d.nodePool.GetNodeID())
+		d.logger.Infof("dcron running, nodeID is %s", d.nodePool.GetNodeID())
 		d.cr.Run()
 	} else {
 		d.logger.Infof("dcron already running")
@@ -199,3 +216,18 @@ func (d *Dcron) Stop() {
 		}
 	}
 }
+
+func (d *Dcron) reRunRecentJobs(jobNames []string) {
+	d.logger.Infof("reRunRecentJobs: length=%d", len(jobNames))
+	for _, jobName := range jobNames {
+		if job, ok := d.jobs[jobName]; ok {
+			if ok, _ := d.nodePool.CheckJobAvailable(jobName); ok {
+				job.Execute()
+			}
+		}
+	}
+}
+
+func (d *Dcron) NodeID() string {
+	return d.nodePool.GetNodeID()
+}

dcron_test.go

@@ -197,6 +197,50 @@ func Test_SecondJobWithStopAndSwapNode(t *testing.T) {
 	wg.Wait()
 }
 
+func Test_WithClusterStableNodes(t *testing.T) {
+	wg := &sync.WaitGroup{}
+	wg.Add(5)
+
+	runningTime := 60 * time.Second
+	startFunc := func(id string, timeWindow time.Duration, t *testing.T) {
+		redisCli := redis.NewClient(&redis.Options{
+			Addr: DefaultRedisAddr,
+		})
+		drv := driver.NewRedisDriver(redisCli)
+		dcr := dcron.NewDcronWithOption(t.Name(), drv,
+			dcron.CronOptionSeconds(),
+			dcron.WithLogger(&dlog.StdLogger{
+				Log: log.New(os.Stdout, "["+id+"]", log.LstdFlags),
+			}),
+			dcron.WithClusterStable(timeWindow),
+			dcron.WithNodeUpdateDuration(timeWindow),
+		)
+		var err error
+		err = dcr.AddFunc("job1", "*/3 * * * * *", func() {
+			t.Log(time.Now())
+		})
+		require.Nil(t, err)
+		err = dcr.AddFunc("job2", "*/8 * * * * *", func() {
+			t.Logf("job2: %v", time.Now())
+		})
+		require.Nil(t, err)
+		err = dcr.AddFunc("job3", "* * * * * *", func() {
+			t.Log("job3:", time.Now())
+		})
+		require.Nil(t, err)
+		dcr.Start()
+		<-time.After(runningTime)
+		dcr.Stop()
+		wg.Done()
+	}
+
+	go startFunc("1", time.Second*6, t)
+	go startFunc("2", time.Second*6, t)
+	go startFunc("3", time.Second*6, t)
+	go startFunc("4", time.Second*6, t)
+	go startFunc("5", time.Second*6, t)
+}
+
 func Test_SecondJobLog_Issue68(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	wg.Add(5)

driver/etcddriver.go

@@ -134,14 +134,15 @@ func (e *EtcdDriver) keepAlive(ctx context.Context, nodeID string) (<-chan *clie
 func (e *EtcdDriver) revoke(ctx context.Context) {
 	_, err := e.cli.Lease.Revoke(ctx, e.leaseID)
 	if err != nil {
-		e.logger.Printf("lease revoke error: %v", err)
+		e.logger.Infof("lease revoke error: %v", err)
 	}
 }
 
 func (e *EtcdDriver) heartBeat(ctx context.Context) {
 label:
 	leaseCh, err := e.keepAlive(ctx, e.nodeID)
 	if err != nil {
+		e.logger.Errorf("keep alive error, %v", err)
 		return
 	}
 	for {

inodepool.go

@@ -1,11 +1,20 @@
 package dcron
 
-import "context"
+import (
+	"context"
+	"errors"
+	"time"
+)
+
+var (
+	ErrNodePoolIsUpgrading = errors.New("nodePool is upgrading")
+)
 
 type INodePool interface {
 	Start(ctx context.Context) error
-	CheckJobAvailable(jobName string) bool
+	CheckJobAvailable(jobName string) (bool, error)
 	Stop(ctx context.Context) error
 
 	GetNodeID() string
+	GetLastNodesUpdateTime() time.Time
 }

inodepool_test.go

@@ -81,9 +81,7 @@ func (ts *TestINodePoolSuite) declareRedisZSetDrivers(clients *[]*redis.Client,
 func (ts *TestINodePoolSuite) runCheckJobAvailable(numberOfNodes int, ServiceName string, nodePools *[]dcron.INodePool, updateDuration time.Duration) {
 	for i := 0; i < numberOfNodes; i++ {
 		err := (*nodePools)[i].Start(context.Background())
-		if err != nil {
-			ts.T().Fail()
-		}
+		ts.Require().Nil(err)
 	}
 	<-time.After(updateDuration * 2)
 	ring := consistenthash.New(ts.defaultHashReplicas, nil)
@@ -93,13 +91,14 @@ func (ts *TestINodePoolSuite) runCheckJobAvailable(numberOfNodes int, ServiceNam
 
 	for i := 0; i < 10000; i++ {
 		for j := 0; j < numberOfNodes; j++ {
+			ok, err := (*nodePools)[j].CheckJobAvailable(strconv.Itoa(i))
+			ts.Require().Nil(err)
 			ts.Require().Equal(
-				(*nodePools)[j].CheckJobAvailable(strconv.Itoa(i)),
+				ok,
 				(ring.Get(strconv.Itoa(i)) == (*nodePools)[j].GetNodeID()),
 			)
 		}
 	}
-
 }
 
 func (ts *TestINodePoolSuite) TestMultiNodesRedis() {
@@ -143,7 +142,7 @@ func (ts *TestINodePoolSuite) TestMultiNodesRedisZSet() {
 	var nodePools []dcron.INodePool
 
 	numberOfNodes := 5
-	ServiceName := "TestMultiNodesEtcd"
+	ServiceName := "TestMultiNodesZSet"
 	updateDuration := 2 * time.Second
 
 	ts.setUpRedis()

irecentjobpacker.go

@@ -0,0 +1,19 @@
+package dcron
+
+import "time"
+
+// IRecentJobPacker
+// this is an interface which be used to
+// pack the jobs running in the cluster state
+// is `unstable`.
+// like some nodes broken or new nodes were add.
+type IRecentJobPacker interface {
+	// goroutine safety.
+	// Add a job to packer
+	// will save recent jobs (like 2 * heartbeat duration)
+	AddJob(jobName string, t time.Time) error
+
+	// goroutine safety.
+	// Pop out all jobs in packer
+	PopAllJobs() (jobNames []string)
+}

irecentjobpacker_test.go

@@ -0,0 +1,65 @@
+package dcron_test
+
+import (
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/libi/dcron"
+	"github.com/stretchr/testify/suite"
+)
+
+type IRecentJobPackerTestSuite struct {
+	suite.Suite
+}
+
+func (s *IRecentJobPackerTestSuite) checkPopJobs(l, r int, wg *sync.WaitGroup, tAfter time.Duration, recentPacker dcron.IRecentJobPacker) {
+	<-time.After(tAfter)
+	jobNames := recentPacker.PopAllJobs()
+	expectJobNames := make([]string, 0)
+	for i := l; i < r; i++ {
+		expectJobNames = append(expectJobNames, "i_"+strconv.Itoa(i))
+	}
+	s.Equal(expectJobNames, jobNames)
+	wg.Done()
+}
+
+func (s *IRecentJobPackerTestSuite) TestNormal() {
+	// the timeWindow is 6s and we will add a job each 0.5s
+	// and check the jobs list after 8s.
+	// expect job name is i_4 ~ i_16
+	timeLimit := time.Second * 6
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	recentPacker := dcron.NewRecentJobPacker(timeLimit)
+
+	go s.checkPopJobs(4, 16, wg, 8*time.Second, recentPacker)
+
+	for i := 0; i < 16; i++ {
+		recentPacker.AddJob("i_"+strconv.Itoa(i), time.Now())
+		<-time.After(500 * time.Millisecond)
+	}
+	wg.Wait()
+}
+
+func (s *IRecentJobPackerTestSuite) Test_CallMultiTimes() {
+	timeLimit := time.Second * 6
+	wg := &sync.WaitGroup{}
+	wg.Add(2)
+	recentPacker := dcron.NewRecentJobPacker(timeLimit)
+
+	go s.checkPopJobs(4, 16, wg, 8*time.Second, recentPacker)
+	go s.checkPopJobs(20, 32, wg, 16*time.Second, recentPacker)
+
+	for i := 0; i < 32; i++ {
+		recentPacker.AddJob("i_"+strconv.Itoa(i), time.Now())
+		<-time.After(500 * time.Millisecond)
+	}
+	wg.Wait()
+}
+
+func TestIRecentJobPackerTestSuite(t *testing.T) {
+	s := new(IRecentJobPackerTestSuite)
+	suite.Run(t, s)
+}

job_warpper.go

@@ -31,11 +31,15 @@ type JobWarpper struct {
 func (job JobWarpper) Run() {
 	//如果该任务分配给了这个节点 则允许执行
 	if job.Dcron.allowThisNodeRun(job.Name) {
-		if job.Func != nil {
-			job.Func()
-		}
-		if job.Job != nil {
-			job.Job.Run()
-		}
+		job.Execute()
+	}
+}
+
+func (job JobWarpper) Execute() {
+	if job.Func != nil {
+		job.Func()
+	}
+	if job.Job != nil {
+		job.Job.Run()
 	}
 }