Merge branch 'main' into issue-fix-6695

This commit is contained in:
Lyndon-Li
2023-11-29 15:12:21 +08:00
80 changed files with 1970 additions and 248 deletions
+1
View File
@@ -0,0 +1 @@
Change controller-runtime List option from MatchingFields to ListOptions
+1
View File
@@ -0,0 +1 @@
improve discoveryHelper.Refresh() in restore
+1
View File
@@ -0,0 +1 @@
Add hooks status to backup/restore CR
+1
View File
@@ -0,0 +1 @@
Node agent restart enhancement
+1
View File
@@ -0,0 +1 @@
Treat namespace as a regular restorable item
+1
View File
@@ -0,0 +1 @@
Add more linters part 2.
@@ -544,6 +544,22 @@ spec:
description: FormatVersion is the backup format version, including
major, minor, and patch version.
type: string
hookStatus:
description: HookStatus contains information about the status of the
hooks.
nullable: true
properties:
hooksAttempted:
description: HooksAttempted is the total number of attempted hooks
Specifically, HooksAttempted represents the number of hooks
that failed to execute and the number of hooks that executed
successfully.
type: integer
hooksFailed:
description: HooksFailed is the total number of hooks which ended
with an error
type: integer
type: object
phase:
description: Phase is the current state of the Backup.
enum:
@@ -440,6 +440,22 @@ spec:
description: FailureReason is an error that caused the entire restore
to fail.
type: string
hookStatus:
description: HookStatus contains information about the status of the
hooks.
nullable: true
properties:
hooksAttempted:
description: HooksAttempted is the total number of attempted hooks
Specifically, HooksAttempted represents the number of hooks
that failed to execute and the number of hooks that executed
successfully.
type: integer
hooksFailed:
description: HooksFailed is the total number of hooks which ended
with an error
type: integer
type: object
phase:
description: Phase is the current state of the Restore
enum:
File diff suppressed because one or more lines are too long
+6
View File
@@ -326,6 +326,12 @@ linters:
- unused
- usestdlibvars
- whitespace
- dupword
- errchkjson
- ginkgolinter
- nilerr
- noctx
- nolintlint
fast: false
+1 -1
View File
@@ -56,7 +56,7 @@ RUN wget --quiet https://github.com/goreleaser/goreleaser/releases/download/v1.1
chmod +x /usr/bin/goreleaser
# get golangci-lint
RUN curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.51.0
RUN curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.54.2
# install kubectl
RUN curl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/linux/amd64/kubectl
+137
View File
@@ -0,0 +1,137 @@
/*
Copyright 2020 the Velero contributors.
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 hook
import "sync"
const (
HookSourceAnnotation = "annotation"
HookSourceSpec = "spec"
)
// hookTrackerKey identifies a backup/restore hook
type hookTrackerKey struct {
// PodNamespace indicates the namespace of pod where hooks are executed.
// For hooks specified in the backup/restore spec, this field is the namespace of an applicable pod.
// For hooks specified in pod annotation, this field is the namespace of pod where hooks are annotated.
podNamespace string
// PodName indicates the pod where hooks are executed.
// For hooks specified in the backup/restore spec, this field is an applicable pod name.
// For hooks specified in pod annotation, this field is the pod where hooks are annotated.
podName string
// HookPhase is only for backup hooks, for restore hooks, this field is empty.
hookPhase hookPhase
// HookName is only for hooks specified in the backup/restore spec.
// For hooks specified in pod annotation, this field is empty or "<from-annotation>".
hookName string
// HookSource indicates where hooks come from.
hookSource string
// Container indicates the container hooks use.
// For hooks specified in the backup/restore spec, the container might be the same under different hookName.
container string
}
// hookTrackerVal records the execution status of a specific hook.
// hookTrackerVal is extensible to accommodate additional fields as needs develop.
type hookTrackerVal struct {
// HookFailed indicates if hook failed to execute.
hookFailed bool
// hookExecuted indicates if hook already execute.
hookExecuted bool
}
// HookTracker tracks all hooks' execution status
type HookTracker struct {
lock *sync.RWMutex
tracker map[hookTrackerKey]hookTrackerVal
}
// NewHookTracker creates a hookTracker.
func NewHookTracker() *HookTracker {
return &HookTracker{
lock: &sync.RWMutex{},
tracker: make(map[hookTrackerKey]hookTrackerVal),
}
}
// Add adds a hook to the tracker
func (ht *HookTracker) Add(podNamespace, podName, container, source, hookName string, hookPhase hookPhase) {
ht.lock.Lock()
defer ht.lock.Unlock()
key := hookTrackerKey{
podNamespace: podNamespace,
podName: podName,
hookSource: source,
container: container,
hookPhase: hookPhase,
hookName: hookName,
}
if _, ok := ht.tracker[key]; !ok {
ht.tracker[key] = hookTrackerVal{
hookFailed: false,
hookExecuted: false,
}
}
}
// Record records the hook's execution status
func (ht *HookTracker) Record(podNamespace, podName, container, source, hookName string, hookPhase hookPhase, hookFailed bool) {
ht.lock.Lock()
defer ht.lock.Unlock()
key := hookTrackerKey{
podNamespace: podNamespace,
podName: podName,
hookSource: source,
container: container,
hookPhase: hookPhase,
hookName: hookName,
}
if _, ok := ht.tracker[key]; ok {
ht.tracker[key] = hookTrackerVal{
hookFailed: hookFailed,
hookExecuted: true,
}
}
}
// Stat calculates the number of attempted hooks and failed hooks
func (ht *HookTracker) Stat() (hookAttemptedCnt int, hookFailed int) {
ht.lock.RLock()
defer ht.lock.RUnlock()
for _, hookInfo := range ht.tracker {
if hookInfo.hookExecuted {
hookAttemptedCnt++
if hookInfo.hookFailed {
hookFailed++
}
}
}
return
}
// GetTracker gets the tracker inside HookTracker
func (ht *HookTracker) GetTracker() map[hookTrackerKey]hookTrackerVal {
ht.lock.RLock()
defer ht.lock.RUnlock()
return ht.tracker
}
+88
View File
@@ -0,0 +1,88 @@
/*
Copyright 2020 the Velero contributors.
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 hook
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestNewHookTracker(t *testing.T) {
tracker := NewHookTracker()
assert.NotNil(t, tracker)
assert.Empty(t, tracker.tracker)
}
func TestHookTracker_Add(t *testing.T) {
tracker := NewHookTracker()
tracker.Add("ns1", "pod1", "container1", HookSourceAnnotation, "h1", PhasePre)
key := hookTrackerKey{
podNamespace: "ns1",
podName: "pod1",
container: "container1",
hookPhase: PhasePre,
hookSource: HookSourceAnnotation,
hookName: "h1",
}
_, ok := tracker.tracker[key]
assert.True(t, ok)
}
func TestHookTracker_Record(t *testing.T) {
tracker := NewHookTracker()
tracker.Add("ns1", "pod1", "container1", HookSourceAnnotation, "h1", PhasePre)
tracker.Record("ns1", "pod1", "container1", HookSourceAnnotation, "h1", PhasePre, true)
key := hookTrackerKey{
podNamespace: "ns1",
podName: "pod1",
container: "container1",
hookPhase: PhasePre,
hookSource: HookSourceAnnotation,
hookName: "h1",
}
info := tracker.tracker[key]
assert.True(t, info.hookFailed)
}
func TestHookTracker_Stat(t *testing.T) {
tracker := NewHookTracker()
tracker.Add("ns1", "pod1", "container1", HookSourceAnnotation, "h1", PhasePre)
tracker.Add("ns2", "pod2", "container1", HookSourceAnnotation, "h2", PhasePre)
tracker.Record("ns1", "pod1", "container1", HookSourceAnnotation, "h1", PhasePre, true)
attempted, failed := tracker.Stat()
assert.Equal(t, 1, attempted)
assert.Equal(t, 1, failed)
}
func TestHookTracker_Get(t *testing.T) {
tracker := NewHookTracker()
tracker.Add("ns1", "pod1", "container1", HookSourceAnnotation, "h1", PhasePre)
tr := tracker.GetTracker()
assert.NotNil(t, tr)
t.Logf("tracker :%+v", tr)
}
+39 -16
View File
@@ -82,6 +82,7 @@ type ItemHookHandler interface {
obj runtime.Unstructured,
resourceHooks []ResourceHook,
phase hookPhase,
hookTracker *HookTracker,
) error
}
@@ -200,6 +201,7 @@ func (h *DefaultItemHookHandler) HandleHooks(
obj runtime.Unstructured,
resourceHooks []ResourceHook,
phase hookPhase,
hookTracker *HookTracker,
) error {
// We only support hooks on pods right now
if groupResource != kuberesource.Pods {
@@ -221,15 +223,21 @@ func (h *DefaultItemHookHandler) HandleHooks(
hookFromAnnotations = getPodExecHookFromAnnotations(metadata.GetAnnotations(), "", log)
}
if hookFromAnnotations != nil {
hookTracker.Add(namespace, name, hookFromAnnotations.Container, HookSourceAnnotation, "", phase)
hookLog := log.WithFields(
logrus.Fields{
"hookSource": "annotation",
"hookSource": HookSourceAnnotation,
"hookType": "exec",
"hookPhase": phase,
},
)
hookTracker.Record(namespace, name, hookFromAnnotations.Container, HookSourceAnnotation, "", phase, false)
if err := h.PodCommandExecutor.ExecutePodCommand(hookLog, obj.UnstructuredContent(), namespace, name, "<from-annotation>", hookFromAnnotations); err != nil {
hookLog.WithError(err).Error("Error executing hook")
hookTracker.Record(namespace, name, hookFromAnnotations.Container, HookSourceAnnotation, "", phase, true)
if hookFromAnnotations.OnError == velerov1api.HookErrorModeFail {
return err
}
@@ -240,6 +248,8 @@ func (h *DefaultItemHookHandler) HandleHooks(
labels := labels.Set(metadata.GetLabels())
// Otherwise, check for hooks defined in the backup spec.
// modeFailError records the error from the hook with "Fail" error mode
var modeFailError error
for _, resourceHook := range resourceHooks {
if !resourceHook.Selector.applicableTo(groupResource, namespace, labels) {
continue
@@ -251,21 +261,30 @@ func (h *DefaultItemHookHandler) HandleHooks(
} else {
hooks = resourceHook.Post
}
for _, hook := range hooks {
if groupResource == kuberesource.Pods {
if hook.Exec != nil {
hookLog := log.WithFields(
logrus.Fields{
"hookSource": "backupSpec",
"hookType": "exec",
"hookPhase": phase,
},
)
err := h.PodCommandExecutor.ExecutePodCommand(hookLog, obj.UnstructuredContent(), namespace, name, resourceHook.Name, hook.Exec)
if err != nil {
hookLog.WithError(err).Error("Error executing hook")
if hook.Exec.OnError == velerov1api.HookErrorModeFail {
return err
hookTracker.Add(namespace, name, hook.Exec.Container, HookSourceSpec, resourceHook.Name, phase)
// The remaining hooks will only be executed if modeFailError is nil.
// Otherwise, execution will stop and only hook collection will occur.
if modeFailError == nil {
hookLog := log.WithFields(
logrus.Fields{
"hookSource": HookSourceSpec,
"hookType": "exec",
"hookPhase": phase,
},
)
hookTracker.Record(namespace, name, hook.Exec.Container, HookSourceSpec, resourceHook.Name, phase, false)
err := h.PodCommandExecutor.ExecutePodCommand(hookLog, obj.UnstructuredContent(), namespace, name, resourceHook.Name, hook.Exec)
if err != nil {
hookLog.WithError(err).Error("Error executing hook")
hookTracker.Record(namespace, name, hook.Exec.Container, HookSourceSpec, resourceHook.Name, phase, true)
if hook.Exec.OnError == velerov1api.HookErrorModeFail {
modeFailError = err
}
}
}
}
@@ -273,7 +292,7 @@ func (h *DefaultItemHookHandler) HandleHooks(
}
}
return nil
return modeFailError
}
// NoOpItemHookHandler is the an itemHookHandler for the Finalize controller where hooks don't run
@@ -285,6 +304,7 @@ func (h *NoOpItemHookHandler) HandleHooks(
obj runtime.Unstructured,
resourceHooks []ResourceHook,
phase hookPhase,
hookTracker *HookTracker,
) error {
return nil
}
@@ -514,6 +534,7 @@ func GroupRestoreExecHooks(
resourceRestoreHooks []ResourceRestoreHook,
pod *corev1api.Pod,
log logrus.FieldLogger,
hookTrack *HookTracker,
) (map[string][]PodExecRestoreHook, error) {
byContainer := map[string][]PodExecRestoreHook{}
@@ -530,10 +551,11 @@ func GroupRestoreExecHooks(
if hookFromAnnotation.Container == "" {
hookFromAnnotation.Container = pod.Spec.Containers[0].Name
}
hookTrack.Add(metadata.GetNamespace(), metadata.GetName(), hookFromAnnotation.Container, HookSourceAnnotation, "<from-annotation>", hookPhase(""))
byContainer[hookFromAnnotation.Container] = []PodExecRestoreHook{
{
HookName: "<from-annotation>",
HookSource: "annotation",
HookSource: HookSourceAnnotation,
Hook: *hookFromAnnotation,
},
}
@@ -554,7 +576,7 @@ func GroupRestoreExecHooks(
named := PodExecRestoreHook{
HookName: rrh.Name,
Hook: *rh.Exec,
HookSource: "backupSpec",
HookSource: HookSourceSpec,
}
// default to false if attr WaitForReady not set
if named.Hook.WaitForReady == nil {
@@ -564,6 +586,7 @@ func GroupRestoreExecHooks(
if named.Hook.Container == "" {
named.Hook.Container = pod.Spec.Containers[0].Name
}
hookTrack.Add(metadata.GetNamespace(), metadata.GetName(), named.Hook.Container, HookSourceSpec, rrh.Name, hookPhase(""))
byContainer[named.Hook.Container] = append(byContainer[named.Hook.Container], named)
}
}
+507 -12
View File
@@ -108,6 +108,7 @@ func TestHandleHooksSkips(t *testing.T) {
},
}
hookTracker := NewHookTracker()
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
podCommandExecutor := &velerotest.MockPodCommandExecutor{}
@@ -118,7 +119,7 @@ func TestHandleHooksSkips(t *testing.T) {
}
groupResource := schema.ParseGroupResource(test.groupResource)
err := h.HandleHooks(velerotest.NewLogger(), groupResource, test.item, test.hooks, PhasePre)
err := h.HandleHooks(velerotest.NewLogger(), groupResource, test.item, test.hooks, PhasePre, hookTracker)
assert.NoError(t, err)
})
}
@@ -485,7 +486,8 @@ func TestHandleHooks(t *testing.T) {
}
groupResource := schema.ParseGroupResource(test.groupResource)
err := h.HandleHooks(velerotest.NewLogger(), groupResource, test.item, test.hooks, test.phase)
hookTracker := NewHookTracker()
err := h.HandleHooks(velerotest.NewLogger(), groupResource, test.item, test.hooks, test.phase, hookTracker)
if test.expectedError != nil {
assert.EqualError(t, err, test.expectedError.Error())
@@ -861,7 +863,7 @@ func TestGroupRestoreExecHooks(t *testing.T) {
"container1": {
{
HookName: "<from-annotation>",
HookSource: "annotation",
HookSource: HookSourceAnnotation,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -892,7 +894,7 @@ func TestGroupRestoreExecHooks(t *testing.T) {
"container1": {
{
HookName: "<from-annotation>",
HookSource: "annotation",
HookSource: HookSourceAnnotation,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -933,7 +935,7 @@ func TestGroupRestoreExecHooks(t *testing.T) {
"container1": {
{
HookName: "hook1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -973,7 +975,7 @@ func TestGroupRestoreExecHooks(t *testing.T) {
"container1": {
{
HookName: "hook1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -1021,7 +1023,7 @@ func TestGroupRestoreExecHooks(t *testing.T) {
"container1": {
{
HookName: "<from-annotation>",
HookSource: "annotation",
HookSource: HookSourceAnnotation,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -1140,7 +1142,7 @@ func TestGroupRestoreExecHooks(t *testing.T) {
"container1": {
{
HookName: "hook1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -1152,7 +1154,7 @@ func TestGroupRestoreExecHooks(t *testing.T) {
},
{
HookName: "hook1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/bar"},
@@ -1164,7 +1166,7 @@ func TestGroupRestoreExecHooks(t *testing.T) {
},
{
HookName: "hook2",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/aaa"},
@@ -1178,7 +1180,7 @@ func TestGroupRestoreExecHooks(t *testing.T) {
"container2": {
{
HookName: "hook1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container2",
Command: []string{"/usr/bin/baz"},
@@ -1192,9 +1194,11 @@ func TestGroupRestoreExecHooks(t *testing.T) {
},
},
}
hookTracker := NewHookTracker()
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
actual, err := GroupRestoreExecHooks(tc.resourceRestoreHooks, tc.pod, velerotest.NewLogger())
actual, err := GroupRestoreExecHooks(tc.resourceRestoreHooks, tc.pod, velerotest.NewLogger(), hookTracker)
assert.Nil(t, err)
assert.Equal(t, tc.expected, actual)
})
@@ -1983,3 +1987,494 @@ func TestValidateContainer(t *testing.T) {
// noCommand string should return expected error as result.
assert.Equal(t, expectedError, ValidateContainer([]byte(noCommand)))
}
func TestBackupHookTracker(t *testing.T) {
type podWithHook struct {
item runtime.Unstructured
hooks []ResourceHook
hookErrorsByContainer map[string]error
expectedPodHook *velerov1api.ExecHook
expectedPodHookError error
expectedError error
}
test1 := []struct {
name string
phase hookPhase
groupResource string
pods []podWithHook
hookTracker *HookTracker
expectedHookAttempted int
expectedHookFailed int
}{
{
name: "a pod with spec hooks, no error",
phase: PhasePre,
groupResource: "pods",
hookTracker: NewHookTracker(),
expectedHookAttempted: 2,
expectedHookFailed: 0,
pods: []podWithHook{
{
item: velerotest.UnstructuredOrDie(`
{
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"namespace": "ns",
"name": "name"
}
}`),
hooks: []ResourceHook{
{
Name: "hook1",
Pre: []velerov1api.BackupResourceHook{
{
Exec: &velerov1api.ExecHook{
Container: "1a",
Command: []string{"pre-1a"},
},
},
{
Exec: &velerov1api.ExecHook{
Container: "1b",
Command: []string{"pre-1b"},
},
},
},
},
},
},
},
},
{
name: "a pod with spec hooks and same container under different hook name, no error",
phase: PhasePre,
groupResource: "pods",
hookTracker: NewHookTracker(),
expectedHookAttempted: 4,
expectedHookFailed: 0,
pods: []podWithHook{
{
item: velerotest.UnstructuredOrDie(`
{
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"namespace": "ns",
"name": "name"
}
}`),
hooks: []ResourceHook{
{
Name: "hook1",
Pre: []velerov1api.BackupResourceHook{
{
Exec: &velerov1api.ExecHook{
Container: "1a",
Command: []string{"pre-1a"},
},
},
{
Exec: &velerov1api.ExecHook{
Container: "1b",
Command: []string{"pre-1b"},
},
},
},
},
{
Name: "hook2",
Pre: []velerov1api.BackupResourceHook{
{
Exec: &velerov1api.ExecHook{
Container: "1a",
Command: []string{"2a"},
},
},
{
Exec: &velerov1api.ExecHook{
Container: "2b",
Command: []string{"2b"},
},
},
},
},
},
},
},
},
{
name: "a pod with spec hooks, on error=fail",
phase: PhasePre,
groupResource: "pods",
hookTracker: NewHookTracker(),
expectedHookAttempted: 3,
expectedHookFailed: 2,
pods: []podWithHook{
{
item: velerotest.UnstructuredOrDie(`
{
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"namespace": "ns",
"name": "name"
}
}`),
hooks: []ResourceHook{
{
Name: "hook1",
Pre: []velerov1api.BackupResourceHook{
{
Exec: &velerov1api.ExecHook{
Container: "1a",
Command: []string{"1a"},
OnError: velerov1api.HookErrorModeContinue,
},
},
{
Exec: &velerov1api.ExecHook{
Container: "1b",
Command: []string{"1b"},
},
},
},
},
{
Name: "hook2",
Pre: []velerov1api.BackupResourceHook{
{
Exec: &velerov1api.ExecHook{
Container: "2",
Command: []string{"2"},
OnError: velerov1api.HookErrorModeFail,
},
},
},
},
{
Name: "hook3",
Pre: []velerov1api.BackupResourceHook{
{
Exec: &velerov1api.ExecHook{
Container: "3",
Command: []string{"3"},
},
},
},
},
},
hookErrorsByContainer: map[string]error{
"1a": errors.New("1a error, but continue"),
"2": errors.New("2 error, fail"),
},
},
},
},
{
name: "a pod with annotation and spec hooks",
phase: PhasePre,
groupResource: "pods",
hookTracker: NewHookTracker(),
expectedHookAttempted: 1,
expectedHookFailed: 0,
pods: []podWithHook{
{
item: velerotest.UnstructuredOrDie(`
{
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"namespace": "ns",
"name": "name",
"annotations": {
"hook.backup.velero.io/container": "c",
"hook.backup.velero.io/command": "/bin/ls"
}
}
}`),
expectedPodHook: &velerov1api.ExecHook{
Container: "c",
Command: []string{"/bin/ls"},
},
hooks: []ResourceHook{
{
Name: "hook1",
Pre: []velerov1api.BackupResourceHook{
{
Exec: &velerov1api.ExecHook{
Container: "1a",
Command: []string{"1a"},
OnError: velerov1api.HookErrorModeContinue,
},
},
{
Exec: &velerov1api.ExecHook{
Container: "1b",
Command: []string{"1b"},
},
},
},
},
},
},
},
},
{
name: "a pod with annotation, on error=fail",
phase: PhasePre,
groupResource: "pods",
hookTracker: NewHookTracker(),
expectedHookAttempted: 1,
expectedHookFailed: 1,
pods: []podWithHook{
{
item: velerotest.UnstructuredOrDie(`
{
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"namespace": "ns",
"name": "name",
"annotations": {
"hook.backup.velero.io/container": "c",
"hook.backup.velero.io/command": "/bin/ls",
"hook.backup.velero.io/on-error": "Fail"
}
}
}`),
expectedPodHook: &velerov1api.ExecHook{
Container: "c",
Command: []string{"/bin/ls"},
OnError: velerov1api.HookErrorModeFail,
},
expectedPodHookError: errors.New("pod hook error"),
},
},
},
{
name: "two pods, one with annotation, the other with spec",
phase: PhasePre,
groupResource: "pods",
hookTracker: NewHookTracker(),
expectedHookAttempted: 3,
expectedHookFailed: 1,
pods: []podWithHook{
{
item: velerotest.UnstructuredOrDie(`
{
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"namespace": "ns",
"name": "name",
"annotations": {
"hook.backup.velero.io/container": "c",
"hook.backup.velero.io/command": "/bin/ls",
"hook.backup.velero.io/on-error": "Fail"
}
}
}`),
expectedPodHook: &velerov1api.ExecHook{
Container: "c",
Command: []string{"/bin/ls"},
OnError: velerov1api.HookErrorModeFail,
},
expectedPodHookError: errors.New("pod hook error"),
},
{
item: velerotest.UnstructuredOrDie(`
{
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"namespace": "ns",
"name": "name"
}
}`),
hooks: []ResourceHook{
{
Name: "hook1",
Pre: []velerov1api.BackupResourceHook{
{
Exec: &velerov1api.ExecHook{
Container: "1a",
Command: []string{"pre-1a"},
},
},
{
Exec: &velerov1api.ExecHook{
Container: "1b",
Command: []string{"pre-1b"},
},
},
},
},
},
},
},
},
}
for _, test := range test1 {
t.Run(test.name, func(t *testing.T) {
podCommandExecutor := &velerotest.MockPodCommandExecutor{}
defer podCommandExecutor.AssertExpectations(t)
h := &DefaultItemHookHandler{
PodCommandExecutor: podCommandExecutor,
}
groupResource := schema.ParseGroupResource(test.groupResource)
hookTracker := test.hookTracker
for _, pod := range test.pods {
if pod.expectedPodHook != nil {
podCommandExecutor.On("ExecutePodCommand", mock.Anything, pod.item.UnstructuredContent(), "ns", "name", "<from-annotation>", pod.expectedPodHook).Return(pod.expectedPodHookError)
} else {
hookLoop:
for _, resourceHook := range pod.hooks {
for _, hook := range resourceHook.Pre {
hookError := pod.hookErrorsByContainer[hook.Exec.Container]
podCommandExecutor.On("ExecutePodCommand", mock.Anything, pod.item.UnstructuredContent(), "ns", "name", resourceHook.Name, hook.Exec).Return(hookError)
if hookError != nil && hook.Exec.OnError == velerov1api.HookErrorModeFail {
break hookLoop
}
}
for _, hook := range resourceHook.Post {
hookError := pod.hookErrorsByContainer[hook.Exec.Container]
podCommandExecutor.On("ExecutePodCommand", mock.Anything, pod.item.UnstructuredContent(), "ns", "name", resourceHook.Name, hook.Exec).Return(hookError)
if hookError != nil && hook.Exec.OnError == velerov1api.HookErrorModeFail {
break hookLoop
}
}
}
}
h.HandleHooks(velerotest.NewLogger(), groupResource, pod.item, pod.hooks, test.phase, hookTracker)
}
actualAtemptted, actualFailed := hookTracker.Stat()
assert.Equal(t, test.expectedHookAttempted, actualAtemptted)
assert.Equal(t, test.expectedHookFailed, actualFailed)
})
}
}
func TestRestoreHookTrackerAdd(t *testing.T) {
testCases := []struct {
name string
resourceRestoreHooks []ResourceRestoreHook
pod *corev1api.Pod
hookTracker *HookTracker
expectedCnt int
}{
{
name: "neither spec hooks nor annotations hooks are set",
resourceRestoreHooks: nil,
pod: builder.ForPod("default", "my-pod").Result(),
hookTracker: NewHookTracker(),
expectedCnt: 0,
},
{
name: "a hook specified in pod annotation",
resourceRestoreHooks: nil,
pod: builder.ForPod("default", "my-pod").
ObjectMeta(builder.WithAnnotations(
podRestoreHookCommandAnnotationKey, "/usr/bin/foo",
podRestoreHookContainerAnnotationKey, "container1",
podRestoreHookOnErrorAnnotationKey, string(velerov1api.HookErrorModeContinue),
podRestoreHookTimeoutAnnotationKey, "1s",
podRestoreHookWaitTimeoutAnnotationKey, "1m",
podRestoreHookWaitForReadyAnnotationKey, "true",
)).
Containers(&corev1api.Container{
Name: "container1",
}).
Result(),
hookTracker: NewHookTracker(),
expectedCnt: 1,
},
{
name: "two hooks specified in restore spec",
resourceRestoreHooks: []ResourceRestoreHook{
{
Name: "hook1",
Selector: ResourceHookSelector{},
RestoreHooks: []velerov1api.RestoreResourceHook{
{
Exec: &velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
OnError: velerov1api.HookErrorModeContinue,
ExecTimeout: metav1.Duration{Duration: time.Second},
WaitTimeout: metav1.Duration{Duration: time.Minute},
},
},
{
Exec: &velerov1api.ExecRestoreHook{
Container: "container2",
Command: []string{"/usr/bin/foo"},
OnError: velerov1api.HookErrorModeContinue,
ExecTimeout: metav1.Duration{Duration: time.Second},
WaitTimeout: metav1.Duration{Duration: time.Minute},
},
},
},
},
},
pod: builder.ForPod("default", "my-pod").
Containers(&corev1api.Container{
Name: "container1",
}, &corev1api.Container{
Name: "container2",
}).
Result(),
hookTracker: NewHookTracker(),
expectedCnt: 2,
},
{
name: "both spec hooks and annotations hooks are set",
resourceRestoreHooks: []ResourceRestoreHook{
{
Name: "hook1",
Selector: ResourceHookSelector{},
RestoreHooks: []velerov1api.RestoreResourceHook{
{
Exec: &velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo2"},
OnError: velerov1api.HookErrorModeContinue,
ExecTimeout: metav1.Duration{Duration: time.Second},
WaitTimeout: metav1.Duration{Duration: time.Minute},
},
},
},
},
},
pod: builder.ForPod("default", "my-pod").
ObjectMeta(builder.WithAnnotations(
podRestoreHookCommandAnnotationKey, "/usr/bin/foo",
podRestoreHookContainerAnnotationKey, "container1",
podRestoreHookOnErrorAnnotationKey, string(velerov1api.HookErrorModeContinue),
podRestoreHookTimeoutAnnotationKey, "1s",
podRestoreHookWaitTimeoutAnnotationKey, "1m",
podRestoreHookWaitForReadyAnnotationKey, "true",
)).
Containers(&corev1api.Container{
Name: "container1",
}).
Result(),
hookTracker: NewHookTracker(),
expectedCnt: 1,
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
_, _ = GroupRestoreExecHooks(tc.resourceRestoreHooks, tc.pod, velerotest.NewLogger(), tc.hookTracker)
tracker := tc.hookTracker.GetTracker()
assert.Equal(t, tc.expectedCnt, len(tracker))
})
}
}
+8
View File
@@ -39,6 +39,7 @@ type WaitExecHookHandler interface {
log logrus.FieldLogger,
pod *v1.Pod,
byContainer map[string][]PodExecRestoreHook,
hookTrack *HookTracker,
) []error
}
@@ -73,6 +74,7 @@ func (e *DefaultWaitExecHookHandler) HandleHooks(
log logrus.FieldLogger,
pod *v1.Pod,
byContainer map[string][]PodExecRestoreHook,
hookTracker *HookTracker,
) []error {
if pod == nil {
return nil
@@ -164,6 +166,8 @@ func (e *DefaultWaitExecHookHandler) HandleHooks(
err := fmt.Errorf("hook %s in container %s expired before executing", hook.HookName, hook.Hook.Container)
hookLog.Error(err)
errors = append(errors, err)
hookTracker.Record(newPod.Namespace, newPod.Name, hook.Hook.Container, hook.HookSource, hook.HookName, hookPhase(""), true)
if hook.Hook.OnError == velerov1api.HookErrorModeFail {
cancel()
return
@@ -175,10 +179,13 @@ func (e *DefaultWaitExecHookHandler) HandleHooks(
OnError: hook.Hook.OnError,
Timeout: hook.Hook.ExecTimeout,
}
hookTracker.Record(newPod.Namespace, newPod.Name, hook.Hook.Container, hook.HookSource, hook.HookName, hookPhase(""), false)
if err := e.PodCommandExecutor.ExecutePodCommand(hookLog, podMap, pod.Namespace, pod.Name, hook.HookName, eh); err != nil {
hookLog.WithError(err).Error("Error executing hook")
err = fmt.Errorf("hook %s in container %s failed to execute, err: %v", hook.HookName, hook.Hook.Container, err)
errors = append(errors, err)
hookTracker.Record(newPod.Namespace, newPod.Name, hook.Hook.Container, hook.HookSource, hook.HookName, hookPhase(""), true)
if hook.Hook.OnError == velerov1api.HookErrorModeFail {
cancel()
return
@@ -226,6 +233,7 @@ func (e *DefaultWaitExecHookHandler) HandleHooks(
"hookPhase": "post",
},
)
hookTracker.Record(pod.Namespace, pod.Name, hook.Hook.Container, hook.HookSource, hook.HookName, hookPhase(""), true)
hookLog.Error(err)
errors = append(errors, err)
}
+263 -12
View File
@@ -98,7 +98,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "<from-annotation>",
HookSource: "annotation",
HookSource: HookSourceAnnotation,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -167,7 +167,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "<from-annotation>",
HookSource: "annotation",
HookSource: HookSourceAnnotation,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -236,7 +236,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "<from-annotation>",
HookSource: "annotation",
HookSource: HookSourceAnnotation,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -305,7 +305,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "<from-annotation>",
HookSource: "annotation",
HookSource: HookSourceAnnotation,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -391,7 +391,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "my-hook-1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -440,7 +440,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "my-hook-1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -471,7 +471,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "my-hook-1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -502,7 +502,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "my-hook-1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -533,7 +533,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "my-hook-1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -574,7 +574,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container1": {
{
HookName: "my-hook-1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
@@ -584,7 +584,7 @@ func TestWaitExecHandleHooks(t *testing.T) {
"container2": {
{
HookName: "my-hook-1",
HookSource: "backupSpec",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container2",
Command: []string{"/usr/bin/bar"},
@@ -744,7 +744,8 @@ func TestWaitExecHandleHooks(t *testing.T) {
defer ctxCancel()
}
errs := h.HandleHooks(ctx, velerotest.NewLogger(), test.initialPod, test.byContainer)
hookTracker := NewHookTracker()
errs := h.HandleHooks(ctx, velerotest.NewLogger(), test.initialPod, test.byContainer, hookTracker)
// for i, ee := range test.expectedErrors {
require.Len(t, errs, len(test.expectedErrors))
@@ -997,3 +998,253 @@ func TestMaxHookWait(t *testing.T) {
})
}
}
func TestRestoreHookTrackerUpdate(t *testing.T) {
type change struct {
// delta to wait since last change applied or pod added
wait time.Duration
updated *v1.Pod
}
type expectedExecution struct {
hook *velerov1api.ExecHook
name string
error error
pod *v1.Pod
}
hookTracker1 := NewHookTracker()
hookTracker1.Add("default", "my-pod", "container1", HookSourceAnnotation, "<from-annotation>", hookPhase(""))
hookTracker2 := NewHookTracker()
hookTracker2.Add("default", "my-pod", "container1", HookSourceSpec, "my-hook-1", hookPhase(""))
hookTracker3 := NewHookTracker()
hookTracker3.Add("default", "my-pod", "container1", HookSourceSpec, "my-hook-1", hookPhase(""))
hookTracker3.Add("default", "my-pod", "container2", HookSourceSpec, "my-hook-2", hookPhase(""))
tests1 := []struct {
name string
initialPod *v1.Pod
groupResource string
byContainer map[string][]PodExecRestoreHook
expectedExecutions []expectedExecution
hookTracker *HookTracker
expectedFailed int
}{
{
name: "a hook executes successfully",
initialPod: builder.ForPod("default", "my-pod").
ObjectMeta(builder.WithAnnotations(
podRestoreHookCommandAnnotationKey, "/usr/bin/foo",
podRestoreHookContainerAnnotationKey, "container1",
podRestoreHookOnErrorAnnotationKey, string(velerov1api.HookErrorModeContinue),
podRestoreHookTimeoutAnnotationKey, "1s",
podRestoreHookWaitTimeoutAnnotationKey, "1m",
)).
Containers(&v1.Container{
Name: "container1",
}).
ContainerStatuses(&v1.ContainerStatus{
Name: "container1",
State: v1.ContainerState{
Running: &v1.ContainerStateRunning{},
},
}).
Result(),
groupResource: "pods",
byContainer: map[string][]PodExecRestoreHook{
"container1": {
{
HookName: "<from-annotation>",
HookSource: HookSourceAnnotation,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
OnError: velerov1api.HookErrorModeContinue,
ExecTimeout: metav1.Duration{Duration: time.Second},
WaitTimeout: metav1.Duration{Duration: time.Minute},
},
},
},
},
expectedExecutions: []expectedExecution{
{
name: "<from-annotation>",
hook: &velerov1api.ExecHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
OnError: velerov1api.HookErrorModeContinue,
Timeout: metav1.Duration{Duration: time.Second},
},
error: nil,
pod: builder.ForPod("default", "my-pod").
ObjectMeta(builder.WithResourceVersion("1")).
ObjectMeta(builder.WithAnnotations(
podRestoreHookCommandAnnotationKey, "/usr/bin/foo",
podRestoreHookContainerAnnotationKey, "container1",
podRestoreHookOnErrorAnnotationKey, string(velerov1api.HookErrorModeContinue),
podRestoreHookTimeoutAnnotationKey, "1s",
podRestoreHookWaitTimeoutAnnotationKey, "1m",
)).
Containers(&v1.Container{
Name: "container1",
}).
ContainerStatuses(&v1.ContainerStatus{
Name: "container1",
State: v1.ContainerState{
Running: &v1.ContainerStateRunning{},
},
}).
Result(),
},
},
hookTracker: hookTracker1,
expectedFailed: 0,
},
{
name: "a hook with OnError mode Fail failed to execute",
groupResource: "pods",
initialPod: builder.ForPod("default", "my-pod").
Containers(&v1.Container{
Name: "container1",
}).
ContainerStatuses(&v1.ContainerStatus{
Name: "container1",
State: v1.ContainerState{
Waiting: &v1.ContainerStateWaiting{},
},
}).
Result(),
byContainer: map[string][]PodExecRestoreHook{
"container1": {
{
HookName: "my-hook-1",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
OnError: velerov1api.HookErrorModeFail,
WaitTimeout: metav1.Duration{Duration: time.Millisecond},
},
},
},
},
hookTracker: hookTracker2,
expectedFailed: 1,
},
{
name: "a hook with OnError mode Continue failed to execute",
groupResource: "pods",
initialPod: builder.ForPod("default", "my-pod").
Containers(&v1.Container{
Name: "container1",
}).
ContainerStatuses(&v1.ContainerStatus{
Name: "container1",
State: v1.ContainerState{
Waiting: &v1.ContainerStateWaiting{},
},
}).
Result(),
byContainer: map[string][]PodExecRestoreHook{
"container1": {
{
HookName: "my-hook-1",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
OnError: velerov1api.HookErrorModeContinue,
WaitTimeout: metav1.Duration{Duration: time.Millisecond},
},
},
},
},
hookTracker: hookTracker2,
expectedFailed: 1,
},
{
name: "two hooks with OnError mode Continue failed to execute",
groupResource: "pods",
initialPod: builder.ForPod("default", "my-pod").
Containers(&v1.Container{
Name: "container1",
}).
Containers(&v1.Container{
Name: "container2",
}).
// initially both are waiting
ContainerStatuses(&v1.ContainerStatus{
Name: "container1",
State: v1.ContainerState{
Waiting: &v1.ContainerStateWaiting{},
},
}).
ContainerStatuses(&v1.ContainerStatus{
Name: "container2",
State: v1.ContainerState{
Waiting: &v1.ContainerStateWaiting{},
},
}).
Result(),
byContainer: map[string][]PodExecRestoreHook{
"container1": {
{
HookName: "my-hook-1",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container1",
Command: []string{"/usr/bin/foo"},
OnError: velerov1api.HookErrorModeContinue,
WaitTimeout: metav1.Duration{Duration: time.Millisecond},
},
},
},
"container2": {
{
HookName: "my-hook-2",
HookSource: HookSourceSpec,
Hook: velerov1api.ExecRestoreHook{
Container: "container2",
Command: []string{"/usr/bin/bar"},
OnError: velerov1api.HookErrorModeContinue,
WaitTimeout: metav1.Duration{Duration: time.Millisecond},
},
},
},
},
hookTracker: hookTracker3,
expectedFailed: 2,
},
}
for _, test := range tests1 {
t.Run(test.name, func(t *testing.T) {
source := fcache.NewFakeControllerSource()
go func() {
// This is the state of the pod that will be seen by the AddFunc handler.
source.Add(test.initialPod)
}()
podCommandExecutor := &velerotest.MockPodCommandExecutor{}
defer podCommandExecutor.AssertExpectations(t)
h := &DefaultWaitExecHookHandler{
PodCommandExecutor: podCommandExecutor,
ListWatchFactory: &fakeListWatchFactory{source},
}
for _, e := range test.expectedExecutions {
obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(e.pod)
assert.Nil(t, err)
podCommandExecutor.On("ExecutePodCommand", mock.Anything, obj, e.pod.Namespace, e.pod.Name, e.name, e.hook).Return(e.error)
}
ctx := context.Background()
_ = h.HandleHooks(ctx, velerotest.NewLogger(), test.initialPod, test.byContainer, test.hookTracker)
_, actualFailed := test.hookTracker.Stat()
assert.Equal(t, test.expectedFailed, actualFailed)
})
}
}
+18
View File
@@ -441,6 +441,11 @@ type BackupStatus struct {
// BackupItemAction operations for this backup which ended with an error.
// +optional
BackupItemOperationsFailed int `json:"backupItemOperationsFailed,omitempty"`
// HookStatus contains information about the status of the hooks.
// +optional
// +nullable
HookStatus *HookStatus `json:"hookStatus,omitempty"`
}
// BackupProgress stores information about the progress of a Backup's execution.
@@ -458,6 +463,19 @@ type BackupProgress struct {
ItemsBackedUp int `json:"itemsBackedUp,omitempty"`
}
// HookStatus stores information about the status of the hooks.
type HookStatus struct {
// HooksAttempted is the total number of attempted hooks
// Specifically, HooksAttempted represents the number of hooks that failed to execute
// and the number of hooks that executed successfully.
// +optional
HooksAttempted int `json:"hooksAttempted,omitempty"`
// HooksFailed is the total number of hooks which ended with an error
// +optional
HooksFailed int `json:"hooksFailed,omitempty"`
}
// +genclient
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
// +kubebuilder:object:root=true
+1 -1
View File
@@ -83,7 +83,7 @@ const (
// AsyncOperationIDLabel is the label key used to identify the async operation ID
AsyncOperationIDLabel = "velero.io/async-operation-id"
// PVCNameLabel is the label key used to identify the the PVC's namespace and name.
// PVCNameLabel is the label key used to identify the PVC's namespace and name.
// The format is <namespace>/<name>.
PVCNamespaceNameLabel = "velero.io/pvc-namespace-name"
+5
View File
@@ -345,6 +345,11 @@ type RestoreStatus struct {
// RestoreItemAction operations for this restore which ended with an error.
// +optional
RestoreItemOperationsFailed int `json:"restoreItemOperationsFailed,omitempty"`
// HookStatus contains information about the status of the hooks.
// +optional
// +nullable
HookStatus *HookStatus `json:"hookStatus,omitempty"`
}
// RestoreProgress stores information about the restore's execution progress
@@ -419,6 +419,11 @@ func (in *BackupStatus) DeepCopyInto(out *BackupStatus) {
*out = new(BackupProgress)
**out = **in
}
if in.HookStatus != nil {
in, out := &in.HookStatus, &out.HookStatus
*out = new(HookStatus)
**out = **in
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BackupStatus.
@@ -802,6 +807,21 @@ func (in *ExecRestoreHook) DeepCopy() *ExecRestoreHook {
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *HookStatus) DeepCopyInto(out *HookStatus) {
*out = *in
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new HookStatus.
func (in *HookStatus) DeepCopy() *HookStatus {
if in == nil {
return nil
}
out := new(HookStatus)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *InitRestoreHook) DeepCopyInto(out *InitRestoreHook) {
*out = *in
@@ -1362,6 +1382,11 @@ func (in *RestoreStatus) DeepCopyInto(out *RestoreStatus) {
*out = new(RestoreProgress)
**out = **in
}
if in.HookStatus != nil {
in, out := &in.HookStatus, &out.HookStatus
*out = new(HookStatus)
**out = **in
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RestoreStatus.
+1 -1
View File
@@ -84,7 +84,7 @@ func (e *Extractor) readBackup(tarRdr *tar.Reader) (string, error) {
return "", err
}
target := filepath.Join(dir, header.Name) //nolint:gosec
target := filepath.Join(dir, header.Name) //nolint:gosec // Internal usage. No need to check.
switch header.Typeflag {
case tar.TypeDir:
+18 -4
View File
@@ -302,6 +302,7 @@ func (kb *kubernetesBackupper) BackupWithResolvers(log logrus.FieldLogger,
itemHookHandler: &hook.DefaultItemHookHandler{
PodCommandExecutor: kb.podCommandExecutor,
},
hookTracker: hook.NewHookTracker(),
}
// helper struct to send current progress between the main
@@ -427,11 +428,23 @@ func (kb *kubernetesBackupper) BackupWithResolvers(log logrus.FieldLogger,
updated.Status.Progress.TotalItems = len(backupRequest.BackedUpItems)
updated.Status.Progress.ItemsBackedUp = len(backupRequest.BackedUpItems)
if err := kube.PatchResource(backupRequest.Backup, updated, kb.kbClient); err != nil {
log.WithError(errors.WithStack((err))).Warn("Got error trying to update backup's status.progress")
// update the hooks execution status
if updated.Status.HookStatus == nil {
updated.Status.HookStatus = &velerov1api.HookStatus{}
}
skippedPVSummary, _ := json.Marshal(backupRequest.SkippedPVTracker.Summary())
log.Infof("Summary for skipped PVs: %s", skippedPVSummary)
updated.Status.HookStatus.HooksAttempted, updated.Status.HookStatus.HooksFailed = itemBackupper.hookTracker.Stat()
log.Infof("hookTracker: %+v, hookAttempted: %d, hookFailed: %d", itemBackupper.hookTracker.GetTracker(), updated.Status.HookStatus.HooksAttempted, updated.Status.HookStatus.HooksFailed)
if err := kube.PatchResource(backupRequest.Backup, updated, kb.kbClient); err != nil {
log.WithError(errors.WithStack((err))).Warn("Got error trying to update backup's status.progress and hook status")
}
if skippedPVSummary, err := json.Marshal(backupRequest.SkippedPVTracker.Summary()); err != nil {
log.WithError(errors.WithStack(err)).Warn("Fail to generate skipped PV summary.")
} else {
log.Infof("Summary for skipped PVs: %s", skippedPVSummary)
}
backupRequest.Status.Progress = &velerov1api.BackupProgress{TotalItems: len(backupRequest.BackedUpItems), ItemsBackedUp: len(backupRequest.BackedUpItems)}
log.WithField("progress", "").Infof("Backed up a total of %d items", len(backupRequest.BackedUpItems))
@@ -598,6 +611,7 @@ func (kb *kubernetesBackupper) FinalizeBackup(log logrus.FieldLogger,
discoveryHelper: kb.discoveryHelper,
itemHookHandler: &hook.NoOpItemHookHandler{},
podVolumeSnapshotTracker: newPVCSnapshotTracker(),
hookTracker: hook.NewHookTracker(),
}
updateFiles := make(map[string]FileForArchive)
backedUpGroupResources := map[schema.GroupResource]bool{}
+4 -3
View File
@@ -78,6 +78,7 @@ type itemBackupper struct {
itemHookHandler hook.ItemHookHandler
snapshotLocationVolumeSnapshotters map[string]vsv1.VolumeSnapshotter
hookTracker *hook.HookTracker
}
type FileForArchive struct {
@@ -184,7 +185,7 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
)
log.Debug("Executing pre hooks")
if err := ib.itemHookHandler.HandleHooks(log, groupResource, obj, ib.backupRequest.ResourceHooks, hook.PhasePre); err != nil {
if err := ib.itemHookHandler.HandleHooks(log, groupResource, obj, ib.backupRequest.ResourceHooks, hook.PhasePre, ib.hookTracker); err != nil {
return false, itemFiles, err
}
if optedOut, podName := ib.podVolumeSnapshotTracker.OptedoutByPod(namespace, name); optedOut {
@@ -234,7 +235,7 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
// if there was an error running actions, execute post hooks and return
log.Debug("Executing post hooks")
if err := ib.itemHookHandler.HandleHooks(log, groupResource, obj, ib.backupRequest.ResourceHooks, hook.PhasePost); err != nil {
if err := ib.itemHookHandler.HandleHooks(log, groupResource, obj, ib.backupRequest.ResourceHooks, hook.PhasePost, ib.hookTracker); err != nil {
backupErrs = append(backupErrs, err)
}
return false, itemFiles, kubeerrs.NewAggregate(backupErrs)
@@ -293,7 +294,7 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
}
log.Debug("Executing post hooks")
if err := ib.itemHookHandler.HandleHooks(log, groupResource, obj, ib.backupRequest.ResourceHooks, hook.PhasePost); err != nil {
if err := ib.itemHookHandler.HandleHooks(log, groupResource, obj, ib.backupRequest.ResourceHooks, hook.PhasePost, ib.hookTracker); err != nil {
backupErrs = append(backupErrs, err)
}
+1 -1
View File
@@ -29,7 +29,7 @@ type ServerStatusRequestBuilder struct {
object *velerov1api.ServerStatusRequest
}
// ForServerStatusRequest is the constructor for for a ServerStatusRequestBuilder.
// ForServerStatusRequest is the constructor for a ServerStatusRequestBuilder.
func ForServerStatusRequest(ns, name, resourceVersion string) *ServerStatusRequestBuilder {
return &ServerStatusRequestBuilder{
object: &velerov1api.ServerStatusRequest{
+7
View File
@@ -73,6 +73,7 @@ type Options struct {
UseVolumeSnapshots bool
DefaultRepoMaintenanceFrequency time.Duration
GarbageCollectionFrequency time.Duration
PodVolumeOperationTimeout time.Duration
Plugins flag.StringArray
NoDefaultBackupLocation bool
CRDsOnly bool
@@ -116,6 +117,7 @@ func (o *Options) BindFlags(flags *pflag.FlagSet) {
flags.BoolVar(&o.Wait, "wait", o.Wait, "Wait for Velero deployment to be ready. Optional.")
flags.DurationVar(&o.DefaultRepoMaintenanceFrequency, "default-repo-maintain-frequency", o.DefaultRepoMaintenanceFrequency, "How often 'maintain' is run for backup repositories by default. Optional.")
flags.DurationVar(&o.GarbageCollectionFrequency, "garbage-collection-frequency", o.GarbageCollectionFrequency, "How often the garbage collection runs for expired backups.(default 1h)")
flags.DurationVar(&o.PodVolumeOperationTimeout, "pod-volume-operation-timeout", o.PodVolumeOperationTimeout, "How long to wait for pod volume operations to complete before timing out(default 4h). Optional.")
flags.Var(&o.Plugins, "plugins", "Plugin container images to install into the Velero Deployment")
flags.BoolVar(&o.CRDsOnly, "crds-only", o.CRDsOnly, "Only generate CustomResourceDefinition resources. Useful for updating CRDs for an existing Velero install.")
flags.StringVar(&o.CACertFile, "cacert", o.CACertFile, "File containing a certificate bundle to use when verifying TLS connections to the object store. Optional.")
@@ -209,6 +211,7 @@ func (o *Options) AsVeleroOptions() (*install.VeleroOptions, error) {
VSLConfig: o.VolumeSnapshotConfig.Data(),
DefaultRepoMaintenanceFrequency: o.DefaultRepoMaintenanceFrequency,
GarbageCollectionFrequency: o.GarbageCollectionFrequency,
PodVolumeOperationTimeout: o.PodVolumeOperationTimeout,
Plugins: o.Plugins,
NoDefaultBackupLocation: o.NoDefaultBackupLocation,
CACertData: caCertData,
@@ -426,5 +429,9 @@ func (o *Options) Validate(c *cobra.Command, args []string, f client.Factory) er
return errors.New("--garbage-collection-frequency must be non-negative")
}
if o.PodVolumeOperationTimeout < 0 {
return errors.New("--pod-volume-operation-timeout must be non-negative")
}
return nil
}
+11 -48
View File
@@ -285,13 +285,13 @@ func (s *nodeAgentServer) run() {
}
dataUploadReconciler := controller.NewDataUploadReconciler(s.mgr.GetClient(), s.kubeClient, s.csiSnapshotClient.SnapshotV1(), s.dataPathMgr, repoEnsurer, clock.RealClock{}, credentialGetter, s.nodeName, s.fileSystem, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
s.markDataUploadsCancel(dataUploadReconciler)
s.attemptDataUploadResume(dataUploadReconciler)
if err = dataUploadReconciler.SetupWithManager(s.mgr); err != nil {
s.logger.WithError(err).Fatal("Unable to create the data upload controller")
}
dataDownloadReconciler := controller.NewDataDownloadReconciler(s.mgr.GetClient(), s.kubeClient, s.dataPathMgr, repoEnsurer, credentialGetter, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
s.markDataDownloadsCancel(dataDownloadReconciler)
s.attemptDataDownloadResume(dataDownloadReconciler)
if err = dataDownloadReconciler.SetupWithManager(s.mgr); err != nil {
s.logger.WithError(err).Fatal("Unable to create the data download controller")
}
@@ -365,71 +365,34 @@ func (s *nodeAgentServer) markInProgressCRsFailed() {
s.markInProgressPVRsFailed(client)
}
func (s *nodeAgentServer) markDataUploadsCancel(r *controller.DataUploadReconciler) {
func (s *nodeAgentServer) attemptDataUploadResume(r *controller.DataUploadReconciler) {
// the function is called before starting the controller manager, the embedded client isn't ready to use, so create a new one here
client, err := ctrlclient.New(s.mgr.GetConfig(), ctrlclient.Options{Scheme: s.mgr.GetScheme()})
if err != nil {
s.logger.WithError(errors.WithStack(err)).Error("failed to create client")
return
}
if dataUploads, err := r.FindDataUploads(s.ctx, client, s.namespace); err != nil {
s.logger.WithError(errors.WithStack(err)).Error("failed to find data uploads")
} else {
for i := range dataUploads {
du := dataUploads[i]
if du.Status.Phase == velerov2alpha1api.DataUploadPhaseAccepted ||
du.Status.Phase == velerov2alpha1api.DataUploadPhasePrepared ||
du.Status.Phase == velerov2alpha1api.DataUploadPhaseInProgress {
err = controller.UpdateDataUploadWithRetry(s.ctx, client, types.NamespacedName{Namespace: du.Namespace, Name: du.Name}, s.logger.WithField("dataupload", du.Name),
func(dataUpload *velerov2alpha1api.DataUpload) {
dataUpload.Spec.Cancel = true
dataUpload.Status.Message = fmt.Sprintf("found a dataupload with status %q during the node-agent starting, mark it as cancel", du.Status.Phase)
})
if err != nil {
s.logger.WithError(errors.WithStack(err)).Errorf("failed to mark dataupload %q cancel", du.GetName())
continue
}
s.logger.WithField("dataupload", du.GetName()).Warn(du.Status.Message)
}
}
if err := r.AttemptDataUploadResume(s.ctx, client, s.logger.WithField("node", s.nodeName), s.namespace); err != nil {
s.logger.WithError(errors.WithStack(err)).Error("failed to attempt data upload resume")
}
}
func (s *nodeAgentServer) markDataDownloadsCancel(r *controller.DataDownloadReconciler) {
func (s *nodeAgentServer) attemptDataDownloadResume(r *controller.DataDownloadReconciler) {
// the function is called before starting the controller manager, the embedded client isn't ready to use, so create a new one here
client, err := ctrlclient.New(s.mgr.GetConfig(), ctrlclient.Options{Scheme: s.mgr.GetScheme()})
if err != nil {
s.logger.WithError(errors.WithStack(err)).Error("failed to create client")
return
}
if dataDownloads, err := r.FindDataDownloads(s.ctx, client, s.namespace); err != nil {
s.logger.WithError(errors.WithStack(err)).Error("failed to find data downloads")
} else {
for i := range dataDownloads {
dd := dataDownloads[i]
if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseAccepted ||
dd.Status.Phase == velerov2alpha1api.DataDownloadPhasePrepared ||
dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress {
err = controller.UpdateDataDownloadWithRetry(s.ctx, client, types.NamespacedName{Namespace: dd.Namespace, Name: dd.Name}, s.logger.WithField("datadownload", dd.Name),
func(dataDownload *velerov2alpha1api.DataDownload) {
dataDownload.Spec.Cancel = true
dataDownload.Status.Message = fmt.Sprintf("found a datadownload with status %q during the node-agent starting, mark it as cancel", dd.Status.Phase)
})
if err != nil {
s.logger.WithError(errors.WithStack(err)).Errorf("failed to mark datadownload %q cancel", dd.GetName())
continue
}
s.logger.WithField("datadownload", dd.GetName()).Warn(dd.Status.Message)
}
}
if err := r.AttemptDataDownloadResume(s.ctx, client, s.logger.WithField("node", s.nodeName), s.namespace); err != nil {
s.logger.WithError(errors.WithStack(err)).Error("failed to attempt data download resume")
}
}
func (s *nodeAgentServer) markInProgressPVBsFailed(client ctrlclient.Client) {
pvbs := &velerov1api.PodVolumeBackupList{}
if err := client.List(s.ctx, pvbs, &ctrlclient.MatchingFields{"metadata.namespace": s.namespace}); err != nil {
if err := client.List(s.ctx, pvbs, &ctrlclient.ListOptions{Namespace: s.namespace}); err != nil {
s.logger.WithError(errors.WithStack(err)).Error("failed to list podvolumebackups")
return
}
@@ -455,7 +418,7 @@ func (s *nodeAgentServer) markInProgressPVBsFailed(client ctrlclient.Client) {
func (s *nodeAgentServer) markInProgressPVRsFailed(client ctrlclient.Client) {
pvrs := &velerov1api.PodVolumeRestoreList{}
if err := client.List(s.ctx, pvrs, &ctrlclient.MatchingFields{"metadata.namespace": s.namespace}); err != nil {
if err := client.List(s.ctx, pvrs, &ctrlclient.ListOptions{Namespace: s.namespace}); err != nil {
s.logger.WithError(errors.WithStack(err)).Error("failed to list podvolumerestores")
return
}
@@ -523,7 +486,7 @@ func (s *nodeAgentServer) getDataPathConcurrentNum(defaultNum int) int {
concurrentNum := math.MaxInt32
for _, rule := range configs.DataPathConcurrency.PerNodeConfig {
selector, err := metav1.LabelSelectorAsSelector(&rule.NodeSelector)
selector, err := metav1.LabelSelectorAsSelector(&(rule.NodeSelector))
if err != nil {
s.logger.WithError(err).Warnf("Failed to parse rule with label selector %s, skip it", rule.NodeSelector.String())
continue
+16 -5
View File
@@ -34,6 +34,7 @@ import (
"github.com/spf13/cobra"
corev1api "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
kubeerrs "k8s.io/apimachinery/pkg/util/errors"
@@ -247,7 +248,7 @@ type server struct {
discoveryHelper velerodiscovery.Helper
dynamicClient dynamic.Interface
// controller-runtime client. the difference from the controller-manager's client
// is that the the controller-manager's client is limited to list namespaced-scoped
// is that the controller-manager's client is limited to list namespaced-scoped
// resources in the namespace where Velero is installed, or the cluster-scoped
// resources. The crClient doesn't have the limitation.
crClient ctrlclient.Client
@@ -990,7 +991,7 @@ func markInProgressCRsFailed(ctx context.Context, cfg *rest.Config, scheme *runt
func markInProgressBackupsFailed(ctx context.Context, client ctrlclient.Client, namespace string, log logrus.FieldLogger) {
backups := &velerov1api.BackupList{}
if err := client.List(ctx, backups, &ctrlclient.MatchingFields{"metadata.namespace": namespace}); err != nil {
if err := client.List(ctx, backups, &ctrlclient.ListOptions{Namespace: namespace}); err != nil {
log.WithError(errors.WithStack(err)).Error("failed to list backups")
return
}
@@ -1015,7 +1016,7 @@ func markInProgressBackupsFailed(ctx context.Context, client ctrlclient.Client,
func markInProgressRestoresFailed(ctx context.Context, client ctrlclient.Client, namespace string, log logrus.FieldLogger) {
restores := &velerov1api.RestoreList{}
if err := client.List(ctx, restores, &ctrlclient.MatchingFields{"metadata.namespace": namespace}); err != nil {
if err := client.List(ctx, restores, &ctrlclient.ListOptions{Namespace: namespace}); err != nil {
log.WithError(errors.WithStack(err)).Error("failed to list restores")
return
}
@@ -1040,7 +1041,12 @@ func markInProgressRestoresFailed(ctx context.Context, client ctrlclient.Client,
func markDataUploadsCancel(ctx context.Context, client ctrlclient.Client, backup velerov1api.Backup, log logrus.FieldLogger) {
dataUploads := &velerov2alpha1api.DataUploadList{}
if err := client.List(ctx, dataUploads, &ctrlclient.MatchingFields{"metadata.namespace": backup.GetNamespace()}, &ctrlclient.MatchingLabels{velerov1api.BackupUIDLabel: string(backup.GetUID())}); err != nil {
if err := client.List(ctx, dataUploads, &ctrlclient.ListOptions{
Namespace: backup.GetNamespace(),
LabelSelector: labels.Set(map[string]string{
velerov1api.BackupUIDLabel: string(backup.GetUID()),
}).AsSelector(),
}); err != nil {
log.WithError(errors.WithStack(err)).Error("failed to list dataUploads")
return
}
@@ -1070,7 +1076,12 @@ func markDataUploadsCancel(ctx context.Context, client ctrlclient.Client, backup
func markDataDownloadsCancel(ctx context.Context, client ctrlclient.Client, restore velerov1api.Restore, log logrus.FieldLogger) {
dataDownloads := &velerov2alpha1api.DataDownloadList{}
if err := client.List(ctx, dataDownloads, &ctrlclient.MatchingFields{"metadata.namespace": restore.GetNamespace()}, &ctrlclient.MatchingLabels{velerov1api.RestoreUIDLabel: string(restore.GetUID())}); err != nil {
if err := client.List(ctx, dataDownloads, &ctrlclient.ListOptions{
Namespace: restore.GetNamespace(),
LabelSelector: labels.Set(map[string]string{
velerov1api.RestoreUIDLabel: string(restore.GetUID()),
}).AsSelector(),
}); err != nil {
log.WithError(errors.WithStack(err)).Error("failed to list dataDownloads")
return
}
@@ -111,7 +111,7 @@ func Stream(ctx context.Context, kbClient kbclient.Client, namespace, name strin
httpClient := new(http.Client)
httpClient.Transport = &http.Transport{
TLSClientConfig: &tls.Config{
InsecureSkipVerify: insecureSkipTLSVerify, //nolint:gosec
InsecureSkipVerify: insecureSkipTLSVerify, //nolint:gosec // This parameter is useful for some scenarios.
RootCAs: caPool,
},
IdleConnTimeout: timeout,
@@ -123,7 +123,7 @@ func Stream(ctx context.Context, kbClient kbclient.Client, namespace, name strin
ExpectContinueTimeout: defaultTransport.ExpectContinueTimeout,
}
httpReq, err := http.NewRequest(http.MethodGet, created.Status.DownloadURL, nil)
httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, created.Status.DownloadURL, nil)
if err != nil {
return err
}
+6
View File
@@ -358,6 +358,12 @@ func DescribeBackupStatus(ctx context.Context, kbClient kbclient.Client, d *Desc
describeBackupVolumes(ctx, kbClient, d, backup, details, insecureSkipTLSVerify, caCertPath, podVolumeBackups)
d.Println()
if status.HookStatus != nil {
d.Println()
d.Printf("HooksAttempted:\t%d\n", status.HookStatus.HooksAttempted)
d.Printf("HooksFailed:\t%d\n", status.HookStatus.HooksFailed)
}
}
func describeBackupItemOperations(ctx context.Context, kbClient kbclient.Client, d *Describer, backup *velerov1api.Backup, details bool, insecureSkipTLSVerify bool, caCertPath string) {
@@ -266,6 +266,11 @@ func DescribeBackupStatusInSF(ctx context.Context, kbClient kbclient.Client, d *
}
describeBackupVolumesInSF(ctx, kbClient, backup, details, insecureSkipTLSVerify, caCertPath, podVolumeBackups, backupStatusInfo)
if status.HookStatus != nil {
backupStatusInfo["hooksAttempted"] = status.HookStatus.HooksAttempted
backupStatusInfo["hooksFailed"] = status.HookStatus.HooksFailed
}
}
func describeBackupResourceListInSF(ctx context.Context, kbClient kbclient.Client, backupStatusInfo map[string]interface{}, backup *velerov1api.Backup, insecureSkipTLSVerify bool, caCertPath string) {
+6 -1
View File
@@ -164,6 +164,11 @@ func (d *StructuredDescriber) JSONEncode() string {
encoder := json.NewEncoder(byteBuffer)
encoder.SetEscapeHTML(false)
encoder.SetIndent("", " ")
_ = encoder.Encode(d.output)
err := encoder.Encode(d.output)
if err != nil {
fmt.Printf("fail to encode %s", err.Error())
return ""
}
return byteBuffer.String()
}
+6
View File
@@ -180,6 +180,12 @@ func DescribeRestore(ctx context.Context, kbClient kbclient.Client, restore *vel
d.Println()
describeRestoreItemOperations(ctx, kbClient, d, restore, details, insecureSkipTLSVerify, caCertFile)
if restore.Status.HookStatus != nil {
d.Println()
d.Printf("HooksAttempted: \t%d\n", restore.Status.HookStatus.HooksAttempted)
d.Printf("HooksFailed: \t%d\n", restore.Status.HookStatus.HooksFailed)
}
if details {
describeRestoreResourceList(ctx, kbClient, d, restore, insecureSkipTLSVerify, caCertFile)
d.Println()
+1 -1
View File
@@ -1080,7 +1080,7 @@ func generateVolumeInfoForCSIVolumeSnapshot(backup *pkgbackup.Request, csiVolume
SnapshotDataMoved: false,
PreserveLocalSnapshot: true,
OperationID: operation.Spec.OperationID,
StartTimestamp: &volumeSnapshot.CreationTimestamp,
StartTimestamp: &(volumeSnapshot.CreationTimestamp),
CSISnapshotInfo: volume.CSISnapshotInfo{
VSCName: *volumeSnapshot.Status.BoundVolumeSnapshotContentName,
Size: size,
+5 -1
View File
@@ -544,7 +544,11 @@ func (r *backupDeletionReconciler) deleteMovedSnapshots(ctx context.Context, bac
for i := range list.Items {
cm := list.Items[i]
snapshot := repository.SnapshotIdentifier{}
b, _ := json.Marshal(cm.Data)
b, err := json.Marshal(cm.Data)
if err != nil {
errs = append(errs, errors.Wrapf(err, "fail to marshal the snapshot info into JSON"))
continue
}
if err := json.Unmarshal(b, &snapshot); err != nil {
errs = append(errs, errors.Wrapf(err, "failed to unmarshal snapshot info"))
continue
+79 -4
View File
@@ -140,7 +140,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
// to help clear up resources instead of clear them directly in case of some conflict with Expose action
if err := UpdateDataDownloadWithRetry(ctx, r.client, req.NamespacedName, log, func(dataDownload *velerov2alpha1api.DataDownload) {
dataDownload.Spec.Cancel = true
dataDownload.Status.Message = fmt.Sprintf("found a dataupload %s/%s is being deleted, mark it as cancel", dd.Namespace, dd.Name)
dataDownload.Status.Message = fmt.Sprintf("found a datadownload %s/%s is being deleted, mark it as cancel", dd.Namespace, dd.Name)
}); err != nil {
log.Errorf("failed to set cancel flag with error %s for %s/%s", err.Error(), dd.Namespace, dd.Name)
return ctrl.Result{}, err
@@ -151,6 +151,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
log.Info("Data download starting")
if _, err := r.getTargetPVC(ctx, dd); err != nil {
log.WithField("error", err).Debugf("Cannot find target PVC for DataDownload yet. Retry later.")
return ctrl.Result{Requeue: true}, nil
}
@@ -192,7 +193,6 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
return r.errorOut(ctx, dd, err, "error to expose snapshot", log)
}
}
log.Info("Restore is exposed")
// we need to get CR again for it may canceled by datadownload controller on other
@@ -205,7 +205,6 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
}
return ctrl.Result{}, errors.Wrap(err, "getting datadownload")
}
// we need to clean up resources as resources created in Expose it may later than cancel action or prepare time
// and need to clean up resources again
if isDataDownloadInFinalState(dd) {
@@ -267,7 +266,6 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
return r.errorOut(ctx, dd, err, "error to create data path", log)
}
}
// Update status to InProgress
original := dd.DeepCopy()
dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseInProgress
@@ -576,6 +574,51 @@ func (r *DataDownloadReconciler) FindDataDownloads(ctx context.Context, cli clie
return dataDownloads, nil
}
func (r *DataDownloadReconciler) findAcceptDataDownloadsByNodeLabel(ctx context.Context, cli client.Client, ns string) ([]velerov2alpha1api.DataDownload, error) {
dataDownloads := &velerov2alpha1api.DataDownloadList{}
if err := cli.List(ctx, dataDownloads, &client.ListOptions{Namespace: ns}); err != nil {
r.logger.WithError(errors.WithStack(err)).Error("failed to list datauploads")
return nil, errors.Wrapf(err, "failed to list datauploads")
}
var result []velerov2alpha1api.DataDownload
for _, dd := range dataDownloads.Items {
if dd.Status.Phase != velerov2alpha1api.DataDownloadPhaseAccepted {
continue
}
if dd.Labels[acceptNodeLabelKey] == r.nodeName {
result = append(result, dd)
}
}
return result, nil
}
// CancelAcceptedDataDownload will cancel the accepted data download
func (r *DataDownloadReconciler) CancelAcceptedDataDownload(ctx context.Context, cli client.Client, ns string) {
r.logger.Infof("Canceling accepted data for node %s", r.nodeName)
dataDownloads, err := r.findAcceptDataDownloadsByNodeLabel(ctx, cli, ns)
if err != nil {
r.logger.WithError(err).Error("failed to find data downloads")
return
}
for _, dd := range dataDownloads {
if dd.Spec.Cancel {
continue
}
err = UpdateDataDownloadWithRetry(ctx, cli, types.NamespacedName{Namespace: dd.Namespace, Name: dd.Name},
r.logger.WithField("dataupload", dd.Name), func(dataDownload *velerov2alpha1api.DataDownload) {
dataDownload.Spec.Cancel = true
dataDownload.Status.Message = fmt.Sprintf("found a datadownload with status %q during the node-agent starting, mark it as cancel", dd.Status.Phase)
})
r.logger.Warn(dd.Status.Message)
if err != nil {
r.logger.WithError(err).Errorf("failed to set cancel flag with error %s", err.Error())
}
}
}
func (r *DataDownloadReconciler) prepareDataDownload(ssb *velerov2alpha1api.DataDownload) {
ssb.Status.Phase = velerov2alpha1api.DataDownloadPhasePrepared
ssb.Status.Node = r.nodeName
@@ -749,3 +792,35 @@ func UpdateDataDownloadWithRetry(ctx context.Context, client client.Client, name
return true, nil
})
}
func (r *DataDownloadReconciler) AttemptDataDownloadResume(ctx context.Context, cli client.Client, logger *logrus.Entry, ns string) error {
if dataDownloads, err := r.FindDataDownloads(ctx, cli, ns); err != nil {
return errors.Wrapf(err, "failed to find data downloads")
} else {
for i := range dataDownloads {
dd := dataDownloads[i]
if dd.Status.Phase == velerov2alpha1api.DataDownloadPhasePrepared {
// keep doing nothing let controller re-download the data
// the Prepared CR could be still handled by datadownload controller after node-agent restart
logger.WithField("datadownload", dd.GetName()).Debug("find a datadownload with status prepared")
} else if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress {
err = UpdateDataDownloadWithRetry(ctx, cli, types.NamespacedName{Namespace: dd.Namespace, Name: dd.Name}, logger.WithField("datadownload", dd.Name),
func(dataDownload *velerov2alpha1api.DataDownload) {
dataDownload.Spec.Cancel = true
dataDownload.Status.Message = fmt.Sprintf("found a datadownload with status %q during the node-agent starting, mark it as cancel", dd.Status.Phase)
})
if err != nil {
logger.WithError(errors.WithStack(err)).Errorf("failed to mark datadownload %q into canceled", dd.GetName())
continue
}
logger.WithField("datadownload", dd.GetName()).Debug("mark datadownload into canceled")
}
}
}
//If the data download is in Accepted status, the expoded PVC may be not created
// so we need to mark the data download as canceled for it may not be recoverable
r.CancelAcceptedDataDownload(ctx, cli, ns)
return nil
}
+113 -1
View File
@@ -69,7 +69,7 @@ func dataDownloadBuilder() *builder.DataDownloadBuilder {
}
func initDataDownloadReconciler(objects []runtime.Object, needError ...bool) (*DataDownloadReconciler, error) {
var errs []error = make([]error, 5)
var errs []error = make([]error, 6)
for k, isError := range needError {
if k == 0 && isError {
errs[0] = fmt.Errorf("Get error")
@@ -81,6 +81,8 @@ func initDataDownloadReconciler(objects []runtime.Object, needError ...bool) (*D
errs[3] = fmt.Errorf("Patch error")
} else if k == 4 && isError {
errs[4] = apierrors.NewConflict(velerov2alpha1api.Resource("datadownload"), dataDownloadName, errors.New("conflict"))
} else if k == 5 && isError {
errs[5] = fmt.Errorf("List error")
}
}
return initDataDownloadReconcilerWithError(objects, errs...)
@@ -116,6 +118,8 @@ func initDataDownloadReconcilerWithError(objects []runtime.Object, needError ...
fakeClient.patchError = needError[3]
} else if k == 4 {
fakeClient.updateConflict = needError[4]
} else if k == 5 {
fakeClient.listError = needError[5]
}
}
@@ -939,3 +943,111 @@ func TestFindDataDownloads(t *testing.T) {
})
}
}
func TestAttemptDataDownloadResume(t *testing.T) {
tests := []struct {
name string
dataUploads []velerov2alpha1api.DataDownload
du *velerov2alpha1api.DataDownload
pod *corev1.Pod
needErrs []bool
acceptedDataDownloads []string
prepareddDataDownloads []string
cancelledDataDownloads []string
expectedError bool
}{
// Test case 1: Process Accepted DataDownload
{
name: "AcceptedDataDownload",
pod: builder.ForPod(velerov1api.DefaultNamespace, dataDownloadName).Volumes(&corev1.Volume{Name: dataDownloadName}).NodeName("node-1").Labels(map[string]string{
velerov1api.DataDownloadLabel: dataDownloadName,
}).Result(),
du: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseAccepted).Result(),
acceptedDataDownloads: []string{dataDownloadName},
expectedError: false,
},
// Test case 2: Cancel an Accepted DataDownload
{
name: "CancelAcceptedDataDownload",
du: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseAccepted).Result(),
},
// Test case 3: Process Accepted Prepared DataDownload
{
name: "PreparedDataDownload",
pod: builder.ForPod(velerov1api.DefaultNamespace, dataDownloadName).Volumes(&corev1.Volume{Name: dataDownloadName}).NodeName("node-1").Labels(map[string]string{
velerov1api.DataDownloadLabel: dataDownloadName,
}).Result(),
du: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhasePrepared).Result(),
prepareddDataDownloads: []string{dataDownloadName},
},
// Test case 4: Process Accepted InProgress DataDownload
{
name: "InProgressDataDownload",
pod: builder.ForPod(velerov1api.DefaultNamespace, dataDownloadName).Volumes(&corev1.Volume{Name: dataDownloadName}).NodeName("node-1").Labels(map[string]string{
velerov1api.DataDownloadLabel: dataDownloadName,
}).Result(),
du: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhasePrepared).Result(),
prepareddDataDownloads: []string{dataDownloadName},
},
// Test case 5: get resume error
{
name: "ResumeError",
pod: builder.ForPod(velerov1api.DefaultNamespace, dataDownloadName).Volumes(&corev1.Volume{Name: dataDownloadName}).NodeName("node-1").Labels(map[string]string{
velerov1api.DataDownloadLabel: dataDownloadName,
}).Result(),
needErrs: []bool{false, false, false, false, false, true},
du: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhasePrepared).Result(),
expectedError: true,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
ctx := context.TODO()
r, err := initDataDownloadReconciler(nil, test.needErrs...)
r.nodeName = "node-1"
require.NoError(t, err)
defer func() {
r.client.Delete(ctx, test.du, &kbclient.DeleteOptions{})
if test.pod != nil {
r.client.Delete(ctx, test.pod, &kbclient.DeleteOptions{})
}
}()
assert.NoError(t, r.client.Create(ctx, test.du))
if test.pod != nil {
assert.NoError(t, r.client.Create(ctx, test.pod))
}
// Run the test
err = r.AttemptDataDownloadResume(ctx, r.client, r.logger.WithField("name", test.name), test.du.Namespace)
if test.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
// Verify DataDownload marked as Cancelled
for _, duName := range test.cancelledDataDownloads {
dataUpload := &velerov2alpha1api.DataDownload{}
err := r.client.Get(context.Background(), types.NamespacedName{Namespace: "velero", Name: duName}, dataUpload)
require.NoError(t, err)
assert.Equal(t, velerov2alpha1api.DataDownloadPhaseCanceled, dataUpload.Status.Phase)
}
// Verify DataDownload marked as Accepted
for _, duName := range test.acceptedDataDownloads {
dataUpload := &velerov2alpha1api.DataDownload{}
err := r.client.Get(context.Background(), types.NamespacedName{Namespace: "velero", Name: duName}, dataUpload)
require.NoError(t, err)
assert.Equal(t, velerov2alpha1api.DataDownloadPhaseAccepted, dataUpload.Status.Phase)
}
// Verify DataDownload marked as Prepared
for _, duName := range test.prepareddDataDownloads {
dataUpload := &velerov2alpha1api.DataDownload{}
err := r.client.Get(context.Background(), types.NamespacedName{Namespace: "velero", Name: duName}, dataUpload)
require.NoError(t, err)
assert.Equal(t, velerov2alpha1api.DataDownloadPhasePrepared, dataUpload.Status.Phase)
}
}
})
}
}
+77 -2
View File
@@ -274,7 +274,6 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
return r.errorOut(ctx, du, err, "error to create data path", log)
}
}
// Update status to InProgress
original := du.DeepCopy()
du.Status.Phase = velerov2alpha1api.DataUploadPhaseInProgress
@@ -581,7 +580,7 @@ func (r *DataUploadReconciler) findDataUploadForPod(podObj client.Object) []reco
return []reconcile.Request{requests}
}
func (r *DataUploadReconciler) FindDataUploads(ctx context.Context, cli client.Client, ns string) ([]velerov2alpha1api.DataUpload, error) {
func (r *DataUploadReconciler) FindDataUploadsByPod(ctx context.Context, cli client.Client, ns string) ([]velerov2alpha1api.DataUpload, error) {
pods := &corev1.PodList{}
var dataUploads []velerov2alpha1api.DataUpload
if err := cli.List(ctx, pods, &client.ListOptions{Namespace: ns}); err != nil {
@@ -605,6 +604,51 @@ func (r *DataUploadReconciler) FindDataUploads(ctx context.Context, cli client.C
return dataUploads, nil
}
func (r *DataUploadReconciler) findAcceptDataUploadsByNodeLabel(ctx context.Context, cli client.Client, ns string) ([]velerov2alpha1api.DataUpload, error) {
dataUploads := &velerov2alpha1api.DataUploadList{}
if err := cli.List(ctx, dataUploads, &client.ListOptions{Namespace: ns}); err != nil {
r.logger.WithError(errors.WithStack(err)).Error("failed to list datauploads")
return nil, errors.Wrapf(err, "failed to list datauploads")
}
var result []velerov2alpha1api.DataUpload
for _, du := range dataUploads.Items {
if du.Status.Phase != velerov2alpha1api.DataUploadPhaseAccepted {
continue
}
if du.Labels[acceptNodeLabelKey] == r.nodeName {
result = append(result, du)
}
}
return result, nil
}
func (r *DataUploadReconciler) CancelAcceptedDataupload(ctx context.Context, cli client.Client, ns string) {
r.logger.Infof("Reset accepted dataupload for node %s", r.nodeName)
dataUploads, err := r.findAcceptDataUploadsByNodeLabel(ctx, cli, ns)
if err != nil {
r.logger.WithError(err).Error("failed to find dataupload")
return
}
for _, du := range dataUploads {
if du.Spec.Cancel {
continue
}
err = UpdateDataUploadWithRetry(ctx, cli, types.NamespacedName{Namespace: du.Namespace, Name: du.Name}, r.logger.WithField("dataupload", du.Name),
func(dataUpload *velerov2alpha1api.DataUpload) {
dataUpload.Spec.Cancel = true
dataUpload.Status.Message = fmt.Sprintf("found a dataupload with status %q during the node-agent starting, mark it as cancel", du.Status.Phase)
})
r.logger.WithField("dataupload", du.GetName()).Warn(du.Status.Message)
if err != nil {
r.logger.WithError(errors.WithStack(err)).Errorf("failed to mark dataupload %q cancel", du.GetName())
continue
}
}
}
func (r *DataUploadReconciler) prepareDataUpload(du *velerov2alpha1api.DataUpload) {
du.Status.Phase = velerov2alpha1api.DataUploadPhasePrepared
du.Status.Node = r.nodeName
@@ -833,3 +877,34 @@ func UpdateDataUploadWithRetry(ctx context.Context, client client.Client, namesp
return true, nil
})
}
func (r *DataUploadReconciler) AttemptDataUploadResume(ctx context.Context, cli client.Client, logger *logrus.Entry, ns string) error {
if dataUploads, err := r.FindDataUploadsByPod(ctx, cli, ns); err != nil {
return errors.Wrap(err, "failed to find data uploads")
} else {
for _, du := range dataUploads {
if du.Status.Phase == velerov2alpha1api.DataUploadPhasePrepared {
// keep doing nothing let controller re-download the data
// the Prepared CR could be still handled by dataupload controller after node-agent restart
logger.WithField("dataupload", du.GetName()).Debug("find a dataupload with status prepared")
} else if du.Status.Phase == velerov2alpha1api.DataUploadPhaseInProgress {
err = UpdateDataUploadWithRetry(ctx, cli, types.NamespacedName{Namespace: du.Namespace, Name: du.Name}, logger.WithField("dataupload", du.Name),
func(dataUpload *velerov2alpha1api.DataUpload) {
dataUpload.Spec.Cancel = true
dataUpload.Status.Message = fmt.Sprintf("found a dataupload with status %q during the node-agent starting, mark it as cancel", du.Status.Phase)
})
if err != nil {
logger.WithError(errors.WithStack(err)).Errorf("failed to mark dataupload %q into canceled", du.GetName())
continue
}
logger.WithField("dataupload", du.GetName()).Debug("mark dataupload into canceled")
}
}
}
//If the data upload is in Accepted status, the volume snapshot may be deleted and the exposed pod may not be created
// so we need to mark the data upload as canceled for it may not be recoverable
r.CancelAcceptedDataupload(ctx, cli, ns)
return nil
}
+123 -3
View File
@@ -69,6 +69,7 @@ type FakeClient struct {
updateError error
patchError error
updateConflict error
listError error
}
func (c *FakeClient) Get(ctx context.Context, key kbclient.ObjectKey, obj kbclient.Object) error {
@@ -107,8 +108,16 @@ func (c *FakeClient) Patch(ctx context.Context, obj kbclient.Object, patch kbcli
return c.Client.Patch(ctx, obj, patch, opts...)
}
func (c *FakeClient) List(ctx context.Context, list kbclient.ObjectList, opts ...kbclient.ListOption) error {
if c.listError != nil {
return c.listError
}
return c.Client.List(ctx, list, opts...)
}
func initDataUploaderReconciler(needError ...bool) (*DataUploadReconciler, error) {
var errs []error = make([]error, 5)
var errs []error = make([]error, 6)
for k, isError := range needError {
if k == 0 && isError {
errs[0] = fmt.Errorf("Get error")
@@ -119,7 +128,9 @@ func initDataUploaderReconciler(needError ...bool) (*DataUploadReconciler, error
} else if k == 3 && isError {
errs[3] = fmt.Errorf("Patch error")
} else if k == 4 && isError {
errs[4] = apierrors.NewConflict(velerov2alpha1api.Resource("datadownload"), dataDownloadName, errors.New("conflict"))
errs[4] = apierrors.NewConflict(velerov2alpha1api.Resource("dataupload"), dataUploadName, errors.New("conflict"))
} else if k == 5 && isError {
errs[5] = fmt.Errorf("List error")
}
}
@@ -199,6 +210,8 @@ func initDataUploaderReconcilerWithError(needError ...error) (*DataUploadReconci
fakeClient.patchError = needError[3]
} else if k == 4 {
fakeClient.updateConflict = needError[4]
} else if k == 5 {
fakeClient.listError = needError[5]
}
}
@@ -984,7 +997,7 @@ func TestFindDataUploads(t *testing.T) {
require.NoError(t, err)
err = r.client.Create(ctx, &test.pod)
require.NoError(t, err)
uploads, err := r.FindDataUploads(context.Background(), r.client, "velero")
uploads, err := r.FindDataUploadsByPod(context.Background(), r.client, "velero")
if test.expectedError {
assert.Error(t, err)
@@ -995,3 +1008,110 @@ func TestFindDataUploads(t *testing.T) {
})
}
}
func TestAttemptDataUploadResume(t *testing.T) {
tests := []struct {
name string
dataUploads []velerov2alpha1api.DataUpload
du *velerov2alpha1api.DataUpload
pod *corev1.Pod
needErrs []bool
acceptedDataUploads []string
prepareddDataUploads []string
cancelledDataUploads []string
expectedError bool
}{
// Test case 1: Process Accepted DataUpload
{
name: "AcceptedDataUpload",
pod: builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Volumes(&corev1.Volume{Name: "dataupload-1"}).NodeName("node-1").Labels(map[string]string{
velerov1api.DataUploadLabel: dataUploadName,
}).Result(),
du: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseAccepted).Result(),
acceptedDataUploads: []string{dataUploadName},
expectedError: false,
},
// Test case 2: Cancel an Accepted DataUpload
{
name: "CancelAcceptedDataUpload",
du: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseAccepted).Result(),
},
// Test case 3: Process Accepted Prepared DataUpload
{
name: "PreparedDataUpload",
pod: builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Volumes(&corev1.Volume{Name: "dataupload-1"}).NodeName("node-1").Labels(map[string]string{
velerov1api.DataUploadLabel: dataUploadName,
}).Result(),
du: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhasePrepared).Result(),
prepareddDataUploads: []string{dataUploadName},
},
// Test case 4: Process Accepted InProgress DataUpload
{
name: "InProgressDataUpload",
pod: builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Volumes(&corev1.Volume{Name: "dataupload-1"}).NodeName("node-1").Labels(map[string]string{
velerov1api.DataUploadLabel: dataUploadName,
}).Result(),
du: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhasePrepared).Result(),
prepareddDataUploads: []string{dataUploadName},
},
// Test case 5: get resume error
{
name: "ResumeError",
pod: builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Volumes(&corev1.Volume{Name: "dataupload-1"}).NodeName("node-1").Labels(map[string]string{
velerov1api.DataUploadLabel: dataUploadName,
}).Result(),
needErrs: []bool{false, false, false, false, false, true},
du: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhasePrepared).Result(),
expectedError: true,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
ctx := context.TODO()
r, err := initDataUploaderReconciler(test.needErrs...)
r.nodeName = "node-1"
require.NoError(t, err)
defer func() {
r.client.Delete(ctx, test.du, &kbclient.DeleteOptions{})
if test.pod != nil {
r.client.Delete(ctx, test.pod, &kbclient.DeleteOptions{})
}
}()
assert.NoError(t, r.client.Create(ctx, test.du))
if test.pod != nil {
assert.NoError(t, r.client.Create(ctx, test.pod))
}
// Run the test
err = r.AttemptDataUploadResume(ctx, r.client, r.logger.WithField("name", test.name), test.du.Namespace)
if test.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
// Verify DataUploads marked as Cancelled
for _, duName := range test.cancelledDataUploads {
dataUpload := &velerov2alpha1api.DataUpload{}
err := r.client.Get(context.Background(), types.NamespacedName{Namespace: "velero", Name: duName}, dataUpload)
require.NoError(t, err)
assert.Equal(t, velerov2alpha1api.DataUploadPhaseCanceled, dataUpload.Status.Phase)
}
// Verify DataUploads marked as Accepted
for _, duName := range test.acceptedDataUploads {
dataUpload := &velerov2alpha1api.DataUpload{}
err := r.client.Get(context.Background(), types.NamespacedName{Namespace: "velero", Name: duName}, dataUpload)
require.NoError(t, err)
assert.Equal(t, velerov2alpha1api.DataUploadPhaseAccepted, dataUpload.Status.Phase)
}
// Verify DataUploads marked as Prepared
for _, duName := range test.prepareddDataUploads {
dataUpload := &velerov2alpha1api.DataUpload{}
err := r.client.Get(context.Background(), types.NamespacedName{Namespace: "velero", Name: duName}, dataUpload)
require.NoError(t, err)
assert.Equal(t, velerov2alpha1api.DataUploadPhasePrepared, dataUpload.Status.Phase)
}
}
})
}
}
@@ -303,7 +303,7 @@ func (c *PodVolumeRestoreReconciler) OnDataPathCompleted(ctx context.Context, na
// Write a done file with name=<restore-uid> into the just-created .velero dir
// within the volume. The velero init container on the pod is waiting
// for this file to exist in each restored volume before completing.
if err := os.WriteFile(filepath.Join(volumePath, ".velero", string(restoreUID)), nil, 0644); err != nil { //nolint:gosec
if err := os.WriteFile(filepath.Join(volumePath, ".velero", string(restoreUID)), nil, 0644); err != nil { //nolint:gosec // Internal usage. No need to check.
_, _ = c.errorOut(ctx, &pvr, err, "error writing done file", log)
return
}
+7 -2
View File
@@ -58,9 +58,14 @@ func genConfigmap(bak *velerov1.Backup, du velerov2alpha1.DataUpload) *corev1api
SnapshotID: du.Status.SnapshotID,
RepositoryType: GetUploaderType(du.Spec.DataMover),
}
b, _ := json.Marshal(snapshot)
b, err := json.Marshal(snapshot)
if err != nil {
return nil
}
data := make(map[string]string)
_ = json.Unmarshal(b, &data)
if err := json.Unmarshal(b, &data); err != nil {
return nil
}
return &corev1api.ConfigMap{
TypeMeta: metav1.TypeMeta{
APIVersion: corev1api.SchemeGroupVersion.String(),
+11
View File
@@ -41,6 +41,7 @@ type podTemplateConfig struct {
withSecret bool
defaultRepoMaintenanceFrequency time.Duration
garbageCollectionFrequency time.Duration
podVolumeOperationTimeout time.Duration
plugins []string
features []string
defaultVolumesToFsBackup bool
@@ -115,6 +116,12 @@ func WithGarbageCollectionFrequency(val time.Duration) podTemplateOption {
}
}
func WithPodVolumeOperationTimeout(val time.Duration) podTemplateOption {
return func(c *podTemplateConfig) {
c.podVolumeOperationTimeout = val
}
}
func WithPlugins(plugins []string) podTemplateOption {
return func(c *podTemplateConfig) {
c.plugins = plugins
@@ -212,6 +219,10 @@ func Deployment(namespace string, opts ...podTemplateOption) *appsv1.Deployment
args = append(args, fmt.Sprintf("--garbage-collection-frequency=%v", c.garbageCollectionFrequency))
}
if c.podVolumeOperationTimeout > 0 {
args = append(args, fmt.Sprintf("--fs-backup-timeout=%v", c.podVolumeOperationTimeout))
}
deployment := &appsv1.Deployment{
ObjectMeta: objectMeta(namespace, "velero"),
TypeMeta: metav1.TypeMeta{
+2
View File
@@ -246,6 +246,7 @@ type VeleroOptions struct {
VSLConfig map[string]string
DefaultRepoMaintenanceFrequency time.Duration
GarbageCollectionFrequency time.Duration
PodVolumeOperationTimeout time.Duration
Plugins []string
NoDefaultBackupLocation bool
CACertData []byte
@@ -335,6 +336,7 @@ func AllResources(o *VeleroOptions) *unstructured.UnstructuredList {
WithDefaultRepoMaintenanceFrequency(o.DefaultRepoMaintenanceFrequency),
WithServiceAccountName(serviceAccountName),
WithGarbageCollectionFrequency(o.GarbageCollectionFrequency),
WithPodVolumeOperationTimeout(o.PodVolumeOperationTimeout),
WithUploaderType(o.UploaderType),
}
+1 -1
View File
@@ -69,7 +69,7 @@ const (
// data mover metrics
DataUploadSuccessTotal = "data_upload_success_total"
DataUploadFailureTotal = "data_upload_failure_total"
DataUploadCancelTotal = "data_upload_cancel_total"
DataUploadCancelTotal = "data_upload_cancel_total" //nolint:gosec // Not a hard code secret.
DataDownloadSuccessTotal = "data_download_success_total"
DataDownloadFailureTotal = "data_download_failure_total"
DataDownloadCancelTotal = "data_download_cancel_total"
@@ -80,7 +80,7 @@ func (b *clientBuilder) clientConfig() *hcplugin.ClientConfig {
string(common.PluginKindDeleteItemAction): framework.NewDeleteItemActionPlugin(common.ClientLogger(b.clientLogger)),
},
Logger: b.pluginLogger,
Cmd: exec.Command(b.commandName, b.commandArgs...), //nolint
Cmd: exec.Command(b.commandName, b.commandArgs...), //nolint:gosec // Internal call. No need to check the command line.
}
}
+1
View File
@@ -199,6 +199,7 @@ func (r *restorer) RestorePodVolumes(data RestoreData) []error {
err = kube.IsPodScheduled(newObj)
if err != nil {
r.log.WithField("error", err).Debugf("Pod %s/%s is not scheduled yet", newObj.GetNamespace(), newObj.GetName())
return false, nil
}
return true, nil
+1 -1
View File
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
//nolint:gosec
//nolint:gosec // Internal usage. No need to check.
package config
import (
+1 -1
View File
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
//nolint:gosec
//nolint:gosec // Internal usage. No need to check.
package config
import "os"
+1 -1
View File
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
//nolint:gosec
//nolint:gosec // Internal call. No need to check.
package keys
import (
+1 -1
View File
@@ -77,7 +77,7 @@ func (c *Command) String() string {
// Cmd returns an exec.Cmd for the command.
func (c *Command) Cmd() *exec.Cmd {
parts := c.StringSlice()
cmd := exec.Command(parts[0], parts[1:]...) //nolint
cmd := exec.Command(parts[0], parts[1:]...) //nolint:gosec // Internal call. No need to check the parameter.
cmd.Dir = c.Dir
if len(c.Env) > 0 {
+61 -34
View File
@@ -325,6 +325,7 @@ func (kr *kubernetesRestorer) RestoreWithResolvers(
resourceModifiers: req.ResourceModifiers,
disableInformerCache: req.DisableInformerCache,
featureVerifier: kr.featureVerifier,
hookTracker: hook.NewHookTracker(),
}
return restoreCtx.execute()
@@ -377,6 +378,7 @@ type restoreContext struct {
resourceModifiers *resourcemodifiers.ResourceModifiers
disableInformerCache bool
featureVerifier features.Verifier
hookTracker *hook.HookTracker
}
type resourceClientKey struct {
@@ -544,6 +546,23 @@ func (ctx *restoreContext) execute() (results.Result, results.Result) {
errs.Merge(&e)
}
var createdOrUpdatedCRDs bool
for _, restoredItem := range ctx.restoredItems {
if restoredItem.action == itemRestoreResultCreated || restoredItem.action == itemRestoreResultUpdated {
createdOrUpdatedCRDs = true
break
}
}
// If we just restored custom resource definitions (CRDs), refresh
// discovery because the restored CRDs may have created or updated new APIs that
// didn't previously exist in the cluster, and we want to be able to
// resolve & restore instances of them in subsequent loop iterations.
if createdOrUpdatedCRDs {
if err := ctx.discoveryHelper.Refresh(); err != nil {
warnings.Add("", errors.Wrap(err, "refresh discovery after restoring CRDs"))
}
}
// Restore everything else
selectedResourceCollection, _, w, e := ctx.getOrderedResourceCollection(
backupResources,
@@ -629,11 +648,6 @@ func (ctx *restoreContext) execute() (results.Result, results.Result) {
updated.Status.Progress.TotalItems = len(ctx.restoredItems)
updated.Status.Progress.ItemsRestored = len(ctx.restoredItems)
err = kube.PatchResource(ctx.restore, updated, ctx.kbClient)
if err != nil {
ctx.log.WithError(errors.WithStack((err))).Warn("Updating restore status.progress")
}
// Wait for all of the pod volume restore goroutines to be done, which is
// only possible once all of their errors have been received by the loop
// below, then close the podVolumeErrs channel so the loop terminates.
@@ -668,6 +682,19 @@ func (ctx *restoreContext) execute() (results.Result, results.Result) {
}
ctx.log.Info("Done waiting for all post-restore exec hooks to complete")
// update hooks execution status
if updated.Status.HookStatus == nil {
updated.Status.HookStatus = &velerov1api.HookStatus{}
}
updated.Status.HookStatus.HooksAttempted, updated.Status.HookStatus.HooksFailed = ctx.hookTracker.Stat()
ctx.log.Infof("hookTracker: %+v, hookAttempted: %d, hookFailed: %d", ctx.hookTracker.GetTracker(), updated.Status.HookStatus.HooksAttempted, updated.Status.HookStatus.HooksFailed)
// patch the restore status
err = kube.PatchResource(ctx.restore, updated, ctx.kbClient)
if err != nil {
ctx.log.WithError(errors.WithStack((err))).Warn("Updating restore status")
}
return warnings, errs
}
@@ -686,6 +713,9 @@ func (ctx *restoreContext) processSelectedResource(
for namespace, selectedItems := range selectedResource.selectedItemsByNamespace {
for _, selectedItem := range selectedItems {
if groupResource == kuberesource.Namespaces {
namespace = selectedItem.name
}
// If we don't know whether this namespace exists yet, attempt to create
// it in order to ensure it exists. Try to get it from the backup tarball
// (in order to get any backed-up metadata), but if we don't find it there,
@@ -722,6 +752,10 @@ func (ctx *restoreContext) processSelectedResource(
// have to try to create them multiple times.
existingNamespaces.Insert(selectedItem.targetNamespace)
}
// For namespaces resources we don't need to following steps
if groupResource == kuberesource.Namespaces {
continue
}
obj, err := archive.Unmarshal(ctx.fileSystem, selectedItem.path)
if err != nil {
@@ -762,15 +796,6 @@ func (ctx *restoreContext) processSelectedResource(
}
}
// If we just restored custom resource definitions (CRDs), refresh
// discovery because the restored CRDs may have created new APIs that
// didn't previously exist in the cluster, and we want to be able to
// resolve & restore instances of them in subsequent loop iterations.
if groupResource == kuberesource.CustomResourceDefinitions {
if err := ctx.discoveryHelper.Refresh(); err != nil {
warnings.Add("", errors.Wrap(err, "refresh discovery after restoring CRDs"))
}
}
return processedItems, warnings, errs
}
@@ -1875,7 +1900,7 @@ func shouldRenamePV(ctx *restoreContext, obj *unstructured.Unstructured, client
// remapClaimRefNS remaps a PersistentVolume's claimRef.Namespace based on a
// restore's NamespaceMappings, if necessary. Returns true if the namespace was
// remapped, false if it was not required.
func remapClaimRefNS(ctx *restoreContext, obj *unstructured.Unstructured) (bool, error) { //nolint:unparam
func remapClaimRefNS(ctx *restoreContext, obj *unstructured.Unstructured) (bool, error) { //nolint:unparam // ignore the result 0 (bool) is never used warning.
if len(ctx.restore.Spec.NamespaceMapping) == 0 {
ctx.log.Debug("Persistent volume does not need to have the claimRef.namespace remapped because restore is not remapping any namespaces")
return false, nil
@@ -1963,6 +1988,7 @@ func (ctx *restoreContext) waitExec(createdObj *unstructured.Unstructured) {
ctx.resourceRestoreHooks,
pod,
ctx.log,
ctx.hookTracker,
)
if err != nil {
ctx.log.WithError(err).Errorf("error getting exec hooks for pod %s/%s", pod.Namespace, pod.Name)
@@ -1970,7 +1996,7 @@ func (ctx *restoreContext) waitExec(createdObj *unstructured.Unstructured) {
return
}
if errs := ctx.waitExecHookHandler.HandleHooks(ctx.hooksContext, ctx.log, pod, execHooksByContainer); len(errs) > 0 {
if errs := ctx.waitExecHookHandler.HandleHooks(ctx.hooksContext, ctx.log, pod, execHooksByContainer, ctx.hookTracker); len(errs) > 0 {
ctx.log.WithError(kubeerrs.NewAggregate(errs)).Error("unable to successfully execute post-restore hooks")
ctx.hooksCancelFunc()
@@ -2248,12 +2274,6 @@ func (ctx *restoreContext) getOrderedResourceCollection(
continue
}
// We don't want to explicitly restore namespace API objs because we'll handle
// them as a special case prior to restoring anything into them
if groupResource == kuberesource.Namespaces {
continue
}
// Check if the resource is present in the backup
resourceList := backupResources[groupResource.String()]
if resourceList == nil {
@@ -2269,24 +2289,17 @@ func (ctx *restoreContext) getOrderedResourceCollection(
continue
}
// get target namespace to restore into, if different
// from source namespace
targetNamespace := namespace
if target, ok := ctx.restore.Spec.NamespaceMapping[namespace]; ok {
targetNamespace = target
}
if targetNamespace == "" && boolptr.IsSetToFalse(ctx.restore.Spec.IncludeClusterResources) {
if namespace == "" && boolptr.IsSetToFalse(ctx.restore.Spec.IncludeClusterResources) {
ctx.log.Infof("Skipping resource %s because it's cluster-scoped", resource)
continue
}
if targetNamespace == "" && !boolptr.IsSetToTrue(ctx.restore.Spec.IncludeClusterResources) && !ctx.namespaceIncludesExcludes.IncludeEverything() {
if namespace == "" && !boolptr.IsSetToTrue(ctx.restore.Spec.IncludeClusterResources) && !ctx.namespaceIncludesExcludes.IncludeEverything() {
ctx.log.Infof("Skipping resource %s because it's cluster-scoped and only specific namespaces are included in the restore", resource)
continue
}
res, w, e := ctx.getSelectedRestoreableItems(groupResource.String(), targetNamespace, namespace, items)
res, w, e := ctx.getSelectedRestoreableItems(groupResource.String(), ctx.restore.Spec.NamespaceMapping, namespace, items)
warnings.Merge(&w)
errs.Merge(&e)
@@ -2302,7 +2315,7 @@ func (ctx *restoreContext) getOrderedResourceCollection(
// getSelectedRestoreableItems applies Kubernetes selectors on individual items
// of each resource type to create a list of items which will be actually
// restored.
func (ctx *restoreContext) getSelectedRestoreableItems(resource, targetNamespace, originalNamespace string, items []string) (restoreableResource, results.Result, results.Result) {
func (ctx *restoreContext) getSelectedRestoreableItems(resource string, namespaceMapping map[string]string, originalNamespace string, items []string) (restoreableResource, results.Result, results.Result) { //nolint:unparam // Ignore the warnings is always nil warning.
warnings, errs := results.Result{}, results.Result{}
restorable := restoreableResource{
@@ -2313,6 +2326,11 @@ func (ctx *restoreContext) getSelectedRestoreableItems(resource, targetNamespace
restorable.selectedItemsByNamespace = make(map[string][]restoreableItem)
}
targetNamespace := originalNamespace
if target, ok := namespaceMapping[originalNamespace]; ok {
targetNamespace = target
}
if targetNamespace != "" {
ctx.log.Infof("Resource '%s' will be restored into namespace '%s'", resource, targetNamespace)
} else {
@@ -2374,6 +2392,15 @@ func (ctx *restoreContext) getSelectedRestoreableItems(resource, targetNamespace
continue
}
if resource == kuberesource.Namespaces.String() {
// handle remapping for namespace resource
if target, ok := namespaceMapping[item]; ok {
targetNamespace = target
} else {
targetNamespace = item
}
}
selectedItem := restoreableItem{
path: itemPath,
name: item,
@@ -2403,7 +2430,7 @@ func removeRestoreLabels(obj metav1.Object) {
}
// updates the backup/restore labels
func (ctx *restoreContext) updateBackupRestoreLabels(fromCluster, fromClusterWithLabels *unstructured.Unstructured, namespace string, resourceClient client.Dynamic) (warnings, errs results.Result) {
func (ctx *restoreContext) updateBackupRestoreLabels(fromCluster, fromClusterWithLabels *unstructured.Unstructured, namespace string, resourceClient client.Dynamic) (warnings, errs results.Result) { //nolint:unparam // Ignore the warnings is nil warning.
patchBytes, err := generatePatch(fromCluster, fromClusterWithLabels)
if err != nil {
ctx.log.Errorf("error generating patch for %s %s: %v", fromCluster.GroupVersionKind().Kind, kube.NamespaceAndName(fromCluster), err)
+1 -1
View File
@@ -33,7 +33,7 @@ type Throttle struct {
func (t *Throttle) ShouldOutput() bool {
nextOutputTimeUnixNano := atomic.LoadInt64(&t.throttle)
if nowNano := time.Now().UnixNano(); nowNano > nextOutputTimeUnixNano { //nolint:forbidigo
if nowNano := time.Now().UnixNano(); nowNano > nextOutputTimeUnixNano {
if atomic.CompareAndSwapInt64(&t.throttle, nextOutputTimeUnixNano, nowNano+t.interval.Nanoseconds()) {
return true
}
+1 -1
View File
@@ -37,7 +37,7 @@ When naming your plugin, keep in mind that the full name needs to conform to the
- have two parts, prefix + name, separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same prefix + name cannot not already exist
- a plugin with the same prefix + name cannot already exist
### Some examples:
+1 -1
View File
@@ -15,7 +15,7 @@ When naming your plugin, keep in mind that the name needs to conform to these ru
- have two parts separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same name cannot not already exist
- a plugin with the same name cannot already exist
### Some examples:
+1 -1
View File
@@ -15,7 +15,7 @@ When naming your plugin, keep in mind that the name needs to conform to these ru
- have two parts separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same name cannot not already exist
- a plugin with the same name cannot already exist
### Some examples:
+1 -1
View File
@@ -37,7 +37,7 @@ When naming your plugin, keep in mind that the full name needs to conform to the
- have two parts, prefix + name, separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same prefix + name cannot not already exist
- a plugin with the same prefix + name cannot already exist
### Some examples:
+1 -1
View File
@@ -37,7 +37,7 @@ When naming your plugin, keep in mind that the full name needs to conform to the
- have two parts, prefix + name, separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same prefix + name cannot not already exist
- a plugin with the same prefix + name cannot already exist
### Some examples:
+1 -1
View File
@@ -15,7 +15,7 @@ When naming your plugin, keep in mind that the name needs to conform to these ru
- have two parts separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same name cannot not already exist
- a plugin with the same name cannot already exist
### Some examples:
+1 -1
View File
@@ -15,7 +15,7 @@ When naming your plugin, keep in mind that the name needs to conform to these ru
- have two parts separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same name cannot not already exist
- a plugin with the same name cannot already exist
### Some examples:
+1 -1
View File
@@ -15,7 +15,7 @@ When naming your plugin, keep in mind that the name needs to conform to these ru
- have two parts separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same name cannot not already exist
- a plugin with the same name cannot already exist
### Some examples:
+1 -1
View File
@@ -15,7 +15,7 @@ When naming your plugin, keep in mind that the name needs to conform to these ru
- have two parts separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same name cannot not already exist
- a plugin with the same name cannot already exist
### Some examples:
+1 -1
View File
@@ -15,7 +15,7 @@ When naming your plugin, keep in mind that the name needs to conform to these ru
- have two parts separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same name cannot not already exist
- a plugin with the same name cannot already exist
### Some examples:
+1 -1
View File
@@ -37,7 +37,7 @@ When naming your plugin, keep in mind that the full name needs to conform to the
- have two parts, prefix + name, separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same prefix + name cannot not already exist
- a plugin with the same prefix + name cannot already exist
### Some examples:
+1 -1
View File
@@ -37,7 +37,7 @@ When naming your plugin, keep in mind that the full name needs to conform to the
- have two parts, prefix + name, separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same prefix + name cannot not already exist
- a plugin with the same prefix + name cannot already exist
### Some examples:
+1 -1
View File
@@ -37,7 +37,7 @@ When naming your plugin, keep in mind that the full name needs to conform to the
- have two parts, prefix + name, separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same prefix + name cannot not already exist
- a plugin with the same prefix + name cannot already exist
### Some examples:
+1 -1
View File
@@ -37,7 +37,7 @@ When naming your plugin, keep in mind that the full name needs to conform to the
- have two parts, prefix + name, separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same prefix + name cannot not already exist
- a plugin with the same prefix + name cannot already exist
### Some examples:
+1 -1
View File
@@ -37,7 +37,7 @@ When naming your plugin, keep in mind that the full name needs to conform to the
- have two parts, prefix + name, separated by '/'
- none of the above parts can be empty
- the prefix is a valid DNS subdomain name
- a plugin with the same prefix + name cannot not already exist
- a plugin with the same prefix + name cannot already exist
### Some examples:
+2
View File
@@ -28,6 +28,7 @@ import (
"github.com/onsi/ginkgo/reporters"
. "github.com/onsi/gomega"
"github.com/vmware-tanzu/velero/pkg/cmd/cli/install"
. "github.com/vmware-tanzu/velero/test"
. "github.com/vmware-tanzu/velero/test/e2e/backup"
. "github.com/vmware-tanzu/velero/test/e2e/backups"
@@ -49,6 +50,7 @@ import (
)
func init() {
VeleroCfg.Options = &install.Options{}
flag.StringVar(&VeleroCfg.CloudProvider, "cloud-provider", "", "cloud that Velero will be installed into. Required.")
flag.StringVar(&VeleroCfg.ObjectStoreProvider, "object-store-provider", "", "provider of object store plugin. Required if cloud-provider is kind, otherwise ignored.")
flag.StringVar(&VeleroCfg.BSLBucket, "bucket", "", "name of the object storage bucket where backups from e2e tests should be stored. Required.")
+23 -1
View File
@@ -76,6 +76,17 @@ NFS_SERVER_PATH ?=
UPLOADER_TYPE ?=
TEST_CASE_DESCRIBE ?= 'velero performance test'
BACKUP_FOR_RESTORE ?=
Delete_Cluster_Resource ?= false
Debug_Velero_Pod_Restart ?= false
NODE_AGENT_POD_CPU_LIMIT ?= 4
NODE_AGENT_POD_MEM_LIMIT ?= 4Gi
NODE_AGENT_POD_CPU_REQUEST ?= 2
NODE_AGENT_POD_MEM_REQUEST ?= 2Gi
VELERO_POD_CPU_LIMIT ?= 4
VELERO_POD_MEM_LIMIT ?= 4Gi
VELERO_POD_CPU_REQUEST ?= 2
VELERO_POD_MEM_REQUEST ?= 2Gi
POD_VOLUME_OPERATION_TIMEOUT ?= 6h
.PHONY:ginkgo
ginkgo: # Make sure ginkgo is in $GOPATH/bin
@@ -110,7 +121,18 @@ run: ginkgo
-uploader-type=$(UPLOADER_TYPE) \
-nfs-server-path=$(NFS_SERVER_PATH) \
-test-case-describe=$(TEST_CASE_DESCRIBE) \
-backup-for-restore=$(BACKUP_FOR_RESTORE)
-backup-for-restore=$(BACKUP_FOR_RESTORE) \
-delete-cluster-resource=$(Delete_Cluster_Resource) \
-debug-velero-pod-restart=$(Debug_Velero_Pod_Restart) \
-node-agent-pod-cpu-limit=$(NODE_AGENT_POD_CPU_LIMIT) \
-node-agent-pod-mem-limit=$(NODE_AGENT_POD_MEM_LIMIT) \
-node-agent-pod-cpu-request=$(NODE_AGENT_POD_CPU_REQUEST) \
-node-agent-pod-mem-request=$(NODE_AGENT_POD_MEM_REQUEST) \
-velero-pod-cpu-limit=$(VELERO_POD_CPU_LIMIT) \
-velero-pod-mem-limit=$(VELERO_POD_MEM_LIMIT) \
-velero-pod-cpu-request=$(VELERO_POD_CPU_REQUEST) \
-velero-pod-mem-request=$(VELERO_POD_MEM_REQUEST) \
-pod-volume-operation-timeout=$(POD_VOLUME_OPERATION_TIMEOUT)
build: ginkgo
mkdir -p $(OUTPUT_DIR)
+1 -1
View File
@@ -32,7 +32,7 @@ type BackupTest struct {
func (b *BackupTest) Init() error {
b.TestCase.Init()
b.Ctx, b.CtxCancel = context.WithTimeout(context.Background(), 1*time.Hour)
b.Ctx, b.CtxCancel = context.WithTimeout(context.Background(), 6*time.Hour)
b.CaseBaseName = "backup"
b.BackupName = "backup-" + b.CaseBaseName + "-" + b.UUIDgen
+15 -3
View File
@@ -18,12 +18,14 @@ package basic
import (
"context"
"fmt"
"strings"
"time"
"github.com/pkg/errors"
. "github.com/vmware-tanzu/velero/test"
. "github.com/vmware-tanzu/velero/test/perf/test"
"github.com/vmware-tanzu/velero/test/util/k8s"
)
type BasicTest struct {
@@ -32,7 +34,7 @@ type BasicTest struct {
func (b *BasicTest) Init() error {
b.TestCase.Init()
b.Ctx, b.CtxCancel = context.WithTimeout(context.Background(), 1*time.Hour)
b.Ctx, b.CtxCancel = context.WithTimeout(context.Background(), 6*time.Hour)
b.CaseBaseName = "backuprestore"
b.BackupName = "backup-" + b.CaseBaseName + "-" + b.UUIDgen
b.RestoreName = "restore-" + b.CaseBaseName + "-" + b.UUIDgen
@@ -49,10 +51,20 @@ func (b *BasicTest) Init() error {
"--from-backup", b.BackupName, "--wait",
}
if !VeleroCfg.DeleteClusterResource {
joinedNsMapping, err := k8s.GetMappingNamespaces(b.Ctx, b.Client, *b.NSExcluded)
if err != nil {
return errors.Wrapf(err, "failed to get mapping namespaces in init")
}
b.RestoreArgs = append(b.RestoreArgs, "--namespace-mappings")
b.RestoreArgs = append(b.RestoreArgs, joinedNsMapping)
}
b.TestMsg = &TestMSG{
Desc: "Do backup and restore resources for performance test",
FailedMSG: "Failed to backup and restore resources",
Text: fmt.Sprintf("Should backup and restore resources success"),
Text: "Should backup and restore resources success",
}
return nil
}
+14
View File
@@ -21,12 +21,14 @@ import (
"flag"
"fmt"
"testing"
"time"
. "github.com/onsi/ginkgo"
"github.com/onsi/ginkgo/reporters"
. "github.com/onsi/gomega"
"github.com/pkg/errors"
"github.com/vmware-tanzu/velero/pkg/cmd/cli/install"
. "github.com/vmware-tanzu/velero/test"
"github.com/vmware-tanzu/velero/test/perf/backup"
@@ -39,6 +41,7 @@ import (
)
func init() {
VeleroCfg.Options = &install.Options{}
flag.StringVar(&VeleroCfg.CloudProvider, "cloud-provider", "", "cloud that Velero will be installed into. Required.")
flag.StringVar(&VeleroCfg.ObjectStoreProvider, "object-store-provider", "", "provider of object store plugin. Required if cloud-provider is kind, otherwise ignored.")
flag.StringVar(&VeleroCfg.BSLBucket, "bucket", "", "name of the object storage bucket where backups from e2e tests should be stored. Required.")
@@ -56,6 +59,15 @@ func init() {
flag.BoolVar(&VeleroCfg.InstallVelero, "install-velero", true, "install/uninstall velero during the test. Optional.")
flag.BoolVar(&VeleroCfg.UseNodeAgent, "use-node-agent", true, "whether deploy node agent daemonset velero during the test. Optional.")
flag.StringVar(&VeleroCfg.RegistryCredentialFile, "registry-credential-file", "", "file containing credential for the image registry, follows the same format rules as the ~/.docker/config.json file. Optional.")
flag.StringVar(&VeleroCfg.NodeAgentPodCPULimit, "node-agent-pod-cpu-limit", "4", "CPU limit for node agent pod. Optional.")
flag.StringVar(&VeleroCfg.NodeAgentPodMemLimit, "node-agent-pod-mem-limit", "4Gi", "Memory limit for node agent pod. Optional.")
flag.StringVar(&VeleroCfg.NodeAgentPodCPURequest, "node-agent-pod-cpu-request", "2", "CPU request for node agent pod. Optional.")
flag.StringVar(&VeleroCfg.NodeAgentPodMemRequest, "node-agent-pod-mem-request", "2Gi", "Memory request for node agent pod. Optional.")
flag.StringVar(&VeleroCfg.VeleroPodCPULimit, "velero-pod-cpu-limit", "4", "CPU limit for velero pod. Optional.")
flag.StringVar(&VeleroCfg.VeleroPodMemLimit, "velero-pod-mem-limit", "4Gi", "Memory limit for velero pod. Optional.")
flag.StringVar(&VeleroCfg.VeleroPodCPURequest, "velero-pod-cpu-request", "2", "CPU request for velero pod. Optional.")
flag.StringVar(&VeleroCfg.VeleroPodMemRequest, "velero-pod-mem-request", "2Gi", "Memory request for velero pod. Optional.")
flag.DurationVar(&VeleroCfg.PodVolumeOperationTimeout, "pod-volume-operation-timeout", 360*time.Minute, "Timeout for pod volume operations. Optional.")
//vmware-tanzu-experiments
flag.StringVar(&VeleroCfg.Features, "features", "", "Comma-separated list of features to enable for this Velero process.")
flag.StringVar(&VeleroCfg.DefaultCluster, "default-cluster-context", "", "Default cluster context for migration test.")
@@ -65,6 +77,8 @@ func init() {
flag.StringVar(&VeleroCfg.NFSServerPath, "nfs-server-path", "", "the path of nfs server")
flag.StringVar(&VeleroCfg.TestCaseDescribe, "test-case-describe", "velero performance test", "the description for the current test")
flag.StringVar(&VeleroCfg.BackupForRestore, "backup-for-restore", "", "the name of backup for restore")
flag.BoolVar(&VeleroCfg.DeleteClusterResource, "delete-cluster-resource", false, "delete cluster resource after test")
flag.BoolVar(&VeleroCfg.DebugVeleroPodRestart, "debug-velero-pod-restart", false, "Switch for debugging velero pod restart.")
}
func initConfig() error {
+24 -22
View File
@@ -20,6 +20,7 @@ import (
"context"
"fmt"
"strings"
"time"
"github.com/pkg/errors"
@@ -29,6 +30,7 @@ import (
)
const PodResourceDesc = "Resource consumption"
const PodMetricsTimeout = 5 * time.Minute
type PodMetrics struct {
Client *metricsclientset.Clientset
@@ -39,31 +41,31 @@ type PodMetrics struct {
}
func (p *PodMetrics) Update() error {
cpu, mem, err := metrics.GetPodUsageMetrics(p.Ctx, p.Client, p.PodName, p.Namespace)
cpu, mem, err := metrics.GetPodUsageMetrics(p.Ctx, p.Client, p.PodName, p.Namespace, PodMetricsTimeout)
if err != nil {
return errors.WithStack(err)
} else {
keyMaxCPU := p.PodName + ":MaxCPU"
curCPU := cpu.MilliValue()
if curCPU > p.Metrics[keyMaxCPU] {
p.Metrics[keyMaxCPU] = curCPU
}
keyMaxMem := p.PodName + ":MaxMemory"
curMem := mem.MilliValue()
if curMem > p.Metrics[keyMaxMem] {
p.Metrics[keyMaxMem] = curMem
}
keyAvgCPU := p.PodName + ":AverageCPU"
preAvgCPU := p.Metrics[keyAvgCPU]
p.Metrics[keyAvgCPU] = (preAvgCPU*p.count + curCPU) / (p.count + 1)
keyAvgMem := p.PodName + ":AverageMemory"
preAvgMem := p.Metrics[keyAvgMem]
p.Metrics[keyAvgMem] = (preAvgMem*p.count + curMem) / (p.count + 1)
p.count++
}
keyMaxCPU := p.PodName + ":MaxCPU"
curCPU := cpu.MilliValue()
if curCPU > p.Metrics[keyMaxCPU] {
p.Metrics[keyMaxCPU] = curCPU
}
keyMaxMem := p.PodName + ":MaxMemory"
curMem := mem.MilliValue()
if curMem > p.Metrics[keyMaxMem] {
p.Metrics[keyMaxMem] = curMem
}
keyAvgCPU := p.PodName + ":AverageCPU"
preAvgCPU := p.Metrics[keyAvgCPU]
p.Metrics[keyAvgCPU] = (preAvgCPU*p.count + curCPU) / (p.count + 1)
keyAvgMem := p.PodName + ":AverageMemory"
preAvgMem := p.Metrics[keyAvgMem]
p.Metrics[keyAvgMem] = (preAvgMem*p.count + curMem) / (p.count + 1)
p.count++
return nil
}
+29 -16
View File
@@ -16,40 +16,53 @@ limitations under the License.
package metrics
import "time"
import (
"fmt"
"time"
)
const TimeCaseDesc = "Time cost"
type TimeSpan struct {
Start time.Time
End time.Time
}
type TimeMetrics struct {
Name string
TimeInfo map[string]time.Time // metric name : start timestamp
Metrics map[string]float64 // metric name : time duration
TimeInfo map[string]TimeSpan // metric name : start timestamp
}
func (t *TimeMetrics) GetMetrics() map[string]string {
tmpMetrics := make(map[string]string)
for k, v := range t.Metrics {
duration := time.Duration(v) * time.Second
tmpMetrics[k] = duration.String()
for k, v := range t.TimeInfo {
duration := v.End.Sub(v.Start)
if duration < time.Second {
// For those too shoter time difference we should ignored
// as it may not really execute the logic
continue
}
tmpMetrics[k] = duration.String() + fmt.Sprintf(" (%s - %s)", v.Start.Format(time.RFC3339), v.End.Format(time.RFC3339))
}
return tmpMetrics
}
func (t *TimeMetrics) Start(name string) {
t.TimeInfo[name] = time.Now()
}
func (t *TimeMetrics) End(name string) {
t.Metrics[name] = time.Now().Sub(t.TimeInfo[name]).Seconds()
if t.Metrics[name] < 1 {
// For those too shoter time difference we should ignored
// as it may not really execute the logic
delete(t.Metrics, name)
t.TimeInfo[name] = TimeSpan{
Start: time.Now(),
}
}
func (t *TimeMetrics) End(name string) {
if _, ok := t.TimeInfo[name]; !ok {
return
}
timeSpan := t.TimeInfo[name]
timeSpan.End = time.Now()
t.TimeInfo[name] = timeSpan
}
func (t *TimeMetrics) Update() error {
t.Metrics[t.Name] = time.Now().Sub(t.TimeInfo[t.Name]).Seconds()
return nil
}
+18 -2
View File
@@ -25,6 +25,7 @@ import (
. "github.com/vmware-tanzu/velero/test"
. "github.com/vmware-tanzu/velero/test/perf/test"
"github.com/vmware-tanzu/velero/test/util/k8s"
. "github.com/vmware-tanzu/velero/test/util/velero"
)
@@ -34,7 +35,7 @@ type RestoreTest struct {
func (r *RestoreTest) Init() error {
r.TestCase.Init()
r.Ctx, r.CtxCancel = context.WithTimeout(context.Background(), 1*time.Hour)
r.Ctx, r.CtxCancel = context.WithTimeout(context.Background(), 6*time.Hour)
r.CaseBaseName = "restore"
r.RestoreName = "restore-" + r.CaseBaseName + "-" + r.UUIDgen
@@ -43,7 +44,7 @@ func (r *RestoreTest) Init() error {
FailedMSG: "Failed to restore resources",
Text: fmt.Sprintf("Should restore resources success"),
}
return r.clearUpResourcesBeforRestore()
return nil
}
func (r *RestoreTest) clearUpResourcesBeforRestore() error {
@@ -52,6 +53,11 @@ func (r *RestoreTest) clearUpResourcesBeforRestore() error {
}
func (r *RestoreTest) Restore() error {
// we need to clear up all resources before do the restore test
err := r.clearUpResourcesBeforRestore()
if err != nil {
return errors.Wrapf(err, "failed to clear up resources before do the restore test")
}
var backupName string
if VeleroCfg.BackupForRestore != "" {
backupName = VeleroCfg.BackupForRestore
@@ -71,6 +77,16 @@ func (r *RestoreTest) Restore() error {
"--from-backup", r.BackupName, "--wait",
}
if !VeleroCfg.DeleteClusterResource {
joinedNsMapping, err := k8s.GetMappingNamespaces(r.Ctx, r.Client, *r.NSExcluded)
if err != nil {
return errors.Wrapf(err, "failed to get mapping namespaces in init")
}
r.RestoreArgs = append(r.RestoreArgs, "--namespace-mappings")
r.RestoreArgs = append(r.RestoreArgs, joinedNsMapping)
}
return r.TestCase.Restore()
}
func (r *RestoreTest) Destroy() error {
+13 -11
View File
@@ -97,14 +97,15 @@ func TestFunc(test VeleroBackupRestoreTest) func() {
}
func (t *TestCase) Init() error {
t.Ctx, t.CtxCancel = context.WithTimeout(context.Background(), 1*time.Hour)
t.Ctx, t.CtxCancel = context.WithTimeout(context.Background(), 6*time.Hour)
t.NSExcluded = &[]string{"kube-system", "velero", "default", "kube-public", "kube-node-lease"}
t.UUIDgen = t.GenerateUUID()
t.Client = *VeleroCfg.DefaultClient
t.timer = &metrics.TimeMetrics{
Name: "Total time cost",
TimeInfo: map[string]time.Time{"Total time cost": time.Now()},
Metrics: make(map[string]float64),
Name: "Total time cost",
TimeInfo: map[string]metrics.TimeSpan{"Total time cost": {
Start: time.Now(),
}},
}
return nil
}
@@ -131,10 +132,12 @@ func (t *TestCase) Backup() error {
}
func (t *TestCase) Destroy() error {
By(fmt.Sprintf("Start to destroy namespace %s......", t.CaseBaseName), func() {
Expect(CleanupNamespacesFiterdByExcludes(t.GetTestCase().Ctx, t.Client, *t.NSExcluded)).To(Succeed(), "Could cleanup retrieve namespaces")
Expect(ClearClaimRefForFailedPVs(t.Ctx, t.Client)).To(Succeed(), "Failed to make PV status become to available")
})
if VeleroCfg.DeleteClusterResource {
By(fmt.Sprintf("Start to destroy namespace %s......", t.CaseBaseName), func() {
Expect(CleanupNamespacesFiterdByExcludes(t.GetTestCase().Ctx, t.Client, *t.NSExcluded)).To(Succeed(), "Could cleanup retrieve namespaces")
Expect(ClearClaimRefForFailedPVs(t.Ctx, t.Client)).To(Succeed(), "Failed to make PV status become to available")
})
}
return nil
}
@@ -160,7 +163,7 @@ func (t *TestCase) Verify() error {
}
func (t *TestCase) Clean() error {
if !VeleroCfg.Debug {
if !VeleroCfg.Debug || VeleroCfg.DeleteClusterResource {
By("Clean backups and restore after test", func() {
if len(t.BackupArgs) != 0 {
if err := VeleroBackupDelete(t.Ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, t.BackupName); err != nil {
@@ -269,8 +272,7 @@ func (t *TestCase) MonitorMetircs(ctx context.Context, collectors *metrics.Metri
timeMetrics := &metrics.TimeMetrics{
Name: t.CaseBaseName,
TimeInfo: make(map[string]time.Time),
Metrics: make(map[string]float64),
TimeInfo: make(map[string]metrics.TimeSpan),
}
collectors.RegisterOneTimeMetric(timeMetrics)
+6 -9
View File
@@ -21,6 +21,7 @@ import (
"github.com/google/uuid"
"github.com/vmware-tanzu/velero/pkg/cmd/cli/install"
. "github.com/vmware-tanzu/velero/test/util/k8s"
)
@@ -40,6 +41,7 @@ var ReportData *Report
type VeleroConfig struct {
VeleroCfgInPerf
*install.Options
VeleroCLI string
VeleroImage string
VeleroVersion string
@@ -66,7 +68,6 @@ type VeleroConfig struct {
AddBSLPlugins string
InstallVelero bool
KibishiiDirectory string
Features string
Debug bool
GCFrequency string
DefaultCluster string
@@ -74,12 +75,7 @@ type VeleroConfig struct {
ClientToInstallVelero *TestClient
DefaultClient *TestClient
StandbyClient *TestClient
UploaderType string
UseNodeAgent bool
UseRestic bool
ProvideSnapshotsVolumeParam bool
DefaultVolumesToFsBackup bool
UseVolumeSnapshots bool
VeleroServerDebugMode bool
SnapshotMoveData bool
DataMoverPlugin string
@@ -90,9 +86,10 @@ type VeleroConfig struct {
}
type VeleroCfgInPerf struct {
NFSServerPath string
TestCaseDescribe string
BackupForRestore string
NFSServerPath string
TestCaseDescribe string
BackupForRestore string
DeleteClusterResource bool
}
type SnapshotCheckPoint struct {
+39
View File
@@ -194,3 +194,42 @@ func NamespaceShouldNotExist(ctx context.Context, client TestClient, namespace s
}
return nil
}
func GetBackupNamespaces(ctx context.Context, client TestClient, excludeNS []string) ([]string, error) {
namespaces, err := client.ClientGo.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
if err != nil {
return nil, errors.Wrap(err, "Could not retrieve namespaces")
}
var backupNamespaces []string
for _, checkNamespace := range namespaces.Items {
isExclude := false
for k := range excludeNS {
if checkNamespace.Name == excludeNS[k] {
isExclude = true
}
}
if !isExclude {
backupNamespaces = append(backupNamespaces, checkNamespace.Name)
}
}
return backupNamespaces, nil
}
func GetMappingNamespaces(ctx context.Context, client TestClient, excludeNS []string) (string, error) {
ns, err := GetBackupNamespaces(ctx, client, excludeNS)
if err != nil {
return "", errors.Wrap(err, "Could not retrieve namespaces")
} else if len(ns) == 0 {
return "", errors.Wrap(err, "Get empty namespaces in backup")
}
nsMapping := []string{}
for _, n := range ns {
nsMapping = append(nsMapping, n+":mapping-"+n)
}
joinedNsMapping := strings.Join(nsMapping, ",")
if len(joinedNsMapping) > 0 {
joinedNsMapping = joinedNsMapping[:len(joinedNsMapping)-1]
}
return joinedNsMapping, nil
}
+17 -3
View File
@@ -18,21 +18,35 @@ package metrics
import (
"context"
"time"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/metrics/pkg/apis/metrics/v1beta1"
metricsclientset "k8s.io/metrics/pkg/client/clientset/versioned"
)
func GetPodUsageMetrics(ctx context.Context, metricsClient *metricsclientset.Clientset, podName, namespace string) (cpuUsage, memoryUsage resource.Quantity, err error) {
func GetPodUsageMetrics(ctx context.Context, metricsClient *metricsclientset.Clientset, podName, namespace string, podMetricsTimeout time.Duration) (cpuUsage, memoryUsage resource.Quantity, err error) {
ctx, cancel := context.WithTimeout(context.Background(), podMetricsTimeout)
defer cancel()
var podMetrics *v1beta1.PodMetrics
podMetrics, err = metricsClient.MetricsV1beta1().PodMetricses(namespace).Get(ctx, podName, metav1.GetOptions{})
err = wait.PollImmediateUntil(time.Second, func() (bool, error) {
var err error
podMetrics, err = metricsClient.MetricsV1beta1().PodMetricses(namespace).Get(ctx, podName, metav1.GetOptions{})
if err != nil {
return false, nil
}
return true, nil
}, ctx.Done())
if err != nil {
return
} else if podMetrics == nil {
return cpuUsage, memoryUsage, nil
}
// Variables to store the max and sum of CPU and memory usage
// For velero pod we only return the main container
for _, container := range podMetrics.Containers {
+45
View File
@@ -120,6 +120,15 @@ func VeleroInstall(ctx context.Context, veleroCfg *VeleroConfig, isStandbyCluste
veleroInstallOptions.UploaderType = veleroCfg.UploaderType
GCFrequency, _ := time.ParseDuration(veleroCfg.GCFrequency)
veleroInstallOptions.GarbageCollectionFrequency = GCFrequency
veleroInstallOptions.PodVolumeOperationTimeout = veleroCfg.PodVolumeOperationTimeout
veleroInstallOptions.NodeAgentPodCPULimit = veleroCfg.NodeAgentPodCPULimit
veleroInstallOptions.NodeAgentPodCPURequest = veleroCfg.NodeAgentPodCPURequest
veleroInstallOptions.NodeAgentPodMemLimit = veleroCfg.NodeAgentPodMemLimit
veleroInstallOptions.NodeAgentPodMemRequest = veleroCfg.NodeAgentPodMemRequest
veleroInstallOptions.VeleroPodCPULimit = veleroCfg.VeleroPodCPULimit
veleroInstallOptions.VeleroPodCPURequest = veleroCfg.VeleroPodCPURequest
veleroInstallOptions.VeleroPodMemLimit = veleroCfg.VeleroPodMemLimit
veleroInstallOptions.VeleroPodMemRequest = veleroCfg.VeleroPodMemRequest
err = installVeleroServer(ctx, veleroCfg.VeleroCLI, veleroCfg.CloudProvider, &installOptions{
Options: veleroInstallOptions,
@@ -251,6 +260,42 @@ func installVeleroServer(ctx context.Context, cli, cloudProvider string, options
args = append(args, fmt.Sprintf("--garbage-collection-frequency=%v", options.GarbageCollectionFrequency))
}
if options.PodVolumeOperationTimeout > 0 {
args = append(args, fmt.Sprintf("--pod-volume-operation-timeout=%v", options.PodVolumeOperationTimeout))
}
if options.NodeAgentPodCPULimit != "" {
args = append(args, fmt.Sprintf("--node-agent-pod-cpu-limit=%v", options.NodeAgentPodCPULimit))
}
if options.NodeAgentPodCPURequest != "" {
args = append(args, fmt.Sprintf("--node-agent-pod-cpu-request=%v", options.NodeAgentPodCPURequest))
}
if options.NodeAgentPodMemLimit != "" {
args = append(args, fmt.Sprintf("--node-agent-pod-mem-limit=%v", options.NodeAgentPodMemLimit))
}
if options.NodeAgentPodMemRequest != "" {
args = append(args, fmt.Sprintf("--node-agent-pod-mem-request=%v", options.NodeAgentPodMemRequest))
}
if options.VeleroPodCPULimit != "" {
args = append(args, fmt.Sprintf("--velero-pod-cpu-limit=%v", options.VeleroPodCPULimit))
}
if options.VeleroPodCPURequest != "" {
args = append(args, fmt.Sprintf("--velero-pod-cpu-request=%v", options.VeleroPodCPURequest))
}
if options.VeleroPodMemLimit != "" {
args = append(args, fmt.Sprintf("--velero-pod-mem-limit=%v", options.VeleroPodMemLimit))
}
if options.VeleroPodMemRequest != "" {
args = append(args, fmt.Sprintf("--velero-pod-mem-request=%v", options.VeleroPodMemRequest))
}
if len(options.UploaderType) > 0 {
args = append(args, fmt.Sprintf("--uploader-type=%v", options.UploaderType))
}