diff --git a/graph/generated/generated.go b/graph/generated/generated.go index 9b76179..91db1f1 100644 --- a/graph/generated/generated.go +++ b/graph/generated/generated.go @@ -185,7 +185,7 @@ type ComplexityRoot struct { Job func(childComplexity int, id string) int JobMetrics func(childComplexity int, id string, metrics []string, scopes []schema.MetricScope) int Jobs func(childComplexity int, filter []*model.JobFilter, page *model.PageRequest, order *model.OrderByInput) int - JobsCount func(childComplexity int, filter []*model.JobFilter, groupBy model.Aggregate, limit *int) int + JobsCount func(childComplexity int, filter []*model.JobFilter, groupBy model.Aggregate, weight *model.Weights, limit *int) int JobsFootprints func(childComplexity int, filter []*model.JobFilter, metrics []string) int JobsStatistics func(childComplexity int, filter []*model.JobFilter, groupBy *model.Aggregate) int NodeMetrics func(childComplexity int, cluster string, nodes []string, scopes []schema.MetricScope, metrics []string, from time.Time, to time.Time) int @@ -281,7 +281,7 @@ type QueryResolver interface { JobsFootprints(ctx context.Context, filter []*model.JobFilter, metrics []string) (*model.Footprints, error) Jobs(ctx context.Context, filter []*model.JobFilter, page *model.PageRequest, order *model.OrderByInput) (*model.JobResultList, error) JobsStatistics(ctx context.Context, filter []*model.JobFilter, groupBy *model.Aggregate) ([]*model.JobsStatistics, error) - JobsCount(ctx context.Context, filter []*model.JobFilter, groupBy model.Aggregate, limit *int) ([]*model.Count, error) + JobsCount(ctx context.Context, filter []*model.JobFilter, groupBy model.Aggregate, weight *model.Weights, limit *int) ([]*model.Count, error) RooflineHeatmap(ctx context.Context, filter []*model.JobFilter, rows int, cols int, minX float64, minY float64, maxX float64, maxY float64) ([][]float64, error) NodeMetrics(ctx context.Context, cluster string, nodes []string, scopes []schema.MetricScope, metrics []string, from time.Time, to time.Time) ([]*model.NodeMetrics, error) } @@ -958,7 +958,7 @@ func (e *executableSchema) Complexity(typeName, field string, childComplexity in return 0, false } - return e.complexity.Query.JobsCount(childComplexity, args["filter"].([]*model.JobFilter), args["groupBy"].(model.Aggregate), args["limit"].(*int)), true + return e.complexity.Query.JobsCount(childComplexity, args["filter"].([]*model.JobFilter), args["groupBy"].(model.Aggregate), args["weight"].(*model.Weights), args["limit"].(*int)), true case "Query.jobsFootprints": if e.complexity.Query.JobsFootprints == nil { @@ -1481,6 +1481,7 @@ type Footprints { } enum Aggregate { USER, PROJECT, CLUSTER } +enum Weights { NODE_COUNT, NODE_HOURS } type NodeMetrics { host: String! @@ -1512,7 +1513,7 @@ type Query { jobs(filter: [JobFilter!], page: PageRequest, order: OrderByInput): JobResultList! jobsStatistics(filter: [JobFilter!], groupBy: Aggregate): [JobsStatistics!]! - jobsCount(filter: [JobFilter]!, groupBy: Aggregate!, limit: Int): [Count!]! + jobsCount(filter: [JobFilter]!, groupBy: Aggregate!, weight: Weights, limit: Int): [Count!]! rooflineHeatmap(filter: [JobFilter!]!, rows: Int!, cols: Int!, minX: Float!, minY: Float!, maxX: Float!, maxY: Float!): [[Float!]!]! @@ -1826,15 +1827,24 @@ func (ec *executionContext) field_Query_jobsCount_args(ctx context.Context, rawA } } args["groupBy"] = arg1 - var arg2 *int - if tmp, ok := rawArgs["limit"]; ok { - ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("limit")) - arg2, err = ec.unmarshalOInt2ᚖint(ctx, tmp) + var arg2 *model.Weights + if tmp, ok := rawArgs["weight"]; ok { + ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("weight")) + arg2, err = ec.unmarshalOWeights2ᚖgithubᚗcomᚋClusterCockpitᚋccᚑbackendᚋgraphᚋmodelᚐWeights(ctx, tmp) if err != nil { return nil, err } } - args["limit"] = arg2 + args["weight"] = arg2 + var arg3 *int + if tmp, ok := rawArgs["limit"]; ok { + ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("limit")) + arg3, err = ec.unmarshalOInt2ᚖint(ctx, tmp) + if err != nil { + return nil, err + } + } + args["limit"] = arg3 return args, nil } @@ -5324,7 +5334,7 @@ func (ec *executionContext) _Query_jobsCount(ctx context.Context, field graphql. fc.Args = args resTmp, err := ec.ResolverMiddleware(ctx, func(rctx context.Context) (interface{}, error) { ctx = rctx // use context from middleware stack in children - return ec.resolvers.Query().JobsCount(rctx, args["filter"].([]*model.JobFilter), args["groupBy"].(model.Aggregate), args["limit"].(*int)) + return ec.resolvers.Query().JobsCount(rctx, args["filter"].([]*model.JobFilter), args["groupBy"].(model.Aggregate), args["weight"].(*model.Weights), args["limit"].(*int)) }) if err != nil { ec.Error(ctx, err) @@ -11518,6 +11528,22 @@ func (ec *executionContext) marshalOUser2ᚖgithubᚗcomᚋClusterCockpitᚋcc return ec._User(ctx, sel, v) } +func (ec *executionContext) unmarshalOWeights2ᚖgithubᚗcomᚋClusterCockpitᚋccᚑbackendᚋgraphᚋmodelᚐWeights(ctx context.Context, v interface{}) (*model.Weights, error) { + if v == nil { + return nil, nil + } + var res = new(model.Weights) + err := res.UnmarshalGQL(v) + return res, graphql.ErrorOnPath(ctx, err) +} + +func (ec *executionContext) marshalOWeights2ᚖgithubᚗcomᚋClusterCockpitᚋccᚑbackendᚋgraphᚋmodelᚐWeights(ctx context.Context, sel ast.SelectionSet, v *model.Weights) graphql.Marshaler { + if v == nil { + return graphql.Null + } + return v +} + func (ec *executionContext) marshalO__EnumValue2ᚕgithubᚗcomᚋ99designsᚋgqlgenᚋgraphqlᚋintrospectionᚐEnumValueᚄ(ctx context.Context, sel ast.SelectionSet, v []introspection.EnumValue) graphql.Marshaler { if v == nil { return graphql.Null diff --git a/graph/model/models_gen.go b/graph/model/models_gen.go index 16e6590..46baa30 100644 --- a/graph/model/models_gen.go +++ b/graph/model/models_gen.go @@ -258,3 +258,44 @@ func (e *SortDirectionEnum) UnmarshalGQL(v interface{}) error { func (e SortDirectionEnum) MarshalGQL(w io.Writer) { fmt.Fprint(w, strconv.Quote(e.String())) } + +type Weights string + +const ( + WeightsNodeCount Weights = "NODE_COUNT" + WeightsNodeHours Weights = "NODE_HOURS" +) + +var AllWeights = []Weights{ + WeightsNodeCount, + WeightsNodeHours, +} + +func (e Weights) IsValid() bool { + switch e { + case WeightsNodeCount, WeightsNodeHours: + return true + } + return false +} + +func (e Weights) String() string { + return string(e) +} + +func (e *Weights) UnmarshalGQL(v interface{}) error { + str, ok := v.(string) + if !ok { + return fmt.Errorf("enums must be strings") + } + + *e = Weights(str) + if !e.IsValid() { + return fmt.Errorf("%s is not a valid Weights", str) + } + return nil +} + +func (e Weights) MarshalGQL(w io.Writer) { + fmt.Fprint(w, strconv.Quote(e.String())) +} diff --git a/graph/schema.graphqls b/graph/schema.graphqls index 8bab641..793fe69 100644 --- a/graph/schema.graphqls +++ b/graph/schema.graphqls @@ -136,6 +136,7 @@ type Footprints { } enum Aggregate { USER, PROJECT, CLUSTER } +enum Weights { NODE_COUNT, NODE_HOURS } type NodeMetrics { host: String! @@ -167,7 +168,7 @@ type Query { jobs(filter: [JobFilter!], page: PageRequest, order: OrderByInput): JobResultList! jobsStatistics(filter: [JobFilter!], groupBy: Aggregate): [JobsStatistics!]! - jobsCount(filter: [JobFilter]!, groupBy: Aggregate!, limit: Int): [Count!]! + jobsCount(filter: [JobFilter]!, groupBy: Aggregate!, weight: Weights, limit: Int): [Count!]! rooflineHeatmap(filter: [JobFilter!]!, rows: Int!, cols: Int!, minX: Float!, minY: Float!, maxX: Float!, maxY: Float!): [[Float!]!]! diff --git a/graph/schema.resolvers.go b/graph/schema.resolvers.go index ec22f45..46b4d7f 100644 --- a/graph/schema.resolvers.go +++ b/graph/schema.resolvers.go @@ -202,8 +202,8 @@ func (r *queryResolver) JobsStatistics(ctx context.Context, filter []*model.JobF return r.jobsStatistics(ctx, filter, groupBy) } -func (r *queryResolver) JobsCount(ctx context.Context, filter []*model.JobFilter, groupBy model.Aggregate, limit *int) ([]*model.Count, error) { - counts, err := r.Repo.CountGroupedJobs(ctx, groupBy, filter, limit) +func (r *queryResolver) JobsCount(ctx context.Context, filter []*model.JobFilter, groupBy model.Aggregate, weight *model.Weights, limit *int) ([]*model.Count, error) { + counts, err := r.Repo.CountGroupedJobs(ctx, groupBy, filter, weight, limit) if err != nil { return nil, err } diff --git a/repository/job.go b/repository/job.go index 4118153..d6866ca 100644 --- a/repository/job.go +++ b/repository/job.go @@ -193,12 +193,25 @@ func (r *JobRepository) Stop( } // TODO: Use node hours instead: SELECT job.user, sum(job.num_nodes * (CASE WHEN job.job_state = "running" THEN CAST(strftime('%s', 'now') AS INTEGER) - job.start_time ELSE job.duration END)) as x FROM job GROUP BY user ORDER BY x DESC; -func (r *JobRepository) CountGroupedJobs(ctx context.Context, aggreg model.Aggregate, filters []*model.JobFilter, limit *int) (map[string]int, error) { +func (r *JobRepository) CountGroupedJobs(ctx context.Context, aggreg model.Aggregate, filters []*model.JobFilter, weight *model.Weights, limit *int) (map[string]int, error) { if !aggreg.IsValid() { return nil, errors.New("invalid aggregate") } - q := sq.Select("job."+string(aggreg), "count(*) as count").From("job").GroupBy("job." + string(aggreg)).OrderBy("count DESC") + runner := (sq.BaseRunner)(r.stmtCache) + count := "count(*) as count" + if weight != nil { + switch *weight { + case model.WeightsNodeCount: + count = "sum(job.num_nodes) as count" + case model.WeightsNodeHours: + now := time.Now().Unix() + count = fmt.Sprintf(`sum(job.num_nodes * (CASE WHEN job.job_state = "running" THEN %d - job.start_time ELSE job.duration END)) as count`, now) + runner = r.DB + } + } + + q := sq.Select("job."+string(aggreg), count).From("job").GroupBy("job." + string(aggreg)).OrderBy("count DESC") q = SecurityCheck(ctx, q) for _, f := range filters { q = BuildWhereClause(f, q) @@ -208,7 +221,7 @@ func (r *JobRepository) CountGroupedJobs(ctx context.Context, aggreg model.Aggre } counts := map[string]int{} - rows, err := q.RunWith(r.DB).Query() + rows, err := q.RunWith(runner).Query() if err != nil { return nil, err } diff --git a/routes.go b/routes.go index 63350e5..9885b94 100644 --- a/routes.go +++ b/routes.go @@ -53,22 +53,21 @@ func setupHomeRoute(i InfoType, r *http.Request) InfoType { runningJobs, err := jobRepo.CountGroupedJobs(r.Context(), model.AggregateCluster, []*model.JobFilter{{ State: []schema.JobState{schema.JobStateRunning}, - }}, nil) + }}, nil, nil) if err != nil { log.Errorf("failed to count jobs: %s", err.Error()) runningJobs = map[string]int{} } - totalJobs, err := jobRepo.CountGroupedJobs(r.Context(), model.AggregateCluster, nil, nil) + totalJobs, err := jobRepo.CountGroupedJobs(r.Context(), model.AggregateCluster, nil, nil, nil) if err != nil { log.Errorf("failed to count jobs: %s", err.Error()) totalJobs = map[string]int{} } - from := time.Now().Add(-24 * time.Hour) recentShortJobs, err := jobRepo.CountGroupedJobs(r.Context(), model.AggregateCluster, []*model.JobFilter{{ StartTime: &model.TimeRange{From: &from, To: nil}, Duration: &model.IntRange{From: 0, To: graph.ShortJobDuration}, - }}, nil) + }}, nil, nil) if err != nil { log.Errorf("failed to count jobs: %s", err.Error()) recentShortJobs = map[string]int{}