Merge pull request #4025 from technosophos/issue/3971

fix(deisctl): Switch from channels to io.Writers.

Author: technosophos

deisctl/backend/backend.go

@@ -1,14 +1,17 @@
 package backend
 
-import "sync"
+import (
+	"io"
+	"sync"
+)
 
 // Backend interface is used to interact with the cluster control plane
 type Backend interface {
-	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, *sync.WaitGroup, chan string, chan error)
+	Create([]string, *sync.WaitGroup, io.Writer, io.Writer)
+	Destroy([]string, *sync.WaitGroup, io.Writer, io.Writer)
+	Start([]string, *sync.WaitGroup, io.Writer, io.Writer)
+	Stop([]string, *sync.WaitGroup, io.Writer, io.Writer)
+	Scale(string, int, *sync.WaitGroup, io.Writer, io.Writer)
 	SSH(string) error
 	ListUnits() error
 	ListUnitFiles() error

deisctl/backend/fleet/create.go

@@ -2,25 +2,28 @@ package fleet
 
 import (
 	"fmt"
+	"io"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/coreos/fleet/job"
 	"github.com/coreos/fleet/schema"
 	"github.com/coreos/fleet/unit"
+
+	"github.com/deis/deis/pkg/prettyprint"
 )
 
 // Create schedules unit files for the given components.
 func (c *FleetClient) Create(
-	targets []string, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
+	targets []string, wg *sync.WaitGroup, out, ew io.Writer) {
 
 	units := make([]*schema.Unit, len(targets))
 
 	for i, target := range targets {
 		unitName, unitFile, err := c.createUnitFile(target)
 		if err != nil {
-			errchan <- err
+			fmt.Fprintf(ew, "Error creating: %s\n", err)
 			return
 		}
 		units[i] = &schema.Unit{
@@ -31,28 +34,29 @@ func (c *FleetClient) Create(
 
 	for _, unit := range units {
 		wg.Add(1)
-		go doCreate(c, unit, wg, outchan, errchan)
+		go doCreate(c, unit, wg, out, ew)
 	}
 }
 
-func doCreate(c *FleetClient, unit *schema.Unit, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
+func doCreate(c *FleetClient, unit *schema.Unit, wg *sync.WaitGroup, out, ew io.Writer) {
 	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
+			fmt.Fprintln(ew, err.Error())
 			return
 		}
 	}
 
 	desiredState := string(job.JobStateLoaded)
-	out := fmt.Sprintf("\033[0;33m%v:\033[0m loaded                                 \r", unit.Name)
+	tpl := prettyprint.Colorize("{{.Yellow}}%v:{{.Default}} loaded")
+	msg := fmt.Sprintf(tpl, unit.Name)
 
 	// schedule the unit
 	if err := c.Fleet.SetUnitTargetState(unit.Name, desiredState); err != nil {
-		errchan <- err
+		fmt.Fprintln(ew, err)
 		return
 	}
 
@@ -62,7 +66,7 @@ outerLoop:
 		time.Sleep(250 * time.Millisecond)
 		unitStates, err := c.Fleet.UnitStates()
 		if err != nil {
-			errchan <- err
+			fmt.Fprintln(ew, err)
 		}
 		for _, us := range unitStates {
 			if strings.HasPrefix(us.Name, unit.Name) {
@@ -71,7 +75,7 @@ outerLoop:
 		}
 	}
 
-	outchan <- out
+	fmt.Fprintln(out, msg)
 }
 
 func (c *FleetClient) createUnitFile(target string) (unitName string, uf *unit.UnitFile, err error) {

deisctl/backend/fleet/create_test.go

@@ -1,7 +1,6 @@
 package fleet
 
 import (
-	"fmt"
 	"io/ioutil"
 	"path"
 	"sync"
@@ -34,19 +33,14 @@ func TestCreate(t *testing.T) {
 	c := &FleetClient{templatePaths: []string{name}, Fleet: &testFleetClient}
 
 	var errOutput string
-	outchan := make(chan string)
-	errchan := make(chan error)
 	var wg sync.WaitGroup
 
 	logMutex := sync.Mutex{}
 
-	go logState(outchan, errchan, &errOutput, &logMutex)
-
-	c.Create([]string{"controller", "builder", "router@1"}, &wg, outchan, errchan)
+	se := newOutErr()
+	c.Create([]string{"controller", "builder", "router@1"}, &wg, se.out, se.ew)
 
 	wg.Wait()
-	close(errchan)
-	close(outchan)
 
 	logMutex.Lock()
 	if errOutput != "" {
@@ -68,7 +62,7 @@ func TestCreate(t *testing.T) {
 		}
 
 		if !found {
-			t.Error(fmt.Errorf("Expected Unit %s not found in Unit States", expectedUnit))
+			t.Errorf("Expected Unit %s not found in Unit States", expectedUnit)
 		}
 	}
 }

deisctl/backend/fleet/destroy.go

@@ -2,42 +2,46 @@ package fleet
 
 import (
 	"fmt"
+	"io"
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/deis/deis/pkg/prettyprint"
 )
 
 // Destroy units for a given target
-func (c *FleetClient) Destroy(targets []string, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
+func (c *FleetClient) Destroy(targets []string, wg *sync.WaitGroup, out, ew io.Writer) {
 	// expand @* targets
 	expandedTargets, err := c.expandTargets(targets)
 	if err != nil {
-		errchan <- err
+		fmt.Fprintln(ew, err.Error())
 		return
 	}
 
 	for _, target := range expandedTargets {
 		wg.Add(1)
-		go doDestroy(c, target, wg, outchan, errchan)
+		go doDestroy(c, target, wg, out, ew)
 	}
 	return
 }
 
-func doDestroy(c *FleetClient, target string, wg *sync.WaitGroup, outchan chan string, errchan chan error) {
+func doDestroy(c *FleetClient, target string, wg *sync.WaitGroup, out, ew io.Writer) {
 	defer wg.Done()
 
 	// prepare string representation
 	component, num, err := splitTarget(target)
 	if err != nil {
-		errchan <- err
+		fmt.Fprintln(ew, err.Error())
 		return
 	}
 	name, err := formatUnitName(component, num)
 	if err != nil {
-		errchan <- err
+		fmt.Fprintln(ew, err.Error())
 		return
 	}
-	destroyed := fmt.Sprintf("\033[0;33m%v:\033[0m destroyed                                 \r", name)
+	tpl := prettyprint.Colorize("{{.Yellow}}%v:{{.Default}} destroyed")
+	destroyed := fmt.Sprintf(tpl, name)
 
 	// tell fleet to destroy the unit
 	c.Fleet.DestroyUnit(name)
@@ -48,14 +52,14 @@ outerLoop:
 		time.Sleep(250 * time.Millisecond)
 		unitStates, err := c.Fleet.UnitStates()
 		if err != nil {
-			errchan <- err
+			fmt.Fprintln(ew, err.Error())
 		}
 		for _, us := range unitStates {
 			if strings.HasPrefix(us.Name, name) {
 				continue outerLoop
 			}
 		}
-		outchan <- destroyed
+		fmt.Fprintln(out, destroyed)
 		return
 	}
 }

deisctl/backend/fleet/destroy_test.go

@@ -1,7 +1,6 @@
 package fleet
 
 import (
-	"fmt"
 	"sync"
 	"testing"
 
@@ -28,19 +27,14 @@ func TestDestroy(t *testing.T) {
 	c := &FleetClient{Fleet: &testFleetClient}
 
 	var errOutput string
-	outchan := make(chan string)
-	errchan := make(chan error)
 	var wg sync.WaitGroup
 
 	logMutex := sync.Mutex{}
 
-	go logState(outchan, errchan, &errOutput, &logMutex)
-
-	c.Destroy([]string{"controller", "registry", "router@1"}, &wg, outchan, errchan)
+	oe := newOutErr()
+	c.Destroy([]string{"controller", "registry", "router@1"}, &wg, oe.out, oe.ew)
 
 	wg.Wait()
-	close(errchan)
-	close(outchan)
 
 	logMutex.Lock()
 	if errOutput != "" {
@@ -49,6 +43,6 @@ func TestDestroy(t *testing.T) {
 	logMutex.Unlock()
 
 	if len(testFleetClient.testUnits) != 1 || testFleetClient.testUnits[0].Name != "deis-builder.service" {
-		t.Error(fmt.Errorf("Got %d Units (want 1), first unit %s (want builder)", len(testFleetClient.testUnits), testFleetClient.testUnits[0].Name))
+		t.Errorf("Got %d Units (want 1), first unit %s (want builder)", len(testFleetClient.testUnits), testFleetClient.testUnits[0].Name)
 	}
 }

deisctl/backend/fleet/fleet_test.go

@@ -1,6 +1,7 @@
 package fleet
 
 import (
+	"bytes"
 	"fmt"
 	"sync"
 	"testing"
@@ -103,6 +104,52 @@ func (c *stubFleetClient) DestroyUnit(name string) error {
 	return nil
 }
 
+func newOutErr() *outErr {
+	return &outErr{
+		&syncBuffer{},
+		&syncBuffer{},
+	}
+}
+
+// Wrap output and error streams for ease of testing.
+type outErr struct {
+	out, ew buffer
+}
+
+// buffer represents a buffer for collecting written test output.
+//
+// This is used only in testing, so add more bytes.Buffer methods as needed.
+type buffer interface {
+	Bytes() []byte
+	String() string
+	Write([]byte) (int, error)
+}
+
+// syncBuffer simply synchronizes writes on a bytes.Buffer.
+type syncBuffer struct {
+	bytes.Buffer
+	mx sync.RWMutex
+}
+
+func (s *syncBuffer) Write(b []byte) (int, error) {
+	s.mx.Lock()
+	defer s.mx.Unlock()
+	return s.Buffer.Write(b)
+}
+
+func (s *syncBuffer) String() string {
+	s.mx.RLock()
+	defer s.mx.RUnlock()
+	return s.Buffer.String()
+}
+
+func (s *syncBuffer) Bytes() []byte {
+	s.mx.RLock()
+	defer s.mx.RUnlock()
+	return s.Buffer.Bytes()
+}
+
+/*
 func logState(outchan chan string, errchan chan error, errOutput *string, mutex *sync.Mutex) {
 	for {
 		select {
@@ -125,6 +172,7 @@ func logState(outchan chan string, errchan chan error, errOutput *string, mutex
 		}
 	}
 }
+*/
 
 func TestNewClient(t *testing.T) {
 	t.Parallel()

deisctl/backend/fleet/list_unit_files_test.go

@@ -2,7 +2,6 @@ package fleet
 
 import (
 	"bytes"
-	"fmt"
 	"sync"
 	"testing"
 	"text/tabwriter"
@@ -80,6 +79,6 @@ deis-router@1.service	1182ecf	launched	launched	654321.../2.2.2.2
 	actual := testWriter.String()
 
 	if expected != actual {
-		t.Error(fmt.Errorf("Expected '%s', Got '%s'", expected, actual))
+		t.Errorf("Expected '%s', Got '%s'", expected, actual)
 	}
 }

deisctl/backend/fleet/list_units_test.go

@@ -2,7 +2,6 @@ package fleet
 
 import (
 	"bytes"
-	"fmt"
 	"sync"
 	"testing"
 	"text/tabwriter"
@@ -70,6 +69,6 @@ deis-router@1.service	654321.../2.2.2.2	loaded	active	running
 	actual := testWriter.String()
 
 	if expected != actual {
-		t.Error(fmt.Errorf("Expected '%s', Got '%s'", expected, actual))
+		t.Errorf("Expected '%s', Got '%s'", expected, actual)
 	}
 }