forked from FoundationDB/fdb-kubernetes-operator
/
pod_client.go
419 lines (350 loc) · 12.2 KB
/
pod_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
/*
* pod_client.go
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2018-2019 Apple Inc. and the FoundationDB project authors
*
* 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 foundationdbcluster
import (
"crypto/sha256"
"crypto/tls"
"crypto/x509"
"encoding/hex"
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"os"
"strings"
fdbtypes "github.com/foundationdb/fdb-kubernetes-operator/pkg/apis/apps/v1beta1"
corev1 "k8s.io/api/core/v1"
)
// FdbPodClient provides methods for working with a FoundationDB pod
type FdbPodClient interface {
// GetCluster returns the cluster associated with a client
GetCluster() *fdbtypes.FoundationDBCluster
// GetPod returns the pod associated with a client
GetPod() *corev1.Pod
// GetPodIP gets the IP address for a pod.
GetPodIP() string
// IsPresent checks whether a file in the sidecar is present
IsPresent(filename string) (bool, error)
// CheckHash checks whether a file in the sidecar has the expected contents.
CheckHash(filename string, contents string) (bool, error)
// GenerateMonitorConf updates the monitor conf file for a pod
GenerateMonitorConf() error
// CopyFiles copies the files from the config map to the shared dynamic conf
// volume
CopyFiles() error
// GetVariableSubstitutions gets the current keys and values that this
// instance will substitute into its monitor conf.
GetVariableSubstitutions() (map[string]string, error)
}
// realPodClient provides a client for use in real environments.
type realFdbPodClient struct {
// Cluster is the cluster we are connecting to.
Cluster *fdbtypes.FoundationDBCluster
// Pod is the pod we are connecting to.
Pod *corev1.Pod
// useTLS indicates whether this is using a TLS connection to the sidecar.
useTLS bool
// tlsConfig contains the TLS configuration for the connection to the
// sidecar.
tlsConfig *tls.Config
}
// NewFdbPodClient builds a client for working with an FDB Pod
func NewFdbPodClient(cluster *fdbtypes.FoundationDBCluster, pod *corev1.Pod) (FdbPodClient, error) {
if pod.Status.PodIP == "" {
return nil, fdbPodClientErrorNoIP
}
for _, container := range pod.Status.ContainerStatuses {
if !container.Ready {
return nil, fdbPodClientErrorNotReady
}
}
useTLS := podHasSidecarTLS(pod)
var tlsConfig = &tls.Config{}
if useTLS {
cert, err := tls.LoadX509KeyPair(
os.Getenv("FDB_TLS_CERTIFICATE_FILE"),
os.Getenv("FDB_TLS_KEY_FILE"),
)
if err != nil {
return nil, err
}
tlsConfig.Certificates = []tls.Certificate{cert}
if os.Getenv("DISABLE_SIDECAR_TLS_CHECK") == "1" {
tlsConfig.InsecureSkipVerify = true
}
certPool := x509.NewCertPool()
caList, err := ioutil.ReadFile(os.Getenv("FDB_TLS_CA_FILE"))
if err != nil {
return nil, err
}
certPool.AppendCertsFromPEM(caList)
tlsConfig.RootCAs = certPool
}
return &realFdbPodClient{Cluster: cluster, Pod: pod, useTLS: useTLS, tlsConfig: tlsConfig}, nil
}
// GetCluster returns the cluster associated with a client
func (client *realFdbPodClient) GetCluster() *fdbtypes.FoundationDBCluster {
return client.Cluster
}
// GetPod returns the pod associated with a client
func (client *realFdbPodClient) GetPod() *corev1.Pod {
return client.Pod
}
// GetPodIP gets the IP address for a pod.
func (client *realFdbPodClient) GetPodIP() string {
return client.Pod.Status.PodIP
}
// makeRequest submits a request to the sidecar.
func (client *realFdbPodClient) makeRequest(method string, path string) (string, error) {
var protocol string
if client.useTLS {
protocol = "https"
} else {
protocol = "http"
}
url := fmt.Sprintf("%s://%s:8080/%s", protocol, client.GetPodIP(), path)
var resp *http.Response
var err error
httpClient := &http.Client{}
if client.useTLS {
httpClient.Transport = &http.Transport{TLSClientConfig: client.tlsConfig}
}
switch method {
case "GET":
resp, err = httpClient.Get(url)
case "POST":
resp, err = httpClient.Post(url, "application/json", strings.NewReader(""))
default:
return "", fmt.Errorf("Unknown HTTP method %s", method)
}
if err != nil {
return "", err
}
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)
bodyText := string(body)
if err != nil {
return "", err
}
if resp.StatusCode >= 400 {
return "", failedResponse{response: resp, body: bodyText}
}
return bodyText, nil
}
// IsPresent checks whether a file in the sidecar is present.
func (client *realFdbPodClient) IsPresent(filename string) (bool, error) {
_, err := client.makeRequest("GET", fmt.Sprintf("check_hash/%s", filename))
if err == nil {
return true, err
}
response, isResponse := err.(failedResponse)
if isResponse && response.response.StatusCode == 404 {
return false, nil
} else {
return false, err
}
}
// CheckHash checks whether a file in the sidecar has the expected contents.
func (client *realFdbPodClient) CheckHash(filename string, contents string) (bool, error) {
response, err := client.makeRequest("GET", fmt.Sprintf("check_hash/%s", filename))
if err != nil {
return false, err
}
expectedHash := sha256.Sum256([]byte(contents))
expectedHashString := hex.EncodeToString(expectedHash[:])
return strings.Compare(expectedHashString, response) == 0, nil
}
// GenerateMonitorConf updates the monitor conf file for a pod
func (client *realFdbPodClient) GenerateMonitorConf() error {
_, err := client.makeRequest("POST", "copy_monitor_conf")
return err
}
// CopyFiles copies the files from the config map to the shared dynamic conf
// volume
func (client *realFdbPodClient) CopyFiles() error {
_, err := client.makeRequest("POST", "copy_files")
return err
}
// GetVariableSubstitutions gets the current keys and values that this
// instance will substitute into its monitor conf.
func (client *realFdbPodClient) GetVariableSubstitutions() (map[string]string, error) {
contents, err := client.makeRequest("GET", "substitutions")
if err != nil {
return nil, err
}
substitutions := map[string]string{}
err = json.Unmarshal([]byte(contents), &substitutions)
return substitutions, err
}
// MockFdbPodClient provides a mock connection to a pod
type mockFdbPodClient struct {
Cluster *fdbtypes.FoundationDBCluster
Pod *corev1.Pod
}
// NewMockFdbPodClient builds a mock client for working with an FDB pod
func NewMockFdbPodClient(cluster *fdbtypes.FoundationDBCluster, pod *corev1.Pod) (FdbPodClient, error) {
return &mockFdbPodClient{Cluster: cluster, Pod: pod}, nil
}
// GetCluster returns the cluster associated with a client
func (client *mockFdbPodClient) GetCluster() *fdbtypes.FoundationDBCluster {
return client.Cluster
}
// GetPod returns the pod associated with a client
func (client *mockFdbPodClient) GetPod() *corev1.Pod {
return client.Pod
}
// GetPodIP gets the IP address for a pod.
func (client *mockFdbPodClient) GetPodIP() string {
return mockPodIP(client.Pod)
}
// IsPresent checks whether a file in the sidecar is prsent.
func (client *mockFdbPodClient) IsPresent(filename string) (bool, error) {
return true, nil
}
// CheckHash checks whether a file in the sidecar has the expected contents.
func (client *mockFdbPodClient) CheckHash(filename string, contents string) (bool, error) {
return true, nil
}
// GenerateMonitorConf updates the monitor conf file for a pod
func (client *mockFdbPodClient) GenerateMonitorConf() error {
return nil
}
// CopyFiles copies the files from the config map to the shared dynamic conf
// volume
func (client *mockFdbPodClient) CopyFiles() error {
return nil
}
// mockPodIp generates an IP address for a pod in the mock client.
func mockPodIP(pod *corev1.Pod) string {
components := strings.Split(pod.Labels["fdb-instance-id"], "-")
for index, class := range fdbtypes.ProcessClasses {
if class == components[len(components)-2] {
return fmt.Sprintf("1.1.%d.%s", index, components[len(components)-1])
}
}
return "0.0.0.0"
}
// UpdateDynamicFiles checks if the files in the dynamic conf volume match the
// expected contents, and tries to copy the latest files from the input volume
// if they do not.
func UpdateDynamicFiles(client FdbPodClient, filename string, contents string, updateFunc func(client FdbPodClient) error) (bool, error) {
match := false
var err error
match, err = client.CheckHash(filename, contents)
if err != nil {
return false, err
}
if !match {
log.Info("Waiting for config update", "namespace", client.GetPod().Namespace, "pod", client.GetPod().Name, "file", filename)
err = updateFunc(client)
if err != nil {
return false, err
}
return client.CheckHash(filename, contents)
}
return true, nil
}
// CheckDynamicFilePresent waits for a file to be present in the dynamic conf
func CheckDynamicFilePresent(client FdbPodClient, filename string) (bool, error) {
present, err := client.IsPresent(filename)
if !present {
log.Info("Waiting for file", "namespace", client.GetPod().Namespace, "pod", client.GetPod().Name, "file", filename)
}
return present, err
}
// GetVariableSubstitutions gets the current keys and values that this
// instance will substitute into its monitor conf.
func (client *mockFdbPodClient) GetVariableSubstitutions() (map[string]string, error) {
substitutions := map[string]string{}
substitutions["FDB_PUBLIC_IP"] = client.Pod.Status.PodIP
if client.Cluster.Spec.FaultDomain.Key == "foundationdb.org/none" {
substitutions["FDB_MACHINE_ID"] = client.Pod.Name
substitutions["FDB_ZONE_ID"] = client.Pod.Name
} else if client.Cluster.Spec.FaultDomain.Key == "foundationdb.org/kubernetes-cluster" {
substitutions["FDB_MACHINE_ID"] = client.Pod.Spec.NodeName
substitutions["FDB_ZONE_ID"] = client.Cluster.Spec.FaultDomain.Value
} else {
faultDomainSource := client.Cluster.Spec.FaultDomain.ValueFrom
if faultDomainSource == "" {
faultDomainSource = "spec.nodeName"
}
substitutions["FDB_MACHINE_ID"] = client.Pod.Spec.NodeName
if faultDomainSource == "spec.nodeName" {
substitutions["FDB_ZONE_ID"] = client.Pod.Spec.NodeName
} else {
return nil, fmt.Errorf("Unsupported fault domain source %s", faultDomainSource)
}
}
substitutions["FDB_INSTANCE_ID"] = client.Pod.ObjectMeta.Labels["fdb-instance-id"]
version, err := fdbtypes.ParseFdbVersion(client.Cluster.Spec.Version)
if err != nil {
return nil, err
}
if version.SupportsUsingBinariesFromMainContainer() {
if client.Cluster.IsBeingUpgraded() {
substitutions["BINARY_DIR"] = fmt.Sprintf("/var/dynamic-conf/bin/%s", client.Cluster.Spec.Version)
} else {
substitutions["BINARY_DIR"] = "/usr/bin"
}
}
return substitutions, nil
}
// fdbPodClient provides errors that are returned when talking to FDB pods.
type fdbPodClientError int
const (
// fdbPodClientErrorNoIP is returned when the pod has not been assigned an
// IP address.
fdbPodClientErrorNoIP fdbPodClientError = iota
// fdbPodClientErrorNotReady is returned when the pod is not ready to
// recieve requests.
fdbPodClientErrorNotReady fdbPodClientError = iota
)
// Error generates an error message.
func (err fdbPodClientError) Error() string {
switch err {
case fdbPodClientErrorNoIP:
return "Pod does not have an IP address"
default:
return fmt.Sprintf("Unknown error code %d", err)
}
}
// failedResponse is an error thrown when a request to the sidecar fails.
type failedResponse struct {
response *http.Response
body string
}
// Error generates an error message.
func (response failedResponse) Error() string {
return fmt.Sprintf("HTTP request failed. Status=%d; response=%s", response.response.StatusCode, response.body)
}
// podHasSidecarTLS determines whether a pod currently has TLS enabled for the
// sidecar process.
func podHasSidecarTLS(pod *corev1.Pod) bool {
for _, container := range pod.Spec.Containers {
if container.Name == "foundationdb-kubernetes-sidecar" {
for _, arg := range container.Args {
if arg == "--tls" {
return true
}
}
}
}
return false
}