diff --git a/heat/engine/resources/openstack/sahara/cluster.py b/heat/engine/resources/openstack/sahara/cluster.py
index 2a71909bed..b5e2166e07 100644
--- a/heat/engine/resources/openstack/sahara/cluster.py
+++ b/heat/engine/resources/openstack/sahara/cluster.py
@@ -13,6 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import re
+
 from oslo_log import log as logging
 
 from heat.common import exception
@@ -26,6 +28,15 @@ from heat.engine import translation
 
 LOG = logging.getLogger(__name__)
 
+# NOTE(jfreud, pshchelo): copied from sahara/utils/api_validator.py
+SAHARA_NAME_REGEX = (r"^(([a-zA-Z]|[a-zA-Z][a-zA-Z0-9\-]"
+                     r"*[a-zA-Z0-9])\.)*([A-Za-z]|[A-Za-z]"
+                     r"[A-Za-z0-9\-]*[A-Za-z0-9])$")
+
+# NOTE(jfreud): we do not use physical_resource_name_limit attr because we
+# prefer to truncate _after_ removing invalid characters
+SAHARA_CLUSTER_NAME_MAX_LENGTH = 80
+
 
 class SaharaCluster(resource.Resource):
     """A resource for managing Sahara clusters.
@@ -69,6 +80,10 @@ class SaharaCluster(resource.Resource):
         NAME: properties.Schema(
             properties.Schema.STRING,
             _('Hadoop cluster name.'),
+            constraints=[
+                constraints.Length(min=1, max=SAHARA_CLUSTER_NAME_MAX_LENGTH),
+                constraints.AllowedPattern(SAHARA_NAME_REGEX),
+            ],
         ),
         PLUGIN_NAME: properties.Schema(
             properties.Schema.STRING,
@@ -208,7 +223,9 @@ class SaharaCluster(resource.Resource):
         name = self.properties[self.NAME]
         if name:
             return name
-        return self.physical_resource_name()
+        return self.reduce_physical_resource_name(
+            re.sub('[^a-zA-Z0-9-]', '', self.physical_resource_name()),
+            SAHARA_CLUSTER_NAME_MAX_LENGTH)
 
     def handle_create(self):
         plugin_name = self.properties[self.PLUGIN_NAME]
diff --git a/heat/tests/openstack/sahara/test_cluster.py b/heat/tests/openstack/sahara/test_cluster.py
index a09971254c..f12d46f252 100644
--- a/heat/tests/openstack/sahara/test_cluster.py
+++ b/heat/tests/openstack/sahara/test_cluster.py
@@ -47,6 +47,25 @@ resources:
           access_level: ro
 """
 
+# NOTE(jfreud): the resource name contains an invalid character
+cluster_stack_template_without_name = """
+heat_template_version: 2013-05-23
+description: Hadoop Cluster by Sahara
+resources:
+  lots_of_underscore_name:
+    type: OS::Sahara::Cluster
+    properties:
+      plugin_name: vanilla
+      hadoop_version: 2.3.0
+      cluster_template_id: some_cluster_template_id
+      default_image_id: some_image
+      key_name: admin
+      neutron_management_network: some_network
+      shares:
+        - id: some_share_id
+          access_level: ro
+"""
+
 
 class FakeCluster(object):
     def __init__(self, status='Active'):
@@ -79,10 +98,11 @@ class SaharaClusterTest(common.HeatTestCase):
         self.fake_cl = FakeCluster()
 
         self.t = template_format.parse(cluster_stack_template)
+        self.t2 = template_format.parse(cluster_stack_template_without_name)
 
-    def _init_cluster(self, template):
+    def _init_cluster(self, template, name='super-cluster'):
         self.stack = utils.parse_stack(template)
-        cluster = self.stack['super-cluster']
+        cluster = self.stack[name]
         return cluster
 
     def _create_cluster(self, template):
@@ -110,6 +130,14 @@ class SaharaClusterTest(common.HeatTestCase):
                                                    **expected_kwargs)
         self.cl_mgr.get.assert_called_once_with(self.fake_cl.id)
 
+    def test_cluster_create_invalid_name(self):
+        cluster = self._init_cluster(self.t2, 'lots_of_underscore_name')
+        self.cl_mgr.create.return_value = self.fake_cl
+        self.cl_mgr.get.return_value = self.fake_cl
+        scheduler.TaskRunner(cluster.create)()
+        name = self.cl_mgr.create.call_args[0][0]
+        self.assertIn('lotsofunderscorename', name)
+
     def test_cluster_create_fails(self):
         cfg.CONF.set_override('action_retry_limit', 0)
         cluster = self._init_cluster(self.t)