finish etcd backend

backends/etcd/etcd.go

@@ -5,13 +5,13 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"log"
 	"sync"
 	"time"
 
 	"github.com/RichardKnop/machinery/v2/backends/iface"
 	"github.com/RichardKnop/machinery/v2/common"
 	"github.com/RichardKnop/machinery/v2/config"
+	"github.com/RichardKnop/machinery/v2/log"
 	"github.com/RichardKnop/machinery/v2/tasks"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"golang.org/x/sync/errgroup"
@@ -25,7 +25,6 @@ const (
 type etcdBackend struct {
 	common.Backend
 	ctx    context.Context
-	conf   *config.Config
 	client *clientv3.Client
 }
 
@@ -45,7 +44,6 @@ func New(ctx context.Context, conf *config.Config) (iface.Backend, error) {
 		Backend: common.NewBackend(conf),
 		ctx:     ctx,
 		client:  client,
-		conf:    conf,
 	}
 
 	return &backend, nil
@@ -54,26 +52,28 @@ func New(ctx context.Context, conf *config.Config) (iface.Backend, error) {
 
 // Group related functions
 func (b *etcdBackend) InitGroup(groupUUID string, taskUUIDs []string) error {
+	lease, err := b.getLease()
+	if err != nil {
+		return err
+	}
+
 	groupMeta := &tasks.GroupMeta{
 		GroupUUID: groupUUID,
 		TaskUUIDs: taskUUIDs,
 		CreatedAt: time.Now().UTC(),
+		TTL:       lease.TTL,
 	}
 
 	encoded, err := json.Marshal(groupMeta)
 	if err != nil {
 		return err
 	}
-	lease, err := b.getLease()
-	if err != nil {
-		return err
-	}
 
 	ctx, cancel := context.WithTimeout(b.ctx, time.Second*2)
 	defer cancel()
 
 	key := fmt.Sprintf(groupKey, groupUUID)
-	_, err = b.client.Put(ctx, key, string(encoded), clientv3.WithLease(lease))
+	_, err = b.client.Put(ctx, key, string(encoded), clientv3.WithLease(lease.ID))
 	return err
 }
 
@@ -131,8 +131,14 @@ func (b *etcdBackend) TriggerChord(groupUUID string) (bool, error) {
 		return false, nil
 	}
 
+	lease, err := b.getLease()
+	if err != nil {
+		return false, err
+	}
+
 	// Set flag to true
 	meta.ChordTriggered = true
+	meta.TTL = lease.TTL
 
 	// Update the group meta
 	encoded, err := json.Marshal(&meta)
@@ -140,13 +146,8 @@ func (b *etcdBackend) TriggerChord(groupUUID string) (bool, error) {
 		return false, err
 	}
 
-	lease, err := b.getLease()
-	if err != nil {
-		return false, err
-	}
-
 	cmp := clientv3.Compare(clientv3.ModRevision(key), "=", kv.ModRevision)
-	update := clientv3.OpPut(key, string(encoded), clientv3.WithLease(lease))
+	update := clientv3.OpPut(key, string(encoded), clientv3.WithLease(lease.ID))
 
 	b.client.Txn(b.ctx).If(cmp).Then(update)
 	txnresp, err := b.client.Txn(b.ctx).If(cmp).Then(update).Commit()
@@ -300,28 +301,29 @@ func (b *etcdBackend) getStates(taskUUIDs ...string) ([]*tasks.TaskState, error)
 
 // updateState saves current task state
 func (b *etcdBackend) updateState(taskState *tasks.TaskState) error {
+	lease, err := b.getLease()
+	if err != nil {
+		return err
+	}
+	taskState.TTL = lease.TTL
+
 	encoded, err := json.Marshal(taskState)
 	if err != nil {
 		return err
 	}
 
-	lease, err := b.getLease()
-	if err != nil {
-		return err
-	}
-
 	key := fmt.Sprintf(taskKey, taskState.TaskUUID)
-	_, err = b.client.Put(b.ctx, key, string(encoded), clientv3.WithLease(lease))
+	_, err = b.client.Put(b.ctx, key, string(encoded), clientv3.WithLease(lease.ID))
 	if err != nil {
 		return err
 	}
 
-	log.Default().Printf("update taskstate %s %s, %s", taskState.TaskName, taskState.TaskUUID, encoded)
+	log.DEBUG.Printf("update taskstate %s %s, %s", taskState.TaskName, taskState.TaskUUID, encoded)
 	return nil
 }
 
 // getExpiration returns expiration for a stored task state
-func (b *etcdBackend) getLease() (clientv3.LeaseID, error) {
+func (b *etcdBackend) getLease() (*clientv3.LeaseGrantResponse, error) {
 	expiresIn := b.GetConfig().ResultsExpireIn
 	if expiresIn <= 0 {
 		// expire results after 1 hour by default
@@ -330,8 +332,8 @@ func (b *etcdBackend) getLease() (clientv3.LeaseID, error) {
 
 	resp, err := b.client.Grant(b.ctx, int64(expiresIn))
 	if err != nil {
-		return clientv3.NoLease, err
+		return nil, err
 	}
 
-	return resp.ID, nil
+	return resp, nil
 }

examples/etcd/main.go

@@ -161,7 +161,11 @@ func send() error {
 	}
 
 	var (
-		addTask0 tasks.Signature
+		addTask0, addTask1, addTask2                      tasks.Signature
+		multiplyTask0, multiplyTask1                      tasks.Signature
+		sumIntsTask, sumFloatsTask, concatTask, splitTask tasks.Signature
+		panicTask                                         tasks.Signature
+		longRunningTask                                   tasks.Signature
 	)
 
 	var initTasks = func() {
@@ -179,6 +183,95 @@ func send() error {
 			},
 		}
 
+		addTask1 = tasks.Signature{
+			Name: "add",
+			Args: []tasks.Arg{
+				{
+					Type:  "int64",
+					Value: 2,
+				},
+				{
+					Type:  "int64",
+					Value: 2,
+				},
+			},
+		}
+
+		addTask2 = tasks.Signature{
+			Name: "add",
+			Args: []tasks.Arg{
+				{
+					Type:  "int64",
+					Value: 5,
+				},
+				{
+					Type:  "int64",
+					Value: 6,
+				},
+			},
+		}
+
+		multiplyTask0 = tasks.Signature{
+			Name: "multiply",
+			Args: []tasks.Arg{
+				{
+					Type:  "int64",
+					Value: 4,
+				},
+			},
+		}
+
+		multiplyTask1 = tasks.Signature{
+			Name: "multiply",
+		}
+
+		sumIntsTask = tasks.Signature{
+			Name: "sum_ints",
+			Args: []tasks.Arg{
+				{
+					Type:  "[]int64",
+					Value: []int64{1, 2},
+				},
+			},
+		}
+
+		sumFloatsTask = tasks.Signature{
+			Name: "sum_floats",
+			Args: []tasks.Arg{
+				{
+					Type:  "[]float64",
+					Value: []float64{1.5, 2.7},
+				},
+			},
+		}
+
+		concatTask = tasks.Signature{
+			Name: "concat",
+			Args: []tasks.Arg{
+				{
+					Type:  "[]string",
+					Value: []string{"foo", "bar"},
+				},
+			},
+		}
+
+		splitTask = tasks.Signature{
+			Name: "split",
+			Args: []tasks.Arg{
+				{
+					Type:  "string",
+					Value: "foo",
+				},
+			},
+		}
+
+		panicTask = tasks.Signature{
+			Name: "panic_task",
+		}
+
+		longRunningTask = tasks.Signature{
+			Name: "long_running_task",
+		}
 	}
 
 	/*
@@ -194,26 +287,172 @@ func send() error {
 	span.LogFields(opentracinglog.String("batch.id", batchID))
 
 	log.INFO.Println("Starting batch:", batchID)
-
 	/*
 	 * First, let's try sending a single task
 	 */
 	initTasks()
 
-	now := time.Now().Add(time.Second * 10)
+	log.INFO.Println("Single task:")
-	addTask0.ETA = &now
+
 	asyncResult, err := server.SendTaskWithContext(ctx, &addTask0)
 	if err != nil {
 		return fmt.Errorf("Could not send task: %s", err.Error())
 	}
 
-	log.INFO.Println("Single task:", asyncResult.Signature.UUID)
+	results, err := asyncResult.Get(time.Millisecond * 5)
+	if err != nil {
+		return fmt.Errorf("Getting task result failed with error: %s", err.Error())
+	}
+	log.INFO.Printf("1 + 1 = %v\n", tasks.HumanReadableResults(results))
 
-	if err := server.RegisterPeriodicTask("* * * * *", "hello", &addTask0); err != nil {
+	/*
-		return err
+	 * Try couple of tasks with a slice argument and slice return value
+	 */
+	asyncResult, err = server.SendTaskWithContext(ctx, &sumIntsTask)
+	if err != nil {
+		return fmt.Errorf("Could not send task: %s", err.Error())
 	}
 
-	time.Sleep(time.Second * 120000)
+	results, err = asyncResult.Get(time.Millisecond * 5)
+	if err != nil {
+		return fmt.Errorf("Getting task result failed with error: %s", err.Error())
+	}
+	log.INFO.Printf("sum([1, 2]) = %v\n", tasks.HumanReadableResults(results))
+
+	asyncResult, err = server.SendTaskWithContext(ctx, &sumFloatsTask)
+	if err != nil {
+		return fmt.Errorf("Could not send task: %s", err.Error())
+	}
+
+	results, err = asyncResult.Get(time.Millisecond * 5)
+	if err != nil {
+		return fmt.Errorf("Getting task result failed with error: %s", err.Error())
+	}
+	log.INFO.Printf("sum([1.5, 2.7]) = %v\n", tasks.HumanReadableResults(results))
+
+	asyncResult, err = server.SendTaskWithContext(ctx, &concatTask)
+	if err != nil {
+		return fmt.Errorf("Could not send task: %s", err.Error())
+	}
+
+	results, err = asyncResult.Get(time.Millisecond * 5)
+	if err != nil {
+		return fmt.Errorf("Getting task result failed with error: %s", err.Error())
+	}
+	log.INFO.Printf("concat([\"foo\", \"bar\"]) = %v\n", tasks.HumanReadableResults(results))
+
+	asyncResult, err = server.SendTaskWithContext(ctx, &splitTask)
+	if err != nil {
+		return fmt.Errorf("Could not send task: %s", err.Error())
+	}
+
+	results, err = asyncResult.Get(time.Millisecond * 5)
+	if err != nil {
+		return fmt.Errorf("Getting task result failed with error: %s", err.Error())
+	}
+	log.INFO.Printf("split([\"foo\"]) = %v\n", tasks.HumanReadableResults(results))
+
+	/*
+	 * Now let's explore ways of sending multiple tasks
+	 */
+
+	// Now let's try a parallel execution
+	initTasks()
+	log.INFO.Println("Group of tasks (parallel execution):")
+
+	group, err := tasks.NewGroup(&addTask0, &addTask1, &addTask2)
+	if err != nil {
+		return fmt.Errorf("Error creating group: %s", err.Error())
+	}
+
+	asyncResults, err := server.SendGroupWithContext(ctx, group, 10)
+	if err != nil {
+		return fmt.Errorf("Could not send group: %s", err.Error())
+	}
+
+	for _, asyncResult := range asyncResults {
+		results, err = asyncResult.Get(time.Millisecond * 5)
+		if err != nil {
+			return fmt.Errorf("Getting task result failed with error: %s", err.Error())
+		}
+		log.INFO.Printf(
+			"%v + %v = %v\n",
+			asyncResult.Signature.Args[0].Value,
+			asyncResult.Signature.Args[1].Value,
+			tasks.HumanReadableResults(results),
+		)
+	}
+
+	// Now let's try a group with a chord
+	initTasks()
+	log.INFO.Println("Group of tasks with a callback (chord):")
+
+	group, err = tasks.NewGroup(&addTask0, &addTask1, &addTask2)
+	if err != nil {
+		return fmt.Errorf("Error creating group: %s", err.Error())
+	}
+
+	chord, err := tasks.NewChord(group, &multiplyTask1)
+	if err != nil {
+		return fmt.Errorf("Error creating chord: %s", err)
+	}
+
+	chordAsyncResult, err := server.SendChordWithContext(ctx, chord, 10)
+	if err != nil {
+		return fmt.Errorf("Could not send chord: %s", err.Error())
+	}
+
+	results, err = chordAsyncResult.Get(time.Millisecond * 5)
+	if err != nil {
+		return fmt.Errorf("Getting chord result failed with error: %s", err.Error())
+	}
+	log.INFO.Printf("(1 + 1) * (2 + 2) * (5 + 6) = %v\n", tasks.HumanReadableResults(results))
+
+	// Now let's try chaining task results
+	initTasks()
+	log.INFO.Println("Chain of tasks:")
+
+	chain, err := tasks.NewChain(&addTask0, &addTask1, &addTask2, &multiplyTask0)
+	if err != nil {
+		return fmt.Errorf("Error creating chain: %s", err)
+	}
+
+	chainAsyncResult, err := server.SendChainWithContext(ctx, chain)
+	if err != nil {
+		return fmt.Errorf("Could not send chain: %s", err.Error())
+	}
+
+	results, err = chainAsyncResult.Get(time.Millisecond * 5)
+	if err != nil {
+		return fmt.Errorf("Getting chain result failed with error: %s", err.Error())
+	}
+	log.INFO.Printf("(((1 + 1) + (2 + 2)) + (5 + 6)) * 4 = %v\n", tasks.HumanReadableResults(results))
+
+	// Let's try a task which throws panic to make sure stack trace is not lost
+	initTasks()
+	asyncResult, err = server.SendTaskWithContext(ctx, &panicTask)
+	if err != nil {
+		return fmt.Errorf("Could not send task: %s", err.Error())
+	}
+
+	_, err = asyncResult.Get(time.Millisecond * 5)
+	if err == nil {
+		return errors.New("Error should not be nil if task panicked")
+	}
+	log.INFO.Printf("Task panicked and returned error = %v\n", err.Error())
+
+	// Let's try a long running task
+	initTasks()
+	asyncResult, err = server.SendTaskWithContext(ctx, &longRunningTask)
+	if err != nil {
+		return fmt.Errorf("Could not send task: %s", err.Error())
+	}
+
+	results, err = asyncResult.Get(time.Millisecond * 5)
+	if err != nil {
+		return fmt.Errorf("Getting long running task result failed with error: %s", err.Error())
+	}
+	log.INFO.Printf("Long running task returned = %v\n", tasks.HumanReadableResults(results))
 
 	return nil
 }