-
Notifications
You must be signed in to change notification settings - Fork 126
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Make legacyScheduler generic * linting * linting * wip * wip * wip * wip * wip * remove nats from e2e * remove more nats config * fixed submit test * linting * go mod tidy * increase sleep to 60 * wip * wip * linting * unused context * remove extra queued-jobs-iterator * wip * wip * linting passes * documentation * unit tests * doc * more tests * linting * more doc, revert files modified in error * more linting * more linting * fix tests * merge master * doc improvements * wip * Update internal/scheduler/scheduler.go Co-authored-by: Albin Severinson <albin@severinson.org> * fixes following review * wip * renamed publisher_test * wip * linting * fixes after merge * Update internal/scheduler/scheduling_algo.go Co-authored-by: Clif Houck <me@clifhouck.com> * code review comments * mocks * wip * wip * wip * doc for job repo * doc for job repo * addressed some warnings * restore nodes table * filter out queues with no jobs to schedule * wip * wip * remived nats from main makefiles * removed extra config * renamed file * code review comments * linting * linting * more linting * more linting * another nats reference * another nats reference * linting * fix tests * fix tests * fix tests * job repo test done * wip * merged master * merged master * fix changes following merge * restore package * fixed proto package names * linting * linting * fix null pointer in test * linting * linting * doc * unit tests * moved mock generation into its owen package * moved mock generation into its owen package * wip * go lint * linting * fixed package names * wip * linting * linting * linting * move master check into a separate function * don't modify restapi * doc * Update internal/armada/configuration/types.go Co-authored-by: Albin Severinson <albin@severinson.org> * Update internal/scheduler/database/job_repository.go Co-authored-by: Albin Severinson <albin@severinson.org> * Update internal/scheduler/reports.go Co-authored-by: Albin Severinson <albin@severinson.org> * wip * compilation fixes * updated proto * add job ids * wip * tests for api * linting * add grpc mock * fix tests * fix tests * fix tests * formatting * formatting * doc * more doc * wip * remove active_job_ids from streaming lease call * wip * nodes should be pointer not struct * wip * added job run lookup * wip * linting * linting * linting * fix tests * fix tests * executor tests * linting * linting * doc * fix tests * wip * wip * wip * wip * linting * import order * custom pulsar marshallers * config * more changes * commands * linting * added tests for hooks * lots of compilation fixes * go mod tidy * fix failing test * test for queue_repository * remove uneeded file * linting * flush db always * flush db always * review comments * linting Co-authored-by: Chris Martin <chris@cmartinit.co.uk> Co-authored-by: Albin Severinson <albin@severinson.org> Co-authored-by: Clif Houck <me@clifhouck.com>
- Loading branch information
4 people
committed
Jan 25, 2023
1 parent
db473c0
commit 6b27ffe
Showing
45 changed files
with
884 additions
and
313 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| package cmd | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/pkg/errors" | ||
| log "github.com/sirupsen/logrus" | ||
| "github.com/spf13/cobra" | ||
| "github.com/spf13/viper" | ||
|
|
||
| "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, | ||
| } | ||
| cmd.PersistentFlags().Duration( | ||
| "timeout", | ||
| 5*time.Minute, | ||
| "Duration after which the migration will fail if it has not been created") | ||
|
|
||
| return cmd | ||
| } | ||
|
|
||
| func migrateDatabase(_ *cobra.Command, _ []string) error { | ||
| timeout := viper.GetDuration("timeout") | ||
| ctx, cancel := context.WithTimeout(context.Background(), timeout) | ||
| defer cancel() | ||
| 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.WithMessagef(err, "Failed to connect to database") | ||
| } | ||
| err = schedulerdb.Migrate(ctx, db) | ||
| if err != nil { | ||
| return errors.WithMessagef(err, "Failed to migrate scheduler database") | ||
| } | ||
| taken := time.Since(start) | ||
| log.Infof("Scheduler database migrated in %s", taken) | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,67 +1,12 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "os" | ||
| "time" | ||
|
|
||
| "github.com/pkg/errors" | ||
| log "github.com/sirupsen/logrus" | ||
| "github.com/spf13/pflag" | ||
| "github.com/spf13/viper" | ||
|
|
||
| "github.com/armadaproject/armada/cmd/scheduler/cmd" | ||
| "github.com/armadaproject/armada/internal/common" | ||
| "github.com/armadaproject/armada/internal/common/database" | ||
| "github.com/armadaproject/armada/internal/scheduler" | ||
| schedulerdb "github.com/armadaproject/armada/internal/scheduler/database" | ||
| ) | ||
|
|
||
| const ( | ||
| CustomConfigLocation string = "config" | ||
| MigrateDatabase string = "migrateDatabase" | ||
| ) | ||
|
|
||
| func init() { | ||
| pflag.StringSlice( | ||
| CustomConfigLocation, | ||
| []string{}, | ||
| "Fully qualified path to application configuration file (for multiple config files repeat this arg or separate paths with commas)", | ||
| ) | ||
| pflag.Bool(MigrateDatabase, false, "Migrate database instead of running scheduler") | ||
| pflag.Parse() | ||
| } | ||
|
|
||
| func main() { | ||
| common.ConfigureLogging() | ||
| common.BindCommandlineArguments() | ||
|
|
||
| var config scheduler.Configuration | ||
| userSpecifiedConfigs := viper.GetStringSlice(CustomConfigLocation) | ||
|
|
||
| common.LoadConfig(&config, "./config/scheduler", userSpecifiedConfigs) | ||
|
|
||
| if viper.GetBool(MigrateDatabase) { | ||
| migrateDatabase(&config) | ||
| } else { | ||
| if err := scheduler.Run(&config); err != nil { | ||
| log.Errorf("failed to run scheduler: %s", err) | ||
| os.Exit(1) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func migrateDatabase(config *scheduler.Configuration) { | ||
| start := time.Now() | ||
| log.Info("Beginning scheduler database migration") | ||
| db, err := database.OpenPgxPool(config.Postgres) | ||
| if err != nil { | ||
| panic(errors.WithMessage(err, "Failed to connect to database")) | ||
| } | ||
| err = schedulerdb.Migrate(context.Background(), db) | ||
| if err != nil { | ||
| panic(errors.WithMessage(err, "Failed to migrate scheduler database")) | ||
| } | ||
| taken := time.Now().Sub(start) | ||
| log.Infof("Scheduler database migrated in %dms", taken.Milliseconds()) | ||
| os.Exit(0) | ||
| _ = cmd.RootCmd().Execute() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.