Add service-specific resource monitoring and heartbeats

Author: jhkolb

.gitignore

@@ -4,3 +4,5 @@ spawnd/spawnd
 spawnctl/spawnctl
 .bw2bind.log
 containers/spawnd/spawnd
+*.yml
+spawnd/.manifests

service/types.go

@@ -5,8 +5,8 @@ type Configuration struct {
 	BaseImage           string   `yaml:"image"`
 	Source              string   `yaml:"source"`
 	BW2Entity           string   `yaml:"bw2Entity"`
-	CPUShares           uint32   `yaml:"cpuShares"`
-	Memory              uint32   `yaml:"memory"`
+	CPUShares           uint64   `yaml:"cpuShares"`
+	Memory              uint64   `yaml:"memory"`
 	Build               []string `yaml:"build,omitempty"`
 	Run                 []string `yaml:"run"`
 	IncludedFiles       []string `yaml:"includedFiles,omitempty"`

spawnd/backend/docker.go

@@ -5,11 +5,13 @@ import (
 	"bytes"
 	"context"
 	"encoding/base64"
+	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"path/filepath"
 	"strings"
+	"time"
 
 	"github.com/SoftwareDefinedBuildings/spawnpoint/service"
 	"github.com/docker/docker/api/types"
@@ -23,13 +25,35 @@ import (
 
 const defaultSpawnpointImage = "jhkolb/spawnable:amd64"
 const logMaxSize = "50m"
+const cpuSharesPerCore = 1024
 
 type Docker struct {
 	Alias     string
 	bw2Router string
 	client    *docker.Client
 }
 
+type dockerStatsResponse struct {
+	Read     time.Time `json:"read"`
+	PreRead  time.Time `json:"preread"`
+	CPUStats struct {
+		CPUUsage struct {
+			TotalUsage uint64 `json:"total_usage"`
+		} `json:"cpu_usage"`
+		SystemCPUUsage uint64 `json:"system_cpu_usage"`
+		OnlineCPUs     uint64 `json:"online_cpus"`
+	} `json:"cpu_stats"`
+	PreCPUStats struct {
+		CPUUsage struct {
+			TotalUsage uint64 `json:"total_usage"`
+		} `json:"cpu_usage"`
+		SystemCPUUsage uint64 `json:"system_cpu_usage"`
+	} `json:"precpu_stats"`
+	MemoryStats struct {
+		Usage uint64 `json:"usage"`
+	} `json:"memory_stats"`
+}
+
 func NewDocker(alias, bw2Router string) (*Docker, error) {
 	client, err := docker.NewEnvClient()
 	if err != nil {
@@ -208,6 +232,60 @@ func (dkr *Docker) ListServices(ctx context.Context) ([]string, error) {
 	return IDs, nil
 }
 
+func (dkr *Docker) ProfileService(ctx context.Context, id string, period time.Duration) (<-chan Stats, <-chan error) {
+	statChan := make(chan Stats, 10)
+	errChan := make(chan error, 1)
+	response, err := dkr.client.ContainerStats(ctx, id, true)
+	if err != nil {
+		close(statChan)
+		errChan <- errors.Wrap(err, "Failed to initialize container stat collection")
+		return statChan, errChan
+	}
+
+	go func() {
+		defer response.Body.Close()
+		decoder := json.NewDecoder(response.Body)
+		lastEmitted := time.Now()
+		lastCPUCores := 0.0
+		for {
+			select {
+			case <-ctx.Done():
+				close(statChan)
+				return
+			default:
+				var statEntry dockerStatsResponse
+				if err := decoder.Decode(&statEntry); err != nil {
+					close(statChan)
+					if err != io.EOF {
+						errChan <- errors.Wrap(err, "Failed to read and decode container stats entry")
+					}
+					return
+				}
+				if statEntry.Read.Sub(lastEmitted) > period {
+					lastEmitted = time.Now()
+
+					// Logic is based on Docker's `calculateCPUPercent` function
+					containerCPUDelta := float64(statEntry.CPUStats.CPUUsage.TotalUsage -
+						statEntry.PreCPUStats.CPUUsage.TotalUsage)
+					systemCPUDelta := float64(statEntry.CPUStats.SystemCPUUsage -
+						statEntry.PreCPUStats.SystemCPUUsage)
+					if systemCPUDelta > 0.0 {
+						numCores := float64(statEntry.CPUStats.OnlineCPUs)
+						lastCPUCores = (containerCPUDelta / systemCPUDelta) * numCores
+					}
+
+					statChan <- Stats{
+						Memory:    float64(statEntry.MemoryStats.Usage) / (1024.0 * 1024.0),
+						CPUShares: lastCPUCores * cpuSharesPerCore,
+					}
+				}
+			}
+		}
+	}()
+
+	return statChan, errChan
+}
+
 func (dkr *Docker) buildImage(ctx context.Context, svcConfig *service.Configuration) (string, error) {
 	buildCtxt, err := generateBuildContext(svcConfig)
 	if err != nil {

spawnd/backend/types.go

@@ -2,6 +2,7 @@ package backend
 
 import (
 	"context"
+	"time"
 
 	"github.com/SoftwareDefinedBuildings/spawnpoint/service"
 )
@@ -14,10 +15,16 @@ type ServiceBackend interface {
 	ListServices(ctx context.Context) ([]string, error)
 	TailService(ctx context.Context, id string, log bool) (<-chan string, <-chan error)
 	MonitorService(ctx context.Context, id string) (<-chan Event, <-chan error)
+	ProfileService(ctx context.Context, id string, period time.Duration) (<-chan Stats, <-chan error)
 }
 
 type Event int
 
 const (
 	Die = iota
 )
+
+type Stats struct {
+	Memory    float64
+	CPUShares float64
+}

spawnd/daemon/core.go

@@ -22,8 +22,8 @@ type Config struct {
 	BW2Entity string `yaml:"bw2Entity"`
 	BW2Agent  string `yaml:"bw2Agent"`
 	Path      string `yaml:"path"`
-	CPUShares uint32 `yaml:"cpuShares"`
-	Memory    uint32 `yaml:"memory"`
+	CPUShares uint64 `yaml:"cpuShares"`
+	Memory    uint64 `yaml:"memory"`
 	Backend   string `yaml:"backend"`
 }
 
@@ -34,10 +34,10 @@ type SpawnpointDaemon struct {
 	logger             *logging.Logger
 	path               string
 	alias              string
-	totalCPUShares     uint32
-	totalMemory        uint32
-	availableCPUShares uint32
-	availableMemory    uint32
+	totalCPUShares     uint64
+	totalMemory        uint64
+	availableCPUShares uint64
+	availableMemory    uint64
 	resourceLock       sync.RWMutex
 	serviceRegistry    map[string]*serviceManifest
 	registryLock       sync.RWMutex

spawnd/daemon/heartbeat.go

@@ -13,13 +13,21 @@ type Heartbeat struct {
 	Alias           string
 	Version         string
 	Time            int64
-	TotalMemory     uint32
-	TotalCPU        uint32
-	AvailableMemory uint32
-	AvailableCPU    uint32
+	TotalMemory     uint64
+	TotalCPU        uint64
+	AvailableMemory uint64
+	AvailableCPU    uint64
 	Services        []string
 }
 
+type ServiceHeartbeat struct {
+	Time          int64
+	Memory        uint64
+	CPUShares     uint64
+	UsedMemory    float64
+	UsedCPUShares float64
+}
+
 func (daemon *SpawnpointDaemon) publishHearbeats(ctx context.Context, delay time.Duration) {
 	bw2Iface := daemon.bw2Service.RegisterInterface("daemon", "i.spawnpoint")
 	tick := time.Tick(delay)
@@ -67,6 +75,40 @@ func (daemon *SpawnpointDaemon) publishHearbeats(ctx context.Context, delay time
 	}
 }
 
+func (daemon *SpawnpointDaemon) publishServiceHeartbeats(ctx context.Context, svc *serviceManifest, period time.Duration) {
+	statChan, errChan := daemon.backend.ProfileService(ctx, svc.ID, period)
+	bw2Iface := daemon.bw2Service.RegisterInterface(svc.Name, "i.spawnable")
+	for stats := range statChan {
+		daemon.logger.Debugf("(%s) Publishing service heartbeat", svc.Name)
+		daemon.logger.Debugf("(%s) CPU Shares: ~%.2f/%d, Memory: %.2f/%d", svc.Name,
+			stats.CPUShares, svc.CPUShares, stats.Memory, svc.Memory)
+		svcHb := ServiceHeartbeat{
+			Time:          time.Now().UnixNano(),
+			Memory:        svc.Memory,
+			CPUShares:     svc.CPUShares,
+			UsedMemory:    stats.Memory,
+			UsedCPUShares: stats.CPUShares,
+		}
+
+		po, err := bw2.CreateMsgPackPayloadObject(bw2.PONumSpawnpointSvcHb, svcHb)
+		if err != nil {
+			daemon.logger.Errorf("(%s) Failed to marshal service heartbeat: %s", svc.Name, err)
+			continue
+		}
+
+		if err := bw2Iface.PublishSignal("heartbeat", po); err != nil {
+			daemon.logger.Errorf("(%s) Failed to publish service heartbeat: %s", svc.Name, err)
+		}
+	}
+	daemon.logger.Debugf("(%s) Service heartbeat publication terminated", svc.Name)
+
+	select {
+	case err := <-errChan:
+		daemon.logger.Errorf("(%s) Error while profiling service: %s", svc.Name, err)
+	default:
+	}
+}
+
 func (daemon *SpawnpointDaemon) Decommission() error {
 	bw2Iface := daemon.bw2Service.RegisterInterface("daemon", "i.spawnpoint")
 	daemon.logger.Debugf("Decomissioning spawnpoint %s", daemon.alias)

spawnd/daemon/management.go

@@ -69,6 +69,7 @@ func (daemon *SpawnpointDaemon) manageService(svc *serviceManifest, done chan<-
 
 			go daemon.tailLogs(ctx, svc, true)
 			go daemon.monitorEvents(ctx, svc)
+			go daemon.publishServiceHeartbeats(ctx, svc, heartbeatInterval)
 
 		case service.Adopt:
 			daemon.logger.Debugf("(%s) State machine received service adopt event", svc.Name)
@@ -96,6 +97,7 @@ func (daemon *SpawnpointDaemon) manageService(svc *serviceManifest, done chan<-
 
 			go daemon.tailLogs(ctx, svc, false)
 			go daemon.monitorEvents(ctx, svc)
+			go daemon.publishServiceHeartbeats(ctx, svc, heartbeatInterval)
 
 		case service.Restart:
 			daemon.logger.Debugf("(%s) State machine received service restart event", svc.Name)
@@ -187,7 +189,7 @@ func (daemon *SpawnpointDaemon) monitorEvents(ctx context.Context, svc *serviceM
 	}
 	select {
 	case err := <-errChan:
-		fmt.Printf("(%s) Error while monitoring docker events: %s", svc.Name, err)
+		daemon.logger.Errorf("(%s) Error while monitoring docker events: %s", svc.Name, err)
 	default:
 	}
 }