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

Sync commit

Browse Source
This commit is contained in:
Jan Eitzinger 2023-06-20 10:38:53 +02:00
parent 1f60963cbb
commit 33613cdda0
2 changed files with 27 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -32,32 +32,15 @@ func (r *jobResolver) Tags(ctx context.Context, obj *schema.Job) ([]*schema.Tag,
} }
// ConcurrentJobs is the resolver for the concurrentJobs field. // 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(
exc := int(obj.Exclusive) ctx context.Context, obj *schema.Job) (*model.JobLinkResultList, error) {
if exc != 1 {
filter := []*model.JobFilter{}
jid := fmt.Sprint(obj.JobID)
jdu := int(obj.Duration)
filter = append(filter, &model.JobFilter{Exclusive: &exc})
filter = append(filter, &model.JobFilter{SharedNode: &model.StringInput{Contains: &obj.Resources[0].Hostname}})
filter = append(filter, &model.JobFilter{SelfJobID: &model.StringInput{Neq: &jid}})
filter = append(filter, &model.JobFilter{SelfStartTime: &obj.StartTime, SelfDuration: &jdu})
jobLinks, err := r.Repo.QueryJobLinks(ctx, filter) if obj.State == schema.JobStateRunning {
if err != nil { obj.Duration = int32(time.Now().Unix() - obj.StartTimeUnix)
log.Warn("Error while querying jobLinks") }
return nil, err
}
count, err := r.Repo.CountJobs(ctx, filter) if obj.Exclusive != 1 && obj.Duration > 600 {
if err != nil { return r.Repo.FindConcurrentJobs(obj)
log.Warn("Error while counting jobLinks")
return nil, err
}
result := &model.JobLinkResultList{Items: jobLinks, Count: &count}
return result, nil
} }
return nil, nil return nil, nil

20
internal/repository/job.go
View File

@ -292,6 +292,26 @@ func (r *JobRepository) FindById(jobId int64) (*schema.Job, error) {
return scanJob(q.RunWith(r.stmtCache).QueryRow()) return scanJob(q.RunWith(r.stmtCache).QueryRow())
} }
func (r *JobRepository) FindConcurrentJobs(
job *schema.Job) (*model.JobLinkResultList, error) {
query := sq.Select("job.id","job.job_id").From("job").Where("cluster = ?",job.Cluster)
var startTime := job.StartTimeUnix
var stopTime int64
if job.State == schema.JobStateRunning {
stopTime = time.Now().Unix()
} else {
stopTime = startTime + int64(job.Duration)
}
// Add 5m overlap for jobs start time at the end
stopTime -= 300
query = query.Where("start_time BETWEEN ? AND ?", startTime, stopTime)
}
// Start inserts a new job in the table, returning the unique job ID. // Start inserts a new job in the table, returning the unique job ID.
// Statistics are not transfered! // Statistics are not transfered!
func (r *JobRepository) Start(job *schema.JobMeta) (id int64, err error) { func (r *JobRepository) Start(job *schema.JobMeta) (id int64, err error) {