From 835971c6fdaf6ea35a0e7e45f6d9a09fd5f03ce1 Mon Sep 17 00:00:00 2001 From: Vincent Demeester Date: Wed, 4 Jan 2017 18:01:59 +0100 Subject: [PATCH] Extract daemon statsCollector to its own package Signed-off-by: Vincent Demeester --- daemon/daemon.go | 13 +- daemon/daemon_unix.go | 4 +- daemon/delete.go | 2 +- daemon/stats.go | 4 +- daemon/stats/collector.go | 101 ++++++++++++++ daemon/stats/collector_solaris.go | 29 ++++ .../collector_unix.go} | 13 +- .../collector_windows.go} | 8 +- daemon/stats/types.go | 42 ++++++ daemon/stats_collector.go | 130 ++---------------- daemon/stats_collector_solaris.go | 34 ----- 11 files changed, 205 insertions(+), 175 deletions(-) create mode 100644 daemon/stats/collector.go create mode 100644 daemon/stats/collector_solaris.go rename daemon/{stats_collector_unix.go => stats/collector_unix.go} (81%) rename daemon/{stats_collector_windows.go => stats/collector_windows.go} (61%) create mode 100644 daemon/stats/types.go delete mode 100644 daemon/stats_collector_solaris.go diff --git a/daemon/daemon.go b/daemon/daemon.go index e402d6f2e3ad6..052d0d461434f 100644 --- a/daemon/daemon.go +++ b/daemon/daemon.go @@ -26,14 +26,13 @@ import ( "github.com/docker/docker/container" "github.com/docker/docker/daemon/events" "github.com/docker/docker/daemon/exec" - "github.com/docker/docker/daemon/initlayer" - "github.com/docker/docker/dockerversion" - "github.com/docker/docker/plugin" - "github.com/docker/libnetwork/cluster" // register graph drivers _ "github.com/docker/docker/daemon/graphdriver/register" + "github.com/docker/docker/daemon/initlayer" + "github.com/docker/docker/daemon/stats" dmetadata "github.com/docker/docker/distribution/metadata" "github.com/docker/docker/distribution/xfer" + "github.com/docker/docker/dockerversion" "github.com/docker/docker/image" "github.com/docker/docker/layer" "github.com/docker/docker/libcontainerd" @@ -46,6 +45,7 @@ import ( "github.com/docker/docker/pkg/sysinfo" "github.com/docker/docker/pkg/system" "github.com/docker/docker/pkg/truncindex" + "github.com/docker/docker/plugin" "github.com/docker/docker/reference" "github.com/docker/docker/registry" "github.com/docker/docker/runconfig" @@ -53,6 +53,7 @@ import ( "github.com/docker/docker/volume/local" "github.com/docker/docker/volume/store" "github.com/docker/libnetwork" + "github.com/docker/libnetwork/cluster" nwconfig "github.com/docker/libnetwork/config" "github.com/docker/libtrust" "github.com/pkg/errors" @@ -82,7 +83,7 @@ type Daemon struct { trustKey libtrust.PrivateKey idIndex *truncindex.TruncIndex configStore *Config - statsCollector *statsCollector + statsCollector *stats.Collector defaultLogConfig containertypes.LogConfig RegistryService registry.Service EventsService *events.Events @@ -106,6 +107,8 @@ type Daemon struct { clusterProvider cluster.Provider cluster Cluster + machineMemory uint64 + seccompProfile []byte seccompProfilePath string } diff --git a/daemon/daemon_unix.go b/daemon/daemon_unix.go index e8efb95be3c3e..98ca569a7f66c 100644 --- a/daemon/daemon_unix.go +++ b/daemon/daemon_unix.go @@ -1125,8 +1125,8 @@ func (daemon *Daemon) stats(c *container.Container) (*types.StatsJSON, error) { Limit: mem.Limit, } // if the container does not set memory limit, use the machineMemory - if mem.Limit > daemon.statsCollector.machineMemory && daemon.statsCollector.machineMemory > 0 { - s.MemoryStats.Limit = daemon.statsCollector.machineMemory + if mem.Limit > daemon.machineMemory && daemon.machineMemory > 0 { + s.MemoryStats.Limit = daemon.machineMemory } if cgs.PidsStats != nil { s.PidsStats = types.PidsStats{ diff --git a/daemon/delete.go b/daemon/delete.go index 6b622bde370f0..92272903574b7 100644 --- a/daemon/delete.go +++ b/daemon/delete.go @@ -89,7 +89,7 @@ func (daemon *Daemon) cleanupContainer(container *container.Container, forceRemo // stop collection of stats for the container regardless // if stats are currently getting collected. - daemon.statsCollector.stopCollection(container) + daemon.statsCollector.StopCollection(container) if err = daemon.containerStop(container, 3); err != nil { return err diff --git a/daemon/stats.go b/daemon/stats.go index 51f5962d17c20..d5aa43d3d9f18 100644 --- a/daemon/stats.go +++ b/daemon/stats.go @@ -133,11 +133,11 @@ func (daemon *Daemon) ContainerStats(ctx context.Context, prefixOrName string, c } func (daemon *Daemon) subscribeToContainerStats(c *container.Container) chan interface{} { - return daemon.statsCollector.collect(c) + return daemon.statsCollector.Collect(c) } func (daemon *Daemon) unsubscribeToContainerStats(c *container.Container, ch chan interface{}) { - daemon.statsCollector.unsubscribe(c, ch) + daemon.statsCollector.Unsubscribe(c, ch) } // GetContainerStats collects all the stats published by a container diff --git a/daemon/stats/collector.go b/daemon/stats/collector.go new file mode 100644 index 0000000000000..e7eec7631ec92 --- /dev/null +++ b/daemon/stats/collector.go @@ -0,0 +1,101 @@ +// +build !solaris + +package stats + +import ( + "time" + + "github.com/Sirupsen/logrus" + "github.com/docker/docker/container" + "github.com/docker/docker/pkg/pubsub" +) + +// Collect registers the container with the collector and adds it to +// the event loop for collection on the specified interval returning +// a channel for the subscriber to receive on. +func (s *Collector) Collect(c *container.Container) chan interface{} { + s.m.Lock() + defer s.m.Unlock() + publisher, exists := s.publishers[c] + if !exists { + publisher = pubsub.NewPublisher(100*time.Millisecond, 1024) + s.publishers[c] = publisher + } + return publisher.Subscribe() +} + +// StopCollection closes the channels for all subscribers and removes +// the container from metrics collection. +func (s *Collector) StopCollection(c *container.Container) { + s.m.Lock() + if publisher, exists := s.publishers[c]; exists { + publisher.Close() + delete(s.publishers, c) + } + s.m.Unlock() +} + +// Unsubscribe removes a specific subscriber from receiving updates for a container's stats. +func (s *Collector) Unsubscribe(c *container.Container, ch chan interface{}) { + s.m.Lock() + publisher := s.publishers[c] + if publisher != nil { + publisher.Evict(ch) + if publisher.Len() == 0 { + delete(s.publishers, c) + } + } + s.m.Unlock() +} + +// Run starts the collectors and will indefinitely collect stats from the supervisor +func (s *Collector) Run() { + type publishersPair struct { + container *container.Container + publisher *pubsub.Publisher + } + // we cannot determine the capacity here. + // it will grow enough in first iteration + var pairs []publishersPair + + for range time.Tick(s.interval) { + // it does not make sense in the first iteration, + // but saves allocations in further iterations + pairs = pairs[:0] + + s.m.Lock() + for container, publisher := range s.publishers { + // copy pointers here to release the lock ASAP + pairs = append(pairs, publishersPair{container, publisher}) + } + s.m.Unlock() + if len(pairs) == 0 { + continue + } + + systemUsage, err := s.getSystemCPUUsage() + if err != nil { + logrus.Errorf("collecting system cpu usage: %v", err) + continue + } + + for _, pair := range pairs { + stats, err := s.supervisor.GetContainerStats(pair.container) + if err != nil { + if _, ok := err.(notRunningErr); !ok { + logrus.Errorf("collecting stats for %s: %v", pair.container.ID, err) + } + continue + } + // FIXME: move to containerd on Linux (not Windows) + stats.CPUStats.SystemUsage = systemUsage + + pair.publisher.Publish(*stats) + } + } +} + +type notRunningErr interface { + error + ContainerIsRunning() bool +} diff --git a/daemon/stats/collector_solaris.go b/daemon/stats/collector_solaris.go new file mode 100644 index 0000000000000..3699d08c10421 --- /dev/null +++ b/daemon/stats/collector_solaris.go @@ -0,0 +1,29 @@ +package stats + +import ( + "github.com/docker/docker/container" +) + +// platformNewStatsCollector performs platform specific initialisation of the +// Collector structure. This is a no-op on Windows. +func platformNewStatsCollector(s *Collector) { +} + +// Collect registers the container with the collector and adds it to +// the event loop for collection on the specified interval returning +// a channel for the subscriber to receive on. +// Currently not supported on Solaris +func (s *Collector) Collect(c *container.Container) chan interface{} { + return nil +} + +// StopCollection closes the channels for all subscribers and removes +// the container from metrics collection. +// Currently not supported on Solaris +func (s *Collector) StopCollection(c *container.Container) { +} + +// Unsubscribe removes a specific subscriber from receiving updates for a container's stats. +// Currently not supported on Solaris +func (s *Collector) Unsubscribe(c *container.Container, ch chan interface{}) { +} diff --git a/daemon/stats_collector_unix.go b/daemon/stats/collector_unix.go similarity index 81% rename from daemon/stats_collector_unix.go rename to daemon/stats/collector_unix.go index 0fcc9c5828829..611df1ed69569 100644 --- a/daemon/stats_collector_unix.go +++ b/daemon/stats/collector_unix.go @@ -1,6 +1,6 @@ // +build !windows,!solaris -package daemon +package stats import ( "fmt" @@ -8,18 +8,13 @@ import ( "strconv" "strings" - sysinfo "github.com/docker/docker/pkg/system" "github.com/opencontainers/runc/libcontainer/system" ) // platformNewStatsCollector performs platform specific initialisation of the -// statsCollector structure. -func platformNewStatsCollector(s *statsCollector) { +// Collector structure. +func platformNewStatsCollector(s *Collector) { s.clockTicksPerSecond = uint64(system.GetClockTicks()) - meminfo, err := sysinfo.ReadMemInfo() - if err == nil && meminfo.MemTotal > 0 { - s.machineMemory = uint64(meminfo.MemTotal) - } } const nanoSecondsPerSecond = 1e9 @@ -32,7 +27,7 @@ const nanoSecondsPerSecond = 1e9 // statistics line and then sums up the first seven fields // provided. See `man 5 proc` for details on specific field // information. -func (s *statsCollector) getSystemCPUUsage() (uint64, error) { +func (s *Collector) getSystemCPUUsage() (uint64, error) { var line string f, err := os.Open("/proc/stat") if err != nil { diff --git a/daemon/stats_collector_windows.go b/daemon/stats/collector_windows.go similarity index 61% rename from daemon/stats_collector_windows.go rename to daemon/stats/collector_windows.go index 41731b9c14c68..e51495f048cd2 100644 --- a/daemon/stats_collector_windows.go +++ b/daemon/stats/collector_windows.go @@ -1,15 +1,15 @@ // +build windows -package daemon +package stats // platformNewStatsCollector performs platform specific initialisation of the -// statsCollector structure. This is a no-op on Windows. -func platformNewStatsCollector(s *statsCollector) { +// Collector structure. This is a no-op on Windows. +func platformNewStatsCollector(s *Collector) { } // getSystemCPUUsage returns the host system's cpu usage in // nanoseconds. An error is returned if the format of the underlying // file does not match. This is a no-op on Windows. -func (s *statsCollector) getSystemCPUUsage() (uint64, error) { +func (s *Collector) getSystemCPUUsage() (uint64, error) { return 0, nil } diff --git a/daemon/stats/types.go b/daemon/stats/types.go new file mode 100644 index 0000000000000..e48783c9a9411 --- /dev/null +++ b/daemon/stats/types.go @@ -0,0 +1,42 @@ +package stats + +import ( + "bufio" + "sync" + "time" + + "github.com/docker/docker/api/types" + "github.com/docker/docker/container" + "github.com/docker/docker/pkg/pubsub" +) + +type supervisor interface { + // GetContainerStats collects all the stats related to a container + GetContainerStats(container *container.Container) (*types.StatsJSON, error) +} + +// NewCollector creates a stats collector that will poll the supervisor with the specified interval +func NewCollector(supervisor supervisor, interval time.Duration) *Collector { + s := &Collector{ + interval: interval, + supervisor: supervisor, + publishers: make(map[*container.Container]*pubsub.Publisher), + bufReader: bufio.NewReaderSize(nil, 128), + } + + platformNewStatsCollector(s) + + return s +} + +// Collector manages and provides container resource stats +type Collector struct { + m sync.Mutex + supervisor supervisor + interval time.Duration + publishers map[*container.Container]*pubsub.Publisher + bufReader *bufio.Reader + + // The following fields are not set on Windows currently. + clockTicksPerSecond uint64 +} diff --git a/daemon/stats_collector.go b/daemon/stats_collector.go index dc6825e705239..7daf26f9f7433 100644 --- a/daemon/stats_collector.go +++ b/daemon/stats_collector.go @@ -1,132 +1,26 @@ -// +build !solaris - package daemon import ( - "bufio" - "sync" + "runtime" "time" - "github.com/Sirupsen/logrus" - "github.com/docker/docker/api/types" - "github.com/docker/docker/container" - "github.com/docker/docker/pkg/pubsub" + "github.com/docker/docker/daemon/stats" + "github.com/docker/docker/pkg/system" ) -type statsSupervisor interface { - // GetContainerStats collects all the stats related to a container - GetContainerStats(container *container.Container) (*types.StatsJSON, error) -} - // newStatsCollector returns a new statsCollector that collections // stats for a registered container at the specified interval. // The collector allows non-running containers to be added // and will start processing stats when they are started. -func (daemon *Daemon) newStatsCollector(interval time.Duration) *statsCollector { - s := &statsCollector{ - interval: interval, - supervisor: daemon, - publishers: make(map[*container.Container]*pubsub.Publisher), - bufReader: bufio.NewReaderSize(nil, 128), - } - platformNewStatsCollector(s) - go s.run() - return s -} - -// statsCollector manages and provides container resource stats -type statsCollector struct { - m sync.Mutex - supervisor statsSupervisor - interval time.Duration - publishers map[*container.Container]*pubsub.Publisher - bufReader *bufio.Reader - - // The following fields are not set on Windows currently. - clockTicksPerSecond uint64 - machineMemory uint64 -} - -// collect registers the container with the collector and adds it to -// the event loop for collection on the specified interval returning -// a channel for the subscriber to receive on. -func (s *statsCollector) collect(c *container.Container) chan interface{} { - s.m.Lock() - defer s.m.Unlock() - publisher, exists := s.publishers[c] - if !exists { - publisher = pubsub.NewPublisher(100*time.Millisecond, 1024) - s.publishers[c] = publisher - } - return publisher.Subscribe() -} - -// stopCollection closes the channels for all subscribers and removes -// the container from metrics collection. -func (s *statsCollector) stopCollection(c *container.Container) { - s.m.Lock() - if publisher, exists := s.publishers[c]; exists { - publisher.Close() - delete(s.publishers, c) - } - s.m.Unlock() -} - -// unsubscribe removes a specific subscriber from receiving updates for a container's stats. -func (s *statsCollector) unsubscribe(c *container.Container, ch chan interface{}) { - s.m.Lock() - publisher := s.publishers[c] - if publisher != nil { - publisher.Evict(ch) - if publisher.Len() == 0 { - delete(s.publishers, c) - } - } - s.m.Unlock() -} - -func (s *statsCollector) run() { - type publishersPair struct { - container *container.Container - publisher *pubsub.Publisher - } - // we cannot determine the capacity here. - // it will grow enough in first iteration - var pairs []publishersPair - - for range time.Tick(s.interval) { - // it does not make sense in the first iteration, - // but saves allocations in further iterations - pairs = pairs[:0] - - s.m.Lock() - for container, publisher := range s.publishers { - // copy pointers here to release the lock ASAP - pairs = append(pairs, publishersPair{container, publisher}) - } - s.m.Unlock() - if len(pairs) == 0 { - continue - } - - systemUsage, err := s.getSystemCPUUsage() - if err != nil { - logrus.Errorf("collecting system cpu usage: %v", err) - continue - } - - for _, pair := range pairs { - stats, err := s.supervisor.GetContainerStats(pair.container) - if err != nil { - if _, ok := err.(errNotRunning); !ok { - logrus.Errorf("collecting stats for %s: %v", pair.container.ID, err) - } - continue - } - // FIXME: move to containerd on Linux (not Windows) - stats.CPUStats.SystemUsage = systemUsage - - pair.publisher.Publish(*stats) +func (daemon *Daemon) newStatsCollector(interval time.Duration) *stats.Collector { + // FIXME(vdemeester) move this elsewhere + if runtime.GOOS == "linux" { + meminfo, err := system.ReadMemInfo() + if err == nil && meminfo.MemTotal > 0 { + daemon.machineMemory = uint64(meminfo.MemTotal) } } + s := stats.NewCollector(daemon, interval) + go s.Run() + return s } diff --git a/daemon/stats_collector_solaris.go b/daemon/stats_collector_solaris.go deleted file mode 100644 index 9cf9f0a94e310..0000000000000 --- a/daemon/stats_collector_solaris.go +++ /dev/null @@ -1,34 +0,0 @@ -package daemon - -import ( - "github.com/docker/docker/container" - "time" -) - -// newStatsCollector returns a new statsCollector for collection stats -// for a registered container at the specified interval. The collector allows -// non-running containers to be added and will start processing stats when -// they are started. -func (daemon *Daemon) newStatsCollector(interval time.Duration) *statsCollector { - return &statsCollector{} -} - -// statsCollector manages and provides container resource stats -type statsCollector struct { -} - -// collect registers the container with the collector and adds it to -// the event loop for collection on the specified interval returning -// a channel for the subscriber to receive on. -func (s *statsCollector) collect(c *container.Container) chan interface{} { - return nil -} - -// stopCollection closes the channels for all subscribers and removes -// the container from metrics collection. -func (s *statsCollector) stopCollection(c *container.Container) { -} - -// unsubscribe removes a specific subscriber from receiving updates for a container's stats. -func (s *statsCollector) unsubscribe(c *container.Container, ch chan interface{}) { -}