mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-09-03 14:47:08 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Support pause/unpause schedules

Browse Source 
Support pause/unpause schedule

Fixes #2363

Signed-off-by: Wenkai Yin(尹文开) <yinw@vmware.com>
This commit is contained in:
Wenkai Yin(尹文开)
2022-09-15 10:42:48 +08:00
parent 100d6b4430
commit 4b9dbfa416
21 changed files with 440 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/util/kube/periodical_enqueue_source.go
+11 -11
View File
@@ -23,13 +23,13 @@ import (
"github.com/sirupsen/logrus"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/util/workqueue"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/handler"
"sigs.k8s.io/controller-runtime/pkg/predicate"
)
@@ -61,11 +61,11 @@ type PeriodicalEnqueueSource struct {
}
type PeriodicalEnqueueSourceOption struct {
FilterFuncs []func(object client.Object) bool
OrderFunc func(objList client.ObjectList) client.ObjectList
OrderFunc func(objList client.ObjectList) client.ObjectList
}
func (p *PeriodicalEnqueueSource) Start(ctx context.Context, h handler.EventHandler, q workqueue.RateLimitingInterface, pre ...predicate.Predicate) error {
// Start enqueue items periodically. The predicates only apply to the GenericEvent
func (p *PeriodicalEnqueueSource) Start(ctx context.Context, h handler.EventHandler, q workqueue.RateLimitingInterface, predicates ...predicate.Predicate) error {
go wait.Until(func() {
p.logger.Debug("enqueueing resources ...")
if err := p.List(ctx, p.objList); err != nil {
@@ -80,19 +80,19 @@ func (p *PeriodicalEnqueueSource) Start(ctx context.Context, h handler.EventHand
p.objList = p.option.OrderFunc(p.objList)
}
if err := meta.EachListItem(p.objList, func(object runtime.Object) error {
obj, ok := object.(metav1.Object)
obj, ok := object.(client.Object)
if !ok {
p.logger.Error("%s's type isn't metav1.Object", object.GetObjectKind().GroupVersionKind().String())
return nil
}
for _, filter := range p.option.FilterFuncs {
if filter != nil {
if enqueueObj := filter(object.(client.Object)); !enqueueObj {
p.logger.Debugf("skip enqueue object %s/%s due to filter function.", obj.GetNamespace(), obj.GetName())
return nil
}
event := event.GenericEvent{Object: obj}
for _, predicate := range predicates {
if !predicate.Generic(event) {
p.logger.Debugf("skip enqueue object %s/%s due to the predicate.", obj.GetNamespace(), obj.GetName())
return nil
}
}
q.Add(ctrl.Request{
NamespacedName: types.NamespacedName{
Namespace: obj.GetNamespace(),
pkg/util/kube/periodical_enqueue_source_test.go
+6 -8
View File
@@ -68,7 +68,7 @@ func TestStart(t *testing.T) {
require.Equal(t, 0, queue.Len())
}
func TestFilter(t *testing.T) {
func TestPredicate(t *testing.T) {
require.Nil(t, velerov1.AddToScheme(scheme.Scheme))
ctx, cancelFunc := context.WithCancel(context.TODO())
@@ -79,15 +79,13 @@ func TestFilter(t *testing.T) {
client,
&velerov1.BackupStorageLocationList{},
1*time.Second,
PeriodicalEnqueueSourceOption{
FilterFuncs: []func(object crclient.Object) bool{func(object crclient.Object) bool {
location := object.(*velerov1.BackupStorageLocation)
return storage.IsReadyToValidate(location.Spec.ValidationFrequency, location.Status.LastValidationTime, 1*time.Minute, logrus.WithContext(ctx).WithField("BackupStorageLocation", location.Name))
}},
},
PeriodicalEnqueueSourceOption{},
)
require.Nil(t, source.Start(ctx, nil, queue))
require.Nil(t, source.Start(ctx, nil, queue, NewGenericEventPredicate(func(object crclient.Object) bool {
location := object.(*velerov1.BackupStorageLocation)
return storage.IsReadyToValidate(location.Spec.ValidationFrequency, location.Status.LastValidationTime, 1*time.Minute, logrus.WithContext(ctx).WithField("BackupStorageLocation", location.Name))
})))
// Should not patch a backup storage location object status phase
// if the location's validation frequency is specifically set to zero
pkg/util/kube/predicate.go
+51
View File
@@ -19,6 +19,7 @@ package kube
import (
"reflect"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/predicate"
)
@@ -45,3 +46,53 @@ func (SpecChangePredicate) Update(e event.UpdateEvent) bool {
newSpec := reflect.ValueOf(e.ObjectNew).Elem().FieldByName("Spec")
return !reflect.DeepEqual(oldSpec.Interface(), newSpec.Interface())
}
// NewGenericEventPredicate creates a new Predicate that checks the Generic event with the provided func
func NewGenericEventPredicate(f func(object client.Object) bool) predicate.Predicate {
return predicate.Funcs{
GenericFunc: func(event event.GenericEvent) bool {
return f(event.Object)
},
}
}
// NewAllEventPredicate creates a new Predicate that checks all the events with the provided func
func NewAllEventPredicate(f func(object client.Object) bool) predicate.Predicate {
return predicate.Funcs{
CreateFunc: func(event event.CreateEvent) bool {
return f(event.Object)
},
DeleteFunc: func(event event.DeleteEvent) bool {
return f(event.Object)
},
UpdateFunc: func(event event.UpdateEvent) bool {
return f(event.ObjectNew)
},
GenericFunc: func(event event.GenericEvent) bool {
return f(event.Object)
},
}
}
// FalsePredicate always returns false for all kinds of events
type FalsePredicate struct{}
// Create always returns false
func (f FalsePredicate) Create(event.CreateEvent) bool {
return false
}
// Delete always returns false
func (f FalsePredicate) Delete(event.DeleteEvent) bool {
return false
}
// Update always returns false
func (f FalsePredicate) Update(event.UpdateEvent) bool {
return false
}
// Generic always returns false
func (f FalsePredicate) Generic(event.GenericEvent) bool {
return false
}
pkg/util/kube/predicate_test.go
+19
View File
@@ -178,3 +178,22 @@ func TestSpecChangePredicate(t *testing.T) {
})
}
}
func TestNewGenericEventPredicate(t *testing.T) {
predicate := NewGenericEventPredicate(func(object client.Object) bool {
return false
})
assert.False(t, predicate.Generic(event.GenericEvent{}))
}
func TestNewAllEventPredicate(t *testing.T) {
predicate := NewAllEventPredicate(func(object client.Object) bool {
return false
})
assert.False(t, predicate.Create(event.CreateEvent{}))
assert.False(t, predicate.Update(event.UpdateEvent{}))
assert.False(t, predicate.Delete(event.DeleteEvent{}))
assert.False(t, predicate.Generic(event.GenericEvent{}))
}