Merge pull request #53 from mboersma/parallel-start-containers

feat(start/stop): allow starting or stopping > 1 unit at a time

Author: mboersma

client/api.go

@@ -2,10 +2,10 @@ package client
 
 // Client interface used to interact with the cluster control plane
 type Client interface {
-	Create(string) error
-	Destroy(string) error
-	Start(string) error
-	Stop(string) error
+	Create([]string) error
+	Destroy([]string) error
+	Start([]string) error
+	Stop([]string) error
 	Scale(string, int) error
 	ListUnits() error
 	ListUnitFiles() error

client/create.go

@@ -11,37 +11,37 @@ import (
 
 // Create schedules a new unit for the given component
 // and blocks until the unit is loaded
-func (c *FleetClient) Create(target string) (err error) {
-	var (
-		unitName string
-		unitFile *unit.UnitFile
-	)
-	// create unit file
-	unitName, unitFile, err = c.createUnitFile(target)
-	if err != nil {
-		return err
-	}
-	// define unit
-	u := &schema.Unit{
-		Name:    unitName,
-		Options: schema.MapUnitFileToSchemaUnitOptions(unitFile),
-	}
-	// schedule unit
-	if err := c.Fleet.CreateUnit(u); err != nil {
-		// ignore units that already exist
-		if err.Error() != "job already exists" {
-			return fmt.Errorf("failed creating job %s: %v", unitName, err)
+func (c *FleetClient) Create(targets []string) error {
+	units := make([]*schema.Unit, len(targets))
+	for i, target := range targets {
+		unitName, unitFile, err := c.createUnitFile(target)
+		if err != nil {
+			return err
+		}
+		units[i] = &schema.Unit{
+			Name:    unitName,
+			Options: schema.MapUnitFileToSchemaUnitOptions(unitFile),
 		}
 	}
-	desiredState := string(job.JobStateLoaded)
-	err = c.Fleet.SetUnitTargetState(unitName, desiredState)
-	if err != nil {
-		return err
+	for _, unit := range units {
+		// schedule unit
+		if err := c.Fleet.CreateUnit(unit); err != nil {
+			// ignore units that already exist
+			if err.Error() != "job already exists" {
+				return fmt.Errorf("failed creating job %s: %v", unit.Name, err)
+			}
+		}
+		desiredState := string(job.JobStateLoaded)
+		if err := c.Fleet.SetUnitTargetState(unit.Name, desiredState); err != nil {
+			return err
+		}
 	}
-	outchan, errchan := waitForUnitStates([]string{unitName}, desiredState)
-	err = printUnitState(unitName, outchan, errchan)
-	if err != nil {
-		return err
+	for _, unit := range units {
+		desiredState := string(job.JobStateLoaded)
+		outchan, errchan := waitForUnitStates([]string{unit.Name}, desiredState)
+		if err := printUnitState(unit.Name, outchan, errchan); err != nil {
+			return err
+		}
 	}
 	return nil
 }

client/destroy.go

@@ -8,29 +8,26 @@ import (
 )
 
 // Destroy units for a given target
-func (c *FleetClient) Destroy(target string) (err error) {
-	// check if the unit exists
-	units, err := c.Units(target)
-	if err != nil {
-		return err
-	}
-	component, num, err := splitTarget(target)
-	if err != nil {
-		return
-	}
-	// if no number is specified, destroy ALL THE UNITS!
-	if num == 0 {
-		num = len(units)
-	}
-	if strings.HasSuffix(component, "-data") {
-		err = c.destroyDataUnit(component)
-	} else {
-		err = c.destroyServiceUnit(component, num)
-	}
-	if err != nil {
-		return err
+func (c *FleetClient) Destroy(targets []string) error {
+	for _, target := range targets {
+		// check if the unit exists
+		if _, err := c.Units(target); err != nil {
+			return err
+		}
+		component, num, err := splitTarget(target)
+		if err != nil {
+			return err
+		}
+		if strings.HasSuffix(component, "-data") {
+			err = c.destroyDataUnit(component)
+		} else {
+			err = c.destroyServiceUnit(component, num)
+		}
+		if err != nil {
+			return err
+		}
 	}
-	return
+	return nil
 }
 
 func (c *FleetClient) destroyServiceUnit(component string, num int) (err error) {

client/scale.go

@@ -17,8 +17,7 @@ func (c *FleetClient) Scale(component string, requested int) (err error) {
 			if err != nil {
 				return err
 			}
-			err = c.Create(component + "@" + strconv.Itoa(num))
-			if err != nil {
+			if err = c.Create([]string{component + "@" + strconv.Itoa(num)}); err != nil {
 				return err
 			}
 			continue
@@ -28,8 +27,7 @@ func (c *FleetClient) Scale(component string, requested int) (err error) {
 			if err != nil {
 				return err
 			}
-			err = c.Destroy(component + "@" + strconv.Itoa(num))
-			if err != nil {
+			if err = c.Destroy([]string{component + "@" + strconv.Itoa(num)}); err != nil {
 				return err
 			}
 			continue

client/start.go

@@ -8,29 +8,37 @@ import (
 )
 
 // Start launches target units and blocks until active
-func (c *FleetClient) Start(target string) (err error) {
-	units, err := c.Units(target)
-	if err != nil {
-		return
-	}
-	desiredState := string(job.JobStateLaunched)
-	for _, name := range units {
-		err = c.Fleet.SetUnitTargetState(name, desiredState)
+func (c *FleetClient) Start(targets []string) error {
+	units := make([][]string, len(targets))
+	for i, target := range targets {
+		u, err := c.Units(target)
 		if err != nil {
 			return err
 		}
-		// wait for systemd to tell us that it's running, not fleet
-		var errchan chan error
-		var outchan chan *schema.Unit
-		// data containers are special snowflakes who just exit
-		if strings.Contains(name, "-data.service") {
-			outchan, errchan = waitForUnitSubStates(units, "exited")
-		} else {
-			outchan, errchan = waitForUnitSubStates(units, "running")
+		units[i] = u
+	}
+	desiredState := string(job.JobStateLaunched)
+	for _, names := range units {
+		for _, name := range names {
+			if err := c.Fleet.SetUnitTargetState(name, desiredState); err != nil {
+				return err
+			}
 		}
-		err = printUnitSubState(name, outchan, errchan)
-		if err != nil {
-			return err
+	}
+	var errchan chan error
+	var outchan chan *schema.Unit
+	for _, names := range units {
+		for _, name := range names {
+			// wait for systemd to tell us that it's running, not fleet
+			// data containers are special snowflakes who just exit
+			if strings.Contains(name, "-data.service") {
+				outchan, errchan = waitForUnitSubStates(names, "exited")
+			} else {
+				outchan, errchan = waitForUnitSubStates(names, "running")
+			}
+			if err := printUnitSubState(name, outchan, errchan); err != nil {
+				return err
+			}
 		}
 	}
 	return nil

client/stop.go

@@ -1,24 +1,38 @@
 package client
 
-import "github.com/coreos/fleet/job"
+import (
+	"github.com/coreos/fleet/job"
+	"github.com/coreos/fleet/schema"
+)
 
 // Stop sets target units to inactive and blocks until complete
-func (c *FleetClient) Stop(target string) (err error) {
-	units, err := c.Units(target)
-	if err != nil {
-		return
-	}
-	desiredState := string(job.JobStateLoaded)
-	for _, name := range units {
-		err = c.Fleet.SetUnitTargetState(name, desiredState)
+func (c *FleetClient) Stop(targets []string) error {
+	units := make([][]string, len(targets))
+	for i, target := range targets {
+		u, err := c.Units(target)
 		if err != nil {
 			return err
 		}
-		outchan, errchan := waitForUnitStates(units, desiredState)
-		err = printUnitState(name, outchan, errchan)
-		if err != nil {
-			return err
+		units[i] = u
+	}
+	desiredState := string(job.JobStateLoaded)
+	for _, names := range units {
+		for _, name := range names {
+			if err := c.Fleet.SetUnitTargetState(name, desiredState); err != nil {
+				return err
+			}
+		}
+	}
+	var errchan chan error
+	var outchan chan *schema.Unit
+	for _, names := range units {
+		for _, name := range names {
+			outchan, errchan = waitForUnitSubStates(names, "dead")
+			if err := printUnitSubState(name, outchan, errchan); err != nil {
+				return err
+			}
 		}
 	}
 	return nil
+
 }