(choria-io#2029) Extract some common code and reuse in CLI

Also improve the plugins pack cli.

The external agents provider will check frequently at
first for new agents then settle into a 30s kind of interval,
with jitter.  This optimise bootup to deliver agents quickly
but with a more relaxed settled state

Signed-off-by: R.I.Pienaar <rip@devco.net>

Author: ripienaar

aagent/watchers/pluginswatcher/plugins.go

@@ -54,11 +54,6 @@ var stateNames = map[State]string{
 	Unchanged: "unchanged",
 }
 
-type Specification struct {
-	Plugins   []byte `json:"plugins"`
-	Signature string `json:"signature,omitempty"`
-}
-
 type ManagedPlugin struct {
 	Name                     string `json:"name" yaml:"name"`
 	NamePrefix               string `json:"-" yaml:"-"`
@@ -253,7 +248,7 @@ func (w *Watcher) watch(ctx context.Context) (state State, err error) {
 			}
 		}
 
-		w.Warnf("Deploying Choria Autonomous Agent %s from %s", m.Name, m.Source)
+		w.Warnf("Deploying plugin %s from %s info %s", m.Name, m.Source, targetDir)
 
 		err = os.MkdirAll(targetDir, 0700)
 		if err != nil {

aagent/watchers/pluginswatcher/specification.go

@@ -0,0 +1,50 @@
+// Copyright (c) 2023, R.I. Pienaar and the Choria Project contributors
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package machines
+
+import (
+	"crypto/ed25519"
+	"encoding/hex"
+	"encoding/json"
+
+	iu "github.com/choria-io/go-choria/internal/util"
+)
+
+// Specification holds []ManagedPlugin marshaled to JSON with an optional ed25519 signature
+type Specification struct {
+	Plugins   []byte `json:"plugins"`
+	Signature string `json:"signature,omitempty"`
+}
+
+// Encode sets the signature and Marshals the specification to JSON, if key is empty signature is not made
+func (s *Specification) Encode(key string) ([]byte, error) {
+	var pk ed25519.PrivateKey
+	var err error
+
+	data, err := json.Marshal(s)
+	if err != nil {
+		return nil, err
+	}
+
+	if key != "" {
+		if iu.IsEncodedEd25519KeyString(key) {
+			pk, err = hex.DecodeString(key)
+		} else {
+			_, pk, err = iu.Ed25519KeyPairFromSeedFile(key)
+		}
+		if err != nil {
+			return nil, err
+		}
+
+		sig, err := iu.Ed25519Sign(pk, data)
+		if err != nil {
+			return nil, err
+		}
+
+		s.Signature = hex.EncodeToString(sig)
+	}
+
+	return json.Marshal(s)
+}

broker/network/choria_auth.go

@@ -663,7 +663,7 @@ func (a *ChoriaAuth) cachedEd25519Token(token string) (ed25519.PublicKey, error)
 func (a *ChoriaAuth) parseServerJWTWithSigners(jwts string) (claims *tokens.ServerClaims, err error) {
 	for _, s := range a.serverJwtSigners {
 		// its a token
-		if tokens.IsEncodedEd25519Key([]byte(s)) {
+		if util.IsEncodedEd25519KeyString(s) {
 			var pk ed25519.PublicKey
 			pk, err = a.cachedEd25519Token(s)
 			if err != nil {
@@ -765,7 +765,7 @@ func (a *ChoriaAuth) parseServerJWT(jwts string) (claims *tokens.ServerClaims, e
 func (a *ChoriaAuth) parseClientJWTWithSigners(jwts string) (claims *tokens.ClientIDClaims, err error) {
 	for _, s := range a.clientJwtSigners {
 		// its a token
-		if tokens.IsEncodedEd25519Key([]byte(s)) {
+		if util.IsEncodedEd25519KeyString(s) {
 			var pk ed25519.PublicKey
 			pk, err = a.cachedEd25519Token(s)
 			if err != nil {

cmd/machine_plugins.go

@@ -0,0 +1,34 @@
+// Copyright (c) 2023, R.I. Pienaar and the Choria Project contributors
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package cmd
+
+import (
+	"sync"
+)
+
+type machinePluginsCommand struct {
+	command
+}
+
+func (p *machinePluginsCommand) Setup() (err error) {
+	if machine, ok := cmdWithFullCommand("machine"); ok {
+		p.cmd = machine.Cmd().Command("plugins", "Manage specifications for the plugins watcher")
+	}
+	return nil
+}
+
+func (p *machinePluginsCommand) Configure() error {
+	return nil
+}
+
+func (p *machinePluginsCommand) Run(wg *sync.WaitGroup) (err error) {
+	defer wg.Done()
+
+	return nil
+}
+
+func init() {
+	cli.commands = append(cli.commands, &machinePluginsCommand{})
+}

cmd/machine_pack.go renamed to cmd/machine_plugins_pack.go

@@ -5,8 +5,6 @@
 package cmd
 
 import (
-	"crypto/ed25519"
-	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -18,18 +16,18 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-type mPackCommand struct {
+type mPluginsPackCommand struct {
 	command
-	machines string
-	key      string
-	out      string
-	force    bool
+	source string
+	key    string
+	out    string
+	force  bool
 }
 
-func (r *mPackCommand) Setup() (err error) {
-	if machine, ok := cmdWithFullCommand("machine"); ok {
-		r.cmd = machine.Cmd().Command("plugins", "Encodes and signs data for the plugins watcher")
-		r.cmd.Arg("source", "File containing the plugins definition").Required().ExistingFileVar(&r.machines)
+func (r *mPluginsPackCommand) Setup() (err error) {
+	if machine, ok := cmdWithFullCommand("machine plugins"); ok {
+		r.cmd = machine.Cmd().Command("pack", "Encodes and signs data for the plugins watcher")
+		r.cmd.Arg("source", "File containing the plugins definition").Required().ExistingFileVar(&r.source)
 		r.cmd.Arg("key", "The ed25519 private key to encode with").StringVar(&r.key)
 		r.cmd.Flag("force", "Do not warn about no ed25519 key and support writing empty files").BoolVar(&r.force)
 		r.cmd.Flag("output", "Write result to a file").StringVar(&r.out)
@@ -38,7 +36,7 @@ func (r *mPackCommand) Setup() (err error) {
 	return nil
 }
 
-func (r *mPackCommand) Configure() error {
+func (r *mPluginsPackCommand) Configure() error {
 	if debug {
 		logrus.SetOutput(os.Stdout)
 		logrus.SetLevel(logrus.DebugLevel)
@@ -56,10 +54,10 @@ func (r *mPackCommand) Configure() error {
 	return err
 }
 
-func (r *mPackCommand) Run(wg *sync.WaitGroup) (err error) {
+func (r *mPluginsPackCommand) Run(wg *sync.WaitGroup) (err error) {
 	defer wg.Done()
 
-	data, err := os.ReadFile(r.machines)
+	data, err := os.ReadFile(r.source)
 	if err != nil {
 		return err
 	}
@@ -74,25 +72,12 @@ func (r *mPackCommand) Run(wg *sync.WaitGroup) (err error) {
 		return fmt.Errorf("no plugins listed in specification, use --force to write an empty list")
 	}
 
-	spec := watcher.Specification{Plugins: data}
-
-	if r.key != "" {
-		var key []byte
-		if iu.FileExist(r.key) {
-			key, err = os.ReadFile(r.key)
-		} else {
-			key, err = hex.DecodeString(r.key)
-		}
-		if err != nil {
-			return err
-		}
-
-		spec.Signature = hex.EncodeToString(ed25519.Sign(key, data))
-	} else if !r.force {
+	if r.key == "" && !r.force {
 		logrus.Warn("No ed25519 private key given, encoding without signing")
 	}
 
-	j, err := json.Marshal(spec)
+	spec := &watcher.Specification{Plugins: data}
+	j, err := spec.Encode(r.key)
 	if err != nil {
 		return err
 	}
@@ -110,5 +95,5 @@ func (r *mPackCommand) Run(wg *sync.WaitGroup) (err error) {
 }
 
 func init() {
-	cli.commands = append(cli.commands, &mPackCommand{})
+	cli.commands = append(cli.commands, &mPluginsPackCommand{})
 }

internal/util/ed25519.go

@@ -10,6 +10,8 @@ import (
 	"encoding/hex"
 	"fmt"
 	"os"
+
+	"github.com/choria-io/tokens"
 )
 
 func Ed25519SignWithSeedFile(f string, msg []byte) ([]byte, error) {
@@ -78,3 +80,11 @@ func Ed25519KeyPairToFile(f string) (ed25519.PublicKey, ed25519.PrivateKey, erro
 
 	return pubK, priK, nil
 }
+
+func IsEncodedEd25519Key(k []byte) bool {
+	return tokens.IsEncodedEd25519Key(k)
+}
+
+func IsEncodedEd25519KeyString(k string) bool {
+	return tokens.IsEncodedEd25519Key([]byte(k))
+}

providers/agent/mcorpc/external/provider.go

@@ -14,6 +14,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/choria-io/go-choria/backoff"
 	"github.com/sirupsen/logrus"
 
 	"github.com/choria-io/go-choria/config"
@@ -26,10 +27,8 @@ import (
 var (
 	// agents we do not ever wish to load from external agents
 	denyList = []string{"rpcutil", "choria_util", "choria_provision", "choria_registry", "discovery", "scout"}
-	// how frequently agents are reconciled
-	watchInterval = time.Minute
 	// we only consider ddl files modified longer than this ago for reconciliation
-	fileChangeGrace = 20 * time.Second
+	fileChangeGrace = 5 * time.Second
 )
 
 // Provider is a Choria Agent Provider that supports calling agents external to the
@@ -211,8 +210,8 @@ func (p *Provider) watchAgents(ctx context.Context, mgr server.AgentManager, con
 		p.log.Errorf("Initial agent reconcile failed: %v", err)
 	}
 
-	ticker := time.NewTicker(watchInterval)
-	p.log.Debugf("Watching for agent updates every %v", watchInterval)
+	count := 1
+	ticker := time.NewTicker(backoff.TwentySec.Duration(count))
 
 	for {
 		select {
@@ -222,6 +221,9 @@ func (p *Provider) watchAgents(ctx context.Context, mgr server.AgentManager, con
 				p.log.Errorf("Reconciling agents failed: %v", err)
 			}
 
+			count++
+			ticker.Reset(backoff.TwentySec.Duration(count))
+
 		case <-ctx.Done():
 			return
 		}