Adds flags for configuring instance port pool.

This commit adds new flags for configuring the instance port pool that
Empire uses to allocate instance ports to applications and processes.

cmd/empire/factories.go

@@ -37,7 +37,19 @@ import (
 // DB ===================================
 
 func newDB(c *Context) (*empire.DB, error) {
-	return empire.OpenDB(c.String(FlagDB))
+	db, err := empire.OpenDB(c.String(FlagDB))
+	if err != nil {
+		return nil, err
+	}
+
+	db.Schema = &empire.Schema{
+		InstancePortPool: &empire.InstancePortPool{
+			Start: uint(c.Int(FlagInstancePortPoolStart)),
+			End:   uint(c.Int(FlagInstancePortPoolEnd)),
+		},
+	}
+
+	return db, nil
 }
 
 // Empire ===============================

cmd/empire/main.go

@@ -71,6 +71,9 @@ const (
 	FlagEC2SubnetsPrivate = "ec2.subnets.private"
 	FlagEC2SubnetsPublic  = "ec2.subnets.public"
 
+	FlagInstancePortPoolStart = "instance-port-pool.start"
+	FlagInstancePortPoolEnd   = "instance-port-pool.end"
+
 	FlagRoute53InternalZoneID = "route53.zoneid.internal"
 
 	FlagCloudFormationStackNameTemplate = "cloudformation.stack-name-template"
@@ -359,6 +362,18 @@ var EmpireFlags = []cli.Flag{
 		Usage:  "The comma separated public subnet ids",
 		EnvVar: "EMPIRE_EC2_SUBNETS_PUBLIC",
 	},
+	cli.IntFlag{
+		Name:   FlagInstancePortPoolStart,
+		Value:  empire.DefaultInstancePortPoolStart,
+		Usage:  "The start of the range of instance ports to allocate from.",
+		EnvVar: "EMPIRE_INSTANCE_PORT_POOL_START",
+	},
+	cli.IntFlag{
+		Name:   FlagInstancePortPoolEnd,
+		Value:  empire.DefaultInstancePortPoolEnd,
+		Usage:  "The end of the range of instance ports to allocate from.",
+		EnvVar: "EMPIRE_INSTANCE_PORT_POOL_END",
+	},
 	cli.StringFlag{
 		Name:   FlagSecret,
 		Value:  "<change this>",

db.go

@@ -27,6 +27,10 @@ func (e *IncompatibleSchemaError) Error() string {
 
 // DB wraps a gorm.DB and provides the datastore layer for Empire.
 type DB struct {
+	// Schema is the Schema instance that will be used to migrate the
+	// database to the latest schema. The zero value is DefaultSchema.
+	Schema *Schema
+
 	*gorm.DB
 
 	uri string
@@ -72,7 +76,11 @@ func NewDB(conn *sql.DB) (*DB, error) {
 
 // MigrateUp migrates the database to the latest version of the schema.
 func (db *DB) MigrateUp() error {
-	return db.migrator.Exec(migrate.Up, migrations...)
+	return db.migrator.Exec(migrate.Up, db.migrations()...)
+}
+
+func (db *DB) migrations() []migrate.Migration {
+	return db.schema().migrations()
 }
 
 // Reset resets the database to a pristine state.
@@ -87,7 +95,7 @@ func (db *DB) Reset() error {
 	exec(`TRUNCATE TABLE apps CASCADE`)
 	exec(`TRUNCATE TABLE ports CASCADE`)
 	exec(`TRUNCATE TABLE slugs CASCADE`)
-	exec(`INSERT INTO ports (port) (SELECT generate_series(9000,10000))`)
+	exec(`UPDATE ports SET app_id = NULL`)
 
 	return err
 }
@@ -113,7 +121,7 @@ func (db *DB) CheckSchemaVersion() error {
 		return fmt.Errorf("error fetching schema version: %v", err)
 	}
 
-	expectedSchemaVersion := latestSchema()
+	expectedSchemaVersion := db.schema().latestSchema()
 	if schemaVersion != expectedSchemaVersion {
 		return &IncompatibleSchemaError{
 			SchemaVersion:         schemaVersion,
@@ -137,6 +145,13 @@ func (db *DB) Debug() {
 	db.DB = db.DB.Debug()
 }
 
+func (db *DB) schema() *Schema {
+	if db.Schema == nil {
+		return DefaultSchema
+	}
+	return db.Schema
+}
+
 // scope is an interface that scopes a gorm.DB. Scopes are used in
 // ThingsFirst and ThingsAll methods on the store for filtering/querying.
 type scope interface {

internal/migrate/migrate.go

@@ -59,13 +59,13 @@ type Migration struct {
 	Down func(tx *sql.Tx) error
 }
 
-// byID implements the sort.Interface interface for sorting migrations by
+// ByID implements the sort.Interface interface for sorting migrations by
 // ID.
-type byID []Migration
+type ByID []Migration
 
-func (m byID) Len() int           { return len(m) }
-func (m byID) Less(i, j int) bool { return m[i].ID < m[j].ID }
-func (m byID) Swap(i, j int)      { m[i], m[j] = m[j], m[i] }
+func (m ByID) Len() int           { return len(m) }
+func (m ByID) Less(i, j int) bool { return m[i].ID < m[j].ID }
+func (m ByID) Swap(i, j int)      { m[i], m[j] = m[j], m[i] }
 
 // Migrator performs migrations.
 type Migrator struct {
@@ -277,16 +277,16 @@ func Queries(queries []string) func(*sql.Tx) error {
 // When the direction is "Up", the migrations will be sorted by ID ascending.
 // When the direction is "Down", the migrations will be sorted by ID descending.
 func sortMigrations(dir MigrationDirection, migrations []Migration) []Migration {
-	var m byID
+	var m ByID
 	for _, migration := range migrations {
 		m = append(m, migration)
 	}
 
 	switch dir {
 	case Up:
-		sort.Sort(byID(m))
+		sort.Sort(ByID(m))
 	default:
-		sort.Sort(sort.Reverse(byID(m)))
+		sort.Sort(sort.Reverse(ByID(m)))
 	}
 
 	return m

migrations.go

@@ -4,13 +4,78 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
+	"sort"
 
 	"github.com/lib/pq/hstore"
 	"github.com/remind101/empire/internal/migrate"
 	"github.com/remind101/empire/pkg/constraints"
 	"github.com/remind101/empire/procfile"
 )
 
+const (
+	DefaultInstancePortPoolStart = 9000
+	DefaultInstancePortPoolEnd   = 10000
+)
+
+// DefaultSchema is the default Schema that will be used to migrate the database
+// if none is provided.
+var DefaultSchema = &Schema{
+	InstancePortPool: &InstancePortPool{
+		Start: DefaultInstancePortPoolStart,
+		End:   DefaultInstancePortPoolEnd,
+	},
+}
+
+type InstancePortPool struct {
+	Start, End uint
+}
+
+type Schema struct {
+	// For legacy ELB's (not ALB) Empire manages a pool of host ports to
+	// ensure that all applications have a unique port on the EC2 instance.
+	// This option specifies the beginning and end of that range.
+	InstancePortPool *InstancePortPool
+}
+
+// latestSchema returns the schema version that this version of Empire should be
+// using.
+func (s *Schema) latestSchema() int {
+	migrations := s.migrations()
+	return migrations[len(migrations)-1].ID
+}
+
+func (s *Schema) migrations() []migrate.Migration {
+	ports := migrate.Migration{
+		ID: 4,
+		Up: func(tx *sql.Tx) error {
+			const table = `CREATE TABLE ports (
+  id uuid NOT NULL DEFAULT uuid_generate_v4() primary key,
+  port integer,
+  app_id uuid references apps(id) ON DELETE SET NULL
+)`
+			if _, err := tx.Exec(table); err != nil {
+				return fmt.Errorf("unable to create ports table: %v", err)
+			}
+
+			ports := fmt.Sprintf(`INSERT INTO ports (port) (SELECT generate_series(%d,%d))`, s.InstancePortPool.Start, s.InstancePortPool.End)
+
+			if _, err := tx.Exec(ports); err != nil {
+				return fmt.Errorf("unable to generate a series of instance ports: %v", err)
+			}
+
+			return nil
+		},
+		Down: migrate.Queries([]string{
+			`DROP TABLE ports CASCADE`,
+		}),
+	}
+
+	migrations := migrate.ByID(append(migrations, ports))
+	sort.Sort(migrations)
+
+	return migrations
+}
+
 var migrations = []migrate.Migration{
 	{
 		ID: 1,
@@ -128,21 +193,6 @@ var migrations = []migrate.Migration{
 )`,
 		}),
 	},
-	{
-		ID: 4,
-		Up: migrate.Queries([]string{
-			`CREATE TABLE ports (
-  id uuid NOT NULL DEFAULT uuid_generate_v4() primary key,
-  port integer,
-  app_id uuid references apps(id) ON DELETE SET NULL
-)`,
-			`-- Insert 1000 ports
-INSERT INTO ports (port) (SELECT generate_series(9000,10000))`,
-		}),
-		Down: migrate.Queries([]string{
-			`DROP TABLE ports CASCADE`,
-		}),
-	},
 	{
 		ID: 5,
 		Up: migrate.Queries([]string{
@@ -630,9 +680,3 @@ ALTER TABLE apps ADD COLUMN exposure TEXT NOT NULL default 'private'`,
 		}),
 	},
 }
-
-// latestSchema returns the schema version that this version of Empire should be
-// using.
-func latestSchema() int {
-	return migrations[len(migrations)-1].ID
-}

migrations_test.go

@@ -16,6 +16,8 @@ func TestMigrations(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	migrations := DefaultSchema.migrations()
+
 	err = db.migrator.Exec(migrate.Up, migrations...)
 	assert.NoError(t, err)
 
@@ -30,13 +32,13 @@ func TestMigrations(t *testing.T) {
 }
 
 func TestLatestSchema(t *testing.T) {
-	assert.Equal(t, 20, latestSchema())
+	assert.Equal(t, 20, DefaultSchema.latestSchema())
 }
 
 func TestNoDuplicateMigrations(t *testing.T) {
 	visited := make(map[int]bool)
 	expectedID := 1
-	for _, m := range migrations {
+	for _, m := range DefaultSchema.migrations() {
 		if visited[m.ID] {
 			t.Fatalf("Migration %d appears more than once", m.ID)
 		}