From c1f54b1c4a7b080dda62bc5b9197f5707b4440fb Mon Sep 17 00:00:00 2001 From: Santosh Pillai Date: Fri, 13 Jun 2025 16:09:23 +0530 Subject: [PATCH] pool: blocklist rns image watchers When force-deletion annotation is added to CephRadosNamespace CR, then blocklist any watchers on the Namespace images in order to remove the image from the trash Signed-off-by: Santosh Pillai (cherry picked from commit 98475e847ed94cda2804b20116ccf95858206c5b) --- pkg/daemon/ceph/cleanup/radosnamespace.go | 57 +++++++++++++++++++ .../ceph/cleanup/radosnamespace_test.go | 39 +++++++++++++ pkg/daemon/ceph/client/image.go | 41 +++++++++++++ pkg/daemon/ceph/client/image_test.go | 20 +++++++ pkg/daemon/ceph/client/osd.go | 10 ++++ 5 files changed, 167 insertions(+) diff --git a/pkg/daemon/ceph/cleanup/radosnamespace.go b/pkg/daemon/ceph/cleanup/radosnamespace.go index a7381b0d1..df16be48c 100644 --- a/pkg/daemon/ceph/cleanup/radosnamespace.go +++ b/pkg/daemon/ceph/cleanup/radosnamespace.go @@ -23,6 +23,13 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/client" cephclient "github.com/rook/rook/pkg/daemon/ceph/client" + "k8s.io/apimachinery/pkg/util/sets" +) + +const ( + + // ClientBlocklistDuration is the duration (in seconds) for which the client IP will be blocklisted + ClientBlocklistDuration = "1200" ) func RadosNamespaceCleanup(context *clusterd.Context, clusterInfo *client.ClusterInfo, poolName, radosNamespace string) error { @@ -48,6 +55,11 @@ func cleanupImages(context *clusterd.Context, clusterInfo *client.ClusterInfo, p return errors.Wrapf(err, "failed to list images in %s", msg) } + err = blocklistClientIPs(context, clusterInfo, images, poolName, radosNamespace) + if err != nil { + return errors.Wrap(err, "failed to add client IPs to the blocklist") + } + var retErr error for _, image := range images { snaps, err := cephclient.ListSnapshotsInRadosNamespace(context, clusterInfo, poolName, image.Name, radosNamespace) @@ -92,3 +104,48 @@ func BlockPoolCleanup(context *clusterd.Context, clusterInfo *client.ClusterInfo logger.Infof("successfully cleaned up CephBlockPool %q resource", poolName) return nil } + +func blocklistClientIPs(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, images []cephclient.CephBlockImage, poolName, radosNamespace string) error { + ips, err := getClientIPs(context, clusterInfo, images, poolName, radosNamespace) + if err != nil { + return errors.Wrapf(err, "failed to get client IPs for all the images in pool %q in namespace %q", poolName, radosNamespace) + } + + if len(ips) == 0 { + logger.Info("no client IPs found for images in pool %q in rados namespace %q", poolName, radosNamespace) + } + + for ip := range ips { + logger.Infof("blocklist client IP %q for images in pool %q in namespace %q", ip, poolName, radosNamespace) + err = cephclient.BlocklistIP(context, clusterInfo, ip, ClientBlocklistDuration) + if err != nil { + return errors.Wrapf(err, "failed to blocklist IP %q in pool %q in namespace %q", ip, poolName, radosNamespace) + } + logger.Infof("successfully blocklisted client IP %q in pool %q in namespace %q", ip, poolName, radosNamespace) + } + + return nil +} + +func getClientIPs(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, images []cephclient.CephBlockImage, poolName, radosNamespace string) (sets.Set[string], error) { + clientIPs := sets.New[string]() + for _, image := range images { + rbdStatus, err := cephclient.GetRBDImageStatus(context, clusterInfo, poolName, image.Name, radosNamespace) + if err != nil { + return clientIPs, errors.Wrapf(err, "failed to list watchers for the image %q in %s", image.Name, radosNamespace) + } + ips := rbdStatus.GetWatcherIPs() + if len(ips) == 0 { + logger.Infof("no watcher IPs found for image %q in pool %q in namespace %q", image.Name, poolName, radosNamespace) + continue + } + + logger.Infof("watcher IPs for image %q in pool %q in namespace %q: %v", image.Name, poolName, radosNamespace, ips) + for _, ip := range ips { + clientIPs.Insert(ip) + } + } + + logger.Infof("client IPs : %v", clientIPs) + return clientIPs, nil +} diff --git a/pkg/daemon/ceph/cleanup/radosnamespace_test.go b/pkg/daemon/ceph/cleanup/radosnamespace_test.go index 6010982e4..c17c177b7 100644 --- a/pkg/daemon/ceph/cleanup/radosnamespace_test.go +++ b/pkg/daemon/ceph/cleanup/radosnamespace_test.go @@ -83,6 +83,10 @@ func TestRadosNamespace(t *testing.T) { assert.Equal(t, "test-pool/test-namespace/16e35cfa56a7", args[5]) return "", nil } + if args[0] == "status" { + assert.Equal(t, "test-pool/csi-vol-136268e8-5386-4453-a6bd-9dca381d187d", args[1]) + return `{}`, nil + } return "", errors.New("unknown command") } context := &clusterd.Context{Executor: executor} @@ -136,6 +140,10 @@ func TestBlockPoolCleanup(t *testing.T) { assert.Equal(t, "test-pool/16e35cfa56a7", args[5]) return "", nil } + if args[0] == "status" { + assert.Equal(t, "test-pool/csi-vol-136268e8-5386-4453-a6bd-9dca381d187d", args[1]) + return `{}`, nil + } return "", errors.New("unknown command") } context := &clusterd.Context{Executor: executor} @@ -143,3 +151,34 @@ func TestBlockPoolCleanup(t *testing.T) { assert.NoError(t, err) }) } + +func TestGetClientIPs(t *testing.T) { + images := []cephclient.CephBlockImage{ + { + Name: "image1", + }, + { + Name: "image2", + }, + } + + // assert that only unique client IPs are returned + executor := &exectest.MockExecutor{} + executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) { + logger.Infof("Command: %s %v", command, args) + if args[0] == "status" && args[1] == "test-pool/image1" { + return `{"watchers":[{"address":"192.168.39.137:0/3762982934"}, {"address":"192.168.39.137:0/3762982934"}]}`, nil + } + if args[0] == "status" && args[1] == "test-pool/image2" { + return `{"watchers":[{"address":"192.168.39.137:0/3762982934"}, {"address":"192.168.39.137:0/3762982934"}]}`, nil + } + return "", errors.New("unknown command") + } + context := &clusterd.Context{Executor: executor} + clusterInfo := cephclient.AdminTestClusterInfo("mycluster") + ips, err := getClientIPs(context, clusterInfo, images, "test-pool", "testrns") + assert.NoError(t, err) + assert.Equal(t, 1, len(ips)) + actualIP, _ := ips.PopAny() + assert.Equal(t, "192.168.39.137", actualIP) +} diff --git a/pkg/daemon/ceph/client/image.go b/pkg/daemon/ceph/client/image.go index 2b2340dbc..8ef180b8a 100644 --- a/pkg/daemon/ceph/client/image.go +++ b/pkg/daemon/ceph/client/image.go @@ -20,6 +20,7 @@ import ( "fmt" "regexp" "strconv" + "strings" "syscall" "github.com/pkg/errors" @@ -292,3 +293,43 @@ func getImageSpecInRadosNamespace(poolName, namespace, imageID string) string { func getImageSnapshotSpec(poolName, imageName, snapshot string) string { return fmt.Sprintf("%s/%s@%s", poolName, imageName, snapshot) } + +type RBDStatus struct { + Watchers []struct { + Address string `json:"address"` + } `json:"watchers"` +} + +// GetWatcherIPs returns a list of watcher IPs of the RBD image. +// The RBD status output is parsed to get the desired IPs. +// For example: `192.168.39.137:0/3762982934“ IP format returned in the RBD status output is parsed as 192.168.39.137 +func (r RBDStatus) GetWatcherIPs() []string { + watcherIPList := []string{} + for _, watcher := range r.Watchers { + watcherIP := strings.Split(watcher.Address, ":0")[0] + watcherIPList = append(watcherIPList, watcherIP) + } + return watcherIPList +} + +// GetRBDImageStatus returns the status of the RDB image. +func GetRBDImageStatus(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, imageName, namespace string) (RBDStatus, error) { + args := []string{"status", getImageSpec(imageName, poolName)} + if namespace != "" { + args = append(args, "--namespace", namespace) + } + var rbdStatusObj RBDStatus + cmd := NewRBDCommand(context, clusterInfo, args) + cmd.JsonOutput = true + buf, err := cmd.Run() + if err != nil { + return rbdStatusObj, errors.Wrapf(err, "failed to get status of the image %q in cephblockpool %q", imageName, poolName) + } + + err = json.Unmarshal(buf, &rbdStatusObj) + if err != nil { + return rbdStatusObj, errors.Wrap(err, "failed to unmarshal rbd status output") + } + + return rbdStatusObj, nil +} diff --git a/pkg/daemon/ceph/client/image_test.go b/pkg/daemon/ceph/client/image_test.go index fd5e52aad..d26bda7f4 100644 --- a/pkg/daemon/ceph/client/image_test.go +++ b/pkg/daemon/ceph/client/image_test.go @@ -271,3 +271,23 @@ func TestListImageLogLevelDebug(t *testing.T) { assert.True(t, listCalled) listCalled = false } + +func TestGetWatcherIPs(t *testing.T) { + rbdStatus := RBDStatus{ + Watchers: []struct { + Address string "json:\"address\"" + }{ + { + Address: "192.168.39.137:0/3762982934", + }, + { + Address: "192.168.39.136:0/3762982934", + }, + }, + } + + res := rbdStatus.GetWatcherIPs() + assert.Equal(t, 2, len(res)) + assert.Equal(t, "192.168.39.137", res[0]) + assert.Equal(t, "192.168.39.136", res[1]) +} diff --git a/pkg/daemon/ceph/client/osd.go b/pkg/daemon/ceph/client/osd.go index 7f161bd53..da99f8df2 100644 --- a/pkg/daemon/ceph/client/osd.go +++ b/pkg/daemon/ceph/client/osd.go @@ -466,3 +466,13 @@ func GetOSDMetadata(context *clusterd.Context, clusterInfo *ClusterInfo) (*[]OSD } return &osdMetadata, nil } + +// BlocklistIP blocklists the IP for predefined duration +func BlocklistIP(context *clusterd.Context, clusterInfo *ClusterInfo, ip, duration string) error { + args := []string{"osd", "blocklist", "add", ip, duration} + _, err := NewCephCommand(context, clusterInfo, args).Run() + if err != nil { + return errors.Wrapf(err, "failed to blockist IP %q", ip) + } + return nil +}