ridv/gorealis-v2
1
0
Fork 0
Code Issues Pull requests Projects Releases 13 Packages Wiki Activity Actions

Adding support for using different update strategies.

Browse source
This commit is contained in:
Renan DelValle 2020-05-05 20:46:11 -07:00
parent 22013b1be3
commit 45eac75e47
No known key found for this signature in database
GPG key ID: C240AD6D6F443EC9
1 changed files with 39 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

55
jobUpdate.go
View file

@ -31,32 +31,35 @@ type JobUpdate struct {
func NewJobUpdate() *JobUpdate {
newTask := NewTask()
req := aurora.JobUpdateRequest{}
req.TaskConfig = newTask.TaskConfig()
req.Settings = newUpdateSettings()
return &JobUpdate{task: newTask, request: &req}
return &JobUpdate{
task: newTask,
request: &aurora.JobUpdateRequest{TaskConfig: newTask.TaskConfig(), Settings: newUpdateSettings()},
}
}
// Creates an update with default values using an AuroraTask as the underlying task configuration.
// This function has a high level understanding of Aurora Tasks and thus will support copying a task that is configured
// to use Thermos.
func JobUpdateFromAuroraTask(task *AuroraTask) *JobUpdate {
newTask := task.Clone()
req := aurora.JobUpdateRequest{}
req.TaskConfig = newTask.TaskConfig()
req.Settings = newUpdateSettings()
return &JobUpdate{task: newTask, request: &req}
return &JobUpdate{
task: newTask,
request: &aurora.JobUpdateRequest{TaskConfig: newTask.TaskConfig(), Settings: newUpdateSettings()},
}
}
// JobUpdateFromConfig creates an update with default values using an aurora.TaskConfig
// primitive as the underlying task configuration.
// This function should not be used unless the implications of using a primitive value are understood.
// For example, the primitive has no concept of Thermos.
func JobUpdateFromConfig(task *aurora.TaskConfig) *JobUpdate {
// Perform a deep copy to avoid unexpected behavior
newTask := TaskFromThrift(task)
req := aurora.JobUpdateRequest{}
req.TaskConfig = newTask.TaskConfig()
req.Settings = newUpdateSettings()
return &JobUpdate{task: newTask, request: &req}
return &JobUpdate{
task: newTask,
request: &aurora.JobUpdateRequest{TaskConfig: newTask.TaskConfig(), Settings: newUpdateSettings()},
}
}
// Set instance count the job will have after the update.
@ -106,6 +109,26 @@ func (j *JobUpdate) PulseIntervalTimeout(timeout time.Duration) *JobUpdate {
j.request.Settings.BlockIfNoPulsesAfterMs = thrift.Int32Ptr(int32(timeout.Seconds() * 1000))
return j
}
func (j *JobUpdate) BatchUpdateStrategy(autoPause bool, batchSize int32) *JobUpdate {
j.request.Settings.UpdateStrategy = &aurora.JobUpdateStrategy{
BatchStrategy: &aurora.BatchJobUpdateStrategy{GroupSize: batchSize, AutopauseAfterBatch: autoPause},
}
return j
}
func (j *JobUpdate) QueueUpdateStrategy(groupSize int32) *JobUpdate {
j.request.Settings.UpdateStrategy = &aurora.JobUpdateStrategy{
QueueStrategy: &aurora.QueueJobUpdateStrategy{GroupSize: groupSize},
}
return j
}
func (j *JobUpdate) VariableBatchStrategy(autoPause bool, batchSizes... int32) *JobUpdate {
j.request.Settings.UpdateStrategy = &aurora.JobUpdateStrategy{
VarBatchStrategy: &aurora.VariableBatchJobUpdateStrategy{GroupSizes: batchSizes, AutopauseAfterBatch: autoPause},
}
return j
}
func newUpdateSettings() *aurora.JobUpdateSettings {