Reviewed-on: #182 Co-authored-by: Tyr Mactire <tyr@pettingzoo.co> Co-committed-by: Tyr Mactire <tyr@pettingzoo.co>pull/183/head v0.10.0
parent
52e7a747fe
commit
cd97960f32
Signed by:
PettingZoo Gitea
PettingZoo Gitea
GPG Key ID:
39788A4390A1372F
@ -0,0 +1,30 @@
|
||||
package worker
|
||||
|
||||
import (
|
||||
"git.ptzo.gdn/feditools/relay/internal/logic"
|
||||
"git.ptzo.gdn/feditools/relay/internal/notification/manager"
|
||||
"git.ptzo.gdn/feditools/relay/internal/notification/telegram"
|
||||
)
|
||||
|
||||
func newNotifier(
|
||||
logicMod logic.Logic,
|
||||
) (*manager.Manager, error) {
|
||||
l := logger.WithField("func", "newNotifier")
|
||||
|
||||
newManager, err := manager.New()
|
||||
if err != nil {
|
||||
l.Errorf("notification manager: %s", err.Error())
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
telegramMod, err := telegram.New(logicMod)
|
||||
if err != nil {
|
||||
l.Errorf("notification telegram: %s", err.Error())
|
||||
|
||||
return nil, err
|
||||
}
|
||||
newManager.AddService(telegramMod)
|
||||
|
||||
return newManager, nil
|
||||
}
|
||||
@ -0,0 +1,4 @@
|
||||
FROM grafana/grafana:latest
|
||||
|
||||
ADD grafana.ini /etc/grafana/
|
||||
ADD datasource.yaml /etc/grafana/provisioning/datasources/
|
||||
@ -0,0 +1,9 @@
|
||||
apiVersion: 1
|
||||
|
||||
datasources:
|
||||
- name: Prometheus
|
||||
type: prometheus
|
||||
access: proxy
|
||||
url: http://prometheus:9090
|
||||
isDefault: true
|
||||
editable: true
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,3 @@
|
||||
FROM prom/prometheus
|
||||
|
||||
ADD prometheus.yml /etc/prometheus/
|
||||
@ -0,0 +1,11 @@
|
||||
global:
|
||||
scrape_interval: 15s
|
||||
evaluation_interval: 15s
|
||||
|
||||
scrape_configs:
|
||||
- job_name: "prometheus"
|
||||
static_configs:
|
||||
- targets: ["localhost:9090"]
|
||||
- job_name: "asynq"
|
||||
static_configs:
|
||||
- targets: ["asynqmon:8080"]
|
||||
@ -0,0 +1,59 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"git.ptzo.gdn/feditools/go-lib/fedihelper"
|
||||
"git.ptzo.gdn/feditools/relay/internal/runner"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeDeliverActivity = "activity:deliver"
|
||||
|
||||
type inboxDeliverPayload struct {
|
||||
InstanceID int64
|
||||
Activity fedihelper.Activity
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueDeliverActivity(ctx context.Context, instanceID int64, activity fedihelper.Activity) error {
|
||||
_, span := r.tracer.Start(ctx, "EnqueueDeliverActivity")
|
||||
defer span.End()
|
||||
|
||||
l := logger.WithField("func", "EnqueueDeliverActivity")
|
||||
|
||||
payload, err := json.Marshal(inboxDeliverPayload{
|
||||
InstanceID: instanceID,
|
||||
Activity: activity,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
task := asynq.NewTask(TypeDeliverActivity, payload, asynq.Queue(runner.QueueDelivery))
|
||||
|
||||
info, err := r.client.Enqueue(task, asynq.MaxRetry(5))
|
||||
if err != nil {
|
||||
l.Debugf("can't enqueue: %s", err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
l.Debugf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runner) handleDeliverActivity(ctx context.Context, t *asynq.Task) error {
|
||||
_, span := r.tracer.Start(ctx, "handleDeliverActivity")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleDeliverActivity")
|
||||
|
||||
var p inboxDeliverPayload
|
||||
if err := json.Unmarshal(t.Payload(), &p); err != nil {
|
||||
return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
// process activity
|
||||
return r.logic.DeliverActivity(ctx, "", p.InstanceID, p.Activity, false)
|
||||
}
|
||||
@ -0,0 +1,68 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"git.ptzo.gdn/feditools/go-lib/fedihelper"
|
||||
"github.com/hibiken/asynq"
|
||||
"net/url"
|
||||
)
|
||||
|
||||
const TypeInboxActivity = "activity:inbox"
|
||||
|
||||
type inboxActivityPayload struct {
|
||||
InstanceID int64
|
||||
ActorIRI *url.URL
|
||||
Activity fedihelper.Activity
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueInboxActivity(ctx context.Context, instanceID int64, actorIRI string, activity fedihelper.Activity) error {
|
||||
_, span := r.tracer.Start(ctx, "EnqueueInboxAction")
|
||||
defer span.End()
|
||||
|
||||
l := logger.WithField("func", "EnqueueInboxActivity")
|
||||
|
||||
actorURL, err := url.Parse(actorIRI)
|
||||
if err != nil {
|
||||
l.Errorf("cant parse url from argument 1: %s", err.Error())
|
||||
|
||||
return fmt.Errorf("cant parse url from argument 1: %s", err.Error())
|
||||
}
|
||||
|
||||
payload, err := json.Marshal(inboxActivityPayload{
|
||||
InstanceID: instanceID,
|
||||
ActorIRI: actorURL,
|
||||
Activity: activity,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
task := asynq.NewTask(TypeInboxActivity, payload)
|
||||
|
||||
info, err := r.client.Enqueue(task, asynq.MaxRetry(5))
|
||||
if err != nil {
|
||||
l.Debugf("can't enqueue: %s", err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
l.Debugf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runner) handleInboxActivity(ctx context.Context, t *asynq.Task) error {
|
||||
_, span := r.tracer.Start(ctx, "handleInboxActivity")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleInboxAction")
|
||||
|
||||
var p inboxActivityPayload
|
||||
if err := json.Unmarshal(t.Payload(), &p); err != nil {
|
||||
return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
// process activity
|
||||
return r.logic.ProcessActivity(ctx, "", p.InstanceID, p.ActorIRI, p.Activity)
|
||||
}
|
||||
@ -0,0 +1,55 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeProcessBlockAdd = "block:add"
|
||||
|
||||
type processBlockAddPayload struct {
|
||||
BlockID int64
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueProcessBlockAdd(ctx context.Context, blockID int64) error {
|
||||
_, span := r.tracer.Start(ctx, "EnqueueProcessBlockAdd")
|
||||
defer span.End()
|
||||
|
||||
l := logger.WithField("func", "EnqueueProcessBlockAdd")
|
||||
|
||||
payload, err := json.Marshal(processBlockAddPayload{
|
||||
BlockID: blockID,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
task := asynq.NewTask(TypeProcessBlockAdd, payload)
|
||||
|
||||
info, err := r.client.Enqueue(task)
|
||||
if err != nil {
|
||||
l.Debugf("can't enqueue: %s", err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
l.Debugf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runner) handleProcessBlockAdd(ctx context.Context, t *asynq.Task) error {
|
||||
_, span := r.tracer.Start(ctx, "handleProcessBlockAdd")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleProcessBlockAdd")
|
||||
|
||||
var p processBlockAddPayload
|
||||
if err := json.Unmarshal(t.Payload(), &p); err != nil {
|
||||
return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
// process activity
|
||||
return r.logic.ProcessBlockAdd(ctx, p.BlockID)
|
||||
}
|
||||
@ -0,0 +1,55 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeProcessBlockDelete = "block:delete"
|
||||
|
||||
type processBlockDeletePayload struct {
|
||||
BlockID int64
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueProcessBlockDelete(ctx context.Context, blockID int64) error {
|
||||
_, span := r.tracer.Start(ctx, "EnqueueProcessBlockDelete")
|
||||
defer span.End()
|
||||
|
||||
l := logger.WithField("func", "EnqueueProcessBlockDelete")
|
||||
|
||||
payload, err := json.Marshal(processBlockDeletePayload{
|
||||
BlockID: blockID,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
task := asynq.NewTask(TypeProcessBlockDelete, payload)
|
||||
|
||||
info, err := r.client.Enqueue(task)
|
||||
if err != nil {
|
||||
l.Debugf("can't enqueue: %s", err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
l.Debugf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runner) handleProcessBlockDelete(ctx context.Context, t *asynq.Task) error {
|
||||
_, span := r.tracer.Start(ctx, "handleProcessBlockDelete")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleProcessBlockDelete")
|
||||
|
||||
var p processBlockDeletePayload
|
||||
if err := json.Unmarshal(t.Payload(), &p); err != nil {
|
||||
return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
// process activity
|
||||
return r.logic.ProcessBlockDelete(ctx, p.BlockID)
|
||||
}
|
||||
@ -0,0 +1,55 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeProcessBlockUpdate = "block:update"
|
||||
|
||||
type processBlockUpdatePayload struct {
|
||||
BlockID int64
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueProcessBlockUpdate(ctx context.Context, blockID int64) error {
|
||||
_, span := r.tracer.Start(ctx, "EnqueueInboxAction")
|
||||
defer span.End()
|
||||
|
||||
l := logger.WithField("func", "EnqueueInboxActivity")
|
||||
|
||||
payload, err := json.Marshal(processBlockUpdatePayload{
|
||||
BlockID: blockID,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
task := asynq.NewTask(TypeProcessBlockUpdate, payload)
|
||||
|
||||
info, err := r.client.Enqueue(task)
|
||||
if err != nil {
|
||||
l.Debugf("can't enqueue: %s", err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
l.Debugf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runner) handleProcessBlockUpdate(ctx context.Context, t *asynq.Task) error {
|
||||
_, span := r.tracer.Start(ctx, "handleProcessBlockUpdate")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleProcessBlockUpdate")
|
||||
|
||||
var p processBlockUpdatePayload
|
||||
if err := json.Unmarshal(t.Payload(), &p); err != nil {
|
||||
return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
// process activity
|
||||
return r.logic.ProcessBlockUpdate(ctx, p.BlockID)
|
||||
}
|
||||
@ -0,0 +1,12 @@
|
||||
package asynq
|
||||
|
||||
import "git.ptzo.gdn/feditools/relay/internal/logic"
|
||||
|
||||
type Config struct {
|
||||
Logic logic.Logic
|
||||
|
||||
Concurrency int
|
||||
Address string
|
||||
Password string
|
||||
DB int
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"git.ptzo.gdn/feditools/relay/internal/log"
|
||||
)
|
||||
|
||||
type empty struct{}
|
||||
|
||||
var logger = log.WithPackageField(empty{})
|
||||
@ -0,0 +1,36 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeMaintDeliveryErrorTimeout = "maint:delivery-timeout"
|
||||
|
||||
func (r *Runner) EnqueueMaintDeliveryErrorTimeout(ctx context.Context) error {
|
||||
_, span := r.tracer.Start(ctx, "EnqueueMaintDeliveryErrorTimeout")
|
||||
defer span.End()
|
||||
|
||||
l := logger.WithField("func", "EnqueueMaintDeliveryErrorTimeout")
|
||||
|
||||
task := asynq.NewTask(TypeMaintDeliveryErrorTimeout, nil)
|
||||
|
||||
info, err := r.client.Enqueue(task)
|
||||
if err != nil {
|
||||
l.Debugf("can't enqueue: %s", err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
l.Debugf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runner) handleMaintDeliveryErrorTimeout(ctx context.Context, _ *asynq.Task) error {
|
||||
_, span := r.tracer.Start(ctx, "handleMaintDeliveryErrorTimeout")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleMaintDeliveryErrorTimeout")
|
||||
|
||||
return r.logic.MaintDeliveryErrorTimeout(ctx, "")
|
||||
}
|
||||
@ -0,0 +1,80 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"git.ptzo.gdn/feditools/relay/internal/logic"
|
||||
"git.ptzo.gdn/feditools/relay/internal/runner"
|
||||
"github.com/hibiken/asynq"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
)
|
||||
|
||||
func New(c *Config) (*Runner, error) {
|
||||
client := asynq.NewClient(asynq.RedisClientOpt{
|
||||
Addr: c.Address,
|
||||
Password: c.Password,
|
||||
DB: c.DB,
|
||||
})
|
||||
|
||||
return &Runner{
|
||||
client: client,
|
||||
logic: c.Logic,
|
||||
tracer: otel.Tracer("internal/runner/asynq"),
|
||||
|
||||
concurrency: c.Concurrency,
|
||||
address: c.Address,
|
||||
password: c.Password,
|
||||
db: c.DB,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type Runner struct {
|
||||
client *asynq.Client
|
||||
logic logic.Logic
|
||||
server *asynq.Server
|
||||
tracer trace.Tracer
|
||||
|
||||
concurrency int
|
||||
address string
|
||||
password string
|
||||
db int
|
||||
}
|
||||
|
||||
var _ runner.Runner = (*Runner)(nil)
|
||||
|
||||
func (r *Runner) Start(_ context.Context) error {
|
||||
r.server = asynq.NewServer(
|
||||
asynq.RedisClientOpt{
|
||||
Addr: r.address,
|
||||
Password: r.password,
|
||||
DB: r.db,
|
||||
},
|
||||
asynq.Config{
|
||||
Concurrency: r.concurrency,
|
||||
Queues: map[string]int{
|
||||
runner.QueuePriority: 6,
|
||||
runner.QueueDefault: 3,
|
||||
runner.QueueDelivery: 1,
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
mux := asynq.NewServeMux()
|
||||
mux.HandleFunc(TypeDeliverActivity, r.handleDeliverActivity)
|
||||
mux.HandleFunc(TypeInboxActivity, r.handleInboxActivity)
|
||||
mux.HandleFunc(TypeProcessBlockAdd, r.handleProcessBlockAdd)
|
||||
mux.HandleFunc(TypeProcessBlockDelete, r.handleProcessBlockDelete)
|
||||
mux.HandleFunc(TypeProcessBlockUpdate, r.handleProcessBlockUpdate)
|
||||
mux.HandleFunc(TypeMaintDeliveryErrorTimeout, r.handleMaintDeliveryErrorTimeout)
|
||||
mux.HandleFunc(TypeSendNotification, r.handleSendNotification)
|
||||
mux.HandleFunc(TypeUpdateAccountInfo, r.handleUpdateAccountInfo)
|
||||
mux.HandleFunc(TypeUpdateInstanceInfo, r.handleUpdateInstanceInfo)
|
||||
|
||||
return r.server.Run(mux)
|
||||
}
|
||||
|
||||
func (r *Runner) Stop() error {
|
||||
r.server.Stop()
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,63 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"git.ptzo.gdn/feditools/relay/internal/models"
|
||||
"git.ptzo.gdn/feditools/relay/internal/runner"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeSendNotification = "notification:send"
|
||||
|
||||
type sendNotificationPayload struct {
|
||||
Event models.EventType
|
||||
Metadata map[string]interface{}
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueSendNotification(ctx context.Context, event models.EventType, metadata map[string]interface{}) error {
|
||||
_, span := r.tracer.Start(ctx, "EnqueueSendNotification")
|
||||
defer span.End()
|
||||
|
||||
l := logger.WithField("func", "EnqueueSendNotification")
|
||||
|
||||
payload, err := json.Marshal(sendNotificationPayload{
|
||||
Event: event,
|
||||
Metadata: metadata,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
task := asynq.NewTask(TypeSendNotification, payload, asynq.Queue(runner.QueuePriority))
|
||||
|
||||
info, err := r.client.Enqueue(task)
|
||||
if err != nil {
|
||||
l.Debugf("can't enqueue: %s", err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
l.Debugf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runner) handleSendNotification(ctx context.Context, t *asynq.Task) error {
|
||||
_, span := r.tracer.Start(ctx, "handleSendNotification")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleSendNotification")
|
||||
|
||||
var p sendNotificationPayload
|
||||
if err := json.Unmarshal(t.Payload(), &p); err != nil {
|
||||
return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
// process activity
|
||||
if err := r.logic.SendNotification(ctx, "", p.Event, p.Metadata); err != nil {
|
||||
return fmt.Errorf("send notification failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,59 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeUpdateAccountInfo = "update:account"
|
||||
|
||||
type updateAccountInfoPayload struct {
|
||||
AccountID int64
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueUpdateAccountInfo(ctx context.Context, accountID int64) error {
|
||||
_, span := r.tracer.Start(ctx, "EnqueueUpdateAccountInfo")
|
||||
defer span.End()
|
||||
|
||||
l := logger.WithField("func", "EnqueueUpdateAccountInfo")
|
||||
|
||||
payload, err := json.Marshal(updateAccountInfoPayload{
|
||||
AccountID: accountID,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
task := asynq.NewTask(TypeUpdateAccountInfo, payload)
|
||||
|
||||
info, err := r.client.Enqueue(task)
|
||||
if err != nil {
|
||||
l.Debugf("can't enqueue: %s", err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
l.Debugf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runner) handleUpdateAccountInfo(ctx context.Context, t *asynq.Task) error {
|
||||
_, span := r.tracer.Start(ctx, "handleUpdateAccountInfo")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleUpdateAccountInfo")
|
||||
|
||||
var p updateAccountInfoPayload
|
||||
if err := json.Unmarshal(t.Payload(), &p); err != nil {
|
||||
return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
// process activity
|
||||
if err := r.logic.UpdateAccountInfo(ctx, "", p.AccountID); err != nil {
|
||||
return fmt.Errorf("account update failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,59 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeUpdateInstanceInfo = "update:instance"
|
||||
|
||||
type updateInstanceInfoPayload struct {
|
||||
InstanceID int64
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueUpdateInstanceInfo(ctx context.Context, instanceID int64) error {
|
||||
_, span := r.tracer.Start(ctx, "EnqueueUpdateInstanceInfo")
|
||||
defer span.End()
|
||||
|
||||
l := logger.WithField("func", "EnqueueUpdateInstanceInfo")
|
||||
|
||||
payload, err := json.Marshal(updateInstanceInfoPayload{
|
||||
InstanceID: instanceID,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
task := asynq.NewTask(TypeUpdateInstanceInfo, payload)
|
||||
|
||||
info, err := r.client.Enqueue(task)
|
||||
if err != nil {
|
||||
l.Debugf("can't enqueue: %s", err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
l.Debugf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runner) handleUpdateInstanceInfo(ctx context.Context, t *asynq.Task) error {
|
||||
_, span := r.tracer.Start(ctx, "handleUpdateInstanceInfo")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleUpdateInstanceInfo")
|
||||
|
||||
var p updateInstanceInfoPayload
|
||||
if err := json.Unmarshal(t.Payload(), &p); err != nil {
|
||||
return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
// process activity
|
||||
if err := r.logic.UpdateInstanceInfo(ctx, "", p.InstanceID); err != nil {
|
||||
return fmt.Errorf("account update failed: %v: %w", err, asynq.SkipRetry)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,27 @@
|
||||
vendor
|
||||
# Binaries for programs and plugins
|
||||
*.exe
|
||||
*.exe~
|
||||
*.dll
|
||||
*.so
|
||||
*.dylib
|
||||
|
||||
# Test binary, built with `go test -c`
|
||||
*.test
|
||||
|
||||
# Output of the go coverage tool, specifically when used with LiteIDE
|
||||
*.out
|
||||
|
||||
# Ignore examples for now
|
||||
/examples
|
||||
|
||||
# Ignore tool binaries
|
||||
/tools/asynq/asynq
|
||||
/tools/metrics_exporter/metrics_exporter
|
||||
|
||||
# Ignore asynq config file
|
||||
.asynq.*
|
||||
|
||||
# Ignore editor config files
|
||||
.vscode
|
||||
.idea
|
||||
@ -0,0 +1,526 @@
|
||||
# Changelog
|
||||
|
||||
All notable changes to this project will be documented in this file.
|
||||
|
||||
The format is based on ["Keep a Changelog"](https://keepachangelog.com/en/1.0.0/),
|
||||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [0.24.0] - 2023-01-02
|
||||
|
||||
### Added
|
||||
- `PreEnqueueFunc`, `PostEnqueueFunc` is added in `Scheduler` and deprecated `EnqueueErrorHandler` (PR: https://github.com/hibiken/asynq/pull/476)
|
||||
|
||||
### Changed
|
||||
- Removed error log when `Scheduler` failed to enqueue a task. Use `PostEnqueueFunc` to check for errors and task actions if needed.
|
||||
- Changed log level from ERROR to WARNINING when `Scheduler` failed to record `SchedulerEnqueueEvent`.
|
||||
|
||||
## [0.23.0] - 2022-04-11
|
||||
|
||||
### Added
|
||||
|
||||
- `Group` option is introduced to enqueue task in a group.
|
||||
- `GroupAggregator` and related types are introduced for task aggregation feature.
|
||||
- `GroupGracePeriod`, `GroupMaxSize`, `GroupMaxDelay`, and `GroupAggregator` fields are added to `Config`.
|
||||
- `Inspector` has new methods related to "aggregating tasks".
|
||||
- `Group` field is added to `TaskInfo`.
|
||||
- (CLI): `group ls` command is added
|
||||
- (CLI): `task ls` supports listing aggregating tasks via `--state=aggregating --group=<GROUP>` flags
|
||||
- Enable rediss url parsing support
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fixed overflow issue with 32-bit systems (For details, see https://github.com/hibiken/asynq/pull/426)
|
||||
|
||||
## [0.22.1] - 2022-02-20
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fixed Redis version compatibility: Keep support for redis v4.0+
|
||||
|
||||
## [0.22.0] - 2022-02-19
|
||||
|
||||
### Added
|
||||
|
||||
- `BaseContext` is introduced in `Config` to specify callback hook to provide a base `context` from which `Handler` `context` is derived
|
||||
- `IsOrphaned` field is added to `TaskInfo` to describe a task left in active state with no worker processing it.
|
||||
|
||||
### Changed
|
||||
|
||||
- `Server` now recovers tasks with an expired lease. Recovered tasks are retried/archived with `ErrLeaseExpired` error.
|
||||
|
||||
## [0.21.0] - 2022-01-22
|
||||
|
||||
### Added
|
||||
|
||||
- `PeriodicTaskManager` is added. Prefer using this over `Scheduler` as it has better support for dynamic periodic tasks.
|
||||
- The `asynq stats` command now supports a `--json` option, making its output a JSON object
|
||||
- Introduced new configuration for `DelayedTaskCheckInterval`. See [godoc](https://godoc.org/github.com/hibiken/asynq) for more details.
|
||||
|
||||
## [0.20.0] - 2021-12-19
|
||||
|
||||
### Added
|
||||
|
||||
- Package `x/metrics` is added.
|
||||
- Tool `tools/metrics_exporter` binary is added.
|
||||
- `ProcessedTotal` and `FailedTotal` fields were added to `QueueInfo` struct.
|
||||
|
||||
## [0.19.1] - 2021-12-12
|
||||
|
||||
### Added
|
||||
|
||||
- `Latency` field is added to `QueueInfo`.
|
||||
- `EnqueueContext` method is added to `Client`.
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fixed an error when user pass a duration less than 1s to `Unique` option
|
||||
|
||||
## [0.19.0] - 2021-11-06
|
||||
|
||||
### Changed
|
||||
|
||||
- `NewTask` takes `Option` as variadic argument
|
||||
- Bumped minimum supported go version to 1.14 (i.e. go1.14 or higher is required).
|
||||
|
||||
### Added
|
||||
|
||||
- `Retention` option is added to allow user to specify task retention duration after completion.
|
||||
- `TaskID` option is added to allow user to specify task ID.
|
||||
- `ErrTaskIDConflict` sentinel error value is added.
|
||||
- `ResultWriter` type is added and provided through `Task.ResultWriter` method.
|
||||
- `TaskInfo` has new fields `CompletedAt`, `Result` and `Retention`.
|
||||
|
||||
### Removed
|
||||
|
||||
- `Client.SetDefaultOptions` is removed. Use `NewTask` instead to pass default options for tasks.
|
||||
|
||||
## [0.18.6] - 2021-10-03
|
||||
|
||||
### Changed
|
||||
|
||||
- Updated `github.com/go-redis/redis` package to v8
|
||||
|
||||
## [0.18.5] - 2021-09-01
|
||||
|
||||
### Added
|
||||
|
||||
- `IsFailure` config option is added to determine whether error returned from Handler counts as a failure.
|
||||
|
||||
## [0.18.4] - 2021-08-17
|
||||
|
||||
### Fixed
|
||||
|
||||
- Scheduler methods are now thread-safe. It's now safe to call `Register` and `Unregister` concurrently.
|
||||
|
||||
## [0.18.3] - 2021-08-09
|
||||
|
||||
### Changed
|
||||
|
||||
- `Client.Enqueue` no longer enqueues tasks with empty typename; Error message is returned.
|
||||
|
||||
## [0.18.2] - 2021-07-15
|
||||
|
||||
### Changed
|
||||
|
||||
- Changed `Queue` function to not to convert the provided queue name to lowercase. Queue names are now case-sensitive.
|
||||
- `QueueInfo.MemoryUsage` is now an approximate usage value.
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fixed latency issue around memory usage (see https://github.com/hibiken/asynq/issues/309).
|
||||
|
||||
## [0.18.1] - 2021-07-04
|
||||
|
||||
### Changed
|
||||
|
||||
- Changed to execute task recovering logic when server starts up; Previously it needed to wait for a minute for task recovering logic to exeucte.
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fixed task recovering logic to execute every minute
|
||||
|
||||
## [0.18.0] - 2021-06-29
|
||||
|
||||
### Changed
|
||||
|
||||
- NewTask function now takes array of bytes as payload.
|
||||
- Task `Type` and `Payload` should be accessed by a method call.
|
||||
- `Server` API has changed. Renamed `Quiet` to `Stop`. Renamed `Stop` to `Shutdown`. _Note:_ As a result of this renaming, the behavior of `Stop` has changed. Please update the exising code to call `Shutdown` where it used to call `Stop`.
|
||||
- `Scheduler` API has changed. Renamed `Stop` to `Shutdown`.
|
||||
- Requires redis v4.0+ for multiple field/value pair support
|
||||
- `Client.Enqueue` now returns `TaskInfo`
|
||||
- `Inspector.RunTaskByKey` is replaced with `Inspector.RunTask`
|
||||
- `Inspector.DeleteTaskByKey` is replaced with `Inspector.DeleteTask`
|
||||
- `Inspector.ArchiveTaskByKey` is replaced with `Inspector.ArchiveTask`
|
||||
- `inspeq` package is removed. All types and functions from the package is moved to `asynq` package.
|
||||