forked from influxdata/kapacitor
/
client.go
211 lines (186 loc) · 4.86 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
package client
import (
"bytes"
"crypto/tls"
"encoding/json"
"fmt"
"log"
"net/http"
"net/url"
"path"
"strconv"
"sync"
"sync/atomic"
"github.com/docker/docker/api/types"
"github.com/docker/docker/api/types/swarm"
"github.com/pkg/errors"
)
// Version is the Docker Engine API version needed for this package.
// Since docker versions their API this can be a constant value and newer docker daemons should still support this version once it becomes older.
// If at some point Docker removes this version then we will need to update this code.
const version = "v1.30"
type Config struct {
URLs []string
TLSConfig *tls.Config
}
type Client interface {
Update(c Config) error
Version() (string, error)
Service(name string) (*swarm.Service, error)
UpdateService(service *swarm.Service) error
}
type httpClient struct {
mu sync.RWMutex
config Config
urls []url.URL
client *http.Client
index int32
}
func New(c Config) (Client, error) {
urls, err := parseURLs(c.URLs)
if err != nil {
return nil, err
}
return &httpClient{
config: c,
urls: urls,
client: &http.Client{
Transport: &http.Transport{
TLSClientConfig: c.TLSConfig,
},
},
}, nil
}
func (c *httpClient) pickURL(urls []url.URL) url.URL {
i := atomic.LoadInt32(&c.index)
i = (i + 1) % int32(len(urls))
atomic.StoreInt32(&c.index, i)
return urls[i]
}
func parseURLs(urlStrs []string) ([]url.URL, error) {
urls := make([]url.URL, len(urlStrs))
for i, urlStr := range urlStrs {
u, err := url.Parse(urlStr)
if err != nil {
return nil, errors.Wrapf(err, "invalid url %q", urlStr)
}
urls[i] = *u
}
return urls, nil
}
func (c *httpClient) Update(new Config) error {
c.mu.Lock()
defer c.mu.Unlock()
old := c.config
c.config = new
urls, err := parseURLs(new.URLs)
if err != nil {
return err
}
c.urls = urls
if old.TLSConfig != new.TLSConfig {
c.client = &http.Client{
Transport: &http.Transport{
TLSClientConfig: new.TLSConfig,
},
}
}
return nil
}
func (c *httpClient) Do(r http.Request) (*http.Response, error) {
c.mu.RLock()
u := c.pickURL(c.urls)
client := c.client
c.mu.RUnlock()
r.URL.Host = u.Host
r.URL.Scheme = u.Scheme
resp, err := client.Do(&r)
return resp, errors.Wrap(err, "swarm client request failed")
}
func (c *httpClient) decodeResponse(resp *http.Response, response interface{}, successfulCodes ...int) error {
dec := json.NewDecoder(resp.Body)
successful := false
for _, code := range successfulCodes {
if code == resp.StatusCode {
successful = true
break
}
}
// Unsuccessful response code, decode status result
if !successful {
var e types.ErrorResponse
err := dec.Decode(&e)
if err != nil {
return errors.Wrapf(err, "failed to understand swarm server response: Code: %d", resp.StatusCode)
}
if e.Message == "" {
return fmt.Errorf("failed to understand swarm server error response: Code: %d", resp.StatusCode)
}
return errors.New(e.Message)
}
if response != nil {
// Decode response body into provided response object
if err := dec.Decode(response); err != nil {
return errors.Wrapf(err, "failed to decode swarm server response into %T: Code: %d", response, resp.StatusCode)
}
}
return nil
}
func (c *httpClient) Version() (string, error) {
r, err := http.NewRequest("GET", "/info", nil)
if err != nil {
return "", err
}
resp, err := c.Do(*r)
if err != nil {
return "", err
}
defer resp.Body.Close()
info := new(types.Info)
if err := c.decodeResponse(resp, info, http.StatusOK); err != nil {
return "", err
}
return info.ServerVersion, nil
}
func (c *httpClient) Service(id string) (*swarm.Service, error) {
p := path.Join("/", version, "services", id)
r, err := http.NewRequest("GET", p, nil)
if err != nil {
return nil, errors.Wrapf(err, "failed create GET request for %q", p)
}
resp, err := c.Do(*r)
if err != nil {
return nil, errors.Wrapf(err, "failed create GET request for %q", p)
}
defer resp.Body.Close()
service := new(swarm.Service)
if err := c.decodeResponse(resp, service, http.StatusOK); err != nil {
return nil, err
}
return service, nil
}
func (c *httpClient) UpdateService(service *swarm.Service) error {
var buf bytes.Buffer
err := json.NewEncoder(&buf).Encode(service.Spec)
if err != nil {
return errors.Wrapf(err, "failed to json encode service %q", service.ID)
}
p := path.Join("/", version, "services", service.ID, "update")
params := &url.Values{}
params.Set("version", strconv.Itoa(int(service.Version.Index)))
u := url.URL{
Path: p,
RawQuery: params.Encode(),
}
log.Println("D!", buf.String())
r, err := http.NewRequest("POST", u.String(), &buf)
if err != nil {
return errors.Wrapf(err, "failed to create POST request for %q", service.ID)
}
resp, err := c.Do(*r)
if err != nil {
return errors.Wrapf(err, "update service request failed for %q", service.ID)
}
defer resp.Body.Close()
return c.decodeResponse(resp, nil, http.StatusOK)
}