Improve Retrieve

closes vmware-tanzu#274 and closes vmware-tanzu#313

The UX is much cleaner with no channels.

The errors from the command line no longer show extraneous information.

Signed-off-by: Chuck Ha <[email protected]>

cmd/sonobuoy/app/retrieve.go

@@ -20,7 +20,6 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"sync"
 
 	"github.com/heptio/sonobuoy/pkg/client"
 	"github.com/heptio/sonobuoy/pkg/errlog"
@@ -33,10 +32,12 @@ var (
 	defaultOutDir = "."
 )
 
-var (
-	cpKubecfg   Kubeconfig
-	cpNamespace string
-)
+type receiveFlags struct {
+	namespace string
+	kubecfg   Kubeconfig
+}
+
+var rcvFlags receiveFlags
 
 func init() {
 	cmd := &cobra.Command{
@@ -46,26 +47,19 @@ func init() {
 		Args:  cobra.MaximumNArgs(1),
 	}
 
-	AddKubeconfigFlag(&cpKubecfg, cmd.Flags())
-	AddNamespaceFlag(&cpNamespace, cmd.Flags())
+	AddKubeconfigFlag(&rcvFlags.kubecfg, cmd.Flags())
+	AddNamespaceFlag(&rcvFlags.namespace, cmd.Flags())
 
 	RootCmd.AddCommand(cmd)
 }
 
-// TODO (timothysc) abstract retrieve details into a lower level function.
 func retrieveResults(cmd *cobra.Command, args []string) {
-	namespace, err := cmd.Flags().GetString("namespace")
-	if err != nil {
-		errlog.LogError(fmt.Errorf("failed to get namespace flag: %v", err))
-		os.Exit(1)
-	}
-
 	outDir := defaultOutDir
 	if len(args) > 0 {
 		outDir = args[0]
 	}
 
-	restConfig, err := cpKubecfg.Get()
+	restConfig, err := rcvFlags.kubecfg.Get()
 	if err != nil {
 		errlog.LogError(fmt.Errorf("failed to get kubernetes client: %v", err))
 		os.Exit(1)
@@ -76,50 +70,16 @@ func retrieveResults(cmd *cobra.Command, args []string) {
 		os.Exit(1)
 	}
 
-	// TODO(chuckha) try to catch some errors and present user friendly messages.
-	// Setup error channel and synchronization so that all errors get reported before exiting.
-	errc := make(chan error)
-	errcount := 0
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		for err := range errc {
-			errcount++
-			errlog.LogError(err)
-		}
-	}()
-
-	cfg := &client.RetrieveConfig{
-		Namespace: namespace,
-		CmdErr:    os.Stderr,
-		Errc:      errc,
-	}
-
 	// Get a reader that contains the tar output of the results directory.
-	reader := sbc.RetrieveResults(cfg)
-
-	// RetrieveResults bailed early and will report an error.
-	if reader == nil {
-		close(errc)
-		wg.Wait()
+	reader, err := sbc.RetrieveResults(&client.RetrieveConfig{Namespace: rcvFlags.namespace})
+	if err != nil {
+		errlog.LogError(err)
 		os.Exit(1)
 	}
 
 	// Extract the tar output into a local directory under the prefix.
 	err = client.UntarAll(reader, outDir, prefix)
 	if err != nil {
-		close(errc)
-		wg.Wait()
-		errlog.LogError(fmt.Errorf("error untarring output: %v", err))
-		os.Exit(1)
-	}
-
-	// Everything has been written from the reader which means we're done.
-	close(errc)
-	wg.Wait()
-
-	if errcount != 0 {
 		os.Exit(1)
 	}
 }

pkg/client/interfaces.go

@@ -69,10 +69,6 @@ type DeleteConfig struct {
 
 // RetrieveConfig are the options passed to RetrieveResults.
 type RetrieveConfig struct {
-	// CmdErr is the place to write errors to.
-	CmdErr io.Writer
-	// Errc reports errors from go routines that retrieve may spawn.
-	Errc chan error
 	// Namespace is the namespace the sonobuoy aggregator is running in.
 	Namespace string
 }
@@ -127,7 +123,7 @@ type Interface interface {
 	// GenerateManifest fills in a template with a Sonobuoy config
 	GenerateManifest(cfg *GenConfig) ([]byte, error)
 	// RetrieveResults copies results from a sonobuoy run into a Reader in tar format.
-	RetrieveResults(cfg *RetrieveConfig) io.Reader
+	RetrieveResults(cfg *RetrieveConfig) (io.Reader, error)
 	// GetStatus determines the status of the sonobuoy run in order to assist the user.
 	GetStatus(namespace string) (*aggregation.Status, error)
 	// LogReader returns a reader that contains a merged stream of sonobuoy logs.

pkg/client/retrieve.go

@@ -24,6 +24,8 @@ import (
 	"path"
 	"path/filepath"
 
+	"github.com/sirupsen/logrus"
+
 	"github.com/heptio/sonobuoy/pkg/config"
 	"github.com/pkg/errors"
 
@@ -32,13 +34,8 @@ import (
 	"k8s.io/client-go/tools/remotecommand"
 )
 
-func (c *SonobuoyClient) RetrieveResults(cfg *RetrieveConfig) io.Reader {
-	kubeClient, err := c.Client()
-	if err != nil {
-		cfg.Errc <- err
-		return nil
-	}
-	client := kubeClient.CoreV1().RESTClient()
+func (c *SonobuoyClient) RetrieveResults(cfg *RetrieveConfig) (io.Reader, error) {
+	client := c.client.CoreV1().RESTClient()
 	req := client.Post().
 		Resource("pods").
 		Name(config.MasterPodName).
@@ -54,25 +51,27 @@ func (c *SonobuoyClient) RetrieveResults(cfg *RetrieveConfig) io.Reader {
 	}, scheme.ParameterCodec)
 	executor, err := remotecommand.NewSPDYExecutor(c.RestConfig, "POST", req.URL())
 	if err != nil {
-		cfg.Errc <- errors.Wrap(err, "unable to get remote executor")
-		return nil
+		return nil, err
 	}
 	reader, writer := io.Pipe()
-	// We use a goroutine here to allow this function to return a reader that lets the caller
-	// deal with what to do with this stream of data and keep that detail out of this function.
-	go func() {
+	go func(writer *io.PipeWriter) {
 		defer writer.Close()
-
 		err = executor.Stream(remotecommand.StreamOptions{
 			Stdout: writer,
-			Stderr: cfg.CmdErr,
+			Stderr: os.Stderr,
 			Tty:    false,
 		})
 		if err != nil {
-			cfg.Errc <- fmt.Errorf("error streaming: %v", err)
+			// Since this function returns an io.Reader to the consumer and does
+			// not buffer the entire (potentially large) output, RetrieveResults
+			// has to return the reader first to be read from. This means we
+			// either lose this error (easy) or provide a significantly more
+			// complex error mechanism for the consumer (hard).
+			logrus.Error(err)
 		}
-	}()
-	return reader
+	}(writer)
+
+	return reader, nil
 }
 
 /** Everything below this marker has been copy/pasta'd from k8s/k8s. The only modification is exporting UntarAll **/