Fixed race and added support for all option in docker stats

In some cases a race could be encountered when executing docker
stats.  The race would result in an orphaned go routine that
would be a memory leak.

In addition if stats --no-stream -all was requested the persona
would hang on any stopped containers.  Stats now will consider
the container state when processing these requests.

Finally, the unit testing coverage was increased to just over
90% and integration tests were created.

Fixes #4549, #4585, #4421

lib/apiservers/engine/backends/container.go

@@ -51,6 +51,7 @@ import (
 
 	"github.com/vmware/vic/lib/apiservers/engine/backends/cache"
 	viccontainer "github.com/vmware/vic/lib/apiservers/engine/backends/container"
+	"github.com/vmware/vic/lib/apiservers/engine/backends/convert"
 	"github.com/vmware/vic/lib/apiservers/engine/backends/filter"
 	"github.com/vmware/vic/lib/apiservers/engine/backends/portmap"
 	"github.com/vmware/vic/lib/apiservers/portlayer/client/containers"
@@ -1367,7 +1368,25 @@ func (c *Container) ContainerStats(ctx context.Context, name string, config *bac
 		out = io.Writer(wf)
 	}
 
-	err = c.containerProxy.StreamContainerStats(ctx, vc.ContainerID, out, config.Stream, cpuMhz, vc.HostConfig.Memory)
+	// stats configuration
+	statsConfig := &convert.ContainerStatsConfig{
+		VchMhz:      cpuMhz,
+		Stream:      config.Stream,
+		ContainerID: vc.ContainerID,
+		Out:         out,
+		Memory:      vc.HostConfig.Memory,
+	}
+
+	// if we are not streaming then we need to get the container state
+	if !config.Stream {
+		statsConfig.ContainerState, err = c.containerProxy.State(vc)
+		if err != nil {
+			return InternalServerError(err.Error())
+		}
+
+	}
+
+	err = c.containerProxy.StreamContainerStats(ctx, statsConfig)
 	if err != nil {
 		log.Errorf("error while streaming container (%s) stats: %s", vc.ContainerID, err)
 	}

lib/apiservers/engine/backends/container_proxy.go

@@ -89,7 +89,7 @@ type VicContainerProxy interface {
 	AddInteractionToContainer(handle string, config types.ContainerCreateConfig) (string, error)
 	CommitContainerHandle(handle, containerID string, waitTime int32) error
 	StreamContainerLogs(name string, out io.Writer, started chan struct{}, showTimestamps bool, followLogs bool, since int64, tailLines int64) error
-	StreamContainerStats(ctx context.Context, id string, out io.Writer, stream bool, CPUMhz int64, memory int64) error
+	StreamContainerStats(ctx context.Context, config *convert.ContainerStatsConfig) error
 
 	Stop(vc *viccontainer.VicContainer, name string, seconds *int, unbound bool) error
 	State(vc *viccontainer.VicContainer) (*types.ContainerState, error)
@@ -523,8 +523,8 @@ func (c *ContainerProxy) StreamContainerLogs(name string, out io.Writer, started
 // StreamContainerStats will provide a stream of container stats written to the provided
 // io.Writer.  Prior to writing to the provided io.Writer there will be a transformation
 // from the portLayer representation of stats to the docker format
-func (c *ContainerProxy) StreamContainerStats(ctx context.Context, id string, out io.Writer, stream bool, CPUMhz int64, mem int64) error {
-	defer trace.End(trace.Begin(id))
+func (c *ContainerProxy) StreamContainerStats(ctx context.Context, config *convert.ContainerStatsConfig) error {
+	defer trace.End(trace.Begin(config.ContainerID))
 
 	plClient, transport := c.createNewAttachClientWithTimeouts(attachConnectTimeout, 0, attachAttemptTimeout)
 	defer transport.Close()
@@ -534,33 +534,26 @@ func (c *ContainerProxy) StreamContainerStats(ctx context.Context, id string, ou
 	defer cancel()
 
 	params := containers.NewGetContainerStatsParamsWithContext(ctx)
-	params.ID = id
-	params.Stream = stream
-
-	// converter config
-	config := convert.ContainerStatsConfig{
-		Ctx:         ctx,
-		Cancel:      cancel,
-		VchMhz:      CPUMhz,
-		Stream:      stream,
-		ContainerID: id,
-		Out:         out,
-		Memory:      mem,
-	}
+	params.ID = config.ContainerID
+	params.Stream = config.Stream
+
+	config.Ctx = ctx
+	config.Cancel = cancel
+
 	// create our converter
 	containerConverter := convert.NewContainerStats(config)
 	// provide the writer for the portLayer and start listening for metrics
 	writer := containerConverter.Listen()
 	if writer == nil {
 		// problem with the listener
-		return InternalServerError(fmt.Sprintf("unable to gather container(%s) statistics", id))
+		return InternalServerError(fmt.Sprintf("unable to gather container(%s) statistics", config.ContainerID))
 	}
 
 	_, err := plClient.Containers.GetContainerStats(params, writer)
 	if err != nil {
 		switch err := err.(type) {
 		case *containers.GetContainerStatsNotFound:
-			return NotFoundError(id)
+			return NotFoundError(config.ContainerID)
 		case *containers.GetContainerStatsInternalServerError:
 			return InternalServerError("Server error from the interaction port layer")
 		default:

lib/apiservers/engine/backends/container_test.go

@@ -38,6 +38,7 @@ import (
 
 	"github.com/vmware/vic/lib/apiservers/engine/backends/cache"
 	viccontainer "github.com/vmware/vic/lib/apiservers/engine/backends/container"
+	"github.com/vmware/vic/lib/apiservers/engine/backends/convert"
 	plclient "github.com/vmware/vic/lib/apiservers/portlayer/client"
 	plscopes "github.com/vmware/vic/lib/apiservers/portlayer/client/scopes"
 	plmodels "github.com/vmware/vic/lib/apiservers/portlayer/models"
@@ -338,7 +339,7 @@ func (m *MockContainerProxy) Rename(vc *viccontainer.VicContainer, newName strin
 func (m *MockContainerProxy) AttachStreams(ctx context.Context, vc *viccontainer.VicContainer, clStdin io.ReadCloser, clStdout, clStderr io.Writer, ca *backend.ContainerAttachConfig) error {
 	return nil
 }
-func (m *MockContainerProxy) StreamContainerStats(ctx context.Context, id string, out io.Writer, stream bool, CPUMhz int64, memory int64) error {
+func (m *MockContainerProxy) StreamContainerStats(ctx context.Context, config *convert.ContainerStatsConfig) error {
 	return nil
 }
 func (m *MockContainerProxy) exitCode(vc *viccontainer.VicContainer) (string, error) {