forked from CrunchyData/postgres-operator
-
Notifications
You must be signed in to change notification settings - Fork 0
/
pgbouncerimpl.go
527 lines (431 loc) · 17.6 KB
/
pgbouncerimpl.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
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
package pgbouncerservice
/*
Copyright 2018 - 2021 Crunchy Data Solutions, Inc.
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.
*/
import (
"fmt"
"strings"
"github.com/crunchydata/postgres-operator/internal/apiserver"
"github.com/crunchydata/postgres-operator/internal/config"
clusteroperator "github.com/crunchydata/postgres-operator/internal/operator/cluster"
"github.com/crunchydata/postgres-operator/internal/util"
crv1 "github.com/crunchydata/postgres-operator/pkg/apis/crunchydata.com/v1"
msgs "github.com/crunchydata/postgres-operator/pkg/apiservermsgs"
log "github.com/sirupsen/logrus"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
const pgBouncerServiceSuffix = "-pgbouncer"
// CreatePgbouncer ...
// pgo create pgbouncer mycluster
// pgo create pgbouncer --selector=name=mycluster
func CreatePgbouncer(request *msgs.CreatePgbouncerRequest, ns, pgouser string) msgs.CreatePgbouncerResponse {
var err error
resp := msgs.CreatePgbouncerResponse{}
resp.Status.Code = msgs.Ok
resp.Status.Msg = ""
resp.Results = make([]string, 0)
// validate the CPU/Memory request parameters, if they are passed in
if err := apiserver.ValidateResourceRequestLimit(request.CPURequest, request.CPULimit, resource.Quantity{}); err != nil {
resp.Status.Code = msgs.Error
resp.Status.Msg = err.Error()
return resp
}
if err := apiserver.ValidateResourceRequestLimit(request.MemoryRequest, request.MemoryLimit,
apiserver.Pgo.Cluster.DefaultPgBouncerResourceMemory); err != nil {
resp.Status.Code = msgs.Error
resp.Status.Msg = err.Error()
return resp
}
// validate the number of replicas being requested
if request.Replicas < 0 {
resp.Status.Code = msgs.Error
resp.Status.Msg = fmt.Sprintf(apiserver.ErrMessageReplicas, 1)
return resp
}
log.Debugf("createPgbouncer selector is [%s]", request.Selector)
// try to get the list of clusters. if there is an error, put it into the
// status and return
clusterList, err := getClusterList(request.Namespace, request.Args, request.Selector)
if err != nil {
resp.Status.Code = msgs.Error
resp.Status.Msg = err.Error()
return resp
}
for _, cluster := range clusterList.Items {
// check if the current cluster is not upgraded to the deployed
// Operator version. If not, do not allow the command to complete
if cluster.Annotations[config.ANNOTATION_IS_UPGRADED] == config.ANNOTATIONS_FALSE {
resp.Status.Code = msgs.Error
resp.Status.Msg = cluster.Name + msgs.UpgradeError
return resp
}
log.Debugf("adding pgbouncer to cluster [%s]", cluster.Name)
resources := v1.ResourceList{}
limits := v1.ResourceList{}
// Set the value that enables the pgBouncer, which is the replicas
// Set the default value, and if there is a custom number of replicas
// provided, set it to that
cluster.Spec.PgBouncer.Replicas = config.DefaultPgBouncerReplicas
if request.Replicas > 0 {
cluster.Spec.PgBouncer.Replicas = request.Replicas
}
// if the request has overriding CPU/memory parameters,
// these will take precedence over the defaults
if request.CPULimit != "" {
// as this was already validated, we can ignore the error
quantity, _ := resource.ParseQuantity(request.CPULimit)
limits[v1.ResourceCPU] = quantity
}
if request.CPURequest != "" {
// as this was already validated, we can ignore the error
quantity, _ := resource.ParseQuantity(request.CPURequest)
resources[v1.ResourceCPU] = quantity
}
if request.MemoryLimit != "" {
// as this was already validated, we can ignore the error
quantity, _ := resource.ParseQuantity(request.MemoryLimit)
limits[v1.ResourceMemory] = quantity
}
if request.MemoryRequest != "" {
// as this was already validated, we can ignore the error
quantity, _ := resource.ParseQuantity(request.MemoryRequest)
resources[v1.ResourceMemory] = quantity
} else {
resources[v1.ResourceMemory] = apiserver.Pgo.Cluster.DefaultPgBouncerResourceMemory
}
cluster.Spec.PgBouncer.Resources = resources
// update the cluster CRD with these udpates. If there is an error
if _, err := apiserver.Clientset.CrunchydataV1().Pgclusters(request.Namespace).Update(&cluster); err != nil {
log.Error(err)
resp.Results = append(resp.Results, err.Error())
continue
}
resp.Results = append(resp.Results, fmt.Sprintf("%s pgbouncer added", cluster.Name))
}
return resp
}
// DeletePgbouncer ...
// pgo delete pgbouncer mycluster
// pgo delete pgbouncer --selector=name=mycluster
func DeletePgbouncer(request *msgs.DeletePgbouncerRequest, ns string) msgs.DeletePgbouncerResponse {
var err error
resp := msgs.DeletePgbouncerResponse{}
resp.Status.Code = msgs.Ok
resp.Status.Msg = ""
resp.Results = make([]string, 0)
log.Debugf("deletePgbouncer selector is [%s]", request.Selector)
// try to get the list of clusters. if there is an error, put it into the
// status and return
clusterList, err := getClusterList(request.Namespace, request.Args, request.Selector)
if err != nil {
resp.Status.Code = msgs.Error
resp.Status.Msg = err.Error()
return resp
}
// Return an error if any clusters identified to have pgbouncer fully deleted (as specified
// using the uninstall parameter) have standby mode enabled and the 'uninstall' option selected.
// This because while in standby mode the cluster is read-only, preventing the execution of the
// SQL required to remove pgBouncer.
if hasStandby, standbyClusters := apiserver.PGClusterListHasStandby(clusterList); hasStandby &&
request.Uninstall {
resp.Status.Code = msgs.Error
resp.Status.Msg = fmt.Sprintf("Request rejected, unable to delete pgbouncer using the "+
"'uninstall' parameter for clusters %s: %s.", strings.Join(standbyClusters, ","),
apiserver.ErrStandbyNotAllowed.Error())
return resp
}
for _, cluster := range clusterList.Items {
log.Debugf("deleting pgbouncer from cluster [%s]", cluster.Name)
// check to see if the uninstall flag was set. If it was, apply the update
// inline
if request.Uninstall {
if err := clusteroperator.UninstallPgBouncer(apiserver.Clientset, apiserver.RESTConfig, &cluster); err != nil {
log.Error(err)
resp.Status.Code = msgs.Error
resp.Results = append(resp.Results, err.Error())
return resp
}
}
// Disable the pgBouncer Deploymnet, which means setting Replicas to 0
cluster.Spec.PgBouncer.Replicas = 0
// Set the resources/limits to their default values
cluster.Spec.PgBouncer.Resources = v1.ResourceList{}
cluster.Spec.PgBouncer.Limits = v1.ResourceList{}
// update the cluster CRD with these udpates. If there is an error
if _, err := apiserver.Clientset.CrunchydataV1().Pgclusters(request.Namespace).Update(&cluster); err != nil {
log.Error(err)
resp.Status.Code = msgs.Error
resp.Results = append(resp.Results, err.Error())
return resp
}
// follow the legacy format for returning this information
result := fmt.Sprintf("%s pgbouncer deleted", cluster.Name)
resp.Results = append(resp.Results, result)
}
return resp
}
// ShowPgBouncer gets information about a PostgreSQL cluster's pgBouncer
// deployment
//
// pgo show pgbouncer
// pgo show pgbouncer --selector
func ShowPgBouncer(request *msgs.ShowPgBouncerRequest, namespace string) msgs.ShowPgBouncerResponse {
// set up a dummy response
response := msgs.ShowPgBouncerResponse{
Results: []msgs.ShowPgBouncerDetail{},
Status: msgs.Status{
Code: msgs.Ok,
Msg: "",
},
}
log.Debugf("show pgbouncer called, cluster [%v], selector [%s]", request.ClusterNames, request.Selector)
// try to get the list of clusters. if there is an error, put it into the
// status and return
clusterList, err := getClusterList(request.Namespace, request.ClusterNames, request.Selector)
if err != nil {
response.Status.Code = msgs.Error
response.Status.Msg = err.Error()
return response
}
// iterate through the list of clusters to get the relevant pgBouncer
// information about them
for _, cluster := range clusterList.Items {
result := msgs.ShowPgBouncerDetail{
ClusterName: cluster.Spec.Name,
HasPgBouncer: true,
}
// first, check if the cluster has pgBouncer enabled
if !cluster.Spec.PgBouncer.Enabled() {
result.HasPgBouncer = false
response.Results = append(response.Results, result)
continue
}
// only set the pgBouncer user if we know this is a pgBouncer enabled
// cluster...even though, yes, this is a constant
result.Username = crv1.PGUserPgBouncer
// set the pgBouncer service information on this record
setPgBouncerServiceDetail(cluster, &result)
// get the user information about the pgBouncer deployment
setPgBouncerPasswordDetail(cluster, &result)
// append the result to the list
response.Results = append(response.Results, result)
}
return response
}
// UpdatePgBouncer updates a cluster's pgBouncer deployment based on the
// parameters passed in. This includes:
//
// - password rotation
// - updating CPU/memory resources
func UpdatePgBouncer(request *msgs.UpdatePgBouncerRequest, namespace, pgouser string) msgs.UpdatePgBouncerResponse {
// set up a dummy response
response := msgs.UpdatePgBouncerResponse{
// Results: []msgs.ShowPgBouncerDetail{},
Status: msgs.Status{
Code: msgs.Ok,
Msg: "",
},
}
// validate the CPU/Memory parameters, if they are passed in
zeroQuantity := resource.Quantity{}
if err := apiserver.ValidateResourceRequestLimit(request.CPURequest, request.CPULimit, zeroQuantity); err != nil {
response.Status.Code = msgs.Error
response.Status.Msg = err.Error()
return response
}
// Don't check the default value as pgBouncer is already deployed
if err := apiserver.ValidateResourceRequestLimit(request.MemoryRequest, request.MemoryLimit, zeroQuantity); err != nil {
response.Status.Code = msgs.Error
response.Status.Msg = err.Error()
return response
}
// validate the number of replicas being requested
if request.Replicas < 0 {
response.Status.Code = msgs.Error
response.Status.Msg = fmt.Sprintf(apiserver.ErrMessageReplicas, 1)
return response
}
log.Debugf("update pgbouncer called, cluster [%v], selector [%s]", request.ClusterNames, request.Selector)
// try to get the list of clusters. if there is an error, put it into the
// status and return
clusterList, err := getClusterList(request.Namespace, request.ClusterNames, request.Selector)
if err != nil {
response.Status.Code = msgs.Error
response.Status.Msg = err.Error()
return response
}
// Return an error if any clusters selected to have pgbouncer updated have standby mode enabled.
// This is because while in standby mode the cluster is read-only, preventing the execution of the
// SQL required to update pgbouncer.
if hasStandby, standbyClusters := apiserver.PGClusterListHasStandby(clusterList); hasStandby {
response.Status.Code = msgs.Error
response.Status.Msg = fmt.Sprintf("Request rejected, unable to update pgbouncer for "+
"clusters %s: %s.", strings.Join(standbyClusters, ","),
apiserver.ErrStandbyNotAllowed.Error())
return response
}
// iterate through the list of clusters to get the relevant pgBouncer
// information about them
for _, cluster := range clusterList.Items {
result := msgs.UpdatePgBouncerDetail{
ClusterName: cluster.Spec.Name,
HasPgBouncer: true,
}
// first, check if the cluster has pgBouncer enabled
if !cluster.Spec.PgBouncer.Enabled() {
result.HasPgBouncer = false
response.Results = append(response.Results, result)
continue
}
// if we are rotating the password, perform the request inline
if request.RotatePassword {
if err := clusteroperator.RotatePgBouncerPassword(apiserver.Clientset, apiserver.RESTConfig, &cluster); err != nil {
log.Error(err)
result.Error = true
result.ErrorMessage = err.Error()
response.Results = append(response.Results, result)
}
}
// ensure the Resources/Limits are non-nil
if cluster.Spec.PgBouncer.Resources == nil {
cluster.Spec.PgBouncer.Resources = v1.ResourceList{}
}
if cluster.Spec.PgBouncer.Limits == nil {
cluster.Spec.PgBouncer.Limits = v1.ResourceList{}
}
// if the request has overriding CPU/Memory parameters,
// add them to the cluster's pgbouncer resource list
if request.CPULimit != "" {
// as this was already validated, we can ignore the error
quantity, _ := resource.ParseQuantity(request.CPULimit)
cluster.Spec.PgBouncer.Limits[v1.ResourceCPU] = quantity
}
if request.CPURequest != "" {
// as this was already validated, we can ignore the error
quantity, _ := resource.ParseQuantity(request.CPURequest)
cluster.Spec.PgBouncer.Resources[v1.ResourceCPU] = quantity
}
if request.MemoryLimit != "" {
// as this was already validated, we can ignore the error
quantity, _ := resource.ParseQuantity(request.MemoryLimit)
cluster.Spec.PgBouncer.Limits[v1.ResourceMemory] = quantity
}
if request.MemoryRequest != "" {
// as this was already validated, we can ignore the error
quantity, _ := resource.ParseQuantity(request.MemoryRequest)
cluster.Spec.PgBouncer.Resources[v1.ResourceMemory] = quantity
}
// apply the replica count number if there is a change, i.e. replicas is not
// 0
if request.Replicas > 0 {
cluster.Spec.PgBouncer.Replicas = request.Replicas
}
if _, err := apiserver.Clientset.CrunchydataV1().Pgclusters(cluster.Namespace).Update(&cluster); err != nil {
log.Error(err)
result.Error = true
result.ErrorMessage = err.Error()
response.Results = append(response.Results, result)
continue
}
// append the result to the list
response.Results = append(response.Results, result)
}
return response
}
// getClusterList tries to return a list of clusters based on either having an
// argument list of cluster names, or a Kubernetes selector
func getClusterList(namespace string, clusterNames []string, selector string) (crv1.PgclusterList, error) {
clusterList := crv1.PgclusterList{}
// see if there are any values in the cluster name list or in the selector
// if nothing exists, return an error
if len(clusterNames) == 0 && selector == "" {
err := fmt.Errorf("either a list of cluster names or a selector needs to be supplied for this comment")
return clusterList, err
}
// try to build the cluster list based on either the selector or the list
// of arguments...or both. First, start with the selector
if selector != "" {
cl, err := apiserver.Clientset.CrunchydataV1().Pgclusters(namespace).List(metav1.ListOptions{LabelSelector: selector})
// if there is an error, return here with an empty cluster list
if err != nil {
return crv1.PgclusterList{}, err
}
clusterList = *cl
}
// now try to get clusters based specific cluster names
for _, clusterName := range clusterNames {
cluster, err := apiserver.Clientset.CrunchydataV1().Pgclusters(namespace).Get(clusterName, metav1.GetOptions{})
// if there is an error, capture it here and return here with an empty list
if err != nil {
return crv1.PgclusterList{}, err
}
// if successful, append to the cluster list
clusterList.Items = append(clusterList.Items, *cluster)
}
log.Debugf("clusters founds: [%d]", len(clusterList.Items))
// if after all this, there are no clusters found, return an error
if len(clusterList.Items) == 0 {
err := fmt.Errorf("no clusters found")
return clusterList, err
}
// all set! return the cluster list with error
return clusterList, nil
}
// setPgBouncerPasswordDetail applies the password that is used by the pgbouncer
// service account
func setPgBouncerPasswordDetail(cluster crv1.Pgcluster, result *msgs.ShowPgBouncerDetail) {
pgBouncerSecretName := util.GeneratePgBouncerSecretName(cluster.Spec.Name)
// attempt to get the secret, but only get the password
password, err := util.GetPasswordFromSecret(apiserver.Clientset,
cluster.Spec.Namespace, pgBouncerSecretName)
if err != nil {
log.Warn(err)
}
// and set the password. Easy!
result.Password = password
}
// setPgBouncerServiceDetail applies the information about the pgBouncer service
// to the result for the pgBouncer show
func setPgBouncerServiceDetail(cluster crv1.Pgcluster, result *msgs.ShowPgBouncerDetail) {
// get the service information about the pgBouncer deployment
selector := fmt.Sprintf("%s=%s", config.LABEL_PG_CLUSTER, cluster.Spec.Name)
// have to go through a bunch of services because "current design"
services, err := apiserver.Clientset.
CoreV1().Services(cluster.Spec.Namespace).
List(metav1.ListOptions{LabelSelector: selector})
// if there is an error, return without making any adjustments
if err != nil {
log.Warn(err)
return
}
log.Debugf("cluster [%s] has [%d] services", cluster.Spec.Name, len(services.Items))
// adding the service information was borrowed from the ShowCluster
// resource
for _, service := range services.Items {
// if this service is not for pgBouncer, then skip
if !strings.HasSuffix(service.Name, pgBouncerServiceSuffix) {
continue
}
// this is the pgBouncer service!
result.ServiceClusterIP = service.Spec.ClusterIP
result.ServiceName = service.Name
// try to get the exterinal IP based on the formula used in show cluster
if len(service.Spec.ExternalIPs) > 0 {
result.ServiceExternalIP = service.Spec.ExternalIPs[0]
}
if len(service.Status.LoadBalancer.Ingress) > 0 {
result.ServiceExternalIP = service.Status.LoadBalancer.Ingress[0].IP
}
}
}