ref(deisctl): switch to async interface for backends

Author: Gabriel Monroy

deisctl/backend/backend.go

@@ -1,11 +1,13 @@
 package backend
 
+import "sync"
+
 // Backend interface is used to interact with the cluster control plane
 type Backend interface {
-	Create([]string) error
-	Destroy([]string) error
-	Start([]string) error
-	Stop([]string) error
+	Create([]string, *sync.WaitGroup, chan string, chan error)
+	Destroy([]string, *sync.WaitGroup, chan string, chan error)
+	Start([]string, *sync.WaitGroup, chan string, chan error)
+	Stop([]string, *sync.WaitGroup, chan string, chan error)
 	Scale(string, int) error
 	ListUnits() error
 	ListUnitFiles() error

deisctl/backend/fleet/create.go

@@ -3,46 +3,58 @@ package fleet
 import (
 	"fmt"
 	"strings"
+	"sync"
 
 	"github.com/coreos/fleet/job"
 	"github.com/coreos/fleet/schema"
 	"github.com/coreos/fleet/unit"
 )
 
 // Create schedules a new unit for the given component
-// and blocks until the unit is loaded
-func (c *FleetClient) Create(targets []string) error {
+func (c *FleetClient) Create(targets []string, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
+
 	units := make([]*schema.Unit, len(targets))
-	desiredState := string(job.JobStateLoaded)
+
 	for i, target := range targets {
 		unitName, unitFile, err := c.createUnitFile(target)
 		if err != nil {
-			return err
+			errchan <- err
+			return
 		}
 		units[i] = &schema.Unit{
 			Name:    unitName,
 			Options: schema.MapUnitFileToSchemaUnitOptions(unitFile),
 		}
 	}
+
 	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)
-			}
-		}
-		if err := c.Fleet.SetUnitTargetState(unit.Name, desiredState); err != nil {
-			return err
-		}
+		wg.Add(1)
+		go doCreate(c, unit, wg, outchan, errchan)
 	}
-	for _, unit := range units {
-		outchan, errchan := waitForUnitStates([]string{unit.Name}, desiredState)
-		if err := printUnitState(unit.Name, outchan, errchan); err != nil {
-			return err
+}
+
+func doCreate(c *FleetClient, unit *schema.Unit, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
+	defer wg.Done()
+
+	// create unit definition
+	if err := c.Fleet.CreateUnit(unit); err != nil {
+		// ignore units that already exist
+		if err.Error() != "job already exists" {
+			errchan <- err
+			return
 		}
 	}
-	return nil
+
+	desiredState := string(job.JobStateLoaded)
+	out := fmt.Sprintf("\033[0;33m%v:\033[0m loaded                                 \r", unit.Name)
+
+	// schedule the unit
+	if err := c.Fleet.SetUnitTargetState(unit.Name, desiredState); err != nil {
+		errchan <- err
+		return
+	}
+
+	outchan <- out
 }
 
 func (c *FleetClient) createUnitFile(target string) (unitName string, uf *unit.UnitFile, err error) {

deisctl/backend/fleet/destroy.go

@@ -3,72 +3,62 @@ package fleet
 import (
 	"fmt"
 	"strings"
-
-	"github.com/coreos/fleet/job"
+	"sync"
+	"time"
 )
 
 // Destroy units for a given target
-func (c *FleetClient) Destroy(targets []string) error {
+func (c *FleetClient) Destroy(targets []string, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
 	for _, target := range targets {
-		// check if the unit exists
-		_, err := c.Units(target)
-		if err != nil {
-			if strings.Contains(err.Error(), "could not find unit") {
-				return 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 && !strings.Contains(err.Error(), "could not find unit") {
-			return err
-		}
+		wg.Add(1)
+		go doDestroy(c, target, wg, outchan, errchan)
 	}
-	return nil
+	return
 }
 
-func (c *FleetClient) destroyServiceUnit(component string, num int) error {
-	name, err := formatUnitName(component, num)
+func doDestroy(c *FleetClient, target string, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
+	defer wg.Done()
+
+	// prepare string representation
+	component, num, err := splitTarget(target)
 	if err != nil {
-		return err
+		errchan <- err
+		return
 	}
-	desiredState := string(job.JobStateInactive)
-	err = c.Fleet.SetUnitTargetState(name, desiredState)
+	name, err := formatUnitName(component, num)
 	if err != nil {
-		return err
-	}
-	outchan, errchan := waitForUnitStates([]string{name}, desiredState)
-	if err := printUnitState(name, outchan, errchan); err != nil {
-		return err
+		errchan <- err
+		return
 	}
-	if err = c.Fleet.DestroyUnit(name); err != nil {
-		return fmt.Errorf("failed destroying job %s: %v", name, err)
-	}
-	return nil
-}
+	destroyed := fmt.Sprintf("\033[0;33m%v:\033[0m destroyed                                 \r", name)
 
-func (c *FleetClient) destroyDataUnit(component string) error {
-	name, err := formatUnitName(component, 0)
-	desiredState := string(job.JobStateInactive)
+	// bail early if unit doesn't exist
+	_, err = c.Units(name)
 	if err != nil {
-		return err
-	}
-	if err := c.Fleet.SetUnitTargetState(name, desiredState); err != nil {
-		return err
+		if strings.Contains(err.Error(), "could not find unit") {
+			outchan <- destroyed
+		}
+		return
 	}
-	outchan, errchan := waitForUnitStates([]string{name}, desiredState)
-	if err := printUnitState(name, outchan, errchan); err != nil {
-		return err
+
+	// otherwise destroy it
+	if err = c.Fleet.DestroyUnit(name); err != nil {
+		// ignore already destroyed units
+		if !strings.Contains(err.Error(), "could not find unit") {
+			errchan <- err
+			return
+		}
 	}
-	if err := c.Fleet.DestroyUnit(name); err != nil {
-		return fmt.Errorf("failed destroying job %s: %v", name, err)
+
+	// loop until actually destroyed
+	for {
+		_, err = c.Units(name)
+		if err != nil {
+			if strings.Contains(err.Error(), "could not find unit") {
+				outchan <- destroyed
+				return
+			}
+		}
+		time.Sleep(250 * time.Millisecond)
 	}
-	return nil
 }

deisctl/backend/fleet/scale.go

@@ -5,10 +5,16 @@ import (
 	"math"
 	"strconv"
 	"strings"
+	"sync"
 )
 
 // Scale creates or destroys units to match the desired number
 func (c *FleetClient) Scale(component string, requested int) error {
+
+	outchan := make(chan string)
+	errchan := make(chan error)
+	var wg sync.WaitGroup
+
 	if requested < 0 {
 		return errors.New("cannot scale below 0")
 	}
@@ -22,27 +28,29 @@ func (c *FleetClient) Scale(component string, requested int) error {
 	}
 
 	timesToScale := int(math.Abs(float64(requested - len(components))))
+	if timesToScale == 0 {
+		return nil
+	}
 	if requested-len(components) > 0 {
-		return scaleUp(c, component, len(components), timesToScale)
-	} else {
-		return scaleDown(c, component, len(components), timesToScale)
+		return scaleUp(c, component, len(components), timesToScale, &wg, outchan, errchan)
 	}
+	return scaleDown(c, component, len(components), timesToScale, &wg, outchan, errchan)
 }
 
-func scaleUp(c *FleetClient, component string, numExistingContainers, numTimesToScale int) error {
+func scaleUp(c *FleetClient, component string, numExistingContainers, numTimesToScale int,
+	wg *sync.WaitGroup, outchan chan string, errchan chan error) error {
 	for i := 0; i < numTimesToScale; i++ {
-		if err := c.Create([]string{component + "@" + strconv.Itoa(numExistingContainers+i+1)}); err != nil {
-			return err
-		}
+		target := component + "@" + strconv.Itoa(numExistingContainers+i+1)
+		c.Create([]string{target}, wg, outchan, errchan)
 	}
 	return nil
 }
 
-func scaleDown(c *FleetClient, component string, numExistingContainers, numTimesToScale int) error {
+func scaleDown(c *FleetClient, component string, numExistingContainers, numTimesToScale int,
+	wg *sync.WaitGroup, outchan chan string, errchan chan error) error {
 	for i := 0; i < numTimesToScale; i++ {
-		if err := c.Destroy([]string{component + "@" + strconv.Itoa(numExistingContainers-i)}); err != nil {
-			return err
-		}
+		target := component + "@" + strconv.Itoa(numExistingContainers-i)
+		c.Destroy([]string{target}, wg, outchan, errchan)
 	}
 	return nil
 }

deisctl/backend/fleet/start.go

@@ -1,45 +1,87 @@
 package fleet
 
 import (
+	"fmt"
 	"strings"
+	"sync"
+	"time"
 
-	"github.com/coreos/fleet/job"
 	"github.com/coreos/fleet/schema"
 )
 
-// Start launches target units and blocks until active
-func (c *FleetClient) Start(targets []string) error {
-	units := make([][]string, len(targets))
-	for i, target := range targets {
-		u, err := c.Units(target)
+// Start units and wait for their desiredState
+func (c *FleetClient) Start(targets []string, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
+	for _, target := range targets {
+		wg.Add(1)
+		go doStart(c, target, wg, outchan, errchan)
+	}
+	return
+}
+
+func doStart(c *FleetClient, target string, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
+	defer wg.Done()
+
+	// prepare string representation
+	component, num, err := splitTarget(target)
+	if err != nil {
+		errchan <- err
+		return
+	}
+	name, err := formatUnitName(component, num)
+	if err != nil {
+		errchan <- err
+		return
+	}
+
+	requestState := "launched"
+	var desiredState string
+	if strings.Contains(name, "-data.service") {
+		desiredState = "exited"
+	} else {
+		desiredState = "running"
+	}
+
+	if err := c.Fleet.SetUnitTargetState(name, requestState); err != nil {
+		errchan <- err
+		return
+	}
+
+	// start with the likely subState to avoid sending it across the channel
+	lastSubState := "dead"
+
+	for {
+		// poll for unit states
+		states, err := c.Fleet.UnitStates()
 		if err != nil {
-			return err
+			errchan <- err
+			return
 		}
-		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
+
+		// FIXME: fleet UnitStates API forces us to iterate for now
+		var currentState *schema.UnitState
+		for _, s := range states {
+			if name == s.Name {
+				currentState = s
+				break
 			}
 		}
-	}
-	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
-			}
+		if currentState == nil {
+			errchan <- fmt.Errorf("could not find unit: %v", name)
+			return
 		}
+
+		// if subState changed, send it across the output channel
+		if lastSubState != currentState.SystemdSubState {
+			outchan <- fmt.Sprintf("\033[0;33m%v:\033[0m %v/%v                                 \r",
+				name, currentState.SystemdActiveState, currentState.SystemdSubState)
+		}
+
+		// break when desired state is reached
+		if currentState.SystemdSubState == desiredState {
+			return
+		}
+
+		lastSubState = currentState.SystemdSubState
+		time.Sleep(250 * time.Millisecond)
 	}
-	return nil
 }