Skip to content

Commit f85cbd3

Browse files
dxyinmeAH-darkwithrjplibi
authored
Develop to Master (#92)
* chore: use `redis.UniversalClient` instead of `*redis.Client` (#88) * make dcron running locally, update cron test, add GetJobs / GetJob function for dcron. (#89) * 增加code coverage, 修复GetJob bug，增加devcontainer方便开发者 (#91) * add example app to readme * add NodeID function into dcron * Deps: updated github.com/go-redis/redis/v8 to github.com/redis/go-redis/v9 (#73) * add cron lib * fix warning * logger removal : phares1 * update * update * update * add robfig/cron to dcron (#75) * add NodeID function into dcron * add cron lib * fix warning * logger removal : phares1 * update * update * update * update * update * update * update * update test workflow * revert 1.22 * update etcd driver * update * fix get nodes * update * update for fix TAT * Revert "update etcd driver" This reverts commit a21ebf7. * Revert "deps: updated go-redis to v9 (#79)" This reverts commit 0b85b24. * update * update * refact etcd * Revert "refact etcd" This reverts commit 049bed1. * Revert "update" This reverts commit 9c71fd6. * update * refactor etcddriver * fix error * update * Revert "update" This reverts commit 6cfcfe6. * Revert "fix error" This reverts commit 99b2d82. * Revert "refactor etcddriver" This reverts commit a576ac3. * update * update * remove comments, and fix * add comments * Add comments, add split the E2E test cases and other test cases. (#80) * add example app to readme * add NodeID function into dcron * add cron lib * fix warning * logger removal : phares1 * update * update * update * update * update * update * update * update test workflow * revert 1.22 * update etcd driver * update * fix get nodes * update * update for fix TAT * Revert "update etcd driver" This reverts commit a21ebf7. * Revert "deps: updated go-redis to v9 (#79)" This reverts commit 0b85b24. * update * update * refact etcd * Revert "refact etcd" This reverts commit 049bed1. * Revert "update" This reverts commit 9c71fd6. * update * refactor etcddriver * fix error * update * Revert "update" This reverts commit 6cfcfe6. * Revert "fix error" This reverts commit 99b2d82. * Revert "refactor etcddriver" This reverts commit a576ac3. * update * update * remove comments, and fix * add comments * merge e2e test and normal test * move e2etest to origin path * add timeout to avoid pipeline timeout * add getjobs related function * update * remove redis config in test workflow * update * update chain test in windows * update action version for node 16 -> node 20 * add test for dcron locally * update readme * update * fix code bug * Update devcontainer and etcd driver test --------- Co-authored-by: Ava Ackerman <withrjp@gmail.com> Co-authored-by: libi <7769922+libi@users.noreply.github.com> --------- Co-authored-by: AHdark <ahdark@outlook.com> Co-authored-by: Ava Ackerman <withrjp@gmail.com> Co-authored-by: libi <7769922+libi@users.noreply.github.com>
1 parent 68982eb commit f85cbd3

File tree

16 files changed

+479
-95
lines changed

16 files changed

+479
-95
lines changed

‎.devcontainer/devcontainer.json‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
{
2+
"image": "mcr.microsoft.com/devcontainers/go:1-1.22-bookworm",
3+
"customizations": {
4+
"vscode": {
5+
"extensions": [
6+
"golang.go",
7+
"eamodio.gitlens"
8+
]
9+
}
10+
}
11+
}

‎.github/workflows/test.yml‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,20 +20,18 @@ jobs:
2020
runs-on: ubuntu-latest
2121

2222
steps:
23-
- uses: actions/checkout@v3
23+
- uses: actions/checkout@v4
2424

25-
- name: Redis Server in GitHub Actions
26-
uses: supercharge/redis-github-action@1.7.0
2725
- name: Set up Go
28-
uses: actions/setup-go@v3
26+
uses: actions/setup-go@v5
2927
with:
3028
go-version: ${{ matrix.go_version }}
3129

3230
- name: Test
3331
run: go test -v -timeout 30m -coverprofile=coverage.txt -covermode=atomic ./...
3432

3533
- name: Upload coverage reports to Codecov
36-
uses: codecov/codecov-action@v3
34+
uses: codecov/codecov-action@v4
3735
env:
3836
CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
3937

‎cron/README.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
11
# cron
22

3-
fork from [robfig/cron](github.com/robfig/cron)
3+
fork from [robfig/cron](github.com/robfig/cron)
4+
5+
maintain by Dcron opensource team.

‎cron/chain_test.go‎

Lines changed: 26 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -116,37 +116,45 @@ func TestChainDelayIfStillRunning(t *testing.T) {
116116
t.Run("second run immediate if first done", func(t *testing.T) {
117117
var j countJob
118118
wrappedJob := NewChain(DelayIfStillRunning(DiscardLogger)).Then(&j)
119+
wg := &sync.WaitGroup{}
120+
wg.Add(1)
119121
go func() {
120122
go wrappedJob.Run()
121-
time.Sleep(time.Millisecond)
123+
<-time.After(time.Millisecond)
122124
go wrappedJob.Run()
125+
wg.Done()
123126
}()
124-
time.Sleep(3 * time.Millisecond) // Give both jobs 3ms to complete.
127+
wg.Wait()
128+
<-time.After(3 * time.Millisecond)
125129
if c := j.Done(); c != 2 {
126130
t.Errorf("expected job run twice, immediately, got %d", c)
127131
}
128132
})
129133

130134
t.Run("second run delayed if first not done", func(t *testing.T) {
131135
var j countJob
132-
j.delay = 10 * time.Millisecond
136+
j.delay = 100 * time.Millisecond
133137
wrappedJob := NewChain(DelayIfStillRunning(DiscardLogger)).Then(&j)
138+
wg := &sync.WaitGroup{}
139+
wg.Add(1)
134140
go func() {
135141
go wrappedJob.Run()
136-
time.Sleep(time.Millisecond)
142+
<-time.After(10 * time.Millisecond)
137143
go wrappedJob.Run()
144+
wg.Done()
138145
}()
139146

140-
// After 5ms, the first job is still in progress, and the second job was
147+
wg.Wait()
148+
// After 50ms, the first job is still in progress, and the second job was
141149
// run but should be waiting for it to finish.
142-
time.Sleep(5 * time.Millisecond)
150+
<-time.After(50 * time.Millisecond)
143151
started, done := j.Started(), j.Done()
144152
if started != 1 || done != 0 {
145153
t.Error("expected first job started, but not finished, got", started, done)
146154
}
147155

148156
// Verify that the second job completes.
149-
time.Sleep(25 * time.Millisecond)
157+
<-time.After(220 * time.Millisecond)
150158
started, done = j.Started(), j.Done()
151159
if started != 2 || done != 2 {
152160
t.Error("expected both jobs done, got", started, done)
@@ -169,37 +177,41 @@ func TestChainSkipIfStillRunning(t *testing.T) {
169177
t.Run("second run immediate if first done", func(t *testing.T) {
170178
var j countJob
171179
wrappedJob := NewChain(SkipIfStillRunning(DiscardLogger)).Then(&j)
180+
wg := &sync.WaitGroup{}
181+
wg.Add(1)
172182
go func() {
173183
go wrappedJob.Run()
174-
time.Sleep(time.Millisecond)
184+
<-time.After(time.Millisecond)
175185
go wrappedJob.Run()
186+
wg.Done()
176187
}()
177-
time.Sleep(3 * time.Millisecond) // Give both jobs 3ms to complete.
188+
wg.Wait()
189+
<-time.After(3 * time.Millisecond) // Give both jobs 3ms to complete.
178190
if c := j.Done(); c != 2 {
179191
t.Errorf("expected job run twice, immediately, got %d", c)
180192
}
181193
})
182194

183195
t.Run("second run skipped if first not done", func(t *testing.T) {
184196
var j countJob
185-
j.delay = 10 * time.Millisecond
197+
j.delay = 100 * time.Millisecond
186198
wrappedJob := NewChain(SkipIfStillRunning(DiscardLogger)).Then(&j)
187199
go func() {
188200
go wrappedJob.Run()
189-
time.Sleep(time.Millisecond)
201+
<-time.After(10 * time.Millisecond)
190202
go wrappedJob.Run()
191203
}()
192204

193-
// After 5ms, the first job is still in progress, and the second job was
205+
// After 50ms, the first job is still in progress, and the second job was
194206
// aleady skipped.
195-
time.Sleep(5 * time.Millisecond)
207+
<-time.After(50 * time.Millisecond)
196208
started, done := j.Started(), j.Done()
197209
if started != 1 || done != 0 {
198210
t.Error("expected first job started, but not finished, got", started, done)
199211
}
200212

201213
// Verify that the first job completes and second does not run.
202-
time.Sleep(25 * time.Millisecond)
214+
<-time.After(220 * time.Millisecond)
203215
started, done = j.Started(), j.Done()
204216
if started != 1 || done != 1 {
205217
t.Error("expected second job skipped, got", started, done)

‎dcron.go‎

Lines changed: 95 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -25,12 +25,18 @@ const (
2525
dcronStateUpgrade = "dcronStateUpgrade"
2626
)
2727

28+
var (
29+
ErrJobExist = errors.New("jobName already exist")
30+
ErrJobNotExist = errors.New("jobName not exist")
31+
ErrJobWrongNode = errors.New("job is not running in this node")
32+
)
33+
2834
type RecoverFuncType func(d *Dcron)
2935

3036
// Dcron is main struct
3137
type Dcron struct {
3238
jobs map[string]*JobWarpper
33-
jobsRWMut sync.Mutex
39+
jobsRWMut sync.RWMutex
3440

3541
ServerName string
3642
nodePool INodePool
@@ -48,6 +54,8 @@ type Dcron struct {
4854

4955
recentJobs IRecentJobPacker
5056
state atomic.Value
57+
58+
runningLocally bool
5159
}
5260

5361
// NewDcron create a Dcron
@@ -68,7 +76,9 @@ func NewDcronWithOption(serverName string, driver driver.DriverV2, dcronOpts ...
6876
}
6977

7078
dcron.cr = cron.New(dcron.crOptions...)
71-
dcron.nodePool = NewNodePool(serverName, driver, dcron.nodeUpdateDuration, dcron.hashReplicas, dcron.logger)
79+
if !dcron.runningLocally {
80+
dcron.nodePool = NewNodePool(serverName, driver, dcron.nodeUpdateDuration, dcron.hashReplicas, dcron.logger)
81+
}
7282
return dcron
7383
}
7484

@@ -110,9 +120,9 @@ func (d *Dcron) addJob(jobName, cronStr string, job Job) (err error) {
110120
d.jobsRWMut.Lock()
111121
defer d.jobsRWMut.Unlock()
112122
if _, ok := d.jobs[jobName]; ok {
113-
return errors.New("jobName already exist")
123+
return ErrJobExist
114124
}
115-
innerJob := JobWarpper{
125+
innerJob := &JobWarpper{
116126
Name: jobName,
117127
CronStr: cronStr,
118128
Job: job,
@@ -123,11 +133,11 @@ func (d *Dcron) addJob(jobName, cronStr string, job Job) (err error) {
123133
return err
124134
}
125135
innerJob.ID = entryID
126-
d.jobs[jobName] = &innerJob
136+
d.jobs[jobName] = innerJob
127137
return nil
128138
}
129139

130-
// Remove Job
140+
// Remove Job by jobName
131141
func (d *Dcron) Remove(jobName string) {
132142
d.jobsRWMut.Lock()
133143
defer d.jobsRWMut.Unlock()
@@ -138,7 +148,71 @@ func (d *Dcron) Remove(jobName string) {
138148
}
139149
}
140150

151+
// Get job by jobName
152+
// if this jobName not exist, will return error.
153+
//
154+
// if `thisNodeOnly` is true
155+
// if this job is not available in this node, will return error.
156+
// otherwise return the struct of JobWarpper whose name is jobName.
157+
func (d *Dcron) GetJob(jobName string, thisNodeOnly bool) (*JobWarpper, error) {
158+
d.jobsRWMut.RLock()
159+
defer d.jobsRWMut.RUnlock()
160+
161+
job, ok := d.jobs[jobName]
162+
if !ok {
163+
d.logger.Warnf("job: %s, not exist", jobName)
164+
return nil, ErrJobNotExist
165+
}
166+
if !thisNodeOnly {
167+
return job, nil
168+
}
169+
isRunningHere, err := d.nodePool.CheckJobAvailable(jobName)
170+
if err != nil {
171+
return nil, err
172+
}
173+
if !isRunningHere {
174+
return nil, ErrJobWrongNode
175+
}
176+
return job, nil
177+
}
178+
179+
// Get job list.
180+
//
181+
// if `thisNodeOnly` is true
182+
// return all jobs available in this node.
183+
// otherwise return all jobs added to dcron.
184+
//
185+
// we never return nil. If there is no job.
186+
// this func will return an empty slice.
187+
func (d *Dcron) GetJobs(thisNodeOnly bool) []*JobWarpper {
188+
d.jobsRWMut.RLock()
189+
defer d.jobsRWMut.RUnlock()
190+
191+
ret := make([]*JobWarpper, 0)
192+
for _, v := range d.jobs {
193+
var (
194+
isRunningHere bool
195+
ok bool = true
196+
err error
197+
)
198+
if thisNodeOnly {
199+
isRunningHere, err = d.nodePool.CheckJobAvailable(v.Name)
200+
if err != nil {
201+
continue
202+
}
203+
ok = isRunningHere
204+
}
205+
if ok {
206+
ret = append(ret, v)
207+
}
208+
}
209+
return ret
210+
}
211+
141212
func (d *Dcron) allowThisNodeRun(jobName string) (ok bool) {
213+
if d.runningLocally {
214+
return true
215+
}
142216
ok, err := d.nodePool.CheckJobAvailable(jobName)
143217
if err != nil {
144218
d.logger.Errorf("allow this node run error, err=%v", err)
@@ -165,12 +239,14 @@ func (d *Dcron) Start() {
165239
d.RecoverFunc(d)
166240
}
167241
if atomic.CompareAndSwapInt32(&d.running, dcronStopped, dcronRunning) {
168-
if err := d.startNodePool(); err != nil {
169-
atomic.StoreInt32(&d.running, dcronStopped)
170-
return
242+
if !d.runningLocally {
243+
if err := d.startNodePool(); err != nil {
244+
atomic.StoreInt32(&d.running, dcronStopped)
245+
return
246+
}
247+
d.logger.Infof("dcron started, nodeID is %s", d.nodePool.GetNodeID())
171248
}
172249
d.cr.Start()
173-
d.logger.Infof("dcron started, nodeID is %s", d.nodePool.GetNodeID())
174250
} else {
175251
d.logger.Infof("dcron have started")
176252
}
@@ -183,11 +259,13 @@ func (d *Dcron) Run() {
183259
d.RecoverFunc(d)
184260
}
185261
if atomic.CompareAndSwapInt32(&d.running, dcronStopped, dcronRunning) {
186-
if err := d.startNodePool(); err != nil {
187-
atomic.StoreInt32(&d.running, dcronStopped)
188-
return
262+
if !d.runningLocally {
263+
if err := d.startNodePool(); err != nil {
264+
atomic.StoreInt32(&d.running, dcronStopped)
265+
return
266+
}
267+
d.logger.Infof("dcron running, nodeID is %s", d.nodePool.GetNodeID())
189268
}
190-
d.logger.Infof("dcron running, nodeID is %s", d.nodePool.GetNodeID())
191269
d.cr.Run()
192270
} else {
193271
d.logger.Infof("dcron already running")
@@ -205,7 +283,9 @@ func (d *Dcron) startNodePool() error {
205283
// Stop job
206284
func (d *Dcron) Stop() {
207285
tick := time.NewTicker(time.Millisecond)
208-
d.nodePool.Stop(context.Background())
286+
if !d.runningLocally {
287+
d.nodePool.Stop(context.Background())
288+
}
209289
for range tick.C {
210290
if atomic.CompareAndSwapInt32(&d.running, dcronRunning, dcronStopped) {
211291
d.cr.Stop()

‎dcron_locally_test.go‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
package dcron_test
2+
3+
import (
4+
"testing"
5+
"time"
6+
7+
"github.com/libi/dcron"
8+
"github.com/libi/dcron/cron"
9+
"github.com/stretchr/testify/require"
10+
"github.com/stretchr/testify/suite"
11+
)
12+
13+
type DcronLocallyTestSuite struct {
14+
suite.Suite
15+
}
16+
17+
func (s *DcronLocallyTestSuite) TestNormal() {
18+
dcr := dcron.NewDcronWithOption(
19+
"not a necessary servername",
20+
nil,
21+
dcron.RunningLocally(),
22+
dcron.CronOptionSeconds(),
23+
dcron.CronOptionChain(
24+
cron.Recover(
25+
cron.DefaultLogger,
26+
)))
27+
28+
s.Assert().NotNil(dcr)
29+
runningTime := 60 * time.Second
30+
t := s.T()
31+
var err error
32+
err = dcr.AddFunc("job1", "*/5 * * * * *", func() {
33+
t.Log(time.Now())
34+
})
35+
require.Nil(t, err)
36+
err = dcr.AddFunc("job2", "*/8 * * * * *", func() {
37+
panic("test panic")
38+
})
39+
require.Nil(t, err)
40+
err = dcr.AddFunc("job3", "*/2 * * * * *", func() {
41+
t.Log("job3:", time.Now())
42+
})
43+
require.Nil(t, err)
44+
dcr.Start()
45+
<-time.After(runningTime)
46+
dcr.Stop()
47+
}
48+
49+
func TestDcronLocallyTestSuite(t *testing.T) {
50+
suite.Run(t, &DcronLocallyTestSuite{})
51+
}

0 commit comments

Comments
 (0)