stats refactor

Refactor the entities' stats API to simplify using it and reduce the
risk of running into concurrency issues at the call sites.  Further
simplify the stats code by de-spaghetti-ing the logic and reducing
duplicate code.

`ContainerStats` now returns a data channel and an error.  If the error
is nil, callers can read from the channel.

Signed-off-by: Valentin Rothberg <rothberg@redhat.com>

cmd/podman/containers/stats.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"os"
 	"strings"
-	"sync"
 	"text/tabwriter"
 	"text/template"
 
@@ -107,32 +106,31 @@ func stats(cmd *cobra.Command, args []string) error {
 			return errors.New("stats is not supported in rootless mode without cgroups v2")
 		}
 	}
-	statsOptions.StatChan = make(chan []*define.ContainerStats, 1)
-	wg := sync.WaitGroup{}
-	wg.Add(1)
-	go func() {
-		for reports := range statsOptions.StatChan {
-			if err := outputStats(reports); err != nil {
-				logrus.Error(err)
-			}
-		}
-		wg.Done()
 
-	}()
+	statsChan, err := registry.ContainerEngine().ContainerStats(registry.Context(), args, statsOptions)
-	err := registry.ContainerEngine().ContainerStats(registry.Context(), args, statsOptions)
+	if err != nil {
-	wg.Wait()
+		return err
-	return err
+	}
+	for report := range statsChan {
+		if report.Error != nil {
+			return report.Error
+		}
+		if err := outputStats(report.Stats); err != nil {
+			logrus.Error(err)
+		}
+	}
+	return nil
 }
 
-func outputStats(reports []*define.ContainerStats) error {
+func outputStats(reports []define.ContainerStats) error {
 	if len(statsOptions.Format) < 1 && !statsOptions.NoReset {
 		tm.Clear()
 		tm.MoveCursor(1, 1)
 		tm.Flush()
 	}
-	stats := make([]*containerStats, 0, len(reports))
+	stats := make([]containerStats, 0, len(reports))
 	for _, r := range reports {
-		stats = append(stats, &containerStats{r})
+		stats = append(stats, containerStats{r})
 	}
 	if statsOptions.Format == "json" {
 		return outputJSON(stats)
@@ -163,7 +161,7 @@ func outputStats(reports []*define.ContainerStats) error {
 }
 
 type containerStats struct {
-	*define.ContainerStats
+	define.ContainerStats
 }
 
 func (s *containerStats) ID() string {
@@ -213,7 +211,7 @@ func combineHumanValues(a, b uint64) string {
 	return fmt.Sprintf("%s / %s", units.HumanSize(float64(a)), units.HumanSize(float64(b)))
 }
 
-func outputJSON(stats []*containerStats) error {
+func outputJSON(stats []containerStats) error {
 	type jstat struct {
 		Id         string `json:"id"` //nolint
 		Name       string `json:"name"`

pkg/domain/entities/containers.go

@@ -416,5 +416,10 @@ type ContainerStatsOptions struct {
 	Latest   bool
 	NoReset  bool
 	NoStream bool
-	StatChan chan []*define.ContainerStats
+}
+
+// ContainerStatsReport is used for streaming container stats.
+type ContainerStatsReport struct {
+	Error error
+	Stats []define.ContainerStats
 }

pkg/domain/entities/engine_container.go

@@ -38,7 +38,7 @@ type ContainerEngine interface {
 	ContainerRun(ctx context.Context, opts ContainerRunOptions) (*ContainerRunReport, error)
 	ContainerRunlabel(ctx context.Context, label string, image string, args []string, opts ContainerRunlabelOptions) error
 	ContainerStart(ctx context.Context, namesOrIds []string, options ContainerStartOptions) ([]*ContainerStartReport, error)
-	ContainerStats(ctx context.Context, namesOrIds []string, options ContainerStatsOptions) error
+	ContainerStats(ctx context.Context, namesOrIds []string, options ContainerStatsOptions) (chan ContainerStatsReport, error)
 	ContainerStop(ctx context.Context, namesOrIds []string, options StopOptions) ([]*StopReport, error)
 	ContainerTop(ctx context.Context, options TopOptions) (*StringSliceReport, error)
 	ContainerUnmount(ctx context.Context, nameOrIDs []string, options ContainerUnmountOptions) ([]*ContainerUnmountReport, error)

pkg/domain/infra/abi/containers.go

@@ -1142,12 +1142,11 @@ func (ic *ContainerEngine) Shutdown(_ context.Context) {
 	})
 }
 
-func (ic *ContainerEngine) ContainerStats(ctx context.Context, namesOrIds []string, options entities.ContainerStatsOptions) error {
+func (ic *ContainerEngine) ContainerStats(ctx context.Context, namesOrIds []string, options entities.ContainerStatsOptions) (statsChan chan entities.ContainerStatsReport, err error) {
-	defer close(options.StatChan)
+	statsChan = make(chan entities.ContainerStatsReport, 1)
+
 	containerFunc := ic.Libpod.GetRunningContainers
 	switch {
-	case len(namesOrIds) > 0:
-		containerFunc = func() ([]*libpod.Container, error) { return ic.Libpod.GetContainersByList(namesOrIds) }
 	case options.Latest:
 		containerFunc = func() ([]*libpod.Container, error) {
 			lastCtr, err := ic.Libpod.GetLatestContainer()
@@ -1156,62 +1155,74 @@ func (ic *ContainerEngine) ContainerStats(ctx context.Context, namesOrIds []stri
 			}
 			return []*libpod.Container{lastCtr}, nil
 		}
+	case len(namesOrIds) > 0:
+		containerFunc = func() ([]*libpod.Container, error) { return ic.Libpod.GetContainersByList(namesOrIds) }
 	case options.All:
 		containerFunc = ic.Libpod.GetAllContainers
 	}
 
-	ctrs, err := containerFunc()
+	go func() {
-	if err != nil {
+		defer close(statsChan)
-		return errors.Wrapf(err, "unable to get list of containers")
+		var (
-	}
+			err            error
-	containerStats := map[string]*define.ContainerStats{}
+			containers     []*libpod.Container
-	for _, ctr := range ctrs {
+			containerStats map[string]*define.ContainerStats
-		initialStats, err := ctr.GetContainerStats(&define.ContainerStats{})
+		)
-		if err != nil {
+		containerStats = make(map[string]*define.ContainerStats)
-			// when doing "all", don't worry about containers that are not running
+
-			cause := errors.Cause(err)
+	stream: // label to flatten the scope
-			if options.All && (cause == define.ErrCtrRemoved || cause == define.ErrNoSuchCtr || cause == define.ErrCtrStateInvalid) {
+		select {
-				continue
+		case <-ctx.Done():
-			}
+			// client cancelled
-			if cause == cgroups.ErrCgroupV1Rootless {
+			logrus.Debugf("Container stats stopped: context cancelled")
-				err = cause
+			return
-			}
+		default:
-			return err
+			// just fall through and do work
 		}
-		containerStats[ctr.ID()] = initialStats
+
-	}
+		// Anonymous func to easily use the return values for streaming.
-	for {
+		computeStats := func() ([]define.ContainerStats, error) {
-		reportStats := []*define.ContainerStats{}
+			containers, err = containerFunc()
-		for _, ctr := range ctrs {
+			if err != nil {
-			id := ctr.ID()
+				return nil, errors.Wrapf(err, "unable to get list of containers")
-			if _, ok := containerStats[ctr.ID()]; !ok {
+			}
-				initialStats, err := ctr.GetContainerStats(&define.ContainerStats{})
+
-				if errors.Cause(err) == define.ErrCtrRemoved || errors.Cause(err) == define.ErrNoSuchCtr || errors.Cause(err) == define.ErrCtrStateInvalid {
+			reportStats := []define.ContainerStats{}
-					// skip dealing with a container that is gone
+			for _, ctr := range containers {
-					continue
+				prev, ok := containerStats[ctr.ID()]
+				if !ok {
+					prev = &define.ContainerStats{}
 				}
+
+				stats, err := ctr.GetContainerStats(prev)
 				if err != nil {
-					return err
+					cause := errors.Cause(err)
+					if options.All && (cause == define.ErrCtrRemoved || cause == define.ErrNoSuchCtr || cause == define.ErrCtrStateInvalid) {
+						continue
+					}
+					if cause == cgroups.ErrCgroupV1Rootless {
+						err = cause
+					}
+					return nil, err
 				}
-				containerStats[id] = initialStats
+
+				containerStats[ctr.ID()] = stats
+				reportStats = append(reportStats, *stats)
 			}
-			stats, err := ctr.GetContainerStats(containerStats[id])
+			return reportStats, nil
-			if err != nil && errors.Cause(err) != define.ErrNoSuchCtr {
-				return err
-			}
-			// replace the previous measurement with the current one
-			containerStats[id] = stats
-			reportStats = append(reportStats, stats)
 		}
-		ctrs, err = containerFunc()
+
-		if err != nil {
+		report := entities.ContainerStatsReport{}
-			return err
+		report.Stats, report.Error = computeStats()
-		}
+		statsChan <- report
-		options.StatChan <- reportStats
+
 		if options.NoStream {
-			break
+			return
 		}
+
 		time.Sleep(time.Second)
-	}
+		goto stream
-	return nil
+	}()
+
+	return statsChan, nil
 }

pkg/domain/infra/tunnel/containers.go

@@ -682,6 +682,6 @@ func (ic *ContainerEngine) ContainerCp(ctx context.Context, source, dest string,
 func (ic *ContainerEngine) Shutdown(_ context.Context) {
 }
 
-func (ic *ContainerEngine) ContainerStats(ctx context.Context, namesOrIds []string, options entities.ContainerStatsOptions) error {
+func (ic *ContainerEngine) ContainerStats(ctx context.Context, namesOrIds []string, options entities.ContainerStatsOptions) (statsChan chan entities.ContainerStatsReport, err error) {
-	return errors.New("not implemented")
+	return nil, errors.New("not implemented")
 }