diff --git a/plugins/doc_fragments/azure_rm.py b/plugins/doc_fragments/azure_rm.py
index 620f30f69..b96037624 100644
--- a/plugins/doc_fragments/azure_rm.py
+++ b/plugins/doc_fragments/azure_rm.py
@@ -79,4 +79,15 @@ class ModuleDocFragment(object):
         type: bool
         default: False
         version_added: '2.8'
+    hostnames:
+        description:
+        - A list of Jinja2 expressions in order of precedence to compose inventory_hostname.
+        - Ignores expression if result is an empty string or None value.
+        - By default, inventory_hostname is generated to be globally unique based on the VM host name.
+            See C(plain_host_names) for more details on the default.
+        - An expression of 'default' will force using the default hostname generator if no previous hostname expression
+            resulted in a valid hostname.
+        - Use ``default_inventory_hostname`` to access the default hostname generator's value in any of the Jinja2 expressions.
+        type: list
+        default: [default]
 '''
diff --git a/plugins/inventory/azure_rm.py b/plugins/inventory/azure_rm.py
index 6e4118400..cc7c7bf3d 100644
--- a/plugins/inventory/azure_rm.py
+++ b/plugins/inventory/azure_rm.py
@@ -74,6 +74,11 @@
   # if none are found, the first public IP address.
   ansible_host: (public_dns_hostnames + public_ipv4_addresses) | first
 
+# change how inventory_hostname is generated. Each item is a jinja2 expression similar to hostvar_expressions.
+hostnames:
+  - tags.vm_name
+  - default  # special var that uses the default hashed name
+
 # places hosts in dynamically-created groups based on a variable value.
 keyed_groups:
 # places each host in a group named 'tag_(tag name)_(tag value)' for each tag on a VM.
@@ -116,7 +121,7 @@
 from ansible_collections.azure.azcollection.plugins.module_utils.azure_rm_common import AzureRMAuth
 from ansible.errors import AnsibleParserError, AnsibleError
 from ansible.module_utils.parsing.convert_bool import boolean
-from ansible.module_utils._text import to_native, to_bytes
+from ansible.module_utils._text import to_native, to_bytes, to_text
 from itertools import chain
 from msrest import ServiceClient, Serializer, Deserializer
 from msrestazure import AzureConfiguration
@@ -264,8 +269,11 @@ def _get_hosts(self):
         constructable_config_groups = self.get_option('conditional_groups')
         constructable_config_keyed_groups = self.get_option('keyed_groups')
 
+        constructable_hostnames = self.get_option('hostnames')
+
         for h in self._hosts:
-            inventory_hostname = self._get_hostname(h)
+            # FUTURE: track hostnames to warn if a hostname is repeated (can happen for legacy and for composed inventory_hostname)
+            inventory_hostname = self._get_hostname(h, hostnames=constructable_hostnames, strict=constructable_config_strict)
             if self._filter_host(inventory_hostname, h.hostvars):
                 continue
             self.inventory.add_host(inventory_hostname)
@@ -298,9 +306,30 @@ def _filter_host(self, inventory_hostname, hostvars):
 
         return False
 
-    def _get_hostname(self, host):
-        # FUTURE: configurable hostname sources
-        return host.default_inventory_hostname
+    def _get_hostname(self, host, hostnames=None, strict=False):
+        hostname = None
+        errors = []
+
+        for preference in hostnames:
+            if preference == 'default':
+                return host.default_inventory_hostname
+            try:
+                hostname = self._compose(preference, host.hostvars)
+            except Exception as e:  # pylint: disable=broad-except
+                if strict:
+                    raise AnsibleError("Could not compose %s as hostnames - %s" % (preference, to_native(e)))
+                else:
+                    errors.append(
+                        (preference, str(e))
+                    )
+            if hostname:
+                return to_text(hostname)
+
+        raise AnsibleError(
+            'Could not template any hostname for host, errors for each preference: %s' % (
+                ', '.join(['%s: %s' % (pref, err) for pref, err in errors])
+            )
+        )
 
     def _process_queue_serial(self):
         try:
@@ -499,7 +528,8 @@ def hostvars(self):
             ) if self._vmss else {},
             virtual_machine_size=self._vm_model['properties']['hardwareProfile']['vmSize'] if self._vm_model['properties'].get('hardwareProfile') else None,
             plan=self._vm_model['properties']['plan']['name'] if self._vm_model['properties'].get('plan') else None,
-            resource_group=parse_resource_id(self._vm_model['id']).get('resource_group').lower()
+            resource_group=parse_resource_id(self._vm_model['id']).get('resource_group').lower(),
+            default_inventory_hostname=self.default_inventory_hostname,
         )
 
         # set nic-related values from the primary NIC first