diff --git a/cmd/argocd/commands/cluster.go b/cmd/argocd/commands/cluster.go
index 1c32a6eaeed6..5690307ee6b9 100644
--- a/cmd/argocd/commands/cluster.go
+++ b/cmd/argocd/commands/cluster.go
@@ -240,12 +240,13 @@ func printClusterDetails(clusters []argoappv1.Cluster) {
 	}
 }
 
-// NewClusterRemoveCommand returns a new instance of an `argocd cluster list` command
+// NewClusterRemoveCommand returns a new instance of an `argocd cluster rm` command
 func NewClusterRemoveCommand(clientOpts *argocdclient.ClientOptions) *cobra.Command {
 	var command = &cobra.Command{
-		Use:     "rm SERVER",
-		Short:   "Remove cluster credentials",
-		Example: `argocd cluster rm https://12.34.567.89`,
+		Use:   "rm SERVER/NAME",
+		Short: "Remove cluster credentials",
+		Example: `argocd cluster rm https://12.34.567.89
+argocd cluster rm cluster-name`,
 		Run: func(c *cobra.Command, args []string) {
 			if len(args) == 0 {
 				c.HelpFunc()(c, args)
@@ -261,7 +262,7 @@ func NewClusterRemoveCommand(clientOpts *argocdclient.ClientOptions) *cobra.Comm
 				// TODO(jessesuen): find the right context and remove manager RBAC artifacts
 				// err := clusterauth.UninstallClusterManagerRBAC(clientset)
 				// errors.CheckError(err)
-				_, err := clusterIf.Delete(context.Background(), &clusterpkg.ClusterQuery{Server: clusterName})
+				_, err := clusterIf.Delete(context.Background(), getQueryBySelector(clusterName))
 				errors.CheckError(err)
 				fmt.Printf("Cluster '%s' removed\n", clusterName)
 			}
diff --git a/docs/user-guide/commands/argocd_cluster_rm.md b/docs/user-guide/commands/argocd_cluster_rm.md
index 203101ca2a44..5c1028795def 100644
--- a/docs/user-guide/commands/argocd_cluster_rm.md
+++ b/docs/user-guide/commands/argocd_cluster_rm.md
@@ -3,13 +3,14 @@
 Remove cluster credentials
 
 ```
-argocd cluster rm SERVER [flags]
+argocd cluster rm SERVER/NAME [flags]
 ```
 
 ### Examples
 
 ```
 argocd cluster rm https://12.34.567.89
+argocd cluster rm cluster-name
 ```
 
 ### Options
diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go
index 4a212416db01..70aad9ed1b06 100644
--- a/server/cluster/cluster.go
+++ b/server/cluster/cluster.go
@@ -256,17 +256,40 @@ func (s *Server) Update(ctx context.Context, q *cluster.ClusterUpdateRequest) (*
 	return s.toAPIResponse(clust), nil
 }
 
-// Delete deletes a cluster by name
+// Delete deletes a cluster by server/name
 func (s *Server) Delete(ctx context.Context, q *cluster.ClusterQuery) (*cluster.ClusterResponse, error) {
 	c, err := s.getClusterWith403IfNotExist(ctx, q)
 	if err != nil {
 		return nil, err
 	}
-	if err := s.enf.EnforceErr(ctx.Value("claims"), rbacpolicy.ResourceClusters, rbacpolicy.ActionDelete, createRBACObject(c.Project, c.Server)); err != nil {
-		return nil, err
+
+	if q.Name != "" {
+		servers, err := s.db.GetClusterServersByName(ctx, q.Name)
+		if err != nil {
+			return nil, err
+		}
+		for _, server := range servers {
+			if err := enforceAndDelete(s, ctx, server, c.Project); err != nil {
+				return nil, err
+			}
+		}
+	} else {
+		if err := enforceAndDelete(s, ctx, q.Server, c.Project); err != nil {
+			return nil, err
+		}
+	}
+
+	return &cluster.ClusterResponse{}, nil
+}
+
+func enforceAndDelete(s *Server, ctx context.Context, server, project string) error {
+	if err := s.enf.EnforceErr(ctx.Value("claims"), rbacpolicy.ResourceClusters, rbacpolicy.ActionDelete, createRBACObject(project, server)); err != nil {
+		return err
+	}
+	if err := s.db.DeleteCluster(ctx, server); err != nil {
+		return err
 	}
-	err = s.db.DeleteCluster(ctx, q.Server)
-	return &cluster.ClusterResponse{}, err
+	return nil
 }
 
 // RotateAuth rotates the bearer token used for a cluster
diff --git a/server/cluster/cluster_test.go b/server/cluster/cluster_test.go
index 44d794c6b4b6..bb27eecd679d 100644
--- a/server/cluster/cluster_test.go
+++ b/server/cluster/cluster_test.go
@@ -5,7 +5,10 @@ import (
 	"testing"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	"github.com/argoproj/gitops-engine/pkg/utils/kube/kubetest"
 	"github.com/stretchr/testify/assert"
@@ -21,8 +24,10 @@ import (
 	"github.com/argoproj/argo-cd/v2/test"
 	cacheutil "github.com/argoproj/argo-cd/v2/util/cache"
 	appstatecache "github.com/argoproj/argo-cd/v2/util/cache/appstate"
+	"github.com/argoproj/argo-cd/v2/util/db"
 	dbmocks "github.com/argoproj/argo-cd/v2/util/db/mocks"
 	"github.com/argoproj/argo-cd/v2/util/rbac"
+	"github.com/argoproj/argo-cd/v2/util/settings"
 )
 
 func newServerInMemoryCache() *servercache.Cache {
@@ -158,3 +163,68 @@ func TestUpdateCluster_FieldsPathSet(t *testing.T) {
 	assert.Equal(t, updated.Namespaces, []string{"default", "kube-system"})
 	assert.Equal(t, updated.Project, "new-project")
 }
+
+func TestDeleteClusterByName(t *testing.T) {
+	testNamespace := "default"
+	clientset := getClientset(nil, testNamespace, &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "my-cluster-secret",
+			Namespace: testNamespace,
+			Labels: map[string]string{
+				common.LabelKeySecretType: common.LabelValueSecretTypeCluster,
+			},
+			Annotations: map[string]string{
+				common.AnnotationKeyManagedBy: common.AnnotationValueManagedByArgoCD,
+			},
+		},
+		Data: map[string][]byte{
+			"name":   []byte("my-cluster-name"),
+			"server": []byte("https://my-cluster-server"),
+			"config": []byte("{}"),
+		},
+	})
+	db := db.NewDB(testNamespace, settings.NewSettingsManager(context.Background(), clientset, testNamespace), clientset)
+	server := NewServer(db, newNoopEnforcer(), newServerInMemoryCache(), &kubetest.MockKubectlCmd{})
+
+	t.Run("Delete Fails When Deleting by Unknown Name", func(t *testing.T) {
+		_, err := server.Delete(context.Background(), &clusterapi.ClusterQuery{
+			Name: "foo",
+		})
+
+		assert.EqualError(t, err, `rpc error: code = PermissionDenied desc = permission denied`)
+	})
+
+	t.Run("Delete Succeeds When Deleting by Name", func(t *testing.T) {
+		_, err := server.Delete(context.Background(), &clusterapi.ClusterQuery{
+			Name: "my-cluster-name",
+		})
+		assert.Nil(t, err)
+
+		_, err = db.GetCluster(context.Background(), "https://my-cluster-server")
+		assert.EqualError(t, err, `rpc error: code = NotFound desc = cluster "https://my-cluster-server" not found`)
+	})
+}
+
+func getClientset(config map[string]string, ns string, objects ...runtime.Object) *fake.Clientset {
+	secret := corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "argocd-secret",
+			Namespace: ns,
+		},
+		Data: map[string][]byte{
+			"admin.password":   []byte("test"),
+			"server.secretkey": []byte("test"),
+		},
+	}
+	cm := corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "argocd-cm",
+			Namespace: ns,
+			Labels: map[string]string{
+				"app.kubernetes.io/part-of": "argocd",
+			},
+		},
+		Data: config,
+	}
+	return fake.NewSimpleClientset(append(objects, &cm, &secret)...)
+}
diff --git a/test/e2e/cluster_test.go b/test/e2e/cluster_test.go
index e0b77d01bbd0..eb2552eb6392 100644
--- a/test/e2e/cluster_test.go
+++ b/test/e2e/cluster_test.go
@@ -182,3 +182,48 @@ func TestClusterURLInRestAPI(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, map[string]string{"test": "val"}, cluster.Labels)
 }
+
+func TestClusterDeleteDenied(t *testing.T) {
+	EnsureCleanState(t)
+
+	accountFixture.Given(t).
+		Name("test").
+		When().
+		Create().
+		Login().
+		SetPermissions([]fixture.ACL{
+			{
+				Resource: "clusters",
+				Action:   "create",
+				Scope:    ProjectName + "/*",
+			},
+		}, "org-admin")
+
+	// Attempt to remove cluster creds by name
+	clusterFixture.
+		GivenWithSameState(t).
+		Project(ProjectName).
+		Upsert(true).
+		Server(KubernetesInternalAPIServerAddr).
+		When().
+		Create().
+		DeleteByName().
+		Then().
+		AndCLIOutput(func(output string, err error) {
+			assert.True(t, strings.Contains(err.Error(), "PermissionDenied desc = permission denied: clusters, delete"))
+		})
+
+	// Attempt to remove cluster creds by server
+	clusterFixture.
+		GivenWithSameState(t).
+		Project(ProjectName).
+		Upsert(true).
+		Server(KubernetesInternalAPIServerAddr).
+		When().
+		Create().
+		DeleteByServer().
+		Then().
+		AndCLIOutput(func(output string, err error) {
+			assert.True(t, strings.Contains(err.Error(), "PermissionDenied desc = permission denied: clusters, delete"))
+		})
+}
diff --git a/test/e2e/fixture/cluster/actions.go b/test/e2e/fixture/cluster/actions.go
index 25de5e01f8c4..c328e7028bd9 100644
--- a/test/e2e/fixture/cluster/actions.go
+++ b/test/e2e/fixture/cluster/actions.go
@@ -75,6 +75,20 @@ func (a *Actions) Get() *Actions {
 	return a
 }
 
+func (a *Actions) DeleteByName() *Actions {
+	a.context.t.Helper()
+
+	a.runCli("cluster", "rm", a.context.name)
+	return a
+}
+
+func (a *Actions) DeleteByServer() *Actions {
+	a.context.t.Helper()
+
+	a.runCli("cluster", "rm", a.context.server)
+	return a
+}
+
 func (a *Actions) Then() *Consequences {
 	a.context.t.Helper()
 	return &Consequences{a.context, a}