Update go-marathon

This commit is contained in:
Timo Reimann 2017-12-19 16:00:09 +01:00 committed by Traefiker
parent 3142a4f4b3
commit 877770f7cf
24 changed files with 450 additions and 97 deletions

2
glide.lock generated
View file

@ -261,7 +261,7 @@ imports:
- name: github.com/fatih/color
version: 62e9147c64a1ed519147b62a56a14e83e2be02c1
- name: github.com/gambol99/go-marathon
version: dd6cbd4c2d71294a19fb89158f2a00d427f174ab
version: 03b46169666c53b9cc953b875ac5714e5103e064
- name: github.com/ghodss/yaml
version: 73d445a93680fa1a78ae23a5839bad48f32ba1ee
- name: github.com/go-ini/ini

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -56,15 +56,16 @@ type Port struct {
// Application is the definition for an application in marathon
type Application struct {
ID string `json:"id,omitempty"`
Cmd *string `json:"cmd,omitempty"`
Args *[]string `json:"args,omitempty"`
Constraints *[][]string `json:"constraints,omitempty"`
Container *Container `json:"container,omitempty"`
CPUs float64 `json:"cpus,omitempty"`
GPUs *float64 `json:"gpus,omitempty"`
Disk *float64 `json:"disk,omitempty"`
Env *map[string]string `json:"env,omitempty"`
ID string `json:"id,omitempty"`
Cmd *string `json:"cmd,omitempty"`
Args *[]string `json:"args,omitempty"`
Constraints *[][]string `json:"constraints,omitempty"`
Container *Container `json:"container,omitempty"`
CPUs float64 `json:"cpus,omitempty"`
GPUs *float64 `json:"gpus,omitempty"`
Disk *float64 `json:"disk,omitempty"`
// Contains non-secret environment variables. Secrets environment variables are part of the Secrets map.
Env *map[string]string `json:"-"`
Executor *string `json:"executor,omitempty"`
HealthChecks *[]HealthCheck `json:"healthChecks,omitempty"`
ReadinessChecks *[]ReadinessCheck `json:"readinessChecks,omitempty"`
@ -99,6 +100,8 @@ type Application struct {
LastTaskFailure *LastTaskFailure `json:"lastTaskFailure,omitempty"`
Fetch *[]Fetch `json:"fetch,omitempty"`
IPAddressPerTask *IPAddressPerTask `json:"ipAddress,omitempty"`
Residency *Residency `json:"residency,omitempty"`
Secrets *map[string]Secret `json:"-"`
}
// ApplicationVersions is a collection of application versions for a specific app in marathon
@ -149,6 +152,14 @@ type Stats struct {
LifeTime map[string]float64 `json:"lifeTime"`
}
// Secret is the environment variable and secret store path associated with a secret.
// The value for EnvVar is populated from the env field, and Source is populated from
// the secrets field of the application json.
type Secret struct {
EnvVar string
Source string
}
// SetIPAddressPerTask defines that the application will have a IP address defines by a external agent.
// This configuration is not allowed to be used with Port or PortDefinitions. Thus, the implementation
// clears both.
@ -355,8 +366,8 @@ func (r *Application) EmptyLabels() *Application {
}
// AddEnv adds an environment variable to the application
// name: the name of the variable
// value: go figure, the value associated to the above
// name: the name of the variable
// value: go figure, the value associated to the above
func (r *Application) AddEnv(name, value string) *Application {
if r.Env == nil {
r.EmptyEnvs()
@ -375,6 +386,28 @@ func (r *Application) EmptyEnvs() *Application {
return r
}
// AddSecret adds a secret declaration
// envVar: the name of the environment variable
// name: the name of the secret
// source: the source ID of the secret
func (r *Application) AddSecret(envVar, name, source string) *Application {
if r.Secrets == nil {
r.EmptySecrets()
}
(*r.Secrets)[name] = Secret{EnvVar: envVar, Source: source}
return r
}
// EmptySecrets explicitly empties the secrets -- use this if you need to empty
// the secrets of an application that already has secrets set (setting secrets to nil will
// keep the current value)
func (r *Application) EmptySecrets() *Application {
r.Secrets = &map[string]Secret{}
return r
}
// SetExecutor sets the executor
func (r *Application) SetExecutor(executor string) *Application {
r.Executor = &executor
@ -571,6 +604,23 @@ func (r *Application) EmptyUnreachableStrategy() *Application {
return r
}
// SetResidency sets behavior for resident applications, an application is resident when
// it has local persistent volumes set
func (r *Application) SetResidency(whenLost TaskLostBehaviorType) *Application {
r.Residency = &Residency{
TaskLostBehavior: whenLost,
}
return r
}
// EmptyResidency explicitly empties the residency -- use this if
// you need to empty the residency of an application that already has
// the residency set (setting it to nil will keep the current value).
func (r *Application) EmptyResidency() *Application {
r.Residency = &Residency{}
return r
}
// String returns the json representation of this application
func (r *Application) String() string {
s, err := json.MarshalIndent(r, "", " ")
@ -639,7 +689,7 @@ func (r *marathonClient) ApplicationVersions(name string) (*ApplicationVersions,
// name: the id used to identify the application
// version: the version (normally a timestamp) you wish to change to
func (r *marathonClient) SetApplicationVersion(name string, version *ApplicationVersion) (*DeploymentID, error) {
path := fmt.Sprintf(buildPath(name))
path := buildPath(name)
deploymentID := new(DeploymentID)
if err := r.apiPut(path, version, deploymentID); err != nil {
return nil, err

View file

@ -0,0 +1,106 @@
/*
Copyright 2017 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package marathon
import (
"encoding/json"
"fmt"
)
// Alias aliases the Application struct so that it will be marshaled/unmarshaled automatically
type Alias Application
// TmpEnvSecret holds the secret values deserialized from the environment variables field
type TmpEnvSecret struct {
Secret string `json:"secret,omitempty"`
}
// TmpSecret holds the deserialized secrets field in a Marathon application configuration
type TmpSecret struct {
Source string `json:"source,omitempty"`
}
// UnmarshalJSON unmarshals the given Application JSON as expected except for environment variables and secrets.
// Environment varialbes are stored in the Env field. Secrets, including the environment variable part,
// are stored in the Secrets field.
func (app *Application) UnmarshalJSON(b []byte) error {
aux := &struct {
*Alias
Env map[string]interface{} `json:"env"`
Secrets map[string]TmpSecret `json:"secrets"`
}{
Alias: (*Alias)(app),
}
if err := json.Unmarshal(b, aux); err != nil {
return fmt.Errorf("malformed application definition %v", err)
}
env := &map[string]string{}
secrets := &map[string]Secret{}
for envName, genericEnvValue := range aux.Env {
switch envValOrSecret := genericEnvValue.(type) {
case string:
(*env)[envName] = envValOrSecret
case map[string]interface{}:
for secret, secretStore := range envValOrSecret {
if secStore, ok := secretStore.(string); ok && secret == "secret" {
(*secrets)[secStore] = Secret{EnvVar: envName}
break
}
return fmt.Errorf("unexpected secret field %v or value type %T", secret, envValOrSecret[secret])
}
default:
return fmt.Errorf("unexpected environment variable type %T", envValOrSecret)
}
}
app.Env = env
for k, v := range aux.Secrets {
tmp := (*secrets)[k]
tmp.Source = v.Source
(*secrets)[k] = tmp
}
app.Secrets = secrets
return nil
}
// MarshalJSON marshals the given Application as expected except for environment variables and secrets,
// which are marshaled from specialized structs. The environment variable piece of the secrets and other
// normal environment variables are combined and marshaled to the env field. The secrets and the related
// source are marshaled into the secrets field.
func (app *Application) MarshalJSON() ([]byte, error) {
env := make(map[string]interface{})
secrets := make(map[string]TmpSecret)
if app.Env != nil {
for k, v := range *app.Env {
env[string(k)] = string(v)
}
}
if app.Secrets != nil {
for k, v := range *app.Secrets {
env[v.EnvVar] = TmpEnvSecret{Secret: k}
secrets[k] = TmpSecret{v.Source}
}
}
aux := &struct {
*Alias
Env map[string]interface{} `json:"env,omitempty"`
Secrets map[string]TmpSecret `json:"secrets,omitempty"`
}{Alias: (*Alias)(app), Env: env, Secrets: secrets}
return json.Marshal(aux)
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -24,6 +24,7 @@ import (
"io"
"io/ioutil"
"log"
"net"
"net/http"
"net/url"
"regexp"
@ -154,6 +155,24 @@ var (
ErrMarathonDown = errors.New("all the Marathon hosts are presently down")
// ErrTimeoutError is thrown when the operation has timed out
ErrTimeoutError = errors.New("the operation has timed out")
// Default HTTP client used for SSE subscription requests
// It is invalid to set client.Timeout because it includes time to read response so
// set dial, tls handshake and response header timeouts instead
defaultHTTPSSEClient = &http.Client{
Transport: &http.Transport{
Dial: (&net.Dialer{
Timeout: 5 * time.Second,
}).Dial,
ResponseHeaderTimeout: 10 * time.Second,
TLSHandshakeTimeout: 5 * time.Second,
},
}
// Default HTTP client used for non SSE requests
defaultHTTPClient = &http.Client{
Timeout: 10 * time.Second,
}
)
// EventsChannelContext holds contextual data for an EventsChannel.
@ -177,8 +196,8 @@ type marathonClient struct {
hosts *cluster
// a map of service you wish to listen to
listeners map[EventsChannel]EventsChannelContext
// a custom logger for debug log messages
debugLog *log.Logger
// a custom log function for debug messages
debugLog func(format string, v ...interface{})
// the marathon HTTP client to ensure consistency in requests
client *httpClient
}
@ -196,9 +215,18 @@ type newRequestError struct {
// NewClient creates a new marathon client
// config: the configuration to use
func NewClient(config Config) (Marathon, error) {
// step: if no http client, set to default
// step: if the SSE HTTP client is missing, prefer a configured regular
// client, and otherwise use the default SSE HTTP client.
if config.HTTPSSEClient == nil {
config.HTTPSSEClient = defaultHTTPSSEClient
if config.HTTPClient != nil {
config.HTTPSSEClient = config.HTTPClient
}
}
// step: if a regular HTTP client is missing, use the default one.
if config.HTTPClient == nil {
config.HTTPClient = http.DefaultClient
config.HTTPClient = defaultHTTPClient
}
// step: if no polling wait time is set, default to 500 milliseconds.
@ -215,16 +243,19 @@ func NewClient(config Config) (Marathon, error) {
return nil, err
}
debugLogOutput := config.LogOutput
if debugLogOutput == nil {
debugLogOutput = ioutil.Discard
debugLog := func(string, ...interface{}) {}
if config.LogOutput != nil {
logger := log.New(config.LogOutput, "", 0)
debugLog = func(format string, v ...interface{}) {
logger.Printf(format, v...)
}
}
return &marathonClient{
config: config,
listeners: make(map[EventsChannel]EventsChannelContext),
hosts: hosts,
debugLog: log.New(debugLogOutput, "", 0),
debugLog: debugLog,
client: client,
}, nil
}
@ -280,7 +311,7 @@ func (r *marathonClient) apiCall(method, path string, body, result interface{})
if err != nil {
r.hosts.markDown(member)
// step: attempt the request on another member
r.debugLog.Printf("apiCall(): request failed on host: %s, error: %s, trying another\n", member, err)
r.debugLog("apiCall(): request failed on host: %s, error: %s, trying another", member, err)
continue
}
defer response.Body.Close()
@ -292,9 +323,9 @@ func (r *marathonClient) apiCall(method, path string, body, result interface{})
}
if len(requestBody) > 0 {
r.debugLog.Printf("apiCall(): %v %v %s returned %v %s\n", request.Method, request.URL.String(), requestBody, response.Status, oneLogLine(respBody))
r.debugLog("apiCall(): %v %v %s returned %v %s", request.Method, request.URL.String(), requestBody, response.Status, oneLogLine(respBody))
} else {
r.debugLog.Printf("apiCall(): %v %v returned %v %s\n", request.Method, request.URL.String(), response.Status, oneLogLine(respBody))
r.debugLog("apiCall(): %v %v returned %v %s", request.Method, request.URL.String(), response.Status, oneLogLine(respBody))
}
// step: check for a successfull response
@ -311,7 +342,7 @@ func (r *marathonClient) apiCall(method, path string, body, result interface{})
if response.StatusCode >= 500 && response.StatusCode <= 599 {
// step: mark the host as down
r.hosts.markDown(member)
r.debugLog.Printf("apiCall(): request failed, host: %s, status: %d, trying another\n", member, response.StatusCode)
r.debugLog("apiCall(): request failed, host: %s, status: %d, trying another", member, response.StatusCode)
continue
}
@ -329,16 +360,28 @@ func (r *marathonClient) buildAPIRequest(method, path string, reader io.Reader)
}
// Build the HTTP request to Marathon
request, err = r.client.buildMarathonRequest(method, member, path, reader)
request, err = r.client.buildMarathonJSONRequest(method, member, path, reader)
if err != nil {
return nil, member, newRequestError{err}
}
return request, member, nil
}
// buildMarathonJSONRequest is like buildMarathonRequest but sets the
// Content-Type and Accept headers to application/json.
func (rc *httpClient) buildMarathonJSONRequest(method, member, path string, reader io.Reader) (request *http.Request, err error) {
req, err := rc.buildMarathonRequest(method, member, path, reader)
if err == nil {
req.Header.Add("Content-Type", "application/json")
req.Header.Add("Accept", "application/json")
}
return req, err
}
// buildMarathonRequest creates a new HTTP request and configures it according to the *httpClient configuration.
// The path must not contain a leading "/", otherwise buildMarathonRequest will panic.
func (rc *httpClient) buildMarathonRequest(method string, member string, path string, reader io.Reader) (request *http.Request, err error) {
func (rc *httpClient) buildMarathonRequest(method, member, path string, reader io.Reader) (request *http.Request, err error) {
if strings.HasPrefix(path, "/") {
panic(fmt.Sprintf("Path '%s' must not start with a leading slash", path))
}
@ -361,9 +404,6 @@ func (rc *httpClient) buildMarathonRequest(method string, member string, path st
request.Header.Add("Authorization", "token="+rc.config.DCOSToken)
}
request.Header.Add("Content-Type", "application/json")
request.Header.Add("Accept", "application/json")
return request, nil
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -39,6 +39,9 @@ type cluster struct {
members []*member
// the marathon HTTP client to ensure consistency in requests
client *httpClient
// healthCheckInterval is the interval by which we probe down nodes for
// availability again.
healthCheckInterval time.Duration
}
// member represents an individual endpoint
@ -94,8 +97,9 @@ func newCluster(client *httpClient, marathonURL string, isDCOS bool) (*cluster,
}
return &cluster{
client: client,
members: members,
client: client,
members: members,
healthCheckInterval: 5 * time.Second,
}, nil
}
@ -130,20 +134,21 @@ func (c *cluster) markDown(endpoint string) {
// healthCheckNode performs a health check on the node and when active updates the status
func (c *cluster) healthCheckNode(node *member) {
// step: wait for the node to become active ... we are assuming a /ping is enough here
for {
ticker := time.NewTicker(c.healthCheckInterval)
defer ticker.Stop()
for range ticker.C {
req, err := c.client.buildMarathonRequest("GET", node.endpoint, "ping", nil)
if err == nil {
res, err := c.client.Do(req)
if err == nil && res.StatusCode == 200 {
// step: mark the node as active again
c.Lock()
node.status = memberStatusUp
c.Unlock()
break
}
}
<-time.After(time.Duration(5 * time.Second))
}
// step: mark the node as active again
c.Lock()
defer c.Unlock()
node.status = memberStatusUp
}
// activeMembers returns a list of active members

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -50,8 +50,10 @@ type Config struct {
DCOSToken string
// LogOutput the output for debug log messages
LogOutput io.Writer
// HTTPClient is the http client
// HTTPClient is the HTTP client
HTTPClient *http.Client
// HTTPSSEClient is the HTTP client used for SSE subscriptions, can't have client.Timeout set
HTTPSSEClient *http.Client
// wait time (in milliseconds) between repetitive requests to the API during polling
PollingWaitTime time.Duration
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -46,10 +46,71 @@ type Parameters struct {
// Volume is the docker volume details associated to the container
type Volume struct {
ContainerPath string `json:"containerPath,omitempty"`
HostPath string `json:"hostPath,omitempty"`
External *ExternalVolume `json:"external,omitempty"`
Mode string `json:"mode,omitempty"`
ContainerPath string `json:"containerPath,omitempty"`
HostPath string `json:"hostPath,omitempty"`
External *ExternalVolume `json:"external,omitempty"`
Mode string `json:"mode,omitempty"`
Persistent *PersistentVolume `json:"persistent,omitempty"`
}
type PersistentVolumeType string
const (
PersistentVolumeTypeRoot PersistentVolumeType = "root"
PersistentVolumeTypePath PersistentVolumeType = "path"
PersistentVolumeTypeMount PersistentVolumeType = "mount"
)
// PersistentVolume declares a Volume to be Persistent, and sets
// the size (in MiB) and optional type, max size (MiB) and constraints for the Volume.
type PersistentVolume struct {
Type PersistentVolumeType `json:"type,omitempty"`
Size int `json:"size"`
MaxSize int `json:"maxSize,omitempty"`
Constraints *[][]string `json:"constraints,omitempty"`
}
// SetType sets the type of mesos disk resource to use
// type: PersistentVolumeType enum
func (p *PersistentVolume) SetType(tp PersistentVolumeType) *PersistentVolume {
p.Type = tp
return p
}
// SetSize sets size of the persistent volume
// size: size in MiB
func (p *PersistentVolume) SetSize(size int) *PersistentVolume {
p.Size = size
return p
}
// SetMaxSize sets maximum size of an exclusive mount-disk resource to consider;
// does not apply to root or path disk resource types
// maxSize: size in MiB
func (p *PersistentVolume) SetMaxSize(maxSize int) *PersistentVolume {
p.MaxSize = maxSize
return p
}
// AddConstraint adds a new constraint
// constraints: the constraint definition, one constraint per array element
func (p *PersistentVolume) AddConstraint(constraints ...string) *PersistentVolume {
if p.Constraints == nil {
p.EmptyConstraints()
}
c := *p.Constraints
c = append(c, constraints)
p.Constraints = &c
return p
}
// EmptyConstraints explicitly empties constraints -- use this if you need to empty
// constraints of an application that already has constraints set (setting constraints to nil will
// keep the current value)
func (p *PersistentVolume) EmptyConstraints() *PersistentVolume {
p.Constraints = &[][]string{}
return p
}
// ExternalVolume is an external volume definition
@ -98,6 +159,19 @@ func (container *Container) EmptyVolumes() *Container {
return container
}
// SetPersistentVolume defines persistent properties for volume
func (v *Volume) SetPersistentVolume() *PersistentVolume {
ev := &PersistentVolume{}
v.Persistent = ev
return ev
}
// EmptyPersistentVolume empties the persistent volume definition
func (v *Volume) EmptyPersistentVolume() *Volume {
v.Persistent = &PersistentVolume{}
return v
}
// SetExternalVolume define external elements for a volume
// name: the name of the volume
// provider: the provider of the volume (e.g. dvdi)

View file

@ -1,5 +1,5 @@
/*
Copyright 2015 Rohith All rights reserved.
Copyright 2015 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -136,7 +136,7 @@ func (r *marathonClient) GroupBy(name string, opts *GetGroupOpts) (*Group, error
// name: the identifier for the group
func (r *marathonClient) HasGroup(name string) (bool, error) {
path := fmt.Sprintf("%s/%s", marathonAPIGroups, trimRootPath(name))
err := r.apiCall("GET", path, "", nil)
err := r.apiGet(path, "", nil)
if err != nil {
if apiErr, ok := err.(*APIError); ok && apiErr.ErrCode == ErrCodeNotFound {
return false, nil

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -31,37 +31,37 @@ type HealthCheck struct {
}
// SetCommand sets the given command on the health check.
func (h HealthCheck) SetCommand(c Command) HealthCheck {
func (h *HealthCheck) SetCommand(c Command) *HealthCheck {
h.Command = &c
return h
}
// SetPortIndex sets the given port index on the health check.
func (h HealthCheck) SetPortIndex(i int) HealthCheck {
func (h *HealthCheck) SetPortIndex(i int) *HealthCheck {
h.PortIndex = &i
return h
}
// SetPort sets the given port on the health check.
func (h HealthCheck) SetPort(i int) HealthCheck {
func (h *HealthCheck) SetPort(i int) *HealthCheck {
h.Port = &i
return h
}
// SetPath sets the given path on the health check.
func (h HealthCheck) SetPath(p string) HealthCheck {
func (h *HealthCheck) SetPath(p string) *HealthCheck {
h.Path = &p
return h
}
// SetMaxConsecutiveFailures sets the maximum consecutive failures on the health check.
func (h HealthCheck) SetMaxConsecutiveFailures(i int) HealthCheck {
func (h *HealthCheck) SetMaxConsecutiveFailures(i int) *HealthCheck {
h.MaxConsecutiveFailures = &i
return h
}
// SetIgnoreHTTP1xx sets ignore http 1xx on the health check.
func (h HealthCheck) SetIgnoreHTTP1xx(ignore bool) HealthCheck {
func (h *HealthCheck) SetIgnoreHTTP1xx(ignore bool) *HealthCheck {
h.IgnoreHTTP1xx = &ignore
return h
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.

View file

@ -1,4 +1,5 @@
/*
Copyright 2015 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -20,6 +21,7 @@ type LastTaskFailure struct {
AppID string `json:"appId,omitempty"`
Host string `json:"host,omitempty"`
Message string `json:"message,omitempty"`
SlaveID string `json:"slaveId,omitempty"`
State string `json:"state,omitempty"`
TaskID string `json:"taskId,omitempty"`
Timestamp string `json:"timestamp,omitempty"`

View file

@ -1,5 +1,5 @@
/*
Copyright 2016 Rohith All rights reserved.
Copyright 2016 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -27,15 +27,39 @@ type PortDefinition struct {
}
// SetPort sets the given port for the PortDefinition
func (p PortDefinition) SetPort(port int) PortDefinition {
func (p *PortDefinition) SetPort(port int) *PortDefinition {
if p.Port == nil {
p.EmptyPort()
}
p.Port = &port
return p
}
// EmptyPort sets the port to 0 for the PortDefinition
func (p *PortDefinition) EmptyPort() *PortDefinition {
port := 0
p.Port = &port
return p
}
// SetProtocol sets the protocol for the PortDefinition
// protocol: the protocol as a string
func (p *PortDefinition) SetProtocol(protocol string) *PortDefinition {
p.Protocol = protocol
return p
}
// SetName sets the name for the PortDefinition
// name: the name of the PortDefinition
func (p *PortDefinition) SetName(name string) *PortDefinition {
p.Name = name
return p
}
// AddLabel adds a label to the PortDefinition
// name: the name of the label
// value: value for this label
func (p PortDefinition) AddLabel(name, value string) PortDefinition {
func (p *PortDefinition) AddLabel(name, value string) *PortDefinition {
if p.Labels == nil {
p.EmptyLabels()
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2016 Rohith All rights reserved.
Copyright 2016 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -52,9 +52,5 @@ func (r *marathonClient) Queue() (*Queue, error) {
// appID: the ID of the application
func (r *marathonClient) DeleteQueueDelay(appID string) error {
path := fmt.Sprintf("%s/%s/delay", marathonAPIQueue, trimRootPath(appID))
err := r.apiDelete(path, nil, nil)
if err != nil {
return err
}
return nil
return r.apiDelete(path, nil, nil)
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2017 Rohith All rights reserved.
Copyright 2017 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.

48
vendor/github.com/gambol99/go-marathon/residency.go generated vendored Normal file
View file

@ -0,0 +1,48 @@
/*
Copyright 2017 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package marathon
import "time"
// TaskLostBehaviorType sets action taken when the resident task is lost
type TaskLostBehaviorType string
const (
// TaskLostBehaviorTypeWaitForever indicates to not take any action when the resident task is lost
TaskLostBehaviorTypeWaitForever TaskLostBehaviorType = "WAIT_FOREVER"
// TaskLostBehaviorTypeWaitForever indicates to try relaunching the lost resident task on
// another node after the relaunch escalation timeout has elapsed
TaskLostBehaviorTypeRelaunchAfterTimeout TaskLostBehaviorType = "RELAUNCH_AFTER_TIMEOUT"
)
// Residency defines how terminal states of tasks with local persistent volumes are handled
type Residency struct {
TaskLostBehavior TaskLostBehaviorType `json:"taskLostBehavior,omitempty"`
RelaunchEscalationTimeoutSeconds int `json:"relaunchEscalationTimeoutSeconds,omitempty"`
}
// SetTaskLostBehavior sets the residency behavior
func (r *Residency) SetTaskLostBehavior(behavior TaskLostBehaviorType) *Residency {
r.TaskLostBehavior = behavior
return r
}
// SetRelaunchEscalationTimeout sets the residency relaunch escalation timeout with seconds precision
func (r *Residency) SetRelaunchEscalationTimeout(timeout time.Duration) *Residency {
r.RelaunchEscalationTimeoutSeconds = int(timeout.Seconds())
return r
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -103,8 +103,7 @@ func (r *marathonClient) registerSubscription() error {
case EventsTransportCallback:
return r.registerCallbackSubscription()
case EventsTransportSSE:
r.registerSSESubscription()
return nil
return r.registerSSESubscription()
default:
return fmt.Errorf("the events transport: %d is not supported", r.config.EventsTransport)
}
@ -167,27 +166,34 @@ func (r *marathonClient) registerCallbackSubscription() error {
// connect to the SSE stream and to process the received events. To establish
// the connection it tries the active cluster members until no more member is
// active. When this happens it will retry to get a connection every 5 seconds.
func (r *marathonClient) registerSSESubscription() {
func (r *marathonClient) registerSSESubscription() error {
if r.subscribedToSSE {
return
return nil
}
if r.config.HTTPSSEClient.Timeout != 0 {
return fmt.Errorf(
"global timeout must not be set for SSE connections (found %s) -- remove global timeout from HTTP client or provide separate SSE HTTP client without global timeout",
r.config.HTTPSSEClient.Timeout,
)
}
go func() {
for {
stream, err := r.connectToSSE()
if err != nil {
r.debugLog.Printf("Error connecting SSE subscription: %s", err)
r.debugLog("Error connecting SSE subscription: %s", err)
<-time.After(5 * time.Second)
continue
}
err = r.listenToSSE(stream)
stream.Close()
r.debugLog.Printf("Error on SSE subscription: %s", err)
r.debugLog("Error on SSE subscription: %s", err)
}
}()
r.subscribedToSSE = true
return nil
}
// connectToSSE tries to establish an *eventsource.Stream to any of the Marathon cluster members, marking the
@ -209,15 +215,15 @@ func (r *marathonClient) connectToSSE() (*eventsource.Stream, error) {
// its underlying fields for performance reasons. See note that at least the Transport
// should be reused here: https://golang.org/pkg/net/http/#Client
httpClient := &http.Client{
Transport: r.config.HTTPClient.Transport,
CheckRedirect: r.config.HTTPClient.CheckRedirect,
Jar: r.config.HTTPClient.Jar,
Timeout: r.config.HTTPClient.Timeout,
Transport: r.config.HTTPSSEClient.Transport,
CheckRedirect: r.config.HTTPSSEClient.CheckRedirect,
Jar: r.config.HTTPSSEClient.Jar,
Timeout: r.config.HTTPSSEClient.Timeout,
}
stream, err := eventsource.SubscribeWith("", httpClient, request)
if err != nil {
r.debugLog.Printf("Error subscribing to Marathon event stream: %s", err)
r.debugLog("Error subscribing to Marathon event stream: %s", err)
r.hosts.markDown(member)
continue
}
@ -231,7 +237,7 @@ func (r *marathonClient) listenToSSE(stream *eventsource.Stream) error {
select {
case ev := <-stream.Events:
if err := r.handleEvent(ev.Data()); err != nil {
r.debugLog.Printf("listenToSSE(): failed to handle event: %v", err)
r.debugLog("listenToSSE(): failed to handle event: %v", err)
}
case err := <-stream.Errors:
return err
@ -319,12 +325,12 @@ func (r *marathonClient) handleCallbackEvent(writer http.ResponseWriter, request
body, err := ioutil.ReadAll(request.Body)
if err != nil {
// TODO should this return a 500?
r.debugLog.Printf("handleCallbackEvent(): failed to read request body, error: %s\n", err)
r.debugLog("handleCallbackEvent(): failed to read request body, error: %s", err)
return
}
if err := r.handleEvent(string(body[:])); err != nil {
// TODO should this return a 500?
r.debugLog.Printf("handleCallbackEvent(): failed to handle event: %v\n", err)
r.debugLog("handleCallbackEvent(): failed to handle event: %v", err)
}
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -217,7 +217,7 @@ func (r *Task) allHealthChecksAlive() bool {
}
// step: check the health results then
for _, check := range r.HealthCheckResults {
if check.Alive == false {
if !check.Alive {
return false
}
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2017 Rohith All rights reserved.
Copyright 2017 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -65,13 +65,13 @@ func (us *UnreachableStrategy) MarshalJSON() ([]byte, error) {
}
// SetInactiveAfterSeconds sets the period after which instance will be marked as inactive.
func (us UnreachableStrategy) SetInactiveAfterSeconds(cap float64) UnreachableStrategy {
func (us *UnreachableStrategy) SetInactiveAfterSeconds(cap float64) *UnreachableStrategy {
us.InactiveAfterSeconds = &cap
return us
}
// SetExpungeAfterSeconds sets the period after which instance will be expunged.
func (us UnreachableStrategy) SetExpungeAfterSeconds(cap float64) UnreachableStrategy {
func (us *UnreachableStrategy) SetExpungeAfterSeconds(cap float64) *UnreachableStrategy {
us.ExpungeAfterSeconds = &cap
return us
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
@ -23,13 +23,13 @@ type UpgradeStrategy struct {
}
// SetMinimumHealthCapacity sets the minimum health capacity.
func (us UpgradeStrategy) SetMinimumHealthCapacity(cap float64) UpgradeStrategy {
func (us *UpgradeStrategy) SetMinimumHealthCapacity(cap float64) *UpgradeStrategy {
us.MinimumHealthCapacity = &cap
return us
}
// SetMaximumOverCapacity sets the maximum over capacity.
func (us UpgradeStrategy) SetMaximumOverCapacity(cap float64) UpgradeStrategy {
func (us *UpgradeStrategy) SetMaximumOverCapacity(cap float64) *UpgradeStrategy {
us.MaximumOverCapacity = &cap
return us
}

View file

@ -1,5 +1,5 @@
/*
Copyright 2014 Rohith All rights reserved.
Copyright 2014 The go-marathon Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.