feat: port conflict validation

Adapts the `ValidateInstanceSpec` activity to take and validate multiple
specs in order to detect when a port has been allocated multiple times
on a single host.

Note that this will not consistently detect when a port has been
allocated multiple times on a machine when there are multiple
control-plane instances with different host IDs on the same machine - as
in our development configuration. End users should not run the control
plane this way, so we donn't need to handle that case explicitly.

PLAT-86

Author: jason-lynch

server/internal/database/orchestrator.go

@@ -20,8 +20,11 @@ type InstanceResources struct {
 }
 
 type ValidationResult struct {
-	Valid bool
-	Error string
+	InstanceID string   `json:"instance_id"`
+	HostID     string   `json:"host_id"`
+	NodeName   string   `json:"node_name"`
+	Valid      bool     `json:"valid"`
+	Errors     []string `json:"errors"`
 }
 
 func NewInstanceResources(instance *InstanceResource, resources []resource.Resource) (*InstanceResources, error) {
@@ -88,5 +91,5 @@ type Orchestrator interface {
 	GenerateInstanceRestoreResources(spec *InstanceSpec, taskID uuid.UUID) (*InstanceResources, error)
 	GetInstanceConnectionInfo(ctx context.Context, databaseID, instanceID string) (*ConnectionInfo, error)
 	CreatePgBackRestBackup(ctx context.Context, w io.Writer, instanceID string, options *pgbackrest.BackupOptions) error
-	ValidateInstanceSpec(ctx context.Context, spec *InstanceSpec) (*ValidationResult, error)
+	ValidateInstanceSpecs(ctx context.Context, specs []*InstanceSpec) ([]*ValidationResult, error)
 }

server/internal/docker/docker.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net/netip"
+	"regexp"
 	"strings"
 	"time"
 
@@ -527,20 +528,28 @@ func ExtractBindMountErrorMsg(err error) string {
 	return strings.TrimPrefix(msg[idx:], bindMountErrPrefix)
 }
 
-// Another internal error message:
+// ExtractPortErrorMsg extracts the port bind error message from the given error
+// if it is a port bind or allocation error. Otherwise, returns an empty string.
+func ExtractPortErrorMsg(err error) string {
+	if err == nil {
+		return ""
+	}
+	if msg := extractPortBindErrorMsg(err); msg != "" {
+		return msg
+	}
+	if msg := extractPortAlreadyAllocatedErrorMsg(err); msg != "" {
+		return msg
+	}
+	return ""
+}
+
+// More internal error messages:
 // https://github.com/moby/moby/blob/cab4ac834e8bf36aa38a2ca49599773df6e6805a/libnetwork/drivers/bridge/port_mapping_linux.go#L622-L627
 // This one is less stable, so we'll do our best. In the worst case, we return a
 // 500 with a longer error message, which will still be helpful to the user.
 const portBindErrPrefix = `failed to bind`
 
-// ExtractPortBindError extracts the port bind error message from the given
-// error if it is a port bind error. Otherwise, returns an empty string.
-// ExtractBindError extracts the bind error message from the given error if it
-// is a bind error. Otherwise, returns an empty string.
-func ExtractPortBindErrorMsg(err error) string {
-	if err == nil {
-		return ""
-	}
+func extractPortBindErrorMsg(err error) string {
 	msg := err.Error()
 	idx := strings.Index(msg, portBindErrPrefix)
 	if idx < 0 {
@@ -550,6 +559,14 @@ func ExtractPortBindErrorMsg(err error) string {
 	return msg[idx:]
 }
 
+// https://github.com/moby/moby/blob/9b4f68d64cde951e5b985a0c589f16f1416d3968/libnetwork/portallocator/portallocator.go#L33
+// This message has been stable for 10 years.
+var portAlreadyAllocatedPattern = regexp.MustCompile(`Bind for .* failed: port is already allocated`)
+
+func extractPortAlreadyAllocatedErrorMsg(err error) string {
+	return portAlreadyAllocatedPattern.FindString(err.Error())
+}
+
 // The docker errors are annoying to check further up in the stack since they
 // rely on type checks. Wrapping them in our own errors makes it easier for
 // callers to explicitly handle specific errors.

server/internal/orchestrator/swarm/orchestrator.go

@@ -26,6 +26,7 @@ import (
 	"github.com/pgEdge/control-plane/server/internal/config"
 	"github.com/pgEdge/control-plane/server/internal/database"
 	"github.com/pgEdge/control-plane/server/internal/docker"
+	"github.com/pgEdge/control-plane/server/internal/ds"
 	"github.com/pgEdge/control-plane/server/internal/filesystem"
 	"github.com/pgEdge/control-plane/server/internal/host"
 	"github.com/pgEdge/control-plane/server/internal/patroni"
@@ -448,10 +449,40 @@ func (o *Orchestrator) CreatePgBackRestBackup(ctx context.Context, w io.Writer,
 	return nil
 }
 
-func (o *Orchestrator) ValidateInstanceSpec(ctx context.Context, spec *database.InstanceSpec) (*database.ValidationResult, error) {
+func (o *Orchestrator) ValidateInstanceSpecs(ctx context.Context, specs []*database.InstanceSpec) ([]*database.ValidationResult, error) {
+	results := make([]*database.ValidationResult, len(specs))
+
+	occupiedPorts := ds.NewSet[int]()
+	for i, instance := range specs {
+		result := &database.ValidationResult{
+			InstanceID: instance.InstanceID,
+			HostID:     instance.HostID,
+			NodeName:   instance.NodeName,
+			Valid:      true,
+		}
+		if instance.Port > 0 {
+			if occupiedPorts.Has(instance.Port) {
+				result.Valid = false
+				result.Errors = append(
+					result.Errors,
+					fmt.Sprintf("port %d allocated to multiple instances on this host", instance.Port),
+				)
+			}
+			occupiedPorts.Add(instance.Port)
+		}
+		if err := o.validateInstanceSpec(ctx, instance, result); err != nil {
+			return nil, err
+		}
+		results[i] = result
+	}
+
+	return results, nil
+}
+
+func (o *Orchestrator) validateInstanceSpec(ctx context.Context, spec *database.InstanceSpec, result *database.ValidationResult) error {
 	// Short-circuit if there's nothing to validate
 	if len(spec.ExtraVolumes) < 1 && spec.Port == 0 {
-		return &database.ValidationResult{Valid: true}, nil
+		return nil
 	}
 
 	specVersion := spec.PgEdgeVersion
@@ -462,7 +493,7 @@ func (o *Orchestrator) ValidateInstanceSpec(ctx context.Context, spec *database.
 
 	images, err := GetImages(o.cfg, specVersion)
 	if err != nil {
-		return nil, fmt.Errorf("image fetch error: %w", err)
+		return fmt.Errorf("image fetch error: %w", err)
 	}
 
 	var mounts []mount.Mount
@@ -474,30 +505,23 @@ func (o *Orchestrator) ValidateInstanceSpec(ctx context.Context, spec *database.
 
 	cmd := buildVolumeCheckCommand(mountTargets)
 	output, err := o.runValidationContainer(ctx, images.PgEdgeImage, cmd, mounts, spec.Port)
-	if msg := docker.ExtractBindMountErrorMsg(err); msg != "" {
-		return &database.ValidationResult{
-			Valid: false,
-			Error: msg,
-		}, nil
-	}
-	if msg := docker.ExtractPortBindErrorMsg(err); msg != "" {
-		return &database.ValidationResult{
-			Valid: false,
-			Error: msg,
-		}, nil
-	}
-	if err != nil {
-		return nil, err
-	}
-
-	if len(output) > 0 {
-		return &database.ValidationResult{
-			Valid: false,
-			Error: output,
-		}, nil
+	bindMsg := docker.ExtractBindMountErrorMsg(err)
+	portMsg := docker.ExtractPortErrorMsg(err)
+	switch {
+	case bindMsg != "":
+		result.Valid = false
+		result.Errors = append(result.Errors, bindMsg)
+	case portMsg != "":
+		result.Valid = false
+		result.Errors = append(result.Errors, portMsg)
+	case err != nil:
+		return err
+	case len(output) > 0:
+		result.Valid = false
+		result.Errors = append(result.Errors, output)
 	}
 
-	return &database.ValidationResult{Valid: true}, nil
+	return nil
 }
 
 func (o *Orchestrator) runValidationContainer(ctx context.Context, image string, cmd []string, mounts []mount.Mount, port int) (string, error) {
@@ -563,6 +587,18 @@ func validationContainerOpts(image string, cmd []string, mounts []mount.Mount, p
 	return opts
 }
 
+const cmdTemplate = `
+for d in %s; do
+	if [ ! -d "$d" ]; then
+		echo "$d is not a directory"
+	fi
+done
+`
+
 func buildVolumeCheckCommand(mountTargets []string) []string {
-	return []string{"sh", "-c", fmt.Sprintf(`for d in %s; do if [ -d "$d" ]; then echo "$d is not a directory"; fi; done"`, strings.Join(mountTargets, " "))}
+	if len(mountTargets) < 1 {
+		return []string{"true"}
+	}
+
+	return []string{"sh", "-c", fmt.Sprintf(cmdTemplate, strings.Join(mountTargets, " "))}
 }

server/internal/workflows/activities/activities.go

@@ -34,7 +34,7 @@ func (a *Activities) Register(work *worker.Worker) error {
 		work.RegisterActivity(a.RestoreSpec),
 		work.RegisterActivity(a.UpdateDbState),
 		work.RegisterActivity(a.UpdateTask),
-		work.RegisterActivity(a.ValidateInstanceSpec),
+		work.RegisterActivity(a.ValidateInstanceSpecs),
 	}
 	return errors.Join(errs...)
 }

server/internal/workflows/activities/validate_instance_spec.go

@@ -0,0 +1,58 @@
+package activities
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/activity"
+	"github.com/cschleiden/go-workflows/workflow"
+	"github.com/pgEdge/control-plane/server/internal/database"
+)
+
+type ValidateInstanceSpecsInput struct {
+	DatabaseID string                   `json:"database_id"`
+	Specs      []*database.InstanceSpec `json:"spec"`
+}
+
+type ValidateInstanceSpecsOutput struct {
+	Results []*database.ValidationResult `json:"results"`
+}
+
+func (a *Activities) ExecuteValidateInstanceSpecs(
+	ctx workflow.Context,
+	hostID string,
+	input *ValidateInstanceSpecsInput,
+) workflow.Future[*ValidateInstanceSpecsOutput] {
+	options := workflow.ActivityOptions{
+		Queue: workflow.Queue(hostID),
+		RetryOptions: workflow.RetryOptions{
+			MaxAttempts: 1,
+		},
+	}
+	return workflow.ExecuteActivity[*ValidateInstanceSpecsOutput](ctx, options, a.ValidateInstanceSpecs, input)
+}
+
+func (a *Activities) ValidateInstanceSpecs(ctx context.Context, input *ValidateInstanceSpecsInput) (*ValidateInstanceSpecsOutput, error) {
+	logger := activity.Logger(ctx)
+
+	if input == nil {
+		return nil, errors.New("input is nil")
+	}
+
+	logger = logger.With(
+		"database_id", input.DatabaseID,
+		"host_id", a.Config.HostID,
+	)
+	logger.Info("starting instance spec validation")
+
+	results, err := a.Orchestrator.ValidateInstanceSpecs(ctx, input.Specs)
+	if err != nil {
+		return nil, fmt.Errorf("instance spec validation failed: %w", err)
+	}
+
+	logger.Info("instance spec validation completed")
+	return &ValidateInstanceSpecsOutput{
+		Results: results,
+	}, nil
+}

server/internal/workflows/activities/validate_instance_specs.go

@@ -251,18 +251,14 @@ func (s *Service) ValidateSpec(ctx context.Context, spec *database.Spec) (*Valid
 
 	instance, err := s.client.CreateWorkflowInstance(ctx, opts, s.workflows.ValidateSpec, input)
 	if err != nil {
-		s.logger.Error().Err(err).Str("database_id", databaseID).Msg("failed to create volume validation workflow")
+		s.logger.Error().Err(err).Str("database_id", databaseID).Msg("failed to create spec validation workflow")
 		return nil, fmt.Errorf("failed to create workflow instance: %w", err)
 	}
 
 	output, err := client.GetWorkflowResult[*ValidateSpecOutput](ctx, s.client, instance, 5*time.Minute)
 	if err != nil {
-
-	}
-
-	if err != nil {
-		s.logger.Error().Err(err).Str("database_id", databaseID).Msg("volume validation workflow failed")
-		return nil, fmt.Errorf("volume validation workflow failed: %w", err)
+		s.logger.Error().Err(err).Str("database_id", databaseID).Msg("spec validation workflow failed")
+		return nil, fmt.Errorf("spec validation workflow failed: %w", err)
 	}
 
 	return output, nil