From d97e59b9e6863b7e982ca48df5f1ebddd3fcbec7 Mon Sep 17 00:00:00 2001 From: Kumar Krishna Date: Thu, 16 Mar 2017 23:19:30 -0700 Subject: [PATCH 1/3] inital commit gorealis resiliency --- examples/client.go | 76 ++++++++---- monitors.go | 58 ++++++++-- realis.go | 282 ++++++++++++++++++++++++++++++++++++++++----- 3 files changed, 355 insertions(+), 61 deletions(-) diff --git a/examples/client.go b/examples/client.go index 23a04a5..a25f049 100644 --- a/examples/client.go +++ b/examples/client.go @@ -34,6 +34,7 @@ func main() { updateId := flag.String("updateId", "", "Update ID to operate on") username := flag.String("username", "aurora", "Username to use for authorization") password := flag.String("password", "secret", "Password to use for authorization") + zkUrl := flag.String("zkurl", "", "zookeeper url") flag.Parse() // Attempt to load leader from zookeeper @@ -57,20 +58,46 @@ func main() { } } - //Create new configuration with default transport layer - config, err := realis.NewDefaultConfig(*url, 10000) - if err != nil { - fmt.Println(err) - os.Exit(1) - } - - // Configured for vagrant - realis.AddBasicAuth(&config, *username, *password) - r := realis.NewClient(config) - defer r.Close() - - monitor := &realis.Monitor{r} var job realis.Job + var err error + var config *realis.RealisConfig + var monitor *realis.Monitor + var r realis.Realis + + //check if zkUrl is available. + if *zkUrl != "" { + fmt.Println("zkUrl: ", *zkUrl) + cluster := &realis.Cluster{Name: "example", + AuthMechanism: "UNAUTHENTICATED", + ZK: *zkUrl, + SchedZKPath: "/aurora/scheduler", + AgentRunDir: "latest", + AgentRoot: "/var/lib/mesos", + } + fmt.Printf("cluster: %+v \n", cluster) + + r, err = realis.NewClientUsingCluster(cluster, *username, *password) + if err != nil { + fmt.Println(err) + os.Exit(1) + } + monitor = &realis.Monitor{r} + + } else { + //Create new configuration with default transport layer + config, err = realis.NewDefaultConfig(*url, 10000) + if err != nil { + fmt.Println(err) + os.Exit(1) + } + + // Configured for vagrant + realis.AddBasicAuth(config, *username, *password) + r = realis.NewClient(config) + + monitor = &realis.Monitor{r} + } + defer r.Close() switch *executor { case "thermos": @@ -95,17 +122,21 @@ func main() { break case "compose": job = realis.NewJob(). - Environment("prod"). - Role("vagrant"). - Name("docker-compose"). - ExecutorName("docker-compose-executor"). + //Environment("prod"). + //Role("vagrant"). + //Name("docker-compose"). + Role("gorealis"). + Environment("k2"). + Name("testapp"). + ExecutorName("sampleapp"). + ExecutorName("dce-regular"). ExecutorData("{}"). - CPU(0.5). + CPU(0.25). RAM(64). Disk(100). - IsService(false). - InstanceCount(3). - AddPorts(1). + IsService(true). + InstanceCount(4). + AddPorts(4). AddLabel("fileName", "sample-app/docker-compose.yml"). AddURIs(true, true, "https://github.com/mesos/docker-compose-executor/releases/download/0.1.0/sample-app.tar.gz") break @@ -143,7 +174,10 @@ func main() { fmt.Println(err) os.Exit(1) } + fmt.Println("ok: ", ok) + fmt.Println("err: ", err) } + } break case "createDocker": diff --git a/monitors.go b/monitors.go index 5719c96..7bdb14f 100644 --- a/monitors.go +++ b/monitors.go @@ -17,10 +17,11 @@ package realis import ( "fmt" + "time" + "github.com/pkg/errors" "github.com/rdelval/gorealis/gen-go/apache/aurora" "github.com/rdelval/gorealis/response" - "time" ) type Monitor struct { @@ -35,10 +36,31 @@ func (m *Monitor) JobUpdate(updateKey aurora.JobUpdateKey, interval int, timeout Limit: 1, } + duration := defaultBackoff.Duration + var err error + var respDetail *aurora.Response + for i := 0; i*interval <= timeout; i++ { - respDetail, err := m.Client.JobUpdateDetails(updateQ) + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if respDetail, err = m.Client.JobUpdateDetails(updateQ); err == nil { + break + } + err1 := m.Client.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } + } + if err != nil { - fmt.Println(err) return false, err } @@ -46,7 +68,7 @@ func (m *Monitor) JobUpdate(updateKey aurora.JobUpdateKey, interval int, timeout if len(updateDetail) == 0 { fmt.Println("No update found") - return false, errors.New("No update found for "+updateKey.String()) + return false, errors.New("No update found for " + updateKey.String()) } status := updateDetail[0].Update.Summary.State.Status @@ -72,19 +94,39 @@ func (m *Monitor) JobUpdate(updateKey aurora.JobUpdateKey, interval int, timeout } func (m *Monitor) Instances(key *aurora.JobKey, instances int32, interval int, timeout int) (bool, error) { + duration := defaultBackoff.Duration + var err error + var live map[int32]bool for i := 0; i*interval < timeout; i++ { + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + fmt.Println(" sleeping done") + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if live, err = m.Client.GetInstanceIds(key, aurora.LIVE_STATES); err == nil { + fmt.Println(" live: ", live) + break + } + err1 := m.Client.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } + } - live, err := m.Client.GetInstanceIds(key, aurora.LIVE_STATES) - + //live, err := m.Client.GetInstanceIds(key, aurora.LIVE_STATES) if err != nil { return false, errors.Wrap(err, "Unable to communicate with Aurora") } - if len(live) == int(instances) { return true, nil } - fmt.Println("Polling, instances running: ", len(live)) time.Sleep(time.Duration(interval) * time.Second) } diff --git a/realis.go b/realis.go index 5bd1c0b..74a912d 100644 --- a/realis.go +++ b/realis.go @@ -23,6 +23,8 @@ import ( "fmt" + "math/rand" + "git.apache.org/thrift.git/lib/go/thrift" "github.com/pkg/errors" "github.com/rdelval/gorealis/gen-go/apache/aurora" @@ -48,25 +50,83 @@ type Realis interface { StartJobUpdate(updateJob *UpdateJob, message string) (*aurora.Response, error) StartCronJob(key *aurora.JobKey) (*aurora.Response, error) GetJobUpdateSummaries(jobUpdateQuery *aurora.JobUpdateQuery) (*aurora.Response, error) + ReestablishConn() error Close() } type realisClient struct { + config *RealisConfig client *aurora.AuroraSchedulerManagerClient readonlyClient *aurora.ReadOnlySchedulerClient } // Wrapper object to provide future flexibility type RealisConfig struct { - transport thrift.TTransport - protoFactory thrift.TProtocolFactory + username, password string + cluster *Cluster + transport thrift.TTransport + protoFactory thrift.TProtocolFactory +} + +type Backoff struct { + Duration time.Duration // the base duration + Factor float64 // Duration is multipled by factor each iteration + Jitter float64 // The amount of jitter applied each iteration + Steps int // Exit with error after this many steps +} + +var defaultBackoff = Backoff{ + Steps: 10, + Duration: 5 * time.Second, + Factor: 5.0, + Jitter: 0.1, +} + +// Jitter returns a time.Duration between duration and duration + maxFactor * +// duration. +// +// This allows clients to avoid converging on periodic behavior. If maxFactor +// is 0.0, a suggested default value will be chosen. +func Jitter(duration time.Duration, maxFactor float64) time.Duration { + if maxFactor <= 0.0 { + maxFactor = 1.0 + } + wait := duration + time.Duration(rand.Float64()*maxFactor*float64(duration)) + return wait +} + +// Create a new Client with Cluster information and other details. + +func NewClientUsingCluster(cluster *Cluster, user, passwd string) (Realis, error) { + + url, err := LeaderFromZK(*cluster) + if err != nil { + fmt.Println(err) + return nil, err + } + fmt.Printf(" url: %s\n", url) + + //Create new configuration with default transport layer + config, err := NewDefaultConfig("http://localhost:18000", 10000) + if err != nil { + fmt.Println(err) + return nil, err + } + config.username = user + config.password = passwd + config.cluster = cluster + // Configured for vagrant + AddBasicAuth(config, user, passwd) + r := NewClient(config) + return r, nil } // Create a new Client with a default transport layer -func NewClient(config RealisConfig) Realis { +func NewClient(realisconfig *RealisConfig) Realis { return &realisClient{ - client: aurora.NewAuroraSchedulerManagerClientFactory(config.transport, config.protoFactory), - readonlyClient: aurora.NewReadOnlySchedulerClientFactory(config.transport, config.protoFactory)} + config: realisconfig, + client: aurora.NewAuroraSchedulerManagerClientFactory(realisconfig.transport, realisconfig.protoFactory), + readonlyClient: aurora.NewReadOnlySchedulerClientFactory(realisconfig.transport, realisconfig.protoFactory)} } // Creates a default Thrift Transport object for communications in gorealis using an HTTP Post Client @@ -92,28 +152,28 @@ func defaultTTransport(urlstr string, timeoutms int) (thrift.TTransport, error) // Create a default configuration of the transport layer, requires a URL to test connection with. // Uses HTTP Post as transport layer and Thrift JSON as the wire protocol by default. -func NewDefaultConfig(url string, timeoutms int) (RealisConfig, error) { +func NewDefaultConfig(url string, timeoutms int) (*RealisConfig, error) { return NewTJSONConfig(url, timeoutms) } // Creates a realis config object using HTTP Post and Thrift JSON protocol to communicate with Aurora. -func NewTJSONConfig(url string, timeoutms int) (RealisConfig, error) { +func NewTJSONConfig(url string, timeoutms int) (*RealisConfig, error) { trans, err := defaultTTransport(url, timeoutms) if err != nil { - return RealisConfig{}, errors.Wrap(err, "Error creating realis config") + return &RealisConfig{}, errors.Wrap(err, "Error creating realis config") } httpTrans := (trans).(*thrift.THttpClient) httpTrans.SetHeader("Content-Type", "application/x-thrift") - return RealisConfig{transport: trans, protoFactory: thrift.NewTJSONProtocolFactory()}, nil + return &RealisConfig{transport: trans, protoFactory: thrift.NewTJSONProtocolFactory()}, nil } // Creates a realis config config using HTTP Post and Thrift Binary protocol to communicate with Aurora. -func NewTBinaryConfig(url string, timeoutms int) (RealisConfig, error) { +func NewTBinaryConfig(url string, timeoutms int) (*RealisConfig, error) { trans, err := defaultTTransport(url, timeoutms) if err != nil { - return RealisConfig{}, errors.Wrap(err, "Error creating realis config") + return &RealisConfig{}, errors.Wrap(err, "Error creating realis config") } httpTrans := (trans).(*thrift.THttpClient) @@ -121,12 +181,14 @@ func NewTBinaryConfig(url string, timeoutms int) (RealisConfig, error) { httpTrans.SetHeader("Content-Type", "application/vnd.apache.thrift.binary") httpTrans.SetHeader("User-Agent", "GoRealis v1.0.4") - return RealisConfig{transport: trans, protoFactory: thrift.NewTBinaryProtocolFactoryDefault()}, nil + return &RealisConfig{transport: trans, protoFactory: thrift.NewTBinaryProtocolFactoryDefault()}, nil } // Helper function to add basic authorization needed to communicate with Apache Aurora. func AddBasicAuth(config *RealisConfig, username string, password string) { + config.username = username + config.password = password httpTrans := (config.transport).(*thrift.THttpClient) httpTrans.SetHeader("Authorization", "Basic "+basicAuth(username, password)) } @@ -136,6 +198,38 @@ func basicAuth(username, password string) string { return base64.StdEncoding.EncodeToString([]byte(auth)) } +func (r *realisClient) ReestablishConn() error { + //close existing connection.. + fmt.Println("ReestablishConn begin ....") + r.Close() + if r.config.cluster != nil && r.config.username != "" && r.config.password != "" { + url, err := LeaderFromZK(*r.config.cluster) + if err != nil { + fmt.Println("LeaderFromZK error: ", err) + return err + } + fmt.Println("ReestablishConn url: ", url) + config, err := NewDefaultConfig("http://localhost:18000", 10000) + if err != nil { + fmt.Println(err) + return err + } + // Configured for basic-auth + AddBasicAuth(config, r.config.username, r.config.password) + config.cluster = r.config.cluster + r.config = config + r.client = aurora.NewAuroraSchedulerManagerClientFactory(config.transport, config.protoFactory) + r.readonlyClient = aurora.NewReadOnlySchedulerClientFactory(config.transport, config.protoFactory) + } else { + fmt.Println(" Missing Data for ReestablishConn ") + fmt.Println(" r.config.cluster: ", r.config.cluster) + fmt.Println(" r.config.username: ", r.config.username) + fmt.Println(" r.config.passwd: ", r.config.password) + return errors.New(" Missing Data for ReestablishConn ") + } + return nil +} + // Releases resources associated with the realis client. func (r *realisClient) Close() { r.client.Transport.Close() @@ -192,33 +286,94 @@ func (r *realisClient) KillInstances(key *aurora.JobKey, instances ...int32) (*a // Sends a kill message to the scheduler for all active tasks under a job. func (r *realisClient) KillJob(key *aurora.JobKey) (*aurora.Response, error) { - instanceIds, err := r.GetInstanceIds(key, aurora.ACTIVE_STATES) + var instanceIds map[int32]bool + var err error + var resp *aurora.Response + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + + if instanceIds, err = r.GetInstanceIds(key, aurora.ACTIVE_STATES); err == nil { + fmt.Println("instanceIds: ", instanceIds) + break + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } + } + if err != nil { return nil, errors.Wrap(err, "Could not retrieve relevant task instance IDs") } if len(instanceIds) > 0 { - resp, err := r.client.KillTasks(key, instanceIds) + + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + + if resp, err = r.client.KillTasks(key, instanceIds); err == nil { + return response.ResponseCodeCheck(resp) + } + + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } + } if err != nil { return nil, errors.Wrap(err, "Error sending Kill command to Aurora Scheduler") } - return response.ResponseCodeCheck(resp) - } else { - return nil, errors.New("No tasks in the Active state") } + return nil, errors.New("No tasks in the Active state") + } // Sends a create job message to the scheduler with a specific job configuration. func (r *realisClient) CreateJob(auroraJob Job) (*aurora.Response, error) { - resp, err := r.client.CreateJob(auroraJob.JobConfig()) + var resp *aurora.Response + var err error - if err != nil { - return nil, errors.Wrap(err, "Error sending Create command to Aurora Scheduler") + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + + if resp, err = r.client.CreateJob(auroraJob.JobConfig()); err == nil { + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } } - - return response.ResponseCodeCheck(resp) + return nil, errors.Wrap(err, "Error sending Create command to Aurora Scheduler") } func (r *realisClient) ScheduleCronJob(auroraJob Job) (*aurora.Response, error) { @@ -291,13 +446,32 @@ func (r *realisClient) RestartJob(key *aurora.JobKey) (*aurora.Response, error) // Update all tasks under a job configuration. Currently gorealis doesn't support for canary deployments. func (r *realisClient) StartJobUpdate(updateJob *UpdateJob, message string) (*aurora.Response, error) { - resp, err := r.client.StartJobUpdate(updateJob.req, message) + var resp *aurora.Response + var err error - if err != nil { - return nil, errors.Wrap(err, "Error sending StartJobUpdate command to Aurora Scheduler") + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if resp, err = r.client.StartJobUpdate(updateJob.req, message); err == nil { + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } } - return response.ResponseCodeCheck(resp) + //resp, err = r.client.StartJobUpdate(updateJob.req, message) + return nil, errors.Wrap(err, "Error sending StartJobUpdate command to Aurora Scheduler") + } // Abort Job Update on Aurora. Requires the updateId which can be obtained on the Aurora web UI. @@ -374,7 +548,31 @@ func (r *realisClient) FetchTaskConfig(instKey aurora.InstanceKey) (*aurora.Task InstanceIds: ids, Statuses: aurora.ACTIVE_STATES} - resp, err := r.client.GetTasksStatus(taskQ) + var resp *aurora.Response + var err error + + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + + if resp, err = r.client.GetTasksStatus(taskQ); err == nil { + fmt.Println("resp: ", resp) + break + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } + } + if err != nil { return nil, errors.Wrap(err, "Error querying Aurora Scheduler for task configuration") } @@ -394,7 +592,7 @@ func (r *realisClient) FetchTaskConfig(instKey aurora.InstanceKey) (*aurora.Task instKey.JobKey.Name) } - // Currently, instance 0 is always picked + // Currently, instance 0 is always picked.. return tasks[0].AssignedTask.Task, nil } @@ -408,11 +606,31 @@ func (r *realisClient) JobUpdateDetails(updateQuery aurora.JobUpdateQuery) (*aur return response.ResponseCodeCheck(resp) } func (r *realisClient) RollbackJobUpdate(key aurora.JobUpdateKey, message string) (*aurora.Response, error) { + var resp *aurora.Response + var err error - resp, err := r.client.RollbackJobUpdate(&key, message) - if err != nil { - return nil, errors.Wrap(err, "Unable to roll back job update") + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if resp, err = r.client.RollbackJobUpdate(&key, message); err == nil { + fmt.Println(" resp: ", resp) + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } } - return response.ResponseCodeCheck(resp) + //resp, err = r.client.RollbackJobUpdate(&key, message) + return nil, errors.Wrap(err, "Unable to roll back job update") + } From 3add32a5858e50fc3f9d5a7ad6abdad201a85d6e Mon Sep 17 00:00:00 2001 From: Kumar Krishna Date: Mon, 20 Mar 2017 22:34:45 -0700 Subject: [PATCH 2/3] gorealis resiliency --- examples/client.go | 27 +-- monitors.go | 12 +- realis.go | 442 +++++++++++++++++++++++++++++++++++---------- 3 files changed, 367 insertions(+), 114 deletions(-) diff --git a/examples/client.go b/examples/client.go index a25f049..98a2370 100644 --- a/examples/client.go +++ b/examples/client.go @@ -60,7 +60,6 @@ func main() { var job realis.Job var err error - var config *realis.RealisConfig var monitor *realis.Monitor var r realis.Realis @@ -76,7 +75,7 @@ func main() { } fmt.Printf("cluster: %+v \n", cluster) - r, err = realis.NewClientUsingCluster(cluster, *username, *password) + r, err = realis.NewDefaultClientUsingCluster(cluster, *username, *password) if err != nil { fmt.Println(err) os.Exit(1) @@ -84,17 +83,11 @@ func main() { monitor = &realis.Monitor{r} } else { - //Create new configuration with default transport layer - config, err = realis.NewDefaultConfig(*url, 10000) + r, err = realis.NewDefaultClientUsingUrl(*url, *username, *password) if err != nil { fmt.Println(err) os.Exit(1) } - - // Configured for vagrant - realis.AddBasicAuth(config, *username, *password) - r = realis.NewClient(config) - monitor = &realis.Monitor{r} } defer r.Close() @@ -122,20 +115,16 @@ func main() { break case "compose": job = realis.NewJob(). - //Environment("prod"). - //Role("vagrant"). - //Name("docker-compose"). - Role("gorealis"). - Environment("k2"). - Name("testapp"). - ExecutorName("sampleapp"). - ExecutorName("dce-regular"). + Environment("prod"). + Role("vagrant"). + Name("docker-compose"). + ExecutorName("docker-compose-executor"). ExecutorData("{}"). CPU(0.25). RAM(64). Disk(100). IsService(true). - InstanceCount(4). + InstanceCount(2). AddPorts(4). AddLabel("fileName", "sample-app/docker-compose.yml"). AddURIs(true, true, "https://github.com/mesos/docker-compose-executor/releases/download/0.1.0/sample-app.tar.gz") @@ -307,7 +296,7 @@ func main() { case "flexUp": fmt.Println("Flexing up job") - numOfInstances := int32(2) + numOfInstances := int32(4) live, err := r.GetInstanceIds(job.JobKey(), aurora.ACTIVE_STATES) if err != nil { diff --git a/monitors.go b/monitors.go index 7bdb14f..d7f863f 100644 --- a/monitors.go +++ b/monitors.go @@ -59,7 +59,7 @@ func (m *Monitor) JobUpdate(updateKey aurora.JobUpdateKey, interval int, timeout fmt.Println("error in ReestablishConn: ", err1) } } - + // if error remains then return (false, err). if err != nil { return false, err } @@ -114,10 +114,14 @@ func (m *Monitor) Instances(key *aurora.JobKey, instances int32, interval int, t fmt.Println(" live: ", live) break } - err1 := m.Client.ReestablishConn() - if err1 != nil { - fmt.Println("error in ReestablishConn: ", err1) + + if err != nil { + err1 := m.Client.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } } + } //live, err := m.Client.GetInstanceIds(key, aurora.LIVE_STATES) diff --git a/realis.go b/realis.go index 74a912d..a7e5b1f 100644 --- a/realis.go +++ b/realis.go @@ -63,6 +63,7 @@ type realisClient struct { // Wrapper object to provide future flexibility type RealisConfig struct { username, password string + url string cluster *Cluster transport thrift.TTransport protoFactory thrift.TProtocolFactory @@ -76,8 +77,8 @@ type Backoff struct { } var defaultBackoff = Backoff{ - Steps: 10, - Duration: 5 * time.Second, + Steps: 3, + Duration: 10 * time.Second, Factor: 5.0, Jitter: 0.1, } @@ -97,7 +98,7 @@ func Jitter(duration time.Duration, maxFactor float64) time.Duration { // Create a new Client with Cluster information and other details. -func NewClientUsingCluster(cluster *Cluster, user, passwd string) (Realis, error) { +func NewDefaultClientUsingCluster(cluster *Cluster, user, passwd string) (Realis, error) { url, err := LeaderFromZK(*cluster) if err != nil { @@ -115,6 +116,61 @@ func NewClientUsingCluster(cluster *Cluster, user, passwd string) (Realis, error config.username = user config.password = passwd config.cluster = cluster + config.url = "" + // Configured for vagrant + AddBasicAuth(config, user, passwd) + r := NewClient(config) + return r, nil +} + +//This api would create default cluster object.. +func NewDefaultClientUsingZKUrl(zkUrl, user, passwd string) (Realis, error) { + + fmt.Println(" zkUrl: %s", zkUrl) + cluster := &Cluster{Name: "testCluster", + AuthMechanism: "UNAUTHENTICATED", + ZK: zkUrl, + SchedZKPath: "/aurora/scheduler", + AgentRunDir: "latest", + AgentRoot: "/var/lib/mesos", + } + + url, err := LeaderFromZK(*cluster) + if err != nil { + fmt.Println(err) + return nil, err + } + fmt.Printf(" url: %s\n", url) + + //Create new configuration with default transport layer + config, err := NewDefaultConfig("http://localhost:18000", 10000) + if err != nil { + fmt.Println(err) + return nil, err + } + config.username = user + config.password = passwd + config.cluster = cluster + config.url = "" + // Configured for vagrant + AddBasicAuth(config, user, passwd) + r := NewClient(config) + return r, nil +} + +func NewDefaultClientUsingUrl(url, user, passwd string) (Realis, error) { + + fmt.Printf(" url: %s\n", url) + //Create new configuration with default transport layer + config, err := NewDefaultConfig("http://localhost:18000", 10000) + if err != nil { + fmt.Println(err) + return nil, err + } + config.username = user + config.password = passwd + config.url = url + config.cluster = nil // Configured for vagrant AddBasicAuth(config, user, passwd) r := NewClient(config) @@ -202,17 +258,17 @@ func (r *realisClient) ReestablishConn() error { //close existing connection.. fmt.Println("ReestablishConn begin ....") r.Close() + //First check cluster object for re-establish; if not available then try with scheduler url. if r.config.cluster != nil && r.config.username != "" && r.config.password != "" { + //Re-establish using cluster object. url, err := LeaderFromZK(*r.config.cluster) if err != nil { - fmt.Println("LeaderFromZK error: ", err) - return err + fmt.Errorf("LeaderFromZK error: %+v\n ", err) } fmt.Println("ReestablishConn url: ", url) - config, err := NewDefaultConfig("http://localhost:18000", 10000) + config, err := NewDefaultConfig(url, 10000) if err != nil { fmt.Println(err) - return err } // Configured for basic-auth AddBasicAuth(config, r.config.username, r.config.password) @@ -220,11 +276,23 @@ func (r *realisClient) ReestablishConn() error { r.config = config r.client = aurora.NewAuroraSchedulerManagerClientFactory(config.transport, config.protoFactory) r.readonlyClient = aurora.NewReadOnlySchedulerClientFactory(config.transport, config.protoFactory) + } else if r.config.url != "" && r.config.username != "" && r.config.password != "" { + //Re-establish using scheduler url. + //Create new configuration with default transport layer + config, err := NewDefaultConfig(r.config.url, 10000) + if err != nil { + fmt.Println(err) + } + AddBasicAuth(config, r.config.username, r.config.password) + r.config = config + r.client = aurora.NewAuroraSchedulerManagerClientFactory(config.transport, config.protoFactory) + r.readonlyClient = aurora.NewReadOnlySchedulerClientFactory(config.transport, config.protoFactory) } else { fmt.Println(" Missing Data for ReestablishConn ") fmt.Println(" r.config.cluster: ", r.config.cluster) fmt.Println(" r.config.username: ", r.config.username) fmt.Println(" r.config.passwd: ", r.config.password) + fmt.Println(" r.config.url: ", r.config.url) return errors.New(" Missing Data for ReestablishConn ") } return nil @@ -243,7 +311,30 @@ func (r *realisClient) GetInstanceIds(key *aurora.JobKey, states map[aurora.Sche JobName: key.Name, Statuses: states} - resp, err := r.client.GetTasksWithoutConfigs(taskQ) + var resp *aurora.Response + var err error + + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if resp, err = r.client.GetTasksWithoutConfigs(taskQ); err == nil { + fmt.Println("resp: ", resp) + break + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } + } + if err != nil { return nil, errors.Wrap(err, "Error querying Aurora Scheduler for active IDs") } @@ -259,36 +350,9 @@ func (r *realisClient) GetInstanceIds(key *aurora.JobKey, states map[aurora.Sche } func (r *realisClient) GetJobUpdateSummaries(jobUpdateQuery *aurora.JobUpdateQuery) (*aurora.Response, error) { - resp, err := r.readonlyClient.GetJobUpdateSummaries(jobUpdateQuery) - if err != nil { - return nil, errors.Wrap(err, "Error getting job update summaries from Aurora Scheduler") - } - return response.ResponseCodeCheck(resp) -} - -// Kill specific instances of a job. -func (r *realisClient) KillInstances(key *aurora.JobKey, instances ...int32) (*aurora.Response, error) { - - instanceIds := make(map[int32]bool) - - for _, instId := range instances { - instanceIds[instId] = true - } - - resp, err := r.client.KillTasks(key, instanceIds) - if err != nil { - return nil, errors.Wrap(err, "Error sending Kill command to Aurora Scheduler") - } - - return response.ResponseCodeCheck(resp) -} - -// Sends a kill message to the scheduler for all active tasks under a job. -func (r *realisClient) KillJob(key *aurora.JobKey) (*aurora.Response, error) { - - var instanceIds map[int32]bool - var err error var resp *aurora.Response + var err error + duration := defaultBackoff.Duration for i := 0; i < defaultBackoff.Steps; i++ { if i != 0 { @@ -300,17 +364,61 @@ func (r *realisClient) KillJob(key *aurora.JobKey) (*aurora.Response, error) { time.Sleep(adjusted) duration = time.Duration(float64(duration) * defaultBackoff.Factor) } - - if instanceIds, err = r.GetInstanceIds(key, aurora.ACTIVE_STATES); err == nil { - fmt.Println("instanceIds: ", instanceIds) - break + if resp, err = r.readonlyClient.GetJobUpdateSummaries(jobUpdateQuery); err == nil { + fmt.Println("resp: ", resp) + return response.ResponseCodeCheck(resp) } err1 := r.ReestablishConn() if err1 != nil { fmt.Println("error in ReestablishConn: ", err1) } + + } + return nil, errors.Wrap(err, "Error getting job update summaries from Aurora Scheduler") +} + +// Kill specific instances of a job. +func (r *realisClient) KillInstances(key *aurora.JobKey, instances ...int32) (*aurora.Response, error) { + + instanceIds := make(map[int32]bool) + var resp *aurora.Response + var err error + duration := defaultBackoff.Duration + + for _, instId := range instances { + instanceIds[instId] = true } + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if resp, err = r.client.KillTasks(key, instanceIds); err == nil { + fmt.Println("resp: ", resp) + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } + + } + return nil, errors.Wrap(err, "Error sending Kill command to Aurora Scheduler") +} + +// Sends a kill message to the scheduler for all active tasks under a job. +func (r *realisClient) KillJob(key *aurora.JobKey) (*aurora.Response, error) { + + var instanceIds map[int32]bool + var err error + var resp *aurora.Response + instanceIds, err = r.GetInstanceIds(key, aurora.ACTIVE_STATES) if err != nil { return nil, errors.Wrap(err, "Could not retrieve relevant task instance IDs") } @@ -338,14 +446,11 @@ func (r *realisClient) KillJob(key *aurora.JobKey) (*aurora.Response, error) { fmt.Println("error in ReestablishConn: ", err1) } } - if err != nil { return nil, errors.Wrap(err, "Error sending Kill command to Aurora Scheduler") } - } return nil, errors.New("No tasks in the Active state") - } // Sends a create job message to the scheduler with a specific job configuration. @@ -377,33 +482,87 @@ func (r *realisClient) CreateJob(auroraJob Job) (*aurora.Response, error) { } func (r *realisClient) ScheduleCronJob(auroraJob Job) (*aurora.Response, error) { - resp, err := r.client.ScheduleCronJob(auroraJob.JobConfig()) + var resp *aurora.Response + var err error - if err != nil { - return nil, errors.Wrap(err, "Error sending Cron Job Schedule message to Aurora Scheduler") + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + + if resp, err = r.client.ScheduleCronJob(auroraJob.JobConfig()); err == nil { + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } } - - return response.ResponseCodeCheck(resp) + return nil, errors.Wrap(err, "Error sending Cron Job Schedule message to Aurora Scheduler") } func (r *realisClient) DescheduleCronJob(key *aurora.JobKey) (*aurora.Response, error) { - resp, err := r.client.DescheduleCronJob(key) - if err != nil { - return nil, errors.Wrap(err, "Error sending Cron Job De-schedule message to Aurora Scheduler") + var resp *aurora.Response + var err error + + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + + if resp, err = r.client.DescheduleCronJob(key); err == nil { + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } } - return response.ResponseCodeCheck(resp) + return nil, errors.Wrap(err, "Error sending Cron Job De-schedule message to Aurora Scheduler") } func (r *realisClient) StartCronJob(key *aurora.JobKey) (*aurora.Response, error) { - resp, err := r.client.StartCronJob(key) + var resp *aurora.Response + var err error - if err != nil { - return nil, errors.Wrap(err, "Error sending Start Cron Job message to Aurora Scheduler") + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + + if resp, err = r.client.StartCronJob(key); err == nil { + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } } - return response.ResponseCodeCheck(resp) + return nil, errors.Wrap(err, "Error sending Start Cron Job message to Aurora Scheduler") } // Restarts specific instances specified @@ -413,31 +572,66 @@ func (r *realisClient) RestartInstances(key *aurora.JobKey, instances ...int32) for _, instId := range instances { instanceIds[instId] = true } + var resp *aurora.Response + var err error + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + + if resp, err = r.client.RestartShards(key, instanceIds); err == nil { + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } - resp, err := r.client.RestartShards(key, instanceIds) - if err != nil { - return nil, errors.Wrap(err, "Error sending Restart command to Aurora Scheduler") } - - return response.ResponseCodeCheck(resp) + return nil, errors.Wrap(err, "Error sending Restart command to Aurora Scheduler") } // Restarts all active tasks under a job configuration. func (r *realisClient) RestartJob(key *aurora.JobKey) (*aurora.Response, error) { - instanceIds, err := r.GetInstanceIds(key, aurora.ACTIVE_STATES) - if err != nil { - return nil, errors.Wrap(err, "Could not retrieve relevant task instance IDs") + instanceIds, err1 := r.GetInstanceIds(key, aurora.ACTIVE_STATES) + if err1 != nil { + return nil, errors.Wrap(err1, "Could not retrieve relevant task instance IDs") } + var resp *aurora.Response + var err error if len(instanceIds) > 0 { - resp, err := r.client.RestartShards(key, instanceIds) + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + + if resp, err = r.client.RestartShards(key, instanceIds); err == nil { + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } - if err != nil { - return nil, errors.Wrap(err, "Error sending Restart command to Aurora Scheduler") } + return nil, errors.Wrap(err, "Error sending Restart command to Aurora Scheduler") - return response.ResponseCodeCheck(resp) } else { return nil, errors.New("No tasks in the Active state") } @@ -468,10 +662,7 @@ func (r *realisClient) StartJobUpdate(updateJob *UpdateJob, message string) (*au fmt.Println("error in ReestablishConn: ", err1) } } - - //resp, err = r.client.StartJobUpdate(updateJob.req, message) return nil, errors.Wrap(err, "Error sending StartJobUpdate command to Aurora Scheduler") - } // Abort Job Update on Aurora. Requires the updateId which can be obtained on the Aurora web UI. @@ -479,26 +670,57 @@ func (r *realisClient) AbortJobUpdate( updateKey aurora.JobUpdateKey, message string) (*aurora.Response, error) { - resp, err := r.client.AbortJobUpdate(&updateKey, message) - - if err != nil { - return nil, errors.Wrap(err, "Error sending AbortJobUpdate command to Aurora Scheduler") + var resp *aurora.Response + var err error + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if resp, err = r.client.AbortJobUpdate(&updateKey, message); err == nil { + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } } - return response.ResponseCodeCheck(resp) + return nil, errors.Wrap(err, "Error sending AbortJobUpdate command to Aurora Scheduler") } // Scale up the number of instances under a job configuration using the configuration for specific // instance to scale up. func (r *realisClient) AddInstances(instKey aurora.InstanceKey, count int32) (*aurora.Response, error) { - resp, err := r.client.AddInstances(&instKey, count) - - if err != nil { - return nil, errors.Wrap(err, "Error sending AddInstances command to Aurora Scheduler") + var resp *aurora.Response + var err error + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if resp, err = r.client.AddInstances(&instKey, count); err == nil { + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } } - - return response.ResponseCodeCheck(resp) + return nil, errors.Wrap(err, "Error sending AddInstances command to Aurora Scheduler") } //Scale down the number of instances under a job configuration using the configuratipn of a specific instance @@ -525,7 +747,28 @@ func (r *realisClient) RemoveInstances(key *aurora.JobKey, count int32) (*aurora func (r *realisClient) GetTaskStatus(query *aurora.TaskQuery) (tasks []*aurora.ScheduledTask, e error) { - resp, err := r.client.GetTasksStatus(query) + var resp *aurora.Response + var err error + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if resp, err = r.client.GetTasksStatus(query); err == nil { + break + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } + } + if err != nil { return nil, errors.Wrap(err, "Error querying Aurora Scheduler for task status") } @@ -598,13 +841,32 @@ func (r *realisClient) FetchTaskConfig(instKey aurora.InstanceKey) (*aurora.Task func (r *realisClient) JobUpdateDetails(updateQuery aurora.JobUpdateQuery) (*aurora.Response, error) { - resp, err := r.client.GetJobUpdateDetails(&updateQuery) - if err != nil { - return nil, errors.Wrap(err, "Unable to get job update details") - } + var resp *aurora.Response + var err error - return response.ResponseCodeCheck(resp) + duration := defaultBackoff.Duration + for i := 0; i < defaultBackoff.Steps; i++ { + if i != 0 { + adjusted := duration + if defaultBackoff.Jitter > 0.0 { + adjusted = Jitter(duration, defaultBackoff.Jitter) + } + fmt.Println(" sleeping for: ", adjusted) + time.Sleep(adjusted) + duration = time.Duration(float64(duration) * defaultBackoff.Factor) + } + if resp, err = r.client.GetJobUpdateDetails(&updateQuery); err == nil { + fmt.Println(" resp: ", resp) + return response.ResponseCodeCheck(resp) + } + err1 := r.ReestablishConn() + if err1 != nil { + fmt.Println("error in ReestablishConn: ", err1) + } + } + return nil, errors.Wrap(err, "Unable to get job update details") } + func (r *realisClient) RollbackJobUpdate(key aurora.JobUpdateKey, message string) (*aurora.Response, error) { var resp *aurora.Response var err error @@ -630,7 +892,5 @@ func (r *realisClient) RollbackJobUpdate(key aurora.JobUpdateKey, message string } } - //resp, err = r.client.RollbackJobUpdate(&key, message) return nil, errors.Wrap(err, "Unable to roll back job update") - } From 9c4c2fce44e07fb8f8f35649d7aa8558a4567553 Mon Sep 17 00:00:00 2001 From: Kumar Krishna Date: Tue, 21 Mar 2017 16:59:27 -0700 Subject: [PATCH 3/3] making some methods private. --- examples/client.go | 2 +- realis.go | 50 ++++++++++++++++++++++------------------------ 2 files changed, 25 insertions(+), 27 deletions(-) diff --git a/examples/client.go b/examples/client.go index 98a2370..00f4159 100644 --- a/examples/client.go +++ b/examples/client.go @@ -124,7 +124,7 @@ func main() { RAM(64). Disk(100). IsService(true). - InstanceCount(2). + InstanceCount(1). AddPorts(4). AddLabel("fileName", "sample-app/docker-compose.yml"). AddURIs(true, true, "https://github.com/mesos/docker-compose-executor/releases/download/0.1.0/sample-app.tar.gz") diff --git a/realis.go b/realis.go index a7e5b1f..339f07e 100644 --- a/realis.go +++ b/realis.go @@ -108,7 +108,7 @@ func NewDefaultClientUsingCluster(cluster *Cluster, user, passwd string) (Realis fmt.Printf(" url: %s\n", url) //Create new configuration with default transport layer - config, err := NewDefaultConfig("http://localhost:18000", 10000) + config, err := newDefaultConfig(url, 10000) if err != nil { fmt.Println(err) return nil, err @@ -119,21 +119,25 @@ func NewDefaultClientUsingCluster(cluster *Cluster, user, passwd string) (Realis config.url = "" // Configured for vagrant AddBasicAuth(config, user, passwd) - r := NewClient(config) + r := newClient(config) return r, nil } +func GetDefaultClusterFromZKUrl(zkurl string) *Cluster { + return &Cluster{Name: "defaultCluster", + AuthMechanism: "UNAUTHENTICATED", + ZK: zkurl, + SchedZKPath: "/aurora/scheduler", + AgentRunDir: "latest", + AgentRoot: "/var/lib/mesos", + } +} + //This api would create default cluster object.. func NewDefaultClientUsingZKUrl(zkUrl, user, passwd string) (Realis, error) { fmt.Println(" zkUrl: %s", zkUrl) - cluster := &Cluster{Name: "testCluster", - AuthMechanism: "UNAUTHENTICATED", - ZK: zkUrl, - SchedZKPath: "/aurora/scheduler", - AgentRunDir: "latest", - AgentRoot: "/var/lib/mesos", - } + cluster := GetDefaultClusterFromZKUrl(zkUrl) url, err := LeaderFromZK(*cluster) if err != nil { @@ -143,7 +147,7 @@ func NewDefaultClientUsingZKUrl(zkUrl, user, passwd string) (Realis, error) { fmt.Printf(" url: %s\n", url) //Create new configuration with default transport layer - config, err := NewDefaultConfig("http://localhost:18000", 10000) + config, err := newDefaultConfig(url, 10000) if err != nil { fmt.Println(err) return nil, err @@ -154,7 +158,7 @@ func NewDefaultClientUsingZKUrl(zkUrl, user, passwd string) (Realis, error) { config.url = "" // Configured for vagrant AddBasicAuth(config, user, passwd) - r := NewClient(config) + r := newClient(config) return r, nil } @@ -162,7 +166,7 @@ func NewDefaultClientUsingUrl(url, user, passwd string) (Realis, error) { fmt.Printf(" url: %s\n", url) //Create new configuration with default transport layer - config, err := NewDefaultConfig("http://localhost:18000", 10000) + config, err := newDefaultConfig(url, 10000) if err != nil { fmt.Println(err) return nil, err @@ -173,12 +177,12 @@ func NewDefaultClientUsingUrl(url, user, passwd string) (Realis, error) { config.cluster = nil // Configured for vagrant AddBasicAuth(config, user, passwd) - r := NewClient(config) + r := newClient(config) return r, nil } // Create a new Client with a default transport layer -func NewClient(realisconfig *RealisConfig) Realis { +func newClient(realisconfig *RealisConfig) Realis { return &realisClient{ config: realisconfig, client: aurora.NewAuroraSchedulerManagerClientFactory(realisconfig.transport, realisconfig.protoFactory), @@ -208,12 +212,12 @@ func defaultTTransport(urlstr string, timeoutms int) (thrift.TTransport, error) // Create a default configuration of the transport layer, requires a URL to test connection with. // Uses HTTP Post as transport layer and Thrift JSON as the wire protocol by default. -func NewDefaultConfig(url string, timeoutms int) (*RealisConfig, error) { - return NewTJSONConfig(url, timeoutms) +func newDefaultConfig(url string, timeoutms int) (*RealisConfig, error) { + return newTJSONConfig(url, timeoutms) } // Creates a realis config object using HTTP Post and Thrift JSON protocol to communicate with Aurora. -func NewTJSONConfig(url string, timeoutms int) (*RealisConfig, error) { +func newTJSONConfig(url string, timeoutms int) (*RealisConfig, error) { trans, err := defaultTTransport(url, timeoutms) if err != nil { return &RealisConfig{}, errors.Wrap(err, "Error creating realis config") @@ -226,7 +230,7 @@ func NewTJSONConfig(url string, timeoutms int) (*RealisConfig, error) { } // Creates a realis config config using HTTP Post and Thrift Binary protocol to communicate with Aurora. -func NewTBinaryConfig(url string, timeoutms int) (*RealisConfig, error) { +func newTBinaryConfig(url string, timeoutms int) (*RealisConfig, error) { trans, err := defaultTTransport(url, timeoutms) if err != nil { return &RealisConfig{}, errors.Wrap(err, "Error creating realis config") @@ -266,7 +270,7 @@ func (r *realisClient) ReestablishConn() error { fmt.Errorf("LeaderFromZK error: %+v\n ", err) } fmt.Println("ReestablishConn url: ", url) - config, err := NewDefaultConfig(url, 10000) + config, err := newDefaultConfig(url, 10000) if err != nil { fmt.Println(err) } @@ -279,7 +283,7 @@ func (r *realisClient) ReestablishConn() error { } else if r.config.url != "" && r.config.username != "" && r.config.password != "" { //Re-establish using scheduler url. //Create new configuration with default transport layer - config, err := NewDefaultConfig(r.config.url, 10000) + config, err := newDefaultConfig(r.config.url, 10000) if err != nil { fmt.Println(err) } @@ -326,7 +330,6 @@ func (r *realisClient) GetInstanceIds(key *aurora.JobKey, states map[aurora.Sche duration = time.Duration(float64(duration) * defaultBackoff.Factor) } if resp, err = r.client.GetTasksWithoutConfigs(taskQ); err == nil { - fmt.Println("resp: ", resp) break } err1 := r.ReestablishConn() @@ -365,7 +368,6 @@ func (r *realisClient) GetJobUpdateSummaries(jobUpdateQuery *aurora.JobUpdateQue duration = time.Duration(float64(duration) * defaultBackoff.Factor) } if resp, err = r.readonlyClient.GetJobUpdateSummaries(jobUpdateQuery); err == nil { - fmt.Println("resp: ", resp) return response.ResponseCodeCheck(resp) } err1 := r.ReestablishConn() @@ -400,7 +402,6 @@ func (r *realisClient) KillInstances(key *aurora.JobKey, instances ...int32) (*a duration = time.Duration(float64(duration) * defaultBackoff.Factor) } if resp, err = r.client.KillTasks(key, instanceIds); err == nil { - fmt.Println("resp: ", resp) return response.ResponseCodeCheck(resp) } err1 := r.ReestablishConn() @@ -807,7 +808,6 @@ func (r *realisClient) FetchTaskConfig(instKey aurora.InstanceKey) (*aurora.Task } if resp, err = r.client.GetTasksStatus(taskQ); err == nil { - fmt.Println("resp: ", resp) break } err1 := r.ReestablishConn() @@ -856,7 +856,6 @@ func (r *realisClient) JobUpdateDetails(updateQuery aurora.JobUpdateQuery) (*aur duration = time.Duration(float64(duration) * defaultBackoff.Factor) } if resp, err = r.client.GetJobUpdateDetails(&updateQuery); err == nil { - fmt.Println(" resp: ", resp) return response.ResponseCodeCheck(resp) } err1 := r.ReestablishConn() @@ -883,7 +882,6 @@ func (r *realisClient) RollbackJobUpdate(key aurora.JobUpdateKey, message string duration = time.Duration(float64(duration) * defaultBackoff.Factor) } if resp, err = r.client.RollbackJobUpdate(&key, message); err == nil { - fmt.Println(" resp: ", resp) return response.ResponseCodeCheck(resp) } err1 := r.ReestablishConn()