-
Notifications
You must be signed in to change notification settings - Fork 18.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #29887 from vdemeester/daemon-extract-stats-collector
Extract daemon statsCollector to its own package
- Loading branch information
Showing
11 changed files
with
205 additions
and
175 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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{}) { | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
8 changes: 4 additions & 4 deletions
8
daemon/stats_collector_windows.go → daemon/stats/collector_windows.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |
Oops, something went wrong.