chore: 并入 iotlcm-app 历史到 app/（原提交 f40bda9）
This commit is contained in:
MiFaZhan
@@ -0,0 +1,144 @@
|
||||
package allocatedispatch
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/gtime"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-app/internal/consts"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-app/internal/dao"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-app/internal/library/batchmetadata"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-app/internal/library/rabbitmq"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-app/internal/model/entity"
|
||||
)
|
||||
|
||||
type message struct {
|
||||
BatchNo string `json:"batchNo"`
|
||||
BatchType int `json:"batchType"`
|
||||
}
|
||||
|
||||
func Start(ctx context.Context) context.CancelFunc {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
if !g.Cfg().MustGet(ctx, "workerBatchAsync.enabled", false).Bool() || !g.Cfg().MustGet(ctx, "workerBatchAsync.allocateRecovery.enabled", true).Bool() {
|
||||
return cancel
|
||||
}
|
||||
interval := g.Cfg().MustGet(ctx, "workerBatchAsync.allocateRecovery.intervalSeconds", 60).Int()
|
||||
if interval < 30 {
|
||||
interval = 30
|
||||
}
|
||||
go func() {
|
||||
ticker := time.NewTicker(time.Duration(interval) * time.Second)
|
||||
defer ticker.Stop()
|
||||
var cursor int64
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
next, err := recoverPage(ctx, cursor, publish)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "划拨任务补投失败: %+v", err)
|
||||
} else {
|
||||
cursor = next
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return cancel
|
||||
}
|
||||
|
||||
func due(batch *entity.BatchMain, now time.Time, processingDelay time.Duration) bool {
|
||||
if batch == nil || batch.BatchType != consts.BatchTypeAllocate || batch.SourceChannel != batchmetadata.MiniProgram {
|
||||
return false
|
||||
}
|
||||
delay := time.Minute
|
||||
switch batch.BatchAsyncStatus {
|
||||
case consts.BatchAsyncStatusPending:
|
||||
case consts.BatchAsyncStatusProcessing:
|
||||
delay = processingDelay
|
||||
default:
|
||||
return false
|
||||
}
|
||||
timestamp := batch.UpdatedAt
|
||||
if batch.BatchAsyncStatus == consts.BatchAsyncStatusProcessing && batch.StartTime != nil && !batch.StartTime.IsZero() {
|
||||
timestamp = batch.StartTime
|
||||
}
|
||||
if timestamp == nil || timestamp.IsZero() {
|
||||
timestamp = batch.CreatedAt
|
||||
}
|
||||
if timestamp == nil || timestamp.IsZero() {
|
||||
return false
|
||||
}
|
||||
return !timestamp.Time.Add(delay).After(now)
|
||||
}
|
||||
|
||||
func recoverPage(ctx context.Context, cursor int64, send func(context.Context, string, string) error) (int64, error) {
|
||||
if ctx.Err() != nil {
|
||||
return cursor, ctx.Err()
|
||||
}
|
||||
columns := dao.BatchMain.Columns()
|
||||
var parents []*entity.BatchMain
|
||||
err := dao.BatchMain.Ctx(ctx).Where(columns.Id+" > ?", cursor).
|
||||
Where(columns.BatchType, consts.BatchTypeAllocate).Where(columns.SourceChannel, batchmetadata.MiniProgram).
|
||||
Where("COALESCE(source_batch_no, '') = ''").
|
||||
WhereIn(columns.BatchAsyncStatus, []int{consts.BatchAsyncStatusPending, consts.BatchAsyncStatusProcessing}).
|
||||
OrderAsc(columns.Id).Limit(50).Scan(&parents)
|
||||
if err != nil {
|
||||
return cursor, err
|
||||
}
|
||||
if len(parents) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
minutes := g.Cfg().MustGet(ctx, "workerBatchAsync.allocateRecovery.processingRetryMinutes", 30).Int()
|
||||
if minutes < 30 {
|
||||
minutes = 30
|
||||
}
|
||||
delay := time.Duration(minutes) * time.Minute
|
||||
queue := strings.TrimSpace(g.Cfg().MustGet(ctx, "workerBatchAsync.queueName", "iotlcm.batch.execute").String())
|
||||
if queue == "" {
|
||||
queue = "iotlcm.batch.execute"
|
||||
}
|
||||
for _, parent := range parents {
|
||||
if ctx.Err() != nil {
|
||||
return cursor, ctx.Err()
|
||||
}
|
||||
cursor = parent.Id
|
||||
if parent.BatchInfo == nil || parent.BatchInfo.Get("jobRole").String() != "allocate_parent" || !due(parent, time.Now(), delay) {
|
||||
continue
|
||||
}
|
||||
var children []*entity.BatchMain
|
||||
err = dao.BatchMain.Ctx(ctx).Where(columns.SourceBatchNo, parent.BatchNo).
|
||||
Where(columns.BatchType, consts.BatchTypeAllocate).Where(columns.TenantId, parent.TenantId).
|
||||
Where(columns.CreatedBy, parent.CreatedBy).Where(columns.SourceChannel, batchmetadata.MiniProgram).Scan(&children)
|
||||
if err != nil {
|
||||
return cursor, err
|
||||
}
|
||||
for _, child := range children {
|
||||
if child.BatchInfo == nil || child.BatchInfo.Get("jobRole").String() != "allocate_chunk" || !due(child, time.Now(), delay) {
|
||||
continue
|
||||
}
|
||||
if err := send(ctx, queue, child.BatchNo); err != nil {
|
||||
g.Log().Errorf(ctx, "划拨子任务补投失败 batchNo:%s err:%+v", child.BatchNo, err)
|
||||
}
|
||||
}
|
||||
if err := send(ctx, queue, parent.BatchNo); err != nil {
|
||||
g.Log().Errorf(ctx, "划拨父任务补投失败 batchNo:%s err:%+v", parent.BatchNo, err)
|
||||
}
|
||||
}
|
||||
if len(parents) < 50 {
|
||||
return 0, nil
|
||||
}
|
||||
return cursor, nil
|
||||
}
|
||||
|
||||
func publish(ctx context.Context, queue, batchNo string) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
err := rabbitmq.Publish(ctx, queue, message{BatchNo: batchNo, BatchType: consts.BatchTypeAllocate})
|
||||
if err == nil {
|
||||
g.Log().Infof(ctx, "划拨持久任务补投 batchNo:%s time:%s", batchNo, gtime.Now())
|
||||
}
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package allocatedispatch
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
_ "github.com/gogf/gf/contrib/drivers/sqlite/v2"
|
||||
"github.com/gogf/gf/v2/database/gdb"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/gtime"
|
||||
"testing"
|
||||
"time"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-app/internal/model/entity"
|
||||
)
|
||||
|
||||
func TestDue(t *testing.T) {
|
||||
now := time.Now()
|
||||
for _, test := range []struct {
|
||||
status int
|
||||
age time.Duration
|
||||
due bool
|
||||
}{{1, 59 * time.Second, false}, {1, 2 * time.Minute, true}, {2, 29 * time.Minute, false}, {2, 31 * time.Minute, true}, {3, time.Hour, false}, {4, time.Hour, false}, {5, time.Hour, false}} {
|
||||
batch := &entity.BatchMain{BatchType: 3, SourceChannel: "mini_program", BatchAsyncStatus: test.status, UpdatedAt: gtime.New(now.Add(-test.age))}
|
||||
if due(batch, now, 30*time.Minute) != test.due {
|
||||
t.Fatal(test)
|
||||
}
|
||||
}
|
||||
if due(&entity.BatchMain{BatchType: 3, SourceChannel: "console", BatchAsyncStatus: 1, UpdatedAt: gtime.New(now.Add(-time.Hour))}, now, 30*time.Minute) {
|
||||
t.Fatal("foreign channel replayed")
|
||||
}
|
||||
if due(&entity.BatchMain{BatchType: 3, SourceChannel: "mini_program", BatchAsyncStatus: 1}, now, 30*time.Minute) {
|
||||
t.Fatal("missing time replayed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecoveryPersistsFailureAndIsolatesParents(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
if err := gdb.SetConfig(gdb.Config{"default": gdb.ConfigGroup{{Type: "sqlite", Link: "sqlite::@file(:memory:)", MaxOpenConnCount: 1}}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = g.DB().Close(ctx) })
|
||||
for _, sql := range []string{
|
||||
`CREATE TABLE batch_main (id INTEGER PRIMARY KEY,batch_no TEXT,batch_type INTEGER,source_batch_no TEXT,source_channel TEXT,batch_async_status INTEGER,tenant_id TEXT,created_by INTEGER,batch_info TEXT,created_at TEXT,updated_at TEXT,start_time TEXT)`,
|
||||
`INSERT INTO batch_main VALUES (1,'PARENT',3,'','mini_program',1,'A',7,'{"jobRole":"allocate_parent"}','2020-01-01','2020-01-01',NULL), (2,'CHILD',3,'PARENT','mini_program',1,'A',7,'{"jobRole":"allocate_chunk"}','2020-01-01','2020-01-01',NULL), (3,'FOREIGN',3,'PARENT','mini_program',1,'B',7,'{"jobRole":"allocate_chunk"}','2020-01-01','2020-01-01',NULL), (4,'DONE',3,'','mini_program',3,'A',7,'{"jobRole":"allocate_parent"}','2020-01-01','2020-01-01',NULL), (5,'OTHER',101,'','mini_program',1,'A',7,'{}','2020-01-01','2020-01-01',NULL), (6,'CONSOLE',3,'','console',1,'A',7,'{"jobRole":"allocate_parent"}','2020-01-01','2020-01-01',NULL)`,
|
||||
} {
|
||||
if _, err := g.DB().Exec(ctx, sql); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, fail := range []bool{true, false} {
|
||||
var calls []string
|
||||
cursor, err := recoverPage(ctx, 0, func(ctx context.Context, queue, batchNo string) error {
|
||||
calls = append(calls, batchNo)
|
||||
if queue != "iotlcm.batch.execute" {
|
||||
t.Fatal(queue)
|
||||
}
|
||||
if fail {
|
||||
return errors.New("queue offline")
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil || cursor != 0 || len(calls) != 2 || calls[0] != "CHILD" || calls[1] != "PARENT" {
|
||||
t.Fatal(calls, cursor, err)
|
||||
}
|
||||
}
|
||||
if _, err := g.DB().Exec(ctx, `UPDATE batch_main SET batch_async_status=3 WHERE batch_no='PARENT'`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err := recoverPage(ctx, 0, func(context.Context, string, string) error { t.Fatal("terminal parent replayed"); return nil })
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cancelled, cancel := context.WithCancel(ctx)
|
||||
cancel()
|
||||
if _, err := recoverPage(cancelled, 0, func(context.Context, string, string) error { t.Fatal("cancelled replay"); return nil }); err == nil {
|
||||
t.Fatal("cancel ignored")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user