-
Notifications
You must be signed in to change notification settings - Fork 586
/
pgadminimpl.go
298 lines (249 loc) · 9.31 KB
/
pgadminimpl.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
package pgadminservice
/*
Copyright 2020 - 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"
"github.com/crunchydata/postgres-operator/internal/apiserver"
"github.com/crunchydata/postgres-operator/internal/config"
"github.com/crunchydata/postgres-operator/internal/kubeapi"
"github.com/crunchydata/postgres-operator/internal/pgadmin"
crv1 "github.com/crunchydata/postgres-operator/pkg/apis/crunchydata.com/v1"
msgs "github.com/crunchydata/postgres-operator/pkg/apiservermsgs"
log "github.com/sirupsen/logrus"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
const pgAdminServiceSuffix = "-pgadmin"
// CreatePgAdmin ...
// pgo create pgadmin mycluster
// pgo create pgadmin --selector=name=mycluster
func CreatePgAdmin(request *msgs.CreatePgAdminRequest, ns, pgouser string) msgs.CreatePgAdminResponse {
var err error
resp := msgs.CreatePgAdminResponse{
Status: msgs.Status{Code: msgs.Ok},
Results: []string{},
}
log.Debugf("createPgAdmin 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.SetError(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 pgAdmin to cluster [%s]", cluster.Name)
// generate the pgtask, starting with spec
spec := crv1.PgtaskSpec{
Namespace: cluster.Namespace,
Name: fmt.Sprintf("%s-%s", config.LABEL_PGADMIN_TASK_ADD, cluster.Name),
TaskType: crv1.PgtaskPgAdminAdd,
StorageSpec: cluster.Spec.PrimaryStorage,
Parameters: map[string]string{
config.LABEL_PGADMIN_TASK_CLUSTER: cluster.Name,
},
}
task := &crv1.Pgtask{
ObjectMeta: metav1.ObjectMeta{
Name: spec.Name,
Labels: map[string]string{
config.LABEL_PG_CLUSTER: cluster.Name,
config.LABEL_PGADMIN_TASK_ADD: "true",
config.LABEL_PGOUSER: pgouser,
},
},
Spec: spec,
}
if err := kubeapi.Createpgtask(apiserver.RESTClient, task, cluster.Namespace); err != nil {
log.Error(err)
resp.SetError("error creating tasks for one or more clusters")
resp.Results = append(resp.Results, fmt.Sprintf("%s: error - %s", cluster.Name, err.Error()))
continue
} else {
resp.Results = append(resp.Results, fmt.Sprintf("%s pgAdmin addition scheduled", cluster.Name))
}
}
return resp
}
// DeletePgAdmin ...
// pgo delete pgadmin mycluster
// pgo delete pgadmin --selector=name=mycluster
func DeletePgAdmin(request *msgs.DeletePgAdminRequest, ns string) msgs.DeletePgAdminResponse {
var err error
resp := msgs.DeletePgAdminResponse{
Status: msgs.Status{Code: msgs.Ok},
Results: []string{},
}
log.Debugf("deletePgAdmin 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.SetError(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("deleting pgAdmin from cluster [%s]", cluster.Name)
// generate the pgtask, starting with spec
spec := crv1.PgtaskSpec{
Namespace: cluster.Namespace,
Name: config.LABEL_PGADMIN_TASK_DELETE + "-" + cluster.Name,
TaskType: crv1.PgtaskPgAdminDelete,
Parameters: map[string]string{
config.LABEL_PGADMIN_TASK_CLUSTER: cluster.Name,
},
}
task := &crv1.Pgtask{
ObjectMeta: metav1.ObjectMeta{
Name: spec.Name,
Labels: map[string]string{
config.LABEL_PG_CLUSTER: cluster.Name,
config.LABEL_PGADMIN_TASK_DELETE: "true",
},
},
Spec: spec,
}
if err := kubeapi.Createpgtask(apiserver.RESTClient, task, cluster.Namespace); err != nil {
log.Error(err)
resp.SetError("error creating tasks for one or more clusters")
resp.Results = append(resp.Results, fmt.Sprintf("%s: error - %s", cluster.Name, err.Error()))
return resp
} else {
resp.Results = append(resp.Results, cluster.Name+" pgAdmin delete scheduled")
}
}
return resp
}
// ShowPgAdmin gets information about a PostgreSQL cluster's pgAdmin
// deployment
//
// pgo show pgadmin
// pgo show pgadmin --selector
func ShowPgAdmin(request *msgs.ShowPgAdminRequest, namespace string) msgs.ShowPgAdminResponse {
log.Debugf("show pgAdmin called, cluster [%v], selector [%s]", request.ClusterNames, request.Selector)
response := msgs.ShowPgAdminResponse{
Results: []msgs.ShowPgAdminDetail{},
Status: msgs.Status{Code: msgs.Ok},
}
// 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.SetError(err.Error())
return response
}
// iterate through the list of clusters to get the relevant pgAdmin
// information about them
for _, cluster := range clusterList.Items {
result := msgs.ShowPgAdminDetail{
ClusterName: cluster.Spec.Name,
HasPgAdmin: true,
}
// first, check if the cluster has the pgAdmin label. If it does not, we
// add it to the list and keep iterating
clusterLabels := cluster.GetLabels()
if clusterLabels[config.LABEL_PGADMIN] != "true" {
result.HasPgAdmin = false
response.Results = append(response.Results, result)
continue
}
// This takes advantage of pgadmin deployment and pgadmin service
// sharing a name that is clustername + pgAdminServiceSuffix
service, err := apiserver.Clientset.
CoreV1().Services(cluster.Namespace).
Get(cluster.Name+pgAdminServiceSuffix, metav1.GetOptions{})
if err != nil {
response.SetError(err.Error())
return response
}
result.ServiceClusterIP = service.Spec.ClusterIP
result.ServiceName = service.Name
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
}
// In the future, construct results to contain individual error stati
// for now log and return empty content if encountered
qr, err := pgadmin.GetPgAdminQueryRunner(apiserver.Clientset, apiserver.RESTConfig, &cluster)
if err != nil {
log.Error(err)
continue
} else if qr != nil {
names, err := pgadmin.GetUsernames(qr)
if err != nil {
log.Error(err)
continue
}
result.Users = names
}
// 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 != "" {
err := kubeapi.GetpgclustersBySelector(apiserver.RESTClient, &clusterList,
selector, namespace)
// if there is an error, return here with an empty cluster list
if err != nil {
return crv1.PgclusterList{}, err
}
}
// now try to get clusters based specific cluster names
for _, clusterName := range clusterNames {
cluster := crv1.Pgcluster{}
_, err := kubeapi.Getpgcluster(apiserver.RESTClient, &cluster,
clusterName, namespace)
// 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
}