Initialise New Scheduler

cmd/scheduler/cmd/main.go

@@ -0,0 +1,24 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+
+	"github.com/armadaproject/armada/internal/scheduler"
+)
+
+func runCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "run",
+		Short: "Runs the scheduler",
+		RunE:  runScheduler,
+	}
+	return cmd
+}
+
+func runScheduler(_ *cobra.Command, _ []string) error {
+	config, err := loadConfig()
+	if err != nil {
+		return err
+	}
+	return scheduler.Run(config)
+}

cmd/scheduler/cmd/migrate_database.go

@@ -0,0 +1,43 @@
+package cmd
+
+import (
+	"context"
+	"time"
+
+	"github.com/pkg/errors"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+
+	"github.com/armadaproject/armada/internal/common/database"
+	schedulerdb "github.com/armadaproject/armada/internal/scheduler/database"
+)
+
+func migrateDbCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "migrateDatabase",
+		Short: "migrates the scheduler database to the latest version",
+		RunE:  migrateDatabase,
+	}
+	return cmd
+}
+
+func migrateDatabase(_ *cobra.Command, _ []string) error {
+	config, err := loadConfig()
+	if err != nil {
+		return err
+	}
+	start := time.Now()
+	log.Info("Beginning scheduler database migration")
+	db, err := database.OpenPgxConn(config.Postgres)
+	if err != nil {
+		return errors.Wrapf(err, "Failed to connect to database")
+	}
+	err = schedulerdb.Migrate(context.Background(), db)
+	if err != nil {
+		return errors.Wrapf(err, "Failed to migrate scheduler database")
+	}
+	taken := time.Now().Sub(start)
+	log.Infof("Scheduler database migrated in %s", taken)
+	return nil
+}

cmd/scheduler/cmd/root.go

@@ -0,0 +1,48 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+
+	"github.com/armadaproject/armada/internal/common"
+	commonconfig "github.com/armadaproject/armada/internal/common/config"
+	"github.com/armadaproject/armada/internal/scheduler"
+)
+
+const (
+	CustomConfigLocation string = "config"
+)
+
+func RootCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:          "scheduler",
+		SilenceUsage: true,
+		Short:        "The main armada scheduler",
+	}
+
+	cmd.PersistentFlags().StringSlice(
+		"armadaUrl",
+		[]string{},
+		"Fully qualified path to application configuration file (for multiple config files repeat this arg or separate paths with commas)")
+
+	cmd.AddCommand(
+		runCmd(),
+		migrateDbCmd(),
+	)
+
+	return cmd
+}
+
+func loadConfig() (scheduler.Configuration, error) {
+	var config scheduler.Configuration
+	userSpecifiedConfigs := viper.GetStringSlice(CustomConfigLocation)
+
+	common.LoadConfig(&config, "./config/scheduler", userSpecifiedConfigs)
+
+	// TODO: once we're happy with this we can move it to common app startup
+	err := commonconfig.Validate(config)
+	if err != nil {
+		commonconfig.LogValidationErrors(err)
+	}
+	return config, err
+}

cmd/scheduler/main.go

@@ -11,6 +11,7 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/armadaproject/armada/internal/common"
+	commonconfig "github.com/armadaproject/armada/internal/common/config"
 	"github.com/armadaproject/armada/internal/common/database"
 	"github.com/armadaproject/armada/internal/scheduler"
 	schedulerdb "github.com/armadaproject/armada/internal/scheduler/database"
@@ -40,11 +41,18 @@ func main() {
 
 	common.LoadConfig(&config, "./config/scheduler", userSpecifiedConfigs)
 
+	// TODO: once we're happy wirth this we can move it to common app startup
+	err := commonconfig.Validate(config)
+	if err != nil {
+		commonconfig.LogValidationErrors(err)
+		os.Exit(1)
+	}
+
 	if viper.GetBool(MigrateDatabase) {
 		migrateDatabase(&config)
 	} else {
-		if err := scheduler.Run(&config); err != nil {
-			log.Errorf("failed to run scheduler: %s", err)
+		if err := scheduler.Run(config); err != nil {
+			log.WithError(err).Errorf("Scheduler failed")
 			os.Exit(1)
 		}
 	}

config/scheduler/config.yaml

@@ -1,25 +1,94 @@
+cyclePeriod: 10s
+executorTimeout: 1h
+databaseFetchSize: 1000
+pulsarSendTimeout: 5s
+pulsar:
+  URL: "pulsar://pulsar:6650"
+  jobsetEventsTopic: "events"
+  maxConnectionsPerBroker: 1
+  compressionType: zlib
+  compressionLevel: faster
 postgres:
-  maxOpenConns: 20
-  maxIdleConns: 5
-  connMaxLifetime: 30m
   connection:
-    host: localhost
+    host: postgres
     port: 5432
     user: postgres
     password: psw
     dbname: postgres
     sslmode: disable
-
-metrics:
-  port: 9003
-
-pulsar:
-  URL: "pulsar://localhost:6650"
-  jobsetEventsTopic: "events"
-  receiveTimeout: 5s
-  backoffTime: 1s
-
-subscriptionName: "scheduler"
-batchSize: 10000
-batchDuration: 500ms
+leader:
+  mode: standalone
+grpc:
+  #port: 50052
+  keepaliveParams:
+    maxConnectionIdle: 5m
+    time: 120s
+    timeout: 20s
+  keepaliveEnforcementPolicy:
+    minTime: 10s
+    permitWithoutStream: true
+scheduling:
+  preemption:
+    enabled: true
+    priorityClasses:
+      armada-default:
+        priority: 1000
+        maximalResourceFractionPerQueue:
+          memory: 0.99
+          cpu: 0.99
+      armada-preemptible:
+        priority: 900
+        maximalResourceFractionPerQueue:
+          memory: 0.99
+          cpu: 0.99
+    defaultPriorityClass: armada-default
+  queueLeaseBatchSize: 1000
+  minimumResourceToSchedule:
+    memory: 1000000 # 1Mb
+    cpu: 0.1
+  maximalResourceFractionToSchedulePerQueue:
+    memory: 1.0
+    cpu: 1.0
+  maximalResourceFractionPerQueue:
+    memory: 1.0
+    cpu: 1.0
+  maximalClusterFractionToSchedule:
+    memory: 1.0
+    cpu: 1.0
+  maximumJobsToSchedule: 5000
+  maxQueueReportsToStore: 1000
+  MaxJobReportsToStore: 10000
+  defaultJobLimits:
+    cpu: 1
+    memory: 1Gi
+    ephemeral-storage: 8Gi
+  defaultJobTolerations:
+    - key: "armadaproject.io/armada"
+      operator: "Equal"
+      value: "true"
+      effect: "NoSchedule"
+  defaultJobTolerationsByPriorityClass:
+    "":
+      - key: "armadaproject.io/pc-armada-default"
+        operator: "Equal"
+        value: "true"
+        effect: "NoSchedule"
+    armada-default:
+      - key: "armadaproject.io/pc-armada-default"
+        operator: "Equal"
+        value: "true"
+        effect: "NoSchedule"
+    armada-preemptible:
+      - key: "armadaproject.io/pc-armada-preemptible"
+        operator: "Equal"
+        value: "true"
+        effect: "NoSchedule"
+  maxRetries: 5
+  resourceScarcity:
+    cpu: 1.0
+  indexedResources:
+    - cpu
+    - memory
+  gangIdAnnotation: armadaproject.io/gangId
+  gangCardinalityAnnotation: armadaproject.io/gangCardinality
 

e2e/pulsartest_client/app_test.go

@@ -33,26 +33,6 @@ func TestNew(t *testing.T) {
 	assert.Error(t, err)
 	assert.Nil(t, app)
 
-	// Invalid compression type
-	pc = cfg.PulsarConfig{
-		URL:               "pulsar://localhost:6650",
-		CompressionType:   "nocompression",
-		JobsetEventsTopic: "events",
-	}
-	app, err = pt.New(pt.Params{Pulsar: pc}, "submit")
-	assert.Error(t, err)
-	assert.Nil(t, app)
-
-	// Invalid compression level
-	pc = cfg.PulsarConfig{
-		URL:               "pulsar://localhost:6650",
-		CompressionLevel:  "veryCompressed",
-		JobsetEventsTopic: "events",
-	}
-	app, err = pt.New(pt.Params{Pulsar: pc}, "submit")
-	assert.Error(t, err)
-	assert.Nil(t, app)
-
 	// Invalid command type
 	pc = cfg.PulsarConfig{
 		URL:               "pulsar://localhost:6650",

go.mod

@@ -87,6 +87,7 @@ require (
 	github.com/go-openapi/strfmt v0.21.3
 	github.com/go-openapi/swag v0.22.3
 	github.com/go-openapi/validate v0.21.0
+	github.com/go-playground/validator/v10 v10.11.1
 	github.com/golang/mock v1.6.0
 	github.com/goreleaser/goreleaser v1.11.5
 	github.com/jessevdk/go-flags v1.5.0
@@ -199,6 +200,8 @@ require (
 	github.com/go-logr/logr v0.4.0 // indirect
 	github.com/go-openapi/inflect v0.19.0 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
+	github.com/go-playground/locales v0.14.0 // indirect
+	github.com/go-playground/universal-translator v0.18.0 // indirect
 	github.com/go-telegram-bot-api/telegram-bot-api v4.6.4+incompatible // indirect
 	github.com/gobwas/glob v0.2.3 // indirect
 	github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2 // indirect
@@ -256,6 +259,7 @@ require (
 	github.com/klauspost/pgzip v1.2.5 // indirect
 	github.com/kr/pretty v0.3.0 // indirect
 	github.com/kr/text v0.2.0 // indirect
+	github.com/leodido/go-urn v1.2.1 // indirect
 	github.com/linkedin/goavro/v2 v2.9.8 // indirect
 	github.com/lucasb-eyer/go-colorful v1.2.0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect

go.sum

@@ -487,11 +487,18 @@ github.com/go-openapi/swag v0.22.3 h1:yMBqmnQ0gyZvEb/+KzuWZOXgllrXT4SADYbvDaXHv/
 github.com/go-openapi/swag v0.22.3/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14=
 github.com/go-openapi/validate v0.21.0 h1:+Wqk39yKOhfpLqNLEC0/eViCkzM5FVXVqrvt526+wcI=
 github.com/go-openapi/validate v0.21.0/go.mod h1:rjnrwK57VJ7A8xqfpAOEKRH8yQSGUriMu5/zuPSQ1hg=
+github.com/go-playground/assert/v2 v2.0.1 h1:MsBgLAaY856+nPRTKrp3/OZK38U/wa0CcBYNjji3q3A=
 github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
 github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
+github.com/go-playground/locales v0.14.0 h1:u50s323jtVGugKlcYeyzC0etD1HifMjqmJqb8WugfUU=
+github.com/go-playground/locales v0.14.0/go.mod h1:sawfccIbzZTqEDETgFXqTho0QybSa7l++s0DH+LDiLs=
 github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
+github.com/go-playground/universal-translator v0.18.0 h1:82dyy6p4OuJq4/CByFNOn/jYrnRPArHwAcmLoJZxyho=
+github.com/go-playground/universal-translator v0.18.0/go.mod h1:UvRDBj+xPUEGrFYl+lu/H90nyDXpg0fqeB/AQUGNTVA=
 github.com/go-playground/validator/v10 v10.2.0/go.mod h1:uOYAAleCW8F/7oMFd6aG0GOhaH6EGOAJShg8Id5JGkI=
 github.com/go-playground/validator/v10 v10.4.1/go.mod h1:nlOn6nFhuKACm19sB/8EGNn9GlaMV7XkbRSipzJ0Ii4=
+github.com/go-playground/validator/v10 v10.11.1 h1:prmOlTVv+YjZjmRmNSF3VmspqJIxJWXmqUsHwfTRRkQ=
+github.com/go-playground/validator/v10 v10.11.1/go.mod h1:i+3WkQ1FvaUjjxh1kSvIA4dMGDBiPU55YFDl0WbKdWU=
 github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg=
 github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
@@ -918,6 +925,8 @@ github.com/kyleconroy/sqlc v1.16.0/go.mod h1:m+cX/UyBRnKP58lFfUsq+0gw87UUw9Amxwq
 github.com/kylelemons/godebug v0.0.0-20170820004349-d65d576e9348/go.mod h1:B69LEHPfb2qLo0BaaOLcbitczOKLWTsrBG9LczfCD4k=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
+github.com/leodido/go-urn v1.2.1 h1:BqpAaACuzVSgi/VLzGZIobT2z4v53pjosyNd9Yv6n/w=
+github.com/leodido/go-urn v1.2.1/go.mod h1:zt4jvISO2HfUBqxjfIshjdMTYS56ZS/qv49ictyFfxY=
 github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/lib/pq v1.1.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
@@ -1141,6 +1150,7 @@ github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFR
 github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
+github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
 github.com/rogpeppe/go-internal v1.8.1 h1:geMPLpDpQOgVyCg5z5GoRwLHepNdb71NXb67XFkP+Eg=
 github.com/rogpeppe/go-internal v1.8.1/go.mod h1:JeRgkft04UBgHMgCIwADu4Pn6Mtm5d4nPKWu0nJ5d+o=
 github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
@@ -1368,6 +1378,7 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20211108221036-ceb1ce70b4fa/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20211115234514-b4de73f9ece8/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220331220935-ae2d96664a29/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa h1:zuSxTR4o9y82ebqCUJYNGJbGPo6sKVl54f/TVDObg1c=

internal/armada/configuration/types.go

@@ -3,6 +3,7 @@ package configuration
 import (
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsar"
 	"github.com/go-redis/redis"
 	v1 "k8s.io/api/core/v1"
 
@@ -38,7 +39,7 @@ type ArmadaConfig struct {
 
 type PulsarConfig struct {
 	// Pulsar URL
-	URL string
+	URL string `validate:"required"`
 	// Path to the trusted TLS certificate file (must exist)
 	TLSTrustCertsFilePath string
 	// Whether Pulsar client accept untrusted TLS certificate from broker
@@ -56,9 +57,9 @@ type PulsarConfig struct {
 	JobsetEventsTopic           string
 	RedisFromPulsarSubscription string
 	// Compression to use.  Valid values are "None", "LZ4", "Zlib", "Zstd".  Default is "None"
-	CompressionType string
+	CompressionType pulsar.CompressionType
 	// Compression Level to use.  Valid values are "Default", "Better", "Faster".  Default is "Default"
-	CompressionLevel string
+	CompressionLevel pulsar.CompressionLevel
 	// Used to construct an executorconfig.IngressConfiguration,
 	// which is used when converting Armada-specific IngressConfig and ServiceConfig objects into k8s objects.
 	HostnameSuffix string