Feature: Pre aggregate metrics to reduce cost (#33)

.azure-pipelines.yml

@@ -12,6 +12,9 @@ jobs:
       - script: go build
         displayName: Build
 
+      - script: go test ./...
+        displayName: Test
+
       - script: ls
 
       - task: CopyFiles@2
@@ -36,6 +39,9 @@ jobs:
       - script: go build
         displayName: Build
 
+      - script: go test ./...
+        displayName: Test
+
       - script: ls
 
       - task: CopyFiles@2

go.mod

@@ -12,5 +12,6 @@ require (
 	github.com/satori/go.uuid v1.2.0 // indirect
 	github.com/shirou/gopsutil v2.18.12+incompatible
 	github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 // indirect
+	github.com/stretchr/testify v1.3.0
 	golang.org/x/sys v0.0.0-20180907202204-917fdcba135d // indirect
 )

go.sum

@@ -7,6 +7,8 @@ github.com/NVIDIA/gpu-monitoring-tools v0.0.0-20181114021304-b70474fb8511 h1:A9x
 github.com/NVIDIA/gpu-monitoring-tools v0.0.0-20181114021304-b70474fb8511/go.mod h1:nMOvShGpWaf0bXwXmeu4k+O4uziuaEI8pWzIj3BUrOA=
 github.com/StackExchange/wmi v0.0.0-20180725035823-b12b22c5341f h1:5ZfJxyXo8KyX8DgGXC5B7ILL8y51fci/qYz2B4j8iLY=
 github.com/StackExchange/wmi v0.0.0-20180725035823-b12b22c5341f/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d h1:lDrio3iIdNb0Gw9CgH7cQF+iuB5mOOjdJ9ERNJCBgb4=
 github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/go-ole/go-ole v1.2.1 h1:2lOsA72HgjxAuMlKpFiCbHTvu44PIVkZ5hqm3RSdI/E=
@@ -17,11 +19,16 @@ github.com/mxpv/nvml-go v0.0.0-20180227003457-e07f8c26812d h1:lQo1zUtnGr52K2a+Ll
 github.com/mxpv/nvml-go v0.0.0-20180227003457-e07f8c26812d/go.mod h1:PS1oTOPfvtFjl9T7nduA/RYrIpqtRh2Nvk++rQCZ2q8=
 github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/shirou/gopsutil v2.18.12+incompatible h1:1eaJvGomDnH74/5cF4CTmTbLHAriGFsTZppLXDX93OM=
 github.com/shirou/gopsutil v2.18.12+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
 github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 h1:udFKJ0aHUL60LboW/A+DfgoHVedieIzIXE8uylPue0U=
 github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 golang.org/x/sys v0.0.0-20180907202204-917fdcba135d h1:kWn1hlsqeUrk6JsLJO0ZFyz9bMg8u85voZlIuc68ZU4=
 golang.org/x/sys v0.0.0-20180907202204-917fdcba135d/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

pkg/appinsights.go

@@ -1,66 +1,104 @@
 package batchinsights
 
 import (
+	"bytes"
+	"fmt"
 	"strconv"
+	"time"
 
 	"github.com/Microsoft/ApplicationInsights-Go/appinsights"
 )
 
 type AppInsightsService struct {
-	client appinsights.TelemetryClient
+	client                   appinsights.TelemetryClient
+	aggregateCollectionStart *time.Time
+	aggregates               map[string]*appinsights.AggregateMetricTelemetry
 }
 
+const AGGREGATE_TIME = time.Duration(1) * time.Minute
+
 func NewAppInsightsService(instrumentationKey string, poolId string, nodeId string) AppInsightsService {
 	client := appinsights.NewTelemetryClient(instrumentationKey)
 	client.Context().Tags.Cloud().SetRole(poolId)
 	client.Context().Tags.Cloud().SetRoleInstance(nodeId)
 
 	return AppInsightsService{
-		client: client,
+		client:     client,
+		aggregates: make(map[string]*appinsights.AggregateMetricTelemetry),
 	}
 }
 
-func (service AppInsightsService) UploadStats(stats NodeStats) {
+func (service *AppInsightsService) track(metric *appinsights.MetricTelemetry) {
+	t := time.Now()
+
+	if service.aggregateCollectionStart != nil {
+		elapsed := t.Sub(*service.aggregateCollectionStart)
+
+		if elapsed > AGGREGATE_TIME {
+			for k, aggregate := range service.aggregates {
+				fmt.Printf(" - %s: %f\n", k, aggregate.Value)
+				service.client.Track(aggregate)
+			}
+			service.aggregates = make(map[string]*appinsights.AggregateMetricTelemetry)
+			service.aggregateCollectionStart = &t
+		}
+	} else {
+		service.aggregateCollectionStart = &t
+	}
+
+	id := GetMetricId(metric)
+
+	aggregate, ok := service.aggregates[id]
+	if !ok {
+		aggregate = appinsights.NewAggregateMetricTelemetry(metric.Name)
+		aggregate.Properties = metric.Properties
+		service.aggregates[id] = aggregate
+	}
+	aggregate.AddData([]float64{metric.Value})
+}
+
+func (service *AppInsightsService) UploadStats(stats NodeStats) {
 	client := service.client
 
 	for cpuN, percent := range stats.cpuPercents {
 		metric := appinsights.NewMetricTelemetry("Cpu usage", percent)
 		metric.Properties["CPU #"] = strconv.Itoa(cpuN)
-		client.Track(metric)
+		metric.Properties["Core count"] = strconv.Itoa(len(stats.cpuPercents))
+		service.track(metric)
 	}
 
 	for _, usage := range stats.diskUsage {
 		usedMetric := appinsights.NewMetricTelemetry("Disk usage", float64(usage.Used))
 		usedMetric.Properties["Disk"] = usage.Path
-		client.Track(usedMetric)
+		service.track(usedMetric)
 		freeMetric := appinsights.NewMetricTelemetry("Disk free", float64(usage.Free))
 		freeMetric.Properties["Disk"] = usage.Path
-		client.Track(freeMetric)
+		service.track(freeMetric)
 	}
 
 	if stats.memory != nil {
-		client.TrackMetric("Memory used", float64(stats.memory.Used))
-		client.TrackMetric("Memory available", float64(stats.memory.Total-stats.memory.Used))
+		service.track(appinsights.NewMetricTelemetry("Memory used", float64(stats.memory.Used)))
+		service.track(appinsights.NewMetricTelemetry("Memory available", float64(stats.memory.Total-stats.memory.Used)))
 	}
 	if stats.diskIO != nil {
-		client.TrackMetric("Disk read", float64(stats.diskIO.ReadBps))
-		client.TrackMetric("Disk write", float64(stats.diskIO.WriteBps))
+		service.track(appinsights.NewMetricTelemetry("Disk read", float64(stats.diskIO.ReadBps)))
+		service.track(appinsights.NewMetricTelemetry("Disk write", float64(stats.diskIO.WriteBps)))
 	}
 
 	if stats.netIO != nil {
-		client.TrackMetric("Network read", float64(stats.netIO.ReadBps))
-		client.TrackMetric("Network write", float64(stats.netIO.WriteBps))
+		service.track(appinsights.NewMetricTelemetry("Network read", float64(stats.netIO.ReadBps)))
+		service.track(appinsights.NewMetricTelemetry("Network write", float64(stats.netIO.WriteBps)))
 	}
 
 	if len(stats.gpus) > 0 {
 		for cpuN, usage := range stats.gpus {
 			gpuMetric := appinsights.NewMetricTelemetry("Gpu usage", usage.GPU)
 			gpuMetric.Properties["GPU #"] = strconv.Itoa(cpuN)
-			client.Track(gpuMetric)
+			service.track(gpuMetric)
 
 			gpuMemoryMetric := appinsights.NewMetricTelemetry("Gpu memory usage", usage.Memory)
 			gpuMemoryMetric.Properties["GPU #"] = strconv.Itoa(cpuN)
-			client.Track(gpuMemoryMetric)
+			service.track(gpuMemoryMetric)
 		}
 	}
 
@@ -73,18 +111,37 @@ func (service AppInsightsService) UploadStats(stats NodeStats) {
 				cpuMetric := appinsights.NewMetricTelemetry("Process CPU", processStats.cpu)
 				cpuMetric.Properties["Process Name"] = processStats.name
 				cpuMetric.Properties["PID"] = pidStr
-				client.Track(cpuMetric)
+				service.track(cpuMetric)
 			}
 
 			{
 				memMetric := appinsights.NewMetricTelemetry("Process Memory", float64(processStats.memory))
 				memMetric.Properties["Process Name"] = processStats.name
 				memMetric.Properties["PID"] = pidStr
-				client.Track(memMetric)
+				service.track(memMetric)
 			}
 
 		}
 	}
 
 	client.Channel().Flush()
 }
+
+func GetMetricId(metric *appinsights.MetricTelemetry) string {
+	groupBy := createKeyValuePairs(metric.Properties)
+	return fmt.Sprintf("%s/%s", metric.Name, groupBy)
+}
+
+func createKeyValuePairs(m map[string]string) string {
+	b := new(bytes.Buffer)
+	first := true
+	for key, value := range m {
+		if first {
+			first = false
+		} else {
+			fmt.Fprintf(b, ",")
+		}
+		fmt.Fprintf(b, "%s=%s", key, value)
+	}
+	return b.String()
+}

pkg/appinsights_test.go

@@ -0,0 +1,19 @@
+package batchinsights_test
+
+import (
+	"github.com/Azure/batch-insights/pkg"
+	"github.com/Microsoft/ApplicationInsights-Go/appinsights"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestGetMetricId(t *testing.T) {
+	metric := appinsights.NewMetricTelemetry("Disk usage", 134)
+	metric.Properties["Some #"] = "4"
+	metric.Properties["Other #"] = "5"
+
+	assert.Equal(t, "Disk usage/Some #=4,Other #=5", batchinsights.GetMetricId(metric))
+
+	metric = appinsights.NewMetricTelemetry("Disk IO", 543)
+	assert.Equal(t, "Disk IO/", batchinsights.GetMetricId(metric))
+}

pkg/disk/disk_linux.go

@@ -22,7 +22,6 @@ func DiskIO() *utils.IOStats {
 	var writeBytes uint64 = 0
 
 	for _, v := range counters {
-		fmt.Println("stats", v.WriteBytes, v.WriteTime, v.WriteCount)
 		readBytes += v.ReadBytes
 		writeBytes += v.WriteBytes
 	}

pkg/wmi/wmi.go

@@ -1,3 +1,5 @@
+// +build windows
+
 package wmi
 
 import (

scripts/README.md

@@ -1,11 +1,27 @@
-## Script to be used as one liner
+# Script to be used as one liner
 
-# For linux
+### For linux
 
 * `run-linux.sh`: Run a published version for linux
 * `dev.sh`: Will install go, git then build and run on the fly
 
 
-# For windows
+### For windows
 * `run-windows.ps1`: Run a publush version for windows
-* `dev-windows.ps1`: Will install go, git then build and run on the fly
+* `dev-windows.ps1`: Will install go, git then build and run on the fly
+
+
+## Development
+There is some dev script that will install go and other needed dependencies to build and run this project on the fly.
+Set `BATCH_INSIGHTS_BRANCH` environment variable to the branch you are testing
+
+On linux 
+```bash
+/bin/bash -c 'wget  -O - https://raw.githubusercontent.com/Azure/batch-insights/$BATCH_INSIGHTS_BRANCH/scripts/dev.sh | bash'
+```
+
+On windows
+
+```powershell
+cmd /c @"%SystemRoot%\System32\WindowsPowerShell\v1.0\powershell.exe" -NoProfile -InputFormat None -ExecutionPolicy Bypass -Command "iex ((New-Object System.Net.WebClient).DownloadString('https://raw.githubusercontent.com/Azure/batch-insights/$env:BATCH_INSIGHTS_BRANCH/dev-windows.ps1'))"
+```

scripts/dev.sh

@@ -1,5 +1,8 @@
 set -e
 
+branch=$BATCH_INSIGHTS_BRANCH
+echo "Running Batch insights dev script for linux from branch $branch"
+
 apt-get update  
 apt-get install -y git binutils bison build-essential
 
@@ -13,7 +16,7 @@ export PATH=$GOPATH/bin:$GOROOT/bin:$PATH
 
 echo GO version $(go version)
 
-git clone https://github.com/Azure/batch-insights
+git clone https://github.com/Azure/batch-insights -b $branch
 
 cd batch-insights
 go build