Skip to content

Commit a0f2385

Browse files
r-yabyabr-yabyab
authored
Implement watch for packagerevisionresources (#388)
* Enable watch events for packagerevisionresources Change selector * Create generic watch * Edit comments Edit comments Edit comments * Unit test for packagerevisionresources watch Unit test for packagerevisionresources watch Edit comments Edit test case * packagerevisionresources test coverage
1 parent 1bb33e6 commit a0f2385

3 files changed

Lines changed: 92 additions & 13 deletions

File tree

‎pkg/registry/porch/packagerevisionresources.go‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import (
2929
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3030
"k8s.io/apimachinery/pkg/runtime"
3131
"k8s.io/apimachinery/pkg/types"
32+
"k8s.io/apimachinery/pkg/watch"
3233
genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
3334
"k8s.io/apiserver/pkg/registry/rest"
3435
"k8s.io/klog/v2"
@@ -45,6 +46,7 @@ var _ rest.Getter = &packageRevisionResources{}
4546
var _ rest.Scoper = &packageRevisionResources{}
4647
var _ rest.Updater = &packageRevisionResources{}
4748
var _ rest.SingularNameProvider = &packageRevisionResources{}
49+
var _ rest.Watcher = &packageRevisionResources{}
4850

4951
// GetSingularName implements the SingularNameProvider interface
5052
func (r *packageRevisionResources) GetSingularName() string {
@@ -202,3 +204,25 @@ func (r *packageRevisionResources) Update(ctx context.Context, name string, objI
202204

203205
return created, false, nil
204206
}
207+
208+
// Watch supports watching for PackageRevisionResources changes.
209+
func (r *packageRevisionResources) Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error) {
210+
ctx, span := tracer.Start(ctx, "[START]::packageRevisionResources::Watch", trace.WithAttributes())
211+
defer span.End()
212+
213+
filter, err := parsePackageRevisionResourcesFieldSelector(options)
214+
if err != nil {
215+
return nil, err
216+
}
217+
218+
if namespace, namespaced := genericapirequest.NamespaceFrom(ctx); namespaced {
219+
if filter.Key.RKey().Namespace != "" && namespace != filter.Key.RKey().Namespace {
220+
return nil, fmt.Errorf("conflicting namespaces specified: %q and %q", namespace, filter.Key.RKey().Namespace)
221+
}
222+
filter.Key.PkgKey.RepoKey.Namespace = namespace
223+
}
224+
225+
return createGenericWatch(ctx, r, *filter, func(ctx context.Context, pr repository.PackageRevision) (runtime.Object, error) {
226+
return pr.GetResources(ctx)
227+
})
228+
}

‎pkg/registry/porch/packagerevisionresources_test.go‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import (
3030
"k8s.io/apimachinery/pkg/apis/meta/internalversion"
3131
"k8s.io/apimachinery/pkg/fields"
3232
"k8s.io/apimachinery/pkg/runtime"
33+
genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
3334
"sigs.k8s.io/controller-runtime/pkg/client"
3435
)
3536

@@ -92,3 +93,35 @@ func TestListResources(t *testing.T) {
9293
assert.True(t, isList)
9394
assert.Equal(t, 0, len(resultList.Items))
9495
}
96+
97+
func TestWatchResources(t *testing.T) {
98+
_, mockEngine := setupResourcesTest(t)
99+
mockWatcherManager := mockengine.NewMockWatcherManager(t)
100+
mockEngine.On("ObjectCache").Return(mockWatcherManager).Maybe()
101+
102+
mockWatcherManager.On("WatchPackageRevisions", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe()
103+
mockEngine.On("ListPackageRevisions", mock.Anything, mock.Anything, mock.Anything).Return([]repository.PackageRevision{}, nil).Maybe()
104+
105+
watcher, err := packagerevisionresources.Watch(context.TODO(), &internalversion.ListOptions{})
106+
assert.NoError(t, err)
107+
if watcher != nil {
108+
watcher.Stop()
109+
}
110+
111+
//=========================================================================================
112+
113+
watcher, err = packagerevisionresources.Watch(context.TODO(), &internalversion.ListOptions{
114+
FieldSelector: fields.OneTermEqualSelector("invalid.field", "somethingOffTheWall"),
115+
})
116+
assert.Equal(t, nil, watcher)
117+
assert.ErrorContains(t, err, "unknown fieldSelector field")
118+
119+
//=========================================================================================
120+
121+
ctxWithConflictNamespace := genericapirequest.WithNamespace(context.TODO(), "foo")
122+
watcher, err = packagerevisionresources.Watch(ctxWithConflictNamespace, &internalversion.ListOptions{
123+
FieldSelector: fields.OneTermEqualSelector("metadata.namespace", "somethingOffTheWall"),
124+
})
125+
assert.Equal(t, nil, watcher)
126+
assert.ErrorContains(t, err, "conflicting namespaces specified")
127+
}

‎pkg/registry/porch/watch.go‎

Lines changed: 35 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -23,11 +23,27 @@ import (
2323
"github.com/nephio-project/porch/pkg/repository"
2424
"go.opentelemetry.io/otel/trace"
2525
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
26+
"k8s.io/apimachinery/pkg/runtime"
2627
"k8s.io/apimachinery/pkg/watch"
2728
genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
2829
"k8s.io/klog/v2"
2930
)
3031

32+
// createGenericWatch creates a watch.Interface that monitors package changes.
33+
func createGenericWatch(ctx context.Context, r packageReader, filter repository.ListPackageRevisionFilter, extractor objectExtractor) (watch.Interface, error) {
34+
ctx, cancel := context.WithCancel(ctx)
35+
36+
w := &watcher{
37+
cancel: cancel,
38+
resultChan: make(chan watch.Event, 64),
39+
extractor: extractor,
40+
}
41+
42+
go w.listAndWatch(ctx, r, filter)
43+
44+
return w, nil
45+
}
46+
3147
// Watch supports watching for changes.
3248
func (r *packageRevisions) Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error) {
3349
// 'label' selects on labels; 'field' selects on the object's fields. Not all fields
@@ -49,16 +65,9 @@ func (r *packageRevisions) Watch(ctx context.Context, options *metainternalversi
4965
}
5066
}
5167

52-
ctx, cancel := context.WithCancel(ctx)
53-
54-
w := &watcher{
55-
cancel: cancel,
56-
resultChan: make(chan watch.Event, 64),
57-
}
58-
59-
go w.listAndWatch(ctx, r, *filter)
60-
61-
return w, nil
68+
return createGenericWatch(ctx, r, *filter, func(ctx context.Context, pr repository.PackageRevision) (runtime.Object, error) {
69+
return pr.GetPackageRevision(ctx)
70+
})
6271
}
6372

6473
// watcher implements watch.Interface, and holds the state for an active watch.
@@ -72,6 +81,9 @@ type watcher struct {
7281
eventCallback func(eventType watch.EventType, pr repository.PackageRevision) bool
7382
done bool
7483
totalSent int
84+
85+
// objectExtractor function to get the appropriate object from PackageRevision
86+
extractor objectExtractor
7587
}
7688

7789
var _ watch.Interface = &watcher{}
@@ -94,10 +106,20 @@ type packageReader interface {
94106
listPackageRevisions(ctx context.Context, filter repository.ListPackageRevisionFilter, callback func(ctx context.Context, p repository.PackageRevision) error) error
95107
}
96108

109+
// objectExtractor transforms a repository.PackageRevision into the appropriate
110+
// resource (PackageRevision or PackageRevisionResources).
111+
type objectExtractor func(ctx context.Context, pr repository.PackageRevision) (runtime.Object, error)
112+
97113
// listAndWatch implements watch by doing a list, then sending any observed changes.
98114
// This is not a compliant implementation of watch, but it is a good-enough start for most controllers.
99115
// One trick is that we start the watch _before_ we perform the list, so we don't miss changes that happen immediately after the list.
100116
func (w *watcher) listAndWatch(ctx context.Context, r packageReader, filter repository.ListPackageRevisionFilter) {
117+
if w.extractor == nil {
118+
w.extractor = func(ctx context.Context, pr repository.PackageRevision) (runtime.Object, error) {
119+
return pr.GetPackageRevision(ctx)
120+
}
121+
}
122+
101123
if err := w.listAndWatchInner(ctx, r, filter); err != nil {
102124
// TODO: We need to populate the object on this error
103125
klog.Warningf("sending error to watch stream: %v", err)
@@ -121,7 +143,7 @@ func (w *watcher) listAndWatchInner(ctx context.Context, r packageReader, filter
121143
if w.done {
122144
return false
123145
}
124-
obj, err := pr.GetPackageRevision(ctx)
146+
obj, err := w.extractor(ctx, pr)
125147
if err != nil {
126148
w.done = true
127149
errorResult <- err
@@ -144,7 +166,7 @@ func (w *watcher) listAndWatchInner(ctx context.Context, r packageReader, filter
144166
sentAdd := 0
145167
// TODO: Only if rv == 0?
146168
if err := r.listPackageRevisions(ctx, filter, func(ctx context.Context, p repository.PackageRevision) error {
147-
obj, err := p.GetPackageRevision(ctx)
169+
obj, err := w.extractor(ctx, p)
148170
if err != nil {
149171
w.mutex.Lock()
150172
w.done = true
@@ -200,7 +222,7 @@ func (w *watcher) listAndWatchInner(ctx context.Context, r packageReader, filter
200222
if w.done {
201223
return false
202224
}
203-
obj, err := pr.GetPackageRevision(ctx)
225+
obj, err := w.extractor(ctx, pr)
204226
if err != nil {
205227
w.done = true
206228
errorResult <- err

0 commit comments

Comments
 (0)