-
Notifications
You must be signed in to change notification settings - Fork 1.9k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Batch Job Garbage Collection #3982
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -615,8 +615,7 @@ func (j *Job) Deregister(args *structs.JobDeregisterRequest, reply *structs.JobD | |
|
||
// Create a new evaluation | ||
// XXX: The job priority / type is strange for this, since it's not a high | ||
// priority even if the job was. The scheduler itself also doesn't matter, | ||
// since all should be able to handle deregistration in the same way. | ||
// priority even if the job was. | ||
eval := &structs.Evaluation{ | ||
ID: uuid.Generate(), | ||
Namespace: args.RequestNamespace(), | ||
|
@@ -646,6 +645,88 @@ func (j *Job) Deregister(args *structs.JobDeregisterRequest, reply *structs.JobD | |
return nil | ||
} | ||
|
||
// BatchDeregister is used to remove a set of jobs from the cluster. | ||
func (j *Job) BatchDeregister(args *structs.JobBatchDeregisterRequest, reply *structs.JobBatchDeregisterResponse) error { | ||
if done, err := j.srv.forward("Job.BatchDeregister", args, args, reply); done { | ||
return err | ||
} | ||
defer metrics.MeasureSince([]string{"nomad", "job", "batch_deregister"}, time.Now()) | ||
|
||
// Resolve the ACL token | ||
aclObj, err := j.srv.ResolveToken(args.AuthToken) | ||
if err != nil { | ||
return err | ||
} | ||
|
||
// Validate the arguments | ||
if len(args.Jobs) == 0 { | ||
return fmt.Errorf("given no jobs to deregister") | ||
} | ||
if len(args.Evals) != 0 { | ||
return fmt.Errorf("evaluations should not be populated") | ||
} | ||
|
||
// Loop through checking for permissions | ||
for jobNS := range args.Jobs { | ||
// Check for submit-job permissions | ||
if aclObj != nil && !aclObj.AllowNsOp(jobNS.Namespace, acl.NamespaceCapabilitySubmitJob) { | ||
return structs.ErrPermissionDenied | ||
} | ||
} | ||
|
||
// Grab a snapshot | ||
snap, err := j.srv.fsm.State().Snapshot() | ||
if err != nil { | ||
return err | ||
} | ||
|
||
// Loop through to create evals | ||
for jobNS, options := range args.Jobs { | ||
if options == nil { | ||
return fmt.Errorf("no deregister options provided for %v", jobNS) | ||
} | ||
|
||
job, err := snap.JobByID(nil, jobNS.Namespace, jobNS.ID) | ||
if err != nil { | ||
return err | ||
} | ||
|
||
// If the job is periodic or parameterized, we don't create an eval. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The scheduler doesn't do anything for those job types. It does work for the derived batch jobs from them There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ah sure, that makes sense. Maybe in future comments add a |
||
if job != nil && (job.IsPeriodic() || job.IsParameterized()) { | ||
continue | ||
} | ||
|
||
priority := structs.JobDefaultPriority | ||
jtype := structs.JobTypeService | ||
if job != nil { | ||
priority = job.Priority | ||
jtype = job.Type | ||
} | ||
|
||
// Create a new evaluation | ||
eval := &structs.Evaluation{ | ||
ID: uuid.Generate(), | ||
Namespace: jobNS.Namespace, | ||
Priority: priority, | ||
Type: jtype, | ||
TriggeredBy: structs.EvalTriggerJobDeregister, | ||
JobID: jobNS.ID, | ||
Status: structs.EvalStatusPending, | ||
} | ||
args.Evals = append(args.Evals, eval) | ||
} | ||
|
||
// Commit this update via Raft | ||
_, index, err := j.srv.raftApply(structs.JobBatchDeregisterRequestType, args) | ||
if err != nil { | ||
j.srv.logger.Printf("[ERR] nomad.job: batch deregister failed: %v", err) | ||
return err | ||
} | ||
|
||
reply.Index = index | ||
return nil | ||
} | ||
|
||
// GetJob is used to request information about a specific job | ||
func (j *Job) GetJob(args *structs.JobSpecificRequest, | ||
reply *structs.SingleJobResponse) error { | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Little nervous about sharing a pointer to the same struct here, but nothing mutates it and it seems unlikely anything would ever go back through and mutate it so I think it's safe to leave.