forked from gambol99/go-marathon
-
Notifications
You must be signed in to change notification settings - Fork 0
/
cluster.go
177 lines (151 loc) · 4.4 KB
/
cluster.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
/*
Copyright 2014 Rohith 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 (
"fmt"
"net/http"
"net/url"
"strings"
"sync"
"time"
)
const (
memberStatusUp = 0
memberStatusDown = 1
)
// the status of a member node
type memberStatus int
// cluster is a collection of marathon nodes
type cluster struct {
sync.RWMutex
// a collection of nodes
members []*member
// the http client
client *http.Client
}
// member represents an individual endpoint
type member struct {
// the name / ip address of the host
endpoint string
// the status of the host
status memberStatus
}
// newCluster returns a new marathon cluster
func newCluster(client *http.Client, marathonURL string) (*cluster, error) {
// step: extract and basic validate the endpoints
var members []*member
var defaultProto string
for _, endpoint := range strings.Split(marathonURL, ",") {
// step: check for nothing
if endpoint == "" {
return nil, newInvalidEndpointError("endpoint is blank")
}
// step: prepend scheme if missing on (non-initial) endpoint.
if !strings.HasPrefix(endpoint, "http://") && !strings.HasPrefix(endpoint, "https://") {
if defaultProto == "" {
return nil, newInvalidEndpointError("missing scheme on (first) endpoint")
}
endpoint = fmt.Sprintf("%s://%s", defaultProto, endpoint)
}
// step: parse the url
u, err := url.Parse(endpoint)
if err != nil {
return nil, newInvalidEndpointError("invalid endpoint '%s': %s", endpoint, err)
}
if defaultProto == "" {
defaultProto = u.Scheme
}
// step: check for empty hosts
if u.Host == "" {
return nil, newInvalidEndpointError("endpoint: %s must have a host", endpoint)
}
// step: create a new node for this endpoint
members = append(members, &member{endpoint: u.String()})
}
return &cluster{
client: client,
members: members,
}, nil
}
// retrieve the current member, i.e. the current endpoint in use
func (c *cluster) getMember() (string, error) {
c.RLock()
defer c.RUnlock()
for _, n := range c.members {
if n.status == memberStatusUp {
return n.endpoint, nil
}
}
return "", ErrMarathonDown
}
// markDown marks down the current endpoint
func (c *cluster) markDown(endpoint string) {
c.Lock()
defer c.Unlock()
for _, n := range c.members {
// step: check if this is the node and it's marked as up - The double checking on the
// nodes status ensures the multiple calls don't create multiple checks
if n.status == memberStatusUp && n.endpoint == endpoint {
n.status = memberStatusDown
go c.healthCheckNode(n)
break
}
}
}
// 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 {
res, err := c.client.Get(fmt.Sprintf("%s/ping", node.endpoint))
if err == nil && res.StatusCode == 200 {
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
func (c *cluster) activeMembers() []string {
return c.membersList(memberStatusUp)
}
// nonActiveMembers returns a list of non-active members in the cluster
func (c *cluster) nonActiveMembers() []string {
return c.membersList(memberStatusDown)
}
// memberList returns a list of members of a specified status
func (c *cluster) membersList(status memberStatus) []string {
c.RLock()
defer c.RUnlock()
var list []string
for _, m := range c.members {
if m.status == status {
list = append(list, m.endpoint)
}
}
return list
}
// size returns the size of the cluster
func (c *cluster) size() int {
return len(c.members)
}
// String returns a string representation
func (m member) String() string {
status := "UP"
if m.status == memberStatusDown {
status = "DOWN"
}
return fmt.Sprintf("member: %s:%s", m.endpoint, status)
}