-
Notifications
You must be signed in to change notification settings - Fork 127
/
client.go
498 lines (427 loc) · 15.2 KB
/
client.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
/*
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.
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 (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"io/ioutil"
"log"
"net"
"net/http"
"net/url"
"regexp"
"strings"
"sync"
"time"
)
// Marathon is the interface to the marathon API
type Marathon interface {
// -- GENERIC API ACCESS ---
ApiPost(path string, post, result interface{}) error
// -- APPLICATIONS ---
// get a listing of the application ids
ListApplications(url.Values) ([]string, error)
// a list of application versions
ApplicationVersions(name string) (*ApplicationVersions, error)
// check a application version exists
HasApplicationVersion(name, version string) (bool, error)
// change an application to a different version
SetApplicationVersion(name string, version *ApplicationVersion) (*DeploymentID, error)
// check if an application is ok
ApplicationOK(name string) (bool, error)
// create an application in marathon
CreateApplication(application *Application) (*Application, error)
// delete an application
DeleteApplication(name string, force bool) (*DeploymentID, error)
// update an application in marathon
UpdateApplication(application *Application, force bool) (*DeploymentID, error)
// a list of deployments on a application
ApplicationDeployments(name string) ([]*DeploymentID, error)
// scale a application
ScaleApplicationInstances(name string, instances int, force bool) (*DeploymentID, error)
// restart an application
RestartApplication(name string, force bool) (*DeploymentID, error)
// get a list of applications from marathon
Applications(url.Values) (*Applications, error)
// get an application by name
Application(name string) (*Application, error)
// get an application by options
ApplicationBy(name string, opts *GetAppOpts) (*Application, error)
// get an application by name and version
ApplicationByVersion(name, version string) (*Application, error)
// wait of application
WaitOnApplication(name string, timeout time.Duration) error
// -- PODS ---
// whether this version of Marathon supports pods
SupportsPods() (bool, error)
// get pod status
PodStatus(name string) (*PodStatus, error)
// get all pod statuses
PodStatuses() ([]*PodStatus, error)
// get pod
Pod(name string) (*Pod, error)
// get all pods
Pods() ([]Pod, error)
// create pod
CreatePod(pod *Pod) (*Pod, error)
// update pod
UpdatePod(pod *Pod, force bool) (*Pod, error)
// delete pod
DeletePod(name string, force bool) (*DeploymentID, error)
// wait on pod to be deployed
WaitOnPod(name string, timeout time.Duration) error
// check if a pod is running
PodIsRunning(name string) bool
// get versions of a pod
PodVersions(name string) ([]string, error)
// get pod by version
PodByVersion(name, version string) (*Pod, error)
// delete instances of a pod
DeletePodInstances(name string, instances []string) ([]*PodInstance, error)
// delete pod instance
DeletePodInstance(name, instance string) (*PodInstance, error)
// -- TASKS ---
// get a list of tasks for a specific application
Tasks(application string) (*Tasks, error)
// get a list of all tasks
AllTasks(opts *AllTasksOpts) (*Tasks, error)
// get the endpoints for a service on a application
TaskEndpoints(name string, port int, healthCheck bool) ([]string, error)
// kill all the tasks for any application
KillApplicationTasks(applicationID string, opts *KillApplicationTasksOpts) (*Tasks, error)
// kill a single task
KillTask(taskID string, opts *KillTaskOpts) (*Task, error)
// kill the given array of tasks
KillTasks(taskIDs []string, opts *KillTaskOpts) error
// --- GROUPS ---
// list all the groups in the system
Groups() (*Groups, error)
// retrieve a specific group from marathon
Group(name string) (*Group, error)
// list all groups in marathon by options
GroupsBy(opts *GetGroupOpts) (*Groups, error)
// retrieve a specific group from marathon by options
GroupBy(name string, opts *GetGroupOpts) (*Group, error)
// create a group deployment
CreateGroup(group *Group) error
// delete a group
DeleteGroup(name string, force bool) (*DeploymentID, error)
// update a groups
UpdateGroup(id string, group *Group, force bool) (*DeploymentID, error)
// check if a group exists
HasGroup(name string) (bool, error)
// wait for an group to be deployed
WaitOnGroup(name string, timeout time.Duration) error
// --- DEPLOYMENTS ---
// get a list of the deployments
Deployments() ([]*Deployment, error)
// delete a deployment
DeleteDeployment(id string, force bool) (*DeploymentID, error)
// check to see if a deployment exists
HasDeployment(id string) (bool, error)
// wait of a deployment to finish
WaitOnDeployment(id string, timeout time.Duration) error
// --- SUBSCRIPTIONS ---
// a list of current subscriptions
Subscriptions() (*Subscriptions, error)
// add a events listener
AddEventsListener(filter int) (EventsChannel, error)
// remove a events listener
RemoveEventsListener(channel EventsChannel)
// Subscribe a callback URL
Subscribe(string) error
// Unsubscribe a callback URL
Unsubscribe(string) error
// --- QUEUE ---
// get marathon launch queue
Queue() (*Queue, error)
// resets task launch delay of the specific application
DeleteQueueDelay(appID string) error
// --- MISC ---
// get the marathon url
GetMarathonURL() string
// ping the marathon
Ping() (bool, error)
// grab the marathon server info
Info() (*Info, error)
// retrieve the leader info
Leader() (string, error)
// cause the current leader to abdicate
AbdicateLeader() (string, error)
}
var (
// ErrMarathonDown is thrown when all the marathon endpoints are down
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.
type EventsChannelContext struct {
filter int
done chan struct{}
completion *sync.WaitGroup
}
type marathonClient struct {
sync.RWMutex
// the configuration for the client
config Config
// the flag used to prevent multiple SSE subscriptions
subscribedToSSE bool
// the ip address of the client
ipAddress string
// the http server
eventsHTTP *http.Server
// the marathon hosts
hosts *cluster
// a map of service you wish to listen to
listeners map[EventsChannel]EventsChannelContext
// a custom log function for debug messages
debugLog func(format string, v ...interface{})
// the marathon HTTP client to ensure consistency in requests
client *httpClient
}
type httpClient struct {
// the configuration for the marathon HTTP client
config Config
}
// newRequestError signals that creating a new http.Request failed
type newRequestError struct {
error
}
// NewClient creates a new marathon client
// config: the configuration to use
func NewClient(config Config) (Marathon, error) {
// 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 = defaultHTTPClient
}
// step: if no polling wait time is set, default to 500 milliseconds.
if config.PollingWaitTime == 0 {
config.PollingWaitTime = defaultPollingWaitTime
}
// step: setup shared client
client := &httpClient{config: config}
// step: create a new cluster
hosts, err := newCluster(client, config.URL, config.DCOSToken != "")
if err != nil {
return nil, err
}
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: debugLog,
client: client,
}, nil
}
// GetMarathonURL retrieves the marathon url
func (r *marathonClient) GetMarathonURL() string {
return r.config.URL
}
// Ping pings the current marathon endpoint (note, this is not a ICMP ping, but a rest api call)
func (r *marathonClient) Ping() (bool, error) {
if err := r.apiGet(marathonAPIPing, nil, nil); err != nil {
return false, err
}
return true, nil
}
func (r *marathonClient) apiHead(path string, result interface{}) error {
return r.apiCall("HEAD", path, nil, result)
}
func (r *marathonClient) apiGet(path string, post, result interface{}) error {
return r.apiCall("GET", path, post, result)
}
func (r *marathonClient) apiPut(path string, post, result interface{}) error {
return r.apiCall("PUT", path, post, result)
}
func (r *marathonClient) ApiPost(path string, post, result interface{}) error {
return r.apiCall("POST", path, post, result)
}
func (r *marathonClient) apiDelete(path string, post, result interface{}) error {
return r.apiCall("DELETE", path, post, result)
}
func (r *marathonClient) apiCall(method, path string, body, result interface{}) error {
const deploymentHeader = "Marathon-Deployment-Id"
for {
// step: marshall the request to json
var requestBody []byte
var err error
if body != nil {
if requestBody, err = json.Marshal(body); err != nil {
return err
}
}
// step: create the API request
request, member, err := r.buildAPIRequest(method, path, bytes.NewReader(requestBody))
if err != nil {
return err
}
// step: perform the API request
response, err := r.client.Do(request)
if err != nil {
r.hosts.markDown(member)
// step: attempt the request on another member
r.debugLog("apiCall(): request failed on host: %s, error: %s, trying another", member, err)
continue
}
defer response.Body.Close()
// step: read the response body
respBody, err := ioutil.ReadAll(response.Body)
if err != nil {
return err
}
if len(requestBody) > 0 {
r.debugLog("apiCall(): %v %v %s returned %v %s", request.Method, request.URL.String(), requestBody, response.Status, oneLogLine(respBody))
} else {
r.debugLog("apiCall(): %v %v returned %v %s", request.Method, request.URL.String(), response.Status, oneLogLine(respBody))
}
// step: check for a successful response
if response.StatusCode >= 200 && response.StatusCode <= 299 {
if result != nil {
// If we have a deployment ID header and no response body, give them that
// This specifically handles the use case of a DELETE on an app/pod
// We need a way to retrieve the deployment ID
deploymentID := response.Header.Get(deploymentHeader)
if len(respBody) == 0 && deploymentID != "" {
d := DeploymentID{
DeploymentID: deploymentID,
}
if deployID, ok := result.(*DeploymentID); ok {
*deployID = d
}
} else {
if err := json.Unmarshal(respBody, result); err != nil {
return fmt.Errorf("failed to unmarshal response from Marathon: %s", err)
}
}
}
return nil
}
// step: if the member node returns a >= 500 && <= 599 we should try another node?
if response.StatusCode >= 500 && response.StatusCode <= 599 {
// step: mark the host as down
r.hosts.markDown(member)
r.debugLog("apiCall(): request failed, host: %s, status: %d, trying another", member, response.StatusCode)
continue
}
return NewAPIError(response.StatusCode, respBody)
}
}
// wait waits until the provided function returns true (or times out)
func (r *marathonClient) wait(name string, timeout time.Duration, fn func(string) bool) error {
timer := time.NewTimer(timeout)
defer timer.Stop()
ticker := time.NewTicker(r.config.PollingWaitTime)
defer ticker.Stop()
for {
if fn(name) {
return nil
}
select {
case <-timer.C:
return ErrTimeoutError
case <-ticker.C:
continue
}
}
}
// buildAPIRequest creates a default API request.
// It fails when there is no available member in the cluster anymore or when the request can not be built.
func (r *marathonClient) buildAPIRequest(method, path string, reader io.Reader) (request *http.Request, member string, err error) {
// Grab a member from the cluster
member, err = r.hosts.getMember()
if err != nil {
return nil, "", ErrMarathonDown
}
// Build the HTTP request to Marathon
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, 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))
}
// Create the endpoint URL
url := fmt.Sprintf("%s/%s", member, path)
// Instantiate an HTTP request
request, err = http.NewRequest(method, url, reader)
if err != nil {
return nil, err
}
// Add any basic auth and the content headers
if rc.config.HTTPBasicAuthUser != "" && rc.config.HTTPBasicPassword != "" {
request.SetBasicAuth(rc.config.HTTPBasicAuthUser, rc.config.HTTPBasicPassword)
}
if rc.config.DCOSToken != "" {
request.Header.Add("Authorization", "token="+rc.config.DCOSToken)
}
return request, nil
}
func (rc *httpClient) Do(request *http.Request) (response *http.Response, err error) {
return rc.config.HTTPClient.Do(request)
}
var oneLogLineRegex = regexp.MustCompile(`(?m)^\s*`)
// oneLogLine removes indentation at the beginning of each line and
// escapes new line characters.
func oneLogLine(in []byte) []byte {
return bytes.Replace(oneLogLineRegex.ReplaceAll(in, nil), []byte("\n"), []byte("\\n "), -1)
}