-
Notifications
You must be signed in to change notification settings - Fork 586
/
backup.go
340 lines (295 loc) · 12.6 KB
/
backup.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
package backrest
/*
Copyright 2018 - 2020 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 (
"bytes"
"encoding/json"
"fmt"
"os"
"regexp"
"strings"
"time"
"github.com/crunchydata/postgres-operator/internal/config"
"github.com/crunchydata/postgres-operator/internal/kubeapi"
"github.com/crunchydata/postgres-operator/internal/operator"
"github.com/crunchydata/postgres-operator/internal/util"
crv1 "github.com/crunchydata/postgres-operator/pkg/apis/crunchydata.com/v1"
"github.com/crunchydata/postgres-operator/pkg/events"
pgo "github.com/crunchydata/postgres-operator/pkg/generated/clientset/versioned"
log "github.com/sirupsen/logrus"
v1batch "k8s.io/api/batch/v1"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/client-go/kubernetes"
)
type backrestJobTemplateFields struct {
JobName string
Name string
ClusterName string
Command string
CommandOpts string
PITRTarget string
PodName string
PGOImagePrefix string
PGOImageTag string
SecurityContext string
PgbackrestStanza string
PgbackrestDBPath string
PgbackrestRepoPath string
PgbackrestRepoType string
BackrestLocalAndS3Storage bool
PgbackrestS3VerifyTLS string
PgbackrestRestoreVolumes string
PgbackrestRestoreVolumeMounts string
}
var backrestPgHostRegex = regexp.MustCompile("--db-host|--pg1-host")
var backrestPgPathRegex = regexp.MustCompile("--db-path|--pg1-path")
// Backrest ...
func Backrest(namespace string, clientset kubernetes.Interface, task *crv1.Pgtask) {
//create the Job to run the backrest command
cmd := task.Spec.Parameters[config.LABEL_BACKREST_COMMAND]
jobFields := backrestJobTemplateFields{
JobName: task.Spec.Parameters[config.LABEL_JOB_NAME],
ClusterName: task.Spec.Parameters[config.LABEL_PG_CLUSTER],
PodName: task.Spec.Parameters[config.LABEL_POD_NAME],
SecurityContext: "{}",
Command: cmd,
CommandOpts: task.Spec.Parameters[config.LABEL_BACKREST_OPTS],
PITRTarget: "",
PGOImagePrefix: util.GetValueOrDefault(task.Spec.Parameters[config.LABEL_IMAGE_PREFIX], operator.Pgo.Pgo.PGOImagePrefix),
PGOImageTag: operator.Pgo.Pgo.PGOImageTag,
PgbackrestStanza: task.Spec.Parameters[config.LABEL_PGBACKREST_STANZA],
PgbackrestDBPath: task.Spec.Parameters[config.LABEL_PGBACKREST_DB_PATH],
PgbackrestRepoPath: task.Spec.Parameters[config.LABEL_PGBACKREST_REPO_PATH],
PgbackrestRestoreVolumes: "",
PgbackrestRestoreVolumeMounts: "",
PgbackrestRepoType: operator.GetRepoType(task.Spec.Parameters[config.LABEL_BACKREST_STORAGE_TYPE]),
BackrestLocalAndS3Storage: operator.IsLocalAndS3Storage(task.Spec.Parameters[config.LABEL_BACKREST_STORAGE_TYPE]),
PgbackrestS3VerifyTLS: task.Spec.Parameters[config.LABEL_BACKREST_S3_VERIFY_TLS],
}
podCommandOpts, err := getCommandOptsFromPod(clientset, task, namespace)
if err != nil {
log.Error(err.Error())
return
}
jobFields.CommandOpts = jobFields.CommandOpts + " " + podCommandOpts
var doc2 bytes.Buffer
if err := config.BackrestjobTemplate.Execute(&doc2, jobFields); err != nil {
log.Error(err.Error())
return
}
if operator.CRUNCHY_DEBUG {
config.BackrestjobTemplate.Execute(os.Stdout, jobFields)
}
newjob := v1batch.Job{}
err = json.Unmarshal(doc2.Bytes(), &newjob)
if err != nil {
log.Error("error unmarshalling json into Job " + err.Error())
return
}
// set the container image to an override value, if one exists
operator.SetContainerImageOverride(config.CONTAINER_IMAGE_PGO_BACKREST,
&newjob.Spec.Template.Spec.Containers[0])
newjob.ObjectMeta.Labels[config.LABEL_PGOUSER] = task.ObjectMeta.Labels[config.LABEL_PGOUSER]
newjob.ObjectMeta.Labels[config.LABEL_PG_CLUSTER_IDENTIFIER] = task.ObjectMeta.Labels[config.LABEL_PG_CLUSTER_IDENTIFIER]
backupType := task.Spec.Parameters[config.LABEL_PGHA_BACKUP_TYPE]
if backupType != "" {
newjob.ObjectMeta.Labels[config.LABEL_PGHA_BACKUP_TYPE] = backupType
}
clientset.BatchV1().Jobs(namespace).Create(&newjob)
//publish backrest backup event
if cmd == "backup" {
topics := make([]string, 1)
topics[0] = events.EventTopicBackup
f := events.EventCreateBackupFormat{
EventHeader: events.EventHeader{
Namespace: namespace,
Username: task.ObjectMeta.Labels[config.LABEL_PGOUSER],
Topic: topics,
Timestamp: time.Now(),
EventType: events.EventCreateBackup,
},
Clustername: jobFields.ClusterName,
BackupType: "pgbackrest",
}
err := events.Publish(f)
if err != nil {
log.Error(err.Error())
}
}
}
// CreateInitialBackup creates a Pgtask in order to initiate the initial pgBackRest backup for a cluster
// as needed to support replica creation
func CreateInitialBackup(clientset pgo.Interface, namespace, clusterName, podName string) (*crv1.Pgtask, error) {
var params map[string]string
params = make(map[string]string)
params[config.LABEL_PGHA_BACKUP_TYPE] = crv1.BackupTypeBootstrap
return CreateBackup(clientset, namespace, clusterName, podName, params, "--type=full")
}
// CreatePostFailoverBackup creates a Pgtask in order to initiate the a pgBackRest backup following a failure
// event to ensure proper replica creation and/or reinitialization
func CreatePostFailoverBackup(clientset pgo.Interface, namespace, clusterName, podName string) (*crv1.Pgtask, error) {
var params map[string]string
params = make(map[string]string)
params[config.LABEL_PGHA_BACKUP_TYPE] = crv1.BackupTypeFailover
return CreateBackup(clientset, namespace, clusterName, podName, params, "")
}
// CreateBackup creates a Pgtask in order to initiate a pgBackRest backup
func CreateBackup(clientset pgo.Interface, namespace, clusterName, podName string, params map[string]string,
backupOpts string) (*crv1.Pgtask, error) {
log.Debug("pgBackRest operator CreateBackup called")
cluster, err := clientset.CrunchydataV1().Pgclusters(namespace).Get(clusterName, metav1.GetOptions{})
if err != nil {
log.Error(err)
return nil, err
}
var newInstance *crv1.Pgtask
taskName := "backrest-backup-" + cluster.Name
spec := crv1.PgtaskSpec{}
spec.Name = taskName
spec.Namespace = namespace
spec.TaskType = crv1.PgtaskBackrest
spec.Parameters = make(map[string]string)
spec.Parameters[config.LABEL_JOB_NAME] = "backrest-" + crv1.PgtaskBackrestBackup + "-" + cluster.Name
spec.Parameters[config.LABEL_PG_CLUSTER] = cluster.Name
spec.Parameters[config.LABEL_POD_NAME] = podName
spec.Parameters[config.LABEL_CONTAINER_NAME] = "database"
// pass along the appropriate image prefix for the backup task
// this will be used by the associated backrest job
spec.Parameters[config.LABEL_IMAGE_PREFIX] = util.GetValueOrDefault(cluster.Spec.PGOImagePrefix, operator.Pgo.Pgo.PGOImagePrefix)
spec.Parameters[config.LABEL_BACKREST_COMMAND] = crv1.PgtaskBackrestBackup
spec.Parameters[config.LABEL_BACKREST_OPTS] = backupOpts
spec.Parameters[config.LABEL_BACKREST_STORAGE_TYPE] = cluster.Spec.UserLabels[config.LABEL_BACKREST_STORAGE_TYPE]
// Get 'true' or 'false' for setting the pgBackRest S3 verify TLS value
spec.Parameters[config.LABEL_BACKREST_S3_VERIFY_TLS] = operator.GetS3VerifyTLSSetting(cluster)
for k, v := range params {
spec.Parameters[k] = v
}
newInstance = &crv1.Pgtask{
ObjectMeta: metav1.ObjectMeta{
Name: taskName,
},
Spec: spec,
}
newInstance.ObjectMeta.Labels = make(map[string]string)
newInstance.ObjectMeta.Labels[config.LABEL_PG_CLUSTER] = cluster.Name
newInstance.ObjectMeta.Labels[config.LABEL_PG_CLUSTER_IDENTIFIER] = cluster.ObjectMeta.Labels[config.LABEL_PG_CLUSTER_IDENTIFIER]
newInstance.ObjectMeta.Labels[config.LABEL_PGOUSER] = cluster.ObjectMeta.Labels[config.LABEL_PGOUSER]
_, err = clientset.CrunchydataV1().Pgtasks(cluster.Namespace).Create(newInstance)
if err != nil {
log.Error(err)
return nil, err
}
return newInstance, nil
}
// CleanBackupResources is responsible for cleaning up Kubernetes resources from a previous
// pgBackRest backup. Specifically, this function deletes the pgptask and job associate with a
// previous pgBackRest backup for the cluster.
func CleanBackupResources(clientset kubeapi.Interface, namespace, clusterName string) error {
taskName := "backrest-backup-" + clusterName
err := clientset.CrunchydataV1().Pgtasks(namespace).Delete(taskName, &metav1.DeleteOptions{})
if err != nil && !kubeapi.IsNotFound(err) {
log.Error(err)
return err
}
//remove previous backup job
selector := config.LABEL_BACKREST_COMMAND + "=" + crv1.PgtaskBackrestBackup + "," +
config.LABEL_PG_CLUSTER + "=" + clusterName + "," + config.LABEL_BACKREST + "=true"
deletePropagation := metav1.DeletePropagationForeground
err = clientset.
BatchV1().Jobs(namespace).
DeleteCollection(
&metav1.DeleteOptions{PropagationPolicy: &deletePropagation},
metav1.ListOptions{LabelSelector: selector})
if err != nil {
log.Error(err)
}
timeout := time.After(30 * time.Second)
tick := time.NewTicker(1 * time.Second)
defer tick.Stop()
for {
select {
case <-timeout:
return fmt.Errorf("Timed out waiting for deletion of pgBackRest backup job for "+
"cluster %s", clusterName)
case <-tick.C:
jobList, err := clientset.
BatchV1().Jobs(namespace).
List(metav1.ListOptions{LabelSelector: selector})
if err != nil {
log.Error(err)
return err
}
if len(jobList.Items) == 0 {
return nil
}
}
}
}
// getCommandOptsFromPod adds command line options from the primary pod to a backrest job.
// If not already specified in the command options provided in the pgtask, add the IP of the
// primary pod as the value for the "--db-host" parameter. This will ensure direct
// communication between the repo pod and the primary via the primary's IP, instead of going
// through the primary pod's service (which could be unreliable). also if not already specified
// in the command options provided in the pgtask, then lookup the primary pod for the cluster
// and add the PGDATA dir of the pod as the value for the "--db-path" parameter
func getCommandOptsFromPod(clientset kubernetes.Interface, task *crv1.Pgtask,
namespace string) (commandOpts string, err error) {
// lookup the primary pod in order to determine the IP of the primary and the PGDATA directory for
// the current primaty
selector := fmt.Sprintf("%s=%s,%s in (%s,%s)", config.LABEL_PG_CLUSTER,
task.Spec.Parameters[config.LABEL_PG_CLUSTER], config.LABEL_PGHA_ROLE,
"promoted", config.LABEL_PGHA_ROLE_PRIMARY)
options := metav1.ListOptions{
FieldSelector: fields.OneTermEqualSelector("status.phase", string(v1.PodRunning)).String(),
LabelSelector: selector,
}
// only consider pods that are running
pods, err := clientset.CoreV1().Pods(namespace).List(options)
if err != nil {
return
} else if len(pods.Items) > 1 {
err = fmt.Errorf("More than one primary found when creating backrest job %s",
task.Spec.Parameters[config.LABEL_JOB_NAME])
return
} else if len(pods.Items) == 0 {
err = fmt.Errorf("Unable to find primary when creating backrest job %s",
task.Spec.Parameters[config.LABEL_JOB_NAME])
return
}
pod := pods.Items[0]
var cmdOpts []string
if !backrestPgHostRegex.MatchString(task.Spec.Parameters[config.LABEL_BACKREST_OPTS]) {
cmdOpts = append(cmdOpts, fmt.Sprintf("--db-host=%s", pod.Status.PodIP))
}
if !backrestPgPathRegex.MatchString(task.Spec.Parameters[config.LABEL_BACKREST_OPTS]) {
var podDbPath string
for _, envVar := range pod.Spec.Containers[0].Env {
if envVar.Name == "PGBACKREST_DB_PATH" {
podDbPath = envVar.Value
break
}
}
if podDbPath != "" {
cmdOpts = append(cmdOpts, fmt.Sprintf("--db-path=%s", podDbPath))
} else {
log.Errorf("Unable to find PGBACKREST_DB_PATH on primary pod %s for backrest job %s",
pod.Name, task.Spec.Parameters[config.LABEL_JOB_NAME])
return
}
}
// join options using a space
commandOpts = strings.Join(cmdOpts, " ")
return
}