Merge pull request #17101 from cobaltcore-dev/pr1-race-condition-fix

rgw: force objectStore controller to wait until zone and sharedPools are
This commit is contained in:
Travis Nielsen
2026-02-26 10:49:27 -07:00
committed by GitHub
8 changed files with 345 additions and 28 deletions
@@ -0,0 +1,41 @@
apiVersion: ceph.rook.io/v1
kind: CephObjectRealm
metadata:
name: test-realm
namespace: rook-ceph
---
apiVersion: ceph.rook.io/v1
kind: CephObjectZoneGroup
metadata:
name: test-zonegroup
namespace: rook-ceph
spec:
realm: test-realm
---
apiVersion: ceph.rook.io/v1
kind: CephObjectZone
metadata:
name: test-zone
namespace: rook-ceph
spec:
zoneGroup: test-zonegroup
metadataPool:
failureDomain: host
replicated:
size: 1
requireSafeReplicaSize: false
dataPool:
failureDomain: host
replicated:
size: 1
requireSafeReplicaSize: false
sharedPools:
poolPlacements:
- name: placement-a
metadataPoolName: rgw-meta-pool
dataPoolName: rgw-data-pool
default: true
- name: placement-b
metadataPoolName: rgw-meta-pool
dataPoolName: rgw-data-pool
preserveRadosNamespaceDataOnDelete: true
@@ -0,0 +1,11 @@
apiVersion: ceph.rook.io/v1
kind: CephObjectStore
metadata:
name: test-store
namespace: rook-ceph
spec:
gateway:
port: 80
instances: 1
zone:
name: test-zone
+47 -14
View File
@@ -375,22 +375,32 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci
return reconcile.Result{}, *cephObjectStore, errors.Wrapf(err, "failed to get latest CephObjectStore %q", request.NamespacedName.String())
}
objCtx, err := NewMultisiteContext(r.context, r.clusterInfo, cephObjectStore)
if err != nil {
if err != nil && !kerrors.IsNotFound(err) {
return reconcile.Result{}, *cephObjectStore, errors.Wrapf(err, "failed to get object context")
}
opsCtx, err := NewMultisiteAdminOpsContext(objCtx, &cephObjectStore.Spec)
if err != nil {
return reconcile.Result{}, *cephObjectStore, errors.Wrapf(err, "failed to get admin ops API context")
// Check dependents only if we have a working multisite context.
// Context may be nil when zone/zonegroup/realm CRDs are already deleted.
if objCtx != nil {
if !poolsExistForNonRawOps(objCtx) {
log.NamedInfo(request.NamespacedName, logger, "skipping bucket dependency check during deletion")
} else {
opsCtx, err := NewMultisiteAdminOpsContext(objCtx, &cephObjectStore.Spec)
if err != nil {
log.NamedWarning(request.NamespacedName, logger, "failed to get admin ops context during deletion, skipping dependency check: %v", err)
} else {
deps, err := cephObjectStoreDependents(r.context, r.clusterInfo, cephObjectStore, objCtx, opsCtx)
if err != nil {
return reconcile.Result{}, *cephObjectStore, err
}
if !deps.Empty() {
err := reporting.ReportDeletionBlockedDueToDependents(r.opManagerContext, logger, r.client, cephObjectStore, deps)
return opcontroller.WaitForRequeueIfFinalizerBlocked, *cephObjectStore, err
}
reporting.ReportDeletionNotBlockedDueToDependents(r.opManagerContext, logger, r.client, r.recorder, cephObjectStore)
}
}
}
deps, err := cephObjectStoreDependents(r.context, r.clusterInfo, cephObjectStore, objCtx, opsCtx)
if err != nil {
return reconcile.Result{}, *cephObjectStore, err
}
if !deps.Empty() {
err := reporting.ReportDeletionBlockedDueToDependents(r.opManagerContext, logger, r.client, cephObjectStore, deps)
return opcontroller.WaitForRequeueIfFinalizerBlocked, *cephObjectStore, err
}
reporting.ReportDeletionNotBlockedDueToDependents(r.opManagerContext, logger, r.client, r.recorder, cephObjectStore)
cfg := clusterConfig{
context: r.context,
@@ -479,13 +489,20 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci
}
// CREATE/UPDATE
_, err = r.reconcileCreateObjectStore(cephObjectStore, request.NamespacedName, cfg)
reconcileResult, err := r.reconcileCreateObjectStore(cephObjectStore, request.NamespacedName, cfg)
if err != nil && kerrors.IsNotFound(err) {
// A not found error may mean ceph is still initializing, but there might be some other error
// so we log the error and requeue
log.NamedWarning(request.NamespacedName, logger, "object store %q reconcile failed, ceph may still be initializing. %v", request.NamespacedName.String(), err)
return opcontroller.WaitForRequeueIfOperatorNotInitialized, *cephObjectStore, nil
} else if err != nil {
// If reconcileCreateObjectStore returned a requeue result (e.g. waiting for zone Ready),
// propagate it instead of setting Failure status. This avoids flapping between
// Progressing and Failure while waiting for dependencies.
if reconcileResult.RequeueAfter > 0 {
log.NamedWarning(request.NamespacedName, logger, "object store %q reconcile: %v. requeueing", request.NamespacedName.String(), err)
return reconcileResult, *cephObjectStore, nil
}
result, err := r.setFailedStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, "failed to create object store deployments", err)
return result, *cephObjectStore, err
}
@@ -556,6 +573,11 @@ func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *c
if err := UpdateEndpointForAdminOps(objContext, cephObjectStore); err != nil {
return r.setFailedStatus(k8sutil.ObservedGenerationNotAvailable, namespacedName, "failed to set endpoint", err)
}
// Validate that the admin ops user credentials exist
if _, _, err := GetAdminOPSUserCredentials(objContext, &cephObjectStore.Spec); err != nil {
return r.setFailedStatus(k8sutil.ObservedGenerationNotAvailable, namespacedName, "failed to get admin ops user credentials", err)
}
} else {
log.NamedInfo(namespacedName, logger, "reconciling object store deployments")
@@ -677,6 +699,17 @@ func (r *ReconcileCephObjectStore) getMultisiteResourceNames(cephObjectStore *ce
}
log.NamedDebug(nsName, logger, "CephObjectZone resource %s found", zone.Name)
// Wait for the zone to be fully configured (pools, placements, period commit)
// to avoid race condition in period commit in objectStore and Zone controllers
if zone.Status == nil || zone.Status.Phase != k8sutil.ReadyStatus {
gotPhase := ""
if zone.Status != nil {
gotPhase = zone.Status.Phase
}
logger.Infof("CephObjectZone %q is not ready yet (phase %q). waiting for it to be configured", zoneName, gotPhase)
return "", "", "", nil, waitForRequeueIfObjectStoreNotReady, fmt.Errorf("CephObjectZone %q is not ready", zoneName)
}
zonegroup := &cephv1.CephObjectZoneGroup{}
err = r.client.Get(r.opManagerContext, types.NamespacedName{Name: zone.Spec.ZoneGroup, Namespace: cephObjectStore.Namespace}, zonegroup)
if err != nil {
+192 -2
View File
@@ -606,6 +606,9 @@ func TestCephObjectStoreControllerMultisite(t *testing.T) {
MetadataPool: metadataPool,
DataPool: dataPool,
},
Status: &cephv1.Status{
Phase: k8sutil.ReadyStatus,
},
}
objectZoneGroup := &cephv1.CephObjectZoneGroup{
@@ -666,6 +669,9 @@ func TestCephObjectStoreControllerMultisite(t *testing.T) {
if args[0] == "osd" && args[1] == "pool" && args[2] == "get" {
return "", errors.New("test pool does not exit yet")
}
if args[0] == "osd" && args[1] == "lspools" {
return `[{"poolnum":1,"poolname":"zone-a.rgw.control"},{"poolnum":2,"poolname":"zone-a.rgw.meta"},{"poolnum":3,"poolname":"zone-a.rgw.log"},{"poolnum":4,"poolname":"zone-a.rgw.buckets.index"},{"poolnum":5,"poolname":"zone-a.rgw.buckets.non-ec"},{"poolnum":6,"poolname":"zone-a.rgw.otp"},{"poolnum":7,"poolname":"zone-a.rgw.buckets.data"},{"poolnum":8,"poolname":".rgw.root"}]`, nil
}
if args[0] == "versions" {
return dummyVersionsRaw, nil
}
@@ -758,6 +764,8 @@ func TestCephObjectStoreControllerMultisite(t *testing.T) {
})
t.Run("delete the same store", func(t *testing.T) {
calledCommitConfigChanges = false
// Simulate that the store was already created and has endpoints in the zone
zoneGroupGetMultisiteJSON = zoneGroupGetMultisiteJSONWithEndpoint
// no dependents
dependentsChecked := false
@@ -799,6 +807,176 @@ func TestCephObjectStoreControllerMultisite(t *testing.T) {
})
}
func TestCephObjectStoreControllerZoneNotReady(t *testing.T) {
// Verify that the object store controller requeues when the CephObjectZone
// is not yet Ready. This prevents a race condition where RGW starts before
// the zone controller has configured shared pool placements, causing RGW to
// auto-create unwanted default pools (see https://github.com/rook/rook/issues/17013).
ctx := context.TODO()
capnslog.SetGlobalLogLevel(capnslog.DEBUG)
os.Setenv("ROOK_LOG_LEVEL", "DEBUG")
zoneName := "zone-a"
zoneGroupName := "zonegroup-a"
realmName := "realm-a"
cephCluster := &cephv1.CephCluster{
ObjectMeta: metav1.ObjectMeta{
Name: namespace,
Namespace: namespace,
},
Status: cephv1.ClusterStatus{
Phase: k8sutil.ReadyStatus,
CephStatus: &cephv1.CephStatus{
Health: "HEALTH_OK",
},
},
}
secrets := map[string][]byte{
"fsid": []byte(name),
"mon-secret": []byte("monsecret"),
"admin-secret": []byte("adminsecret"),
}
secret := &v1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: "rook-ceph-mon",
Namespace: namespace,
},
Data: secrets,
Type: k8sutil.RookType,
}
objectZoneGroup := &cephv1.CephObjectZoneGroup{
ObjectMeta: metav1.ObjectMeta{
Name: zoneGroupName,
Namespace: namespace,
},
TypeMeta: metav1.TypeMeta{
Kind: "CephObjectZoneGroup",
},
Spec: cephv1.ObjectZoneGroupSpec{},
}
objectZoneGroup.Spec.Realm = realmName
objectRealm := &cephv1.CephObjectRealm{
ObjectMeta: metav1.ObjectMeta{
Name: realmName,
Namespace: namespace,
},
TypeMeta: metav1.TypeMeta{
Kind: "CephObjectRealm",
},
Spec: cephv1.ObjectRealmSpec{},
}
objectStore := &cephv1.CephObjectStore{
ObjectMeta: metav1.ObjectMeta{
Name: store,
Namespace: namespace,
Finalizers: []string{"cephobjectstore.ceph.rook.io"},
},
TypeMeta: metav1.TypeMeta{
Kind: "CephObjectStore",
},
Spec: cephv1.ObjectStoreSpec{},
}
objectStore.Spec.Zone.Name = zoneName
objectStore.Spec.Gateway.Port = 80
executor := &exectest.MockExecutor{
MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
if args[0] == "status" {
return `{"fsid":"c47cac40-9bee-4d52-823b-ccd803ba5bfe","health":{"checks":{},"status":"HEALTH_OK"},"pgmap":{"num_pgs":100,"pgs_by_state":[{"state_name":"active+clean","count":100}]}}`, nil
}
if args[0] == "versions" {
return dummyVersionsRaw, nil
}
return "", nil
},
}
clientset := test.New(t, 3)
s := scheme.Scheme
s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephObjectZone{}, &cephv1.CephObjectZoneList{}, &cephv1.CephCluster{}, &cephv1.CephClusterList{}, &cephv1.CephObjectStore{}, &cephv1.CephObjectStoreList{})
req := reconcile.Request{
NamespacedName: types.NamespacedName{
Name: store,
Namespace: namespace,
},
}
currentAndDesiredCephVersion = func(ctx context.Context, rookImage string, namespace string, jobName string, ownerInfo *k8sutil.OwnerInfo, context *clusterd.Context, cephClusterSpec *cephv1.ClusterSpec, clusterInfo *client.ClusterInfo) (*cephver.CephVersion, *cephver.CephVersion, error) {
return &cephver.Squid, &cephver.Squid, nil
}
zoneNotReadyPhases := []struct {
name string
status *cephv1.Status
}{
{"zone status is nil", nil},
{"zone is reconciling", &cephv1.Status{Phase: k8sutil.ReconcilingStatus}},
{"zone reconcile failed", &cephv1.Status{Phase: k8sutil.ReconcileFailedStatus}},
{"zone status is empty", &cephv1.Status{Phase: k8sutil.EmptyStatus}},
}
for _, tc := range zoneNotReadyPhases {
t.Run(tc.name, func(t *testing.T) {
objectZone := &cephv1.CephObjectZone{
ObjectMeta: metav1.ObjectMeta{
Name: zoneName,
Namespace: namespace,
},
TypeMeta: metav1.TypeMeta{
Kind: "CephObjectZone",
},
Spec: cephv1.ObjectZoneSpec{
ZoneGroup: zoneGroupName,
},
Status: tc.status,
}
object := []runtime.Object{
objectZone,
objectStore,
objectZoneGroup,
objectRealm,
cephCluster,
}
cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
c := &clusterd.Context{
Executor: executor,
RookClientset: rookfake.NewSimpleClientset(objectRealm, objectZoneGroup, objectZone, objectStore),
Clientset: clientset,
}
r := &ReconcileCephObjectStore{
client: cl,
scheme: s,
context: c,
recorder: events.NewFakeRecorder(50),
opManagerContext: ctx,
}
_, err := r.context.Clientset.CoreV1().Secrets(namespace).Create(ctx, secret, metav1.CreateOptions{})
if k8serrors.IsAlreadyExists(err) {
err = nil
}
assert.NoError(t, err)
res, err := r.Reconcile(ctx, req)
// Reconcile should requeue because the zone is not ready.
// The store stays in Progressing status, preventing RGW from
// starting before zone shared pools are configured.
assert.NoError(t, err)
assert.True(t, res.RequeueAfter > 0, "expected requeue while waiting for zone Ready")
})
}
}
func TestCephObjectExternalStoreController(t *testing.T) {
ctx := context.TODO()
capnslog.SetGlobalLogLevel(capnslog.DEBUG)
@@ -867,6 +1045,9 @@ func TestCephObjectExternalStoreController(t *testing.T) {
if args[0] == "osd" && args[1] == "pool" && args[2] == "get" {
return "", errors.New("test pool does not exit yet")
}
if args[0] == "osd" && args[1] == "lspools" {
return `[{"poolnum":1,"poolname":"default.rgw.control"},{"poolnum":2,"poolname":"default.rgw.meta"},{"poolnum":3,"poolname":"default.rgw.log"},{"poolnum":4,"poolname":"default.rgw.buckets.index"},{"poolnum":5,"poolname":"default.rgw.buckets.non-ec"},{"poolnum":6,"poolname":"default.rgw.otp"},{"poolnum":7,"poolname":"default.rgw.buckets.data"},{"poolnum":8,"poolname":".rgw.root"}]`, nil
}
if args[0] == "versions" {
return dummyVersionsRaw, nil
}
@@ -964,14 +1145,20 @@ func TestCephObjectExternalStoreController(t *testing.T) {
}
t.Run("create an external object store with missing secret", func(t *testing.T) {
// Clear DeletionTimestamp that was set by the delete subtest above, since
// externalObjectStore is a shared pointer.
externalObjectStore.DeletionTimestamp = nil
externalObjectStore.Finalizers = nil
objects := []runtime.Object{
cephCluster,
externalObjectStore,
}
r := getReconciler(objects)
res, err := r.Reconcile(ctx, req)
assert.Error(t, err)
assert.False(t, res.Requeue)
// Missing secret is a NotFound error, which triggers a requeue
// (the secret may be created later by another controller or manually).
assert.NoError(t, err)
assert.True(t, res.Requeue)
})
t.Run("create an external object store with no external RGW endpoints", func(t *testing.T) {
@@ -1203,6 +1390,9 @@ func TestKeyRotation(t *testing.T) {
MetadataPool: metadataPool,
DataPool: dataPool,
},
Status: &cephv1.Status{
Phase: k8sutil.ReadyStatus,
},
}
objectZoneGroup := &cephv1.CephObjectZoneGroup{
+27 -5
View File
@@ -595,12 +595,14 @@ func configureObjectStore(objContext *Context, store *cephv1.CephObjectStore, zo
}
func deleteRealm(context *Context) error {
// <name>
realmArg := fmt.Sprintf("--rgw-realm=%s", context.Name)
zoneGroupArg := fmt.Sprintf("--rgw-zonegroup=%s", context.Name)
_, err := RunAdminCommandNoMultisite(context, false, "realm", "delete", realmArg)
// Delete in reverse order: zone → zonegroup → realm.
// Note: radosgw-admin uses "delete" for zone and zonegroup, but "rm" for realm.
_, err := runAdminCommand(context, false, "zone", "delete")
if err != nil {
log.NamedWarning(context.NsName(), logger, "failed to delete rgw realm %q. %v", context.Name, err)
log.NamedWarning(context.NsName(), logger, "failed to delete rgw zone %q. %v", context.Name, err)
}
_, err = RunAdminCommandNoMultisite(context, false, "zonegroup", "delete", realmArg, zoneGroupArg)
@@ -608,9 +610,9 @@ func deleteRealm(context *Context) error {
log.NamedWarning(context.NsName(), logger, "failed to delete rgw zonegroup %q. %v", context.Name, err)
}
_, err = runAdminCommand(context, false, "zone", "delete")
_, err = RunAdminCommandNoMultisite(context, false, "realm", "rm", realmArg)
if err != nil {
log.NamedWarning(context.NsName(), logger, "failed to delete rgw zone %q. %v", context.Name, err)
log.NamedWarning(context.NsName(), logger, "failed to delete rgw realm %q. %v", context.Name, err)
}
return nil
@@ -747,6 +749,26 @@ func missingPools(context *Context) ([]string, error) {
return missingPools, nil
}
// poolsExistForNonRawOps checks whether all object store pools exist.
// Non-raw radosgw-admin commands (ones not in RGW raw_storage_ops_list, see:
// https://github.com/ceph/ceph/blob/649c3131862d310ce0b273dcbe08ebca0c2ec7ec/src/rgw/radosgw-admin/radosgw-admin.cc#L4540-L4567
// )
// trigger full RGW initialization which auto-creates default pools — for example user info called by disableRGWDashboard and
// user create called by NewMultisiteAdminOpsContext during store deletion.
// Use this function on delete reconciliation path to skip non-raw commands if pools were not created.
func poolsExistForNonRawOps(context *Context) bool {
missing, err := missingPools(context)
if err != nil {
log.NamedWarning(context.NsName(), logger, "failed to check for missing pools, assuming they don't exist to avoid ghost pool creation: %v", err)
return false
}
if len(missing) > 0 {
log.NamedInfo(context.NsName(), logger, "some object store pools are missing, skipping non-raw radosgw-admin operations to avoid ghost pool creation: %v", missing)
return false
}
return true
}
func CreateObjectStorePools(context *Context, cluster *cephv1.ClusterSpec, metadataPool, dataPool cephv1.PoolSpec) error {
if EmptyPool(dataPool) || EmptyPool(metadataPool) {
log.NamedInfo(context.NsName(), logger, "no pools specified for the CR, checking for their existence...")
+1 -1
View File
@@ -444,7 +444,7 @@ func deleteStore(t *testing.T, name string, existingStores string, expectedDelet
}
}
if args[0] == "realm" {
if args[1] == "delete" {
if args[1] == "rm" {
realmDeleted = true
return "", nil
}
+16 -6
View File
@@ -270,19 +270,29 @@ func (c *clusterConfig) deleteStore() {
}
}
// Delete the realm and pools
// Delete the realm and pools.
// NewMultisiteContext may return nil objContext when the CephObjectStore references a
// CephObjectZone (spec.zone.name is set) and the zone/zonegroup/realm CRDs have already
// been garbage-collected. This happens during simultaneous deletion of all multisite CRDs.
// Guard against nil to avoid a panic in that case — realm/pool cleanup is best-effort anyway.
objContext, err := NewMultisiteContext(c.context, c.clusterInfo, c.store)
if err != nil {
log.NamedError(nsName, logger, "failed to set multisite on object store %q. Error: %v", c.store.Name, err)
}
objContext.Endpoint = c.store.Status.Info["endpoint"]
if objContext != nil {
objContext.Endpoint = c.store.Status.Info["endpoint"]
go disableRGWDashboard(objContext)
if poolsExistForNonRawOps(objContext) {
go disableRGWDashboard(objContext)
}
err = deleteRealmAndPools(objContext, c.store.Spec)
if err != nil {
log.NamedError(nsName, logger, "failed to delete the realm and pools. Error: %v", err)
err = deleteRealmAndPools(objContext, c.store.Spec)
if err != nil {
log.NamedError(nsName, logger, "failed to delete the realm and pools. Error: %v", err)
}
} else {
log.NamedWarning(nsName, logger, "skipping realm/pool deletion: multisite context not available (zone/zonegroup/realm CRDs may already be deleted)")
}
}
@@ -206,6 +206,16 @@ func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Re
// Make sure an ObjectZoneGroup is present
realmName, reconcileResponse, err := r.getCephObjectZoneGroup(cephObjectZone)
if err != nil {
// If the zone is being deleted and the zonegroup CRD is already gone (e.g., all multisite
// CRDs deleted simultaneously), remove the finalizer so the zone CRD doesn't get stuck.
if !cephObjectZone.GetDeletionTimestamp().IsZero() && kerrors.IsNotFound(err) {
log.NamedWarning(request.NamespacedName, logger, "zonegroup CRD not found during zone deletion, removing finalizer")
err := opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephObjectZone)
if err != nil {
return reconcile.Result{}, *cephObjectZone, errors.Wrap(err, "failed to remove finalizer")
}
return reconcile.Result{}, *cephObjectZone, nil
}
return reconcileResponse, *cephObjectZone, err
}