forked from libopenstorage/stork
-
Notifications
You must be signed in to change notification settings - Fork 0
/
applicationbackupschedule.go
349 lines (322 loc) · 12.7 KB
/
applicationbackupschedule.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
package controllers
import (
"context"
"fmt"
"reflect"
"strings"
"time"
"github.com/libopenstorage/stork/pkg/apis/stork"
stork_api "github.com/libopenstorage/stork/pkg/apis/stork/v1alpha1"
"github.com/libopenstorage/stork/pkg/controller"
"github.com/libopenstorage/stork/pkg/log"
"github.com/libopenstorage/stork/pkg/schedule"
"github.com/operator-framework/operator-sdk/pkg/sdk"
"github.com/portworx/sched-ops/k8s"
"k8s.io/api/core/v1"
apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
"k8s.io/apimachinery/pkg/api/errors"
meta "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/tools/record"
)
const (
nameTimeSuffixFormat string = "2006-01-02-150405"
annotationPrefix = "stork.libopenstorage.org/"
// ApplicationBackupScheduleNameAnnotation Annotation used to specify the name of schedule that
// created the backup
ApplicationBackupScheduleNameAnnotation = annotationPrefix + "applicationBackupScheduleName"
// ApplicationBackupSchedulePolicyTypeAnnotation Annotation used to specify the type of the
// policy that triggered the backup
ApplicationBackupSchedulePolicyTypeAnnotation = annotationPrefix + "applicationBackupSchedulePolicyType"
)
// ApplicationBackupScheduleController reconciles ApplicationBackupSchedule objects
type ApplicationBackupScheduleController struct {
Recorder record.EventRecorder
}
// Init Initialize the backup schedule controller
func (s *ApplicationBackupScheduleController) Init() error {
err := s.createCRD()
if err != nil {
return err
}
return controller.Register(
&schema.GroupVersionKind{
Group: stork.GroupName,
Version: stork_api.SchemeGroupVersion.Version,
Kind: reflect.TypeOf(stork_api.ApplicationBackupSchedule{}).Name(),
},
"",
1*time.Minute,
s)
}
// Handle updates for ApplicationBackupSchedule objects
func (s *ApplicationBackupScheduleController) Handle(ctx context.Context, event sdk.Event) error {
switch o := event.Object.(type) {
case *stork_api.ApplicationBackupSchedule:
backupSchedule := o
// Nothing to do for delete
if event.Deleted {
return nil
}
s.setDefaults(backupSchedule)
// First update the status of any pending backups
err := s.updateApplicationBackupStatus(backupSchedule)
if err != nil {
msg := fmt.Sprintf("Error updating backup status: %v", err)
s.Recorder.Event(backupSchedule,
v1.EventTypeWarning,
string(stork_api.ApplicationBackupStatusFailed),
msg)
log.ApplicationBackupScheduleLog(backupSchedule).Error(msg)
return err
}
if backupSchedule.Spec.Suspend == nil || !*backupSchedule.Spec.Suspend {
// Then check if any of the policies require a trigger
policyType, start, err := s.shouldStartApplicationBackup(backupSchedule)
if err != nil {
msg := fmt.Sprintf("Error checking if backup should be triggered: %v", err)
s.Recorder.Event(backupSchedule,
v1.EventTypeWarning,
string(stork_api.ApplicationBackupStatusFailed),
msg)
log.ApplicationBackupScheduleLog(backupSchedule).Error(msg)
return nil
}
// Start a backup for a policy if required
if start {
err := s.startApplicationBackup(backupSchedule, policyType)
if err != nil {
msg := fmt.Sprintf("Error triggering backup for schedule(%v): %v", policyType, err)
s.Recorder.Event(backupSchedule,
v1.EventTypeWarning,
string(stork_api.ApplicationBackupStatusFailed),
msg)
log.ApplicationBackupScheduleLog(backupSchedule).Error(msg)
return err
}
}
}
// Finally, prune any old backups that were triggered for this
// schedule
err = s.pruneApplicationBackups(backupSchedule)
if err != nil {
msg := fmt.Sprintf("Error pruning old backups: %v", err)
s.Recorder.Event(backupSchedule,
v1.EventTypeWarning,
string(stork_api.ApplicationBackupStatusFailed),
msg)
log.ApplicationBackupScheduleLog(backupSchedule).Error(msg)
return err
}
}
return nil
}
func (s *ApplicationBackupScheduleController) setDefaults(backupSchedule *stork_api.ApplicationBackupSchedule) {
if backupSchedule.Spec.ReclaimPolicy == "" {
backupSchedule.Spec.ReclaimPolicy = stork_api.ReclaimPolicyDelete
}
}
func getApplicationBackupStatus(name string, namespace string) (stork_api.ApplicationBackupStatusType, error) {
backup, err := k8s.Instance().GetApplicationBackup(name, namespace)
if err != nil {
return stork_api.ApplicationBackupStatusFailed, err
}
return backup.Status.Status, nil
}
func (s *ApplicationBackupScheduleController) updateApplicationBackupStatus(backupSchedule *stork_api.ApplicationBackupSchedule) error {
updated := false
for _, policyApplicationBackup := range backupSchedule.Status.Items {
for _, backup := range policyApplicationBackup {
// Get the updated status if we see it as not completed
if !s.isApplicationBackupComplete(backup.Status) {
pendingApplicationBackupStatus, err := getApplicationBackupStatus(backup.Name, backupSchedule.Namespace)
if err != nil {
s.Recorder.Event(backupSchedule,
v1.EventTypeWarning,
string(stork_api.ApplicationBackupStatusFailed),
fmt.Sprintf("Error getting status of backup %v: %v", backup.Name, err))
// If there was an error other than not found move to
// the next one. Otherwise we want to mark it as failed
// since the applicationbackup object is no longer
// present
if !errors.IsNotFound(err) {
continue
}
}
// Check again and update the status if it is completed
backup.Status = pendingApplicationBackupStatus
if s.isApplicationBackupComplete(backup.Status) {
backup.FinishTimestamp = meta.NewTime(schedule.GetCurrentTime())
if pendingApplicationBackupStatus == stork_api.ApplicationBackupStatusSuccessful {
s.Recorder.Event(backupSchedule,
v1.EventTypeNormal,
string(stork_api.ApplicationBackupStatusSuccessful),
fmt.Sprintf("Scheduled backup (%v) completed successfully", backup.Name))
} else {
s.Recorder.Event(backupSchedule,
v1.EventTypeWarning,
string(stork_api.ApplicationBackupStatusFailed),
fmt.Sprintf("Scheduled backup (%v) failed", backup.Name))
}
}
updated = true
}
}
}
if updated {
err := sdk.Update(backupSchedule)
if err != nil {
return err
}
}
return nil
}
func (s *ApplicationBackupScheduleController) isApplicationBackupComplete(status stork_api.ApplicationBackupStatusType) bool {
return status == stork_api.ApplicationBackupStatusFailed ||
status == stork_api.ApplicationBackupStatusPartialSuccess ||
status == stork_api.ApplicationBackupStatusSuccessful
}
func (s *ApplicationBackupScheduleController) shouldStartApplicationBackup(backupSchedule *stork_api.ApplicationBackupSchedule) (stork_api.SchedulePolicyType, bool, error) {
// Don't trigger a new backup if one is already in progress
for _, policyType := range stork_api.GetValidSchedulePolicyTypes() {
policyApplicationBackup, present := backupSchedule.Status.Items[policyType]
if present {
for _, backup := range policyApplicationBackup {
if !s.isApplicationBackupComplete(backup.Status) {
return stork_api.SchedulePolicyTypeInvalid, false, nil
}
}
}
}
for _, policyType := range stork_api.GetValidSchedulePolicyTypes() {
var latestApplicationBackupTimestamp meta.Time
policyApplicationBackup, present := backupSchedule.Status.Items[policyType]
if present {
for _, backup := range policyApplicationBackup {
if latestApplicationBackupTimestamp.Before(&backup.CreationTimestamp) {
latestApplicationBackupTimestamp = backup.CreationTimestamp
}
}
}
trigger, err := schedule.TriggerRequired(
backupSchedule.Spec.SchedulePolicyName,
policyType,
latestApplicationBackupTimestamp,
)
if err != nil {
return stork_api.SchedulePolicyTypeInvalid, false, err
}
if trigger {
return policyType, true, nil
}
}
return stork_api.SchedulePolicyTypeInvalid, false, nil
}
func (s *ApplicationBackupScheduleController) formatApplicationBackupName(backupSchedule *stork_api.ApplicationBackupSchedule, policyType stork_api.SchedulePolicyType) string {
return strings.Join([]string{backupSchedule.Name, strings.ToLower(string(policyType)), time.Now().Format(nameTimeSuffixFormat)}, "-")
}
func (s *ApplicationBackupScheduleController) startApplicationBackup(backupSchedule *stork_api.ApplicationBackupSchedule, policyType stork_api.SchedulePolicyType) error {
backupName := s.formatApplicationBackupName(backupSchedule, policyType)
if backupSchedule.Status.Items == nil {
backupSchedule.Status.Items = make(map[stork_api.SchedulePolicyType][]*stork_api.ScheduledApplicationBackupStatus)
}
if backupSchedule.Status.Items[policyType] == nil {
backupSchedule.Status.Items[policyType] = make([]*stork_api.ScheduledApplicationBackupStatus, 0)
}
backupSchedule.Status.Items[policyType] = append(backupSchedule.Status.Items[policyType],
&stork_api.ScheduledApplicationBackupStatus{
Name: backupName,
CreationTimestamp: meta.NewTime(schedule.GetCurrentTime()),
Status: stork_api.ApplicationBackupStatusPending,
})
err := sdk.Update(backupSchedule)
if err != nil {
return err
}
backup := &stork_api.ApplicationBackup{
ObjectMeta: meta.ObjectMeta{
Name: backupName,
Namespace: backupSchedule.Namespace,
Annotations: backupSchedule.Annotations,
Labels: backupSchedule.Labels,
},
Spec: backupSchedule.Spec.Template.Spec,
}
if backup.Annotations == nil {
backup.Annotations = make(map[string]string)
}
backup.Annotations[ApplicationBackupScheduleNameAnnotation] = backupSchedule.Name
backup.Annotations[ApplicationBackupSchedulePolicyTypeAnnotation] = string(policyType)
log.ApplicationBackupScheduleLog(backupSchedule).Infof("Starting backup %v", backupName)
// If reclaim policy is set to Delete, this will delete the backups
// created by this backupschedule when the schedule object is deleted
if backupSchedule.Spec.ReclaimPolicy == stork_api.ReclaimPolicyDelete {
backup.OwnerReferences = []meta.OwnerReference{
{
Name: backupSchedule.Name,
UID: backupSchedule.UID,
Kind: backupSchedule.GetObjectKind().GroupVersionKind().Kind,
APIVersion: backupSchedule.GetObjectKind().GroupVersionKind().GroupVersion().String(),
},
}
}
_, err = k8s.Instance().CreateApplicationBackup(backup)
return err
}
func (s *ApplicationBackupScheduleController) pruneApplicationBackups(backupSchedule *stork_api.ApplicationBackupSchedule) error {
for policyType, policyApplicationBackup := range backupSchedule.Status.Items {
numApplicationBackups := len(policyApplicationBackup)
deleteBefore := 0
retainNum, err := schedule.GetRetain(backupSchedule.Spec.SchedulePolicyName, policyType)
if err != nil {
return err
}
numReady := 0
// Keep up to retainNum successful backup statuses and all failed backups
// until there is a successful one
if numApplicationBackups > int(retainNum) {
// Start from the end and find the retainNum successful backups
for i := range policyApplicationBackup {
if policyApplicationBackup[(numApplicationBackups-1-i)].Status == stork_api.ApplicationBackupStatusSuccessful {
numReady++
if numReady > int(retainNum) {
deleteBefore = numApplicationBackups - i
break
}
}
}
failedDeletes := make([]*stork_api.ScheduledApplicationBackupStatus, 0)
if numReady > int(retainNum) {
for i := 0; i < deleteBefore; i++ {
err := k8s.Instance().DeleteApplicationBackup(policyApplicationBackup[i].Name, backupSchedule.Namespace)
if err != nil && !errors.IsNotFound(err) {
log.ApplicationBackupScheduleLog(backupSchedule).Warnf("Error deleting %v: %v", policyApplicationBackup[i].Name, err)
// Keep a track of the failed deletes
failedDeletes = append(failedDeletes, policyApplicationBackup[i])
}
}
}
// Remove all the ones we tried to delete above
backupSchedule.Status.Items[policyType] = policyApplicationBackup[deleteBefore:]
// And re-add the ones that failed so that we don't lose track
// of them
backupSchedule.Status.Items[policyType] = append(failedDeletes, backupSchedule.Status.Items[policyType]...)
}
}
return sdk.Update(backupSchedule)
}
func (s *ApplicationBackupScheduleController) createCRD() error {
resource := k8s.CustomResource{
Name: stork_api.ApplicationBackupScheduleResourceName,
Plural: stork_api.ApplicationBackupScheduleResourcePlural,
Group: stork.GroupName,
Version: stork_api.SchemeGroupVersion.Version,
Scope: apiextensionsv1beta1.NamespaceScoped,
Kind: reflect.TypeOf(stork_api.ApplicationBackupSchedule{}).Name(),
}
err := k8s.Instance().CreateCRD(resource)
if err != nil && !errors.IsAlreadyExists(err) {
return err
}
return k8s.Instance().ValidateCRD(resource, validateCRDTimeout, validateCRDInterval)
}