diff --git a/doc/source/cli/command-objects/block-storage-cluster.rst b/doc/source/cli/command-objects/block-storage-cluster.rst
new file mode 100644
index 0000000000..318419e079
--- /dev/null
+++ b/doc/source/cli/command-objects/block-storage-cluster.rst
@@ -0,0 +1,8 @@
+=====================
+block storage cluster
+=====================
+
+Block Storage v3
+
+.. autoprogram-cliff:: openstack.volume.v3
+   :command: block storage cluster *
diff --git a/doc/source/cli/commands.rst b/doc/source/cli/commands.rst
index 7e59215271..d9bedb36d4 100644
--- a/doc/source/cli/commands.rst
+++ b/doc/source/cli/commands.rst
@@ -76,6 +76,7 @@ referring to both Compute and Volume quotas.
 * ``address scope``: (**Network**) a scope of IPv4 or IPv6 addresses
 * ``aggregate``: (**Compute**) a grouping of compute hosts
 * ``availability zone``: (**Compute**, **Network**, **Volume**) a logical partition of hosts or block storage or network services
+* ``block storage cluster``: (**Volume**) clusters of volume services
 * ``catalog``: (**Identity**) service catalog
 * ``command``: (**Internal**) installed commands in the OSC process
 * ``compute agent``: (**Compute**) a cloud Compute agent available to a hypervisor
diff --git a/doc/source/cli/data/cinder.csv b/doc/source/cli/data/cinder.csv
index 9d79d1ba97..54ce79676d 100644
--- a/doc/source/cli/data/cinder.csv
+++ b/doc/source/cli/data/cinder.csv
@@ -20,10 +20,10 @@ cgsnapshot-create,consistency group snapshot create,Creates a cgsnapshot.
 cgsnapshot-delete,consistency group snapshot delete,Removes one or more cgsnapshots.
 cgsnapshot-list,consistency group snapshot list,Lists all cgsnapshots.
 cgsnapshot-show,consistency group snapshot show,Shows cgsnapshot details.
-cluster-disable,,Disables clustered services. (Supported by API versions 3.7 - 3.latest)
-cluster-enable,,Enables clustered services. (Supported by API versions 3.7 - 3.latest)
-cluster-list,,Lists clustered services with optional filtering. (Supported by API versions 3.7 - 3.latest)
-cluster-show,,Show detailed information on a clustered service. (Supported by API versions 3.7 - 3.latest)
+cluster-disable,block storage cluster set --disable,Disables clustered services. (Supported by API versions 3.7 - 3.latest)
+cluster-enable,block storage cluster set --enable,Enables clustered services. (Supported by API versions 3.7 - 3.latest)
+cluster-list,block storage cluster list,Lists clustered services with optional filtering. (Supported by API versions 3.7 - 3.latest)
+cluster-show,block storage cluster show,Show detailed information on a clustered service. (Supported by API versions 3.7 - 3.latest)
 consisgroup-create,consistency group create,Creates a consistency group.
 consisgroup-create-from-src,consistency group create --consistency-group-snapshot,Creates a consistency group from a cgsnapshot or a source CG
 consisgroup-delete,consistency group delete,Removes one or more consistency groups.
diff --git a/openstackclient/tests/unit/volume/v3/fakes.py b/openstackclient/tests/unit/volume/v3/fakes.py
index 9040b2be08..81ff0a9885 100644
--- a/openstackclient/tests/unit/volume/v3/fakes.py
+++ b/openstackclient/tests/unit/volume/v3/fakes.py
@@ -32,6 +32,8 @@ class FakeVolumeClient(object):
 
         self.attachments = mock.Mock()
         self.attachments.resource_class = fakes.FakeResource(None, {})
+        self.clusters = mock.Mock()
+        self.clusters.resource_class = fakes.FakeResource(None, {})
         self.groups = mock.Mock()
         self.groups.resource_class = fakes.FakeResource(None, {})
         self.group_snapshots = mock.Mock()
@@ -70,6 +72,58 @@ FakeVolume = volume_v2_fakes.FakeVolume
 FakeVolumeType = volume_v2_fakes.FakeVolumeType
 
 
+class FakeCluster:
+    """Fake one or more clusters."""
+
+    @staticmethod
+    def create_one_cluster(attrs=None):
+        """Create a fake service cluster.
+
+        :param attrs: A dictionary with all attributes of service cluster
+        :return: A FakeResource object with id, name, status, etc.
+        """
+        attrs = attrs or {}
+
+        # Set default attribute
+        cluster_info = {
+            'name': f'cluster-{uuid.uuid4().hex}',
+            'binary': f'binary-{uuid.uuid4().hex}',
+            'state': random.choice(['up', 'down']),
+            'status': random.choice(['enabled', 'disabled']),
+            'disabled_reason': None,
+            'num_hosts': random.randint(1, 64),
+            'num_down_hosts': random.randint(1, 64),
+            'last_heartbeat': '2015-09-16T09:28:52.000000',
+            'created_at': '2015-09-16T09:28:52.000000',
+            'updated_at': '2015-09-16T09:28:52.000000',
+            'replication_status': None,
+            'frozen': False,
+            'active_backend_id': None,
+        }
+
+        # Overwrite default attributes if there are some attributes set
+        cluster_info.update(attrs)
+
+        return fakes.FakeResource(
+            None,
+            cluster_info,
+            loaded=True)
+
+    @staticmethod
+    def create_clusters(attrs=None, count=2):
+        """Create multiple fake service clusters.
+
+        :param attrs: A dictionary with all attributes of service cluster
+        :param count: The number of service clusters to be faked
+        :return: A list of FakeResource objects
+        """
+        clusters = []
+        for n in range(0, count):
+            clusters.append(FakeCluster.create_one_cluster(attrs))
+
+        return clusters
+
+
 class FakeVolumeGroup:
     """Fake one or more volume groups."""
 
diff --git a/openstackclient/tests/unit/volume/v3/test_block_storage_cluster.py b/openstackclient/tests/unit/volume/v3/test_block_storage_cluster.py
new file mode 100644
index 0000000000..d87a946b9f
--- /dev/null
+++ b/openstackclient/tests/unit/volume/v3/test_block_storage_cluster.py
@@ -0,0 +1,434 @@
+# 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.
+
+from cinderclient import api_versions
+from osc_lib import exceptions
+
+from openstackclient.tests.unit.volume.v3 import fakes as volume_fakes
+from openstackclient.volume.v3 import block_storage_cluster
+
+
+class TestBlockStorageCluster(volume_fakes.TestVolume):
+
+    def setUp(self):
+        super().setUp()
+
+        # Get a shortcut to the BlockStorageClusterManager Mock
+        self.cluster_mock = self.app.client_manager.volume.clusters
+        self.cluster_mock.reset_mock()
+
+
+class TestBlockStorageClusterList(TestBlockStorageCluster):
+
+    # The cluster to be listed
+    fake_clusters = volume_fakes.FakeCluster.create_clusters()
+
+    def setUp(self):
+        super().setUp()
+
+        self.cluster_mock.list.return_value = self.fake_clusters
+
+        # Get the command object to test
+        self.cmd = \
+            block_storage_cluster.ListBlockStorageCluster(self.app, None)
+
+    def test_cluster_list(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.7')
+
+        arglist = [
+        ]
+        verifylist = [
+            ('cluster', None),
+            ('binary', None),
+            ('is_up', None),
+            ('is_disabled', None),
+            ('num_hosts', None),
+            ('num_down_hosts', None),
+            ('long', False),
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        expected_columns = ('Name', 'Binary', 'State', 'Status')
+        expected_data = tuple(
+            (
+                cluster.name,
+                cluster.binary,
+                cluster.state,
+                cluster.status,
+            ) for cluster in self.fake_clusters
+        )
+        columns, data = self.cmd.take_action(parsed_args)
+
+        self.assertEqual(expected_columns, columns)
+        self.assertEqual(expected_data, tuple(data))
+
+        # checking if proper call was made to list clusters
+        self.cluster_mock.list.assert_called_with(
+            name=None,
+            binary=None,
+            is_up=None,
+            disabled=None,
+            num_hosts=None,
+            num_down_hosts=None,
+            detailed=False,
+        )
+
+    def test_cluster_list_with_full_options(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.7')
+
+        arglist = [
+            '--cluster', 'foo',
+            '--binary', 'bar',
+            '--up',
+            '--disabled',
+            '--num-hosts', '5',
+            '--num-down-hosts', '0',
+            '--long',
+        ]
+        verifylist = [
+            ('cluster', 'foo'),
+            ('binary', 'bar'),
+            ('is_up', True),
+            ('is_disabled', True),
+            ('num_hosts', 5),
+            ('num_down_hosts', 0),
+            ('long', True),
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        expected_columns = (
+            'Name',
+            'Binary',
+            'State',
+            'Status',
+            'Num Hosts',
+            'Num Down Hosts',
+            'Last Heartbeat',
+            'Disabled Reason',
+            'Created At',
+            'Updated At',
+        )
+        expected_data = tuple(
+            (
+                cluster.name,
+                cluster.binary,
+                cluster.state,
+                cluster.status,
+                cluster.num_hosts,
+                cluster.num_down_hosts,
+                cluster.last_heartbeat,
+                cluster.disabled_reason,
+                cluster.created_at,
+                cluster.updated_at,
+            ) for cluster in self.fake_clusters
+        )
+        columns, data = self.cmd.take_action(parsed_args)
+
+        self.assertEqual(expected_columns, columns)
+        self.assertEqual(expected_data, tuple(data))
+
+        # checking if proper call was made to list clusters
+        self.cluster_mock.list.assert_called_with(
+            name='foo',
+            binary='bar',
+            is_up=True,
+            disabled=True,
+            num_hosts=5,
+            num_down_hosts=0,
+            detailed=True,
+        )
+
+    def test_cluster_list_pre_v37(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.6')
+
+        arglist = [
+        ]
+        verifylist = [
+            ('cluster', None),
+            ('binary', None),
+            ('is_up', None),
+            ('is_disabled', None),
+            ('num_hosts', None),
+            ('num_down_hosts', None),
+            ('long', False),
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        exc = self.assertRaises(
+            exceptions.CommandError,
+            self.cmd.take_action,
+            parsed_args)
+        self.assertIn(
+            '--os-volume-api-version 3.7 or greater is required', str(exc))
+
+
+class TestBlockStorageClusterSet(TestBlockStorageCluster):
+
+    cluster = volume_fakes.FakeCluster.create_one_cluster()
+    columns = (
+        'Name',
+        'Binary',
+        'State',
+        'Status',
+        'Disabled Reason',
+        'Hosts',
+        'Down Hosts',
+        'Last Heartbeat',
+        'Created At',
+        'Updated At',
+        'Replication Status',
+        'Frozen',
+        'Active Backend ID',
+    )
+    data = (
+        cluster.name,
+        cluster.binary,
+        cluster.state,
+        cluster.status,
+        cluster.disabled_reason,
+        cluster.num_hosts,
+        cluster.num_down_hosts,
+        cluster.last_heartbeat,
+        cluster.created_at,
+        cluster.updated_at,
+        cluster.replication_status,
+        cluster.frozen,
+        cluster.active_backend_id,
+    )
+
+    def setUp(self):
+        super().setUp()
+
+        self.cluster_mock.update.return_value = self.cluster
+
+        self.cmd = \
+            block_storage_cluster.SetBlockStorageCluster(self.app, None)
+
+    def test_cluster_set(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.7')
+
+        arglist = [
+            '--enable',
+            self.cluster.name,
+        ]
+        verifylist = [
+            ('cluster', self.cluster.name),
+            ('binary', 'cinder-volume'),
+            ('disabled', False),
+            ('disabled_reason', None),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        columns, data = self.cmd.take_action(parsed_args)
+        self.assertEqual(self.columns, columns)
+        self.assertEqual(self.data, tuple(data))
+
+        self.cluster_mock.update.assert_called_once_with(
+            self.cluster.name,
+            'cinder-volume',
+            disabled=False,
+            disabled_reason=None,
+        )
+
+    def test_cluster_set_disable_with_reason(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.7')
+
+        arglist = [
+            '--binary', self.cluster.binary,
+            '--disable',
+            '--disable-reason', 'foo',
+            self.cluster.name,
+        ]
+        verifylist = [
+            ('cluster', self.cluster.name),
+            ('binary', self.cluster.binary),
+            ('disabled', True),
+            ('disabled_reason', 'foo'),
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        columns, data = self.cmd.take_action(parsed_args)
+        self.assertEqual(self.columns, columns)
+        self.assertEqual(self.data, tuple(data))
+
+        self.cluster_mock.update.assert_called_once_with(
+            self.cluster.name,
+            self.cluster.binary,
+            disabled=True,
+            disabled_reason='foo',
+        )
+
+    def test_cluster_set_only_with_disable_reason(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.7')
+
+        arglist = [
+            '--disable-reason', 'foo',
+            self.cluster.name,
+        ]
+        verifylist = [
+            ('cluster', self.cluster.name),
+            ('binary', 'cinder-volume'),
+            ('disabled', None),
+            ('disabled_reason', 'foo'),
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        exc = self.assertRaises(
+            exceptions.CommandError,
+            self.cmd.take_action,
+            parsed_args)
+        self.assertIn(
+            "Cannot specify --disable-reason without --disable", str(exc))
+
+    def test_cluster_set_enable_with_disable_reason(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.7')
+
+        arglist = [
+            '--enable',
+            '--disable-reason', 'foo',
+            self.cluster.name,
+        ]
+        verifylist = [
+            ('cluster', self.cluster.name),
+            ('binary', 'cinder-volume'),
+            ('disabled', False),
+            ('disabled_reason', 'foo'),
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        exc = self.assertRaises(
+            exceptions.CommandError,
+            self.cmd.take_action,
+            parsed_args)
+        self.assertIn(
+            "Cannot specify --disable-reason without --disable", str(exc))
+
+    def test_cluster_set_pre_v37(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.6')
+
+        arglist = [
+            '--enable',
+            self.cluster.name,
+        ]
+        verifylist = [
+            ('cluster', self.cluster.name),
+            ('binary', 'cinder-volume'),
+            ('disabled', False),
+            ('disabled_reason', None),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        exc = self.assertRaises(
+            exceptions.CommandError,
+            self.cmd.take_action,
+            parsed_args)
+        self.assertIn(
+            '--os-volume-api-version 3.7 or greater is required', str(exc))
+
+
+class TestBlockStorageClusterShow(TestBlockStorageCluster):
+
+    cluster = volume_fakes.FakeCluster.create_one_cluster()
+    columns = (
+        'Name',
+        'Binary',
+        'State',
+        'Status',
+        'Disabled Reason',
+        'Hosts',
+        'Down Hosts',
+        'Last Heartbeat',
+        'Created At',
+        'Updated At',
+        'Replication Status',
+        'Frozen',
+        'Active Backend ID',
+    )
+    data = (
+        cluster.name,
+        cluster.binary,
+        cluster.state,
+        cluster.status,
+        cluster.disabled_reason,
+        cluster.num_hosts,
+        cluster.num_down_hosts,
+        cluster.last_heartbeat,
+        cluster.created_at,
+        cluster.updated_at,
+        cluster.replication_status,
+        cluster.frozen,
+        cluster.active_backend_id,
+    )
+
+    def setUp(self):
+        super().setUp()
+
+        self.cluster_mock.show.return_value = self.cluster
+
+        self.cmd = \
+            block_storage_cluster.ShowBlockStorageCluster(self.app, None)
+
+    def test_cluster_show(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.7')
+
+        arglist = [
+            '--binary', self.cluster.binary,
+            self.cluster.name,
+        ]
+        verifylist = [
+            ('cluster', self.cluster.name),
+            ('binary', self.cluster.binary),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        columns, data = self.cmd.take_action(parsed_args)
+        self.assertEqual(self.columns, columns)
+        self.assertEqual(self.data, tuple(data))
+
+        self.cluster_mock.show.assert_called_once_with(
+            self.cluster.name,
+            binary=self.cluster.binary,
+        )
+
+    def test_cluster_show_pre_v37(self):
+        self.app.client_manager.volume.api_version = \
+            api_versions.APIVersion('3.6')
+
+        arglist = [
+            '--binary', self.cluster.binary,
+            self.cluster.name,
+        ]
+        verifylist = [
+            ('cluster', self.cluster.name),
+            ('binary', self.cluster.binary),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        exc = self.assertRaises(
+            exceptions.CommandError,
+            self.cmd.take_action,
+            parsed_args)
+        self.assertIn(
+            '--os-volume-api-version 3.7 or greater is required', str(exc))
diff --git a/openstackclient/volume/v3/block_storage_cluster.py b/openstackclient/volume/v3/block_storage_cluster.py
new file mode 100644
index 0000000000..34b25efcfd
--- /dev/null
+++ b/openstackclient/volume/v3/block_storage_cluster.py
@@ -0,0 +1,281 @@
+# 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.
+
+from cinderclient import api_versions
+from osc_lib.command import command
+from osc_lib import exceptions
+from osc_lib import utils
+
+from openstackclient.i18n import _
+
+
+def _format_cluster(cluster, detailed=False):
+    columns = (
+        'name',
+        'binary',
+        'state',
+        'status',
+    )
+    column_headers = (
+        'Name',
+        'Binary',
+        'State',
+        'Status',
+    )
+
+    if detailed:
+        columns += (
+            'disabled_reason',
+            'num_hosts',
+            'num_down_hosts',
+            'last_heartbeat',
+            'created_at',
+            'updated_at',
+            # optional columns, depending on whether replication is enabled
+            'replication_status',
+            'frozen',
+            'active_backend_id',
+        )
+        column_headers += (
+            'Disabled Reason',
+            'Hosts',
+            'Down Hosts',
+            'Last Heartbeat',
+            'Created At',
+            'Updated At',
+            # optional columns, depending on whether replication is enabled
+            'Replication Status',
+            'Frozen',
+            'Active Backend ID',
+        )
+
+    return (
+        column_headers,
+        utils.get_item_properties(
+            cluster,
+            columns,
+        ),
+    )
+
+
+class ListBlockStorageCluster(command.Lister):
+    """List block storage clusters.
+
+    This command requires ``--os-volume-api-version`` 3.7 or greater.
+    """
+
+    def get_parser(self, prog_name):
+        parser = super().get_parser(prog_name)
+        parser.add_argument(
+            '--cluster', metavar='<name>', default=None,
+            help=_(
+                'Filter by cluster name, without backend will list '
+                'all clustered services from the same cluster.'
+            ),
+        )
+        parser.add_argument(
+            '--binary',
+            metavar='<binary>',
+            help=_('Cluster binary.'),
+        )
+        parser.add_argument(
+            '--up',
+            action='store_true',
+            dest='is_up',
+            default=None,
+            help=_('Filter by up status.'),
+        )
+        parser.add_argument(
+            '--down',
+            action='store_false',
+            dest='is_up',
+            help=_('Filter by down status.'),
+        )
+        parser.add_argument(
+            '--disabled',
+            action='store_true',
+            dest='is_disabled',
+            default=None,
+            help=_('Filter by disabled status.'),
+        )
+        parser.add_argument(
+            '--enabled',
+            action='store_false',
+            dest='is_disabled',
+            help=_('Filter by enabled status.'),
+        )
+        parser.add_argument(
+            '--num-hosts',
+            metavar='<hosts>',
+            type=int,
+            default=None,
+            help=_('Filter by number of hosts in the cluster.'),
+        )
+        parser.add_argument(
+            '--num-down-hosts',
+            metavar='<hosts>',
+            type=int,
+            default=None,
+            help=_('Filter by number of hosts that are down.'),
+        )
+        parser.add_argument(
+            '--long',
+            action='store_true',
+            default=False,
+            help=_("List additional fields in output")
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        volume_client = self.app.client_manager.volume
+
+        if volume_client.api_version < api_versions.APIVersion('3.7'):
+            msg = _(
+                "--os-volume-api-version 3.7 or greater is required to "
+                "support the 'block storage cluster list' command"
+            )
+            raise exceptions.CommandError(msg)
+
+        columns = ('Name', 'Binary', 'State', 'Status')
+        if parsed_args.long:
+            columns += (
+                'Num Hosts',
+                'Num Down Hosts',
+                'Last Heartbeat',
+                'Disabled Reason',
+                'Created At',
+                'Updated At',
+            )
+
+        data = volume_client.clusters.list(
+            name=parsed_args.cluster,
+            binary=parsed_args.binary,
+            is_up=parsed_args.is_up,
+            disabled=parsed_args.is_disabled,
+            num_hosts=parsed_args.num_hosts,
+            num_down_hosts=parsed_args.num_down_hosts,
+            detailed=parsed_args.long,
+        )
+
+        return (
+            columns,
+            (utils.get_item_properties(s, columns) for s in data),
+        )
+
+
+class SetBlockStorageCluster(command.Command):
+    """Set block storage cluster properties.
+
+    This command requires ``--os-volume-api-version`` 3.7 or greater.
+    """
+
+    def get_parser(self, prog_name):
+        parser = super().get_parser(prog_name)
+        parser.add_argument(
+            'cluster',
+            metavar='<cluster>',
+            help=_('Name of block storage cluster to update (name only)')
+        )
+        parser.add_argument(
+            '--binary',
+            metavar='<binary>',
+            default='cinder-volume',
+            help=_(
+                "Name of binary to filter by; defaults to 'cinder-volume' "
+                "(optional)"
+            )
+        )
+        enabled_group = parser.add_mutually_exclusive_group()
+        enabled_group.add_argument(
+            '--enable',
+            action='store_false',
+            dest='disabled',
+            default=None,
+            help=_('Enable cluster')
+        )
+        enabled_group.add_argument(
+            '--disable',
+            action='store_true',
+            dest='disabled',
+            help=_('Disable cluster')
+        )
+        parser.add_argument(
+            '--disable-reason',
+            metavar='<reason>',
+            dest='disabled_reason',
+            help=_(
+                'Reason for disabling the cluster '
+                '(should be used with --disable option)'
+            )
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        volume_client = self.app.client_manager.volume
+
+        if volume_client.api_version < api_versions.APIVersion('3.7'):
+            msg = _(
+                "--os-volume-api-version 3.7 or greater is required to "
+                "support the 'block storage cluster set' command"
+            )
+            raise exceptions.CommandError(msg)
+
+        if parsed_args.disabled_reason and not parsed_args.disabled:
+            msg = _("Cannot specify --disable-reason without --disable")
+            raise exceptions.CommandError(msg)
+
+        cluster = volume_client.clusters.update(
+            parsed_args.cluster,
+            parsed_args.binary,
+            disabled=parsed_args.disabled,
+            disabled_reason=parsed_args.disabled_reason,
+        )
+
+        return _format_cluster(cluster, detailed=True)
+
+
+class ShowBlockStorageCluster(command.ShowOne):
+    """Show detailed information for a block storage cluster.
+
+    This command requires ``--os-volume-api-version`` 3.7 or greater.
+    """
+
+    def get_parser(self, prog_name):
+        parser = super().get_parser(prog_name)
+        parser.add_argument(
+            'cluster',
+            metavar='<cluster>',
+            help=_('Name of block storage cluster.'),
+        )
+        parser.add_argument(
+            '--binary',
+            metavar='<binary>',
+            help=_('Service binary.'),
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        volume_client = self.app.client_manager.volume
+
+        if volume_client.api_version < api_versions.APIVersion('3.7'):
+            msg = _(
+                "--os-volume-api-version 3.7 or greater is required to "
+                "support the 'block storage cluster show' command"
+            )
+            raise exceptions.CommandError(msg)
+
+        cluster = volume_client.clusters.show(
+            parsed_args.cluster,
+            binary=parsed_args.binary,
+        )
+
+        return _format_cluster(cluster, detailed=True)
diff --git a/releasenotes/notes/add-block-storage-cluster-commands-fae8f686582bbbcf.yaml b/releasenotes/notes/add-block-storage-cluster-commands-fae8f686582bbbcf.yaml
new file mode 100644
index 0000000000..be50aeb491
--- /dev/null
+++ b/releasenotes/notes/add-block-storage-cluster-commands-fae8f686582bbbcf.yaml
@@ -0,0 +1,7 @@
+---
+features:
+  - |
+    Add ``block storage cluster create``, ``block storage cluster delete``,
+    ``block storage cluster list`` and ``block storage cluster show`` commands
+    to create, delete, list, and show block storage service clusters,
+    respectively.
diff --git a/setup.cfg b/setup.cfg
index 79e805f04e..d5eaab1872 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -756,6 +756,10 @@ openstack.volume.v3 =
     volume_message_list = openstackclient.volume.v3.volume_message:ListMessages
     volume_message_show = openstackclient.volume.v3.volume_message:ShowMessage
 
+    block_storage_cluster_list = openstackclient.volume.v3.block_storage_cluster:ListBlockStorageCluster
+    block_storage_cluster_set = openstackclient.volume.v3.block_storage_cluster:SetBlockStorageCluster
+    block_storage_cluster_show = openstackclient.volume.v3.block_storage_cluster:ShowBlockStorageCluster
+
     volume_snapshot_create = openstackclient.volume.v2.volume_snapshot:CreateVolumeSnapshot
     volume_snapshot_delete = openstackclient.volume.v2.volume_snapshot:DeleteVolumeSnapshot
     volume_snapshot_list = openstackclient.volume.v2.volume_snapshot:ListVolumeSnapshot