Skip to content

Commit

Permalink
Change ForDynamic to proper generic type *Unstructured
Browse files Browse the repository at this point in the history 
...instead of runtime.Object, which I think was used originally to
preserve the behavior of allowing any runtime.Object and internally
converting to *Unstructured but we can easily adjust users accordingly.

Some resource syncer unit tests failed b/c previously ForDynamic had the
side effect of actually modifying the resource on Update and the tests
did not explicitly retrieve the updated resource after test.UpdateResource.
The fake Federator was also modified to internally convert to *Unstructured
to simplify users.

Signed-off-by: Tom Pantelis <tompantelis@gmail.com>
  • Loading branch information
@tpantelis
tpantelis committed Nov 16, 2023
1 parent 04321bc commit 72e3ce5
Show file tree
Hide file tree
Showing 8 changed files with 72 additions and 92 deletions.
6 changes: 3 additions & 3 deletions pkg/federate/create_or_update_federator.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,9 +58,9 @@ func (f *createOrUpdateFederator) Distribute(ctx context.Context, obj runtime.Ob

f.prepareResourceForSync(toDistribute)

result, err := util.CreateOrUpdate[runtime.Object](ctx, resource.ForDynamic(resourceClient), toDistribute,
func(obj runtime.Object) (runtime.Object, error) {
return util.CopyImmutableMetadata(obj.(*unstructured.Unstructured), toDistribute), nil
result, err := util.CreateOrUpdate[*unstructured.Unstructured](ctx, resource.ForDynamic(resourceClient), toDistribute,
func(obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
return util.CopyImmutableMetadata(obj, toDistribute), nil
})

if f.eventLogName != "" {
Expand Down
22 changes: 12 additions & 10 deletions pkg/federate/fake/federator.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,26 +24,28 @@ import (
"time"

. "github.com/onsi/gomega"
"github.com/submariner-io/admiral/pkg/resource"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
)

type Federator struct {
distribute chan runtime.Object
delete chan runtime.Object
distribute chan *unstructured.Unstructured
delete chan *unstructured.Unstructured
FailOnDistribute error
FailOnDelete error
ResetOnFailure bool
}

func New() *Federator {
return &Federator{
distribute: make(chan runtime.Object, 100),
delete: make(chan runtime.Object, 100),
distribute: make(chan *unstructured.Unstructured, 100),
delete: make(chan *unstructured.Unstructured, 100),
ResetOnFailure: true,
}
}

func (f *Federator) Distribute(_ context.Context, resource runtime.Object) error {
func (f *Federator) Distribute(_ context.Context, obj runtime.Object) error {
err := f.FailOnDistribute
if err != nil {
if f.ResetOnFailure {
Expand All @@ -53,12 +55,12 @@ func (f *Federator) Distribute(_ context.Context, resource runtime.Object) error
return err
}

f.distribute <- resource
f.distribute <- resource.MustToUnstructured(obj)

return nil
}

func (f *Federator) Delete(_ context.Context, resource runtime.Object) error {
func (f *Federator) Delete(_ context.Context, obj runtime.Object) error {
err := f.FailOnDelete
if err != nil {
if f.ResetOnFailure {
Expand All @@ -68,21 +70,21 @@ func (f *Federator) Delete(_ context.Context, resource runtime.Object) error {
return err
}

f.delete <- resource
f.delete <- resource.MustToUnstructured(obj)

return nil
}

func (f *Federator) VerifyDistribute(expected runtime.Object) {
Eventually(f.distribute, 5).Should(Receive(Equal(expected)), "Distribute was not called")
Eventually(f.distribute, 5).Should(Receive(Equal(resource.MustToUnstructured(expected))), "Distribute was not called")
}

func (f *Federator) VerifyNoDistribute() {
Consistently(f.distribute, 300*time.Millisecond).ShouldNot(Receive(), "Distribute was unexpectedly called")
}

func (f *Federator) VerifyDelete(expected runtime.Object) {
Eventually(f.delete, 5).Should(Receive(Equal(expected)), "Delete was not called")
Eventually(f.delete, 5).Should(Receive(Equal(resource.MustToUnstructured(expected))), "Delete was not called")
}

func (f *Federator) VerifyNoDelete() {
Expand Down
6 changes: 3 additions & 3 deletions pkg/federate/update_federator.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,8 +62,8 @@ func (f *updateFederator) Distribute(ctx context.Context, obj runtime.Object) er

f.prepareResourceForSync(toUpdate)

return util.Update[runtime.Object](ctx, resource.ForDynamic(resourceClient), toUpdate,
func(obj runtime.Object) (runtime.Object, error) {
return f.update(obj.(*unstructured.Unstructured), toUpdate), nil
return util.Update[*unstructured.Unstructured](ctx, resource.ForDynamic(resourceClient), toUpdate,
func(obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
return f.update(obj, toUpdate), nil
})
}
60 changes: 18 additions & 42 deletions pkg/resource/dynamic.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,70 +23,46 @@ import (
"context"

metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/client-go/dynamic"
)

type dynamicType struct {
client dynamic.ResourceInterface
}

func (d *dynamicType) Get(ctx context.Context, name string, options metav1.GetOptions) (runtime.Object, error) {
func (d *dynamicType) Get(ctx context.Context, name string, options metav1.GetOptions) (*unstructured.Unstructured, error) {
return d.client.Get(ctx, name, options)
}

//nolint:gocritic // hugeParam - we're matching K8s API
func (d *dynamicType) Create(ctx context.Context, obj runtime.Object, options metav1.CreateOptions) (runtime.Object, error) {
raw, err := ToUnstructured(obj)
if err != nil {
return nil, err
}

return d.client.Create(ctx, raw, options)
func (d *dynamicType) Create(ctx context.Context, obj *unstructured.Unstructured, options metav1.CreateOptions,
) (*unstructured.Unstructured, error) {
return d.client.Create(ctx, obj, options)
}

//nolint:gocritic // hugeParam - we're matching K8s API
func (d *dynamicType) Update(ctx context.Context, obj runtime.Object, options metav1.UpdateOptions) (runtime.Object, error) {
raw, err := ToUnstructured(obj)
if err != nil {
return nil, err
}

return d.client.Update(ctx, raw, options)
func (d *dynamicType) Update(ctx context.Context, obj *unstructured.Unstructured, options metav1.UpdateOptions,
) (*unstructured.Unstructured, error) {
return d.client.Update(ctx, obj, options)
}

//nolint:gocritic // hugeParam - we're matching K8s API
func (d *dynamicType) UpdateStatus(ctx context.Context, obj runtime.Object, options metav1.UpdateOptions) (runtime.Object, error) {
raw, err := ToUnstructured(obj)
if err != nil {
return nil, err
}

return d.client.UpdateStatus(ctx, raw, options)
func (d *dynamicType) UpdateStatus(ctx context.Context, obj *unstructured.Unstructured, options metav1.UpdateOptions,
) (*unstructured.Unstructured, error) {
return d.client.UpdateStatus(ctx, obj, options)
}

func (d *dynamicType) Delete(ctx context.Context, name string,
options metav1.DeleteOptions, //nolint:gocritic // hugeParam - we're matching K8s API
) error {
//nolint:gocritic // hugeParam - we're matching K8s API
func (d *dynamicType) Delete(ctx context.Context, name string, options metav1.DeleteOptions) error {
return d.client.Delete(ctx, name, options)
}

func (d *dynamicType) List(ctx context.Context,
options metav1.ListOptions, //nolint:gocritic // hugeParam - we're matching K8s API
) ([]runtime.Object, error) {
l, err := d.client.List(ctx, options)
return MustExtractList[runtime.Object](l), err
//nolint:gocritic // hugeParam - we're matching K8s API
func (d *dynamicType) List(ctx context.Context, options metav1.ListOptions) (*unstructured.UnstructuredList, error) {
return d.client.List(ctx, options)
}

func ForDynamic(client dynamic.ResourceInterface) *InterfaceFuncs[runtime.Object] {
t := &dynamicType{client: client}

return &InterfaceFuncs[runtime.Object]{
GetFunc: t.Get,
CreateFunc: t.Create,
UpdateFunc: t.Update,
UpdateStatusFunc: t.UpdateStatus,
DeleteFunc: t.Delete,
ListFunc: t.List,
}
func ForDynamic(client dynamic.ResourceInterface) Interface[*unstructured.Unstructured] {
return kubernetesStatusInterfaceAdapter[*unstructured.Unstructured, *unstructured.UnstructuredList](&dynamicType{client: client})
}
7 changes: 4 additions & 3 deletions pkg/resource/interface_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
rbacv1 "k8s.io/api/rbac/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
Expand Down Expand Up @@ -191,14 +192,14 @@ var _ = Describe("Interface", func() {
})

Context("ForDynamic", func() {
testInterfaceFuncs(func() resource.Interface[runtime.Object] {
testInterfaceFuncs(func() resource.Interface[*unstructured.Unstructured] {
return resource.ForDynamic(dynamicfake.NewSimpleDynamicClient(scheme.Scheme).Resource(
schema.GroupVersionResource{
Group: corev1.SchemeGroupVersion.Group,
Version: corev1.SchemeGroupVersion.Version,
Resource: "pods",
}).Namespace(test.LocalNamespace))
}, runtime.Object(resource.MustToUnstructured(&corev1.Pod{
}, resource.MustToUnstructured(&corev1.Pod{
TypeMeta: metav1.TypeMeta{
Kind: "Pod",
APIVersion: "v1",
Expand All @@ -210,7 +211,7 @@ var _ = Describe("Interface", func() {
Spec: corev1.PodSpec{
Hostname: "my-host",
},
})))
}))
})
})

Expand Down
31 changes: 15 additions & 16 deletions pkg/syncer/resource_syncer_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/submariner-io/admiral/pkg/federate/fake"
. "github.com/submariner-io/admiral/pkg/gomega"
"github.com/submariner-io/admiral/pkg/resource"
"github.com/submariner-io/admiral/pkg/syncer"
"github.com/submariner-io/admiral/pkg/syncer/test"
"github.com/submariner-io/admiral/pkg/util"
Expand Down Expand Up @@ -90,7 +89,7 @@ func testReconcileLocalToRemote() {
When("the resource to reconcile is local and does not exist", func() {
It("should invoke delete", func() {
d.resource.Namespace = test.LocalNamespace
d.federator.VerifyDelete(test.SetClusterIDLabel(resource.MustToUnstructured(d.resource), ""))
d.federator.VerifyDelete(test.SetClusterIDLabel(d.resource, ""))
})
})

Expand Down Expand Up @@ -133,7 +132,7 @@ func testReconcileRemoteToLocal() {

When("the resource to reconcile is remote and does not exist", func() {
It("should invoke delete", func() {
d.federator.VerifyDelete(resource.MustToUnstructured(d.resource))
d.federator.VerifyDelete(d.resource)
})
})

Expand Down Expand Up @@ -177,7 +176,7 @@ func testReconcileNoDirection() {

When("the resource to reconcile does not exist", func() {
It("should invoke delete", func() {
d.federator.VerifyDelete(resource.MustToUnstructured(d.resource))
d.federator.VerifyDelete(d.resource)
})
})

Expand Down Expand Up @@ -355,11 +354,11 @@ func testTransformFunction() {
})

verifyDistribute := func() {
d.federator.VerifyDistribute(test.SetClusterIDLabel(test.ToUnstructured(transformed), "remote"))
d.federator.VerifyDistribute(test.SetClusterIDLabel(transformed.DeepCopy(), "remote"))
}

verifyDelete := func() {
d.federator.VerifyDelete(test.SetClusterIDLabel(test.ToUnstructured(transformed), "remote"))
d.federator.VerifyDelete(test.SetClusterIDLabel(transformed.DeepCopy(), "remote"))
}

When("a resource is created in the datastore", func() {
Expand Down Expand Up @@ -399,7 +398,7 @@ func testTransformFunction() {

d.resource = test.NewPodWithImage(d.config.SourceNamespace, "updated")
test.UpdateResource(d.sourceClient, test.NewPodWithImage(d.config.SourceNamespace, "updated"))
d.federator.VerifyDistribute(test.ToUnstructured(transformed))
d.federator.VerifyDistribute(transformed)
Eventually(expOperation).Should(Receive(Equal(syncer.Update)))
})
})
Expand Down Expand Up @@ -630,7 +629,7 @@ func testOnSuccessfulSyncFunction() {

It("should invoke the OnSuccessfulSync function with the transformed resource", func() {
test.CreateResource(d.sourceClient, d.resource)
d.federator.VerifyDistribute(test.ToUnstructured(expResource))
d.federator.VerifyDistribute(expResource)
Eventually(expOperation).Should(Receive(Equal(syncer.Create)))
})
})
Expand Down Expand Up @@ -883,7 +882,7 @@ func testUpdateSuppression() {
})

It("should distribute it", func() {
d.federator.VerifyDistribute(test.ToUnstructured(d.resource))
d.federator.VerifyDistribute(test.GetResource(d.sourceClient, d.resource))
})
})

Expand All @@ -894,7 +893,7 @@ func testUpdateSuppression() {
})

It("should distribute it", func() {
d.federator.VerifyDistribute(test.ToUnstructured(d.resource))
d.federator.VerifyDistribute(test.GetResource(d.sourceClient, d.resource))
})
})
})
Expand Down Expand Up @@ -931,7 +930,7 @@ func testUpdateSuppression() {
})

It("should distribute it", func() {
d.federator.VerifyDistribute(test.ToUnstructured(d.resource))
d.federator.VerifyDistribute(test.GetResource(d.sourceClient, d.resource))
})
})

Expand All @@ -941,7 +940,7 @@ func testUpdateSuppression() {
})

It("should distribute it", func() {
d.federator.VerifyDistribute(test.ToUnstructured(d.resource))
d.federator.VerifyDistribute(test.GetResource(d.sourceClient, d.resource))
})
})
})
Expand All @@ -960,7 +959,7 @@ func testUpdateSuppression() {
})

It("should distribute it", func() {
d.federator.VerifyDistribute(test.ToUnstructured(d.resource))
d.federator.VerifyDistribute(test.GetResource(d.sourceClient, d.resource))
})
})
})
Expand All @@ -976,7 +975,7 @@ func testUpdateSuppression() {
})

It("should distribute it", func() {
d.federator.VerifyDistribute(test.ToUnstructured(d.resource))
d.federator.VerifyDistribute(test.GetResource(d.sourceClient, d.resource))
})
})

Expand Down Expand Up @@ -1152,11 +1151,11 @@ func testRequeueResource() {
})

It("should requeue it", func() {
d.federator.VerifyDistribute(test.ToUnstructured(transformed))
d.federator.VerifyDistribute(transformed)

d.syncer.RequeueResource(d.resource.Name, d.resource.Namespace)

d.federator.VerifyDistribute(test.ToUnstructured(transformed))
d.federator.VerifyDistribute(transformed)
})
})

Expand Down
10 changes: 4 additions & 6 deletions pkg/syncer/test/util.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,10 +46,7 @@ const (
)

func GetResourceAndError(resourceInterface dynamic.ResourceInterface, obj runtime.Object) (*unstructured.Unstructured, error) {
meta, err := metaapi.Accessor(obj)
Expect(err).To(Succeed())

return resourceInterface.Get(context.TODO(), meta.GetName(), metav1.GetOptions{})
return resourceInterface.Get(context.TODO(), resource.MustToMeta(obj).GetName(), metav1.GetOptions{})
}

func GetResource(resourceInterface dynamic.ResourceInterface, obj runtime.Object) *unstructured.Unstructured {
Expand All @@ -70,8 +67,9 @@ func CreateResource(resourceInterface dynamic.ResourceInterface, obj runtime.Obj
}

func UpdateResource(resourceInterface dynamic.ResourceInterface, obj runtime.Object) *unstructured.Unstructured {
err := util.Update[runtime.Object](context.Background(), resource.ForDynamic(resourceInterface), obj,
util.Replace(obj))
u := resource.MustToUnstructured(obj)
err := util.Update[*unstructured.Unstructured](context.Background(), resource.ForDynamic(resourceInterface), u,
util.Replace(u))
Expect(err).To(Succeed())

return GetResource(resourceInterface, obj)
Expand Down
Loading

0 comments on commit 72e3ce5

Please sign in to comment.