package daemon import ( "bufio" "fmt" "os" "strconv" "strings" "sync" "time" "github.com/Sirupsen/logrus" "github.com/docker/docker/daemon/execdriver" "github.com/docker/docker/pkg/pubsub" "github.com/docker/libcontainer/system" ) // newStatsCollector returns a new statsCollector that collections // network and cgroup 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 newStatsCollector(interval time.Duration) *statsCollector { s := &statsCollector{ interval: interval, publishers: make(map[*Container]*pubsub.Publisher), clockTicks: uint64(system.GetClockTicks()), bufReader: bufio.NewReaderSize(nil, 128), } go s.run() return s } // statsCollector manages and provides container resource stats type statsCollector struct { m sync.Mutex interval time.Duration clockTicks uint64 publishers map[*Container]*pubsub.Publisher bufReader *bufio.Reader } // 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) 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) { 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, 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 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) { systemUsage, err := s.getSystemCpuUsage() if err != nil { logrus.Errorf("collecting system cpu usage: %v", err) continue } // 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() for _, pair := range pairs { stats, err := pair.container.Stats() if err != nil { if err != execdriver.ErrNotRunning { logrus.Errorf("collecting stats for %s: %v", pair.container.ID, err) } continue } stats.SystemUsage = systemUsage pair.publisher.Publish(stats) } } } const nanoSeconds = 1e9 // getSystemCpuUSage returns the host system's cpu usage in nanoseconds // for the system to match the cgroup readings are returned in the same format. func (s *statsCollector) getSystemCpuUsage() (uint64, error) { var line string f, err := os.Open("/proc/stat") if err != nil { return 0, err } defer func() { s.bufReader.Reset(nil) f.Close() }() s.bufReader.Reset(f) err = nil for err == nil { line, err = s.bufReader.ReadString('\n') if err != nil { break } parts := strings.Fields(line) switch parts[0] { case "cpu": if len(parts) < 8 { return 0, fmt.Errorf("invalid number of cpu fields") } var sum uint64 for _, i := range parts[1:8] { v, err := strconv.ParseUint(i, 10, 64) if err != nil { return 0, fmt.Errorf("Unable to convert value %s to int: %s", i, err) } sum += v } return (sum * nanoSeconds) / s.clockTicks, nil } } return 0, fmt.Errorf("invalid stat format") }