508 lines
11 KiB
Go
508 lines
11 KiB
Go
package worker
|
|
|
|
import (
|
|
"context"
|
|
"encoding/binary"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"regexp"
|
|
"runtime"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"nano-run/services/blob"
|
|
"nano-run/services/blob/fsblob"
|
|
"nano-run/services/meta"
|
|
"nano-run/services/meta/micrometa"
|
|
"nano-run/services/queue"
|
|
"nano-run/services/queue/microqueue"
|
|
)
|
|
|
|
type (
|
|
CompleteHandler func(ctx context.Context, requestID string, info *meta.Request)
|
|
ProcessHandler func(ctx context.Context, requestID, attemptID string, info *meta.Request)
|
|
)
|
|
|
|
const (
|
|
defaultAttempts = 3
|
|
defaultInterval = 3 * time.Second
|
|
minimalFailedCode = 500
|
|
nsRequest byte = 0x00
|
|
nsAttempt byte = 0x01
|
|
)
|
|
|
|
func Default(location string) (*Worker, error) {
|
|
path := filepath.Join(location, "blobs")
|
|
err := os.MkdirAll(path, 0755)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
valid, err := regexp.Compile("^[a-zA-Z0-9-]+$")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
storage := fsblob.NewCheck(path, func(id string) bool {
|
|
return valid.MatchString(id)
|
|
})
|
|
|
|
taskQueue, err := microqueue.NewMicroQueue(filepath.Join(location, "queue"))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
requeue, err := microqueue.NewMicroQueue(filepath.Join(location, "requeue"))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
metaStorage, err := micrometa.NewMetaStorage(filepath.Join(location, "meta"))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cleanup := func() {
|
|
_ = requeue.Close()
|
|
_ = taskQueue.Close()
|
|
_ = metaStorage.Close()
|
|
}
|
|
|
|
wrk, err := New(taskQueue, requeue, storage, metaStorage)
|
|
if err != nil {
|
|
cleanup()
|
|
return nil, err
|
|
}
|
|
wrk.cleanup = cleanup
|
|
return wrk, nil
|
|
}
|
|
|
|
func New(tasks, requeue queue.Queue, blobs blob.Blob, meta meta.Meta) (*Worker, error) {
|
|
wrk := &Worker{
|
|
queue: tasks,
|
|
requeue: requeue,
|
|
blob: blobs,
|
|
meta: meta,
|
|
reloadMeta: make(chan struct{}, 1),
|
|
maxAttempts: defaultAttempts,
|
|
interval: defaultInterval,
|
|
concurrency: runtime.NumCPU(),
|
|
handler: http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
|
|
writer.WriteHeader(http.StatusNoContent)
|
|
}),
|
|
}
|
|
err := wrk.init()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return wrk, nil
|
|
}
|
|
|
|
type Worker struct {
|
|
queue queue.Queue
|
|
requeue queue.Queue
|
|
blob blob.Blob
|
|
meta meta.Meta
|
|
handler http.Handler
|
|
cleanup func()
|
|
|
|
onDead CompleteHandler
|
|
onSuccess CompleteHandler
|
|
onProcess ProcessHandler
|
|
maxAttempts int
|
|
concurrency int
|
|
reloadMeta chan struct{}
|
|
interval time.Duration
|
|
sequence uint64
|
|
}
|
|
|
|
func (mgr *Worker) init() error {
|
|
return mgr.meta.Iterate(func(id string, record meta.Request) error {
|
|
if _, v, err := decodeID(id); err == nil && v > mgr.sequence {
|
|
mgr.sequence = v
|
|
} else if err != nil {
|
|
log.Println("found broken id:", id, "-", err)
|
|
}
|
|
if !record.Complete {
|
|
log.Println("found incomplete job", id)
|
|
return mgr.queue.Push([]byte(id))
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// Cleanup internal resource.
|
|
func (mgr *Worker) Close() {
|
|
if fn := mgr.cleanup; fn != nil {
|
|
fn()
|
|
}
|
|
}
|
|
|
|
// Enqueue request to storage, save meta-info to meta storage and push id into processing queue. Generated ID
|
|
// always unique and returns only in case of successful enqueue.
|
|
func (mgr *Worker) Enqueue(req *http.Request) (string, error) {
|
|
id, err := mgr.saveRequest(req)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
log.Println("new request saved:", id)
|
|
err = mgr.queue.Push([]byte(id))
|
|
return id, err
|
|
}
|
|
|
|
// Complete request manually.
|
|
func (mgr *Worker) Complete(requestID string) error {
|
|
err := mgr.meta.Complete(requestID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
select {
|
|
case mgr.reloadMeta <- struct{}{}:
|
|
default:
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (mgr *Worker) OnSuccess(handler CompleteHandler) *Worker {
|
|
mgr.onSuccess = handler
|
|
return mgr
|
|
}
|
|
|
|
func (mgr *Worker) OnDead(handler CompleteHandler) *Worker {
|
|
mgr.onDead = handler
|
|
return mgr
|
|
}
|
|
|
|
func (mgr *Worker) OnProcess(handler ProcessHandler) *Worker {
|
|
mgr.onProcess = handler
|
|
return mgr
|
|
}
|
|
|
|
func (mgr *Worker) Handler(handler http.Handler) *Worker {
|
|
mgr.handler = handler
|
|
return mgr
|
|
}
|
|
|
|
func (mgr *Worker) HandlerFunc(fn http.HandlerFunc) *Worker {
|
|
mgr.handler = fn
|
|
return mgr
|
|
}
|
|
|
|
// Attempts number of 500x requests.
|
|
func (mgr *Worker) Attempts(max int) *Worker {
|
|
mgr.maxAttempts = max
|
|
return mgr
|
|
}
|
|
|
|
// Interval between attempts.
|
|
func (mgr *Worker) Interval(duration time.Duration) *Worker {
|
|
mgr.interval = duration
|
|
return mgr
|
|
}
|
|
|
|
// Concurrency limit (number of parallel tasks). Does not affect already running worker.
|
|
// 0 means num CPU.
|
|
func (mgr *Worker) Concurrency(num int) *Worker {
|
|
mgr.concurrency = num
|
|
if num == 0 {
|
|
mgr.concurrency = runtime.NumCPU()
|
|
}
|
|
return mgr
|
|
}
|
|
|
|
// Meta information about requests.
|
|
func (mgr *Worker) Meta() meta.Meta {
|
|
return mgr.meta
|
|
}
|
|
|
|
// Blobs storage (for large objects).
|
|
func (mgr *Worker) Blobs() blob.Blob {
|
|
return mgr.blob
|
|
}
|
|
|
|
func (mgr *Worker) Run(global context.Context) error {
|
|
if mgr.interval < 0 {
|
|
return fmt.Errorf("negative interval")
|
|
}
|
|
if mgr.maxAttempts < 0 {
|
|
return fmt.Errorf("negative attempts")
|
|
}
|
|
if mgr.handler == nil {
|
|
return fmt.Errorf("nil handler")
|
|
}
|
|
if mgr.concurrency <= 0 {
|
|
return fmt.Errorf("invalid concurrency number")
|
|
}
|
|
ctx, cancel := context.WithCancel(global)
|
|
defer cancel()
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < mgr.concurrency; i++ {
|
|
wg.Add(1)
|
|
go func(i int) {
|
|
defer wg.Done()
|
|
defer cancel()
|
|
err := mgr.runQueue(ctx)
|
|
if err != nil {
|
|
log.Println("worker", i, "stopped due to error:", err)
|
|
} else {
|
|
log.Println("worker", i, "stopped")
|
|
}
|
|
}(i)
|
|
}
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
defer cancel()
|
|
err := mgr.runReQueue(ctx)
|
|
if err != nil {
|
|
log.Println("re-queue process stopped due to error:", err)
|
|
} else {
|
|
log.Println("re-queue process stopped")
|
|
}
|
|
}()
|
|
wg.Wait()
|
|
return ctx.Err()
|
|
}
|
|
|
|
// Retry processing.
|
|
func (mgr *Worker) Retry(ctx context.Context, requestID string) (string, error) {
|
|
info, err := mgr.meta.Get(requestID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
req, err := mgr.restoreRequest(ctx, requestID, info)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer req.Body.Close()
|
|
return mgr.Enqueue(req)
|
|
}
|
|
|
|
func (mgr *Worker) call(ctx context.Context, requestID string, info *meta.Request) error {
|
|
// caller should ensure that request id is valid
|
|
req, err := mgr.restoreRequest(ctx, requestID, info)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer req.Body.Close()
|
|
attemptID := encodeID(nsAttempt, uint64(len(info.Attempts))+1)
|
|
|
|
req.Header.Set("X-Correlation-Id", requestID)
|
|
req.Header.Set("X-Attempt-Id", attemptID)
|
|
req.Header.Set("X-Attempt", strconv.Itoa(len(info.Attempts)+1))
|
|
|
|
var header meta.AttemptHeader
|
|
|
|
err = mgr.blob.Push(attemptID, func(out io.Writer) error {
|
|
res := openResponse(out)
|
|
mgr.handler.ServeHTTP(res, req)
|
|
header = res.meta
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
info, err = mgr.meta.AddAttempt(requestID, attemptID, header)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
mgr.requestProcessed(ctx, requestID, attemptID, info)
|
|
if header.Code >= minimalFailedCode {
|
|
return fmt.Errorf("500 code returned: %d", header.Code)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (mgr *Worker) runQueue(ctx context.Context) error {
|
|
for {
|
|
err := mgr.processQueueItem(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (mgr *Worker) runReQueue(ctx context.Context) error {
|
|
for {
|
|
err := mgr.processReQueueItem(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (mgr *Worker) processQueueItem(ctx context.Context) error {
|
|
bid, err := mgr.queue.Get(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
id := string(bid)
|
|
log.Println("processing request", id)
|
|
info, err := mgr.meta.Get(id)
|
|
if err != nil {
|
|
return fmt.Errorf("get request %s meta info: %w", id, err)
|
|
}
|
|
if info.Complete {
|
|
log.Printf("request %s already complete", id)
|
|
return nil
|
|
}
|
|
err = mgr.call(ctx, id, info)
|
|
if err == nil {
|
|
mgr.requestSuccess(ctx, id, info)
|
|
return nil
|
|
}
|
|
return mgr.requeueItem(ctx, id, info)
|
|
}
|
|
|
|
func (mgr *Worker) processReQueueItem(ctx context.Context) error {
|
|
var item requeueItem
|
|
|
|
data, err := mgr.requeue.Get(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = json.Unmarshal(data, &item)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
d := time.Since(item.At)
|
|
if d < mgr.interval {
|
|
var ok = false
|
|
for !ok {
|
|
info, err := mgr.meta.Get(item.ID)
|
|
if err != nil {
|
|
return fmt.Errorf("re-queue: get meta %s: %w", item.ID, err)
|
|
}
|
|
if info.Complete {
|
|
log.Printf("re-queue: %s already complete", item.ID)
|
|
return nil
|
|
}
|
|
select {
|
|
case <-time.After(mgr.interval - d):
|
|
ok = true
|
|
case <-mgr.reloadMeta:
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
}
|
|
return mgr.queue.Push([]byte(item.ID))
|
|
}
|
|
|
|
func (mgr *Worker) requeueItem(ctx context.Context, id string, info *meta.Request) error {
|
|
if len(info.Attempts) >= mgr.maxAttempts {
|
|
mgr.requestDead(ctx, id, info)
|
|
log.Println("maximum attempts reached for request", id)
|
|
return nil
|
|
}
|
|
data, err := json.Marshal(requeueItem{
|
|
At: time.Now(),
|
|
ID: id,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return mgr.requeue.Push(data)
|
|
}
|
|
|
|
func (mgr *Worker) saveRequest(req *http.Request) (string, error) {
|
|
id := encodeID(nsRequest, atomic.AddUint64(&mgr.sequence, 1))
|
|
err := mgr.blob.Push(id, func(out io.Writer) error {
|
|
_, err := io.Copy(out, req.Body)
|
|
return err
|
|
})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return id, mgr.meta.CreateRequest(id, req.Header, req.URL.RequestURI(), req.Method)
|
|
}
|
|
|
|
func (mgr *Worker) requestDead(ctx context.Context, id string, info *meta.Request) {
|
|
err := mgr.meta.Complete(id)
|
|
if err != nil {
|
|
log.Println("failed complete (dead) request:", err)
|
|
}
|
|
if handler := mgr.onDead; handler != nil {
|
|
handler(ctx, id, info)
|
|
}
|
|
log.Println("request", id, "completely failed")
|
|
}
|
|
|
|
func (mgr *Worker) requestSuccess(ctx context.Context, id string, info *meta.Request) {
|
|
err := mgr.meta.Complete(id)
|
|
if err != nil {
|
|
log.Println("failed complete (success) request:", err)
|
|
}
|
|
if handler := mgr.onSuccess; handler != nil {
|
|
handler(ctx, id, info)
|
|
}
|
|
log.Println("request", id, "complete successfully")
|
|
}
|
|
|
|
func (mgr *Worker) requestProcessed(ctx context.Context, id string, attemptID string, info *meta.Request) {
|
|
if handler := mgr.onProcess; handler != nil {
|
|
handler(ctx, id, attemptID, info)
|
|
}
|
|
log.Println("request", id, "processed with attempt", attemptID)
|
|
}
|
|
|
|
func (mgr *Worker) restoreRequest(ctx context.Context, requestID string, info *meta.Request) (*http.Request, error) {
|
|
f, err := mgr.blob.Get(requestID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, info.Method, info.URI, f)
|
|
if err != nil {
|
|
_ = f.Close()
|
|
return nil, err
|
|
}
|
|
for k, v := range info.Headers {
|
|
req.Header[k] = v
|
|
}
|
|
return req, nil
|
|
}
|
|
|
|
func encodeID(nsId byte, id uint64) string {
|
|
var data [9]byte
|
|
data[0] = nsId
|
|
binary.BigEndian.PutUint64(data[1:], id)
|
|
return strings.ToUpper(hex.EncodeToString(data[:]))
|
|
}
|
|
|
|
func decodeID(val string) (byte, uint64, error) {
|
|
hx, err := hex.DecodeString(val)
|
|
if err != nil {
|
|
return 0, 0, err
|
|
}
|
|
if len(hx) != 9 {
|
|
return 0, 0, errors.New("too short")
|
|
}
|
|
n := binary.BigEndian.Uint64(hx[1:])
|
|
return hx[0], n, nil
|
|
}
|
|
|
|
type requeueItem struct {
|
|
At time.Time
|
|
ID string
|
|
}
|