forked from rook/rook
test: clean up object store tests
In general, make the output from the object store test easier to follow and debug. - Make sure assertions are associated with the appropriate subtest. - Remove `t.Helper()` from complex helpers. - Add info to log messages. - Move long parts of (sub-)test names to log messages. - Move some reused functions back to ceph_base_object_test.go - Refactor runObjectE2ETestLite() to be suitable for use as the second store created in the "run a second object store" test. Signed-off-by: Blaine Gardner <blaine.gardner@redhat.com>
This commit is contained in:
Blaine Gardner
@@ -17,11 +17,25 @@ limitations under the License.
|
||||
package integration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
|
||||
"github.com/rook/rook/pkg/operator/k8sutil"
|
||||
"github.com/rook/rook/tests/framework/clients"
|
||||
"github.com/rook/rook/tests/framework/installer"
|
||||
"github.com/rook/rook/tests/framework/utils"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"github.com/stretchr/testify/suite"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -45,27 +59,219 @@ var (
|
||||
)
|
||||
|
||||
// Test Object StoreCreation on Rook that was installed via helm
|
||||
func runObjectE2ETestLite(helper *clients.TestClient, k8sh *utils.K8sHelper, s suite.Suite, settings *installer.TestCephSettings, name string, replicaSize int, deleteStore bool) {
|
||||
logger.Infof("Object Storage End To End Integration Test - Create Object Store and check if rgw service is Running")
|
||||
logger.Infof("Running on Rook Cluster %s", settings.Namespace)
|
||||
func runObjectE2ETestLite(t *testing.T, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName string, replicaSize int, deleteStore bool, enableTLS bool) {
|
||||
andDeleting := ""
|
||||
if deleteStore {
|
||||
andDeleting = "and deleting"
|
||||
}
|
||||
logger.Infof("test creating %s object store %q in namespace %q", andDeleting, storeName, namespace)
|
||||
|
||||
logger.Infof("Step 1 : Create Object Store")
|
||||
err := helper.ObjectClient.Create(settings.Namespace, name, int32(replicaSize), false)
|
||||
assert.Nil(s.T(), err)
|
||||
|
||||
logger.Infof("Step 2 : check rook-ceph-rgw service status and count")
|
||||
assert.True(s.T(), k8sh.IsPodInExpectedState("rook-ceph-rgw", settings.Namespace, "Running"),
|
||||
"Make sure rook-ceph-rgw is in running state")
|
||||
|
||||
assert.True(s.T(), k8sh.CheckPodCountAndState("rook-ceph-rgw", settings.Namespace, replicaSize, "Running"),
|
||||
"Make sure all rook-ceph-rgw pods are in Running state")
|
||||
|
||||
assert.True(s.T(), k8sh.IsServiceUp("rook-ceph-rgw-"+name, settings.Namespace))
|
||||
createCephObjectStore(t, helper, k8sh, namespace, storeName, replicaSize, enableTLS)
|
||||
|
||||
if deleteStore {
|
||||
logger.Infof("Delete Object Store")
|
||||
err = helper.ObjectClient.Delete(settings.Namespace, name)
|
||||
assert.Nil(s.T(), err)
|
||||
logger.Infof("Done deleting object store")
|
||||
t.Run("delete object store", func(t *testing.T) {
|
||||
deleteObjectStore(t, k8sh, namespace, storeName)
|
||||
assertObjectStoreDeletion(t, k8sh, namespace, storeName)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// create a CephObjectStore and wait for it to report ready status
|
||||
func createCephObjectStore(t *testing.T, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName string, replicaSize int, tlsEnable bool) {
|
||||
logger.Infof("Create Object Store %q with replica count %d", storeName, replicaSize)
|
||||
rgwServiceName := "rook-ceph-rgw-" + storeName
|
||||
if tlsEnable {
|
||||
t.Run("generate TLS certs", func(t *testing.T) {
|
||||
generateRgwTlsCertSecret(t, helper, k8sh, namespace, storeName, rgwServiceName)
|
||||
})
|
||||
}
|
||||
t.Run("create CephObjectStore", func(t *testing.T) {
|
||||
err := helper.ObjectClient.Create(namespace, storeName, int32(replicaSize), tlsEnable)
|
||||
assert.Nil(t, err)
|
||||
})
|
||||
|
||||
t.Run("wait for RGWs to be running", func(t *testing.T) {
|
||||
// check that ObjectStore is created
|
||||
logger.Infof("Check that RGW pods are Running")
|
||||
for i := 0; i < 24 && k8sh.CheckPodCountAndState("rook-ceph-rgw", namespace, 1, "Running") == false; i++ {
|
||||
logger.Infof("(%d) RGW pod check sleeping for 5 seconds ...", i)
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
assert.True(t, k8sh.CheckPodCountAndState("rook-ceph-rgw", namespace, replicaSize, "Running"))
|
||||
logger.Info("RGW pods are running")
|
||||
logger.Infof("Object store %q created successfully", storeName)
|
||||
})
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
// Check object store status
|
||||
t.Run("verify object store status", func(t *testing.T) {
|
||||
retryCount := 30
|
||||
i := 0
|
||||
for i = 0; i < retryCount; i++ {
|
||||
objectStore, err := k8sh.RookClientset.CephV1().CephObjectStores(namespace).Get(ctx, storeName, metav1.GetOptions{})
|
||||
assert.Nil(t, err)
|
||||
if objectStore.Status == nil || objectStore.Status.BucketStatus == nil {
|
||||
logger.Infof("(%d) object status check sleeping for 5 seconds ...%+v", i, objectStore.Status)
|
||||
time.Sleep(5 * time.Second)
|
||||
continue
|
||||
}
|
||||
logger.Info("objectstore status is", objectStore.Status)
|
||||
if objectStore.Status.BucketStatus.Health == cephv1.ConditionFailure {
|
||||
logger.Infof("(%d) bucket status check sleeping for 5 seconds ...%+v", i, objectStore.Status.BucketStatus)
|
||||
time.Sleep(5 * time.Second)
|
||||
continue
|
||||
}
|
||||
assert.Equal(t, cephv1.ConditionConnected, objectStore.Status.BucketStatus.Health)
|
||||
// Info field has the endpoint in it
|
||||
assert.NotEmpty(t, objectStore.Status.Info)
|
||||
assert.NotEmpty(t, objectStore.Status.Info["endpoint"])
|
||||
break
|
||||
}
|
||||
if i == retryCount {
|
||||
t.Fatal("bucket status check failed. status is not connected")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("verify RGW service is up", func(t *testing.T) {
|
||||
assert.True(t, k8sh.IsServiceUp("rook-ceph-rgw-"+storeName, namespace))
|
||||
})
|
||||
}
|
||||
|
||||
func deleteObjectStore(t *testing.T, k8sh *utils.K8sHelper, namespace, storeName string) {
|
||||
err := k8sh.DeleteResourceAndWait(false, "-n", namespace, "CephObjectStore", storeName)
|
||||
assert.NoError(t, err)
|
||||
// wait initially for the controller to detect deletion. Almost always enough, but not
|
||||
// waiting immediately after this will almost always fail the first check in the loop
|
||||
time.Sleep(4 * time.Second)
|
||||
}
|
||||
|
||||
func assertObjectStoreDeletion(t *testing.T, k8sh *utils.K8sHelper, namespace, storeName string) {
|
||||
store := &cephv1.CephObjectStore{}
|
||||
i := 0
|
||||
retry := 10
|
||||
sleepTime := 3 * time.Second
|
||||
for i = 0; i < retry; i++ {
|
||||
storeStr, err := k8sh.GetResource("-n", namespace, "CephObjectStore", storeName, "-o", "json")
|
||||
assert.NoError(t, err)
|
||||
logger.Infof("store: \n%s", storeStr)
|
||||
|
||||
err = json.Unmarshal([]byte(storeStr), &store)
|
||||
assert.NoError(t, err)
|
||||
|
||||
cond := cephv1.FindStatusCondition(store.Status.Conditions, cephv1.ConditionDeletionIsBlocked)
|
||||
if cond == nil {
|
||||
logger.Infof("waiting for CephObjectStore %q to have a deletion condition", storeName)
|
||||
time.Sleep(sleepTime)
|
||||
continue
|
||||
}
|
||||
if cond.Status == v1.ConditionFalse && cond.Reason == cephv1.ObjectHasNoDependentsReason {
|
||||
// no longer blocked by dependents
|
||||
time.Sleep(5 * time.Second) // Let's give some time to the object to be updated
|
||||
break
|
||||
}
|
||||
logger.Infof("waiting 3 more seconds for CephObjectStore %q to be unblocked by dependents", storeName)
|
||||
time.Sleep(sleepTime)
|
||||
}
|
||||
assert.NotEqual(t, retry, i)
|
||||
|
||||
assert.Equal(t, cephv1.ConditionDeleting, store.Status.Phase) // phase == "Deleting"
|
||||
// verify deletion is NOT blocked b/c object has dependents
|
||||
cond := cephv1.FindStatusCondition(store.Status.Conditions, cephv1.ConditionDeletionIsBlocked)
|
||||
assert.Equal(t, v1.ConditionFalse, cond.Status)
|
||||
assert.Equal(t, cephv1.ObjectHasNoDependentsReason, cond.Reason)
|
||||
|
||||
err := k8sh.WaitUntilResourceIsDeleted("CephObjectStore", namespace, storeName)
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
|
||||
func createCephObjectUser(
|
||||
s suite.Suite, helper *clients.TestClient, k8sh *utils.K8sHelper,
|
||||
namespace, storeName, userID string,
|
||||
checkPhase, checkQuotaAndCaps bool) {
|
||||
|
||||
maxObjectInt, err := strconv.Atoi(maxObject)
|
||||
assert.Nil(s.T(), err)
|
||||
logger.Infof("creating CephObjectStore user %q for store %q in namespace %q", userID, storeName, namespace)
|
||||
cosuErr := helper.ObjectUserClient.Create(userID, userdisplayname, storeName, userCap, maxSize, maxBucket, maxObjectInt)
|
||||
assert.Nil(s.T(), cosuErr)
|
||||
logger.Infof("Waiting 5 seconds for the object user %q to be created", userID)
|
||||
time.Sleep(5 * time.Second)
|
||||
logger.Infof("Checking to see if user %q secret has been created", userID)
|
||||
for i := 0; i < 6 && helper.ObjectUserClient.UserSecretExists(namespace, storeName, userID) == false; i++ {
|
||||
logger.Infof("(%d) secret check sleeping for 5 seconds ...", i)
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
|
||||
checkCephObjectUser(s, helper, k8sh, namespace, storeName, userID, checkPhase, checkQuotaAndCaps)
|
||||
}
|
||||
|
||||
func checkCephObjectUser(
|
||||
s suite.Suite, helper *clients.TestClient, k8sh *utils.K8sHelper,
|
||||
namespace, storeName, userID string,
|
||||
checkPhase, checkQuotaAndCaps bool,
|
||||
) {
|
||||
logger.Infof("checking object store \"%s/%s\" user %q", namespace, storeName, userID)
|
||||
assert.True(s.T(), helper.ObjectUserClient.UserSecretExists(namespace, storeName, userID))
|
||||
|
||||
userInfo, err := helper.ObjectUserClient.GetUser(namespace, storeName, userID)
|
||||
assert.NoError(s.T(), err)
|
||||
assert.Equal(s.T(), userID, userInfo.UserID)
|
||||
assert.Equal(s.T(), userdisplayname, *userInfo.DisplayName)
|
||||
|
||||
if checkPhase {
|
||||
// status.phase doesn't exist before Rook v1.6
|
||||
phase, err := k8sh.GetResource("--namespace", namespace, "cephobjectstoreuser", userID, "--output", "jsonpath={.status.phase}")
|
||||
assert.NoError(s.T(), err)
|
||||
assert.Equal(s.T(), k8sutil.ReadyStatus, phase)
|
||||
}
|
||||
if checkQuotaAndCaps {
|
||||
// following fields in CephObjectStoreUser CRD doesn't exist before Rook v1.7.3
|
||||
maxObjectInt, err := strconv.Atoi(maxObject)
|
||||
assert.Nil(s.T(), err)
|
||||
maxSizeInt, err := strconv.Atoi(maxSize)
|
||||
assert.Nil(s.T(), err)
|
||||
assert.Equal(s.T(), maxBucket, userInfo.MaxBuckets)
|
||||
assert.Equal(s.T(), int64(maxObjectInt), *userInfo.UserQuota.MaxObjects)
|
||||
assert.Equal(s.T(), int64(maxSizeInt), *userInfo.UserQuota.MaxSize)
|
||||
assert.Equal(s.T(), userCap, userInfo.Caps[0].Perm)
|
||||
}
|
||||
}
|
||||
|
||||
func objectStoreCleanUp(s suite.Suite, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName string) {
|
||||
logger.Infof("Delete Object Store (will fail if users and buckets still exist)")
|
||||
err := helper.ObjectClient.Delete(namespace, storeName)
|
||||
assert.Nil(s.T(), err)
|
||||
logger.Infof("Done deleting object store")
|
||||
}
|
||||
|
||||
func generateRgwTlsCertSecret(t *testing.T, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName, rgwServiceName string) {
|
||||
ctx := context.TODO()
|
||||
root, err := utils.FindRookRoot()
|
||||
require.NoError(t, err, "failed to get rook root")
|
||||
tlscertdir, err := ioutil.TempDir(root, "tlscertdir")
|
||||
require.NoError(t, err, "failed to create directory for TLS certs")
|
||||
defer os.RemoveAll(tlscertdir)
|
||||
cmdArgs := utils.CommandArgs{Command: filepath.Join(root, "tests/scripts/generate-tls-config.sh"),
|
||||
CmdArgs: []string{tlscertdir, rgwServiceName, namespace}}
|
||||
cmdOut := utils.ExecuteCommand(cmdArgs)
|
||||
require.NoError(t, cmdOut.Err)
|
||||
tlsKeyIn, err := ioutil.ReadFile(filepath.Join(tlscertdir, rgwServiceName+".key"))
|
||||
require.NoError(t, err)
|
||||
tlsCertIn, err := ioutil.ReadFile(filepath.Join(tlscertdir, rgwServiceName+".crt"))
|
||||
require.NoError(t, err)
|
||||
tlsCaCertIn, err := ioutil.ReadFile(filepath.Join(tlscertdir, rgwServiceName+".ca"))
|
||||
require.NoError(t, err)
|
||||
secretCertOut := fmt.Sprintf("%s%s%s", tlsKeyIn, tlsCertIn, tlsCaCertIn)
|
||||
tlsK8sSecret := &v1.Secret{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: storeName,
|
||||
Namespace: namespace,
|
||||
},
|
||||
Data: map[string][]byte{
|
||||
"cert": []byte(secretCertOut),
|
||||
},
|
||||
}
|
||||
_, err = k8sh.Clientset.CoreV1().Secrets(namespace).Create(ctx, tlsK8sSecret, metav1.CreateOptions{})
|
||||
require.Nil(t, err)
|
||||
}
|
||||
|
||||
@@ -105,5 +105,7 @@ func (h *HelmSuite) TestFileStoreOnRookInstalledViaHelm() {
|
||||
|
||||
// Test Object StoreCreation on Rook that was installed via helm
|
||||
func (h *HelmSuite) TestObjectStoreOnRookInstalledViaHelm() {
|
||||
runObjectE2ETestLite(h.helper, h.k8shelper, h.Suite, h.settings, "default", 3, true)
|
||||
deleteStore := true
|
||||
tls := false
|
||||
runObjectE2ETestLite(h.T(), h.helper, h.k8shelper, h.settings.Namespace, "default", 3, deleteStore, tls)
|
||||
}
|
||||
|
||||
@@ -19,23 +19,16 @@ package integration
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
|
||||
"github.com/rook/rook/pkg/daemon/ceph/client"
|
||||
rgw "github.com/rook/rook/pkg/operator/ceph/object"
|
||||
"github.com/rook/rook/pkg/operator/k8sutil"
|
||||
"github.com/rook/rook/tests/framework/clients"
|
||||
"github.com/rook/rook/tests/framework/installer"
|
||||
"github.com/rook/rook/tests/framework/utils"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"github.com/stretchr/testify/suite"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
@@ -125,149 +118,24 @@ func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, s suite
|
||||
createCephObjectStore(s.T(), helper, k8sh, namespace, storeName, 3, tlsEnable)
|
||||
|
||||
// test that a second object store can be created (and deleted) while the first exists
|
||||
testSecondObjectStore(s.T(), helper, k8sh, namespace, storeName+"-second", tlsEnable)
|
||||
s.T().Run("run a second object store", func(t *testing.T) {
|
||||
otherStoreName := "other-" + storeName
|
||||
// The lite e2e test is perfect, as it only creates a cluster, checks that it is healthy,
|
||||
// and then deletes it.
|
||||
deleteStore := true
|
||||
runObjectE2ETestLite(t, helper, k8sh, namespace, otherStoreName, 1, deleteStore, tlsEnable)
|
||||
})
|
||||
|
||||
// now test operation of the first object store
|
||||
testObjectStoreOperations(s, helper, k8sh, namespace, storeName)
|
||||
}
|
||||
|
||||
func objectStoreCleanUp(s suite.Suite, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName string) {
|
||||
logger.Infof("Delete Object Store (will fail if users and buckets still exist)")
|
||||
err := helper.ObjectClient.Delete(namespace, storeName)
|
||||
assert.Nil(s.T(), err)
|
||||
logger.Infof("Done deleting object store")
|
||||
}
|
||||
|
||||
func createCephObjectUser(
|
||||
s suite.Suite, helper *clients.TestClient, k8sh *utils.K8sHelper,
|
||||
namespace, storeName, userID string,
|
||||
checkPhase, checkQuotaAndCaps bool) {
|
||||
s.T().Helper()
|
||||
maxObjectInt, err := strconv.Atoi(maxObject)
|
||||
assert.Nil(s.T(), err)
|
||||
cosuErr := helper.ObjectUserClient.Create(userID, userdisplayname, storeName, userCap, maxSize, maxBucket, maxObjectInt)
|
||||
assert.Nil(s.T(), cosuErr)
|
||||
logger.Infof("Waiting 5 seconds for the object user to be created")
|
||||
time.Sleep(5 * time.Second)
|
||||
logger.Infof("Checking to see if the user secret has been created")
|
||||
for i := 0; i < 6 && helper.ObjectUserClient.UserSecretExists(namespace, storeName, userID) == false; i++ {
|
||||
logger.Infof("(%d) secret check sleeping for 5 seconds ...", i)
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
|
||||
checkCephObjectUser(s, helper, k8sh, namespace, storeName, userID, checkPhase, checkQuotaAndCaps)
|
||||
}
|
||||
|
||||
func checkCephObjectUser(
|
||||
s suite.Suite, helper *clients.TestClient, k8sh *utils.K8sHelper,
|
||||
namespace, storeName, userID string,
|
||||
checkPhase, checkQuotaAndCaps bool,
|
||||
) {
|
||||
s.T().Helper()
|
||||
|
||||
logger.Infof("checking object store \"%s/%s\" user %q", namespace, storeName, userID)
|
||||
assert.True(s.T(), helper.ObjectUserClient.UserSecretExists(namespace, storeName, userID))
|
||||
|
||||
userInfo, err := helper.ObjectUserClient.GetUser(namespace, storeName, userID)
|
||||
assert.NoError(s.T(), err)
|
||||
assert.Equal(s.T(), userID, userInfo.UserID)
|
||||
assert.Equal(s.T(), userdisplayname, *userInfo.DisplayName)
|
||||
|
||||
if checkPhase {
|
||||
// status.phase doesn't exist before Rook v1.6
|
||||
phase, err := k8sh.GetResource("--namespace", namespace, "cephobjectstoreuser", userID, "--output", "jsonpath={.status.phase}")
|
||||
assert.NoError(s.T(), err)
|
||||
assert.Equal(s.T(), k8sutil.ReadyStatus, phase)
|
||||
}
|
||||
if checkQuotaAndCaps {
|
||||
// following fields in CephObjectStoreUser CRD doesn't exist before Rook v1.7.3
|
||||
maxObjectInt, err := strconv.Atoi(maxObject)
|
||||
assert.Nil(s.T(), err)
|
||||
maxSizeInt, err := strconv.Atoi(maxSize)
|
||||
assert.Nil(s.T(), err)
|
||||
assert.Equal(s.T(), maxBucket, userInfo.MaxBuckets)
|
||||
assert.Equal(s.T(), int64(maxObjectInt), *userInfo.UserQuota.MaxObjects)
|
||||
assert.Equal(s.T(), int64(maxSizeInt), *userInfo.UserQuota.MaxSize)
|
||||
assert.Equal(s.T(), userCap, userInfo.Caps[0].Perm)
|
||||
}
|
||||
}
|
||||
|
||||
// create a CephObjectStore and wait for it to report ready status
|
||||
func createCephObjectStore(t *testing.T, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName string, replicaSize int, tlsEnable bool) {
|
||||
logger.Infof("Create Object Store %q with replica count %d", storeName, replicaSize)
|
||||
rgwServiceName := "rook-ceph-rgw-" + storeName
|
||||
if tlsEnable {
|
||||
t.Run("generate TLS certs", func(t *testing.T) {
|
||||
generateRgwTlsCertSecret(t, helper, k8sh, namespace, storeName, rgwServiceName)
|
||||
})
|
||||
}
|
||||
t.Run("create CephObjectStore", func(t *testing.T) {
|
||||
err := helper.ObjectClient.Create(namespace, storeName, 3, tlsEnable)
|
||||
assert.Nil(t, err)
|
||||
})
|
||||
|
||||
t.Run("wait for RGWs to be running", func(t *testing.T) {
|
||||
// check that ObjectStore is created
|
||||
logger.Infof("Check that RGW pods are Running")
|
||||
for i := 0; i < 24 && k8sh.CheckPodCountAndState("rook-ceph-rgw", namespace, 1, "Running") == false; i++ {
|
||||
logger.Infof("(%d) RGW pod check sleeping for 5 seconds ...", i)
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
assert.True(t, k8sh.CheckPodCountAndState("rook-ceph-rgw", namespace, 1, "Running"))
|
||||
logger.Info("RGW pods are running")
|
||||
logger.Infof("Object store %q created successfully", storeName)
|
||||
})
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
// Check object store status
|
||||
t.Run("verify object store status", func(t *testing.T) {
|
||||
retryCount := 30
|
||||
i := 0
|
||||
for i = 0; i < retryCount; i++ {
|
||||
objectStore, err := k8sh.RookClientset.CephV1().CephObjectStores(namespace).Get(ctx, storeName, metav1.GetOptions{})
|
||||
assert.Nil(t, err)
|
||||
if objectStore.Status == nil || objectStore.Status.BucketStatus == nil {
|
||||
logger.Infof("(%d) object status check sleeping for 5 seconds ...%+v", i, objectStore.Status)
|
||||
time.Sleep(5 * time.Second)
|
||||
continue
|
||||
}
|
||||
logger.Info("objectstore status is", objectStore.Status)
|
||||
if objectStore.Status.BucketStatus.Health == cephv1.ConditionFailure {
|
||||
logger.Infof("(%d) bucket status check sleeping for 5 seconds ...%+v", i, objectStore.Status.BucketStatus)
|
||||
time.Sleep(5 * time.Second)
|
||||
continue
|
||||
}
|
||||
assert.Equal(t, cephv1.ConditionConnected, objectStore.Status.BucketStatus.Health)
|
||||
// Info field has the endpoint in it
|
||||
assert.NotEmpty(t, objectStore.Status.Info)
|
||||
assert.NotEmpty(t, objectStore.Status.Info["endpoint"])
|
||||
break
|
||||
}
|
||||
if i == retryCount {
|
||||
t.Fatal("bucket status check failed. status is not connected")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// test creating a second object store while an object store already exists, once the second store
|
||||
// is found to be ready, it will be deleted. "storeName" should be unique.
|
||||
func testSecondObjectStore(t *testing.T, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName string, tlsEnable bool) {
|
||||
t.Run("run a second object store", func(t *testing.T) {
|
||||
createCephObjectStore(t, helper, k8sh, namespace, storeName, 1, tlsEnable)
|
||||
|
||||
t.Run("delete the second object store", func(t *testing.T) {
|
||||
deleteObjectStore(t, k8sh, namespace, storeName)
|
||||
assertObjectStoreDeletion(t, k8sh, namespace, storeName)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName string) {
|
||||
ctx := context.TODO()
|
||||
clusterInfo := client.AdminClusterInfo(namespace)
|
||||
t := s.T()
|
||||
t.Run(fmt.Sprintf("create CephObjectStoreUser %q", storeName), func(t *testing.T) {
|
||||
|
||||
t.Run("create CephObjectStoreUser", func(t *testing.T) {
|
||||
createCephObjectUser(s, helper, k8sh, namespace, storeName, userid, true, true)
|
||||
i := 0
|
||||
for i = 0; i < 4; i++ {
|
||||
@@ -282,19 +150,20 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
|
||||
|
||||
context := k8sh.MakeContext()
|
||||
objectStore, err := k8sh.RookClientset.CephV1().CephObjectStores(namespace).Get(ctx, storeName, metav1.GetOptions{})
|
||||
assert.Nil(s.T(), err)
|
||||
assert.Nil(t, err)
|
||||
rgwcontext, err := rgw.NewMultisiteContext(context, clusterInfo, objectStore)
|
||||
assert.Nil(s.T(), err)
|
||||
t.Run("create ObjectBucketClaim with reclaim policy delete", func(t *testing.T) {
|
||||
assert.Nil(t, err)
|
||||
t.Run("create ObjectBucketClaim", func(t *testing.T) {
|
||||
logger.Infof("create OBC %q with storageclass %q - using reclaim policy 'delete' so buckets don't block deletion", obcName, bucketStorageClassName)
|
||||
cobErr := helper.BucketClient.CreateBucketStorageClass(namespace, storeName, bucketStorageClassName, "Delete", region)
|
||||
assert.Nil(s.T(), cobErr)
|
||||
assert.Nil(t, cobErr)
|
||||
cobcErr := helper.BucketClient.CreateObc(obcName, bucketStorageClassName, bucketname, maxObject, true)
|
||||
assert.Nil(s.T(), cobcErr)
|
||||
assert.Nil(t, cobcErr)
|
||||
|
||||
created := utils.Retry(12, 2*time.Second, "OBC is created", func() bool {
|
||||
return helper.BucketClient.CheckOBC(obcName, "bound")
|
||||
})
|
||||
assert.True(s.T(), created)
|
||||
assert.True(t, created)
|
||||
logger.Info("OBC created successfully")
|
||||
|
||||
var bkt rgw.ObjectBucket
|
||||
@@ -309,12 +178,12 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
|
||||
logger.Infof("(%d) check bucket exists, sleeping for 5 seconds ...", i)
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
assert.NotEqual(s.T(), 4, i)
|
||||
assert.Equal(s.T(), bucketname, bkt.Name)
|
||||
assert.NotEqual(t, 4, i)
|
||||
assert.Equal(t, bucketname, bkt.Name)
|
||||
logger.Info("OBC, Secret and ConfigMap created")
|
||||
})
|
||||
|
||||
t.Run("test S3 get/put/delete", func(t *testing.T) {
|
||||
t.Run("S3 access to OBC bucket", func(t *testing.T) {
|
||||
var s3client *rgw.S3Agent
|
||||
s3endpoint, _ := helper.ObjectClient.GetEndPointUrl(namespace, storeName)
|
||||
s3AccessKey, _ := helper.BucketClient.GetAccessKey(obcName)
|
||||
@@ -325,61 +194,61 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
|
||||
s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, region, true, nil)
|
||||
}
|
||||
|
||||
assert.Nil(s.T(), err)
|
||||
assert.Nil(t, err)
|
||||
logger.Infof("endpoint (%s) Accesskey (%s) secret (%s)", s3endpoint, s3AccessKey, s3SecretKey)
|
||||
|
||||
t.Run("put object on OBC bucket", func(t *testing.T) {
|
||||
t.Run("put object", func(t *testing.T) {
|
||||
_, poErr := s3client.PutObjectInBucket(bucketname, ObjBody, ObjectKey1, contentType)
|
||||
assert.Nil(s.T(), poErr)
|
||||
assert.Nil(t, poErr)
|
||||
})
|
||||
|
||||
t.Run("get object on OBC bucket", func(t *testing.T) {
|
||||
t.Run("get object", func(t *testing.T) {
|
||||
read, err := s3client.GetObjectInBucket(bucketname, ObjectKey1)
|
||||
assert.Nil(s.T(), err)
|
||||
assert.Equal(s.T(), ObjBody, read)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, ObjBody, read)
|
||||
})
|
||||
|
||||
t.Run("test quota enforcement on OBC bucket", func(t *testing.T) {
|
||||
t.Run("quota enforcement", func(t *testing.T) {
|
||||
_, poErr := s3client.PutObjectInBucket(bucketname, ObjBody, ObjectKey2, contentType)
|
||||
assert.Nil(s.T(), poErr)
|
||||
assert.Nil(t, poErr)
|
||||
logger.Infof("Testing the max object limit")
|
||||
_, poErr = s3client.PutObjectInBucket(bucketname, ObjBody, ObjectKey3, contentType)
|
||||
assert.Error(s.T(), poErr)
|
||||
assert.Error(t, poErr)
|
||||
})
|
||||
|
||||
t.Run("test update quota on OBC bucket", func(t *testing.T) {
|
||||
t.Run("update quota limits", func(t *testing.T) {
|
||||
poErr := helper.BucketClient.UpdateObc(obcName, bucketStorageClassName, bucketname, newMaxObject, true)
|
||||
assert.Nil(s.T(), poErr)
|
||||
assert.Nil(t, poErr)
|
||||
updated := utils.Retry(5, 2*time.Second, "OBC is updated", func() bool {
|
||||
return helper.BucketClient.CheckOBMaxObject(obcName, newMaxObject)
|
||||
})
|
||||
assert.True(s.T(), updated)
|
||||
assert.True(t, updated)
|
||||
logger.Infof("Testing the updated object limit")
|
||||
_, poErr = s3client.PutObjectInBucket(bucketname, ObjBody, ObjectKey3, contentType)
|
||||
assert.NoError(s.T(), poErr)
|
||||
assert.NoError(t, poErr)
|
||||
_, poErr = s3client.PutObjectInBucket(bucketname, ObjBody, ObjectKey4, contentType)
|
||||
assert.Error(s.T(), poErr)
|
||||
assert.Error(t, poErr)
|
||||
})
|
||||
|
||||
t.Run("delete objects on OBC bucket", func(t *testing.T) {
|
||||
t.Run("delete objects", func(t *testing.T) {
|
||||
_, delobjErr := s3client.DeleteObjectInBucket(bucketname, ObjectKey1)
|
||||
assert.Nil(s.T(), delobjErr)
|
||||
assert.Nil(t, delobjErr)
|
||||
_, delobjErr = s3client.DeleteObjectInBucket(bucketname, ObjectKey2)
|
||||
assert.Nil(s.T(), delobjErr)
|
||||
assert.Nil(t, delobjErr)
|
||||
_, delobjErr = s3client.DeleteObjectInBucket(bucketname, ObjectKey3)
|
||||
assert.Nil(s.T(), delobjErr)
|
||||
assert.Nil(t, delobjErr)
|
||||
logger.Info("Objects deleted on bucket successfully")
|
||||
})
|
||||
})
|
||||
|
||||
t.Run("Regression check: Verify bucket does not revert to Pending phase", func(t *testing.T) {
|
||||
t.Run("Regression check: OBC does not revert to Pending phase", func(t *testing.T) {
|
||||
// A bug exists in older versions of lib-bucket-provisioner that will revert a bucket and claim
|
||||
// back to "Pending" phase after being created and initially "Bound" by looping infinitely in
|
||||
// the bucket provision/creation loop. Verify that the OBC is "Bound" and stays that way.
|
||||
// The OBC reconcile loop runs again immediately b/c the OBC is modified to refer to its OB.
|
||||
// Wait a short amount of time before checking just to be safe.
|
||||
time.Sleep(15 * time.Second)
|
||||
assert.True(s.T(), helper.BucketClient.CheckOBC(obcName, "bound"))
|
||||
assert.True(t, helper.BucketClient.CheckOBC(obcName, "bound"))
|
||||
})
|
||||
|
||||
t.Run("delete CephObjectStore should be blocked by OBC bucket and CephObjectStoreUser", func(t *testing.T) {
|
||||
@@ -422,13 +291,13 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
|
||||
t.Run("delete OBC", func(t *testing.T) {
|
||||
i := 0
|
||||
dobcErr := helper.BucketClient.DeleteObc(obcName, bucketStorageClassName, bucketname, maxObject, true)
|
||||
assert.Nil(s.T(), dobcErr)
|
||||
logger.Info("Checking to see if the obc, secret and cm have all been deleted")
|
||||
assert.Nil(t, dobcErr)
|
||||
logger.Info("Checking to see if the obc, secret, and cm have all been deleted")
|
||||
for i = 0; i < 4 && !helper.BucketClient.CheckOBC(obcName, "deleted"); i++ {
|
||||
logger.Infof("(%d) obc deleted check, sleeping for 5 seconds ...", i)
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
assert.NotEqual(s.T(), 4, i)
|
||||
assert.NotEqual(t, 4, i)
|
||||
|
||||
logger.Info("ensure OBC bucket was deleted")
|
||||
var rgwErr int
|
||||
@@ -440,16 +309,16 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
|
||||
logger.Infof("(%d) check bucket deleted, sleeping for 5 seconds ...", i)
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
assert.NotEqual(s.T(), 4, i)
|
||||
assert.Equal(s.T(), rgwErr, rgw.RGWErrorNotFound)
|
||||
assert.NotEqual(t, 4, i)
|
||||
assert.Equal(t, rgwErr, rgw.RGWErrorNotFound)
|
||||
|
||||
dobErr := helper.BucketClient.DeleteBucketStorageClass(namespace, storeName, bucketStorageClassName, "Delete", region)
|
||||
assert.Nil(s.T(), dobErr)
|
||||
assert.Nil(t, dobErr)
|
||||
})
|
||||
|
||||
t.Run("delete CephObjectStoreUser", func(t *testing.T) {
|
||||
dosuErr := helper.ObjectUserClient.Delete(namespace, userid)
|
||||
assert.Nil(s.T(), dosuErr)
|
||||
assert.Nil(t, dosuErr)
|
||||
logger.Info("Object store user deleted successfully")
|
||||
logger.Info("Checking to see if the user secret has been deleted")
|
||||
i := 0
|
||||
@@ -457,11 +326,11 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
|
||||
logger.Infof("(%d) secret check sleeping for 5 seconds ...", i)
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
assert.False(s.T(), helper.ObjectUserClient.UserSecretExists(namespace, storeName, userid))
|
||||
assert.False(t, helper.ObjectUserClient.UserSecretExists(namespace, storeName, userid))
|
||||
})
|
||||
|
||||
t.Run("check that mgrs are not in a crashloop", func(t *testing.T) {
|
||||
assert.True(s.T(), k8sh.CheckPodCountAndState("rook-ceph-mgr", namespace, 1, "Running"))
|
||||
t.Run("Regression check: mgrs are not in a crashloop", func(t *testing.T) {
|
||||
assert.True(t, k8sh.CheckPodCountAndState("rook-ceph-mgr", namespace, 1, "Running"))
|
||||
})
|
||||
|
||||
t.Run("CephObjectStore should delete now that dependents are gone", func(t *testing.T) {
|
||||
@@ -473,81 +342,3 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
|
||||
|
||||
// TODO : Add case for brownfield/cleanup s3 client}
|
||||
}
|
||||
|
||||
func deleteObjectStore(t *testing.T, k8sh *utils.K8sHelper, namespace, storeName string) {
|
||||
err := k8sh.DeleteResourceAndWait(false, "-n", namespace, "CephObjectStore", storeName)
|
||||
assert.NoError(t, err)
|
||||
// wait initially for the controller to detect deletion. Almost always enough, but not
|
||||
// waiting immediately after this will almost always fail the first check in the loop
|
||||
time.Sleep(4 * time.Second)
|
||||
}
|
||||
|
||||
func assertObjectStoreDeletion(t *testing.T, k8sh *utils.K8sHelper, namespace, storeName string) {
|
||||
store := &cephv1.CephObjectStore{}
|
||||
i := 0
|
||||
retry := 10
|
||||
sleepTime := 3 * time.Second
|
||||
for i = 0; i < retry; i++ {
|
||||
storeStr, err := k8sh.GetResource("-n", namespace, "CephObjectStore", storeName, "-o", "json")
|
||||
assert.NoError(t, err)
|
||||
logger.Infof("store: \n%s", storeStr)
|
||||
|
||||
err = json.Unmarshal([]byte(storeStr), &store)
|
||||
assert.NoError(t, err)
|
||||
|
||||
cond := cephv1.FindStatusCondition(store.Status.Conditions, cephv1.ConditionDeletionIsBlocked)
|
||||
if cond == nil {
|
||||
logger.Infof("waiting for CephObjectStore %q to have a deletion condition", storeName)
|
||||
time.Sleep(sleepTime)
|
||||
continue
|
||||
}
|
||||
if cond.Status == v1.ConditionFalse && cond.Reason == cephv1.ObjectHasNoDependentsReason {
|
||||
// no longer blocked by dependents
|
||||
time.Sleep(5 * time.Second) // Let's give some time to the object to be updated
|
||||
break
|
||||
}
|
||||
logger.Infof("waiting 3 more seconds for CephObjectStore %q to be unblocked by dependents", storeName)
|
||||
time.Sleep(sleepTime)
|
||||
}
|
||||
assert.NotEqual(t, retry, i)
|
||||
|
||||
assert.Equal(t, cephv1.ConditionDeleting, store.Status.Phase) // phase == "Deleting"
|
||||
// verify deletion is NOT blocked b/c object has dependents
|
||||
cond := cephv1.FindStatusCondition(store.Status.Conditions, cephv1.ConditionDeletionIsBlocked)
|
||||
assert.Equal(t, v1.ConditionFalse, cond.Status)
|
||||
assert.Equal(t, cephv1.ObjectHasNoDependentsReason, cond.Reason)
|
||||
|
||||
err := k8sh.WaitUntilResourceIsDeleted("CephObjectStore", namespace, storeName)
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
|
||||
func generateRgwTlsCertSecret(t *testing.T, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName, rgwServiceName string) {
|
||||
ctx := context.TODO()
|
||||
root, err := utils.FindRookRoot()
|
||||
require.NoError(t, err, "failed to get rook root")
|
||||
tlscertdir, err := ioutil.TempDir(root, "tlscertdir")
|
||||
require.NoError(t, err, "failed to create directory for TLS certs")
|
||||
defer os.RemoveAll(tlscertdir)
|
||||
cmdArgs := utils.CommandArgs{Command: filepath.Join(root, "tests/scripts/generate-tls-config.sh"),
|
||||
CmdArgs: []string{tlscertdir, rgwServiceName, namespace}}
|
||||
cmdOut := utils.ExecuteCommand(cmdArgs)
|
||||
require.NoError(t, cmdOut.Err)
|
||||
tlsKeyIn, err := ioutil.ReadFile(filepath.Join(tlscertdir, rgwServiceName+".key"))
|
||||
require.NoError(t, err)
|
||||
tlsCertIn, err := ioutil.ReadFile(filepath.Join(tlscertdir, rgwServiceName+".crt"))
|
||||
require.NoError(t, err)
|
||||
tlsCaCertIn, err := ioutil.ReadFile(filepath.Join(tlscertdir, rgwServiceName+".ca"))
|
||||
require.NoError(t, err)
|
||||
secretCertOut := fmt.Sprintf("%s%s%s", tlsKeyIn, tlsCertIn, tlsCaCertIn)
|
||||
tlsK8sSecret := &v1.Secret{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: storeName,
|
||||
Namespace: namespace,
|
||||
},
|
||||
Data: map[string][]byte{
|
||||
"cert": []byte(secretCertOut),
|
||||
},
|
||||
}
|
||||
_, err = k8sh.Clientset.CoreV1().Secrets(namespace).Create(ctx, tlsK8sSecret, metav1.CreateOptions{})
|
||||
require.Nil(t, err)
|
||||
}
|
||||
|
||||
@@ -126,7 +126,9 @@ func (s *SmokeSuite) TestObjectStorage_SmokeTest() {
|
||||
s.T().Skip("object store tests skipped on openshift")
|
||||
}
|
||||
storeName := "lite-store"
|
||||
runObjectE2ETestLite(s.helper, s.k8sh, s.Suite, s.settings, storeName, 2, true)
|
||||
deleteStore := true
|
||||
tls := false
|
||||
runObjectE2ETestLite(s.T(), s.helper, s.k8sh, s.settings.Namespace, storeName, 2, deleteStore, tls)
|
||||
}
|
||||
|
||||
// Test to make sure all rook components are installed and Running
|
||||
|
||||
@@ -255,7 +255,9 @@ func (s *UpgradeSuite) deployClusterforUpgrade(objectStoreName, objectUserID, me
|
||||
createFilesystemConsumerPod(s.helper, s.k8sh, s.Suite, s.settings, filesystemName, fsStorageClass)
|
||||
|
||||
logger.Infof("Initializing object before the upgrade")
|
||||
runObjectE2ETestLite(s.helper, s.k8sh, s.Suite, s.settings, objectStoreName, 1, false)
|
||||
deleteStore := false
|
||||
tls := false
|
||||
runObjectE2ETestLite(s.T(), s.helper, s.k8sh, s.settings.Namespace, objectStoreName, 1, deleteStore, tls)
|
||||
|
||||
logger.Infof("Initializing object user before the upgrade")
|
||||
createCephObjectUser(s.Suite, s.helper, s.k8sh, s.namespace, objectStoreName, objectUserID, false, false)
|
||||
|
||||
Reference in New Issue
Block a user