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

Documentation fix (#130)

Browse source 
Fixes documentation so that it is more compliant with godoc format.
This commit is contained in:
Renán I. Del Valle 2021-04-29 10:48:43 -07:00 • committed by GitHub
parent e7f9c0cba9
commit a9d99067ee
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 37 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

4
monitors.go
View file

@ -117,7 +117,7 @@ func (m *Monitor) JobUpdateQuery(
} }
} }
// AutoPaused monitor is a special monitor for auto pause enabled batch updates. This monitor ensures that the update // AutoPausedUpdateMonitor is a special monitor for auto pause enabled batch updates. This monitor ensures that the update
// being monitored is capable of auto pausing and has auto pausing enabled. After verifying this information, // being monitored is capable of auto pausing and has auto pausing enabled. After verifying this information,
// the monitor watches for the job to enter the ROLL_FORWARD_PAUSED state and calculates the current batch // the monitor watches for the job to enter the ROLL_FORWARD_PAUSED state and calculates the current batch
// the update is in using information from the update configuration. // the update is in using information from the update configuration.
@ -183,7 +183,7 @@ func (m *Monitor) AutoPausedUpdateMonitor(key aurora.JobUpdateKey, interval, tim
return calculateCurrentBatch(int32(len(updatingInstances)), batchSizes), nil return calculateCurrentBatch(int32(len(updatingInstances)), batchSizes), nil
} }
// Monitor a Job until all instances enter one of the LIVE_STATES // Instances will monitor a Job until all instances enter one of the LIVE_STATES
func (m *Monitor) Instances(key *aurora.JobKey, instances int32, interval, timeout int) (bool, error) { func (m *Monitor) Instances(key *aurora.JobKey, instances int32, interval, timeout int) (bool, error) {
return m.ScheduleStatus(key, instances, LiveStates, interval, timeout) return m.ScheduleStatus(key, instances, LiveStates, interval, timeout)
} }

64
realis.go
View file

@ -386,7 +386,7 @@ func NewRealisClient(options ...ClientOption) (Realis, error) {
return nil, errors.New("incomplete Options -- url, cluster.json, or Zookeeper address required") return nil, errors.New("incomplete Options -- url, cluster.json, or Zookeeper address required")
} }
config.logger.Println("Addresss obtained: ", url) config.logger.Println("Address obtained: ", url)
url, err = validateAuroraURL(url) url, err = validateAuroraURL(url)
if err != nil { if err != nil {
return nil, errors.Wrap(err, "invalid Aurora url") return nil, errors.Wrap(err, "invalid Aurora url")
@ -428,7 +428,8 @@ func NewRealisClient(options ...ClientOption) (Realis, error) {
adminClient: aurora.NewAuroraAdminClientFactory(config.transport, config.protoFactory), adminClient: aurora.NewAuroraAdminClientFactory(config.transport, config.protoFactory),
logger: LevelLogger{logger: config.logger, debug: config.debug, trace: config.trace}, logger: LevelLogger{logger: config.logger, debug: config.debug, trace: config.trace},
lock: &sync.Mutex{}, lock: &sync.Mutex{},
transport: config.transport}, nil transport: config.transport,
}, nil
} }
// GetDefaultClusterFromZKUrl creates a cluster object from a Zoookeper url. This is deprecated in favor of using // GetDefaultClusterFromZKUrl creates a cluster object from a Zoookeper url. This is deprecated in favor of using
@ -485,11 +486,11 @@ func defaultTTransport(url string, timeoutMs int, config *config) (thrift.TTrans
}) })
if err != nil { if err != nil {
return nil, errors.Wrap(err, "Error creating transport") return nil, errors.Wrap(err, "error creating transport")
} }
if err := trans.Open(); err != nil { if err := trans.Open(); err != nil {
return nil, errors.Wrapf(err, "Error opening connection to %s", url) return nil, errors.Wrapf(err, "error opening connection to %s", url)
} }
return trans, nil return trans, nil
@ -532,16 +533,17 @@ func (r *realisClient) ReestablishConn() error {
return nil return nil
} }
// Releases resources associated with the realis client. // Close releases resources associated with the realis client.
func (r *realisClient) Close() { func (r *realisClient) Close() {
r.lock.Lock() r.lock.Lock()
defer r.lock.Unlock() defer r.lock.Unlock()
r.transport.Close() // The return value of Close here is ignored on purpose because there's nothing that can be done if it fails.
_ = r.transport.Close()
} }
// Uses predefined set of states to retrieve a set of active jobs in Apache Aurora. // GetInstanceIds uses a predefined set of states to retrieve a set of active jobs in the Aurora Scheduler.
func (r *realisClient) GetInstanceIds(key *aurora.JobKey, states []aurora.ScheduleStatus) ([]int32, error) { func (r *realisClient) GetInstanceIds(key *aurora.JobKey, states []aurora.ScheduleStatus) ([]int32, error) {
taskQ := &aurora.TaskQuery{ taskQ := &aurora.TaskQuery{
JobKeys: []*aurora.JobKey{{Environment: key.Environment, Role: key.Role, Name: key.Name}}, JobKeys: []*aurora.JobKey{{Environment: key.Environment, Role: key.Role, Name: key.Name}},
@ -609,7 +611,7 @@ func (r *realisClient) GetJobs(role string) (*aurora.Response, *aurora.GetJobsRe
return resp, result, nil return resp, result, nil
} }
// Kill specific instances of a job. // KillInstances kills specific instances of a job.
func (r *realisClient) KillInstances(key *aurora.JobKey, instances ...int32) (*aurora.Response, error) { func (r *realisClient) KillInstances(key *aurora.JobKey, instances ...int32) (*aurora.Response, error) {
r.logger.debugPrintf("KillTasks Thrift Payload: %+v %v\n", key, instances) r.logger.debugPrintf("KillTasks Thrift Payload: %+v %v\n", key, instances)
@ -629,7 +631,7 @@ func (r *realisClient) RealisConfig() *config {
return r.config return r.config
} }
// Sends a kill message to the scheduler for all active tasks under a job. // KillJob kills all instances of a job.
func (r *realisClient) KillJob(key *aurora.JobKey) (*aurora.Response, error) { func (r *realisClient) KillJob(key *aurora.JobKey) (*aurora.Response, error) {
r.logger.debugPrintf("KillTasks Thrift Payload: %+v\n", key) r.logger.debugPrintf("KillTasks Thrift Payload: %+v\n", key)
@ -647,7 +649,7 @@ func (r *realisClient) KillJob(key *aurora.JobKey) (*aurora.Response, error) {
return resp, nil return resp, nil
} }
// Sends a create job message to the scheduler with a specific job configuration. // CreateJob sends a create job message to the scheduler with a specific job configuration.
// Although this API is able to create service jobs, it is better to use CreateService instead // Although this API is able to create service jobs, it is better to use CreateService instead
// as that API uses the update thrift call which has a few extra features available. // as that API uses the update thrift call which has a few extra features available.
// Use this API to create ad-hoc jobs. // Use this API to create ad-hoc jobs.
@ -667,7 +669,7 @@ func (r *realisClient) CreateJob(auroraJob Job) (*aurora.Response, error) {
return resp, nil return resp, nil
} }
// This API uses an update thrift call to create the services giving a few more robust features. // CreateService uses the scheduler's updating mechanism to create a job.
func (r *realisClient) CreateService( func (r *realisClient) CreateService(
auroraJob Job, auroraJob Job,
settings *aurora.JobUpdateSettings) (*aurora.Response, *aurora.StartJobUpdateResult_, error) { settings *aurora.JobUpdateSettings) (*aurora.Response, *aurora.StartJobUpdateResult_, error) {
@ -741,7 +743,7 @@ func (r *realisClient) StartCronJob(key *aurora.JobKey) (*aurora.Response, error
} }
// Restarts specific instances specified // RestartInstances restarts the specified instances of a Job.
func (r *realisClient) RestartInstances(key *aurora.JobKey, instances ...int32) (*aurora.Response, error) { func (r *realisClient) RestartInstances(key *aurora.JobKey, instances ...int32) (*aurora.Response, error) {
r.logger.debugPrintf("RestartShards Thrift Payload: %+v %v\n", key, instances) r.logger.debugPrintf("RestartShards Thrift Payload: %+v %v\n", key, instances)
@ -757,12 +759,12 @@ func (r *realisClient) RestartInstances(key *aurora.JobKey, instances ...int32)
return resp, nil return resp, nil
} }
// Restarts all active tasks under a job configuration. // RestartJob restarts all active instances of a Job.
func (r *realisClient) RestartJob(key *aurora.JobKey) (*aurora.Response, error) { func (r *realisClient) RestartJob(key *aurora.JobKey) (*aurora.Response, error) {
instanceIds, err1 := r.GetInstanceIds(key, aurora.ACTIVE_STATES) instanceIds, err1 := r.GetInstanceIds(key, aurora.ACTIVE_STATES)
if err1 != nil { if err1 != nil {
return nil, errors.Wrap(err1, "Could not retrieve relevant task instance IDs") return nil, errors.Wrap(err1, "could not retrieve relevant task instance IDs")
} }
r.logger.debugPrintf("RestartShards Thrift Payload: %+v %v\n", key, instanceIds) r.logger.debugPrintf("RestartShards Thrift Payload: %+v %v\n", key, instanceIds)
@ -784,7 +786,7 @@ func (r *realisClient) RestartJob(key *aurora.JobKey) (*aurora.Response, error)
return nil, errors.New("No tasks in the Active state") return nil, errors.New("No tasks in the Active state")
} }
// Update all tasks under a job configuration. Currently gorealis doesn't support for canary deployments. // StartJobUpdate updates all instances under a job configuration.
func (r *realisClient) StartJobUpdate(updateJob *UpdateJob, message string) (*aurora.Response, error) { func (r *realisClient) StartJobUpdate(updateJob *UpdateJob, message string) (*aurora.Response, error) {
r.logger.debugPrintf("StartJobUpdate Thrift Payload: %+v %v\n", updateJob, message) r.logger.debugPrintf("StartJobUpdate Thrift Payload: %+v %v\n", updateJob, message)
@ -806,7 +808,8 @@ func (r *realisClient) StartJobUpdate(updateJob *UpdateJob, message string) (*au
return resp, nil return resp, nil
} }
// Abort Job Update on Aurora. Requires the updateId which can be obtained on the Aurora web UI. // AbortJobUpdate terminates a job update in the scheduler.
// It requires the updateId which can be obtained on the Aurora web UI.
// This API is meant to be synchronous. It will attempt to wait until the update transitions to the aborted state. // This API is meant to be synchronous. It will attempt to wait until the update transitions to the aborted state.
// However, if the job update does not transition to the ABORT state an error will be returned. // However, if the job update does not transition to the ABORT state an error will be returned.
func (r *realisClient) AbortJobUpdate(updateKey aurora.JobUpdateKey, message string) (*aurora.Response, error) { func (r *realisClient) AbortJobUpdate(updateKey aurora.JobUpdateKey, message string) (*aurora.Response, error) {
@ -834,7 +837,8 @@ func (r *realisClient) AbortJobUpdate(updateKey aurora.JobUpdateKey, message str
return resp, err return resp, err
} }
// Pause Job Update. UpdateID is returned from StartJobUpdate or the Aurora web UI. // PauseJobUpdate pauses the progress of an ongoing update.
// The UpdateID value needed for this function is returned from StartJobUpdate or can be obtained from the Aurora web UI.
func (r *realisClient) PauseJobUpdate(updateKey *aurora.JobUpdateKey, message string) (*aurora.Response, error) { func (r *realisClient) PauseJobUpdate(updateKey *aurora.JobUpdateKey, message string) (*aurora.Response, error) {
r.logger.debugPrintf("PauseJobUpdate Thrift Payload: %+v %v\n", updateKey, message) r.logger.debugPrintf("PauseJobUpdate Thrift Payload: %+v %v\n", updateKey, message)
@ -852,7 +856,7 @@ func (r *realisClient) PauseJobUpdate(updateKey *aurora.JobUpdateKey, message st
return resp, nil return resp, nil
} }
// Resume Paused Job Update. UpdateID is returned from StartJobUpdate or the Aurora web UI. // ResumeJobUpdate resumes a previously Paused Job update.
func (r *realisClient) ResumeJobUpdate(updateKey *aurora.JobUpdateKey, message string) (*aurora.Response, error) { func (r *realisClient) ResumeJobUpdate(updateKey *aurora.JobUpdateKey, message string) (*aurora.Response, error) {
r.logger.debugPrintf("ResumeJobUpdate Thrift Payload: %+v %v\n", updateKey, message) r.logger.debugPrintf("ResumeJobUpdate Thrift Payload: %+v %v\n", updateKey, message)
@ -870,7 +874,7 @@ func (r *realisClient) ResumeJobUpdate(updateKey *aurora.JobUpdateKey, message s
return resp, nil return resp, nil
} }
// Pulse Job Update on Aurora. UpdateID is returned from StartJobUpdate or the Aurora web UI. // PulseJobUpdate sends a pulse to an ongoing Job update.
func (r *realisClient) PulseJobUpdate(updateKey *aurora.JobUpdateKey) (*aurora.Response, error) { func (r *realisClient) PulseJobUpdate(updateKey *aurora.JobUpdateKey) (*aurora.Response, error) {
r.logger.debugPrintf("PulseJobUpdate Thrift Payload: %+v\n", updateKey) r.logger.debugPrintf("PulseJobUpdate Thrift Payload: %+v\n", updateKey)
@ -888,8 +892,7 @@ func (r *realisClient) PulseJobUpdate(updateKey *aurora.JobUpdateKey) (*aurora.R
return resp, nil return resp, nil
} }
// Scale up the number of instances under a job configuration using the configuration for specific // AddInstances scales up the number of instances for a Job.
// instance to scale up.
func (r *realisClient) AddInstances(instKey aurora.InstanceKey, count int32) (*aurora.Response, error) { func (r *realisClient) AddInstances(instKey aurora.InstanceKey, count int32) (*aurora.Response, error) {
r.logger.debugPrintf("AddInstances Thrift Payload: %+v %v\n", instKey, count) r.logger.debugPrintf("AddInstances Thrift Payload: %+v %v\n", instKey, count)
@ -907,15 +910,15 @@ func (r *realisClient) AddInstances(instKey aurora.InstanceKey, count int32) (*a
} }
// Scale down the number of instances under a job configuration using the configuration of a specific instance // RemoveInstances scales down the number of instances for a Job.
func (r *realisClient) RemoveInstances(key *aurora.JobKey, count int32) (*aurora.Response, error) { func (r *realisClient) RemoveInstances(key *aurora.JobKey, count int32) (*aurora.Response, error) {
instanceIds, err := r.GetInstanceIds(key, aurora.ACTIVE_STATES) instanceIds, err := r.GetInstanceIds(key, aurora.ACTIVE_STATES)
if err != nil { if err != nil {
return nil, errors.Wrap(err, "RemoveInstances: Could not retrieve relevant instance IDs") return nil, errors.Wrap(err, "could not retrieve relevant instance IDs")
} }
if len(instanceIds) < int(count) { if len(instanceIds) < int(count) {
return nil, errors.Errorf("Insufficient active instances available for killing: "+ return nil, errors.Errorf("insufficient active instances available for killing: "+
" Instances to be killed %d Active instances %d", count, len(instanceIds)) " Instances to be killed %d Active instances %d", count, len(instanceIds))
} }
@ -928,7 +931,7 @@ func (r *realisClient) RemoveInstances(key *aurora.JobKey, count int32) (*aurora
return r.KillInstances(key, instanceIds[:count]...) return r.KillInstances(key, instanceIds[:count]...)
} }
// Get information about task including a fully hydrated task configuration object // GetTaskStatus gets information about task including a fully hydrated task configuration object.
func (r *realisClient) GetTaskStatus(query *aurora.TaskQuery) ([]*aurora.ScheduledTask, error) { func (r *realisClient) GetTaskStatus(query *aurora.TaskQuery) ([]*aurora.ScheduledTask, error) {
r.logger.debugPrintf("GetTasksStatus Thrift Payload: %+v\n", query) r.logger.debugPrintf("GetTasksStatus Thrift Payload: %+v\n", query)
@ -946,7 +949,7 @@ func (r *realisClient) GetTaskStatus(query *aurora.TaskQuery) ([]*aurora.Schedul
return response.ScheduleStatusResult(resp).GetTasks(), nil return response.ScheduleStatusResult(resp).GetTasks(), nil
} }
// Get pending reason // GetPendingReason returns the reason why the an instance of a Job has not been scheduled.
func (r *realisClient) GetPendingReason(query *aurora.TaskQuery) ([]*aurora.PendingReason, error) { func (r *realisClient) GetPendingReason(query *aurora.TaskQuery) ([]*aurora.PendingReason, error) {
r.logger.debugPrintf("GetPendingReason Thrift Payload: %+v\n", query) r.logger.debugPrintf("GetPendingReason Thrift Payload: %+v\n", query)
@ -970,7 +973,8 @@ func (r *realisClient) GetPendingReason(query *aurora.TaskQuery) ([]*aurora.Pend
return pendingReasons, nil return pendingReasons, nil
} }
// Get information about task including without a task configuration object // GetTasksWithoutConfigs gets information about task including without a task configuration object.
// This is a more lightweight version of GetTaskStatus but contains less information as a result.
func (r *realisClient) GetTasksWithoutConfigs(query *aurora.TaskQuery) ([]*aurora.ScheduledTask, error) { func (r *realisClient) GetTasksWithoutConfigs(query *aurora.TaskQuery) ([]*aurora.ScheduledTask, error) {
r.logger.debugPrintf("GetTasksWithoutConfigs Thrift Payload: %+v\n", query) r.logger.debugPrintf("GetTasksWithoutConfigs Thrift Payload: %+v\n", query)
@ -989,7 +993,7 @@ func (r *realisClient) GetTasksWithoutConfigs(query *aurora.TaskQuery) ([]*auror
} }
// Get the task configuration from the aurora scheduler for a job // FetchTaskConfig gets the task configuration from the aurora scheduler for a job.
func (r *realisClient) FetchTaskConfig(instKey aurora.InstanceKey) (*aurora.TaskConfig, error) { func (r *realisClient) FetchTaskConfig(instKey aurora.InstanceKey) (*aurora.TaskConfig, error) {
taskQ := &aurora.TaskQuery{ taskQ := &aurora.TaskQuery{
Role: &instKey.JobKey.Role, Role: &instKey.JobKey.Role,
@ -1014,7 +1018,7 @@ func (r *realisClient) FetchTaskConfig(instKey aurora.InstanceKey) (*aurora.Task
tasks := response.ScheduleStatusResult(resp).GetTasks() tasks := response.ScheduleStatusResult(resp).GetTasks()
if len(tasks) == 0 { if len(tasks) == 0 {
return nil, errors.Errorf("Instance %d for jobkey %s/%s/%s doesn't exist", return nil, errors.Errorf("instance %d for jobkey %s/%s/%s doesn't exist",
instKey.InstanceId, instKey.InstanceId,
instKey.JobKey.Environment, instKey.JobKey.Environment,
instKey.JobKey.Role, instKey.JobKey.Role,
@ -1036,7 +1040,7 @@ func (r *realisClient) JobUpdateDetails(updateQuery aurora.JobUpdateQuery) (*aur
}) })
if retryErr != nil { if retryErr != nil {
return nil, errors.Wrap(retryErr, "Unable to get job update details") return nil, errors.Wrap(retryErr, "unable to get job update details")
} }
return resp, nil return resp, nil

2
retry.go
View file

@ -179,7 +179,7 @@ func (r *realisClient) thriftCallWithRetries(
return nil, errors.Wrap(clientErr, "permanent connection error") return nil, errors.Wrap(clientErr, "permanent connection error")
} }
// Corner case where thrift payload was received by Aurora but connection timedout before Aurora was // Corner case where thrift payload was received by Aurora but connection timed out before Aurora was
// able to reply. In this case we will return whatever response was received and a TimedOut behaving // able to reply. In this case we will return whatever response was received and a TimedOut behaving
// error. Users can take special action on a timeout by using IsTimedout and reacting accordingly. // error. Users can take special action on a timeout by using IsTimedout and reacting accordingly.
if e.Timeout() { if e.Timeout() {