Merge ~troyanov/maas:refactor-workerpool into maas:master

[Anton Troyanov]

refactor: opts pattern to override pool defaults

[MAAS Lander (maas-lander)]

UNIT TESTS
-b refactor-workerpool lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: b283dfb3bfc9758842e6c65026a9b7588a675ec3

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b refactor-workerpool lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: dd7bb0c5a0d8f87e5189156bacfc43fceea276e1

review: Approve

08f30b5... by Anton Troyanov on 2023-08-14

refactor: remove repetitive things

[MAAS Lander (maas-lander)]

UNIT TESTS
-b refactor-workerpool lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: 53f6871aa41b82cd5c16905c728235b924c14f9d

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b refactor-workerpool lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: 51fbefb67768125ac5b6015c9cc3fe7ea3a1addb

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b refactor-workerpool lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/3362/console
COMMIT: 54136ffb92b13109c6b23a143efbd63374ce2aff

review: Needs Fixing

a8b0c1d... by Anton Troyanov on 2023-08-24

refactor: opts pattern to override pool defaults

1f0c468... by Anton Troyanov on 2023-08-24

refactor(temporal): explicit handle client fatal

[Christian Grabowski (cgrabowski)]

One minor nit inline, otherwise, lgtm.

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b refactor-workerpool lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: fef5fdc3e23c06348c292758b105e3d9f07c348c

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b refactor-workerpool lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/3364/console
COMMIT: b6f0afb70385efd11ab45aa1730546b959f00029

review: Needs Fixing

2f3ca5a... by Anton Troyanov on 2023-08-24

chore: add comments

[MAAS Lander (maas-lander)]

UNIT TESTS
-b refactor-workerpool lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/3365/console
COMMIT: 2492ed624f3cb16f825c10f4ee87f01e53aff38b

review: Needs Fixing

e6579d6... by Anton Troyanov on 2023-08-24

fixup! refactor(temporal): explicit handle client fatal

[MAAS Lander (maas-lander)]

UNIT TESTS
-b refactor-workerpool lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: e6579d65203fd664e6f2f0a49f1f9823d548df99

review: Approve

diff --git a/src/maasagent/cmd/maas-agent/main.go b/src/maasagent/cmd/maas-agent/main.go
index 6336006..545272f 100644
--- a/src/maasagent/cmd/maas-agent/main.go
+++ b/src/maasagent/cmd/maas-agent/main.go
@@ -19,6 +19,7 @@ import (
 	"go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/converter"
 	"gopkg.in/yaml.v3"
+	wf "maas.io/core/src/maasagent/internal/workflow"
 	wflog "maas.io/core/src/maasagent/internal/workflow/log"
 	"maas.io/core/src/maasagent/internal/workflow/worker"
 	"maas.io/core/src/maasagent/pkg/workflow/codec"
@@ -84,17 +85,28 @@ func Run() int {
 		return 1
 	}
 
+	workerPool := worker.NewWorkerPool(cfg.SystemID, client,
+		worker.WithAllowedWorkflows(map[string]interface{}{
+			"check_ip":              wf.CheckIP,
+			"commission":            wf.Commission,
+			"deploy":                wf.Deploy,
+			"deployed_os_workflow":  wf.DeployedOS,
+			"ephemeral_os_workflow": wf.EphemeralOS,
+			"power_on":              wf.PowerOn,
+			"power_off":             wf.PowerOff,
+			"power_query":           wf.PowerQuery,
+			"power_cycle":           wf.PowerCycle,
+		}), worker.WithAllowedActivities(map[string]interface{}{
+			"switch_boot_order": wf.SwitchBootOrderActivity,
+			"power":             wf.PowerActivity,
+		}))
+
 	workerPoolBackoff := backoff.NewExponentialBackOff()
 	workerPoolBackoff.MaxElapsedTime = 60 * time.Second
 
-	_, err = backoff.RetryWithData(
-		func() (*worker.WorkerPool, error) {
-			return worker.NewWorkerPool(cfg.SystemID, client)
-		}, workerPoolBackoff,
-	)
-
+	err = backoff.Retry(workerPool.Start, workerPoolBackoff)
 	if err != nil {
-		log.Error().Err(err).Msg("Temporal worker pool creation failure")
+		log.Error().Err(err).Msg("Temporal worker pool failure")
 		return 1
 	}
 
@@ -104,9 +116,13 @@ func Run() int {
 
 	signal.Notify(sigC, syscall.SIGTERM, syscall.SIGINT)
 
-	<-sigC
-
-	return 0
+	select {
+	case err := <-workerPool.Error():
+		log.Fatal().Err(err).Msg("Temporal worker pool failure")
+		return 1
+	case <-sigC:
+		return 0
+	}
 }
 
 // getConfig reads MAAS Agent YAML configuration file
diff --git a/src/maasagent/internal/workflow/worker/pool.go b/src/maasagent/internal/workflow/worker/pool.go
index a342e26..05f91ef 100644
--- a/src/maasagent/internal/workflow/worker/pool.go
+++ b/src/maasagent/internal/workflow/worker/pool.go
@@ -8,71 +8,134 @@ import (
 
 	"go.temporal.io/sdk/activity"
 	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/worker"
 	"go.temporal.io/sdk/workflow"
+)
 
-	wf "maas.io/core/src/maasagent/internal/workflow"
+const (
+	defaultAddWorkerWorkflowName     = "add_worker"
+	defaultRemoveWorkerWorkflowName  = "remove_worker"
+	defaultConfigurePoolWorkflowName = "configure_worker_pool"
+	defaultControlPlaneTaskQueueName = "control_plane"
 )
 
 // WorkerPool contains a collection of Temporal Workers that can be added or
 // removed during runtime by master worker which is responsible for execution of
 // special workflows `add_worker` and `remove_worker`.
 type WorkerPool struct {
+	fatal  chan error
 	client client.Client
 	// worker for control plane workflows like Add or Remove workers
-	master worker.Worker
-	// collection of workflows allowed for registration
-	workflows map[string]interface{}
-	// collection of activities allowed for registration
-	activities map[string]interface{}
-	workers    map[string]worker.Worker
-	systemID   string
-	mutex      sync.Mutex
+	master                    worker.Worker
+	workers                   map[string]worker.Worker
+	allowedWorkflows          map[string]interface{}
+	allowedActivities         map[string]interface{}
+	systemID                  string
+	addWorkerWorkflowName     string
+	removeWorkerWorkflowName  string
+	configurePoolWorkflowName string
+	controlPlaneTaskQueueName string
+	mutex                     sync.Mutex
 }
 
 // NewWorkerPool returns WorkerPool that has a master worker listening to a
 // Temporal Task Queue named after systemID
-func NewWorkerPool(systemID string, client client.Client) (*WorkerPool, error) {
+func NewWorkerPool(systemID string, client client.Client,
+	options ...WorkerPoolOption) *WorkerPool {
 	pool := &WorkerPool{
-		systemID: systemID,
-		client:   client,
-		workers:  make(map[string]worker.Worker),
-		workflows: map[string]interface{}{
-			"check_ip":              wf.CheckIP,
-			"commission":            wf.Commission,
-			"deploy":                wf.Deploy,
-			"deployed_os_workflow":  wf.DeployedOS,
-			"ephemeral_os_workflow": wf.EphemeralOS,
-			"power_on":              wf.PowerOn,
-			"power_off":             wf.PowerOff,
-			"power_query":           wf.PowerQuery,
-			"power_cycle":           wf.PowerCycle,
-		},
-		activities: map[string]interface{}{
-			"switch_boot_order": wf.SwitchBootOrderActivity,
-			"power":             wf.PowerActivity,
-		},
+		systemID:                  systemID,
+		client:                    client,
+		workers:                   make(map[string]worker.Worker),
+		addWorkerWorkflowName:     defaultAddWorkerWorkflowName,
+		removeWorkerWorkflowName:  defaultRemoveWorkerWorkflowName,
+		configurePoolWorkflowName: defaultConfigurePoolWorkflowName,
+		controlPlaneTaskQueueName: defaultControlPlaneTaskQueueName,
+	}
+
+	for _, opt := range options {
+		opt(pool)
 	}
 
 	// master worker is responsible for adding/removing workers to/from the pool
-	pool.master = worker.New(client, systemID, worker.Options{})
+	pool.master = worker.New(client, systemID, worker.Options{
+		Identity:     fmt.Sprintf("%s:master", systemID),
+		OnFatalError: func(err error) { pool.fatal <- err },
+	})
 
-	var opts workflow.RegisterOptions
-	opts = workflow.RegisterOptions{
-		Name: "add_worker",
-	}
 	pool.master.RegisterWorkflowWithOptions(
-		localActivityExec[addWorkerParam](pool.addWorker), opts,
+		localActivityExec[addWorkerParam](pool.addWorker),
+		workflow.RegisterOptions{
+			Name: pool.addWorkerWorkflowName,
+		},
 	)
 
-	opts = workflow.RegisterOptions{
-		Name: "remove_worker",
-	}
 	pool.master.RegisterWorkflowWithOptions(
-		localActivityExec[removeWorkerParam](pool.removeWorker), opts,
+		localActivityExec[removeWorkerParam](pool.removeWorker),
+		workflow.RegisterOptions{
+			Name: pool.removeWorkerWorkflowName,
+		},
 	)
 
-	return pool, pool.master.Start()
+	return pool
+}
+
+// Start starts the master worker process that controls worker pool
+func (p *WorkerPool) Start() error {
+	return p.master.Start()
+}
+
+func (p *WorkerPool) Error() chan error {
+	return p.fatal
+}
+
+// WorkerPoolOption allows to set additional WorkerPool options
+type WorkerPoolOption func(*WorkerPool)
+
+// WithAddWorkerWorkflowName sets custom addWorkerWorkflowName
+// (default: "add_worker")
+func WithAddWorkerWorkflowName(s string) WorkerPoolOption {
+	return func(p *WorkerPool) {
+		p.addWorkerWorkflowName = s
+	}
+}
+
+// WithRemoveWorkerWorkflowName sets custom removeWorkerWorkflowName
+// (default: "remove_worker")
+func WithRemoveWorkerWorkflowName(s string) WorkerPoolOption {
+	return func(p *WorkerPool) {
+		p.removeWorkerWorkflowName = s
+	}
+}
+
+// WithConfigurePoolWorkflowName sets custom configurePoolWorkflowName
+// (default: "configure_worker_pool")
+func WithConfigurePoolWorkflowName(s string) WorkerPoolOption {
+	return func(p *WorkerPool) {
+		p.configurePoolWorkflowName = s
+	}
+}
+
+// WithControlPlaneTaskQueueName sets custom controlPlaneTaskQueueName
+// (default: "control_plane")
+func WithControlPlaneTaskQueueName(s string) WorkerPoolOption {
+	return func(p *WorkerPool) {
+		p.controlPlaneTaskQueueName = s
+	}
+}
+
+// WithAllowedWorkflows sets workflows allowed to be registered
+func WithAllowedWorkflows(workflows map[string]interface{}) WorkerPoolOption {
+	return func(p *WorkerPool) {
+		p.allowedWorkflows = workflows
+	}
+}
+
+// WithAllowedActivities sets activities allowed to be registered
+func WithAllowedActivities(activities map[string]interface{}) WorkerPoolOption {
+	return func(p *WorkerPool) {
+		p.allowedActivities = activities
+	}
 }
 
 // configureWorkerPoolParam is a parameter that should be provided to the
@@ -109,25 +172,32 @@ func (p *WorkerPool) addWorker(param addWorkerParam) error {
 	defer p.mutex.Unlock()
 
 	if _, ok := p.workers[param.TaskQueue]; ok {
-		return fmt.Errorf("worker for TaskQueue %s is already registered in the pool", param.TaskQueue)
+		return failedToAddWorkerError(param.TaskQueue)
 	}
 
-	w := worker.New(p.client, param.TaskQueue, worker.Options{})
+	w := worker.New(p.client, param.TaskQueue, worker.Options{
+		Identity:     fmt.Sprintf("%s:%s", p.systemID, param.TaskQueue),
+		OnFatalError: func(err error) { p.fatal <- err },
+	})
 
-	for _, wf := range param.Workflows {
-		if fn, ok := p.workflows[wf]; ok {
-			w.RegisterWorkflowWithOptions(fn, workflow.RegisterOptions{Name: wf})
-		}
+	if err := register("workflow", param.Workflows, p.allowedWorkflows,
+		func(name string, fn interface{}) {
+			w.RegisterWorkflowWithOptions(fn, workflow.RegisterOptions{Name: name})
+		}); err != nil {
+		return err
 	}
 
-	for _, act := range param.Activities {
-		if fn, ok := p.activities[act]; ok {
-			w.RegisterActivityWithOptions(fn, activity.RegisterOptions{Name: act})
-		}
+	if err := register("activity", param.Activities, p.allowedActivities,
+		func(name string, fn interface{}) {
+			w.RegisterActivityWithOptions(fn, activity.RegisterOptions{Name: name})
+		}); err != nil {
+		return err
 	}
 
 	if err := w.Start(); err != nil {
-		return err
+		w = nil
+
+		return failedToStartWorkerError(err)
 	}
 
 	p.workers[param.TaskQueue] = w
@@ -135,6 +205,24 @@ func (p *WorkerPool) addWorker(param addWorkerParam) error {
 	return nil
 }
 
+func register(t string, s []string, allowed map[string]interface{},
+	reg func(string, interface{})) error {
+	for _, val := range s {
+		fn, ok := allowed[val]
+		if ok {
+			reg(val, fn)
+			continue
+		}
+
+		return temporal.NewNonRetryableApplicationError(
+			fmt.Sprintf("Failed registering %s", t),
+			fmt.Sprintf("%sNotAllowed", t),
+			fmt.Errorf("%s %q is not allowed", t, val))
+	}
+
+	return nil
+}
+
 type removeWorkerParam struct {
 	TaskQueue string `json:"task_queue"`
 }
@@ -145,9 +233,8 @@ func (p *WorkerPool) removeWorker(param removeWorkerParam) error {
 	defer p.mutex.Unlock()
 
 	w, ok := p.workers[param.TaskQueue]
-
 	if !ok {
-		return nil
+		return failedToRemoveWorkerError(param.TaskQueue)
 	}
 
 	w.Stop()
@@ -167,3 +254,23 @@ func localActivityExec[T any](fn any) func(ctx workflow.Context, param T) error 
 		return workflow.ExecuteLocalActivity(ctx, fn, param).Get(ctx, nil)
 	}
 }
+
+// failedToAddWorkerError returns a non retryable error
+func failedToAddWorkerError(taskQueue string) error {
+	return temporal.NewNonRetryableApplicationError("Failed adding worker",
+		"failedToAddWorker",
+		fmt.Errorf("worker for task queue %q already exists", taskQueue))
+}
+
+// failedToRemoveWorkerError returns a non retryable error
+func failedToRemoveWorkerError(taskQueue string) error {
+	return temporal.NewNonRetryableApplicationError("Failed removing worker",
+		"failedToRemoveWorker",
+		fmt.Errorf("worker for task queue %q doesn't exist", taskQueue))
+}
+
+// failedToStartWorkerError returns a non retryable error
+func failedToStartWorkerError(err error) error {
+	return temporal.NewNonRetryableApplicationError("Failed to start worker",
+		"failedToStartWorker", err)
+}