Repository: matrixorigin/matrixcube
Pull request #731 (2022-03-25 08:00:08)

Update long running job related logic;
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [x] Code Refactoring

**Which issue(s) this PR fixes:**

Fixes #

**What this PR does / why we need it:**

1. Respect the original license;
2. Update error handling for long running jobs;
3. Update method signature of `type interface JobStorage` for precise;
4. Fix some subtle nuance logic;

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/components/prophet/id/id.go b/components/prophet/id/id.go
index 0f37acf..96c5c7b 100644
--- a/components/prophet/id/id.go
+++ b/components/prophet/id/id.go
@@ -1,4 +1,5 @@
-// Copyright 2020 MatrixOrigin.
+// Copyright 2020 PingCAP, Inc.
+// Modifications copyright (C) 2021 MatrixOrigin.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/components/prophet/id/id_mem.go b/components/prophet/id/id_mem.go
index 0ac008a..7b6f5ec 100644
--- a/components/prophet/id/id_mem.go
+++ b/components/prophet/id/id_mem.go
@@ -1,4 +1,5 @@
-// Copyright 2020 MatrixOrigin.
+// Copyright 2020 PingCAP, Inc.
+// Modifications copyright (C) 2021 MatrixOrigin.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/components/prophet/join/join.go b/components/prophet/join/join.go
index cb6d49a..002e29e 100644
--- a/components/prophet/join/join.go
+++ b/components/prophet/join/join.go
@@ -1,4 +1,5 @@
-// Copyright 2020 MatrixOrigin.
+// Copyright 2020 PingCAP, Inc.
+// Modifications copyright (C) 2021 MatrixOrigin.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/components/prophet/member/member.go b/components/prophet/member/member.go
index a9c7470..98a4fdc 100644
--- a/components/prophet/member/member.go
+++ b/components/prophet/member/member.go
@@ -1,4 +1,5 @@
-// Copyright 2020 MatrixOrigin.
+// Copyright 2020 PingCAP, Inc.
+// Modifications copyright (C) 2021 MatrixOrigin.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/components/prophet/prophet_job.go b/components/prophet/prophet_job.go
index 248eb9e..771b9ea 100644
--- a/components/prophet/prophet_job.go
+++ b/components/prophet/prophet_job.go
@@ -18,6 +18,7 @@ import (
 	"fmt"
 
 	"github.com/matrixorigin/matrixcube/components/prophet/cluster"
+	"github.com/matrixorigin/matrixcube/components/prophet/util"
 	"github.com/matrixorigin/matrixcube/pb/metapb"
 	"github.com/matrixorigin/matrixcube/pb/rpcpb"
 	"go.uber.org/zap"
@@ -51,7 +52,7 @@ func (p *defaultProphet) startJobs() {
 		}
 
 		p.logger.Info("load jobs", zap.Int("count", len(p.jobMu.jobs)))
-		for _, job := range p.jobMu.jobs {
+		for jobType, job := range p.jobMu.jobs {
 			select {
 			case <-ctx.Done():
 				return
@@ -65,6 +66,8 @@ func (p *defaultProphet) startJobs() {
 						zap.String("type", job.Type.String()),
 						zap.Error(err))
 				}
+				// Get rid of completed job from the cache
+				delete(p.jobMu.jobs, jobType)
 				continue
 			}
 
@@ -112,7 +115,9 @@ func (p *defaultProphet) handleCreateJob(rc *cluster.RaftCluster, req *rpcpb.Pro
 	job := req.CreateJob.Job
 	processor := p.cfg.Prophet.GetJobProcessor(job.Type)
 	if processor == nil {
-		return fmt.Errorf("missing job processor for type %d", job.Type)
+		return util.WrappedError(
+			util.ErrJobProcessorNotFound, fmt.Sprintf("job type = %d", job.Type),
+		)
 	}
 
 	if _, ok := p.jobMu.jobs[job.Type]; ok {
@@ -136,7 +141,9 @@ func (p *defaultProphet) handleRemoveJob(rc *cluster.RaftCluster, req *rpcpb.Pro
 	job := req.RemoveJob.Job
 	processor := p.cfg.Prophet.GetJobProcessor(job.Type)
 	if processor == nil {
-		return fmt.Errorf("missing job processor for type %d, %+v", job.Type, job)
+		return util.WrappedError(
+			util.ErrJobProcessorNotFound, fmt.Sprintf("job type = %d, %+v", job.Type, job),
+		)
 	}
 
 	if _, ok := p.jobMu.jobs[job.Type]; !ok {
@@ -156,11 +163,15 @@ func (p *defaultProphet) handleExecuteJob(rc *cluster.RaftCluster, req *rpcpb.Pr
 	job := req.ExecuteJob.Job
 	processor := p.cfg.Prophet.GetJobProcessor(job.Type)
 	if processor == nil {
-		return fmt.Errorf("missing job processor for type %d", job.Type)
+		return util.WrappedError(
+			util.ErrJobProcessorNotFound, fmt.Sprintf("job type = %d", job.Type),
+		)
 	}
 
 	if _, ok := p.jobMu.jobs[job.Type]; !ok {
-		return fmt.Errorf("missing job for type %d, the job maybe not created or started", job.Type)
+		return util.WrappedError(
+			util.ErrJobNotFound, fmt.Sprintf("not created or not started, job type = %d", job.Type),
+		)
 	}
 
 	data, err := processor.Execute(req.ExecuteJob.Data, p.storage, rc.GetCacheCluster())
diff --git a/components/prophet/storage/storage.go b/components/prophet/storage/storage.go
index d7336c0..b79ea89 100644
--- a/components/prophet/storage/storage.go
+++ b/components/prophet/storage/storage.go
@@ -39,11 +39,11 @@ type JobStorage interface {
 	LoadJobs(limit int64, do func(metapb.Job)) error
 
 	// PutJobData put job data
-	PutJobData(metapb.Job, []byte) error
+	PutJobData(metapb.JobType, []byte) error
 	// GetJobData  returns job data
-	GetJobData(metapb.Job) ([]byte, error)
+	GetJobData(metapb.JobType) ([]byte, error)
 	// RemoveJobData removes job data
-	RemoveJobData(metapb.Job) error
+	RemoveJobData(metapb.JobType) error
 }
 
 // RuleStorage rule storage
@@ -496,12 +496,12 @@ func (s *storage) LoadJobs(limit int64, fn func(metapb.Job)) error {
 	})
 }
 
-func (s *storage) PutJobData(job metapb.Job, data []byte) error {
-	return s.kv.Save(s.jobDataKey(job.Type), string(data))
+func (s *storage) PutJobData(jobType metapb.JobType, data []byte) error {
+	return s.kv.Save(s.jobDataKey(jobType), string(data))
 }
 
-func (s *storage) GetJobData(job metapb.Job) ([]byte, error) {
-	v, err := s.kv.Load(s.jobDataKey(job.Type))
+func (s *storage) GetJobData(jobType metapb.JobType) ([]byte, error) {
+	v, err := s.kv.Load(s.jobDataKey(jobType))
 	if err != nil {
 		return nil, err
 	}
@@ -509,8 +509,8 @@ func (s *storage) GetJobData(job metapb.Job) ([]byte, error) {
 	return []byte(v), nil
 }
 
-func (s *storage) RemoveJobData(job metapb.Job) error {
-	return s.kv.Remove(s.jobDataKey(job.Type))
+func (s *storage) RemoveJobData(jobType metapb.JobType) error {
+	return s.kv.Remove(s.jobDataKey(jobType))
 }
 
 func (s *storage) PutCustomData(key []byte, data []byte) error {
diff --git a/components/prophet/util/errors.go b/components/prophet/util/errors.go
index 5f7c481..7dd8602 100644
--- a/components/prophet/util/errors.go
+++ b/components/prophet/util/errors.go
@@ -15,6 +15,8 @@ package util
 
 import (
 	"errors"
+	"fmt"
+	"strings"
 )
 
 var (
@@ -34,9 +36,25 @@ var (
 	ErrSchedulerExisted = errors.New("scheduler is existed")
 	// ErrSchedulerNotFound error with scheduler is not found
 	ErrSchedulerNotFound = errors.New("scheduler is not found")
+
+	// errors related with long running job
+	// ErrJobProcessorNotFound should keep compatibility with PR 1915 of matrixone
+	ErrJobProcessorNotFound = errors.New("missing job processor")
+	ErrJobProcessorStopped  = errors.New("job processor stopped")
+	ErrJobInvalidCommand    = errors.New("invalid job command")
+	ErrJobNotFound          = errors.New("job not found")
 )
 
 // IsNotLeaderError is not leader error
 func IsNotLeaderError(err string) bool {
 	return err == ErrNotLeader.Error()
 }
+
+// IsJobProcessorNotFoundErr check error via its string content
+func IsJobProcessorNotFoundErr(err string) bool {
+	return strings.Contains(err, ErrJobProcessorNotFound.Error())
+}
+
+func WrappedError(err error, msg string) error {
+	return fmt.Errorf("%w: %s", err, msg)
+}
diff --git a/raftstore/store_shards_pool.go b/raftstore/store_shards_pool.go
index 80ec945..00f1efc 100644
--- a/raftstore/store_shards_pool.go
+++ b/raftstore/store_shards_pool.go
@@ -14,6 +14,7 @@ import (
 	"github.com/matrixorigin/matrixcube/components/prophet"
 	pconfig "github.com/matrixorigin/matrixcube/components/prophet/config"
 	"github.com/matrixorigin/matrixcube/components/prophet/storage"
+	"github.com/matrixorigin/matrixcube/components/prophet/util"
 	"github.com/matrixorigin/matrixcube/config"
 	"github.com/matrixorigin/matrixcube/pb/metapb"
 	"github.com/matrixorigin/matrixcube/util/buf"
@@ -138,7 +139,7 @@ func (dsp *dynamicShardsPool) Start(job metapb.Job, store storage.JobStorage, aw
 	}
 
 	// load or init the job data
-	value, err := store.GetJobData(job)
+	value, err := store.GetJobData(job.Type)
 	if err != nil {
 		return
 	}
@@ -184,14 +185,16 @@ func (dsp *dynamicShardsPool) Remove(job metapb.Job, store storage.JobStorage, a
 
 func (dsp *dynamicShardsPool) Execute(data []byte, store storage.JobStorage, aware pconfig.ShardsAware) ([]byte, error) {
 	if len(data) <= 0 {
-		return nil, errors.New("error execute data")
+		return nil, util.WrappedError(
+			util.ErrJobInvalidCommand, fmt.Sprintf("data empty"),
+		)
 	}
 
 	dsp.mu.Lock()
 	defer dsp.mu.Unlock()
 
 	if !dsp.isStartedLocked() {
-		return nil, fmt.Errorf("job not started")
+		return nil, util.ErrJobProcessorStopped
 	}
 
 	cmd := &metapb.ShardsPoolCmd{}
@@ -200,13 +203,22 @@ func (dsp *dynamicShardsPool) Execute(data []byte, store storage.JobStorage, awa
 	case metapb.ShardsPoolCmdType_AllocShard:
 		return dsp.doAllocLocked(cmd.Alloc, store, aware)
 	default:
-		return nil, fmt.Errorf("invalid execute cmd %d", cmd.Type)
+		return nil, util.WrappedError(
+			util.ErrJobInvalidCommand,
+			fmt.Sprintf("not supported (type=%d)", cmd.Type),
+		)
 	}
 }
 
 func (dsp *dynamicShardsPool) doAllocLocked(cmd *metapb.ShardsPoolAllocCmd, store storage.JobStorage, aware pconfig.ShardsAware) ([]byte, error) {
 	group := cmd.Group
-	p := dsp.mu.pools.Pools[group]
+	p, ok := dsp.mu.pools.Pools[group]
+	if !ok {
+		return nil, util.WrappedError(
+			util.ErrJobInvalidCommand,
+			fmt.Sprintf("missing shard pool for group: %d", group),
+		)
+	}
 
 	// check whether the purpose has been allocated before
 	if len(p.AllocatedShards) > 0 {
@@ -420,7 +432,7 @@ func (dsp *dynamicShardsPool) gcAllocating(store storage.JobStorage, aware pconf
 }
 
 func (dsp *dynamicShardsPool) saveLocked(store storage.JobStorage) error {
-	err := store.PutJobData(dsp.job, protoc.MustMarshal(&dsp.mu.pools))
+	err := store.PutJobData(dsp.job.Type, protoc.MustMarshal(&dsp.mu.pools))
 	if err != nil {
 		dsp.logger.Error("fail to save shard pool metadata, retry later",
 			zap.Error(err))


Test patch:
diff --git a/components/prophet/id/id_test.go b/components/prophet/id/id_test.go
index 4b47f98..8e41c46 100644
--- a/components/prophet/id/id_test.go
+++ b/components/prophet/id/id_test.go
@@ -1,4 +1,5 @@
-// Copyright 2020 MatrixOrigin.
+// Copyright 2020 PingCAP, Inc.
+// Modifications copyright (C) 2021 MatrixOrigin.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/components/prophet/util/errors_test.go b/components/prophet/util/errors_test.go
new file mode 100644
index 0000000..db44bbc
--- /dev/null
+++ b/components/prophet/util/errors_test.go
@@ -0,0 +1,17 @@
+package util
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsJobProcessorNotFoundErr(t *testing.T) {
+	msg := fmt.Sprintf("type = %d", 1)
+	wrappedErr := WrappedError(ErrJobProcessorNotFound, msg)
+	assert.True(
+		t,
+		IsJobProcessorNotFoundErr(wrappedErr.Error()),
+	)
+}
diff --git a/raftstore/integration_create_shard_test.go b/raftstore/integration_create_shard_test.go
index 5d04e5c..0739483 100644
--- a/raftstore/integration_create_shard_test.go
+++ b/raftstore/integration_create_shard_test.go
@@ -160,7 +160,7 @@ func TestAddShardWithShardPool(t *testing.T) {
 	c.WaitShardByCountPerNode(5, testWaitTimeout)
 	c.WaitLeadersByCount(5, testWaitTimeout)
 
-	v, err := c.GetProphet().GetStorage().GetJobData(metapb.Job{Type: metapb.JobType_CreateShardPool})
+	v, err := c.GetProphet().GetStorage().GetJobData(metapb.JobType_CreateShardPool)
 	assert.NoError(t, err)
 	sp := &metapb.ShardsPool{}
 	protoc.MustUnmarshal(sp, v)
@@ -183,7 +183,7 @@ func TestAddShardWithShardPool(t *testing.T) {
 		}
 	})
 
-	v, err = c.GetProphet().GetStorage().GetJobData(metapb.Job{Type: metapb.JobType_CreateShardPool})
+	v, err = c.GetProphet().GetStorage().GetJobData(metapb.JobType_CreateShardPool)
 	assert.NoError(t, err)
 	sp = &metapb.ShardsPool{}
 	protoc.MustUnmarshal(sp, v)
diff --git a/raftstore/store_shards_pool_test.go b/raftstore/store_shards_pool_test.go
index d1756aa..26ab8be 100644
--- a/raftstore/store_shards_pool_test.go
+++ b/raftstore/store_shards_pool_test.go
@@ -135,7 +135,7 @@ func TestGCAllocating(t *testing.T) {
 
 	ss := storage.NewTestStorage()
 	p.gcAllocating(ss, nil)
-	v, err := ss.GetJobData(p.job)
+	v, err := ss.GetJobData(p.job.Type)
 	assert.NoError(t, err)
 	assert.Empty(t, v)
 
@@ -143,7 +143,7 @@ func TestGCAllocating(t *testing.T) {
 	p.mu.pools = metapb.ShardsPool{Pools: make(map[uint64]*metapb.ShardPool)}
 	p.mu.pools.Pools[1] = &metapb.ShardPool{Capacity: 1, Seq: 1, AllocatedOffset: 0}
 	p.gcAllocating(ss, nil)
-	v, err = ss.GetJobData(p.job)
+	v, err = ss.GetJobData(p.job.Type)
 	assert.NoError(t, err)
 	assert.Empty(t, v)
 
@@ -163,7 +163,7 @@ func TestGCAllocating(t *testing.T) {
 		},
 	}}
 	p.gcAllocating(ss, aware)
-	v, err = ss.GetJobData(p.job)
+	v, err = ss.GetJobData(p.job.Type)
 	assert.NoError(t, err)
 	assert.NotEmpty(t, v)
 	assert.Equal(t, 1, len(p.mu.pools.Pools[1].AllocatedShards))
@@ -219,7 +219,7 @@ func TestMaybeCreate(t *testing.T) {
 	assert.Equal(t, c, batchCreateCount)
 	assert.Equal(t, 1, len(p.mu.createC))
 	assert.Equal(t, uint64(batchCreateCount), p.mu.pools.Pools[0].Seq)
-	v, err := ss.GetJobData(p.job)
+	v, err := ss.GetJobData(p.job.Type)
 	assert.NoError(t, err)
 	assert.Equal(t, protoc.MustMarshal(&p.mu.pools), v)
 }
@@ -290,7 +290,7 @@ func TestDoAllocLocked(t *testing.T) {
 	assert.NotEmpty(t, v)
 	assert.Equal(t, 1, len(p.mu.createC))
 	assert.Equal(t, uint64(1), p.mu.pools.Pools[0].AllocatedOffset)
-	v, err = ss.GetJobData(p.job)
+	v, err = ss.GetJobData(p.job.Type)
 	assert.NoError(t, err)
 	assert.Equal(t, protoc.MustMarshal(&p.mu.pools), v)
 }
@@ -332,7 +332,7 @@ func TestShardsPoolStartAndStop(t *testing.T) {
 	job := metapb.Job{Type: metapb.JobType_CreateShardPool}
 	pools := metapb.ShardsPool{Pools: make(map[uint64]*metapb.ShardPool)}
 	pools.Pools[0] = &metapb.ShardPool{Capacity: 4}
-	s.PutJobData(job, protoc.MustMarshal(&pools))
+	s.PutJobData(job.Type, protoc.MustMarshal(&pools))
 	p.Start(job, s, nil)
 
 	p.mu.RLock()