feat: stable random ports

Adds a service to allocate random ports for instances from a
configurable range. These ports are then persisted with the instance
spec. The next time the instance spec is evaluated, we copy the ports
from the persisted copy of the spec.

The end result is that when we assign a random port to an instance, it
will retain that port assignment until it is deleted or the user
sets the port to `null` or a non-zero value in the database/node spec.
Since this mechanism does not rely on Docker or the OS to assign ports,
it will also work for the SystemD orchestrator.

PLAT-417
PLAT-236

Author: jason-lynch

docs/installation/configuration.md

@@ -68,3 +68,4 @@ This is the current list of components that can be configured in the `logging.co
 - `migration_runner`
 - `scheduler_service`
 - `workflows_worker`
+- `ports_service`

server/cmd/root.go

@@ -23,6 +23,7 @@ import (
 	"github.com/pgEdge/control-plane/server/internal/monitor"
 	"github.com/pgEdge/control-plane/server/internal/orchestrator"
 	"github.com/pgEdge/control-plane/server/internal/orchestrator/swarm"
+	"github.com/pgEdge/control-plane/server/internal/ports"
 	"github.com/pgEdge/control-plane/server/internal/resource"
 	"github.com/pgEdge/control-plane/server/internal/scheduler"
 	"github.com/pgEdge/control-plane/server/internal/task"
@@ -76,6 +77,7 @@ func newRootCmd(i *do.Injector) *cobra.Command {
 			logging.Provide(i)
 			migrate.Provide(i)
 			monitor.Provide(i)
+			ports.Provide(i)
 			resource.Provide(i)
 			scheduler.Provide(i)
 			workflows.Provide(i)

server/internal/config/config.go

@@ -194,6 +194,32 @@ const (
 	EtcdModeClient EtcdMode = "client"
 )
 
+type RandomPorts struct {
+	Min int `koanf:"min" json:"min,omitempty"`
+	Max int `koanf:"max" json:"max,omitempty"`
+}
+
+func (r RandomPorts) validate() []error {
+	var errs []error
+	if r.Min < 1 {
+		errs = append(errs, errors.New("min: cannot be less than 1"))
+	}
+	if r.Max > 65535 {
+		errs = append(errs, errors.New("max: cannot be greater than 65535"))
+	}
+	if r.Max <= r.Min {
+		errs = append(errs, errors.New("max: must be greater than min"))
+	}
+	return errs
+}
+
+// We're intentionally using a range that's well below the ephemeral port range
+// to reduce the risk of interference from the OS.
+var defaultRandomPorts = RandomPorts{
+	Min: 5432,
+	Max: 15432,
+}
+
 type Config struct {
 	TenantID               string       `koanf:"tenant_id" json:"tenant_id,omitempty"`
 	HostID                 string       `koanf:"host_id" json:"host_id,omitempty"`
@@ -216,6 +242,7 @@ type Config struct {
 	DockerSwarm            DockerSwarm  `koanf:"docker_swarm" json:"docker_swarm,omitzero"`
 	DatabaseOwnerUID       int          `koanf:"database_owner_uid" json:"database_owner_uid,omitempty"`
 	ProfilingEnabled       bool         `koanf:"profiling_enabled" json:"profiling_enabled,omitempty"`
+	RandomPorts            RandomPorts  `koanf:"random_ports" json:"random_ports,omitzero"`
 }
 
 // ClientAddress is a convenience function to return the first client address.
@@ -310,6 +337,9 @@ func (c Config) Validate() error {
 	for _, err := range c.Logging.validate() {
 		errs = append(errs, fmt.Errorf("logging.%w", err))
 	}
+	for _, err := range c.RandomPorts.validate() {
+		errs = append(errs, fmt.Errorf("random_ports.%w", err))
+	}
 	if c.Orchestrator != OrchestratorSwarm {
 		errs = append(errs, fmt.Errorf("orchestrator: unsupported orchestrator %q", c.Orchestrator))
 	}
@@ -361,6 +391,7 @@ func DefaultConfig() (Config, error) {
 		EtcdClient:             etcdClientDefault,
 		DockerSwarm:            defaultDockerSwarm,
 		DatabaseOwnerUID:       26,
+		RandomPorts:            defaultRandomPorts,
 	}, nil
 }
 

server/internal/database/instance_spec_store.go

@@ -0,0 +1,66 @@
+package database
+
+import (
+	clientv3 "go.etcd.io/etcd/client/v3"
+
+	"github.com/pgEdge/control-plane/server/internal/storage"
+)
+
+type StoredInstanceSpec struct {
+	storage.StoredValue
+	Spec *InstanceSpec `json:"spec"`
+}
+
+type InstanceSpecStore struct {
+	client *clientv3.Client
+	root   string
+}
+
+func NewInstanceSpecStore(client *clientv3.Client, root string) *InstanceSpecStore {
+	return &InstanceSpecStore{
+		client: client,
+		root:   root,
+	}
+}
+
+func (s *InstanceSpecStore) Prefix() string {
+	return storage.Prefix("/", s.root, "instance_specs")
+}
+
+func (s *InstanceSpecStore) DatabasePrefix(databaseID string) string {
+	return storage.Prefix(s.Prefix(), databaseID)
+}
+
+func (s *InstanceSpecStore) Key(databaseID, instanceID string) string {
+	return storage.Key(s.DatabasePrefix(databaseID), instanceID)
+}
+
+func (s *InstanceSpecStore) GetByKey(databaseID, instanceID string) storage.GetOp[*StoredInstanceSpec] {
+	key := s.Key(databaseID, instanceID)
+	return storage.NewGetOp[*StoredInstanceSpec](s.client, key)
+}
+
+func (s *InstanceSpecStore) GetByDatabaseID(databaseID string) storage.GetMultipleOp[*StoredInstanceSpec] {
+	prefix := s.DatabasePrefix(databaseID)
+	return storage.NewGetPrefixOp[*StoredInstanceSpec](s.client, prefix)
+}
+
+func (s *InstanceSpecStore) GetAll() storage.GetMultipleOp[*StoredInstanceSpec] {
+	prefix := s.Prefix()
+	return storage.NewGetPrefixOp[*StoredInstanceSpec](s.client, prefix)
+}
+
+func (s *InstanceSpecStore) Put(item *StoredInstanceSpec) storage.PutOp[*StoredInstanceSpec] {
+	key := s.Key(item.Spec.DatabaseID, item.Spec.InstanceID)
+	return storage.NewPutOp(s.client, key, item)
+}
+
+func (s *InstanceSpecStore) DeleteByKey(databaseID, instanceID string) storage.DeleteOp {
+	key := s.Key(databaseID, instanceID)
+	return storage.NewDeleteKeyOp(s.client, key)
+}
+
+func (s *InstanceSpecStore) DeleteByDatabaseID(databaseID string) storage.DeleteOp {
+	prefix := s.DatabasePrefix(databaseID)
+	return storage.NewDeletePrefixOp(s.client, prefix)
+}

server/internal/database/provide.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/pgEdge/control-plane/server/internal/config"
 	"github.com/pgEdge/control-plane/server/internal/host"
+	"github.com/pgEdge/control-plane/server/internal/ports"
 )
 
 func Provide(i *do.Injector) {
@@ -15,6 +16,10 @@ func Provide(i *do.Injector) {
 
 func provideService(i *do.Injector) {
 	do.Provide(i, func(i *do.Injector) (*Service, error) {
+		cfg, err := do.Invoke[config.Config](i)
+		if err != nil {
+			return nil, err
+		}
 		orch, err := do.Invoke[Orchestrator](i)
 		if err != nil {
 			return nil, err
@@ -27,7 +32,11 @@ func provideService(i *do.Injector) {
 		if err != nil {
 			return nil, err
 		}
-		return NewService(orch, store, hostSvc), nil
+		portsSvc, err := do.Invoke[*ports.Service](i)
+		if err != nil {
+			return nil, err
+		}
+		return NewService(cfg, orch, store, hostSvc, portsSvc), nil
 	})
 }
 

server/internal/database/service.go

@@ -8,8 +8,11 @@ import (
 
 	"github.com/google/uuid"
 
+	"github.com/pgEdge/control-plane/server/internal/config"
 	"github.com/pgEdge/control-plane/server/internal/host"
+	"github.com/pgEdge/control-plane/server/internal/ports"
 	"github.com/pgEdge/control-plane/server/internal/storage"
+	"github.com/pgEdge/control-plane/server/internal/utils"
 )
 
 var (
@@ -24,16 +27,26 @@ var (
 )
 
 type Service struct {
+	cfg          config.Config
 	orchestrator Orchestrator
 	store        *Store
 	hostSvc      *host.Service
+	portsSvc     *ports.Service
 }
 
-func NewService(orchestrator Orchestrator, store *Store, hostSvc *host.Service) *Service {
+func NewService(
+	cfg config.Config,
+	orchestrator Orchestrator,
+	store *Store,
+	hostSvc *host.Service,
+	portsSvc *ports.Service,
+) *Service {
 	return &Service{
+		cfg:          cfg,
 		orchestrator: orchestrator,
 		store:        store,
 		hostSvc:      hostSvc,
+		portsSvc:     portsSvc,
 	}
 }
 
@@ -119,6 +132,18 @@ func (s *Service) UpdateDatabase(ctx context.Context, state DatabaseState, spec
 }
 
 func (s *Service) DeleteDatabase(ctx context.Context, databaseID string) error {
+	specs, err := s.store.InstanceSpec.
+		GetByDatabaseID(databaseID).
+		Exec(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get instance specs: %w", err)
+	}
+	for _, spec := range specs {
+		if err := s.releaseInstancePorts(ctx, spec.Spec); err != nil {
+			return err
+		}
+	}
+
 	var ops []storage.TxnOperation
 
 	spec, err := s.store.Spec.GetByKey(databaseID).Exec(ctx)
@@ -141,6 +166,7 @@ func (s *Service) DeleteDatabase(ctx context.Context, databaseID string) error {
 
 	ops = append(ops,
 		s.store.Instance.DeleteByDatabaseID(databaseID),
+		s.store.InstanceSpec.DeleteByDatabaseID(databaseID),
 		s.store.InstanceStatus.DeleteByDatabaseID(databaseID),
 	)
 
@@ -279,7 +305,7 @@ func (s *Service) DeleteInstance(ctx context.Context, databaseID, instanceID str
 		return fmt.Errorf("failed to delete stored instance status: %w", err)
 	}
 
-	return nil
+	return s.DeleteInstanceSpec(ctx, databaseID, instanceID)
 }
 
 func (s *Service) UpdateInstanceStatus(
@@ -490,6 +516,112 @@ func (s *Service) PopulateSpecDefaults(ctx context.Context, spec *Spec) error {
 	return nil
 }
 
+func (s *Service) ReconcileInstanceSpec(ctx context.Context, spec *InstanceSpec) (*InstanceSpec, error) {
+	if s.cfg.HostID != spec.HostID {
+		return nil, fmt.Errorf("this instance belongs to another host - this host='%s', instance host='%s'", s.cfg.HostID, spec.HostID)
+	}
+	current, err := s.store.InstanceSpec.
+		GetByKey(spec.DatabaseID, spec.InstanceID).
+		Exec(ctx)
+	switch {
+	case errors.Is(err, storage.ErrNotFound):
+		// Nothing to copy from, so we'll just generate any newly-needed random
+		// ports.
+	case err != nil:
+		return nil, fmt.Errorf("failed to get current spec for instance '%s': %w", spec.InstanceID, err)
+	default:
+		spec.CopySettingsFrom(current.Spec)
+	}
+
+	if spec.Port != nil && *spec.Port == 0 {
+		port, err := s.portsSvc.AllocatePort(ctx, spec.HostID)
+		if err != nil {
+			return nil, fmt.Errorf("failed to allocate port: %w", err)
+		}
+		spec.Port = utils.PointerTo(port)
+	}
+	if spec.PatroniPort != nil && *spec.PatroniPort == 0 {
+		port, err := s.portsSvc.AllocatePort(ctx, spec.HostID)
+		if err != nil {
+			return nil, fmt.Errorf("failed to allocate patroni port: %w", err)
+		}
+		spec.PatroniPort = utils.PointerTo(port)
+	}
+
+	err = s.store.InstanceSpec.
+		Put(&StoredInstanceSpec{Spec: spec}).
+		Exec(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to persist updated instance spec: %w", err)
+	}
+
+	if current != nil {
+		if portShouldBeReleased(current.Spec.Port, spec.Port) {
+			err := s.portsSvc.ReleasePortIfDefined(ctx, spec.HostID, current.Spec.Port)
+			if err != nil {
+				return nil, fmt.Errorf("failed to release previous port: %w", err)
+			}
+		}
+		if portShouldBeReleased(current.Spec.PatroniPort, spec.PatroniPort) {
+			err := s.portsSvc.ReleasePortIfDefined(ctx, spec.HostID, current.Spec.PatroniPort)
+			if err != nil {
+				return nil, fmt.Errorf("failed to release previous patroni port: %w", err)
+			}
+		}
+	}
+
+	return spec, nil
+}
+
+func (s *Service) DeleteInstanceSpec(ctx context.Context, databaseID, instanceID string) error {
+	spec, err := s.store.InstanceSpec.
+		GetByKey(databaseID, instanceID).
+		Exec(ctx)
+	if errors.Is(err, storage.ErrNotFound) {
+		// Spec has already been deleted
+		return nil
+	} else if err != nil {
+		return fmt.Errorf("failed to check if instance spec exists: %w", err)
+	}
+
+	if err := s.releaseInstancePorts(ctx, spec.Spec); err != nil {
+		return err
+	}
+
+	_, err = s.store.InstanceSpec.
+		DeleteByKey(databaseID, instanceID).
+		Exec(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to delete instance spec: %w", err)
+	}
+
+	return nil
+}
+
+func (s *Service) releaseInstancePorts(ctx context.Context, spec *InstanceSpec) error {
+	err := s.portsSvc.ReleasePortIfDefined(ctx, spec.HostID, spec.Port, spec.PatroniPort)
+	if err != nil {
+		return fmt.Errorf("failed to release ports for instance '%s': %w", spec.InstanceID, err)
+	}
+
+	return nil
+}
+
+func portShouldBeReleased(current *int, new *int) bool {
+	if current == nil || *current == 0 {
+		// we didn't previously have an assigned port
+		return false
+	}
+	if new == nil || *current != *new {
+		// we had a previously assigned port and now the port is either nil or
+		// different
+		return true
+	}
+
+	// the current and new ports are equal, so it should not be released.
+	return false
+}
+
 func ValidateChangedSpec(current, updated *Spec) error {
 	var errs []error
 

server/internal/database/spec.go

@@ -508,6 +508,25 @@ type InstanceSpec struct {
 	InPlaceRestore   bool                `json:"in_place_restore,omitempty"`
 }
 
+func (s *InstanceSpec) CopySettingsFrom(current *InstanceSpec) {
+	s.Port = reconcilePort(current.Port, s.Port)
+	s.PatroniPort = reconcilePort(current.PatroniPort, s.PatroniPort)
+}
+
+func reconcilePort(current, new *int) *int {
+	if new == nil || *new != 0 {
+		// no action needed if the new port is unexposed or explicitly set
+		return new
+	}
+	if current != nil && *current != 0 {
+		// we've already assigned a stable random port here
+		return utils.PointerTo(*current) // create new pointer
+	}
+
+	// return 0 to signal that we need to assign a new random port
+	return utils.PointerTo(0)
+}
+
 type InstanceSpecChange struct {
 	Previous *InstanceSpec
 	Current  *InstanceSpec