Merge ~troyanov/maas:temporal-go-worker into maas:master

[Anton Troyanov]

feat(temporal): maas-agent with temporal worker

Add worker pool with control plane for dynamic registration of Temporal workers.
Add adapter for zerolog that allows usage of custom logger for Temporal client.
Add snap and deb packaging for maas-agent

Co-authored-by: Christian Grabowski <email address hidden>

Here are some examples:
1. I want to execute Workflow `CheckIP` for Task Queue `vlan-1`

`tctl --ad 10.0.0.62:5271 workflow run --taskqueue vlan-1 --workflow_type CheckIP --input '{"IPs":["172.16.1.1", "172.16.1.11"]}'`

Since I don't have any workers registered for that type of a Workflow and that Task Queue, nothing will happen.

2. I am triggering a *control plane* Workflow `AddWorker` on a Task Queue `amazed-asp` and pass parameter saying: I want to register new worker on a Task Queue `vlan-1` and ask that worker to register Workflow `CheckIP`

`tctl --ad 10.0.0.62:5271 workflow run --taskqueue amazed-asp --workflow_type AddWorker --input '{"TaskQueue":"vlan-1", "Workflows":["CheckIP"]}'`

3. Running command from 1 again will give me result:
```
Result:
  Run Time: 1 seconds
  Status: COMPLETED
  Output: [{"IPs":{"172.16.1.1":"ABY+SvJA","172.16.1.11":"ABY+SvJA"}}]
```

4. If I am not interested in having such Worker on certain host anymore, I simply execute `RemoveWorker` workflow with a Task Queue `amazed-asp` and that will remove the Worker completely. Note, you cannot unregister specific workflow from the worker.

`tctl --ad 10.0.0.62:5271 workflow run --taskqueue amazed-asp --workflow_type RemoveWorker --input '{"TaskQueue":"vlan-1"}'`

[MAAS Lander (maas-lander)]

UNIT TESTS
-b temporal-go-worker lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: 415a890a13ff72513c8a2c1a128fa1edf79064d7

review: Approve

[Christian Grabowski (cgrabowski)]

+1

review: Approve

[MAAS Lander (maas-lander)]

LANDING
-b temporal-go-worker lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci.internal:8080/job/maas-tester/3170/console

[MAAS Lander (maas-lander)]

UNIT TESTS
-b temporal-go-worker lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/3171/console
COMMIT: 7358dffe3344b08759e3299254e5858cfdc644a8

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b temporal-go-worker lp:~troyanov/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: c6b268dcfacd971773921250be8b0f0da14c7f44

review: Approve

diff --git a/debian/extras/99-maas-agent-sudoers b/debian/extras/99-maas-agent-sudoers
new file mode 100644
index 0000000..a8e915a
--- /dev/null
+++ b/debian/extras/99-maas-agent-sudoers
@@ -0,0 +1 @@
+maas ALL= NOPASSWD: /usr/sbin/maas-agent
diff --git a/debian/maas-agent.install b/debian/maas-agent.install
new file mode 100644
index 0000000..71c9e73
--- /dev/null
+++ b/debian/maas-agent.install
@@ -0,0 +1,3 @@
+debian/extras/99-maas-agent-sudoers etc/sudoers.d
+
+bin/maas-agent usr/sbin
diff --git a/debian/maas-agent.maas-agent.service b/debian/maas-agent.maas-agent.service
new file mode 100644
index 0000000..e16fd8a
--- /dev/null
+++ b/debian/maas-agent.maas-agent.service
@@ -0,0 +1,14 @@
+[Unit]
+Description=The MAAS Agent daemon
+Documentation=https://maas.io/docs
+Wants=network-online.target
+After=network-online.target
+BindsTo=maas-rackd.service
+
+[Service]
+User=maas
+Group=maas
+ExecStart=/usr/sbin/maas-agent
+
+[Install]
+WantedBy=multi-user.target
diff --git a/debian/not-installed b/debian/not-installed
index a136782..4bee65a 100644
--- a/debian/not-installed
+++ b/debian/not-installed
@@ -2,6 +2,3 @@
 /usr/bin/test.*
 /usr/bin/maas-sampledata
 
-# Temporary exclude maas-agent
-/bin/maas-agent
-
diff --git a/debian/rules b/debian/rules
index ad435b8..61def6d 100755
--- a/debian/rules
+++ b/debian/rules
@@ -20,6 +20,7 @@ override_dh_installsystemd:
 	dh_installsystemd -p maas-proxy --name=maas-proxy maas-proxy.service
 	dh_installsystemd -p maas-dhcp --name=maas-dhcpd maas-dhcpd.service
 	dh_installsystemd -p maas-dhcp --name=maas-dhcpd6 maas-dhcpd6.service
+	dh_installsystemd -p maas-agent --name=maas-agent maas-agent.service
 
 override_dh_auto_install:
 	dh_auto_install
diff --git a/snap/local/tree/bin/run-maas-agent b/snap/local/tree/bin/run-maas-agent
new file mode 100755
index 0000000..8c37a57
--- /dev/null
+++ b/snap/local/tree/bin/run-maas-agent
@@ -0,0 +1,6 @@
+#!/bin/bash
+
+# Copyright 2023 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+exec "$SNAP/usr/sbin/maas-agent"
diff --git a/snap/local/tree/usr/share/maas/pebble/layers/003-maas-rack-layer.yaml b/snap/local/tree/usr/share/maas/pebble/layers/003-maas-rack-layer.yaml
index ec97214..82ce9a0 100644
--- a/snap/local/tree/usr/share/maas/pebble/layers/003-maas-rack-layer.yaml
+++ b/snap/local/tree/usr/share/maas/pebble/layers/003-maas-rack-layer.yaml
@@ -28,3 +28,8 @@ services:
     override: replace
     startup: disabled
     command: sh -c "exec $SNAP/bin/run-dhcpd6"
+
+  agent:
+    override: replace
+    startup: disabled
+    command: sh -c "exec $SNAP/bin/run-maas-agent"
diff --git a/snap/snapcraft.yaml b/snap/snapcraft.yaml
index b2209a0..04e63b1 100644
--- a/snap/snapcraft.yaml
+++ b/snap/snapcraft.yaml
@@ -268,6 +268,16 @@ parts:
     prime:
       - usr/sbin/maas-netmon
 
+  maas-agent:
+    plugin: make
+    source: src/maasagent
+    build-packages:
+      - golang-go
+    organize:
+      bin/maas-agent: usr/sbin/maas-agent
+    prime:
+      - usr/sbin/maas-agent
+
   tree:
     plugin: dump
     source: snap/local/tree
diff --git a/src/maasagent/cmd/maas-agent/main.go b/src/maasagent/cmd/maas-agent/main.go
index 7905807..3b4623c 100644
--- a/src/maasagent/cmd/maas-agent/main.go
+++ b/src/maasagent/cmd/maas-agent/main.go
@@ -1,5 +1,69 @@
 package main
 
-func main() {
+/*
+	Copyright 2023 Canonical Ltd.  This software is licensed under the
+	GNU Affero General Public License version 3 (see the file LICENSE).
+*/
+
+import (
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+	"go.temporal.io/sdk/client"
+	"launchpad.net/maas/maas/src/maasagent/internal/workflow"
+	wflog "launchpad.net/maas/maas/src/maasagent/internal/workflow/log"
+)
+
+func Run() int {
+	zerolog.SetGlobalLevel(zerolog.InfoLevel)
+
+	log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr})
+
+	if envLogLevel, ok := os.LookupEnv("LOG_LEVEL"); ok {
+		if logLevel, err := zerolog.ParseLevel(envLogLevel); err != nil {
+			log.Warn().Str("LOG_LEVEL", envLogLevel).Msg("Unknown log level, defaulting to INFO")
+		} else {
+			zerolog.SetGlobalLevel(logLevel)
+		}
+	}
+
+	// TODO: add contextual fields to the global logger?
+	log.Info().Msg("Starting MAAS Agent service")
+
+	client, err := client.Dial(client.Options{
+		// TODO: fetch from rack config?
+		HostPort: "localhost:5271",
+		Logger:   wflog.New(log.Logger),
+	})
+
+	if err != nil {
+		log.Error().Err(err).Send()
+		return 1
+	}
 
+	// TODO: init Data Encoder
+	// TODO: read real systemID
+	_, err = workflow.NewWorkerPool("systemID", client)
+
+	if err != nil {
+		log.Error().Err(err).Send()
+		return 1
+	}
+
+	log.Info().Msg("Service MAAS Agent started")
+
+	sigC := make(chan os.Signal, 2)
+
+	signal.Notify(sigC, syscall.SIGTERM, syscall.SIGINT)
+
+	<-sigC
+
+	return 0
+}
+
+func main() {
+	os.Exit(Run())
 }
diff --git a/src/maasagent/internal/workflow/log/adapter.go b/src/maasagent/internal/workflow/log/adapter.go
new file mode 100644
index 0000000..0b16241
--- /dev/null
+++ b/src/maasagent/internal/workflow/log/adapter.go
@@ -0,0 +1,45 @@
+package log
+
+import (
+	"github.com/rs/zerolog"
+)
+
+// Logger is an adapter that allows usage of zerolog with Temporal Client
+type Logger struct {
+	logger zerolog.Logger
+}
+
+// New returns new workflow Logger
+func New(logger zerolog.Logger) *Logger {
+	return &Logger{
+		logger: logger,
+	}
+}
+
+// Debug implements Temporal log.Logger interface
+func (l *Logger) Debug(msg string, keyvals ...interface{}) {
+	sendEvent(l.logger.Debug(), msg, keyvals)
+}
+
+// Info implements Temporal log.Logger interface
+func (l *Logger) Info(msg string, keyvals ...interface{}) {
+	sendEvent(l.logger.Info(), msg, keyvals)
+}
+
+// Warn implements Temporal log.Logger interface
+func (l *Logger) Warn(msg string, keyvals ...interface{}) {
+	sendEvent(l.logger.Warn(), msg, keyvals)
+}
+
+// Error implements Temporal log.Logger interface
+func (l *Logger) Error(msg string, keyvals ...interface{}) {
+	sendEvent(l.logger.Error(), msg, keyvals)
+}
+
+func sendEvent(event *zerolog.Event, msg string, keyvals ...interface{}) {
+	if len(keyvals) > 0 {
+		event.Fields(keyvals[0])
+	}
+
+	event.Msg(msg)
+}
diff --git a/src/maasagent/internal/workflow/pool.go b/src/maasagent/internal/workflow/pool.go
new file mode 100644
index 0000000..bbc6faa
--- /dev/null
+++ b/src/maasagent/internal/workflow/pool.go
@@ -0,0 +1,153 @@
+package workflow
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+	"go.temporal.io/sdk/workflow"
+)
+
+// WorkerPool contains a collection of Temporal Workers that can be added or
+// removed in runtime, through master worker that is responsible for execution of
+// special workflows AddWorker and RemoveWorker.
+// WorkerPool allows to register specific Workflows and Activities for the
+// added workers.
+type WorkerPool struct {
+	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
+}
+
+// NewWorkerPool returns WorkerPool that has a master worker listening to a
+// Temporal Task Queue {systemID}
+func NewWorkerPool(systemID string, client client.Client) (*WorkerPool, error) {
+	pool := &WorkerPool{
+		systemID: systemID,
+		client:   client,
+		workers:  make(map[string]worker.Worker),
+		workflows: map[string]interface{}{
+			"CheckIP": CheckIP,
+		},
+		activities: map[string]interface{}{},
+	}
+
+	// master worker responsible for adding/removing workers from the pool
+	pool.master = worker.New(client, systemID, worker.Options{})
+
+	var opts workflow.RegisterOptions
+	opts = workflow.RegisterOptions{
+		Name: "AddWorker",
+	}
+	pool.master.RegisterWorkflowWithOptions(
+		exec[addWorkerParam](pool.addWorker), opts,
+	)
+
+	opts = workflow.RegisterOptions{
+		Name: "RemoveWorker",
+	}
+	pool.master.RegisterWorkflowWithOptions(
+		exec[removeWorkerParam](pool.removeWorker), opts,
+	)
+
+	return pool, pool.master.Start()
+}
+
+// configureParam is a parameter that should be provided to Configure workflow
+type configureParam struct {
+	SystemID string
+}
+
+// Configure calls Configure workflow to be executed.
+// This workflow will configure WorkerPool with a proper set of workers.
+// E.g. it will call AddWorker and RemoveWorker workflows.
+func (p *WorkerPool) Configure(ctx context.Context) error {
+	workflowOptions := client.StartWorkflowOptions{
+		TaskQueue: "control-plane",
+	}
+
+	workflowRun, err := p.client.ExecuteWorkflow(ctx, workflowOptions,
+		"Configure", configureParam{SystemID: p.systemID})
+	if err != nil {
+		return err
+	}
+
+	return workflowRun.Get(ctx, nil)
+}
+
+type addWorkerParam struct {
+	TaskQueue  string
+	Workflows  []string
+	Activities []string
+}
+
+// addWorker adds worker to the WorkerPool and registers workflows and activities
+func (p *WorkerPool) addWorker(param addWorkerParam) error {
+	p.mutex.Lock()
+	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)
+	}
+
+	w := worker.New(p.client, param.TaskQueue, worker.Options{})
+
+	for _, workflow := range param.Workflows {
+		if fn, ok := p.workflows[workflow]; ok {
+			w.RegisterWorkflow(fn)
+		}
+	}
+
+	for _, activity := range param.Activities {
+		if fn, ok := p.activities[activity]; ok {
+			w.RegisterActivity(fn)
+		}
+	}
+
+	if err := w.Start(); err != nil {
+		return err
+	}
+
+	p.workers[param.TaskQueue] = w
+
+	return nil
+}
+
+type removeWorkerParam struct {
+	TaskQueue string
+}
+
+// removeWorker stops worker of a certain TaskQueue and removes it from the pool
+func (p *WorkerPool) removeWorker(param removeWorkerParam) error {
+	if w, ok := p.workers[param.TaskQueue]; ok {
+		w.Stop()
+	}
+
+	p.mutex.Lock()
+	delete(p.workers, param.TaskQueue)
+	defer p.mutex.Unlock()
+
+	return nil
+}
+
+// exec will execute provide function as Local Activity
+func exec[T any](fn any) func(ctx workflow.Context, param T) error {
+	return func(ctx workflow.Context, param T) error {
+		lao := workflow.LocalActivityOptions{
+			ScheduleToCloseTimeout: 5 * time.Second,
+		}
+		ctx = workflow.WithLocalActivityOptions(ctx, lao)
+
+		return workflow.ExecuteLocalActivity(ctx, fn, param).Get(ctx, nil)
+	}
+}
diff --git a/src/maasserver/models/service.py b/src/maasserver/models/service.py
index 886ee8c..4680ff9 100644
--- a/src/maasserver/models/service.py
+++ b/src/maasserver/models/service.py
@@ -37,6 +37,7 @@ RACK_SERVICES = frozenset(
         "dns_rack",
         "proxy_rack",
         "syslog_rack",
+        "agent",
     }
 )
 
@@ -59,6 +60,7 @@ DEAD_STATUSES = {
     "syslog_rack": SERVICE_STATUS.UNKNOWN,
     "reverse_proxy": SERVICE_STATUS.UNKNOWN,
     "temporal": SERVICE_STATUS.UNKNOWN,
+    "agent": SERVICE_STATUS.UNKNOWN,
 }
 
 
diff --git a/src/provisioningserver/rackdservices/external.py b/src/provisioningserver/rackdservices/external.py
index 4944b02..94b1576 100644
--- a/src/provisioningserver/rackdservices/external.py
+++ b/src/provisioningserver/rackdservices/external.py
@@ -392,6 +392,24 @@ class RackSyslog(RackOnlyExternalService):
         )
 
 
+class RackAgent(RackOnlyExternalService):
+    """External maas-agent service"""
+
+    service_name = "agent"
+
+    def _configure(self, configuration):
+        return
+
+    def _tryUpdate(self, config):
+        d = maybeDeferred(
+            self._getConfiguration,
+        )
+        return d
+
+    def _getConfiguration(self):
+        return
+
+
 class RackExternalService(TimerService):
     # Initial start the interval is low so that forwarders of bind9 gets
     # at least one region controller. When no region controllers are set
@@ -417,6 +435,7 @@ class RackExternalService(TimerService):
                 ("DNS", RackDNS()),
                 ("proxy", RackProxy()),
                 ("syslog", RackSyslog()),
+                ("agent", RackAgent()),
             ]
 
     def _update_interval(self, config):
diff --git a/src/provisioningserver/service_monitor.py b/src/provisioningserver/service_monitor.py
index 20d7642..de2e115 100644
--- a/src/provisioningserver/service_monitor.py
+++ b/src/provisioningserver/service_monitor.py
@@ -66,6 +66,14 @@ class SyslogServiceOnRack(ToggleableService):
     snap_service_name = "syslog"
 
 
+class AgentServiceOnRack(AlwaysOnService):
+    """Monitored MAAS Agent service on a rack controller host."""
+
+    name = "agent"
+    service_name = "maas-agent"
+    snap_service_name = "agent"
+
+
 # Global service monitor for rackd. NOTE that changes to this need to be
 # mirrored in maasserver.model.services.
 service_monitor = ServiceMonitor(
@@ -76,4 +84,5 @@ service_monitor = ServiceMonitor(
     DNSServiceOnRack(),
     ProxyServiceOnRack(),
     SyslogServiceOnRack(),
+    AgentServiceOnRack(),
 )
diff --git a/src/provisioningserver/tests/test_service_monitor.py b/src/provisioningserver/tests/test_service_monitor.py
index 8147d5b..3116eec 100644
--- a/src/provisioningserver/tests/test_service_monitor.py
+++ b/src/provisioningserver/tests/test_service_monitor.py
@@ -6,6 +6,7 @@
 
 from maastesting.testcase import MAASTestCase
 from provisioningserver.service_monitor import (
+    AgentServiceOnRack,
     DHCPv4Service,
     DHCPv6Service,
     DNSServiceOnRack,
@@ -66,6 +67,14 @@ class TestSyslogServiceOnRack(MAASTestCase):
         self.assertEqual("syslog_rack", syslog.name)
 
 
+class TestAgentServiceOnRack(MAASTestCase):
+    def test_name_and_service_name(self):
+        syslog = AgentServiceOnRack()
+        self.assertEqual("maas-agent", syslog.service_name)
+        self.assertEqual("agent", syslog.snap_service_name)
+        self.assertEqual("agent", syslog.name)
+
+
 class TestGlobalServiceMonitor(MAASTestCase):
     def test_includes_all_services(self):
         self.assertEqual(
@@ -77,6 +86,7 @@ class TestGlobalServiceMonitor(MAASTestCase):
                 "ntp_rack",
                 "proxy_rack",
                 "syslog_rack",
+                "agent",
             },
             service_monitor._services.keys(),
         )