From eb404e667718fe83dc1c855640c0afe5497debdb Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Fri, 26 Jun 2026 15:38:16 -0700 Subject: [PATCH] test: rework the shared-store user/bucket/topic suites The object store user (caps/keys/opmask), bucket-owner, and bucket topic tests already shared a CephObjectStore but polled with count-based utils.Retry behind a wide entry signature. Move them onto the wait4 toolkit and a slim (t, k8sh, store) signature: each is an ordered t.Run script using typed clients and watch-based waits, with per-package check helpers for repeated verification. Extend the shared store fixture to provision the realm/zonegroup/zone pools and serve TLS, build its rgw admin and SNS clients through the consolidated util/client package (replacing the separate admin/sns/s3 packages), and run every package in both the TLS and non-TLS passes (now separate parallel jobs). Signed-off-by: Joshua Hoblitt --- tests/integration/ceph_object_test.go | 21 +- .../integration/object/bucket/owner/owner.go | 446 ++++---------- tests/integration/object/topic/kafka/kafka.go | 489 +++++---------- tests/integration/object/user/caps/caps.go | 102 ++-- tests/integration/object/user/keys/keys.go | 556 +++++------------- .../integration/object/user/opmask/opmask.go | 117 ++-- tests/integration/object/util/admin/admin.go | 67 --- tests/integration/object/util/s3/s3.go | 75 --- .../object/util/sharedstore/sharedstore.go | 305 ++++++++-- tests/integration/object/util/sns/sns.go | 90 --- 10 files changed, 745 insertions(+), 1523 deletions(-) delete mode 100644 tests/integration/object/util/admin/admin.go delete mode 100644 tests/integration/object/util/s3/s3.go delete mode 100644 tests/integration/object/util/sns/sns.go diff --git a/tests/integration/ceph_object_test.go b/tests/integration/ceph_object_test.go index 903b9e514..59f53ead5 100644 --- a/tests/integration/ceph_object_test.go +++ b/tests/integration/ceph_object_test.go @@ -209,20 +209,15 @@ func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, install // now test operation of the first object store testObjectStoreOperations(s, helper, k8sh, settings, storeName, swiftAndKeystone) - // The namespaced test packages below all skip when TLS is enabled, so only set up the - // shared store when it will actually be used. - var sharedObjectStore *cephv1.CephObjectStore - if !tlsEnable { - var teardownSharedStore func() - sharedObjectStore, teardownSharedStore = sharedstore.Setup(s.T(), k8sh) - defer teardownSharedStore() + sharedObjectStore := sharedstore.Create(s.T(), k8sh, installer, tlsEnable, + bucketowner.Namespace, userkeys.Namespace, topickafka.Namespace, useropmask.Namespace, usercaps.Namespace) + defer sharedObjectStore.Destroy() - bucketowner.TestObjectBucketClaimBucketOwner(s.T(), k8sh, installer, logger, tlsEnable, sharedObjectStore) - userkeys.TestObjectStoreUserKeys(s.T(), k8sh, installer, logger, tlsEnable, sharedObjectStore) - topickafka.TestBucketTopicKafka(s.T(), k8sh, installer, logger, tlsEnable, sharedObjectStore) - useropmask.TestObjectStoreUserOpMask(s.T(), k8sh, installer, logger, tlsEnable, sharedObjectStore) - usercaps.TestObjectStoreUserCaps(s.T(), k8sh, installer, logger, tlsEnable, sharedObjectStore) - } + bucketowner.TestObjectBucketClaimBucketOwner(s.T(), k8sh, sharedObjectStore) + userkeys.TestObjectStoreUserKeys(s.T(), k8sh, sharedObjectStore) + topickafka.TestBucketTopicKafka(s.T(), k8sh, sharedObjectStore) + useropmask.TestObjectStoreUserOpMask(s.T(), k8sh, sharedObjectStore) + usercaps.TestObjectStoreUserCaps(s.T(), k8sh, sharedObjectStore) bucketNotificationTestStoreName := "bucket-notification-" + storeName createCephObjectStore(s.T(), helper, k8sh, installer, namespace, bucketNotificationTestStoreName, 1, tlsEnable, swiftAndKeystone) diff --git a/tests/integration/object/bucket/owner/owner.go b/tests/integration/object/bucket/owner/owner.go index 5c3243c57..70b8cab18 100644 --- a/tests/integration/object/bucket/owner/owner.go +++ b/tests/integration/object/bucket/owner/owner.go @@ -17,79 +17,34 @@ limitations under the License. package owner import ( - "bufio" "context" "fmt" - "log" "strings" "testing" - "time" "github.com/ceph/go-ceph/rgw/admin" - "github.com/coreos/pkg/capnslog" bktv1alpha1 "github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1" - cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/tests/framework/installer" - "github.com/rook/rook/tests/framework/utils" - utiladmin "github.com/rook/rook/tests/integration/object/util/admin" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" - storagev1 "k8s.io/api/storage/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + + cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" + "github.com/rook/rook/tests/framework/utils" + "github.com/rook/rook/tests/integration/object/util/fixture" + "github.com/rook/rook/tests/integration/object/util/sharedstore" + "github.com/rook/rook/tests/integration/object/util/wait4" ) -func WaitForPodLogContainingText(k8sh *utils.K8sHelper, namespace string, selector *labels.Selector, text string, timeout time.Duration) error { - pods, err := k8sh.Clientset.CoreV1().Pods(namespace).List( - context.Background(), - metav1.ListOptions{ - LabelSelector: (*selector).String(), - }, - ) - if err != nil { - return fmt.Errorf("failed to list pods: %v", err) - } +const Namespace = "test-bucketowner" - if len(pods.Items) == 0 { - return fmt.Errorf("no pods found with labels %v in namespace %q", selector, namespace) - } - - // if there are multiple pods, just pick the first one - selectedPod := pods.Items[0] - log.Printf("Found pod %q (first match) with labels %v\n", selectedPod.Name, *selector) - - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - - req := k8sh.Clientset.CoreV1().Pods(namespace).GetLogs(selectedPod.Name, &corev1.PodLogOptions{}) - - logStream, err := req.Stream(ctx) - if err != nil { - return fmt.Errorf("failed to stream logs from pod %q: %v", selectedPod.Name, err) - } - defer logStream.Close() - - scanner := bufio.NewScanner(logStream) - - for scanner.Scan() { - line := scanner.Text() - if strings.Contains(line, text) { - break - } - } - // Check for scanner error (could be context timeout, etc.) - if err := scanner.Err(); err != nil { - return fmt.Errorf("error reading log stream: %v", err) - } - - return nil -} - -func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, installer *installer.CephInstaller, logger *capnslog.PackageLogger, tlsEnable bool, objectStore *cephv1.CephObjectStore) { +func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, store *sharedstore.Sharedstore) { var ( - defaultName = "test-bucketowner" + defaultName = Namespace + objectStore = store.ObjectStore() + adminClient = store.AdminClient() ns = &corev1.Namespace{ ObjectMeta: metav1.ObjectMeta{ @@ -97,16 +52,7 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta }, } - storageClass = &storagev1.StorageClass{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName, - }, - Provisioner: objectStore.Namespace + ".ceph.rook.io/bucket", - Parameters: map[string]string{ - "objectStoreName": objectStore.Name, - "objectStoreNamespace": objectStore.Namespace, - }, - } + storageClass = fixture.StorageClass(defaultName, objectStore) // test user without any quotas set osu1 = cephv1.CephObjectStoreUser{ @@ -178,102 +124,48 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta }, }, } + + osuClient = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name) + obcClient = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name) + obClient = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets() ) t.Run("OBC bucketOwner", func(t *testing.T) { - if tlsEnable { - // There is lots of coverage of rgw working with tls enabled; skipping to save time. - // If tls is to be enabled, cert generation needs to be added and a - // different CephObjectStore name needs to be set for with/without tls as - // CephObjectStore does not currently cleanup the rgw realm. - t.Skip("skipping test for TLS enabled clusters") - } + ctx := t.Context() - var adminClient *admin.API - ctx := context.TODO() - - t.Run(fmt.Sprintf("create ns %q", ns.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create sc %q", storageClass.Name), func(t *testing.T) { - _, err := k8sh.Clientset.StorageV1().StorageClasses().Create(ctx, storageClass, metav1.CreateOptions{}) - require.NoError(t, err) - }) + fixture.RequireNamespace(t, k8sh, ns) + fixture.RequireStorageClass(t, k8sh, storageClass) // since this is an obc specific subtest we assume that CephObjectStoreUser // is working and the rgw service state does not need to be inspected to // confirm user creation. t.Run(fmt.Sprintf("create CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Create(ctx, &osu1, metav1.CreateOptions{}) - require.NoError(t, err) - // user creation may be slow right after rgw start up - osuReady := utils.Retry(120, time.Second, "CephObjectStoreUser is Ready", func() bool { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - if liveOsu.Status == nil { - return false - } - - return liveOsu.Status.Phase == "Ready" - }) - require.True(t, osuReady) + wait4.RequireCreate(ctx, t, osuClient, &osu1, wait4.ObjectStoreUser, wait4.TimeoutLong) }) t.Run(fmt.Sprintf("create obc %q with bucketOwner %q", obc1.Name, obc1.Spec.AdditionalConfig["bucketOwner"]), func(t *testing.T) { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Create(ctx, &obc1, metav1.CreateOptions{}) + _, err := obcClient.Create(ctx, &obc1, metav1.CreateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("obc %q has bucketOwner %q set", obc1.Name, obc1.Spec.AdditionalConfig["bucketOwner"]), func(t *testing.T) { - var liveObc *bktv1alpha1.ObjectBucketClaim - obcBound := utils.Retry(40, time.Second, "OBC is Bound", func() bool { - var err error - liveObc, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - return liveObc.Status.Phase == bktv1alpha1.ObjectBucketClaimStatusPhaseBound - }) - require.True(t, obcBound) + liveObc := wait4.RequireCondition(ctx, t, obcClient, obc1.Name, wait4.OBCBound, wait4.TimeoutShort) // verify that bucketOwner is set on the live obc assert.Equal(t, obc1.Spec.AdditionalConfig["bucketOwner"], liveObc.Spec.AdditionalConfig["bucketOwner"]) }) t.Run(fmt.Sprintf("ob for obc %q has bucketOwner %q set", obc1.Name, obc1.Spec.AdditionalConfig["bucketOwner"]), func(t *testing.T) { - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) obName := liveObc.Spec.ObjectBucketName - var liveOb *bktv1alpha1.ObjectBucket - obBound := utils.Retry(40, time.Second, "OB is Bound", func() bool { - var err error - liveOb, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{}) - if err != nil { - return false - } - - return liveOb.Status.Phase == bktv1alpha1.ObjectBucketStatusPhaseBound - }) - require.True(t, obBound) + liveOb := wait4.RequireCondition(ctx, t, obClient, obName, wait4.OBBound, wait4.TimeoutShort) // verify that bucketOwner is set on the live ob assert.Equal(t, obc1.Spec.AdditionalConfig["bucketOwner"], liveOb.Spec.Connection.AdditionalState["bucketOwner"]) }) - // the rgw admin api is used to verify bucket ownership - t.Run("setup rgw admin api client", func(t *testing.T) { - var err error - adminClient, err = utiladmin.NewAdminClient(objectStore, installer, k8sh, tlsEnable) - require.NoError(t, err) - }) - t.Run(fmt.Sprintf("bucket created with owner %q", obc1.Spec.AdditionalConfig["bucketOwner"]), func(t *testing.T) { bucket, err := adminClient.GetBucketInfo(ctx, admin.Bucket{Bucket: obc1.Spec.BucketName}) require.NoError(t, err) @@ -294,38 +186,23 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta t.Run(fmt.Sprintf("create CephObjectStoreUser %q", osu2.Name), func(t *testing.T) { // create user2 - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Create(ctx, &osu2, metav1.CreateOptions{}) - require.NoError(t, err) - - osuReady := utils.Retry(40, time.Second, "CephObjectStoreUser is Ready", func() bool { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu2.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - if liveOsu.Status == nil { - return false - } - - return liveOsu.Status.Phase == "Ready" - }) - require.True(t, osuReady) + wait4.RequireCreate(ctx, t, osuClient, &osu2, wait4.ObjectStoreUser, wait4.TimeoutLong) }) t.Run(fmt.Sprintf("update obc %q to bucketOwner %q", obc1.Name, osu2.Name), func(t *testing.T) { // update obc bucketOwner - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) liveObc.Spec.AdditionalConfig["bucketOwner"] = osu2.Name - _, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Update(ctx, liveObc, metav1.UpdateOptions{}) + _, err = obcClient.Update(ctx, liveObc, metav1.UpdateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("obc %q has bucketOwner %q set", obc1.Name, osu2.Name), func(t *testing.T) { // obc .Status.Phase does not appear to change when updating the obc - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) // verify that bucketOwner is set on the live obc @@ -334,39 +211,32 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta t.Run(fmt.Sprintf("ob for obc %q has bucketOwner %q set", obc1.Name, osu2.Name), func(t *testing.T) { // ob .Status.Phase does not appear to change when updating the obc - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) obName := liveObc.Spec.ObjectBucketName - var liveOb *bktv1alpha1.ObjectBucket - inSync := utils.Retry(40, time.Second, "OB is Bound", func() bool { - var err error - liveOb, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{}) - if err != nil { - return false - } - - return osu2.Name == liveOb.Spec.Connection.AdditionalState["bucketOwner"] - }) - require.True(t, inSync) + liveOb := wait4.RequireCondition(ctx, t, obClient, obName, + func(ob *bktv1alpha1.ObjectBucket) bool { + return ob.Spec.Connection.AdditionalState["bucketOwner"] == osu2.Name + }, + wait4.TimeoutShort) // verify that bucketOwner is set on the live ob assert.Equal(t, osu2.Name, liveOb.Spec.Connection.AdditionalState["bucketOwner"]) }) t.Run(fmt.Sprintf("bucket owner changed to %q", osu2.Name), func(t *testing.T) { - var bucket admin.Bucket - ownerSync := utils.Retry(40, time.Second, "bucket owner in sync", func() bool { - var err error - bucket, err = adminClient.GetBucketInfo(ctx, admin.Bucket{Bucket: obc1.Name}) + wait4.AssertEventually(ctx, t, wait4.TimeoutShort, fmt.Sprintf("bucket %q owner is %q", obc1.Name, osu2.Name), func(ctx context.Context) error { + bucket, err := adminClient.GetBucketInfo(ctx, admin.Bucket{Bucket: obc1.Name}) if err != nil { - return false + return err } - return bucket.Owner == osu2.Name + if bucket.Owner != osu2.Name { + return fmt.Errorf("bucket owner is %q, want %q", bucket.Owner, osu2.Name) + } + return nil }) - assert.True(t, ownerSync) - assert.Equal(t, osu2.Name, bucket.Owner) }) // obc should not modify pre-existing users @@ -382,78 +252,68 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta t.Run(fmt.Sprintf("remove obc %q bucketOwner", obc1.Name), func(t *testing.T) { // update/remove obc bucketOwner - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) liveObc.Spec.AdditionalConfig = map[string]string{} - _, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Update(ctx, liveObc, metav1.UpdateOptions{}) + _, err = obcClient.Update(ctx, liveObc, metav1.UpdateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("obc %q has no bucketOwner", obc1.Name), func(t *testing.T) { // verify that bucketOwner is unset on the live obc - notSet := utils.Retry(40, time.Second, "bucketOwner not set", func() bool { - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - _, ok := liveObc.Spec.AdditionalConfig["bucketOwner"] - return !ok - }) - assert.True(t, notSet) + wait4.AssertCondition(ctx, t, obcClient, obc1.Name, + func(obc *bktv1alpha1.ObjectBucketClaim) bool { + _, ok := obc.Spec.AdditionalConfig["bucketOwner"] + return !ok + }, + wait4.TimeoutShort) }) t.Run(fmt.Sprintf("ob for obc %q has no bucketOwner", obc1.Name), func(t *testing.T) { - // ob .Status.Phase does not appear to change when updating the obc - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) obName := liveObc.Spec.ObjectBucketName - notSet := utils.Retry(40, time.Second, "bucketOwner not set", func() bool { - liveOb, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{}) - if err != nil { - return false - } - - _, ok := liveOb.Spec.Connection.AdditionalState["bucketOwner"] - return !ok - }) - assert.True(t, notSet) + wait4.AssertCondition(ctx, t, obClient, obName, + func(ob *bktv1alpha1.ObjectBucket) bool { + _, ok := ob.Spec.Connection.AdditionalState["bucketOwner"] + return !ok + }, + wait4.TimeoutShort) }) // the ob should retain the existing owner and not revert to a generated user t.Run(fmt.Sprintf("bucket owner is still %q", osu2.Name), func(t *testing.T) { - var bucket admin.Bucket - ownerSync := utils.Retry(40, time.Second, "bucket owner in sync", func() bool { - var err error - bucket, err = adminClient.GetBucketInfo(ctx, admin.Bucket{Bucket: obc1.Name}) + wait4.AssertEventually(ctx, t, wait4.TimeoutShort, fmt.Sprintf("bucket %q owner is still %q", obc1.Name, osu2.Name), func(ctx context.Context) error { + bucket, err := adminClient.GetBucketInfo(ctx, admin.Bucket{Bucket: obc1.Name}) if err != nil { - return false + return err } - return bucket.Owner == osu2.Name + if bucket.Owner != osu2.Name { + return fmt.Errorf("bucket owner is %q, want %q", bucket.Owner, osu2.Name) + } + return nil }) - assert.True(t, ownerSync) - assert.Equal(t, osu2.Name, bucket.Owner) }) // this covers setting bucketOwner on an obc initially created without an explicit owner t.Run(fmt.Sprintf("update obc %q to bucketOwner %q", obc1.Name, osu1.Name), func(t *testing.T) { // update obc bucketOwner - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) liveObc.Spec.AdditionalConfig = map[string]string{"bucketOwner": osu1.Name} - _, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Update(ctx, liveObc, metav1.UpdateOptions{}) + _, err = obcClient.Update(ctx, liveObc, metav1.UpdateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("obc %q has bucketOwner %q set", obc1.Name, osu1.Name), func(t *testing.T) { // obc .Status.Phase does not appear to change when updating the obc - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) // verify that bucketOwner is set on the live obc @@ -462,94 +322,52 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta t.Run(fmt.Sprintf("ob for obc %q has bucketOwner %q set", obc1.Name, osu1.Name), func(t *testing.T) { // ob .Status.Phase does not appear to change when updating the obc - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) obName := liveObc.Spec.ObjectBucketName - var liveOb *bktv1alpha1.ObjectBucket - inSync := utils.Retry(40, time.Second, "OB is Bound", func() bool { - var err error - liveOb, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{}) - if err != nil { - return false - } - - return osu1.Name == liveOb.Spec.Connection.AdditionalState["bucketOwner"] - }) - require.True(t, inSync) + liveOb := wait4.RequireCondition(ctx, t, obClient, obName, + func(ob *bktv1alpha1.ObjectBucket) bool { + return ob.Spec.Connection.AdditionalState["bucketOwner"] == osu1.Name + }, + wait4.TimeoutShort) // verify that bucketOwner is set on the live ob assert.Equal(t, osu1.Name, liveOb.Spec.Connection.AdditionalState["bucketOwner"]) }) t.Run(fmt.Sprintf("bucket owner changed to %q", osu1.Name), func(t *testing.T) { - var bucket admin.Bucket - ownerSync := utils.Retry(40, time.Second, "bucket owner in sync", func() bool { - var err error - bucket, err = adminClient.GetBucketInfo(ctx, admin.Bucket{Bucket: obc1.Name}) + wait4.AssertEventually(ctx, t, wait4.TimeoutShort, fmt.Sprintf("bucket %q owner is %q", obc1.Name, osu1.Name), func(ctx context.Context) error { + bucket, err := adminClient.GetBucketInfo(ctx, admin.Bucket{Bucket: obc1.Name}) if err != nil { - return false + return err } - return bucket.Owner == osu1.Name - }) - assert.True(t, ownerSync) - assert.Equal(t, osu1.Name, bucket.Owner) - }) - - t.Run(fmt.Sprintf("bucket owner changed to %q", osu1.Name), func(t *testing.T) { - var bucket admin.Bucket - ownerSync := utils.Retry(40, time.Second, "bucket owner in sync", func() bool { - var err error - bucket, err = adminClient.GetBucketInfo(ctx, admin.Bucket{Bucket: obc1.Name}) - if err != nil { - return false + if bucket.Owner != osu1.Name { + return fmt.Errorf("bucket owner is %q, want %q", bucket.Owner, osu1.Name) } - - return bucket.Owner == osu1.Name + return nil }) - assert.True(t, ownerSync) - assert.Equal(t, osu1.Name, bucket.Owner) }) t.Run(fmt.Sprintf("create obc %q with bucketOwner %q", obc2.Name, obc2.Spec.AdditionalConfig["bucketOwner"]), func(t *testing.T) { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Create(ctx, &obc2, metav1.CreateOptions{}) + _, err := obcClient.Create(ctx, &obc2, metav1.CreateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("obc %q has bucketOwner %q set", obc2.Name, obc2.Spec.AdditionalConfig["bucketOwner"]), func(t *testing.T) { - var liveObc *bktv1alpha1.ObjectBucketClaim - obcBound := utils.Retry(40, time.Second, "OBC is Bound", func() bool { - var err error - liveObc, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc2.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - return liveObc.Status.Phase == bktv1alpha1.ObjectBucketClaimStatusPhaseBound - }) - require.True(t, obcBound) + liveObc := wait4.RequireCondition(ctx, t, obcClient, obc2.Name, wait4.OBCBound, wait4.TimeoutShort) // verify that bucketOwner is set on the live obc assert.Equal(t, obc2.Spec.AdditionalConfig["bucketOwner"], liveObc.Spec.AdditionalConfig["bucketOwner"]) }) t.Run(fmt.Sprintf("ob for obc %q has bucketOwner %q set", obc2.Name, obc2.Spec.AdditionalConfig["bucketOwner"]), func(t *testing.T) { - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc2.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc2.Name, metav1.GetOptions{}) require.NoError(t, err) obName := liveObc.Spec.ObjectBucketName - var liveOb *bktv1alpha1.ObjectBucket - obBound := utils.Retry(40, time.Second, "OB is Bound", func() bool { - var err error - liveOb, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{}) - if err != nil { - return false - } - - return liveOb.Status.Phase == bktv1alpha1.ObjectBucketStatusPhaseBound - }) - require.True(t, obBound) + liveOb := wait4.RequireCondition(ctx, t, obClient, obName, wait4.OBBound, wait4.TimeoutShort) // verify that bucketOwner is set on the live ob assert.Equal(t, obc2.Spec.AdditionalConfig["bucketOwner"], liveOb.Spec.Connection.AdditionalState["bucketOwner"]) @@ -568,48 +386,26 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta t.Run(fmt.Sprintf("delete obc %q", obc1.Name), func(t *testing.T) { // lookup ob name - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) obName := liveObc.Spec.ObjectBucketName - // delete obc - err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Delete(ctx, obc1.Name, metav1.DeleteOptions{}) - require.NoError(t, err) + // delete obc; the backing ob is garbage-collected by the provisioner + wait4.AssertDelete(ctx, t, obcClient, obc1.Name, wait4.TimeoutShort) - absent := utils.Retry(40, time.Second, "OBC is absent", func() bool { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc1.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) - - absent = utils.Retry(40, time.Second, "OB is absent", func() bool { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + wait4.AssertAbsent(ctx, t, obClient, obName, wait4.TimeoutShort) }) t.Run(fmt.Sprintf("delete obc %q", obc2.Name), func(t *testing.T) { // lookup ob name - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc2.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc2.Name, metav1.GetOptions{}) require.NoError(t, err) obName := liveObc.Spec.ObjectBucketName - // delete obc - err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Delete(ctx, obc2.Name, metav1.DeleteOptions{}) - require.NoError(t, err) + // delete obc; the backing ob is garbage-collected by the provisioner + wait4.AssertDelete(ctx, t, obcClient, obc2.Name, wait4.TimeoutShort) - absent := utils.Retry(40, time.Second, "OBC is absent", func() bool { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obc2.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) - - absent = utils.Retry(40, time.Second, "OB is absent", func() bool { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + wait4.AssertAbsent(ctx, t, obClient, obName, wait4.TimeoutShort) }) t.Run(fmt.Sprintf("user %q was not deleted by obc %q", osu1.Name, obc1.Name), func(t *testing.T) { @@ -629,7 +425,7 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta // test obc creation with bucketOwner set to a non-existent user, which should fail // "failure" means the obc remains in Pending state t.Run(fmt.Sprintf("create obc %q with non-existent bucketOwner %q", obcBogusOwner.Name, obcBogusOwner.Spec.AdditionalConfig["bucketOwner"]), func(t *testing.T) { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Create(ctx, &obcBogusOwner, metav1.CreateOptions{}) + _, err := obcClient.Create(ctx, &obcBogusOwner, metav1.CreateOptions{}) require.NoError(t, err) }) @@ -637,14 +433,17 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta selector := labels.SelectorFromSet(labels.Set{ "app": "rook-ceph-operator", }) - text := `error provisioning bucket: unable to get user \"test-bucket-owner-bogus-user\" creds: Ceph object user \"test-bucket-owner-bogus-user\" not found: NoSuchUser` - - err := WaitForPodLogContainingText(k8sh, "object-ns-system", &selector, text, 10*time.Second) - require.NoError(t, err) + // match on the operator's "unable to get user ... creds" error + // mentioning the bogus owner rather than a hardcoded full log line, + // so the logger's quote encoding cannot break the match + bogusOwner := obcBogusOwner.Spec.AdditionalConfig["bucketOwner"] + wait4.RequirePodLog(ctx, t, k8sh, "object-ns-system", selector, wait4.TimeoutShort, func(line string) bool { + return strings.Contains(line, "unable to get user") && strings.Contains(line, bogusOwner) + }) }) t.Run(fmt.Sprintf("obc %q stays Pending", obcBogusOwner.Name), func(t *testing.T) { - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obcBogusOwner.Namespace).Get(ctx, obcBogusOwner.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obcBogusOwner.Name, metav1.GetOptions{}) require.NoError(t, err) assert.True(t, bktv1alpha1.ObjectBucketClaimStatusPhasePending == liveObc.Status.Phase) @@ -657,57 +456,22 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, insta t.Run(fmt.Sprintf("delete obc %q", obcBogusOwner.Name), func(t *testing.T) { // lookup ob name - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obcBogusOwner.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obcBogusOwner.Name, metav1.GetOptions{}) require.NoError(t, err) obName := liveObc.Spec.ObjectBucketName - // delete obc - err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Delete(ctx, obcBogusOwner.Name, metav1.DeleteOptions{}) - require.NoError(t, err) + // delete obc; the backing ob is garbage-collected by the provisioner + wait4.AssertDelete(ctx, t, obcClient, obcBogusOwner.Name, wait4.TimeoutShort) - absent := utils.Retry(40, time.Second, "OBC is absent", func() bool { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).Get(ctx, obcBogusOwner.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) - - absent = utils.Retry(40, time.Second, "OB is absent", func() bool { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + wait4.AssertAbsent(ctx, t, obClient, obName, wait4.TimeoutShort) }) t.Run(fmt.Sprintf("delete CephObjectStoreUser %q", osu2.Name), func(t *testing.T) { - err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Delete(ctx, osu2.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - - absent := utils.Retry(40, time.Second, "CephObjectStoreUser is absent", func() bool { - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu2.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + wait4.AssertDelete(ctx, t, osuClient, osu2.Name, wait4.TimeoutShort) }) t.Run(fmt.Sprintf("delete CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Delete(ctx, osu1.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - - absent := utils.Retry(40, time.Second, "CephObjectStoreUser is absent", func() bool { - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) - }) - - t.Run(fmt.Sprintf("delete sc %q", storageClass.Name), func(t *testing.T) { - err := k8sh.Clientset.StorageV1().StorageClasses().Delete(ctx, storageClass.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("delete ns %q", ns.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Namespaces().Delete(ctx, ns.Name, metav1.DeleteOptions{}) - require.NoError(t, err) + wait4.AssertDelete(ctx, t, osuClient, osu1.Name, wait4.TimeoutShort) }) }) } diff --git a/tests/integration/object/topic/kafka/kafka.go b/tests/integration/object/topic/kafka/kafka.go index f7e556c88..af16287cf 100644 --- a/tests/integration/object/topic/kafka/kafka.go +++ b/tests/integration/object/topic/kafka/kafka.go @@ -22,117 +22,109 @@ import ( "fmt" "net/url" "testing" - "time" "github.com/aws/aws-sdk-go-v2/service/sns" - "github.com/coreos/pkg/capnslog" bktv1alpha1 "github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" - storagev1 "k8s.io/api/storage/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/tests/framework/installer" "github.com/rook/rook/tests/framework/utils" - utilsns "github.com/rook/rook/tests/integration/object/util/sns" + "github.com/rook/rook/tests/integration/object/util/fixture" + "github.com/rook/rook/tests/integration/object/util/secrets" + "github.com/rook/rook/tests/integration/object/util/sharedstore" + "github.com/rook/rook/tests/integration/object/util/wait4" ) -func checkStatusSecrets(t *testing.T, k8sh *utils.K8sHelper, bt *cephv1.CephBucketTopic, expectedSecrets []*corev1.Secret) { - t.Run(fmt.Sprintf("cephBucketTopic %q has .status.secrets set", bt.Name), func(t *testing.T) { - ctx := context.TODO() - - liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt.Namespace).Get(ctx, bt.Name, metav1.GetOptions{}) - require.NoError(t, err) - - require.NotNil(t, liveBt.Status) - assert.Len(t, liveBt.Status.Secrets, len(expectedSecrets)) - - for _, secret := range expectedSecrets { - secretRef, err := func(secretName string, keys []cephv1.SecretReference) (cephv1.SecretReference, error) { - for _, secretRef := range keys { - if secretRef.Name == secretName { - return secretRef, nil - } - } - return cephv1.SecretReference{}, fmt.Errorf("secretReference for secret %q not found in CephObjectStoreUser.status.keys", secret.Name) - }(secret.Name, liveBt.Status.Secrets) - require.NoError(t, err) - - // fetch the live secret for UID and ResourceVersion - liveSecret, err := k8sh.Clientset.CoreV1().Secrets(secret.Namespace).Get(ctx, secret.Name, metav1.GetOptions{}) - require.NoError(t, err) - - assert.Equal(t, liveSecret.Name, secretRef.Name) - assert.Equal(t, liveSecret.Namespace, secretRef.Namespace) - assert.Equal(t, liveSecret.UID, secretRef.UID) - assert.Equal(t, liveSecret.ResourceVersion, secretRef.ResourceVersion) - } - }) +func checkStatusSecrets(t *testing.T, k8sh *utils.K8sHelper, bt *cephv1.CephBucketTopic, expected ...*corev1.Secret) { + secrets.RequireStatusRefs(t, k8sh, + k8sh.RookClientset.CephV1().CephBucketTopics(bt.Namespace), bt.Name, + fmt.Sprintf("cephBucketTopic %q has .status.secrets set", bt.Name), + func(topic *cephv1.CephBucketTopic) []cephv1.SecretReference { + if topic.Status == nil { + return nil + } + return topic.Status.Secrets + }, + expected...) } func checkRgwTopicEndpoint(t *testing.T, snsClient *sns.Client, arn, user, pass string) { t.Run(fmt.Sprintf("rgw topic arn %q has basic auth set", arn), func(t *testing.T) { - ctx := context.TODO() + ctx := t.Context() var uri *url.URL - inSync := utils.Retry(40, time.Second, "rgw topic basic auth in sync", func() bool { + wait4.RequireEventually(ctx, t, wait4.TimeoutShort, "rgw topic basic auth in sync", func(ctx context.Context) error { topicAttrs, err := snsClient.GetTopicAttributes(ctx, &sns.GetTopicAttributesInput{ TopicArn: &arn, }) - require.NoError(t, err) + if err != nil { + return err + } // the sns endpoint attributes are returned as JSON var endpointJSON map[string]interface{} - err = json.Unmarshal([]byte(topicAttrs.Attributes["EndPoint"]), &endpointJSON) - require.NoError(t, err) + if err := json.Unmarshal([]byte(topicAttrs.Attributes["EndPoint"]), &endpointJSON); err != nil { + return err + } - uri, err = url.Parse(string(endpointJSON["EndpointAddress"].(string))) - require.NoError(t, err) + addr, ok := endpointJSON["EndpointAddress"].(string) + if !ok { + return fmt.Errorf("topic endpoint has no EndpointAddress string: %v", endpointJSON["EndpointAddress"]) + } - uriPassword, _ := uri.User.Password() + parsed, err := url.Parse(addr) + if err != nil { + return err + } - return user == uri.User.Username() && pass == uriPassword + uriPassword, _ := parsed.User.Password() + if user != parsed.User.Username() || pass != uriPassword { + return fmt.Errorf("topic basic auth not yet in sync for user %q", parsed.User.Username()) + } + + // capture the matching sample for the post-wait assertions + uri = parsed + return nil }) - require.True(t, inSync) assert.Equal(t, "kafka", uri.Scheme) assert.Equal(t, "kafka.example.com:9094", uri.Host) - assert.Equal(t, user, uri.User.Username()) - uriPassword, _ := uri.User.Password() - assert.Equal(t, pass, uriPassword) }) } -// Note that .status.ARN is nil while .status.phase == Reconciling -func cephBucketTopicReady(t *testing.T, k8sh *utils.K8sHelper, bt *cephv1.CephBucketTopic) *cephv1.CephBucketTopic { - ctx := context.TODO() - - var liveBt *cephv1.CephBucketTopic - btReady := utils.Retry(60, time.Second, fmt.Sprintf("CephBucketTopic %q is Ready", bt.Name), func() bool { - var err error - liveBt, err = k8sh.RookClientset.CephV1().CephBucketTopics(bt.Namespace).Get(ctx, bt.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - if liveBt.Status == nil { - return false - } - - return liveBt.Status.Phase == string(cephv1.ConditionReady) - }) - require.True(t, btReady) - require.NotNil(t, liveBt.Status.ARN) - - return liveBt +// kvSecret returns a Secret holding a single key/value pair. +func kvSecret(ns, name, key, value string) *corev1.Secret { + return &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: ns, + }, + Data: map[string][]byte{ + key: []byte(value), + }, + } } -func TestBucketTopicKafka(t *testing.T, k8sh *utils.K8sHelper, installer *installer.CephInstaller, logger *capnslog.PackageLogger, tlsEnable bool, objectStore *cephv1.CephObjectStore) { +// secretKeySelector returns a selector for key within the named secret. +func secretKeySelector(name, key string) *corev1.SecretKeySelector { + return &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{ + Name: name, + }, + Key: key, + } +} + +const Namespace = "test-topickafka" + +func TestBucketTopicKafka(t *testing.T, k8sh *utils.K8sHelper, store *sharedstore.Sharedstore) { var ( - defaultName = "test-topickafka" + defaultName = Namespace + objectStore = store.ObjectStore() ns = &corev1.Namespace{ ObjectMeta: metav1.ObjectMeta{ @@ -140,56 +132,12 @@ func TestBucketTopicKafka(t *testing.T, k8sh *utils.K8sHelper, installer *instal }, } - secret1 = &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName + "-secret1", - Namespace: ns.Name, - }, - Data: map[string][]byte{ - "user-name": []byte("kafka-user1"), - }, - } + secret1 = kvSecret(ns.Name, defaultName+"-secret1", "user-name", "kafka-user1") + secret2 = kvSecret(ns.Name, defaultName+"-secret2", "password", "kafka-pass2") + secret3 = kvSecret(ns.Name, defaultName+"-secret3", "user-name", "kafka-user3") + secret4 = kvSecret(ns.Name, defaultName+"-secret4", "password", "kafka-pass4") - secret2 = &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName + "-secret2", - Namespace: ns.Name, - }, - Data: map[string][]byte{ - "password": []byte("kafka-pass2"), - }, - } - - secret3 = &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName + "-secret3", - Namespace: ns.Name, - }, - Data: map[string][]byte{ - "user-name": []byte("kafka-user3"), - }, - } - - secret4 = &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName + "-secret4", - Namespace: ns.Name, - }, - Data: map[string][]byte{ - "password": []byte("kafka-pass4"), - }, - } - - storageClass = &storagev1.StorageClass{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName, - }, - Provisioner: objectStore.Namespace + ".ceph.rook.io/bucket", - Parameters: map[string]string{ - "objectStoreName": objectStore.Name, - "objectStoreNamespace": objectStore.Namespace, - }, - } + storageClass = fixture.StorageClass(defaultName, objectStore) bt1 = &cephv1.CephBucketTopic{ ObjectMeta: metav1.ObjectMeta{ @@ -202,21 +150,11 @@ func TestBucketTopicKafka(t *testing.T, k8sh *utils.K8sHelper, installer *instal Persistent: false, Endpoint: cephv1.TopicEndpointSpec{ Kafka: &cephv1.KafkaEndpointSpec{ - URI: "kafka://kafka.example.com:9094", - AckLevel: "broker", - UseSSL: false, - UserSecretRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret1.Name, - }, - Key: "user-name", - }, - PasswordSecretRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret2.Name, - }, - Key: "password", - }, + URI: "kafka://kafka.example.com:9094", + AckLevel: "broker", + UseSSL: false, + UserSecretRef: secretKeySelector(secret1.Name, "user-name"), + PasswordSecretRef: secretKeySelector(secret2.Name, "password"), }, }, }, @@ -249,275 +187,172 @@ func TestBucketTopicKafka(t *testing.T, k8sh *utils.K8sHelper, installer *instal }, } - ctx = context.TODO() - snsClient *sns.Client + snsClient = store.SnsClient() + + btClient = k8sh.RookClientset.CephV1().CephBucketTopics(ns.Name) + bnClient = k8sh.RookClientset.CephV1().CephBucketNotifications(ns.Name) + obcClient = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name) + obClient = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets() + secretClient = k8sh.Clientset.CoreV1().Secrets(ns.Name) ) t.Run("CephBucketTopic kafka", func(t *testing.T) { - if tlsEnable { - // Skip testing with and without TLS to reduce test time - t.Skip("skipping test for TLS enabled clusters") + ctx := t.Context() + + fixture.RequireNamespace(t, k8sh, ns) + + for _, secret := range []*corev1.Secret{secret1, secret2, secret3, secret4} { + t.Run(fmt.Sprintf("create secret %q", secret.Name), func(t *testing.T) { + _, err := secretClient.Create(ctx, secret, metav1.CreateOptions{}) + require.NoError(t, err) + }) } - t.Run(fmt.Sprintf("create ns %q", ns.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create secret %q", secret1.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret1.Namespace).Create(ctx, secret1, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create secret %q", secret2.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret2.Namespace).Create(ctx, secret2, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create secret %q", secret3.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Create(ctx, secret3, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create secret %q", secret4.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Create(ctx, secret4, metav1.CreateOptions{}) - require.NoError(t, err) - }) - // the sc, obc, and CephBucketNotification are essentially unused for testing but are created for completeness - t.Run(fmt.Sprintf("create sc %q", storageClass.Name), func(t *testing.T) { - _, err := k8sh.Clientset.StorageV1().StorageClasses().Create(ctx, storageClass, metav1.CreateOptions{}) - require.NoError(t, err) - }) + fixture.RequireStorageClass(t, k8sh, storageClass) t.Run(fmt.Sprintf("create obc %q", obc1.Name), func(t *testing.T) { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc1.Namespace).Create(ctx, obc1, metav1.CreateOptions{}) + _, err := obcClient.Create(ctx, obc1, metav1.CreateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("create CephBucketTopic %q", bt1.Name), func(t *testing.T) { - _, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Create(ctx, bt1, metav1.CreateOptions{}) - require.NoError(t, err) - - // user creation may be slow right after rgw start up - cephBucketTopicReady(t, k8sh, bt1) + // topic creation may be slow right after rgw start up + wait4.RequireCreate(ctx, t, btClient, bt1, wait4.BucketTopic, wait4.TimeoutMedium) }) t.Run(fmt.Sprintf("create CephBucketNotification %q", bn1.Name), func(t *testing.T) { - _, err := k8sh.RookClientset.CephV1().CephBucketNotifications(bn1.Namespace).Create(ctx, bn1, metav1.CreateOptions{}) + _, err := bnClient.Create(ctx, bn1, metav1.CreateOptions{}) require.NoError(t, err) }) - t.Run("setup sns client", func(t *testing.T) { - var err error - snsClient, err = utilsns.NewClient(objectStore, k8sh, installer, tlsEnable) - require.NoError(t, err) - }) - - { - liveBt := cephBucketTopicReady(t, k8sh, bt1) - - checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(secret1.Data["user-name"]), string(secret2.Data["password"])) - - secrets := []*corev1.Secret{secret1, secret2} - - checkStatusSecrets(t, k8sh, bt1, secrets) - } + liveBt := wait4.RequireCondition(ctx, t, btClient, bt1.Name, wait4.BucketTopic, wait4.TimeoutMedium) + checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(secret1.Data["user-name"]), string(secret2.Data["password"])) + checkStatusSecrets(t, k8sh, bt1, secret1, secret2) t.Run("updating referenced secrets reconciles rgw topic", func(t *testing.T) { t.Run(fmt.Sprintf("update kafka auth on CephBucketTopic %q", bt1.Name), func(t *testing.T) { - liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{}) + liveBt, err := btClient.Get(ctx, bt1.Name, metav1.GetOptions{}) require.NoError(t, err) - liveBt.Spec.Endpoint.Kafka.UserSecretRef = &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret3.Name, - }, - Key: "user-name", - } + liveBt.Spec.Endpoint.Kafka.UserSecretRef = secretKeySelector(secret3.Name, "user-name") + liveBt.Spec.Endpoint.Kafka.PasswordSecretRef = secretKeySelector(secret4.Name, "password") - liveBt.Spec.Endpoint.Kafka.PasswordSecretRef = &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret4.Name, - }, - Key: "password", - } - - _, err = k8sh.RookClientset.CephV1().CephBucketTopics(liveBt.Namespace).Update(ctx, liveBt, metav1.UpdateOptions{}) + _, err = btClient.Update(ctx, liveBt, metav1.UpdateOptions{}) require.NoError(t, err) }) - { - liveBt := cephBucketTopicReady(t, k8sh, bt1) - - checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(secret3.Data["user-name"]), string(secret4.Data["password"])) - - secrets := []*corev1.Secret{secret3, secret4} - - checkStatusSecrets(t, k8sh, bt1, secrets) - } + liveBt := wait4.RequireCondition(ctx, t, btClient, bt1.Name, wait4.BucketTopic, wait4.TimeoutMedium) + checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(secret3.Data["user-name"]), string(secret4.Data["password"])) + checkStatusSecrets(t, k8sh, bt1, secret3, secret4) }) t.Run("removing referenced secrets reconciles rgw topic", func(t *testing.T) { t.Run(fmt.Sprintf("delete kafka auth on CephBucketTopic %q", bt1.Name), func(t *testing.T) { - liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{}) + liveBt, err := btClient.Get(ctx, bt1.Name, metav1.GetOptions{}) require.NoError(t, err) liveBt.Spec.Endpoint.Kafka.UserSecretRef = nil liveBt.Spec.Endpoint.Kafka.PasswordSecretRef = nil - _, err = k8sh.RookClientset.CephV1().CephBucketTopics(liveBt.Namespace).Update(ctx, liveBt, metav1.UpdateOptions{}) + _, err = btClient.Update(ctx, liveBt, metav1.UpdateOptions{}) require.NoError(t, err) }) - { - liveBt := cephBucketTopicReady(t, k8sh, bt1) - - checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, "", "") - - secrets := []*corev1.Secret{} - - checkStatusSecrets(t, k8sh, bt1, secrets) - } + liveBt := wait4.RequireCondition(ctx, t, btClient, bt1.Name, wait4.BucketTopic, wait4.TimeoutMedium) + checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, "", "") + checkStatusSecrets(t, k8sh, bt1) }) t.Run("adding new referenced secrets reconciles rgw topic", func(t *testing.T) { t.Run(fmt.Sprintf("add kafka auth on CephBucketTopic %q", bt1.Name), func(t *testing.T) { - liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{}) + liveBt, err := btClient.Get(ctx, bt1.Name, metav1.GetOptions{}) require.NoError(t, err) - liveBt.Spec.Endpoint.Kafka.UserSecretRef = &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret3.Name, - }, - Key: "user-name", - } + liveBt.Spec.Endpoint.Kafka.UserSecretRef = secretKeySelector(secret3.Name, "user-name") + liveBt.Spec.Endpoint.Kafka.PasswordSecretRef = secretKeySelector(secret4.Name, "password") - liveBt.Spec.Endpoint.Kafka.PasswordSecretRef = &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret4.Name, - }, - Key: "password", - } - - _, err = k8sh.RookClientset.CephV1().CephBucketTopics(liveBt.Namespace).Update(ctx, liveBt, metav1.UpdateOptions{}) + _, err = btClient.Update(ctx, liveBt, metav1.UpdateOptions{}) require.NoError(t, err) }) - { - liveBt := cephBucketTopicReady(t, k8sh, bt1) - - checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(secret3.Data["user-name"]), string(secret4.Data["password"])) - - secrets := []*corev1.Secret{secret3, secret4} - - checkStatusSecrets(t, k8sh, bt1, secrets) - } + liveBt := wait4.RequireCondition(ctx, t, btClient, bt1.Name, wait4.BucketTopic, wait4.TimeoutMedium) + checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(secret3.Data["user-name"]), string(secret4.Data["password"])) + checkStatusSecrets(t, k8sh, bt1, secret3, secret4) }) t.Run("updating secrets reconciles rgw topic", func(t *testing.T) { t.Run(fmt.Sprintf("update secret %q data", secret3.Name), func(t *testing.T) { - liveSecret, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Get(ctx, secret3.Name, metav1.GetOptions{}) + liveSecret, err := secretClient.Get(ctx, secret3.Name, metav1.GetOptions{}) require.NoError(t, err) liveSecret.Data = map[string][]byte{ "user-name": []byte("kafka-user3-updated"), } - _, err = k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Update(ctx, liveSecret, metav1.UpdateOptions{}) + _, err = secretClient.Update(ctx, liveSecret, metav1.UpdateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("update secret %q data", secret4.Name), func(t *testing.T) { - liveSecret, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Get(ctx, secret4.Name, metav1.GetOptions{}) + liveSecret, err := secretClient.Get(ctx, secret4.Name, metav1.GetOptions{}) require.NoError(t, err) liveSecret.Data = map[string][]byte{ "password": []byte("kafka-pass4-updated"), } - _, err = k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Update(ctx, liveSecret, metav1.UpdateOptions{}) + _, err = secretClient.Update(ctx, liveSecret, metav1.UpdateOptions{}) require.NoError(t, err) }) - { - liveBt := cephBucketTopicReady(t, k8sh, bt1) + liveBt := wait4.RequireCondition(ctx, t, btClient, bt1.Name, wait4.BucketTopic, wait4.TimeoutMedium) - liveSecret3, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Get(ctx, secret3.Name, metav1.GetOptions{}) - require.NoError(t, err) + liveSecret3, err := secretClient.Get(ctx, secret3.Name, metav1.GetOptions{}) + require.NoError(t, err) - liveSecret4, err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Get(ctx, secret4.Name, metav1.GetOptions{}) - require.NoError(t, err) + liveSecret4, err := secretClient.Get(ctx, secret4.Name, metav1.GetOptions{}) + require.NoError(t, err) - checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(liveSecret3.Data["user-name"]), string(liveSecret4.Data["password"])) - - secrets := []*corev1.Secret{liveSecret3, liveSecret4} - - checkStatusSecrets(t, k8sh, bt1, secrets) - } + checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(liveSecret3.Data["user-name"]), string(liveSecret4.Data["password"])) + checkStatusSecrets(t, k8sh, bt1, liveSecret3, liveSecret4) }) t.Run("deleting a referenced secret triggers a reconcile, which fails", func(t *testing.T) { t.Run(fmt.Sprintf("delete secret %q", secret4.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Delete(ctx, secret4.Name, metav1.DeleteOptions{}) + err := secretClient.Delete(ctx, secret4.Name, metav1.DeleteOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("CephBucketTopic %q has phase ReconcileFailed", bt1.Name), func(t *testing.T) { - btReady := utils.Retry(40, time.Second, "CephBucketTopic is ReconcileFailed", func() bool { - liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - if liveBt.Status == nil { - return false - } - - return liveBt.Status.Phase == string(cephv1.ReconcileFailed) - }) - require.True(t, btReady) + wait4.RequireCondition(ctx, t, btClient, bt1.Name, wait4.BucketTopicPhase(string(cephv1.ReconcileFailed)), wait4.TimeoutShort) }) t.Run(fmt.Sprintf("create secret %q", secret4.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Create(ctx, secret4, metav1.CreateOptions{}) + _, err := secretClient.Create(ctx, secret4, metav1.CreateOptions{}) require.NoError(t, err) }) - cephBucketTopicReady(t, k8sh, bt1) + wait4.RequireCondition(ctx, t, btClient, bt1.Name, wait4.BucketTopic, wait4.TimeoutMedium) }) t.Run(fmt.Sprintf("delete CephBucketNotification %q", bn1.Name), func(t *testing.T) { - err := k8sh.RookClientset.CephV1().CephBucketNotifications(bn1.Namespace).Delete(ctx, bn1.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - - absent := utils.Retry(40, time.Second, "CephBucketNotification is absent", func() bool { - _, err := k8sh.RookClientset.CephV1().CephBucketNotifications(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + wait4.AssertDelete(ctx, t, bnClient, bn1.Name, wait4.TimeoutShort) }) t.Run(fmt.Sprintf("no CephBucketNotification(s) in ns %q", ns.Name), func(t *testing.T) { - list, err := k8sh.RookClientset.CephV1().CephBucketNotifications(ns.Name).List(ctx, metav1.ListOptions{}) + list, err := bnClient.List(ctx, metav1.ListOptions{}) require.NoError(t, err) assert.Len(t, list.Items, 0) }) t.Run(fmt.Sprintf("delete CephBucketTopic %q", bt1.Name), func(t *testing.T) { - err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Delete(ctx, bt1.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - - absent := utils.Retry(40, time.Second, "CephBucketTopic is absent", func() bool { - _, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + wait4.AssertDelete(ctx, t, btClient, bt1.Name, wait4.TimeoutShort) }) t.Run(fmt.Sprintf("no CephBucketTopic(s) in ns %q", ns.Name), func(t *testing.T) { - list, err := k8sh.RookClientset.CephV1().CephBucketTopics(ns.Name).List(ctx, metav1.ListOptions{}) + list, err := btClient.List(ctx, metav1.ListOptions{}) require.NoError(t, err) assert.Len(t, list.Items, 0) @@ -525,29 +360,17 @@ func TestBucketTopicKafka(t *testing.T, k8sh *utils.K8sHelper, installer *instal t.Run(fmt.Sprintf("delete obc %q", obc1.Name), func(t *testing.T) { // lookup ob name - liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc1.Namespace).Get(ctx, obc1.Name, metav1.GetOptions{}) + liveObc, err := obcClient.Get(ctx, obc1.Name, metav1.GetOptions{}) require.NoError(t, err) obName := liveObc.Spec.ObjectBucketName - // delete obc - err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc1.Namespace).Delete(ctx, obc1.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - - absent := utils.Retry(40, time.Second, "OBC is absent", func() bool { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc1.Namespace).Get(ctx, obc1.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) - - absent = utils.Retry(40, time.Second, "OB is absent", func() bool { - _, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + // delete obc; the backing ob is garbage-collected by the provisioner + wait4.AssertDelete(ctx, t, obcClient, obc1.Name, wait4.TimeoutShort) + wait4.AssertAbsent(ctx, t, obClient, obName, wait4.TimeoutShort) }) t.Run(fmt.Sprintf("no obc(s) in ns %q", ns.Name), func(t *testing.T) { - list, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).List(ctx, metav1.ListOptions{}) + list, err := obcClient.List(ctx, metav1.ListOptions{}) require.NoError(t, err) assert.Len(t, list.Items, 0) @@ -555,61 +378,51 @@ func TestBucketTopicKafka(t *testing.T, k8sh *utils.K8sHelper, installer *instal t.Run("CephBucketTopic deletion did not remove any secrets", func(t *testing.T) { t.Run(fmt.Sprintf("secret %q still exists", secret1.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret1.Namespace).Get(ctx, secret1.Name, metav1.GetOptions{}) + _, err := secretClient.Get(ctx, secret1.Name, metav1.GetOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("secret %q still exists", secret2.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret2.Namespace).Get(ctx, secret2.Name, metav1.GetOptions{}) + _, err := secretClient.Get(ctx, secret2.Name, metav1.GetOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("secret %q still exists", secret3.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Get(ctx, secret3.Name, metav1.GetOptions{}) + _, err := secretClient.Get(ctx, secret3.Name, metav1.GetOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("secret %q still exists", secret4.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Get(ctx, secret4.Name, metav1.GetOptions{}) + _, err := secretClient.Get(ctx, secret4.Name, metav1.GetOptions{}) require.NoError(t, err) }) }) - t.Run(fmt.Sprintf("delete sc %q", storageClass.Name), func(t *testing.T) { - err := k8sh.Clientset.StorageV1().StorageClasses().Delete(ctx, storageClass.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - }) - t.Run(fmt.Sprintf("delete secret %q", secret4.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Delete(ctx, secret4.Name, metav1.DeleteOptions{}) + err := secretClient.Delete(ctx, secret4.Name, metav1.DeleteOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("delete secret %q", secret3.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Delete(ctx, secret3.Name, metav1.DeleteOptions{}) + err := secretClient.Delete(ctx, secret3.Name, metav1.DeleteOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("delete secret %q", secret2.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Delete(ctx, secret2.Name, metav1.DeleteOptions{}) + err := secretClient.Delete(ctx, secret2.Name, metav1.DeleteOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("delete secret %q", secret1.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Secrets(secret1.Namespace).Delete(ctx, secret1.Name, metav1.DeleteOptions{}) + err := secretClient.Delete(ctx, secret1.Name, metav1.DeleteOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("no secrets in ns %q", ns.Name), func(t *testing.T) { - secrets, err := k8sh.Clientset.CoreV1().Secrets(ns.Name).List(ctx, metav1.ListOptions{}) + secrets, err := secretClient.List(ctx, metav1.ListOptions{}) require.NoError(t, err) assert.Len(t, secrets.Items, 0) }) - - t.Run(fmt.Sprintf("delete ns %q", ns.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Namespaces().Delete(ctx, ns.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - }) }) } diff --git a/tests/integration/object/user/caps/caps.go b/tests/integration/object/user/caps/caps.go index 0a3b3fe47..9cb2722e2 100644 --- a/tests/integration/object/user/caps/caps.go +++ b/tests/integration/object/user/caps/caps.go @@ -20,10 +20,8 @@ import ( "context" "fmt" "testing" - "time" "github.com/ceph/go-ceph/rgw/admin" - "github.com/coreos/pkg/capnslog" "github.com/google/go-cmp/cmp" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -31,14 +29,19 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/tests/framework/installer" "github.com/rook/rook/tests/framework/utils" - utiladmin "github.com/rook/rook/tests/integration/object/util/admin" + "github.com/rook/rook/tests/integration/object/util/fixture" + "github.com/rook/rook/tests/integration/object/util/sharedstore" + "github.com/rook/rook/tests/integration/object/util/wait4" ) -func TestObjectStoreUserCaps(t *testing.T, k8sh *utils.K8sHelper, installer *installer.CephInstaller, logger *capnslog.PackageLogger, tlsEnable bool, objectStore *cephv1.CephObjectStore) { +const Namespace = "test-usercaps" + +func TestObjectStoreUserCaps(t *testing.T, k8sh *utils.K8sHelper, store *sharedstore.Sharedstore) { var ( - defaultName = "test-usercaps" + defaultName = Namespace + objectStore = store.ObjectStore() + adminClient = store.AdminClient() ns = &corev1.Namespace{ ObjectMeta: metav1.ObjectMeta{ @@ -56,47 +59,18 @@ func TestObjectStoreUserCaps(t *testing.T, k8sh *utils.K8sHelper, installer *ins ClusterNamespace: objectStore.Namespace, }, } + + osuClient = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name) ) t.Run("ObjectStoreUser caps", func(t *testing.T) { - if tlsEnable { - // Skip testing with and without TLS to reduce test time - t.Skip("skipping test for TLS enabled clusters") - } + ctx := t.Context() - var adminClient *admin.API - ctx := context.TODO() - - t.Run(fmt.Sprintf("create ns %q", ns.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}) - require.NoError(t, err) - }) + fixture.RequireNamespace(t, k8sh, ns) t.Run(fmt.Sprintf("create CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Create(ctx, &osu1, metav1.CreateOptions{}) - require.NoError(t, err) - // user creation may be slow right after rgw start up - osuReady := utils.Retry(120, time.Second, "CephObjectStoreUser is Ready", func() bool { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - if liveOsu.Status == nil { - return false - } - - return liveOsu.Status.Phase == string(cephv1.ConditionReady) - }) - require.True(t, osuReady) - }) - - // wait for cosu user to be ready so we know rgw admin api is ready - t.Run("setup rgw admin api client", func(t *testing.T) { - var err error - adminClient, err = utiladmin.NewAdminClient(objectStore, installer, k8sh, tlsEnable) - require.NoError(t, err) + wait4.RequireCreate(ctx, t, osuClient, &osu1, wait4.ObjectStoreUser, wait4.TimeoutLong) }) t.Run(fmt.Sprintf("verify caps on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { @@ -106,7 +80,7 @@ func TestObjectStoreUserCaps(t *testing.T, k8sh *utils.K8sHelper, installer *ins }) t.Run(fmt.Sprintf("update caps on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + liveOsu, err := osuClient.Get(ctx, osu1.Name, metav1.GetOptions{}) require.NoError(t, err) liveOsu.Spec.Capabilities = &cephv1.ObjectUserCapSpec{ @@ -114,18 +88,18 @@ func TestObjectStoreUserCaps(t *testing.T, k8sh *utils.K8sHelper, installer *ins Usage: "read", } - _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + _, err = osuClient.Update(ctx, liveOsu, metav1.UpdateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("verify caps on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - converged := utils.Retry(30, time.Second, "caps updated", func() bool { + wait4.AssertEventually(ctx, t, wait4.TimeoutShort, "rgw user caps match spec", func(ctx context.Context) error { liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) if err != nil { - return false + return err } - return cmp.Equal(liveUser.Caps, []admin.UserCapSpec{ + if !cmp.Equal(liveUser.Caps, []admin.UserCapSpec{ { Type: "buckets", Perm: "*", @@ -134,56 +108,46 @@ func TestObjectStoreUserCaps(t *testing.T, k8sh *utils.K8sHelper, installer *ins Type: "usage", Perm: "read", }, - }) + }) { + return fmt.Errorf("caps not yet in sync: %+v", liveUser.Caps) + } + return nil }) - - assert.True(t, converged) }) t.Run(fmt.Sprintf("remove caps on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + liveOsu, err := osuClient.Get(ctx, osu1.Name, metav1.GetOptions{}) require.NoError(t, err) liveOsu.Spec.Capabilities = nil - _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + _, err = osuClient.Update(ctx, liveOsu, metav1.UpdateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("verify caps on CephObjectStoreUser %q returns to default", osu1.Name), func(t *testing.T) { - converged := utils.Retry(30, time.Second, "caps updated", func() bool { + wait4.AssertEventually(ctx, t, wait4.TimeoutShort, "rgw user caps are empty", func(ctx context.Context) error { liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) if err != nil { - return false + return err } - return len(liveUser.Caps) == 0 + if len(liveUser.Caps) != 0 { + return fmt.Errorf("caps not yet empty: %+v", liveUser.Caps) + } + return nil }) - - assert.True(t, converged) }) t.Run(fmt.Sprintf("delete CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Delete(ctx, osu1.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - - absent := utils.Retry(40, time.Second, "CephObjectStoreUser is absent", func() bool { - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + wait4.AssertDelete(ctx, t, osuClient, osu1.Name, wait4.TimeoutShort) }) t.Run(fmt.Sprintf("no CephObjectStoreUsers in ns %q", ns.Name), func(t *testing.T) { - osus, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).List(ctx, metav1.ListOptions{}) + osus, err := osuClient.List(ctx, metav1.ListOptions{}) require.NoError(t, err) assert.Len(t, osus.Items, 0) }) - - t.Run(fmt.Sprintf("delete ns %q", ns.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Namespaces().Delete(ctx, ns.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - }) }) } diff --git a/tests/integration/object/user/keys/keys.go b/tests/integration/object/user/keys/keys.go index 9cefcdba7..e6e8905a6 100644 --- a/tests/integration/object/user/keys/keys.go +++ b/tests/integration/object/user/keys/keys.go @@ -20,19 +20,19 @@ import ( "context" "fmt" "testing" - "time" "github.com/ceph/go-ceph/rgw/admin" - "github.com/coreos/pkg/capnslog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/tests/framework/installer" "github.com/rook/rook/tests/framework/utils" - utiladmin "github.com/rook/rook/tests/integration/object/util/admin" + "github.com/rook/rook/tests/integration/object/util/fixture" + "github.com/rook/rook/tests/integration/object/util/secrets" + "github.com/rook/rook/tests/integration/object/util/sharedstore" + "github.com/rook/rook/tests/integration/object/util/wait4" ) // generate the secret name the operator is expected to generate for the CephObjectStoreUser @@ -40,6 +40,40 @@ func generateObjectStoreUserSecretName(osu cephv1.CephObjectStoreUser) string { return "rook-ceph-object-user-" + osu.Spec.Store + "-" + osu.Name } +// awsKeySecret returns a Secret holding an s3 access/secret key pair under the +// conventional AWS env var names. +func awsKeySecret(ns, name, accessKey, secretKey string) *corev1.Secret { + return &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: ns, + }, + Data: map[string][]byte{ + "AWS_ACCESS_KEY_ID": []byte(accessKey), + "AWS_SECRET_ACCESS_KEY": []byte(secretKey), + }, + } +} + +// objectUserKey returns an ObjectUserKey referencing the access/secret key +// pair stored in the named awsKeySecret secret. +func objectUserKey(secretName string) cephv1.ObjectUserKey { + return cephv1.ObjectUserKey{ + AccessKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{ + Name: secretName, + }, + Key: "AWS_ACCESS_KEY_ID", + }, + SecretKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{ + Name: secretName, + }, + Key: "AWS_SECRET_ACCESS_KEY", + }, + } +} + // find a UserKeySpec by AccessKey value func findUserKeySpec(keys []admin.UserKeySpec, accessKey string) (admin.UserKeySpec, error) { for _, k := range keys { @@ -50,71 +84,63 @@ func findUserKeySpec(keys []admin.UserKeySpec, accessKey string) (admin.UserKeyS return admin.UserKeySpec{}, fmt.Errorf("UserKeySpec with AccessKey %q not found", accessKey) } -func checkStatusKeys(t *testing.T, k8sh *utils.K8sHelper, osu cephv1.CephObjectStoreUser, expectedSecrets []*corev1.Secret) { - t.Run(fmt.Sprintf("cephObjectStoreUser %q has .status.keys set", osu.Name), func(t *testing.T) { - ctx := context.TODO() - - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(osu.Namespace).Get(ctx, osu.Name, metav1.GetOptions{}) - require.NoError(t, err) - - require.NotNil(t, liveOsu.Status) - assert.Len(t, liveOsu.Status.Keys, len(expectedSecrets)) - - for _, secret := range expectedSecrets { - secretRef, err := func(secretName string, keys []cephv1.SecretReference) (cephv1.SecretReference, error) { - for _, secretRef := range keys { - if secretRef.Name == secretName { - return secretRef, nil - } - } - return cephv1.SecretReference{}, fmt.Errorf("secretReference for secret %q not found in CephObjectStoreUser.status.keys", secret.Name) - }(secret.Name, liveOsu.Status.Keys) - require.NoError(t, err) - - // fetch the live secret for UID and ResourceVersion - liveSecret, err := k8sh.Clientset.CoreV1().Secrets(osu.Namespace).Get(ctx, secret.Name, metav1.GetOptions{}) - require.NoError(t, err) - - assert.Equal(t, liveSecret.Name, secretRef.Name) - assert.Equal(t, liveSecret.Namespace, secretRef.Namespace) - assert.Equal(t, liveSecret.UID, secretRef.UID) - assert.Equal(t, liveSecret.ResourceVersion, secretRef.ResourceVersion) - } - }) +func checkStatusKeys(t *testing.T, k8sh *utils.K8sHelper, osu cephv1.CephObjectStoreUser, expected ...*corev1.Secret) { + secrets.RequireStatusRefs(t, k8sh, + k8sh.RookClientset.CephV1().CephObjectStoreUsers(osu.Namespace), osu.Name, + fmt.Sprintf("cephObjectStoreUser %q has .status.keys set", osu.Name), + func(u *cephv1.CephObjectStoreUser) []cephv1.SecretReference { + if u.Status == nil { + return nil + } + return u.Status.Keys + }, + expected...) } -func checkRgwUserKeys(t *testing.T, adminClient *admin.API, osu cephv1.CephObjectStoreUser, expectedSecrets []*corev1.Secret, accessKeyName, secretKeyName string) { - t.Run(fmt.Sprintf("rgw user %q has keys set", osu.Name), func(t *testing.T) { - ctx := context.TODO() +// requireRgwUserKeys verifies every expected key on the rgw user — reporting +// all divergences rather than stopping at the first — and then aborts the +// caller if any check failed: the steps that follow assume the keys are in +// sync, so continuing would only cascade noise. +func requireRgwUserKeys(t *testing.T, adminClient *admin.API, osu cephv1.CephObjectStoreUser, accessKeyName, secretKeyName string, expected ...*corev1.Secret) { + t.Helper() - for _, secret := range expectedSecrets { - var keySpec admin.UserKeySpec + if !t.Run(fmt.Sprintf("rgw user %q has keys set", osu.Name), func(t *testing.T) { + ctx := t.Context() + for _, secret := range expected { // assume that the .Phase doesn't change when updating keys - inSync := utils.Retry(40, time.Second, fmt.Sprintf("CephObjectStoreUser has key in sync with secret %q ", secret.Name), func() bool { + wait4.AssertEventually(ctx, t, wait4.TimeoutShort, fmt.Sprintf("rgw user key in sync with secret %q", secret.Name), func(ctx context.Context) error { liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu.Name}) if err != nil { - return false + return err } - keySpec, err = findUserKeySpec(liveUser.Keys, string(secret.Data[accessKeyName])) - return err == nil + keySpec, err := findUserKeySpec(liveUser.Keys, string(secret.Data[accessKeyName])) + if err != nil { + return err + } + if keySpec.SecretKey != string(secret.Data[secretKeyName]) { + return fmt.Errorf("secret key for %q not yet in sync", secret.Name) + } + return nil }) - require.True(t, inSync) - - assert.Equal(t, string(secret.Data[accessKeyName]), keySpec.AccessKey) - assert.Equal(t, string(secret.Data[secretKeyName]), keySpec.SecretKey) } // check that no extra keys are present liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu.Name}) require.NoError(t, err) - assert.Equal(t, len(expectedSecrets), len(liveUser.Keys)) - }) + assert.Equal(t, len(expected), len(liveUser.Keys)) + }) { + t.FailNow() + } } -func TestObjectStoreUserKeys(t *testing.T, k8sh *utils.K8sHelper, installer *installer.CephInstaller, logger *capnslog.PackageLogger, tlsEnable bool, objectStore *cephv1.CephObjectStore) { +const Namespace = "test-userkeys" + +func TestObjectStoreUserKeys(t *testing.T, k8sh *utils.K8sHelper, store *sharedstore.Sharedstore) { var ( - defaultName = "test-userkeys" + defaultName = Namespace + objectStore = store.ObjectStore() + adminClient = store.AdminClient() ns = &corev1.Namespace{ ObjectMeta: metav1.ObjectMeta{ @@ -122,60 +148,11 @@ func TestObjectStoreUserKeys(t *testing.T, k8sh *utils.K8sHelper, installer *ins }, } - secret1 = &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName + "-secret1", - Namespace: ns.Name, - }, - Data: map[string][]byte{ - "AWS_ACCESS_KEY_ID": []byte("P1ANF2BP4K2LPGOR5SBB"), - "AWS_SECRET_ACCESS_KEY": []byte("yfCHPhhOed0vJkqZsGEODyJrdKqHD09OMTWCnwjX"), - }, - } - - secret2 = &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName + "-secret2", - Namespace: ns.Name, - }, - Data: map[string][]byte{ - "AWS_ACCESS_KEY_ID": []byte("2I6RPUTQFMNCSYEXZ6VM"), - "AWS_SECRET_ACCESS_KEY": []byte("uY066SWPfaOVlDeYc7GJyOTfkDejyDdXrqehS6wx"), - }, - } - - secret3 = &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName + "-secret3", - Namespace: ns.Name, - }, - Data: map[string][]byte{ - "AWS_ACCESS_KEY_ID": []byte("J4D0P20F3EDR51OSND7Y"), - "AWS_SECRET_ACCESS_KEY": []byte("jn89OpkXNoDlIHVVQ23mE2DZgPmuDK3WVH5ExOvQ"), - }, - } - - secret4 = &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName + "-secret4", - Namespace: ns.Name, - }, - Data: map[string][]byte{ - "AWS_ACCESS_KEY_ID": []byte("MPZX7DG5WJWQ6VPCSLYT"), - "AWS_SECRET_ACCESS_KEY": []byte("phh7DIxnLPeSD2V6FUouhmnWrKlKRD5dBykyXozX"), - }, - } - - secret5 = &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: defaultName + "-secret5", - Namespace: ns.Name, - }, - Data: map[string][]byte{ - "AWS_ACCESS_KEY_ID": []byte("7TNSSANCO5KXK23IPT91"), - "AWS_SECRET_ACCESS_KEY": []byte("HksEDf0hEh3PtTvl7s9x6CyXfkWuY8eAMYAcvH5l"), - }, - } + secret1 = awsKeySecret(ns.Name, defaultName+"-secret1", "P1ANF2BP4K2LPGOR5SBB", "yfCHPhhOed0vJkqZsGEODyJrdKqHD09OMTWCnwjX") + secret2 = awsKeySecret(ns.Name, defaultName+"-secret2", "2I6RPUTQFMNCSYEXZ6VM", "uY066SWPfaOVlDeYc7GJyOTfkDejyDdXrqehS6wx") + secret3 = awsKeySecret(ns.Name, defaultName+"-secret3", "J4D0P20F3EDR51OSND7Y", "jn89OpkXNoDlIHVVQ23mE2DZgPmuDK3WVH5ExOvQ") + secret4 = awsKeySecret(ns.Name, defaultName+"-secret4", "MPZX7DG5WJWQ6VPCSLYT", "phh7DIxnLPeSD2V6FUouhmnWrKlKRD5dBykyXozX") + secret5 = awsKeySecret(ns.Name, defaultName+"-secret5", "7TNSSANCO5KXK23IPT91", "HksEDf0hEh3PtTvl7s9x6CyXfkWuY8eAMYAcvH5l") osu1 = cephv1.CephObjectStoreUser{ ObjectMeta: metav1.ObjectMeta{ @@ -186,208 +163,85 @@ func TestObjectStoreUserKeys(t *testing.T, k8sh *utils.K8sHelper, installer *ins Store: objectStore.Name, ClusterNamespace: objectStore.Namespace, Keys: []cephv1.ObjectUserKey{ - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret1.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret1.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret2.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret2.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret3.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret3.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, + objectUserKey(secret1.Name), + objectUserKey(secret2.Name), + objectUserKey(secret3.Name), }, }, } + + osuClient = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name) + secretClient = k8sh.Clientset.CoreV1().Secrets(ns.Name) ) t.Run("ObjectStoreUser keys", func(t *testing.T) { - if tlsEnable { - // Skip testing with and without TLS to reduce test time - t.Skip("skipping test for TLS enabled clusters") + ctx := t.Context() + + fixture.RequireNamespace(t, k8sh, ns) + + for _, secret := range []*corev1.Secret{secret1, secret2, secret3, secret4, secret5} { + t.Run(fmt.Sprintf("create secret %q", secret.Name), func(t *testing.T) { + _, err := secretClient.Create(ctx, secret, metav1.CreateOptions{}) + require.NoError(t, err) + }) } - var adminClient *admin.API - ctx := context.TODO() - - t.Run(fmt.Sprintf("create ns %q", ns.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create secret %q", secret1.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(ns.Name).Create(ctx, secret1, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create secret %q", secret2.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(ns.Name).Create(ctx, secret2, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create secret %q", secret3.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(ns.Name).Create(ctx, secret3, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create secret %q", secret4.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(ns.Name).Create(ctx, secret4, metav1.CreateOptions{}) - require.NoError(t, err) - }) - - t.Run(fmt.Sprintf("create secret %q", secret5.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(ns.Name).Create(ctx, secret5, metav1.CreateOptions{}) - require.NoError(t, err) - }) - t.Run(fmt.Sprintf("create CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Create(ctx, &osu1, metav1.CreateOptions{}) - require.NoError(t, err) - // user creation may be slow right after rgw start up - osuReady := utils.Retry(120, time.Second, "CephObjectStoreUser is Ready", func() bool { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - if liveOsu.Status == nil { - return false - } - - return liveOsu.Status.Phase == string(cephv1.ConditionReady) - }) - require.True(t, osuReady) + wait4.RequireCreate(ctx, t, osuClient, &osu1, wait4.ObjectStoreUser, wait4.TimeoutLong) }) - t.Run("setup rgw admin api client", func(t *testing.T) { - var err error - adminClient, err = utiladmin.NewAdminClient(objectStore, installer, k8sh, tlsEnable) - require.NoError(t, err) - }) - - { - secrets := []*corev1.Secret{secret1, secret2, secret3} - - checkRgwUserKeys(t, adminClient, osu1, secrets, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY") - checkStatusKeys(t, k8sh, osu1, secrets) - } + requireRgwUserKeys(t, adminClient, osu1, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", secret1, secret2, secret3) + checkStatusKeys(t, k8sh, osu1, secret1, secret2, secret3) t.Run(fmt.Sprintf("update keys on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + liveOsu, err := osuClient.Get(ctx, osu1.Name, metav1.GetOptions{}) require.NoError(t, err) liveOsu.Spec.Keys = []cephv1.ObjectUserKey{ - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret4.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret4.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret5.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret5.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, + objectUserKey(secret4.Name), + objectUserKey(secret5.Name), } - _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + _, err = osuClient.Update(ctx, liveOsu, metav1.UpdateOptions{}) require.NoError(t, err) }) - { - secrets := []*corev1.Secret{secret4, secret5} - - checkRgwUserKeys(t, adminClient, osu1, secrets, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY") - checkStatusKeys(t, k8sh, osu1, secrets) - } + requireRgwUserKeys(t, adminClient, osu1, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", secret4, secret5) + checkStatusKeys(t, k8sh, osu1, secret4, secret5) // test transition from explicit keys -> automatic secret creation // when all explicit keys are removed from CephObjectStoreUser, one should // be left in place and the operator should [still] create a k8s secret for it t.Run(fmt.Sprintf("remove all keys set on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + liveOsu, err := osuClient.Get(ctx, osu1.Name, metav1.GetOptions{}) require.NoError(t, err) liveOsu.Spec.Keys = nil - _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + _, err = osuClient.Update(ctx, liveOsu, metav1.UpdateOptions{}) require.NoError(t, err) - // wait for the number of keys to drop to 1 - inSync := utils.Retry(40, time.Second, "CephObjectStoreUser has 1 key", func() bool { + wait4.RequireEventually(ctx, t, wait4.TimeoutShort, "rgw user has exactly 1 key", func(ctx context.Context) error { liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) if err != nil { - return false + return err } - return len(liveUser.Keys) == 1 + if len(liveUser.Keys) != 1 { + return fmt.Errorf("rgw user has %d keys, want 1", len(liveUser.Keys)) + } + return nil }) - require.True(t, inSync) }) - // keys updated on user - { - // fetch automatic secret as it should be the only key set on the rgw user - secretName := generateObjectStoreUserSecretName(osu1) - liveSecret, err := k8sh.Clientset.CoreV1().Secrets(ns.Name).Get(ctx, secretName, metav1.GetOptions{}) - require.NoError(t, err) + // fetch the automatic secret; it should hold the only key set on the rgw user + autoSecret, err := secretClient.Get(ctx, generateObjectStoreUserSecretName(osu1), metav1.GetOptions{}) + require.NoError(t, err) - secrets := []*corev1.Secret{liveSecret} - - checkRgwUserKeys(t, adminClient, osu1, secrets, "AccessKey", "SecretKey") - } + requireRgwUserKeys(t, adminClient, osu1, "AccessKey", "SecretKey", autoSecret) t.Run(fmt.Sprintf("cephObjectStoreUser %q .status.keys is unset", osu1.Name), func(t *testing.T) { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + liveOsu, err := osuClient.Get(ctx, osu1.Name, metav1.GetOptions{}) require.NoError(t, err) require.NotNil(t, liveOsu.Status) @@ -396,96 +250,27 @@ func TestObjectStoreUserKeys(t *testing.T, k8sh *utils.K8sHelper, installer *ins // test transition automatic secret creation -> explicit keys t.Run(fmt.Sprintf("add keys to CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + liveOsu, err := osuClient.Get(ctx, osu1.Name, metav1.GetOptions{}) require.NoError(t, err) liveOsu.Spec.Keys = []cephv1.ObjectUserKey{ - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret1.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret1.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret2.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret2.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret3.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret3.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret4.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret4.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, - { - AccessKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret5.Name, - }, - Key: "AWS_ACCESS_KEY_ID", - }, - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{ - Name: secret5.Name, - }, - Key: "AWS_SECRET_ACCESS_KEY", - }, - }, + objectUserKey(secret1.Name), + objectUserKey(secret2.Name), + objectUserKey(secret3.Name), + objectUserKey(secret4.Name), + objectUserKey(secret5.Name), } - _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + _, err = osuClient.Update(ctx, liveOsu, metav1.UpdateOptions{}) require.NoError(t, err) }) - { - secrets := []*corev1.Secret{secret1, secret2, secret3, secret4, secret5} - - checkRgwUserKeys(t, adminClient, osu1, secrets, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY") - checkStatusKeys(t, k8sh, osu1, secrets) - } + requireRgwUserKeys(t, adminClient, osu1, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", secret1, secret2, secret3, secret4, secret5) + checkStatusKeys(t, k8sh, osu1, secret1, secret2, secret3, secret4, secret5) // updating a secret already referenced by a CephObjectStoreUser should trigger a reconcile t.Run(fmt.Sprintf("update secret %q data", secret1.Name), func(t *testing.T) { - liveSecret, err := k8sh.Clientset.CoreV1().Secrets(secret1.Namespace).Get(ctx, secret1.Name, metav1.GetOptions{}) + liveSecret, err := secretClient.Get(ctx, secret1.Name, metav1.GetOptions{}) require.NoError(t, err) liveSecret.Data = map[string][]byte{ @@ -493,59 +278,34 @@ func TestObjectStoreUserKeys(t *testing.T, k8sh *utils.K8sHelper, installer *ins "AWS_SECRET_ACCESS_KEY": []byte("bar"), } - _, err = k8sh.Clientset.CoreV1().Secrets(secret1.Namespace).Update(ctx, liveSecret, metav1.UpdateOptions{}) + _, err = secretClient.Update(ctx, liveSecret, metav1.UpdateOptions{}) require.NoError(t, err) }) - { - // fetch secret that we modified - liveSecret, err := k8sh.Clientset.CoreV1().Secrets(secret1.Namespace).Get(ctx, secret1.Name, metav1.GetOptions{}) - require.NoError(t, err) + updatedSecret1, err := secretClient.Get(ctx, secret1.Name, metav1.GetOptions{}) + require.NoError(t, err) - secrets := []*corev1.Secret{liveSecret, secret2, secret3, secret4, secret5} - - checkRgwUserKeys(t, adminClient, osu1, secrets, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY") - checkStatusKeys(t, k8sh, osu1, secrets) - } + requireRgwUserKeys(t, adminClient, osu1, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", updatedSecret1, secret2, secret3, secret4, secret5) + checkStatusKeys(t, k8sh, osu1, updatedSecret1, secret2, secret3, secret4, secret5) // deleting a secret referenced by a CephObjectStoreUser should trigger a reconcile (which fails) t.Run(fmt.Sprintf("delete secret %q", secret1.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Secrets(ns.Name).Delete(ctx, secret1.Name, metav1.DeleteOptions{}) + err := secretClient.Delete(ctx, secret1.Name, metav1.DeleteOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("CephObjectStoreUser %q has phase ReconcileFailed", osu1.Name), func(t *testing.T) { - // user creation may be slow right after rgw start up - osuReady := utils.Retry(40, time.Second, "CephObjectStoreUser is ReconcileFailed", func() bool { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - if liveOsu.Status == nil { - return false - } - - return liveOsu.Status.Phase == string(cephv1.ReconcileFailed) - }) - require.True(t, osuReady) + wait4.RequireCondition(ctx, t, osuClient, osu1.Name, wait4.ObjectStoreUserPhase(string(cephv1.ReconcileFailed)), wait4.TimeoutShort) }) t.Run("missing referenced secret should not block CephObjectStoreUser deletion", func(t *testing.T) { t.Run(fmt.Sprintf("delete CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Delete(ctx, osu1.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - - absent := utils.Retry(40, time.Second, "CephObjectStoreUser is absent", func() bool { - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + wait4.AssertDelete(ctx, t, osuClient, osu1.Name, wait4.TimeoutShort) }) }) t.Run(fmt.Sprintf("no CephObjectStoreUsers in ns %q", ns.Name), func(t *testing.T) { - osus, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).List(ctx, metav1.ListOptions{}) + osus, err := osuClient.List(ctx, metav1.ListOptions{}) require.NoError(t, err) assert.Len(t, osus.Items, 0) @@ -558,39 +318,25 @@ func TestObjectStoreUserKeys(t *testing.T, k8sh *utils.K8sHelper, installer *ins // check that CephObjectStoreUser removal did not delete any secret resources - { - secrets := []*corev1.Secret{ - // secret1 was deleted in a previous step - secret2, - secret3, - secret4, - secret5, - } + // secret1 was deleted in a previous step + for _, secret := range []*corev1.Secret{secret2, secret3, secret4, secret5} { + t.Run(fmt.Sprintf("secret %q still exists", secret.Name), func(t *testing.T) { + _, err := secretClient.Get(ctx, secret.Name, metav1.GetOptions{}) + require.NoError(t, err) + }) - for _, secret := range secrets { - t.Run(fmt.Sprintf("secret %q still exists", secret.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Secrets(secret.Namespace).Get(ctx, secret.Name, metav1.GetOptions{}) - require.NoError(t, err) - }) - - // cleanup the secrets created for the test - t.Run(fmt.Sprintf("delete secret %q", secret.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Secrets(secret.Namespace).Delete(ctx, secret.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - }) - } + // cleanup the secrets created for the test + t.Run(fmt.Sprintf("delete secret %q", secret.Name), func(t *testing.T) { + err := secretClient.Delete(ctx, secret.Name, metav1.DeleteOptions{}) + require.NoError(t, err) + }) } t.Run(fmt.Sprintf("no secrets in ns %q", ns.Name), func(t *testing.T) { - secrets, err := k8sh.Clientset.CoreV1().Secrets(ns.Name).List(ctx, metav1.ListOptions{}) + secrets, err := secretClient.List(ctx, metav1.ListOptions{}) require.NoError(t, err) assert.Len(t, secrets.Items, 0) }) - - t.Run(fmt.Sprintf("delete ns %q", ns.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Namespaces().Delete(ctx, ns.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - }) }) } diff --git a/tests/integration/object/user/opmask/opmask.go b/tests/integration/object/user/opmask/opmask.go index e89bd8e4e..a0fc252fc 100644 --- a/tests/integration/object/user/opmask/opmask.go +++ b/tests/integration/object/user/opmask/opmask.go @@ -20,24 +20,27 @@ import ( "context" "fmt" "testing" - "time" "github.com/ceph/go-ceph/rgw/admin" - "github.com/coreos/pkg/capnslog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/tests/framework/installer" "github.com/rook/rook/tests/framework/utils" - utiladmin "github.com/rook/rook/tests/integration/object/util/admin" + "github.com/rook/rook/tests/integration/object/util/fixture" + "github.com/rook/rook/tests/integration/object/util/sharedstore" + "github.com/rook/rook/tests/integration/object/util/wait4" ) -func TestObjectStoreUserOpMask(t *testing.T, k8sh *utils.K8sHelper, installer *installer.CephInstaller, logger *capnslog.PackageLogger, tlsEnable bool, objectStore *cephv1.CephObjectStore) { +const Namespace = "test-useropmask" + +func TestObjectStoreUserOpMask(t *testing.T, k8sh *utils.K8sHelper, store *sharedstore.Sharedstore) { var ( - defaultName = "test-useropmask" + defaultName = Namespace + objectStore = store.ObjectStore() + adminClient = store.AdminClient() ns = &corev1.Namespace{ ObjectMeta: metav1.ObjectMeta{ @@ -55,47 +58,18 @@ func TestObjectStoreUserOpMask(t *testing.T, k8sh *utils.K8sHelper, installer *i ClusterNamespace: objectStore.Namespace, }, } + + osuClient = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name) ) - t.Run("ObjectStoreUser keys", func(t *testing.T) { - if tlsEnable { - // Skip testing with and without TLS to reduce test time - t.Skip("skipping test for TLS enabled clusters") - } + t.Run("ObjectStoreUser op_mask", func(t *testing.T) { + ctx := t.Context() - var adminClient *admin.API - ctx := context.TODO() - - t.Run(fmt.Sprintf("create ns %q", ns.Name), func(t *testing.T) { - _, err := k8sh.Clientset.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}) - require.NoError(t, err) - }) + fixture.RequireNamespace(t, k8sh, ns) t.Run(fmt.Sprintf("create CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Create(ctx, &osu1, metav1.CreateOptions{}) - require.NoError(t, err) - // user creation may be slow right after rgw start up - osuReady := utils.Retry(120, time.Second, "CephObjectStoreUser is Ready", func() bool { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) - if err != nil { - return false - } - - if liveOsu.Status == nil { - return false - } - - return liveOsu.Status.Phase == string(cephv1.ConditionReady) - }) - require.True(t, osuReady) - }) - - // wait for cosu user to be ready so we know rgw admin api is ready - t.Run("setup rgw admin api client", func(t *testing.T) { - var err error - adminClient, err = utiladmin.NewAdminClient(objectStore, installer, k8sh, tlsEnable) - require.NoError(t, err) + wait4.RequireCreate(ctx, t, osuClient, &osu1, wait4.ObjectStoreUser, wait4.TimeoutLong) }) t.Run(fmt.Sprintf("verify op_mask set on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { @@ -105,95 +79,86 @@ func TestObjectStoreUserOpMask(t *testing.T, k8sh *utils.K8sHelper, installer *i }) t.Run(fmt.Sprintf("update op-mask on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + liveOsu, err := osuClient.Get(ctx, osu1.Name, metav1.GetOptions{}) require.NoError(t, err) liveOsu.Spec.OpMask = &[]cephv1.ObjectUserOpMask{"read"} - _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + _, err = osuClient.Update(ctx, liveOsu, metav1.UpdateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("verify op_mask set on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - converged := utils.Retry(30, time.Second, "OpMask updated", func() bool { + wait4.AssertEventually(ctx, t, wait4.TimeoutShort, `rgw user op_mask is "read"`, func(ctx context.Context) error { liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) if err != nil { - return false + return err } - return liveUser.OpMask == "read" + if liveUser.OpMask != "read" { + return fmt.Errorf("op_mask not yet %q: %q", "read", liveUser.OpMask) + } + return nil }) - - assert.True(t, converged) }) t.Run(fmt.Sprintf("remove op-mask on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + liveOsu, err := osuClient.Get(ctx, osu1.Name, metav1.GetOptions{}) require.NoError(t, err) liveOsu.Spec.OpMask = nil - _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + _, err = osuClient.Update(ctx, liveOsu, metav1.UpdateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("verify op_mask set on CephObjectStoreUser %q returns to default", osu1.Name), func(t *testing.T) { - converged := utils.Retry(30, time.Second, "OpMask updated", func() bool { + wait4.AssertEventually(ctx, t, wait4.TimeoutShort, "rgw user op_mask returns to default", func(ctx context.Context) error { liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) if err != nil { - return false + return err } - return liveUser.OpMask == "read, write, delete" + if liveUser.OpMask != "read, write, delete" { + return fmt.Errorf("op_mask not yet default: %q", liveUser.OpMask) + } + return nil }) - - assert.True(t, converged) }) t.Run(fmt.Sprintf("set empty op-mask (remove all ops) on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + liveOsu, err := osuClient.Get(ctx, osu1.Name, metav1.GetOptions{}) require.NoError(t, err) liveOsu.Spec.OpMask = &[]cephv1.ObjectUserOpMask{} - _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + _, err = osuClient.Update(ctx, liveOsu, metav1.UpdateOptions{}) require.NoError(t, err) }) t.Run(fmt.Sprintf("verify op_mask set on CephObjectStoreUser %q is empty", osu1.Name), func(t *testing.T) { - converged := utils.Retry(30, time.Second, "OpMask updated", func() bool { + wait4.AssertEventually(ctx, t, wait4.TimeoutShort, "rgw user op_mask is empty", func(ctx context.Context) error { liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) if err != nil { - return false + return err } - return liveUser.OpMask == "" + if liveUser.OpMask != "" { + return fmt.Errorf("op_mask not yet empty: %q", liveUser.OpMask) + } + return nil }) - - assert.True(t, converged) }) t.Run(fmt.Sprintf("delete CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Delete(ctx, osu1.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - - absent := utils.Retry(40, time.Second, "CephObjectStoreUser is absent", func() bool { - _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) - return err != nil - }) - assert.True(t, absent) + wait4.AssertDelete(ctx, t, osuClient, osu1.Name, wait4.TimeoutShort) }) t.Run(fmt.Sprintf("no CephObjectStoreUsers in ns %q", ns.Name), func(t *testing.T) { - osus, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).List(ctx, metav1.ListOptions{}) + osus, err := osuClient.List(ctx, metav1.ListOptions{}) require.NoError(t, err) assert.Len(t, osus.Items, 0) }) - - t.Run(fmt.Sprintf("delete ns %q", ns.Name), func(t *testing.T) { - err := k8sh.Clientset.CoreV1().Namespaces().Delete(ctx, ns.Name, metav1.DeleteOptions{}) - require.NoError(t, err) - }) }) } diff --git a/tests/integration/object/util/admin/admin.go b/tests/integration/object/util/admin/admin.go deleted file mode 100644 index a3e9b2ea6..000000000 --- a/tests/integration/object/util/admin/admin.go +++ /dev/null @@ -1,67 +0,0 @@ -/* -Copyright 2025 The Rook Authors. All rights reserved. - -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 admin - -import ( - "context" - "crypto/tls" - "net/http" - - "github.com/ceph/go-ceph/rgw/admin" - "github.com/pkg/errors" - - cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/tests/framework/installer" - "github.com/rook/rook/tests/framework/utils" - utils3 "github.com/rook/rook/tests/integration/object/util/s3" -) - -func NewAdminClient(objectStore *cephv1.CephObjectStore, installer *installer.CephInstaller, k8sh *utils.K8sHelper, tlsEnable bool) (*admin.API, error) { - accessKey, secretKey, err := utils3.GetS3Credentials(objectStore, installer) - if err != nil { - return nil, errors.Wrap(err, "failed to get s3 credentials") - } - - endpoint, err := utils3.GetS3Endpoint(objectStore, k8sh, tlsEnable) - if err != nil { - return nil, errors.Wrap(err, "failed to get s3 endpoint") - } - - httpClient := &http.Client{} - - if tlsEnable { - httpClient.Transport = &http.Transport{ - TLSClientConfig: &tls.Config{ - // nolint:gosec // skip TLS verification as this is a test - InsecureSkipVerify: true, - }, - } - } - - adminClient, err := admin.New(endpoint, accessKey, secretKey, httpClient) - if err != nil { - return nil, errors.Wrap(err, "failed to setup ceph admin client") - } - - // verify that admin api is working - _, err = adminClient.GetInfo(context.Background()) - if err != nil { - return nil, errors.Wrap(err, "admin client is not working") - } - - return adminClient, nil -} diff --git a/tests/integration/object/util/s3/s3.go b/tests/integration/object/util/s3/s3.go deleted file mode 100644 index 584577aa4..000000000 --- a/tests/integration/object/util/s3/s3.go +++ /dev/null @@ -1,75 +0,0 @@ -/* -Copyright 2025 The Rook Authors. All rights reserved. - -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 s3 - -import ( - "context" - "encoding/json" - "fmt" - - "github.com/pkg/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - - cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/tests/framework/installer" - "github.com/rook/rook/tests/framework/utils" -) - -func GetS3Credentials(objectStore *cephv1.CephObjectStore, installer *installer.CephInstaller) (string, string, error) { - output, err := installer.Execute("radosgw-admin", []string{"user", "info", "--uid=dashboard-admin", fmt.Sprintf("--rgw-realm=%s", objectStore.Name)}, objectStore.Namespace) - if err != nil { - return "", "", errors.Wrap(err, "failed to get user info") - } - - // extract api creds from json output - var userInfo map[string]interface{} - err = json.Unmarshal([]byte(output), &userInfo) - if err != nil { - return "", "", errors.Wrap(err, "failed to unmarshal user info") - } - - accessKey, ok := userInfo["keys"].([]interface{})[0].(map[string]interface{})["access_key"].(string) - if !ok { - return "", "", errors.New("failed to get access key") - } - - secretKey, ok := userInfo["keys"].([]interface{})[0].(map[string]interface{})["secret_key"].(string) - if !ok { - return "", "", errors.New("failed to get secret key") - } - - return accessKey, secretKey, nil -} - -func GetS3Endpoint(objectStore *cephv1.CephObjectStore, k8sh *utils.K8sHelper, tlsEnable bool) (string, error) { - ctx := context.TODO() - - // extract rgw endpoint from k8s svc - svc, err := k8sh.Clientset.CoreV1().Services(objectStore.Namespace).Get(ctx, objectStore.Name, metav1.GetOptions{}) - if err != nil { - return "", errors.Wrap(err, "failed to get objectstore svc") - } - - schema := "http://" - if tlsEnable { - schema = "https://" - } - - endpoint := schema + svc.Spec.ClusterIP + ":80" - - return endpoint, nil -} diff --git a/tests/integration/object/util/sharedstore/sharedstore.go b/tests/integration/object/util/sharedstore/sharedstore.go index 236d03f88..e0b2d0169 100644 --- a/tests/integration/object/util/sharedstore/sharedstore.go +++ b/tests/integration/object/util/sharedstore/sharedstore.go @@ -22,59 +22,154 @@ package sharedstore import ( "context" + "fmt" "testing" "time" - cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/tests/framework/utils" + "github.com/aws/aws-sdk-go-v2/service/sns" + "github.com/ceph/go-ceph/rgw/admin" "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/intstr" + + cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" + "github.com/rook/rook/tests/framework/installer" + "github.com/rook/rook/tests/framework/utils" + "github.com/rook/rook/tests/integration/object/util/client" + "github.com/rook/rook/tests/integration/object/util/wait4" ) +type Sharedstore struct { + adminClient *admin.API + snsClient *sns.Client + objectStore *cephv1.CephObjectStore + tlsEnable bool + destroy func() +} + +func (s *Sharedstore) AdminClient() *admin.API { + return s.adminClient +} + +func (s *Sharedstore) SnsClient() *sns.Client { + return s.snsClient +} + +func (s *Sharedstore) TLSEnabled() bool { + return s.tlsEnable +} + +func (s *Sharedstore) ObjectStore() *cephv1.CephObjectStore { + return s.objectStore +} + +func (s *Sharedstore) Destroy() { + s.destroy() +} + // Setup creates a shared CephObjectStore and its NodePort Service in // ObjectStoreNamespace, waits for the store to become Ready, and returns a // teardown function that deletes both resources. The teardown function should // be deferred by the caller. -func Setup(t *testing.T, k8sh *utils.K8sHelper) (*cephv1.CephObjectStore, func()) { +func Create(t *testing.T, k8sh *utils.K8sHelper, installer *installer.CephInstaller, tlsEnable bool, allowedNamespaces ...string) *Sharedstore { t.Helper() + + s := &Sharedstore{tlsEnable: tlsEnable} ctx := context.TODO() + ns := "object-ns" + storeName := "sharedstore" + + // securePort is the in-container RGW TLS listener port. The NodePort Service + // below exposes the conventional 443 externally and forwards to it. + const securePort int32 = 8443 + certSecretName := storeName + "-tls" + rgwServiceName := "rook-ceph-rgw-" + storeName + + realm := &cephv1.CephObjectRealm{ + ObjectMeta: metav1.ObjectMeta{ + Name: storeName, + Namespace: ns, + }, + Spec: cephv1.ObjectRealmSpec{ + DefaultRealm: true, + }, + } + + zoneGroup := &cephv1.CephObjectZoneGroup{ + ObjectMeta: metav1.ObjectMeta{ + Name: storeName, + Namespace: ns, + }, + Spec: cephv1.ObjectZoneGroupSpec{ + Realm: realm.Name, + }, + } + + zone := &cephv1.CephObjectZone{ + ObjectMeta: metav1.ObjectMeta{ + Name: storeName, + Namespace: ns, + }, + Spec: cephv1.ObjectZoneSpec{ + ZoneGroup: zoneGroup.Name, + SharedPools: cephv1.ObjectSharedPoolsSpec{ + PoolPlacements: []cephv1.PoolPlacementSpec{ + { + Name: "default", + Default: true, + MetadataPoolName: storeName + ".rgw.buckets.index", + DataPoolName: storeName + ".rgw.buckets.data", + StorageClasses: []cephv1.PlacementStorageClassSpec{ + { + Name: "FOO", + DataPoolName: storeName + ".rgw.buckets.data.foo", + }, + }, + }, + }, + }, + }, + } + + poolNames := map[string]string{ + "rgw.root": ".rgw.root", + storeName + ".rgw.control": "", + storeName + ".rgw.meta": "", + storeName + ".rgw.log": "", + storeName + ".rgw.otp": "", + storeName + ".rgw.buckets.index": "", + storeName + ".rgw.buckets.data": "", + storeName + ".rgw.buckets.data.foo": "", + } objectStore := &cephv1.CephObjectStore{ ObjectMeta: metav1.ObjectMeta{ - Name: "test-shared", - Namespace: "object-ns", + Name: storeName, + Namespace: ns, }, Spec: cephv1.ObjectStoreSpec{ - MetadataPool: cephv1.PoolSpec{ - Replicated: cephv1.ReplicatedSpec{ - Size: 1, - RequireSafeReplicaSize: false, - }, - }, - DataPool: cephv1.PoolSpec{ - Replicated: cephv1.ReplicatedSpec{ - Size: 1, - RequireSafeReplicaSize: false, - }, + Zone: cephv1.ZoneSpec{ + Name: zone.Name, }, Gateway: cephv1.GatewaySpec{ Port: 80, Instances: 1, }, - // Allow CephObjectStoreUser resources from every namespace used by - // the sub-package tests under tests/integration/object. - AllowUsersInNamespaces: []string{ - "test-bucketowner", - "test-topickafka", - "test-usercaps", - "test-userkeys", - "test-useropmask", - }, + AllowUsersInNamespaces: allowedNamespaces, }, } + if tlsEnable { + // TLS-only store: omit the plain Port and serve https on securePort, + // referencing the cert secret generated below. + objectStore.Spec.Gateway = cephv1.GatewaySpec{ + SecurePort: securePort, + SSLCertificateRef: certSecretName, + Instances: 1, + } + } + svc := &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ // The service name must match ObjectStoreName so that @@ -101,30 +196,142 @@ func Setup(t *testing.T, k8sh *utils.K8sHelper) (*cephv1.CephObjectStore, func() }, } - _, err := k8sh.RookClientset.CephV1().CephObjectStores(objectStore.Namespace).Create(ctx, objectStore, metav1.CreateOptions{}) - require.NoError(t, err) - - osReady := utils.Retry(180, time.Second, "shared CephObjectStore is Ready", func() bool { - liveOs, err := k8sh.RookClientset.CephV1().CephObjectStores(objectStore.Namespace).Get(ctx, objectStore.Name, metav1.GetOptions{}) - if err != nil { - return false - } - if liveOs.Status == nil { - return false - } - return liveOs.Status.Phase == cephv1.ConditionReady - }) - require.True(t, osReady, "shared CephObjectStore did not become Ready") - - _, err = k8sh.Clientset.CoreV1().Services(objectStore.Namespace).Create(ctx, svc, metav1.CreateOptions{}) - require.NoError(t, err) - - return objectStore, func() { - if err := k8sh.Clientset.CoreV1().Services(objectStore.Namespace).Delete(ctx, objectStore.Name, metav1.DeleteOptions{}); err != nil { - t.Logf("failed to delete shared object store service: %v", err) - } - if err := k8sh.RookClientset.CephV1().CephObjectStores(objectStore.Namespace).Delete(ctx, objectStore.Name, metav1.DeleteOptions{}); err != nil { - t.Logf("failed to delete shared CephObjectStore: %v", err) + if tlsEnable { + // Expose the conventional https port externally, forwarding to the RGW + // TLS container port. + svc.Spec.Ports = []corev1.ServicePort{ + { + Name: "https", + Port: 443, + Protocol: corev1.ProtocolTCP, + TargetPort: intstr.FromInt(int(securePort)), + }, } } + + canaryUser := &cephv1.CephObjectStoreUser{ + ObjectMeta: metav1.ObjectMeta{ + Name: storeName + "-canary", + Namespace: ns, + }, + Spec: cephv1.ObjectStoreUserSpec{ + Store: objectStore.Name, + ClusterNamespace: objectStore.Namespace, + }, + } + + var pools []*cephv1.CephBlockPool + + for k, v := range poolNames { + // .rgw.root is the cluster-global realm metadata pool shared by every + // object store. The operator sets pg_num_min=8 on it when reconciling a + // normal CephObjectStore (e.g. the COSI test store), so it must have + // pg_num>=8 or that reconcile fails with "pg_num_min 8 > pg_num 1". The + // store-private pools are not shared and can stay at the minimal pg_num. + pgNum := "1" + if v == ".rgw.root" { + pgNum = "8" + } + pools = append(pools, &cephv1.CephBlockPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: k, + Namespace: ns, + }, + Spec: cephv1.NamedBlockPoolSpec{ + Name: v, + PoolSpec: cephv1.PoolSpec{ + Replicated: cephv1.ReplicatedSpec{ + Size: 1, + RequireSafeReplicaSize: false, + }, + Parameters: map[string]string{ + "pg_autoscale_mode": "off", + "pg_num": pgNum, + }, + }, + }, + }) + } + + ceph := k8sh.RookClientset.CephV1() + + { + _, err := ceph.CephObjectRealms(ns).Create(ctx, realm, metav1.CreateOptions{}) + require.NoError(t, err) + } + + { + _, err := ceph.CephObjectZoneGroups(ns).Create(ctx, zoneGroup, metav1.CreateOptions{}) + require.NoError(t, err) + } + + { + _, err := ceph.CephObjectZones(ns).Create(ctx, zone, metav1.CreateOptions{}) + require.NoError(t, err) + } + + for _, p := range pools { + _, err := ceph.CephBlockPools(ns).Create(ctx, p, metav1.CreateOptions{}) + require.NoError(t, err) + } + + // the cert secret must exist before the store is created so the operator can + // mount it when the rgw pod first reconciles + if tlsEnable { + client.GenerateRgwTLSCertSecret(t, k8sh, ns, certSecretName, rgwServiceName) + } + + wait4.RequireCreate(ctx, t, ceph.CephObjectStores(ns), objectStore, wait4.ObjectStore, + 3*time.Minute, "shared CephObjectStore did not become Ready") + + { + _, err := k8sh.Clientset.CoreV1().Services(ns).Create(ctx, svc, metav1.CreateOptions{}) + require.NoError(t, err) + } + + t.Run("setup rgw admin api client", func(t *testing.T) { + // user creation may be slow right after rgw start up + wait4.RequireCreate(ctx, t, ceph.CephObjectStoreUsers(ns), canaryUser, wait4.ObjectStoreUser, 2*time.Minute) + + // cleanup the canary user immediately so it does not block the + // CephObjectStore's deletion later + wait4.AssertDelete(ctx, t, ceph.CephObjectStoreUsers(ns), canaryUser.Name, time.Minute) + + // the canary user becoming ready tells us the rgw admin api is ready + var err error + s.adminClient, err = client.NewAdminClient(objectStore, installer, k8sh, tlsEnable) + require.NoError(t, err) + + s.snsClient, err = client.NewSNSClient(objectStore, k8sh, installer, tlsEnable) + require.NoError(t, err) + }) + + s.objectStore = objectStore + + s.destroy = func() { + t.Run(fmt.Sprintf("destroy CephObjectStore %s", objectStore.Name), func(t *testing.T) { + wait4.AssertDelete(ctx, t, k8sh.Clientset.CoreV1().Services(ns), svc.Name, 30*time.Second) + wait4.AssertDelete(ctx, t, ceph.CephObjectStores(ns), objectStore.Name, 5*time.Minute) + + // The multisite CRs must be deleted before the pools. Their deletion + // finalizers run radosgw-admin, which reads the realm/zonegroup/zone + // metadata stored in the .rgw.root pool. Deleting the pools first + // removes that metadata, causing radosgw-admin to fail (exit status 2) + // and the zone controller to loop forever, so the CephObjectZone CR + // never finishes deleting. + wait4.AssertDelete(ctx, t, ceph.CephObjectZones(ns), zone.Name, 2*time.Minute) + wait4.AssertDelete(ctx, t, ceph.CephObjectZoneGroups(ns), zoneGroup.Name, time.Minute) + wait4.AssertDelete(ctx, t, ceph.CephObjectRealms(ns), realm.Name, time.Minute) + + for _, p := range pools { + wait4.AssertDelete(ctx, t, ceph.CephBlockPools(ns), p.Name, time.Minute) + } + + if tlsEnable { + wait4.AssertDelete(ctx, t, k8sh.Clientset.CoreV1().Secrets(ns), certSecretName, 30*time.Second) + } + }) + } + + return s } diff --git a/tests/integration/object/util/sns/sns.go b/tests/integration/object/util/sns/sns.go deleted file mode 100644 index bddbdfdb5..000000000 --- a/tests/integration/object/util/sns/sns.go +++ /dev/null @@ -1,90 +0,0 @@ -/* -Copyright 2025 The Rook Authors. All rights reserved. - -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 sns - -import ( - "context" - "net/url" - - "github.com/aws/aws-sdk-go-v2/config" - "github.com/aws/aws-sdk-go-v2/credentials" - "github.com/aws/aws-sdk-go-v2/service/sns" - smithyendpoints "github.com/aws/smithy-go/endpoints" - "github.com/pkg/errors" - - cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/tests/framework/installer" - "github.com/rook/rook/tests/framework/utils" - utils3 "github.com/rook/rook/tests/integration/object/util/s3" -) - -// based on the s3 endpoint example in the official docs: -// https://docs.aws.amazon.com/sdk-for-go/v2/developer-guide/configure-endpoints.html -type snsResolverV2 struct { - endpoint string -} - -func (r *snsResolverV2) ResolveEndpoint(ctx context.Context, params sns.EndpointParameters) (smithyendpoints.Endpoint, error) { - u, err := url.Parse(r.endpoint) - if err != nil { - return smithyendpoints.Endpoint{}, err - } - return smithyendpoints.Endpoint{ - URI: *u, - }, nil -} - -func NewClient(objectStore *cephv1.CephObjectStore, k8sh *utils.K8sHelper, installer *installer.CephInstaller, tlsEnable bool) (*sns.Client, error) { - ctx := context.TODO() - - accessKey, secretKey, err := utils3.GetS3Credentials(objectStore, installer) - if err != nil { - return nil, errors.Wrap(err, "failed to get s3 credentials") - } - - cfg, err := config.LoadDefaultConfig( - ctx, - config.WithRegion("us-east-1"), - config.WithCredentialsProvider( - credentials.NewStaticCredentialsProvider( - accessKey, - secretKey, - "", - ), - ), - ) - if err != nil { - return nil, errors.Wrap(err, "failed to load default aws config") - } - - endpoint, err := utils3.GetS3Endpoint(objectStore, k8sh, tlsEnable) - if err != nil { - return nil, errors.Wrap(err, "failed to get sns endpoint") - } - - snsClient := sns.NewFromConfig(cfg, func(o *sns.Options) { - o.EndpointResolverV2 = &snsResolverV2{endpoint: endpoint} - }) - - // sanity check that client is working - _, err = snsClient.ListTopics(ctx, &sns.ListTopicsInput{}) - if err != nil { - return nil, errors.Wrap(err, "failed to sns client sanity check") - } - - return snsClient, nil -}