From db743245357ed59df4cbb53a5feab69552c6a581 Mon Sep 17 00:00:00 2001 From: mindon Date: Mon, 21 Nov 2022 12:00:37 +0800 Subject: [PATCH 1/3] Add TaskStateProber to get task state changed events --- client.go | 5 +++ internal/base/base.go | 8 ++++ internal/rdb/rdb.go | 100 +++++++++++++++++++++++++++++++++++++++--- server.go | 57 ++++++++++++++++++++++++ 4 files changed, 165 insertions(+), 5 deletions(-) diff --git a/client.go b/client.go index 7948036f..9ab1db13 100644 --- a/client.go +++ b/client.go @@ -426,3 +426,8 @@ func (c *Client) addToGroup(ctx context.Context, msg *base.TaskMessage, group st } return c.broker.AddToGroup(ctx, msg, group) } + +// StateChanged watchs state updates, with more customized detail +func (c *Client) SetTaskProber(prober base.TaskProber) { + c.broker.SetTaskProber(prober) +} diff --git a/internal/base/base.go b/internal/base/base.go index ec342f88..64be48a0 100644 --- a/internal/base/base.go +++ b/internal/base/base.go @@ -53,6 +53,11 @@ const ( TaskStateAggregating // describes a state where task is waiting in a group to be aggregated ) +type TaskProber interface { + Changed(map[string]interface{}) + Result(state TaskState, raw *TaskInfo) (string, interface{}) +} + func (s TaskState) String() string { switch s { case TaskStateActive: @@ -752,4 +757,7 @@ type Broker interface { PublishCancelation(id string) error WriteResult(qname, id string, data []byte) (n int, err error) + + // StateChanged watch state updates, with more customized detail + SetTaskProber(prober TaskProber) } diff --git a/internal/rdb/rdb.go b/internal/rdb/rdb.go index 7b25f154..c07ccdf6 100644 --- a/internal/rdb/rdb.go +++ b/internal/rdb/rdb.go @@ -7,6 +7,7 @@ package rdb import ( "context" + "encoding/json" "fmt" "math" "time" @@ -131,6 +132,7 @@ func (r *RDB) Enqueue(ctx context.Context, msg *base.TaskMessage) error { if n == 0 { return errors.E(op, errors.AlreadyExists, errors.ErrTaskIdConflict) } + r.state(ctx, msg, base.TaskStatePending) return nil } @@ -198,6 +200,7 @@ func (r *RDB) EnqueueUnique(ctx context.Context, msg *base.TaskMessage, ttl time if n == 0 { return errors.E(op, errors.AlreadyExists, errors.ErrTaskIdConflict) } + r.state(ctx, msg, base.TaskStatePending) return nil } @@ -464,9 +467,13 @@ func (r *RDB) MarkAsComplete(ctx context.Context, msg *base.TaskMessage) error { // Note: We cannot pass empty unique key when running this script in redis-cluster. if len(msg.UniqueKey) > 0 { keys = append(keys, msg.UniqueKey) - return r.runScript(ctx, op, markAsCompleteUniqueCmd, keys, argv...) + err := r.runScript(ctx, op, markAsCompleteUniqueCmd, keys, argv...) + r.state(ctx, msg, base.TaskStateCompleted, err) + return err } - return r.runScript(ctx, op, markAsCompleteCmd, keys, argv...) + err = r.runScript(ctx, op, markAsCompleteCmd, keys, argv...) + r.state(ctx, msg, base.TaskStateCompleted, err) + return err } // KEYS[1] -> asynq:{}:active @@ -495,7 +502,9 @@ func (r *RDB) Requeue(ctx context.Context, msg *base.TaskMessage) error { base.PendingKey(msg.Queue), base.TaskKey(msg.Queue, msg.ID), } - return r.runScript(ctx, op, requeueCmd, keys, msg.ID) + err := r.runScript(ctx, op, requeueCmd, keys, msg.ID) + r.state(ctx, msg, base.TaskStatePending, err) + return err } // KEYS[1] -> asynq:{}:t: @@ -550,6 +559,7 @@ func (r *RDB) AddToGroup(ctx context.Context, msg *base.TaskMessage, groupKey st if n == 0 { return errors.E(op, errors.AlreadyExists, errors.ErrTaskIdConflict) } + r.state(ctx, msg, base.TaskStateAggregating) return nil } @@ -617,6 +627,7 @@ func (r *RDB) AddToGroupUnique(ctx context.Context, msg *base.TaskMessage, group if n == 0 { return errors.E(op, errors.AlreadyExists, errors.ErrTaskIdConflict) } + r.state(ctx, msg, base.TaskStateAggregating) return nil } @@ -667,6 +678,7 @@ func (r *RDB) Schedule(ctx context.Context, msg *base.TaskMessage, processAt tim if n == 0 { return errors.E(op, errors.AlreadyExists, errors.ErrTaskIdConflict) } + r.state(ctx, msg, base.TaskStateScheduled) return nil } @@ -731,6 +743,7 @@ func (r *RDB) ScheduleUnique(ctx context.Context, msg *base.TaskMessage, process if n == 0 { return errors.E(op, errors.AlreadyExists, errors.ErrTaskIdConflict) } + r.state(ctx, msg, base.TaskStateScheduled) return nil } @@ -813,7 +826,9 @@ func (r *RDB) Retry(ctx context.Context, msg *base.TaskMessage, processAt time.T isFailure, int64(math.MaxInt64), } - return r.runScript(ctx, op, retryCmd, keys, argv...) + err = r.runScript(ctx, op, retryCmd, keys, argv...) + r.state(ctx, msg, base.TaskStateRetry, err) + return err } const ( @@ -899,7 +914,9 @@ func (r *RDB) Archive(ctx context.Context, msg *base.TaskMessage, errMsg string) expireAt.Unix(), int64(math.MaxInt64), } - return r.runScript(ctx, op, archiveCmd, keys, argv...) + err = r.runScript(ctx, op, archiveCmd, keys, argv...) + r.state(ctx, msg, base.TaskStateArchived, err) + return err } // ForwardIfReady checks scheduled and retry sets of the given queues @@ -1444,6 +1461,79 @@ func (r *RDB) ClearSchedulerEntries(scheduelrID string) error { return nil } +func (r *RDB) state(ctx context.Context, msg *base.TaskMessage, state base.TaskState, errs ...error) { + var err error + if len(errs) > 0 { + if errs[0] != nil { + err = errs[0] + } + } + out := map[string]interface{}{ + "queue": msg.Queue, + "id": msg.ID, + "state": state, + } + if err != nil { + out["err"] = err.Error() + } + if len(msg.GroupKey) > 0 { + out["group"] = msg.GroupKey + } + if len(msg.UniqueKey) > 0 { + out["unique"] = msg.UniqueKey + } + + payload, _ := json.Marshal(out) + r.client.Publish(ctx, "state-changed", payload) +} + +// StateChanged watch state updates, with more customized detail +func (r *RDB) SetTaskProber(prober base.TaskProber) { + ctx := context.Background() + pubsub := r.client.Subscribe(ctx, "state-changed") + + changed := func(out map[string]interface{}, err error) { + if err != nil { + out["err"] = "prober: " + err.Error() + } + go prober.Changed(out) + } + + handler := func(payload string) { + var out map[string]interface{} + err := json.Unmarshal([]byte(payload), &out) + if err != nil { + changed(out, err) + return + } + s, ok := out["state"].(float64) + if !ok { + changed(out, fmt.Errorf("invalid state %v", out["state"])) + return + } + state := base.TaskState(s) + out["state"] = state.String() + res, err := r.GetTaskInfo(out["queue"].(string), out["id"].(string)) + if err != nil { + changed(out, err) + return + } + msg := res.Message + if state == base.TaskStateCompleted { + out["at"] = msg.CompletedAt + } + key, data := prober.Result(state, res) + if data != nil { + out[key] = data + } + changed(out, nil) + } + + for m := range pubsub.Channel() { + handler(m.Payload) + } +} + // CancelationPubSub returns a pubsub for cancelation messages. func (r *RDB) CancelationPubSub() (*redis.PubSub, error) { var op errors.Op = "rdb.CancelationPubSub" diff --git a/server.go b/server.go index 4bf04e0c..77761e80 100644 --- a/server.go +++ b/server.go @@ -220,6 +220,53 @@ type Config struct { // // If unset or nil, the group aggregation feature will be disabled on the server. GroupAggregator GroupAggregator + + // StateChanged called when a task state changed + // + TaskStateProber *TaskStateProber +} + +// TaskStateProber tell there's a state changed happening +type TaskStateProber struct { + Probers map[string]string // map[state-string]data-name + Handler func(map[string]interface{}) +} + +func (p TaskStateProber) Changed(out map[string]interface{}) { + if p.Handler != nil { + p.Handler(out) + } +} + +func (p TaskStateProber) Result(state base.TaskState, raw *base.TaskInfo) (key string, data interface{}) { + if state == base.TaskStateCompleted { + data = *newTaskInfo(raw.Message, raw.State, raw.NextProcessAt, raw.Result) + return + } + + probers := p.Probers + if len(probers) == 0 { + probers = map[string]string{"*": "task"} + } + key, ok := probers["*"] + if !ok { + key, ok = probers[state.String()] + } + if !ok { + return + } + + switch key { + case "next": + data = raw.NextProcessAt + case "task": + data = *newTaskInfo(raw.Message, raw.State, raw.NextProcessAt, raw.Result) + default: + if len(raw.Result) > 0 { + data = raw.Result + } + } + return } // GroupAggregator aggregates a group of tasks into one before the tasks are passed to the Handler. @@ -458,6 +505,11 @@ func NewServer(r RedisConnOpt, cfg Config) *Server { srvState := &serverState{value: srvStateNew} cancels := base.NewCancelations() + taskStateProber := cfg.TaskStateProber + if taskStateProber != nil { + rdb.SetTaskProber(*taskStateProber) + } + syncer := newSyncer(syncerParams{ logger: logger, requestsCh: syncCh, @@ -697,3 +749,8 @@ func (srv *Server) Stop() { srv.processor.stop() srv.logger.Info("Processor stopped") } + +// StateChanged watch state updates, with more customized detail +func (srv *Server) SetTaskStateProber(prober base.TaskProber) { + srv.broker.SetTaskProber(prober) +} From a2473713666079db55e76f5ea62608deba0943e1 Mon Sep 17 00:00:00 2001 From: mindon Date: Mon, 21 Nov 2022 13:47:35 +0800 Subject: [PATCH 2/3] fix Result lack of key issue --- server.go | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/server.go b/server.go index 77761e80..3590a350 100644 --- a/server.go +++ b/server.go @@ -239,10 +239,14 @@ func (p TaskStateProber) Changed(out map[string]interface{}) { } func (p TaskStateProber) Result(state base.TaskState, raw *base.TaskInfo) (key string, data interface{}) { - if state == base.TaskStateCompleted { - data = *newTaskInfo(raw.Message, raw.State, raw.NextProcessAt, raw.Result) - return - } + defer func() { + if len(key) == 0 { + key = "task" + } + if data == nil { + data = *newTaskInfo(raw.Message, raw.State, raw.NextProcessAt, raw.Result) + } + }() probers := p.Probers if len(probers) == 0 { @@ -259,9 +263,7 @@ func (p TaskStateProber) Result(state base.TaskState, raw *base.TaskInfo) (key s switch key { case "next": data = raw.NextProcessAt - case "task": - data = *newTaskInfo(raw.Message, raw.State, raw.NextProcessAt, raw.Result) - default: + case "result": if len(raw.Result) > 0 { data = raw.Result } From 1fe0d8154497298f24f61b9ae281e44e8b3b2b8e Mon Sep 17 00:00:00 2001 From: mindon Date: Wed, 12 Jul 2023 10:46:55 +0800 Subject: [PATCH 3/3] fix: avoid retry if MaxRetry zero --- processor.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/processor.go b/processor.go index 925b9c90..001619bf 100644 --- a/processor.go +++ b/processor.go @@ -322,6 +322,12 @@ func (p *processor) markAsDone(l *base.Lease, msg *base.TaskMessage) { var SkipRetry = errors.New("skip retry for the task") func (p *processor) handleFailedMessage(ctx context.Context, l *base.Lease, msg *base.TaskMessage, err error) { + if msg.Retry == 0 { + if errors.Is(err, SkipRetry) { + p.markAsDone(l, msg) + return + } + } if p.errHandler != nil { p.errHandler.HandleError(ctx, NewTask(msg.Type, msg.Payload), err) } @@ -339,7 +345,7 @@ func (p *processor) handleFailedMessage(ctx context.Context, l *base.Lease, msg } func (p *processor) retry(l *base.Lease, msg *base.TaskMessage, e error, isFailure bool) { - if !l.IsValid() { + if !l.IsValid() || msg.Retry == 0 { // If lease is not valid, do not write to redis; Let recoverer take care of it. return }