ClusterCockpit/cc-backend
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-backend synced 2024-11-10 08:57:25 +01:00
Code Issues Packages Releases Activity

Add node filter and concurrent job list query

Browse Source
This commit is contained in:
Jan Eitzinger 2023-06-28 13:35:41 +02:00
parent 3828c138b8
commit 7174f27a89
6 changed files with 96 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
api/schema.graphqls
View File

@ -237,10 +237,7 @@ input JobFilter {
memUsedMax: FloatRange
exclusive: Int
sharedNode: StringInput
selfJobId: StringInput
selfStartTime: Time
selfDuration: Int
node: StringInput
}
input OrderByInput {
@ -274,6 +271,7 @@ type JobResultList {
}
type JobLinkResultList {
listQuery: String
items: [JobLink!]!
count: Int
}

93
internal/graph/generated/generated.go
View File

@ -116,6 +116,7 @@ type ComplexityRoot struct {
JobLinkResultList struct {
Count func(childComplexity int) int
Items func(childComplexity int) int
ListQuery func(childComplexity int) int
}
JobMetric struct {
@ -629,6 +630,13 @@ func (e *executableSchema) Complexity(typeName, field string, childComplexity in
return e.complexity.JobLinkResultList.Items(childComplexity), true
case "JobLinkResultList.listQuery":
if e.complexity.JobLinkResultList.ListQuery == nil {
break
}
return e.complexity.JobLinkResultList.ListQuery(childComplexity), true
case "JobMetric.series":
if e.complexity.JobMetric.Series == nil {
break
@ -1739,10 +1747,7 @@ input JobFilter {
memUsedMax: FloatRange
exclusive: Int
sharedNode: StringInput
selfJobId: StringInput
selfStartTime: Time
selfDuration: Int
node: StringInput
}
input OrderByInput {
@ -1776,6 +1781,7 @@ type JobResultList {
}
type JobLinkResultList {
listQuery: String
items: [JobLink!]!
count: Int
}
@ -3951,6 +3957,8 @@ func (ec *executionContext) fieldContext_Job_concurrentJobs(ctx context.Context,
IsResolver: true,
Child: func(ctx context.Context, field graphql.CollectedField) (*graphql.FieldContext, error) {
switch field.Name {
case "listQuery":
return ec.fieldContext_JobLinkResultList_listQuery(ctx, field)
case "items":
return ec.fieldContext_JobLinkResultList_items(ctx, field)
case "count":
@ -4140,6 +4148,47 @@ func (ec *executionContext) fieldContext_JobLink_jobId(ctx context.Context, fiel
return fc, nil
}
func (ec *executionContext) _JobLinkResultList_listQuery(ctx context.Context, field graphql.CollectedField, obj *model.JobLinkResultList) (ret graphql.Marshaler) {
fc, err := ec.fieldContext_JobLinkResultList_listQuery(ctx, field)
if err != nil {
return graphql.Null
}
ctx = graphql.WithFieldContext(ctx, fc)
defer func() {
if r := recover(); r != nil {
ec.Error(ctx, ec.Recover(ctx, r))
ret = graphql.Null
}
}()
resTmp, err := ec.ResolverMiddleware(ctx, func(rctx context.Context) (interface{}, error) {
ctx = rctx // use context from middleware stack in children
return obj.ListQuery, nil
})
if err != nil {
ec.Error(ctx, err)
return graphql.Null
}
if resTmp == nil {
return graphql.Null
}
res := resTmp.(*string)
fc.Result = res
return ec.marshalOString2ᚖstring(ctx, field.Selections, res)
}
func (ec *executionContext) fieldContext_JobLinkResultList_listQuery(ctx context.Context, field graphql.CollectedField) (fc *graphql.FieldContext, err error) {
fc = &graphql.FieldContext{
Object: "JobLinkResultList",
Field: field,
IsMethod: false,
IsResolver: false,
Child: func(ctx context.Context, field graphql.CollectedField) (*graphql.FieldContext, error) {
return nil, errors.New("field of type String does not have child fields")
},
}
return fc, nil
}
func (ec *executionContext) _JobLinkResultList_items(ctx context.Context, field graphql.CollectedField, obj *model.JobLinkResultList) (ret graphql.Marshaler) {
fc, err := ec.fieldContext_JobLinkResultList_items(ctx, field)
if err != nil {
@ -11148,7 +11197,7 @@ func (ec *executionContext) unmarshalInputJobFilter(ctx context.Context, obj int
asMap[k] = v
}
fieldsInOrder := [...]string{"tags", "jobId", "arrayJobId", "user", "project", "jobName", "cluster", "partition", "duration", "minRunningFor", "numNodes", "numAccelerators", "numHWThreads", "startTime", "state", "flopsAnyAvg", "memBwAvg", "loadAvg", "memUsedMax", "exclusive", "sharedNode", "selfJobId", "selfStartTime", "selfDuration"}
fieldsInOrder := [...]string{"tags", "jobId", "arrayJobId", "user", "project", "jobName", "cluster", "partition", "duration", "minRunningFor", "numNodes", "numAccelerators", "numHWThreads", "startTime", "state", "flopsAnyAvg", "memBwAvg", "loadAvg", "memUsedMax", "exclusive", "node"}
for _, k := range fieldsInOrder {
v, ok := asMap[k]
if !ok {
@ -11315,35 +11364,11 @@ func (ec *executionContext) unmarshalInputJobFilter(ctx context.Context, obj int
if err != nil {
return it, err
}
case "sharedNode":
case "node":
var err error
ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("sharedNode"))
it.SharedNode, err = ec.unmarshalOStringInput2ᚖgithubᚗcomᚋClusterCockpitᚋccᚑbackendᚋinternalᚋgraphᚋmodelᚐStringInput(ctx, v)
if err != nil {
return it, err
}
case "selfJobId":
var err error
ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("selfJobId"))
it.SelfJobID, err = ec.unmarshalOStringInput2ᚖgithubᚗcomᚋClusterCockpitᚋccᚑbackendᚋinternalᚋgraphᚋmodelᚐStringInput(ctx, v)
if err != nil {
return it, err
}
case "selfStartTime":
var err error
ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("selfStartTime"))
it.SelfStartTime, err = ec.unmarshalOTime2ᚖtimeᚐTime(ctx, v)
if err != nil {
return it, err
}
case "selfDuration":
var err error
ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("selfDuration"))
it.SelfDuration, err = ec.unmarshalOInt2ᚖint(ctx, v)
ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("node"))
it.Node, err = ec.unmarshalOStringInput2ᚖgithubᚗcomᚋClusterCockpitᚋccᚑbackendᚋinternalᚋgraphᚋmodelᚐStringInput(ctx, v)
if err != nil {
return it, err
}
@ -12055,6 +12080,10 @@ func (ec *executionContext) _JobLinkResultList(ctx context.Context, sel ast.Sele
switch field.Name {
case "__typename":
out.Values[i] = graphql.MarshalString("JobLinkResultList")
case "listQuery":
out.Values[i] = ec._JobLinkResultList_listQuery(ctx, field, obj)
case "items":
out.Values[i] = ec._JobLinkResultList_items(ctx, field, obj)

6
internal/graph/model/models_gen.go
View File

@ -57,10 +57,7 @@ type JobFilter struct {
LoadAvg *FloatRange `json:"loadAvg"`
MemUsedMax *FloatRange `json:"memUsedMax"`
Exclusive *int `json:"exclusive"`
SharedNode *StringInput `json:"sharedNode"`
SelfJobID *StringInput `json:"selfJobId"`
SelfStartTime *time.Time `json:"selfStartTime"`
SelfDuration *int `json:"selfDuration"`
Node *StringInput `json:"node"`
}
type JobLink struct {
@ -69,6 +66,7 @@ type JobLink struct {
}
type JobLinkResultList struct {
ListQuery *string `json:"listQuery"`
Items []*JobLink `json:"items"`
Count *int `json:"count"`
}

4
internal/graph/schema.resolvers.go
View File

@ -32,9 +32,7 @@ func (r *jobResolver) Tags(ctx context.Context, obj *schema.Job) ([]*schema.Tag,
}
// ConcurrentJobs is the resolver for the concurrentJobs field.
func (r *jobResolver) ConcurrentJobs(
ctx context.Context, obj *schema.Job) (*model.JobLinkResultList, error) {
func (r *jobResolver) ConcurrentJobs(ctx context.Context, obj *schema.Job) (*model.JobLinkResultList, error) {
if obj.State == schema.JobStateRunning {
obj.Duration = int32(time.Now().Unix() - obj.StartTimeUnix)
}

28
internal/repository/job.go
View File

@ -10,6 +10,7 @@ import (
"encoding/json"
"errors"
"fmt"
"math"
"strconv"
"sync"
"time"
@ -17,6 +18,7 @@ import (
"github.com/ClusterCockpit/cc-backend/internal/auth"
"github.com/ClusterCockpit/cc-backend/internal/graph/model"
"github.com/ClusterCockpit/cc-backend/internal/metricdata"
"github.com/ClusterCockpit/cc-backend/internal/util"
"github.com/ClusterCockpit/cc-backend/pkg/log"
"github.com/ClusterCockpit/cc-backend/pkg/lrucache"
"github.com/ClusterCockpit/cc-backend/pkg/schema"
@ -298,7 +300,7 @@ func (r *JobRepository) FindConcurrentJobs(
return nil, nil
}
query, qerr := SecurityCheck(ctx, sq.Select("job.id", "job.job_id").From("job"))
query, qerr := SecurityCheck(ctx, sq.Select("job.id", "job.job_id", "job.start_time").From("job"))
if qerr != nil {
return nil, qerr
}
@ -308,6 +310,7 @@ func (r *JobRepository) FindConcurrentJobs(
var stopTime int64
startTime = job.StartTimeUnix
hostname := job.Resources[0].Hostname
if job.State == schema.JobStateRunning {
stopTime = time.Now().Unix()
@ -322,11 +325,11 @@ func (r *JobRepository) FindConcurrentJobs(
queryRunning := query.Where("job.job_state = ?").Where("(job.start_time BETWEEN ? AND ? OR job.start_time < ?)",
"running", startTimeTail, stopTimeTail, startTime)
queryRunning = queryRunning.Where("job.resources LIKE ?", fmt.Sprint("%", job.Resources[0].Hostname, "%"))
queryRunning = queryRunning.Where("job.resources LIKE ?", fmt.Sprint("%", hostname, "%"))
query = query.Where("job.job_state != ?").Where("((job.start_time BETWEEN ? AND ?) OR (job.start_time + job.duration) BETWEEN ? AND ? OR (job.start_time < ?) AND (job.start_time + job.duration) > ?)",
"running", startTimeTail, stopTimeTail, startTimeFront, stopTimeTail, startTime, stopTime)
query = query.Where("job.resources LIKE ?", fmt.Sprint("%", job.Resources[0].Hostname, "%"))
query = query.Where("job.resources LIKE ?", fmt.Sprint("%", hostname, "%"))
rows, err := query.RunWith(r.stmtCache).Query()
if err != nil {
@ -335,16 +338,21 @@ func (r *JobRepository) FindConcurrentJobs(
}
items := make([]*model.JobLink, 0, 10)
minStart := int64(math.MaxInt64)
maxStart := int64(0)
for rows.Next() {
var id, jobId sql.NullInt64
var id, jobId, startTime sql.NullInt64
if err = rows.Scan(&id, &jobId); err != nil {
if err = rows.Scan(&id, &jobId, &startTime); err != nil {
log.Warn("Error while scanning rows")
return nil, err
}
if id.Valid {
minStart = util.Min(minStart, startTime.Int64)
maxStart = util.Max(maxStart, startTime.Int64)
items = append(items,
&model.JobLink{
ID: fmt.Sprint(id.Int64),
@ -360,14 +368,17 @@ func (r *JobRepository) FindConcurrentJobs(
}
for rows.Next() {
var id, jobId sql.NullInt64
var id, jobId, startTime sql.NullInt64
if err := rows.Scan(&id, &jobId); err != nil {
if err := rows.Scan(&id, &jobId, &startTime); err != nil {
log.Warn("Error while scanning rows")
return nil, err
}
if id.Valid {
minStart = util.Min(minStart, startTime.Int64)
maxStart = util.Max(maxStart, startTime.Int64)
items = append(items,
&model.JobLink{
ID: fmt.Sprint(id.Int64),
@ -377,8 +388,11 @@ func (r *JobRepository) FindConcurrentJobs(
}
cnt := len(items)
queryString := fmt.Sprintf("cluster=%s&startTime=%d-%d&node=%s",
job.Cluster, minStart, maxStart, hostname)
return &model.JobLinkResultList{
ListQuery: &queryString,
Items: items,
Count: &cnt,
}, nil

3
internal/repository/query.go
View File

@ -208,6 +208,9 @@ func BuildWhereClause(filter *model.JobFilter, query sq.SelectBuilder) sq.Select
if filter.NumHWThreads != nil {
query = buildIntCondition("job.num_hwthreads", filter.NumHWThreads, query)
}
if filter.Node != nil {
query = buildStringCondition("job.resources", filter.Node, query)
}
if filter.FlopsAnyAvg != nil {
query = buildFloatCondition("job.flops_any_avg", filter.FlopsAnyAvg, query)
}