(#51) additional packager settings

This supports Retry Policy and End State Discard
settings for packages

Did some refactors and updates to make supporting
that easier and so also made the cli generation a
bit more dynamic.

Logging in generated main.go is more detailed

Signed-off-by: R.I.Pienaar <[email protected]>

Author: ripienaar

ajc/package_command.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 
 	"github.com/choria-io/asyncjobs/generators"
 	"gopkg.in/alecthomas/kingpin.v2"
@@ -58,6 +59,9 @@ func (c *packageCommand) dockerAction(_ *kingpin.ParseContext) error {
 	if h.Name == "" {
 		h.Name = "choria.io/asyncjobs/handlers"
 	}
+	if h.RetryPolicy == "" {
+		h.RetryPolicy = "default"
+	}
 
 	if len(h.TaskHandlers) == 0 {
 		return fmt.Errorf("no task handlers specified in %s", c.file)
@@ -68,6 +72,12 @@ func (c *packageCommand) dockerAction(_ *kingpin.ParseContext) error {
 	table.AddRow("NATS Context Name", h.ContextName)
 	table.AddRow("Work Queue", h.WorkQueue)
 	table.AddRow("Task Handlers", len(h.TaskHandlers))
+	table.AddRow("Retry Backoff Policy", h.RetryPolicy)
+	if len(h.DiscardStates) > 0 {
+		table.AddRow("End State Discard", strings.Join(h.DiscardStates, ", "))
+	} else {
+		table.AddRow("End State Discard", "none")
+	}
 	table.AddRow("github.com/choria-io/asyncjobs", h.AJVersion)
 	fmt.Println(table.Render())
 

ajc/task_command.go

@@ -10,9 +10,10 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"strings"
 	"time"
 
-	"github.com/choria-io/asyncjobs"
+	aj "github.com/choria-io/asyncjobs"
 	"github.com/dustin/go-humanize"
 	"github.com/nats-io/jsm.go"
 	"github.com/xlab/tablewriter"
@@ -79,18 +80,20 @@ func configureTaskCommand(app *kingpin.Application) {
 	watch := tasks.Command("watch", "Watch job updates in real time").Action(c.watchAction)
 	watch.Flag("task", "Watch for updates related to a specific task ID").StringVar(&c.id)
 
+	policies := aj.RetryPolicyNames()
+
 	process := tasks.Command("process", "Process Tasks from a given queue").Action(c.processAction)
 	process.Arg("type", "Types of Tasks to process").Required().Envar("AJC_TYPE").StringVar(&c.ttype)
 	process.Arg("queue", "The Queue to consume Tasks from").Required().Envar("AJC_QUEUE").StringVar(&c.queue)
 	process.Arg("concurrency", "How many concurrent Tasks to process").Required().Envar("AJC_CONCURRENCY").IntVar(&c.concurrency)
 	process.Arg("command", "The command to invoke for each Task").Envar("AJC_COMMAND").ExistingFileVar(&c.command)
 	process.Flag("remote", "Process tasks using a remote request-reply callout").BoolVar(&c.remote)
-	process.Flag("monitor", "Runs monitoring on the given port").IntVar(&c.promPort)
-	process.Flag("backoff", "Selects a backoff policy to apply (1m, 10m, 1h)").EnumVar(&c.retry, "1m", "10m", "1h")
+	process.Flag("monitor", "Runs monitoring on the given port").PlaceHolder("PORT").IntVar(&c.promPort)
+	process.Flag("backoff", fmt.Sprintf("Selects a backoff policy to apply (%s)", strings.Join(policies, ", "))).Default("default").EnumVar(&c.retry, policies...)
 }
 
 func (c *taskCommand) retryAction(_ *kingpin.ParseContext) error {
-	err := prepare(asyncjobs.BindWorkQueue(c.queue))
+	err := prepare(aj.BindWorkQueue(c.queue))
 	if err != nil {
 		return err
 	}
@@ -104,7 +107,7 @@ func (c *taskCommand) retryAction(_ *kingpin.ParseContext) error {
 }
 
 func (c *taskCommand) initAction(_ *kingpin.ParseContext) error {
-	err := prepare(asyncjobs.NoStorageInit())
+	err := prepare(aj.NoStorageInit())
 	if err != nil {
 		return err
 	}
@@ -135,9 +138,9 @@ func (c *taskCommand) watchAction(_ *kingpin.ParseContext) error {
 		return err
 	}
 
-	target := asyncjobs.EventsSubjectWildcard
+	target := aj.EventsSubjectWildcard
 	if c.id != "" {
-		target = fmt.Sprintf(asyncjobs.TaskStateChangeEventSubjectPattern, c.id)
+		target = fmt.Sprintf(aj.TaskStateChangeEventSubjectPattern, c.id)
 	}
 
 	sub, err := mgr.NatsConn().SubscribeSync(target)
@@ -151,13 +154,13 @@ func (c *taskCommand) watchAction(_ *kingpin.ParseContext) error {
 			return err
 		}
 
-		event, kind, err := asyncjobs.ParseEventJSON(msg.Data)
+		event, kind, err := aj.ParseEventJSON(msg.Data)
 		if err != nil {
 			fmt.Printf("Could not parse event: %v\n", err)
 		}
 
 		switch e := event.(type) {
-		case asyncjobs.TaskStateChangeEvent:
+		case aj.TaskStateChangeEvent:
 			if e.LastErr == "" {
 				fmt.Printf("[%s] %s: queue: %s type: %s tries: %d state: %s\n", e.TimeStamp.Format("15:04:05"), e.TaskID, e.Queue, e.TaskType, e.Tries, e.State)
 			} else {
@@ -170,7 +173,7 @@ func (c *taskCommand) watchAction(_ *kingpin.ParseContext) error {
 	}
 }
 
-func (c *taskCommand) commandHandlerFunc(ctx context.Context, log asyncjobs.Logger, task *asyncjobs.Task) (interface{}, error) {
+func (c *taskCommand) commandHandlerFunc(ctx context.Context, log aj.Logger, task *aj.Task) (interface{}, error) {
 	tj, err := json.Marshal(task)
 	if err != nil {
 		return nil, err
@@ -210,25 +213,16 @@ func (c *taskCommand) processAction(_ *kingpin.ParseContext) error {
 		return fmt.Errorf("either a command or --remote is required")
 	}
 
-	retryPolicy := asyncjobs.RetryDefault
-	switch c.retry {
-	case "1m":
-		retryPolicy = asyncjobs.RetryLinearOneMinute
-	case "10m":
-		retryPolicy = asyncjobs.RetryLinearTenMinutes
-	case "1h":
-		retryPolicy = asyncjobs.RetryLinearOneHour
-	}
-
 	err := prepare(
-		asyncjobs.BindWorkQueue(c.queue),
-		asyncjobs.PrometheusListenPort(c.promPort),
-		asyncjobs.RetryBackoffPolicy(retryPolicy))
+		aj.BindWorkQueue(c.queue),
+		aj.PrometheusListenPort(c.promPort),
+		aj.RetryBackoffPolicyName(c.retry),
+		aj.ClientConcurrency(c.concurrency))
 	if err != nil {
 		return err
 	}
 
-	router := asyncjobs.NewTaskRouter()
+	router := aj.NewTaskRouter()
 	if c.remote {
 		err = router.RequestReply(c.ttype, client)
 	} else {
@@ -401,17 +395,17 @@ func (c *taskCommand) viewAction(_ *kingpin.ParseContext) error {
 }
 
 func (c *taskCommand) addAction(_ *kingpin.ParseContext) error {
-	err := prepare(asyncjobs.BindWorkQueue(c.queue))
+	err := prepare(aj.BindWorkQueue(c.queue))
 	if err != nil {
 		return err
 	}
 
-	var opts []asyncjobs.TaskOpt
+	var opts []aj.TaskOpt
 	if c.deadline > 0 {
-		opts = append(opts, asyncjobs.TaskDeadline(time.Now().UTC().Add(c.deadline)))
+		opts = append(opts, aj.TaskDeadline(time.Now().UTC().Add(c.deadline)))
 	}
 
-	task, err := asyncjobs.NewTask(c.ttype, c.payload, opts...)
+	task, err := aj.NewTask(c.ttype, c.payload, opts...)
 	if err != nil {
 		return err
 	}

client_options.go

@@ -138,6 +138,18 @@ func RetryBackoffPolicy(p RetryPolicyProvider) ClientOpt {
 	}
 }
 
+// RetryBackoffPolicyName uses the policy named to schedule job retries by using RetryPolicyLookup(name)
+func RetryBackoffPolicyName(name string) ClientOpt {
+	return func(opts *ClientOpts) error {
+		p, err := RetryPolicyLookup(name)
+		if err != nil {
+			return err
+		}
+
+		return RetryBackoffPolicy(p)(opts)
+	}
+}
+
 // ClientConcurrency sets the concurrency to use when executing tasks within this client for horizontal scaling.
 // This is capped by the per-queue maximum concurrency set using the queue setting MaxConcurrent. Generally a
 // queue would have a larger concurrency like 100 (DefaultQueueMaxConcurrent) and an individual task processor
@@ -195,7 +207,7 @@ func BindWorkQueue(queue string) ClientOpt {
 	}
 }
 
-// TaskRetention is the time tasks will be kept with.
+// TaskRetention is the time tasks will be kept for in the task storage
 //
 // Used only when initially creating the underlying streams.
 func TaskRetention(r time.Duration) ClientOpt {

errors.go

@@ -63,6 +63,9 @@ var (
 	// ErrUnknownEventType indicates that while parsing an event an unknown type of event was encountered
 	ErrUnknownEventType = fmt.Errorf("unknown event type")
 
+	// ErrUnknownRetryPolicy indicates the requested retry policy does not exist
+	ErrUnknownRetryPolicy = fmt.Errorf("unknown retry policy")
+
 	// ErrRequestReplyFailed indicates a callout to a remote handler failed due to a timeout, lack of listerners or network error
 	ErrRequestReplyFailed = fmt.Errorf("request-reply callout failed")
 	// ErrRequestReplyNoDeadline indicates a request-reply handler was called without a deadline

generators/fs/godocker/Dockerfile.templ

@@ -4,7 +4,9 @@ WORKDIR /usr/src/app
 
 RUN go mod init "{{ .Package.Name }}" && \
 {{- range $handler := .Package.TaskHandlers }}
+  {{- if not $handler.RequestReply }}
     go get "{{ $handler.Package }}@{{ $handler.Version }}" && \
+  {{- end }}
 {{- end }}
     go get github.com/choria-io/asyncjobs@{{ .Package.AJVersion }}
 

generators/fs/godocker/main.go.templ

@@ -26,9 +26,9 @@ var usage = `This is a generated Handler service for the Choria Async Jobs Proje
 It hosts the following handlers:
 {{ range $handler := .Package.TaskHandlers }}
   {{- if $handler.RequestReply }}
- - {{ $handler.TaskType }}: {{ $handler.Package }}@{{ $handler.Version }}
-  {{- else }}
  - {{ $handler.TaskType}}: Remote Request-Reply Service
+  {{- else }}
+ - {{ $handler.TaskType }}: {{ $handler.Package }}@{{ $handler.Version }}
   {{- end }}
 {{- end }}
 
@@ -37,6 +37,7 @@ The following Environment variables are supported:
  - AJ_WORK_QUEUE: The Work Queue to consume from, defaults to DEFAULT
  - AJ_NATS_CONTEXT: The name of a NATS Context to use for connections
  - AJ_CONCURRENCY: The number of concurrent handlers that can be run
+ - AJ_RETRY_POLICY: The retry policy to use [{{ RetryNamesList }}]
 
 Prometheus statistics are Exposed on port http://0.0.0.0:8080/metrics
 
@@ -95,15 +96,33 @@ func main() {
 		TimestampFormat: "15:04:05",
 	})
 
+	retryPolicy := os.Getenv("AJ_RETRY_POLICY")
+	if retryPolicy == "" {
+		retryPolicy = "{{ .Package.RetryPolicy }}"
+	}
+
 	log := logrus.NewEntry(logger)
 
-	log.Printf("Connecting using Context %s consuming work queue %s with concurrency %d", nctx, wq, concurrency)
+	log.Printf("Choria Async Jobs Handler Service {{.Package.Name}} build settings")
+	log.Printf("NATS Context: %s", nctx)
+	log.Printf("Work Queue: %s", wq)
+	log.Printf("Concurrency: %d", concurrency)
+	log.Printf("Retry Policy: %s", retryPolicy)
+{{- range $state := .Package.DiscardStates }}
+	log.Printf("Discard State: {{$state}}")
+{{- end }}
+	log.Printf("Prometheus Port: 8080")
+	log = log.WithField("queue", wq)
 
 	client, err := aj.NewClient(
 		aj.NatsContext(nctx),
 		aj.BindWorkQueue(wq),
 		aj.ClientConcurrency(concurrency),
 		aj.CustomLogger(log),
+		aj.RetryBackoffPolicyName(retryPolicy),
+{{- range $state := .Package.DiscardStates }}
+		aj.DiscardTaskStates("{{$state}}"),
+{{- end }}
 		aj.PrometheusListenPort(8080))
 	usageIfError(err)
 

generators/godocker.go

@@ -13,6 +13,8 @@ import (
 	"strings"
 	"text/template"
 	"time"
+
+	aj "github.com/choria-io/asyncjobs"
 )
 
 // GoContainer builds docker containers based on the package spec
@@ -43,6 +45,10 @@ func (g *GoContainer) RenderToDirectory(target string) error {
 		return err
 	}
 
+	if g.Package.RetryPolicy == "" {
+		g.Package.RetryPolicy = "default"
+	}
+
 	for _, p := range g.Package.TaskHandlers {
 		if p.RequestReply {
 			continue
@@ -59,6 +65,9 @@ func (g *GoContainer) RenderToDirectory(target string) error {
 	}
 
 	funcs := map[string]interface{}{
+		"RetryNamesList": func() string {
+			return strings.Join(aj.RetryPolicyNames(), ", ")
+		},
 		"TypeToPackageName": func(t string) string {
 			remove := []string{"_", "-", ":", "/", "\\"}
 			res := t

generators/package.go

@@ -22,6 +22,10 @@ type Package struct {
 	Name string `yaml:"name"`
 	// AJVersion is an optional version to use for the choria-io/asyncjobs dependency
 	AJVersion string `yaml:"asyncjobs"`
+	// RetryPolicy is the name of a retry policy, see RetryPolicyNames()
+	RetryPolicy string `yaml:"retry"`
+	// DiscardStates indicates what termination states to discard
+	DiscardStates []string `yaml:"discard"`
 }
 
 // TaskHandler is an individual Task Handler

processor_test.go

@@ -321,6 +321,7 @@ var _ = Describe("Processor", func() {
 
 				router := NewTaskRouter()
 				router.HandleFunc("ginkgo", func(ctx context.Context, log Logger, task *Task) (interface{}, error) {
+					// these will panic as its in a different routine, but they are supposed to pass so thats fine
 					t, err := client.LoadTaskByID(task.ID)
 					Expect(err).ToNot(HaveOccurred())
 					Expect(t.State).To(Equal(TaskStateActive))

retrypolicy.go

@@ -6,7 +6,9 @@ package asyncjobs
 
 import (
 	"context"
+	"fmt"
 	"math/rand"
+	"sort"
 	"time"
 )
 
@@ -40,8 +42,48 @@ var (
 
 	retryLinearTenSeconds = linearPolicy(20, 0.1, 500*time.Millisecond, 10*time.Second)
 	retryForTesting       = linearPolicy(1, 0.1, time.Millisecond, 10*time.Millisecond)
+
+	policies = map[string]RetryPolicyProvider{
+		"default": RetryDefault,
+		"1m":      RetryLinearOneMinute,
+		"10m":     RetryLinearTenMinutes,
+		"1h":      RetryLinearOneHour,
+	}
 )
 
+// RetryPolicyNames returns a list of pre-generated retry policies
+func RetryPolicyNames() []string {
+	var names []string
+	for k := range policies {
+		names = append(names, k)
+	}
+
+	sort.Strings(names)
+
+	return names
+}
+
+// RetryPolicyLookup loads a policy by name
+func RetryPolicyLookup(name string) (RetryPolicyProvider, error) {
+	policy, ok := policies[name]
+	if !ok {
+		return nil, fmt.Errorf("%w: %s", ErrUnknownRetryPolicy, name)
+	}
+
+	return policy, nil
+}
+
+// IsRetryPolicyKnown determines if the named policy exist
+func IsRetryPolicyKnown(name string) bool {
+	for _, p := range RetryPolicyNames() {
+		if p == name {
+			return true
+		}
+	}
+
+	return false
+}
+
 // Duration is the period to sleep for try n, it includes a jitter
 func (p RetryPolicy) Duration(n int) time.Duration {
 	if n >= len(p.Intervals) {