Minimize expensive API calls

connector/docker.go

@@ -1,10 +1,10 @@
 package connector
 
 import (
- "fmt"
  "github.com/op/go-logging"
  "strings"
  "sync"
+ "time"
 
  "github.com/bcicen/ctop/connector/collector"
  "github.com/bcicen/ctop/connector/manager"
@@ -31,7 +31,8 @@ type StatusUpdate struct {
 type Docker struct {
  client *api.Client
  containers map[string]*container.Container
- needsRefresh chan string // container IDs requiring refresh
+ needsRefresh []string // container IDs requiring refresh
+ lastRefresh time.Time
  statuses chan StatusUpdate
  closed chan struct{}
  lock sync.RWMutex
@@ -44,12 +45,11 @@ func NewDocker() (Connector, error) {
  return nil, err
  }
  cm := &Docker{
- client: client,
- containers: make(map[string]*container.Container),
- needsRefresh: make(chan string, 60),
- statuses: make(chan StatusUpdate, 60),
- closed: make(chan struct{}),
- lock: sync.RWMutex{},
+ client: client,
+ containers: make(map[string]*container.Container),
+ statuses: make(chan StatusUpdate, 60),
+ closed: make(chan struct{}),
+ lock: sync.RWMutex{},
  }
 
  // query info as pre-flight healthcheck
@@ -109,7 +109,11 @@ func (cm *Docker) watchEvents() {
  if log.IsEnabledFor(logging.DEBUG) {
  log.Debugf("handling docker event: action=create id=%s", e.ID)
  }
- cm.needsRefresh <- e.ID
+ c := cm.MustGet(e.ID)
+ c.SetMeta("name", manager.ShortName(e.Actor.Attributes["name"]))
+ c.SetMeta("image", e.Actor.Attributes["image"])
+ c.SetState("created")
+ cm.requestRefresh(c)
  case "destroy":
  if log.IsEnabledFor(logging.DEBUG) {
  log.Debugf("handling docker event: action=destroy id=%s", e.ID)
@@ -130,92 +134,99 @@ func (cm *Docker) watchEvents() {
  close(cm.closed)
 }
 
-func portsFormat(ports map[api.Port][]api.PortBinding) string {
- var exposed []string
- var published []string
+// Mark all container IDs for refresh
+func (cm *Docker) refreshAll() {
+ cm.updateContainers(true)
+}
 
- for k, v := range ports {
- if len(v) == 0 {
- exposed = append(exposed, string(k))
- continue
- }
- for _, binding := range v {
- s := fmt.Sprintf("%s:%s -> %s", binding.HostIP, binding.HostPort, k)
- published = append(published, s)
+func (cm *Docker) updateContainers(all bool) {
+ opts := api.ListContainersOptions{All: true}
+ if !all {
+ opts.Filters = map[string][]string{
+ "id": cm.needsRefresh,
  }
  }
+ allContainers, err := cm.client.ListContainers(opts)
+ if err != nil {
+ log.Errorf("%s (%T)", err.Error(), err)
+ return
+ }
+ if all {
+ cm.cleanupDestroyedContainers(allContainers)
+ }
 
- return strings.Join(append(exposed, published...), "\n")
-}
-
-func ipsFormat(networks map[string]api.ContainerNetwork) string {
- var ips []string
-
- for k, v := range networks {
- s := fmt.Sprintf("%s:%s", k, v.IPAddress)
- ips = append(ips, s)
+ for _, i := range allContainers {
+ c := cm.MustGet(i.ID)
+ c.SetMeta("name", manager.ShortName(i.Names[0]))
+ c.SetMeta("image", i.Image)
+  c.SetMeta("IPs", manager.IpsFormat(i.Networks.Networks))
+ c.SetMeta("ports", manager.PortsFormatArr(i.Ports))
+  c.SetMeta("created", time.Unix(i.Created, 0).Format("Mon Jan 2 15:04:05 2006"))
+ parseStatusHealth(c, i.Status)
+ c.SetState(i.State)
  }
+ cm.lastRefresh = time.Now()
+ cm.needsRefresh = nil
+}
 
- return strings.Join(ips, "\n")
+func (cm *Docker) requestRefresh(c *container.Container) {
+ cm.needsRefresh = append(cm.needsRefresh, c.Id)
+ refreshRequestedOn := time.Now()
+ go func() {
+ time.Sleep(5 * time.Second)
+ if refreshRequestedOn.Before(cm.lastRefresh) {
+ return
+ }
+ // batch refresh
+ cm.updateContainers(false)
+ }()
 }
 
-func (cm *Docker) refresh(c *container.Container) {
- insp, found, failed := cm.inspect(c.Id)
- if failed {
- return
+func (cm *Docker) cleanupDestroyedContainers(allContainers []api.APIContainers) {
+ var nonExistingContainers []string
+ for _, oldContainer := range cm.containers {
+ if !cm.hasContainer(oldContainer.Id, allContainers) {
+ nonExistingContainers = append(nonExistingContainers, oldContainer.Id)
+ }
  }
- // remove container if no longer exists
- if !found {
- cm.delByID(c.Id)
- return
+ // remove containers that no longer exists
+ for _, cid := range nonExistingContainers {
+ cm.delByID(cid)
  }
- c.SetMeta("name", shortName(insp.Name))
- c.SetMeta("image", insp.Config.Image)
- c.SetMeta("IPs", ipsFormat(insp.NetworkSettings.Networks))
- c.SetMeta("ports", portsFormat(insp.NetworkSettings.Ports))
- c.SetMeta("created", insp.Created.Format("Mon Jan 2 15:04:05 2006"))
- c.SetMeta("health", insp.State.Health.Status)
- c.SetMeta("[ENV-VAR]", strings.Join(insp.Config.Env, ";"))
- c.SetState(insp.State.Status)
 }
 
-func (cm *Docker) inspect(id string) (insp *api.Container, found bool, failed bool) {
- c, err := cm.client.InspectContainer(id)
- if err != nil {
- if _, notFound := err.(*api.NoSuchContainer); notFound {
- return c, false, false
+func (cm *Docker) hasContainer(oldContainerId string, newContainers []api.APIContainers) bool {
+ for _, newContainer := range newContainers {
+ if newContainer.ID == oldContainerId {
+ return true
  }
- // other error e.g. connection failed
- log.Errorf("%s (%T)", err.Error(), err)
- return c, false, true
  }
- return c, true, false
+ return false
 }
 
-// Mark all container IDs for refresh
-func (cm *Docker) refreshAll() {
- opts := api.ListContainersOptions{All: true}
- allContainers, err := cm.client.ListContainers(opts)
- if err != nil {
- log.Errorf("%s (%T)", err.Error(), err)
+func parseStatusHealth(c *container.Container, status string) {
+ // Status may look like:
+ // Up About a minute (healthy)
+ // Up 7 minutes (unhealthy)
+ var health string
+ if strings.Contains(status, "(healthy)") {
+ health = "healthy"
+ } else if strings.Contains(status, "(unhealthy)") {
+ health = "unhealthy"
+ } else {
  return
  }
-
- for _, i := range allContainers {
- c := cm.MustGet(i.ID)
- c.SetMeta("name", shortName(i.Names[0]))
- c.SetState(i.State)
- cm.needsRefresh <- c.Id
- }
+ c.SetMeta("health", health)
 }
 
 func (cm *Docker) Loop() {
+ ticker := time.NewTicker(5 * time.Minute)
  for {
  select {
- case id := <-cm.needsRefresh:
- c := cm.MustGet(id)
- cm.refresh(c)
+ case <-ticker.C:
+ cm.refreshAll()
  case <-cm.closed:
+ ticker.Stop()
  return
  }
  }
@@ -285,8 +296,3 @@ func (cm *Docker) All() (containers container.Containers) {
  cm.lock.Unlock()
  return containers
 }
-
-// use primary container name
-func shortName(name string) string {
- return strings.TrimPrefix(name, "/")
-}

connector/manager/docker.go

@@ -2,10 +2,17 @@ package manager
 
 import (
  "fmt"
+ "github.com/bcicen/ctop/logging"
+ "github.com/bcicen/ctop/models"
  api "github.com/fsouza/go-dockerclient"
  "github.com/pkg/errors"
  "io"
  "os"
+ "strings"
+)
+
+var (
+ log = logging.Init()
 )
 
 type Docker struct {
@@ -102,6 +109,37 @@ func (dc *Docker) Exec(cmd []string) error {
  })
 }
 
+func (dc *Docker) inspect(id string) (insp *api.Container, found bool, err error) {
+ c, err := dc.client.InspectContainer(id)
+ if err != nil {
+ if _, notFound := err.(*api.NoSuchContainer); notFound {
+ return c, false, nil
+ }
+ // other error e.g. connection failed
+ log.Errorf("%s (%T)", err.Error(), err)
+ return c, false, err
+ }
+ return c, true, nil
+}
+
+func (dc *Docker) Inspect() (models.Meta, error) {
+ insp, found, err := dc.inspect(dc.id)
+ if !found {
+ return nil, err
+ }
+ newMeta := models.Meta{}
+ newMeta["name"] = ShortName(insp.Name)
+ newMeta["image"] = insp.Config.Image
+ newMeta["IPs"] = IpsFormat(insp.NetworkSettings.Networks)
+ newMeta["ports"] = PortsFormat(insp.NetworkSettings.Ports)
+ newMeta["created"] = insp.Created.Format("Mon Jan 2 15:04:05 2006")
+ newMeta["health"] = insp.State.Health.Status
+ newMeta["[ENV-VAR]"] = strings.Join(insp.Config.Env, ";")
+ newMeta["state"] = insp.State.Status
+
+ return newMeta, nil
+}
+
 func (dc *Docker) Start() error {
  c, err := dc.client.InspectContainer(dc.id)
  if err != nil {

connector/manager/docker_utils.go

@@ -0,0 +1,61 @@
+package manager
+
+import (
+ "fmt"
+ api "github.com/fsouza/go-dockerclient"
+ "strings"
+)
+
+func PortsFormat(ports map[api.Port][]api.PortBinding) string {
+ var exposed []string
+ var published []string
+
+ for k, v := range ports {
+ if len(v) == 0 {
+ // 3306/tcp
+ exposed = append(exposed, string(k))
+ continue
+ }
+ for _, binding := range v {
+ // 0.0.0.0:3307 -> 3306/tcp
+ s := fmt.Sprintf("%s:%s -> %s", binding.HostIP, binding.HostPort, k)
+ published = append(published, s)
+ }
+ }
+
+ return strings.Join(append(exposed, published...), "\n")
+}
+
+func PortsFormatArr(ports []api.APIPort) string {
+ var exposed []string
+ var published []string
+ for _, binding := range ports {
+ if binding.PublicPort != 0 {
+ // 0.0.0.0:3307 -> 3306/tcp
+ s := fmt.Sprintf("%s:%d -> %d/%s", binding.IP, binding.PublicPort, binding.PrivatePort, binding.Type)
+ published = append(published, s)
+ } else {
+ // 3306/tcp
+ s := fmt.Sprintf("%d/%s", binding.PrivatePort, binding.Type)
+ exposed = append(exposed, s)
+ }
+ }
+
+ return strings.Join(append(exposed, published...), "\n")
+}
+
+func IpsFormat(networks map[string]api.ContainerNetwork) string {
+ var ips []string
+
+ for k, v := range networks {
+ s := fmt.Sprintf("%s:%s", k, v.IPAddress)
+ ips = append(ips, s)
+ }
+
+ return strings.Join(ips, "\n")
+}
+
+// use primary container name
+func ShortName(name string) string {
+ return strings.TrimPrefix(name, "/")
+}

connector/manager/main.go

@@ -1,6 +1,9 @@
 package manager
 
-import "errors"
+import (
+ "errors"
+ "github.com/bcicen/ctop/models"
+)
 
 var ActionNotImplErr = errors.New("action not implemented")
 
@@ -12,4 +15,5 @@ type Manager interface {
  Unpause() error
  Restart() error
  Exec(cmd []string) error
+ Inspect() (models.Meta, error)
 }

connector/manager/mock.go

@@ -1,5 +1,7 @@
 package manager
 
+import models "github.com/bcicen/ctop/models"
+
 type Mock struct{}
 
 func NewMock() *Mock {
@@ -33,3 +35,7 @@ func (m *Mock) Restart() error {
 func (m *Mock) Exec(cmd []string) error {
  return ActionNotImplErr
 }
+
+func (m *Mock) Inspect() (models.Meta, error) {
+ return nil, nil
+}

connector/manager/runc.go

@@ -1,5 +1,7 @@
 package manager
 
+import "github.com/bcicen/ctop/models"
+
 type Runc struct{}
 
 func NewRunc() *Runc {
@@ -33,3 +35,7 @@ func (rc *Runc) Restart() error {
 func (rc *Runc) Exec(cmd []string) error {
  return ActionNotImplErr
 }
+
+func (rc *Runc) Inspect() (models.Meta, error) {
+ return nil, nil
+}