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 <josh@hoblitt.com>
This commit is contained in:
Joshua Hoblitt committed 2026-06-29 11:10:52 -07:00
1 parent b59101d8e8
commit eb404e6677
10 files changed
+745 -1523

No files matched your search

+8 -13
View File
@@ -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)
+105 -341
View File
@@ -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)
})
})
}
+151 -338
View File
@@ -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)
})
})
}
+33 -69
View File
@@ -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)
})
})
}
+151 -405
View File
@@ -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)
})
})
}
+41 -76
View File
@@ -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 == "<none>"
if liveUser.OpMask != "<none>" {
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)
})
})
}
@@ -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
}
-75
View File
@@ -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
}
@@ -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
}
-90
View File
@@ -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
}