Merge branch 'master' into wip_defrag

Author: JamesMurkin

.mergify.yml

@@ -9,6 +9,6 @@ pull_request_rules:
           - "#approved-reviews-by>=2"
           - and: 
             - "#approved-reviews-by>=1"
-            - "author~=^(d80tb7|dave[-]gantenbein|dejanzele|eleanorpratt|geaere|JamesMurkin|mauriceyap|masipauskas|MustafaI|zuqq|richscott|robertdavidsmith|samclark|suprjinx)"
+            - "author~=^(d80tb7|dave[-]gantenbein|dejanzele|eleanorpratt|geaere|JamesMurkin|mauriceyap|masipauskas|MustafaI|zuqq|richscott|robertdavidsmith|samclark|suprjinx|EnricoMi)$"
         title:
-          Two are checks required.
+          Two approvals required, or one if author is a maintainer.

cmd/lookout/dbloadtester/config.yaml

@@ -0,0 +1,10 @@
+#sample config
+---
+queues:
+  - "queue1"
+ids:
+  - "01JD2FF8TTJAXCNFB0F91053ZD"
+jobsets:
+  - "jobset1"
+queueCounts:
+  queue1: 1000000

cmd/lookout/dbloadtester/main.go

@@ -0,0 +1,85 @@
+package main
+
+import (
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/spf13/pflag"
+	"github.com/spf13/viper"
+
+	"github.com/armadaproject/armada/internal/common"
+	"github.com/armadaproject/armada/internal/common/armadacontext"
+	log "github.com/armadaproject/armada/internal/common/logging"
+	"github.com/armadaproject/armada/internal/common/profiling"
+	"github.com/armadaproject/armada/internal/lookout/configuration"
+	"github.com/armadaproject/armada/internal/lookout/dbloadtester"
+)
+
+const (
+	customConfigLocation string = "config"
+	customParamsLocation string = "params"
+)
+
+func init() {
+	pflag.StringSlice(
+		customConfigLocation,
+		[]string{},
+		"path to the configuration for the Lookout under test",
+	)
+	pflag.StringSlice(
+		customParamsLocation,
+		[]string{},
+		"parameters to the load test, defining the shape of the load",
+	)
+	pflag.Parse()
+}
+
+func makeContext() (*armadacontext.Context, func()) {
+	ctx := armadacontext.Background()
+	ctx, cancel := armadacontext.WithCancel(ctx)
+
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
+
+	go func() {
+		select {
+		case <-c:
+			cancel()
+		case <-ctx.Done():
+		}
+	}()
+
+	return ctx, func() {
+		signal.Stop(c)
+		cancel()
+	}
+}
+
+func main() {
+	common.BindCommandlineArguments()
+
+	var config configuration.LookoutConfig
+	userSpecifiedConfigs := viper.GetStringSlice(customConfigLocation)
+	common.LoadConfig(&config, "./config/lookout", userSpecifiedConfigs)
+
+	var args dbloadtester.ReadTestConfig
+	argsInput := viper.GetStringSlice(customParamsLocation)
+	common.LoadConfig(&args, "./cmd/lookout/dbloadtester", argsInput)
+
+	// Expose profiling endpoints if enabled.
+	err := profiling.SetupPprof(config.Profiling, armadacontext.Background(), nil)
+	if err != nil {
+		log.Fatalf("Pprof setup failed, exiting, %v", err)
+	}
+
+	ctx, cleanup := makeContext()
+	defer cleanup()
+
+	results, err := dbloadtester.DoQueries(config, args, ctx)
+	if err != nil {
+		log.Fatalf("error running queries, exiting, %v", err)
+	}
+
+	log.Infof("results: \n%v", results)
+}

cmd/lookoutingester/dbloadtester/main.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"fmt"
 	"time"
 
 	"github.com/spf13/pflag"
@@ -19,7 +18,7 @@ func init() {
 	pflag.StringSlice(
 		"lookoutIngesterConfig",
 		[]string{},
-		"Fully qualified path to application configuration file (for multiple config files repeat this arg or separate paths with commas)",
+		"path to the configuration for the lookout ingester under test",
 	)
 	pflag.Parse()
 }
@@ -74,7 +73,7 @@ func main() {
 	if err != nil {
 		log.Warn("Failed to marshal lookout ingester config for report output")
 	}
-	fmt.Printf(
+	log.Infof(
 		ReportTemplate,
 		time.Now().Format("2006-01-02"),
 		loadtesterConfig.TotalJobs,

internal/common/resource/resource.go

@@ -204,6 +204,19 @@ func (a ComputeResources) Sub(b ComputeResources) {
 	}
 }
 
+func (a ComputeResources) Div(b map[string]int64) {
+	for k, v := range b {
+		existing, ok := a[k]
+		if ok {
+			if existing.Format == resource.DecimalSI {
+				a[k] = resource.NewMilliQuantity(existing.MilliValue()/v, existing.Format).DeepCopy()
+			} else {
+				a[k] = resource.NewQuantity(existing.Value()/v, existing.Format).DeepCopy()
+			}
+		}
+	}
+}
+
 func (a ComputeResources) DeepCopy() ComputeResources {
 	targetComputeResource := make(ComputeResources)
 

internal/executor/domain/resources.go

@@ -14,11 +14,6 @@ func EmptyUtilisationData() *UtilisationData {
 	}
 }
 
-func (a *UtilisationData) Max(b *UtilisationData) {
-	a.CurrentUsage.Max(b.CurrentUsage)
-	a.CumulativeUsage.Max(b.CumulativeUsage)
-}
-
 func (u *UtilisationData) DeepCopy() *UtilisationData {
 	return &UtilisationData{
 		CurrentUsage:    u.CurrentUsage.DeepCopy(),
@@ -29,3 +24,82 @@ func (u *UtilisationData) DeepCopy() *UtilisationData {
 func (u *UtilisationData) IsEmpty() bool {
 	return len(u.CumulativeUsage) == 0 && len(u.CurrentUsage) == 0
 }
+
+type UtilisationDataAggregation struct {
+	maxUsage        armadaresource.ComputeResources
+	sumOfUsage      armadaresource.ComputeResources
+	numDataPoints   map[string]int64
+	cumulativeUsage armadaresource.ComputeResources
+}
+
+func EmptyUtilisationDataAggregation() *UtilisationDataAggregation {
+	return &UtilisationDataAggregation{
+		maxUsage:        armadaresource.ComputeResources{},
+		sumOfUsage:      armadaresource.ComputeResources{},
+		numDataPoints:   map[string]int64{},
+		cumulativeUsage: armadaresource.ComputeResources{},
+	}
+}
+
+func NewUtilisationDataAggregation(u *UtilisationData) *UtilisationDataAggregation {
+	numDataPoints := make(map[string]int64)
+
+	for k := range u.CurrentUsage {
+		numDataPoints[k] = 1
+	}
+
+	return &UtilisationDataAggregation{
+		maxUsage:        u.CurrentUsage.DeepCopy(),
+		sumOfUsage:      u.CurrentUsage.DeepCopy(),
+		numDataPoints:   numDataPoints,
+		cumulativeUsage: u.CumulativeUsage.DeepCopy(),
+	}
+}
+
+func (u *UtilisationDataAggregation) GetMaxUsage() armadaresource.ComputeResources {
+	return u.maxUsage
+}
+
+func (u *UtilisationDataAggregation) GetAvgUsage() armadaresource.ComputeResources {
+	avg := u.sumOfUsage.DeepCopy()
+	avg.Div(u.numDataPoints)
+	return avg
+}
+
+func (u *UtilisationDataAggregation) GetCumulativeUsage() armadaresource.ComputeResources {
+	return u.cumulativeUsage
+}
+
+func (a *UtilisationDataAggregation) Add(b *UtilisationData) {
+	a.maxUsage.Max(b.CurrentUsage)
+	a.sumOfUsage.Add(b.CurrentUsage)
+	a.cumulativeUsage.Max(b.CumulativeUsage)
+
+	for k := range b.CurrentUsage {
+		existing, ok := a.numDataPoints[k]
+		if ok {
+			a.numDataPoints[k] = existing + 1
+		} else {
+			a.numDataPoints[k] = 1
+		}
+	}
+}
+
+func (u *UtilisationDataAggregation) DeepCopy() *UtilisationDataAggregation {
+	numDataPoints := map[string]int64{}
+
+	for k, v := range u.numDataPoints {
+		numDataPoints[k] = v
+	}
+
+	return &UtilisationDataAggregation{
+		maxUsage:        u.maxUsage.DeepCopy(),
+		sumOfUsage:      u.sumOfUsage.DeepCopy(),
+		numDataPoints:   numDataPoints,
+		cumulativeUsage: u.cumulativeUsage.DeepCopy(),
+	}
+}
+
+func (u *UtilisationDataAggregation) IsEmpty() bool {
+	return len(u.maxUsage) == 0
+}

internal/executor/domain/resources_test.go

@@ -9,37 +9,76 @@ import (
 	armadaresource "github.com/armadaproject/armada/internal/common/resource"
 )
 
-func TestUtilisationData_Max(t *testing.T) {
-	data := &UtilisationData{
-		CurrentUsage:    armadaresource.ComputeResources{"cpu": resource.MustParse("1"), "memory": resource.MustParse("10")},
-		CumulativeUsage: armadaresource.ComputeResources{"cpu": resource.MustParse("5")},
+func compareQuantityMaps(a, b map[string]resource.Quantity) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for key, aVal := range a {
+		bVal, exists := b[key]
+		if !exists {
+			return false
+		}
+		if aVal.String() != bVal.String() {
+			return false
+		}
+	}
+	return true
+}
+
+func TestUtilisationDataAggregation(t *testing.T) {
+	data := &UtilisationDataAggregation{
+		maxUsage:        armadaresource.ComputeResources{"cpu": resource.MustParse("1"), "memory": resource.MustParse("10")},
+		sumOfUsage:      armadaresource.ComputeResources{"cpu": resource.MustParse("1"), "memory": resource.MustParse("10")},
+		numDataPoints:   map[string]int64{"cpu": 1, "memory": 1},
+		cumulativeUsage: armadaresource.ComputeResources{"cpu": resource.MustParse("5")},
 	}
 	data2 := &UtilisationData{
 		CurrentUsage:    armadaresource.ComputeResources{"cpu": resource.MustParse("2"), "memory": resource.MustParse("1")},
 		CumulativeUsage: armadaresource.ComputeResources{"cpu": resource.MustParse("10")},
 	}
-	expected := &UtilisationData{
-		CurrentUsage:    armadaresource.ComputeResources{"cpu": resource.MustParse("2"), "memory": resource.MustParse("10")},
-		CumulativeUsage: armadaresource.ComputeResources{"cpu": resource.MustParse("10")},
+	expected := &UtilisationDataAggregation{
+		maxUsage:        armadaresource.ComputeResources{"cpu": resource.MustParse("2"), "memory": resource.MustParse("10")},
+		sumOfUsage:      armadaresource.ComputeResources{"cpu": resource.MustParse("3"), "memory": resource.MustParse("11")},
+		numDataPoints:   map[string]int64{"cpu": 2, "memory": 2},
+		cumulativeUsage: armadaresource.ComputeResources{"cpu": resource.MustParse("10")},
 	}
 
-	max := data
-	max.Max(data2)
-	assert.Equal(t, expected.CurrentUsage, max.CurrentUsage)
-	assert.Equal(t, expected.CumulativeUsage, max.CumulativeUsage)
+	aggregated := data
+	aggregated.Add(data2)
+	assert.True(t, compareQuantityMaps(expected.maxUsage, aggregated.maxUsage))
+	assert.True(t, compareQuantityMaps(expected.sumOfUsage, aggregated.sumOfUsage))
+	assert.Equal(t, expected.numDataPoints, aggregated.numDataPoints)
+	assert.True(t, compareQuantityMaps(expected.cumulativeUsage, aggregated.cumulativeUsage))
 }
 
-func TestUtilisationData_Max_WithEmpty(t *testing.T) {
+func TestUtilisationDataAggregation_WithEmpty(t *testing.T) {
 	currentUsage := armadaresource.ComputeResources{"cpu": resource.MustParse("1"), "memory": resource.MustParse("1")}
 	cumulativeUsage := armadaresource.ComputeResources{"cpu": resource.MustParse("10")}
 	data := &UtilisationData{
 		CurrentUsage:    currentUsage.DeepCopy(),
 		CumulativeUsage: cumulativeUsage.DeepCopy(),
 	}
-	max := EmptyUtilisationData()
-	max.Max(data)
-	assert.Equal(t, data.CurrentUsage, max.CurrentUsage)
-	assert.Equal(t, data.CumulativeUsage, max.CumulativeUsage)
+	aggregated := EmptyUtilisationDataAggregation()
+	aggregated.Add(data)
+	assert.Equal(t, data.CurrentUsage, aggregated.maxUsage)
+	assert.Equal(t, data.CumulativeUsage, aggregated.cumulativeUsage)
+
+	assert.True(t, compareQuantityMaps(data.CurrentUsage, aggregated.maxUsage))
+	assert.True(t, compareQuantityMaps(data.CurrentUsage, aggregated.sumOfUsage))
+	assert.True(t, compareQuantityMaps(data.CumulativeUsage, aggregated.cumulativeUsage))
+}
+
+func TestUtilisationData_GetAvg(t *testing.T) {
+	data := &UtilisationDataAggregation{
+		maxUsage:        armadaresource.ComputeResources{"cpu": resource.MustParse("100m"), "memory": resource.MustParse("100Mi")},
+		sumOfUsage:      armadaresource.ComputeResources{"cpu": resource.MustParse("160m"), "memory": resource.MustParse("110Mi")},
+		numDataPoints:   map[string]int64{"cpu": 2, "memory": 2},
+		cumulativeUsage: armadaresource.ComputeResources{"cpu": resource.MustParse("160m")},
+	}
+
+	expected := armadaresource.ComputeResources{"cpu": resource.MustParse("80m"), "memory": resource.MustParse("55Mi")}
+
+	assert.True(t, compareQuantityMaps(expected, data.GetAvgUsage()))
 }
 
 func TestUtilisationData_IsEmpty(t *testing.T) {

internal/executor/reporter/event.go

@@ -318,7 +318,7 @@ func CreateReturnLeaseEvent(pod *v1.Pod, reason string, debugMessage string, clu
 	return sequence, nil
 }
 
-func CreateJobUtilisationEvent(pod *v1.Pod, utilisationData *domain.UtilisationData, clusterId string) (*armadaevents.EventSequence, error) {
+func CreateJobUtilisationEvent(pod *v1.Pod, utilisationData *domain.UtilisationDataAggregation, clusterId string) (*armadaevents.EventSequence, error) {
 	sequence := createEmptySequence(pod)
 	jobId, runId, err := extractIds(pod)
 	if err != nil {
@@ -345,8 +345,9 @@ func CreateJobUtilisationEvent(pod *v1.Pod, utilisationData *domain.UtilisationD
 						},
 					},
 				},
-				MaxResourcesForPeriod: utilisationData.CurrentUsage.ToProtoMap(),
-				TotalCumulativeUsage:  utilisationData.CumulativeUsage.ToProtoMap(),
+				MaxResourcesForPeriod: utilisationData.GetMaxUsage().ToProtoMap(),
+				AvgResourcesForPeriod: utilisationData.GetAvgUsage().ToProtoMap(),
+				TotalCumulativeUsage:  utilisationData.GetCumulativeUsage().ToProtoMap(),
 			},
 		},
 	})