diff --git a/src/azure-cli-core/azure/cli/core/_help.py b/src/azure-cli-core/azure/cli/core/_help.py
index 90432397383..d1e568bfb8d 100644
--- a/src/azure-cli-core/azure/cli/core/_help.py
+++ b/src/azure-cli-core/azure/cli/core/_help.py
@@ -183,6 +183,26 @@ def show_help(self, cli_name, nouns, parser, is_group):
             if not nouns:
                 print(UX_SURVEY_PROMPT_COLOR if self.cli_ctx.enable_color else UX_SURVEY_PROMPT)
 
+    def get_examples(self, command, parser, is_group):
+        """Get examples of a certain command from the help file.
+        Get the text of the example, strip the newline character and
+        return a list of commands which start with the given command name.
+        """
+        nouns = command.split(' ')[1:]
+        self.update_loaders_with_help_file_contents(nouns)
+
+        delimiters = ' '.join(nouns)
+        help_file = self.command_help_cls(self, delimiters, parser) if not is_group \
+            else self.group_help_cls(self, delimiters, parser)
+        help_file.load(parser)
+
+        def strip_command(command):
+            command = command.replace('\\\n', '')
+            contents = [item for item in command.split(' ') if item]
+            return ' '.join(contents).strip()
+
+        return [strip_command(example.command) for example in help_file.examples]
+
     def _register_help_loaders(self):
         import azure.cli.core._help_loaders as help_loaders
         import inspect
diff --git a/src/azure-cli-core/azure/cli/core/azclierror.py b/src/azure-cli-core/azure/cli/core/azclierror.py
new file mode 100644
index 00000000000..f1250155655
--- /dev/null
+++ b/src/azure-cli-core/azure/cli/core/azclierror.py
@@ -0,0 +1,88 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import sys
+from enum import Enum
+
+from knack.util import CLIError
+from knack.log import get_logger
+
+logger = get_logger(__name__)
+
+
+class AzCLIErrorType(Enum):
+    """ AzureCLI error types """
+
+    # userfaults
+    CommandNotFoundError = 'CommandNotFoundError'
+    ArgumentParseError = 'ArgumentParseError'
+    ValidationError = 'ValidationError'
+    ManualInterrupt = 'ManualInterrupt'
+    # service side error
+    ServiceError = 'ServiceError'
+    # client side error
+    ClientError = 'ClientError'
+    # unexpected error
+    UnexpectedError = 'UnexpectedError'
+
+
+class AzCLIError(CLIError):
+    """ AzureCLI error definition """
+
+    def __init__(self, error_type, error_msg, raw_exception=None, command=None):
+        """
+        :param error_type: The name of the AzureCLI error type.
+        :type error_type: azure.cli.core.util.AzCLIErrorType
+        :param error_msg: The error message detail.
+        :type error_msg: str
+        :param raw_exception: The raw exception.
+        :type raw_exception: Exception
+        :param command: The command which brings the error.
+        :type command: str
+        :param recommendations: The recommendations to resolve the error.
+        :type recommendations: list
+        """
+        self.error_type = error_type
+        self.error_msg = error_msg
+        self.raw_exception = raw_exception
+        self.command = command
+        self.recommendations = []
+        super().__init__(error_msg)
+
+    def set_recommendation(self, recommendation):
+        self.recommendations.append(recommendation)
+
+    def set_raw_exception(self, raw_exception):
+        self.raw_exception = raw_exception
+
+    def print_error(self):
+        from azure.cli.core.azlogging import CommandLoggerContext
+        with CommandLoggerContext(logger):
+            message = '{}: {}'.format(self.error_type.value, self.error_msg)
+            logger.error(message)
+            if self.raw_exception:
+                logger.exception(self.raw_exception)
+            if self.recommendations:
+                for recommendation in self.recommendations:
+                    print(recommendation, file=sys.stderr)
+
+    def send_telemetry(self):
+        import azure.cli.core.telemetry as telemetry
+        telemetry.set_error_type(self.error_type.value)
+
+        # For userfaults
+        if self.error_type in [AzCLIErrorType.CommandNotFoundError,
+                               AzCLIErrorType.ArgumentParseError,
+                               AzCLIErrorType.ValidationError,
+                               AzCLIErrorType.ManualInterrupt]:
+            telemetry.set_user_fault(self.error_msg)
+
+        # For failures: service side error, client side error, unexpected error
+        else:
+            telemetry.set_failure(self.error_msg)
+
+        # For unexpected error
+        if self.raw_exception:
+            telemetry.set_exception(self.raw_exception, '')
diff --git a/src/azure-cli-core/azure/cli/core/cloud.py b/src/azure-cli-core/azure/cli/core/cloud.py
index 79b997f778a..af183e9c166 100644
--- a/src/azure-cli-core/azure/cli/core/cloud.py
+++ b/src/azure-cli-core/azure/cli/core/cloud.py
@@ -20,8 +20,11 @@
 
 CLOUD_CONFIG_FILE = os.path.join(GLOBAL_CONFIG_DIR, 'clouds.config')
 
-# Add names of clouds that don't allow telemetry data collection here such as JEDI.
-CLOUDS_FORBIDDING_TELEMETRY = []
+# Add names of clouds that don't allow telemetry data collection here such as some air-gapped clouds.
+CLOUDS_FORBIDDING_TELEMETRY = ['USSec', 'USNat']
+
+# Add names of clouds that don't allow Aladdin requests for command recommendations here
+CLOUDS_FORBIDDING_ALADDIN_REQUEST = ['USSec', 'USNat']
 
 
 class CloudNotRegisteredException(Exception):
diff --git a/src/azure-cli-core/azure/cli/core/command_recommender.py b/src/azure-cli-core/azure/cli/core/command_recommender.py
new file mode 100644
index 00000000000..e7c2a69d492
--- /dev/null
+++ b/src/azure-cli-core/azure/cli/core/command_recommender.py
@@ -0,0 +1,306 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import difflib
+from enum import Enum
+
+import azure.cli.core.telemetry as telemetry
+from knack.log import get_logger
+
+
+logger = get_logger(__name__)
+
+
+class AladdinUserFaultType(Enum):
+    """Define the userfault types required by aladdin service
+    to get the command recommendations"""
+
+    ExpectedArgument = 'ExpectedArgument'
+    UnrecognizedArguments = 'UnrecognizedArguments'
+    ValidationError = 'ValidationError'
+    UnknownSubcommand = 'UnknownSubcommand'
+    MissingRequiredParameters = 'MissingRequiredParameters'
+    MissingRequiredSubcommand = 'MissingRequiredSubcommand'
+    StorageAccountNotFound = 'StorageAccountNotFound'
+    Unknown = 'Unknown'
+    InvalidJMESPathQuery = 'InvalidJMESPathQuery'
+    InvalidOutputType = 'InvalidOutputType'
+    InvalidParameterValue = 'InvalidParameterValue'
+    UnableToParseCommandInput = 'UnableToParseCommandInput'
+    ResourceGroupNotFound = 'ResourceGroupNotFound'
+    InvalidDateTimeArgumentValue = 'InvalidDateTimeArgumentValue'
+    InvalidResourceGroupName = 'InvalidResourceGroupName'
+    AzureResourceNotFound = 'AzureResourceNotFound'
+    InvalidAccountName = 'InvalidAccountName'
+
+
+class CommandRecommender():
+    """Recommend a command for user when user's command fails.
+    It combines alladin recommendations and examples in help files."""
+
+    def __init__(self, command, parameters, extension, error_msg, cli_ctx):
+        """
+        :param command: The command name in user's input.
+        :type command: str
+        :param parameters: The parameter arguments in users input.
+        :type parameters: list
+        :param extension: The extension name in user's input if the command comes from an extension.
+        :type extension: str
+        :param error_msg: The error message of the failed command.
+        :type error_msg: str
+        :param cli_ctx: CLI context when parser fails.
+        :type cli_ctx: knack.cli.CLI
+        """
+        self.command = command.strip()
+        self.extension = extension
+        self.error_msg = error_msg
+        self.cli_ctx = cli_ctx
+
+        self.parameters = self._normalize_parameters(parameters)
+        self.help_examples = []
+        self.aladdin_recommendations = []
+
+    def set_help_examples(self, examples):
+        """Set recommendations from help files"""
+
+        self.help_examples.extend(examples)
+
+    def _set_aladdin_recommendations(self):
+        """Set recommendations from aladdin service.
+        Call the aladdin service API, parse the response and set the recommendations.
+        """
+
+        import hashlib
+        import json
+        import requests
+        from requests import RequestException
+        from http import HTTPStatus
+        from azure.cli.core import __version__ as version
+
+        api_url = 'https://app.aladdin.microsoft.com/api/v1.0/suggestions'
+        correlation_id = telemetry._session.correlation_id  # pylint: disable=protected-access
+        subscription_id = telemetry._get_azure_subscription_id()  # pylint: disable=protected-access
+        # Used for DDOS protection and rate limiting
+        user_id = telemetry._get_user_azure_id()  # pylint: disable=protected-access
+        hashed_user_id = hashlib.sha256(user_id.encode('utf-8')).hexdigest()
+
+        headers = {
+            'Content-Type': 'application/json',
+            'X-UserId': hashed_user_id
+        }
+        context = {
+            'versionNumber': version,
+            'errorType': self._get_error_type()
+        }
+
+        if telemetry.is_telemetry_enabled():
+            if correlation_id:
+                context['correlationId'] = correlation_id
+            if subscription_id:
+                context['subscriptionId'] = subscription_id
+
+        parameters = [item for item in self.parameters if item not in ['--debug', '--verbose', '--only-show-errors']]
+        query = {
+            "command": self.command,
+            "parameters": ','.join(parameters)
+        }
+
+        response = None
+        try:
+            response = requests.get(
+                api_url,
+                params={
+                    'query': json.dumps(query),
+                    'clientType': 'AzureCli',
+                    'context': json.dumps(context)
+                },
+                headers=headers,
+                timeout=1)
+        except RequestException as ex:
+            logger.debug('Recommendation requests.get() exception: %s', ex)
+            telemetry.set_debug_info('AladdinRecommendationService', ex.__class__.__name__)
+
+        recommendations = []
+        if response and response.status_code == HTTPStatus.OK:
+            for result in response.json():
+                # parse the reponse and format the recommendation
+                command, parameters, placeholders = result['command'],\
+                    result['parameters'].split(','),\
+                    result['placeholders'].split('♠')
+                recommendation = 'az {} '.format(command)
+                for parameter, placeholder in zip(parameters, placeholders):
+                    recommendation += '{} {} '.format(parameter, placeholder)
+                recommendations.append(recommendation.strip())
+
+        self.aladdin_recommendations.extend(recommendations)
+
+    def recommend_a_command(self):
+        """Recommend a command for user when user's command fails.
+        The recommended command will be the best matched one from
+        both the help files and the aladdin recommendations.
+        """
+        if not self._disable_aladdin_service():
+            self._set_aladdin_recommendations()
+
+        recommend_command = ''
+        if self.help_examples and self.aladdin_recommendations:
+            # all the recommended commands from help examples and aladdin
+            all_commands = self.help_examples + self.aladdin_recommendations
+            all_commands.sort(key=len)
+
+            filtered_commands = []
+            filtered_choices = []
+            target = ''.join(self.parameters)
+            example_command_name = self.help_examples[0].split(' -')[0]
+
+            for command in all_commands:
+                # keep only the commands which begin with a same command name with examples
+                if command.startswith(example_command_name):
+                    parameters = self._get_parameter_list(command)
+                    normalized_parameters = self._normalize_parameters(parameters)
+                    filtered_choices.append(''.join(normalized_parameters))
+                    filtered_commands.append(command)
+
+            # sort the commands by argument matches
+            candidates = difflib.get_close_matches(target, filtered_choices, cutoff=0)
+
+            if candidates:
+                index = filtered_choices.index(candidates[0])
+                recommend_command = filtered_commands[index]
+
+        # fallback to use the first recommended command from Aladdin
+        elif self.aladdin_recommendations:
+            recommend_command = self.aladdin_recommendations[0]
+
+        return recommend_command
+
+    def _disable_aladdin_service(self):
+        """Decide whether to disable aladdin request when a command fails.
+        The possible cases to disable it are:
+            1. In air-gapped clouds
+            2. In testing environments
+        """
+        from azure.cli.core.cloud import CLOUDS_FORBIDDING_ALADDIN_REQUEST
+
+        # CLI is not started well
+        if not self.cli_ctx or not self.cli_ctx.cloud:
+            return True
+
+        # for air-gapped clouds
+        if self.cli_ctx.cloud.name in CLOUDS_FORBIDDING_ALADDIN_REQUEST:
+            return True
+
+        # for testing environments
+        if self.cli_ctx.__class__.__name__ == 'DummyCli':
+            return True
+
+        return False
+
+    def _get_parameter_list(self, raw_command):  # pylint: disable=no-self-use
+        """Get the paramter list from a raw command string
+        An example: 'az group create -n test -l eastus' ==> ['-n', '-l']
+        """
+        contents = raw_command.split(' ')
+        return [item for item in contents if item.startswith('-')]
+
+    def _normalize_parameters(self, parameters):
+        """Normalize a parameter list.
+        Get the standard form of a parameter list, which includes:
+            1. Use long options to replace short options
+            2. Remove the unrecognized parameters
+            3. Sort the result parameter list
+        An example: ['-g', '-n'] ==> ['--name', '--resource-group']
+        """
+
+        from knack.deprecation import Deprecated
+
+        normalized_parameters = []
+        try:
+            cmd_table = self.cli_ctx.invocation.commands_loader.command_table.get(self.command, None)
+            parameter_table = cmd_table.arguments if cmd_table else None
+        except AttributeError:
+            parameter_table = None
+
+        if parameters:
+            rules = {
+                '-h': '--help',
+                '-o': '--output',
+                '--only-show-errors': None,
+                '--help': None,
+                '--output': None,
+                '--query': None,
+                '--debug': None,
+                '--verbose': None
+            }
+
+            if parameter_table:
+                for argument in parameter_table.values():
+                    options = argument.type.settings['options_list']
+                    options = (option for option in options if not isinstance(option, Deprecated))
+                    try:
+                        sorted_options = sorted(options, key=len, reverse=True)
+                        standard_form = sorted_options[0]
+
+                        for option in sorted_options[1:]:
+                            rules[option] = standard_form
+                        rules[standard_form] = None
+                    except TypeError:
+                        logger.debug('Unexpected argument options `%s` of type `%s`.', options, type(options).__name__)
+
+            for parameter in parameters:
+                if parameter in rules:
+                    normalized_form = rules.get(parameter, None) or parameter
+                    normalized_parameters.append(normalized_form)
+                else:
+                    logger.debug('"%s" is an invalid parameter for command "%s".', parameter, self.command)
+
+        return sorted(normalized_parameters)
+
+    def _get_error_type(self):
+        """The the error type of the failed command from the error message.
+        The error types are only consumed by aladdin service for better recommendations.
+        """
+
+        error_type = AladdinUserFaultType.Unknown
+        if not self.error_msg:
+            return error_type.value
+
+        error_msg = self.error_msg.lower()
+        if 'unrecognized' in error_msg:
+            error_type = AladdinUserFaultType.UnrecognizedArguments
+        elif 'expected one argument' in error_msg or 'expected at least one argument' in error_msg \
+                or 'value required' in error_msg:
+            error_type = AladdinUserFaultType.ExpectedArgument
+        elif 'misspelled' in error_msg:
+            error_type = AladdinUserFaultType.UnknownSubcommand
+        elif 'arguments are required' in error_msg or 'argument required' in error_msg:
+            error_type = AladdinUserFaultType.MissingRequiredParameters
+            if '_subcommand' in error_msg:
+                error_type = AladdinUserFaultType.MissingRequiredSubcommand
+            elif '_command_package' in error_msg:
+                error_type = AladdinUserFaultType.UnableToParseCommandInput
+        elif 'not found' in error_msg or 'could not be found' in error_msg \
+                or 'resource not found' in error_msg:
+            error_type = AladdinUserFaultType.AzureResourceNotFound
+            if 'storage_account' in error_msg or 'storage account' in error_msg:
+                error_type = AladdinUserFaultType.StorageAccountNotFound
+            elif 'resource_group' in error_msg or 'resource group' in error_msg:
+                error_type = AladdinUserFaultType.ResourceGroupNotFound
+        elif 'pattern' in error_msg or 'is not a valid value' in error_msg or 'invalid' in error_msg:
+            error_type = AladdinUserFaultType.InvalidParameterValue
+            if 'jmespath_type' in error_msg:
+                error_type = AladdinUserFaultType.InvalidJMESPathQuery
+            elif 'datetime_type' in error_msg:
+                error_type = AladdinUserFaultType.InvalidDateTimeArgumentValue
+            elif '--output' in error_msg:
+                error_type = AladdinUserFaultType.InvalidOutputType
+            elif 'resource_group' in error_msg:
+                error_type = AladdinUserFaultType.InvalidResourceGroupName
+            elif 'storage_account' in error_msg:
+                error_type = AladdinUserFaultType.InvalidAccountName
+        elif "validation error" in error_msg:
+            error_type = AladdinUserFaultType.ValidationError
+
+        return error_type.value
diff --git a/src/azure-cli-core/azure/cli/core/commands/__init__.py b/src/azure-cli-core/azure/cli/core/commands/__init__.py
index d567c85bc3a..4d4e79ac646 100644
--- a/src/azure-cli-core/azure/cli/core/commands/__init__.py
+++ b/src/azure-cli-core/azure/cli/core/commands/__init__.py
@@ -845,9 +845,13 @@ def _validate_cmd_level(self, ns, cmd_validator):  # pylint: disable=no-self-use
             pass
 
     def _validate_arg_level(self, ns, **_):  # pylint: disable=no-self-use
+        from azure.cli.core.azclierror import AzCLIErrorType
+        from azure.cli.core.azclierror import AzCLIError
         for validator in getattr(ns, '_argument_validators', []):
             try:
                 validator(**self._build_kwargs(validator, ns))
+            except AzCLIError:
+                raise
             except Exception as ex:
                 # Delay the import and mimic an exception handler
                 from msrest.exceptions import ValidationError
diff --git a/src/azure-cli-core/azure/cli/core/commands/arm.py b/src/azure-cli-core/azure/cli/core/commands/arm.py
index 296b692c226..d58862760db 100644
--- a/src/azure-cli-core/azure/cli/core/commands/arm.py
+++ b/src/azure-cli-core/azure/cli/core/commands/arm.py
@@ -762,8 +762,12 @@ def show_exception_handler(ex):
     if getattr(getattr(ex, 'response', ex), 'status_code', None) == 404:
         import sys
         from azure.cli.core.azlogging import CommandLoggerContext
+        from azure.cli.core.azclierror import AzCLIErrorType
+        from azure.cli.core.azclierror import AzCLIError
         with CommandLoggerContext(logger):
-            logger.error(getattr(ex, 'message', ex))
+            az_error = AzCLIError(AzCLIErrorType.ValidationError, getattr(ex, 'message', ex))
+            az_error.print_error()
+            az_error.send_telemetry()
             sys.exit(3)
     raise ex
 
diff --git a/src/azure-cli-core/azure/cli/core/parser.py b/src/azure-cli-core/azure/cli/core/parser.py
index 6e56daa5015..6b2d4d1c853 100644
--- a/src/azure-cli-core/azure/cli/core/parser.py
+++ b/src/azure-cli-core/azure/cli/core/parser.py
@@ -17,6 +17,9 @@
 from azure.cli.core.commands import ExtensionCommandSource
 from azure.cli.core.commands import AzCliCommandInvoker
 from azure.cli.core.commands.events import EVENT_INVOKER_ON_TAB_COMPLETION
+from azure.cli.core.azclierror import AzCLIErrorType
+from azure.cli.core.azclierror import AzCLIError
+from azure.cli.core.command_recommender import CommandRecommender
 
 from knack.log import get_logger
 from knack.parser import CLICommandParser
@@ -24,6 +27,15 @@
 
 logger = get_logger(__name__)
 
+EXTENSION_REFERENCE = ("If the command is from an extension, "
+                       "please make sure the corresponding extension is installed. "
+                       "To learn more about extensions, please visit "
+                       "'https://docs.microsoft.com/en-us/cli/azure/azure-cli-extensions-overview'")
+
+OVERVIEW_REFERENCE = ("Still stuck? Run '{command} --help' to view all commands or go to "
+                      "'https://docs.microsoft.com/en-us/cli/azure/reference-index?view=azure-cli-latest' "
+                      "to learn more")
+
 
 class IncorrectUsageError(CLIError):
     '''Raised when a command is incorrectly used and the usage should be
@@ -143,18 +155,30 @@ def load_command_table(self, command_loader):
                 _parser=command_parser)
 
     def validation_error(self, message):
-        telemetry.set_user_fault('validation error: {}'.format(message))
-        return super(AzCliCommandParser, self).error(message)
+        az_error = AzCLIError(AzCLIErrorType.ValidationError, message, command=self.prog)
+        az_error.print_error()
+        az_error.send_telemetry()
+        self.exit(2)
 
     def error(self, message):
-        telemetry.set_user_fault('parse error: {}'.format(message))
-        args = {'prog': self.prog, 'message': message}
-        with CommandLoggerContext(logger):
-            logger.error('%(prog)s: error: %(message)s', args)
-        self.print_usage(sys.stderr)
-        # Manual recommendations
-        self._set_manual_recommendations(args['message'])
-        # AI recommendations
+        # Get a recommended command from the CommandRecommender
+        command_arguments = self._get_failure_recovery_arguments()
+        cli_ctx = self.cli_ctx or (self.cli_help.cli_ctx if self.cli_help else None)
+        recommender = CommandRecommender(*command_arguments, message, cli_ctx)
+        recommender.set_help_examples(self.get_examples(self.prog))
+        recommendation = recommender.recommend_a_command()
+
+        az_error = AzCLIError(AzCLIErrorType.ArgumentParseError, message, command=self.prog)
+        if '--query' in message:
+            from azure.cli.core.util import QUERY_REFERENCE
+            az_error.set_recommendation(QUERY_REFERENCE)
+        elif recommendation:
+            az_error.set_recommendation("Try this: '{}'".format(recommendation))
+            az_error.set_recommendation(OVERVIEW_REFERENCE.format(command=self.prog))
+        az_error.print_error()
+        az_error.send_telemetry()
+
+        # For ai-did-you-mean-this
         failure_recovery_recommendations = self._get_failure_recovery_recommendations()
         self._suggestion_msg.extend(failure_recovery_recommendations)
         self._print_suggestion_msg(sys.stderr)
@@ -177,19 +201,19 @@ def format_help(self):
         telemetry.set_success(summary='show help')
         super(AzCliCommandParser, self).format_help()
 
+    def get_examples(self, command):
+        if not self.cli_help:
+            return []
+        is_group = self.is_group()
+        return self.cli_help.get_examples(command,
+                                          self._actions[-1] if is_group else self,
+                                          is_group)
+
     def enable_autocomplete(self):
         argcomplete.autocomplete = AzCompletionFinder()
         argcomplete.autocomplete(self, validator=lambda c, p: c.lower().startswith(p.lower()),
                                  default_completer=lambda _: ())
 
-    def _set_manual_recommendations(self, error_msg):
-        recommendations = []
-        # recommendation for --query value error
-        if '--query' in error_msg:
-            recommendations.append('To learn more about [--query JMESPATH] usage in AzureCLI, '
-                                   'visit https://aka.ms/CLIQuery')
-        self._suggestion_msg.extend(recommendations)
-
     def _get_failure_recovery_arguments(self, action=None):
         # Strip the leading "az " and any extraneous whitespace.
         command = self.prog[3:].strip()
@@ -230,7 +254,7 @@ def has_extension_name(command_source):
             extension = self.command_source.extension_name
         # Otherwise, the command may have not been in a command group. The command source will not be
         # set in this case.
-        elif action and action.dest == '_subcommand':
+        elif action and action.dest in ('_subcommand', '_command_package'):
             # Get all parsers in the set of possible actions.
             parsers = list(action.choices.values())
             parser = parsers[0] if parsers else None
@@ -349,13 +373,21 @@ def _check_value(self, action, value):  # pylint: disable=too-many-statements, t
         # Override to customize the error message when a argument is not among the available choices
         # converted value must be one of the choices (if specified)
         if action.choices is not None and value not in action.choices:  # pylint: disable=too-many-nested-blocks
+            # self.cli_ctx is None when self.prog is beyond 'az', such as 'az iot'.
+            # use cli_ctx from cli_help which is not lost.
+            cli_ctx = self.cli_ctx or (self.cli_help.cli_ctx if self.cli_help else None)
+
             caused_by_extension_not_installed = False
+            command_name_inferred = self.prog
+            error_msg = None
             if not self.command_source:
                 candidates = difflib.get_close_matches(value, action.choices, cutoff=0.7)
-                error_msg = None
-                # self.cli_ctx is None when self.prog is beyond 'az', such as 'az iot'.
-                # use cli_ctx from cli_help which is not lost.
-                cli_ctx = self.cli_ctx or (self.cli_help.cli_ctx if self.cli_help else None)
+                if candidates:
+                    # use the most likely candidate to replace the misspelled command
+                    args = self.prog.split() + self._raw_arguments
+                    args_inferred = [item if item != value else candidates[0] for item in args]
+                    command_name_inferred = ' '.join(args_inferred).split('-')[0]
+
                 use_dynamic_install = self._get_extension_use_dynamic_install_config()
                 if use_dynamic_install != 'no' and not candidates:
                     # Check if the command is from an extension
@@ -400,46 +432,55 @@ def _check_value(self, action, value):  # pylint: disable=too-many-statements, t
                                 import subprocess
                                 import platform
                                 exit_code = subprocess.call(cmd_list, shell=platform.system() == 'Windows')
-                                telemetry.set_user_fault("Extension {} dynamically installed and commands will be "
-                                                         "rerun automatically.".format(ext_name))
+                                error_msg = ("Extension {} dynamically installed and commands will be "
+                                             "rerun automatically.").format(ext_name)
+                                telemetry.set_user_fault(error_msg)
                                 self.exit(exit_code)
                             else:
-                                error_msg = 'Extension {} installed. Please rerun your command.'.format(ext_name)
+                                with CommandLoggerContext(logger):
+                                    error_msg = 'Extension {} installed. Please rerun your command.'.format(ext_name)
+                                    logger.error(error_msg)
+                                    telemetry.set_user_fault(error_msg)
+                                self.exit(2)
                         else:
                             error_msg = "The command requires the extension {ext_name}. " \
                                 "To install, run 'az extension add -n {ext_name}'.".format(ext_name=ext_name)
                 if not error_msg:
                     # parser has no `command_source`, value is part of command itself
-                    error_msg = "{prog}: '{value}' is not in the '{prog}' command group. See '{prog} --help'." \
-                        .format(prog=self.prog, value=value)
-                    if use_dynamic_install.lower() == 'no':
-                        extensions_link = 'https://docs.microsoft.com/en-us/cli/azure/azure-cli-extensions-overview'
-                        error_msg = ("{msg} "
-                                     "If the command is from an extension, "
-                                     "please make sure the corresponding extension is installed. "
-                                     "To learn more about extensions, please visit "
-                                     "{extensions_link}").format(msg=error_msg, extensions_link=extensions_link)
+                    error_msg = "'{value}' is misspelled or not recognized by the system.".format(value=value)
+                az_error = AzCLIError(AzCLIErrorType.CommandNotFoundError, error_msg, command=self.prog)
+
             else:
                 # `command_source` indicates command values have been parsed, value is an argument
                 parameter = action.option_strings[0] if action.option_strings else action.dest
-                error_msg = "{prog}: '{value}' is not a valid value for '{param}'. See '{prog} --help'.".format(
+                error_msg = "{prog}: '{value}' is not a valid value for '{param}'.".format(
                     prog=self.prog, value=value, param=parameter)
                 candidates = difflib.get_close_matches(value, action.choices, cutoff=0.7)
+                az_error = AzCLIError(AzCLIErrorType.ArgumentParseError, error_msg, command=self.prog)
 
-            telemetry.set_user_fault(error_msg)
-            with CommandLoggerContext(logger):
-                logger.error(error_msg)
-            if not caused_by_extension_not_installed:
-                if candidates:
-                    print_args = {
-                        's': 's' if len(candidates) > 1 else '',
-                        'verb': 'are' if len(candidates) > 1 else 'is',
-                        'value': value
-                    }
-                    self._suggestion_msg.append("\nThe most similar choice{s} to '{value}' {verb}:"
-                                                .format(**print_args))
-                    self._suggestion_msg.append('\n'.join(['\t' + candidate for candidate in candidates]))
+            command_arguments = self._get_failure_recovery_arguments(action)
+            if candidates:
+                az_error.set_recommendation("Did you mean '{}' ?".format(candidates[0]))
+
+            # recommand a command for user
+            recommender = CommandRecommender(*command_arguments, error_msg, cli_ctx)
+            recommender.set_help_examples(self.get_examples(command_name_inferred))
+            recommended_command = recommender.recommend_a_command()
+            if recommended_command:
+                az_error.set_recommendation("Try this: '{}'".format(recommended_command))
 
+            # remind user to check extensions if we can not find a command to recommend
+            if az_error.error_type == AzCLIErrorType.CommandNotFoundError \
+                    and not az_error.recommendations and self.prog == 'az' \
+                    and use_dynamic_install == 'no':
+                az_error.set_recommendation(EXTENSION_REFERENCE)
+
+            az_error.set_recommendation(OVERVIEW_REFERENCE.format(command=self.prog))
+
+            az_error.print_error()
+            az_error.send_telemetry()
+
+            if not caused_by_extension_not_installed:
                 failure_recovery_recommendations = self._get_failure_recovery_recommendations(action)
                 self._suggestion_msg.extend(failure_recovery_recommendations)
                 self._print_suggestion_msg(sys.stderr)
diff --git a/src/azure-cli-core/azure/cli/core/telemetry.py b/src/azure-cli-core/azure/cli/core/telemetry.py
index 2251af42c84..e54ca3682d2 100644
--- a/src/azure-cli-core/azure/cli/core/telemetry.py
+++ b/src/azure-cli-core/azure/cli/core/telemetry.py
@@ -45,8 +45,15 @@ def __init__(self, correlation_id=None, application=None):
         self.extension_management_detail = None
         self.raw_command = None
         self.mode = 'default'
+        # The AzCLIErrorType
+        self.error_type = 'None'
+        # The class name of the raw exception
+        self.exception_name = 'None'
+        # The stacktrace of the raw exception
+        self.stack_trace = 'None'
         self.init_time_elapsed = None
         self.invoke_time_elapsed = None
+        self.debug_info = []
         # A dictionary with the application insight instrumentation key
         # as the key and an array of telemetry events as value
         self.events = defaultdict(list)
@@ -55,6 +62,13 @@ def __init__(self, correlation_id=None, application=None):
         self.suppress_new_event = False
 
     def add_exception(self, exception, fault_type, description=None, message=''):
+        # Move the exception info into userTask record, in order to make one Telemetry record for one command
+        self.exception_name = exception.__class__.__name__
+        self.result_summary = _remove_cmd_chars(message or str(exception))
+        self.stack_trace = _remove_cmd_chars(_get_stack_trace())
+
+        # Backward compatible, so there are duplicated info recorded
+        # The logic below should be removed along with self.exceptions after confirmation
         fault_type = _remove_symbols(fault_type).replace('"', '').replace("'", '').replace(' ', '-')
         details = {
             'Reserved.DataModel.EntityType': 'Fault',
@@ -179,6 +193,10 @@ def _get_azure_cli_properties(self):
         set_custom_properties(result, 'Mode', self.mode)
         from azure.cli.core._environment import _ENV_AZ_INSTALLER
         set_custom_properties(result, 'Installer', os.getenv(_ENV_AZ_INSTALLER))
+        set_custom_properties(result, 'error_type', self.error_type)
+        set_custom_properties(result, 'exception_name', self.exception_name)
+        set_custom_properties(result, 'stack_trace', self.stack_trace)
+        set_custom_properties(result, 'debug_info', ','.join(self.debug_info))
 
         return result
 
@@ -286,6 +304,13 @@ def set_exception(exception, fault_type, summary=None):
     _session.add_exception(exception, fault_type=fault_type, description=summary)
 
 
+@decorators.suppress_all_exceptions()
+def set_error_type(error_type):
+    if _session.result != 'None':
+        return
+    _session.error_type = error_type
+
+
 @decorators.suppress_all_exceptions()
 def set_failure(summary=None):
     if _session.result != 'None':
@@ -316,6 +341,12 @@ def set_user_fault(summary=None):
         _session.result_summary = _remove_cmd_chars(summary)
 
 
+@decorators.suppress_all_exceptions()
+def set_debug_info(key, info):
+    debug_info = '{}: {}'.format(key, info)
+    _session.debug_info.append(debug_info)
+
+
 @decorators.suppress_all_exceptions()
 def set_application(application, arg_complete_env_name):
     _session.application, _session.arg_complete_env_name = application, arg_complete_env_name
diff --git a/src/azure-cli-core/azure/cli/core/tests/test_parser.py b/src/azure-cli-core/azure/cli/core/tests/test_parser.py
index 47708a6d397..3f295db6c68 100644
--- a/src/azure-cli-core/azure/cli/core/tests/test_parser.py
+++ b/src/azure-cli-core/azure/cli/core/tests/test_parser.py
@@ -239,8 +239,7 @@ def mock_add_extension(*args, **kwargs):
         # assert the right type of error msg is logged for command vs argument parsing
         self.assertEqual(len(logger_msgs), 5)
         for msg in logger_msgs[:3]:
-            self.assertIn("not in the", msg)
-            self.assertIn("command group", msg)
+            self.assertIn("CommandNotFoundError", msg)
         for msg in logger_msgs[3:]:
             self.assertIn("not a valid value for '--opt'.", msg)
 
diff --git a/src/azure-cli-core/azure/cli/core/tests/test_util.py b/src/azure-cli-core/azure/cli/core/tests/test_util.py
index 20a31fa2685..0ea8a16d9d4 100644
--- a/src/azure-cli-core/azure/cli/core/tests/test_util.py
+++ b/src/azure-cli-core/azure/cli/core/tests/test_util.py
@@ -392,7 +392,7 @@ def test_b64_to_hex_type(self):
 
 class TestHandleException(unittest.TestCase):
 
-    @mock.patch('azure.cli.core.util.logger.error', autospec=True)
+    @mock.patch('azure.cli.core.azclierror.logger.error', autospec=True)
     def test_handle_exception_keyboardinterrupt(self, mock_logger_error):
         # create test KeyboardInterrupt Exception
         keyboard_interrupt_ex = KeyboardInterrupt("KeyboardInterrupt")
@@ -401,10 +401,10 @@ def test_handle_exception_keyboardinterrupt(self, mock_logger_error):
         ex_result = handle_exception(keyboard_interrupt_ex)
 
         # test behavior
-        self.assertFalse(mock_logger_error.called)
+        self.assertTrue(mock_logger_error.called)
         self.assertEqual(ex_result, 1)
 
-    @mock.patch('azure.cli.core.util.logger.error', autospec=True)
+    @mock.patch('azure.cli.core.azclierror.logger.error', autospec=True)
     def test_handle_exception_clierror(self, mock_logger_error):
         from knack.util import CLIError
 
@@ -417,10 +417,10 @@ def test_handle_exception_clierror(self, mock_logger_error):
 
         # test behavior
         self.assertTrue(mock_logger_error.called)
-        self.assertEqual(mock.call(err_msg), mock_logger_error.call_args)
+        self.assertIn(err_msg, mock_logger_error.call_args.args[0])
         self.assertEqual(ex_result, 1)
 
-    @mock.patch('azure.cli.core.util.logger.error', autospec=True)
+    @mock.patch('azure.cli.core.azclierror.logger.error', autospec=True)
     def test_handle_exception_clouderror(self, mock_logger_error):
         from msrestazure.azure_exceptions import CloudError
 
@@ -435,10 +435,10 @@ def test_handle_exception_clouderror(self, mock_logger_error):
 
         # test behavior
         self.assertTrue(mock_logger_error.called)
-        self.assertEqual(mock.call(mock_cloud_error.args[0]), mock_logger_error.call_args)
+        self.assertIn(mock_cloud_error.args[0], mock_logger_error.call_args.args[0])
         self.assertEqual(ex_result, mock_cloud_error.args[1])
 
-    @mock.patch('azure.cli.core.util.logger.error', autospec=True)
+    @mock.patch('azure.cli.core.azclierror.logger.error', autospec=True)
     def test_handle_exception_httpoperationerror_typical_response_error(self, mock_logger_error):
         # create test HttpOperationError Exception
         err_msg = "Bad Request because of some incorrect param"
@@ -447,17 +447,16 @@ def test_handle_exception_httpoperationerror_typical_response_error(self, mock_l
         response_text = json.dumps(err)
         mock_http_error = self._get_mock_HttpOperationError(response_text)
 
-        expected_call = mock.call("%s%s", "{} - ".format(err_code), err_msg)
-
         # call handle_exception
         ex_result = handle_exception(mock_http_error)
 
         # test behavior
         self.assertTrue(mock_logger_error.called)
-        self.assertEqual(expected_call, mock_logger_error.call_args)
+        self.assertIn(err_msg, mock_logger_error.call_args.args[0])
+        self.assertIn(err_code, mock_logger_error.call_args.args[0])
         self.assertEqual(ex_result, 1)
 
-    @mock.patch('azure.cli.core.util.logger.error', autospec=True)
+    @mock.patch('azure.cli.core.azclierror.logger.error', autospec=True)
     def test_handle_exception_httpoperationerror_error_key_has_string_value(self, mock_logger_error):
         # test error in response, but has str value.
 
@@ -474,10 +473,10 @@ def test_handle_exception_httpoperationerror_error_key_has_string_value(self, mo
 
         # test behavior
         self.assertTrue(mock_logger_error.called)
-        self.assertEqual(mock.call(expected_message), mock_logger_error.call_args)
+        self.assertIn(expected_message, mock_logger_error.call_args.args[0])
         self.assertEqual(ex_result, 1)
 
-    @mock.patch('azure.cli.core.util.logger.error', autospec=True)
+    @mock.patch('azure.cli.core.azclierror.logger.error', autospec=True)
     def test_handle_exception_httpoperationerror_no_error_key(self, mock_logger_error):
         # test error not in response
 
@@ -492,10 +491,10 @@ def test_handle_exception_httpoperationerror_no_error_key(self, mock_logger_erro
 
         # test behavior
         self.assertTrue(mock_logger_error.called)
-        self.assertEqual(mock.call(mock_http_error), mock_logger_error.call_args)
+        self.assertIn(str(mock.call(mock_http_error).args[0]), mock_logger_error.call_args.args[0])
         self.assertEqual(ex_result, 1)
 
-    @mock.patch('azure.cli.core.util.logger.error', autospec=True)
+    @mock.patch('azure.cli.core.azclierror.logger.error', autospec=True)
     def test_handle_exception_httpoperationerror_no_response_text(self, mock_logger_error):
         # test no response text
 
@@ -509,7 +508,7 @@ def test_handle_exception_httpoperationerror_no_response_text(self, mock_logger_
 
         # test behavior
         self.assertTrue(mock_logger_error.called)
-        self.assertEqual(mock.call(mock_http_error), mock_logger_error.call_args)
+        self.assertIn(str(mock.call(mock_http_error).args[0]), mock_logger_error.call_args.args[0])
         self.assertEqual(ex_result, 1)
 
     @staticmethod
diff --git a/src/azure-cli-core/azure/cli/core/util.py b/src/azure-cli-core/azure/cli/core/util.py
index 53c5355d0f5..4fcec496c37 100644
--- a/src/azure-cli-core/azure/cli/core/util.py
+++ b/src/azure-cli-core/azure/cli/core/util.py
@@ -17,6 +17,8 @@
 
 import six
 from six.moves.urllib.request import urlopen  # pylint: disable=import-error
+from azure.cli.core.azclierror import AzCLIErrorType
+from azure.cli.core.azclierror import AzCLIError
 from knack.log import get_logger
 from knack.util import CLIError, to_snake_case
 
@@ -28,8 +30,11 @@
 SSLERROR_TEMPLATE = ('Certificate verification failed. This typically happens when using Azure CLI behind a proxy '
                      'that intercepts traffic with a self-signed certificate. '
                      # pylint: disable=line-too-long
-                     'Please add this certificate to the trusted CA bundle. More info: https://docs.microsoft.com/en-us/cli/azure/use-cli-effectively#work-behind-a-proxy.\n\n'
-                     'Error detail: {}')
+                     'Please add this certificate to the trusted CA bundle. More info: https://docs.microsoft.com/en-us/cli/azure/use-cli-effectively#work-behind-a-proxy.')
+
+QUERY_REFERENCE = ("To learn more about --query, please visit: "
+                   "'https://docs.microsoft.com/cli/azure/query-azure-cli?view=azure-cli-latest'")
+
 
 _PROXYID_RE = re.compile(
     '(?i)/subscriptions/(?P<subscription>[^/]*)(/resourceGroups/(?P<resource_group>[^/]*))?'
@@ -50,7 +55,7 @@
 ]
 
 
-def handle_exception(ex):  # pylint: disable=too-many-return-statements
+def handle_exception(ex):  # pylint: disable=too-many-statements
     # For error code, follow guidelines at https://docs.python.org/2/library/sys.html#sys.exit,
     from jmespath.exceptions import JMESPathTypeError
     from msrestazure.azure_exceptions import CloudError
@@ -66,35 +71,44 @@ def handle_exception(ex):  # pylint: disable=too-many-return-statements
     logger.debug(traceback.format_exc())
 
     with CommandLoggerContext(logger):
-        if isinstance(ex, JMESPathTypeError):
-            logger.error("\nInvalid jmespath query supplied for `--query`:\n%s", ex)
-            logger.error("To learn more about --query, please visit: "
-                         "https://docs.microsoft.com/cli/azure/query-azure-cli?view=azure-cli-latest")
-            return 1
-        if isinstance(ex, (CLIError, CloudError, AzureException, AzureError)):
-            logger.error(ex.args[0])
+        error_msg = getattr(ex, 'message', str(ex))
+        exit_code = 1
+
+        if isinstance(ex, AzCLIError):
+            az_error = ex
+
+        elif isinstance(ex, JMESPathTypeError):
+            error_msg = "Invalid jmespath query supplied for `--query`: {}".format(error_msg)
+            az_error = AzCLIError(AzCLIErrorType.ArgumentParseError, error_msg)
+            az_error.set_recommendation(QUERY_REFERENCE)
+
+        elif isinstance(ex, ValidationError):
+            az_error = AzCLIError(AzCLIErrorType.ValidationError, error_msg)
+
+        # TODO: Fine-grained analysis to decide whether they are ValidationErrors
+        elif isinstance(ex, (CLIError, CloudError, AzureError)):
             try:
+                error_msg = ex.args[0]
                 for detail in ex.args[0].error.details:
-                    logger.error(detail)
-            except (AttributeError, TypeError):
-                pass
-            except:  # pylint: disable=bare-except
+                    error_msg += ('\n' + detail)
+            except Exception:  # pylint: disable=broad-except
                 pass
-            return ex.args[1] if len(ex.args) >= 2 else 1
-        if isinstance(ex, ValidationError):
-            logger.error('validation error: %s', ex)
-            return 1
-        if isinstance(ex, (ClientRequestError, SSLError)):
-            msg = str(ex)
-            if 'SSLError' in msg:
-                # SSL verification failed
-                logger.error("Request failed: %s", SSLERROR_TEMPLATE.format(msg))
-            else:
-                logger.error("Request failed: %s", ex)
-            return 1
-        if isinstance(ex, KeyboardInterrupt):
-            return 1
-        if isinstance(ex, HttpOperationError):
+            az_error = AzCLIError(AzCLIErrorType.ValidationError, error_msg)
+            exit_code = ex.args[1] if len(ex.args) >= 2 else 1
+
+        # TODO: Fine-grained analysis
+        elif isinstance(ex, AzureException):
+            az_error = AzCLIError(AzCLIErrorType.ServiceError, error_msg)
+            exit_code = ex.args[1] if len(ex.args) >= 2 else 1
+
+        # TODO: Fine-grained analysis
+        elif isinstance(ex, (ClientRequestError, SSLError)):
+            az_error = AzCLIError(AzCLIErrorType.ClientError, error_msg)
+            if 'SSLError' in error_msg:
+                az_error.set_recommendation(SSLERROR_TEMPLATE)
+
+        # TODO: Fine-grained analysis
+        elif isinstance(ex, HttpOperationError):
             try:
                 response_dict = json.loads(ex.response.text)
                 error = response_dict['error']
@@ -104,19 +118,29 @@ def handle_exception(ex):  # pylint: disable=too-many-return-statements
                 if isinstance(error, dict):
                     code = "{} - ".format(error.get('code', 'Unknown Code'))
                     message = error.get('message', ex)
-                    logger.error("%s%s", code, message)
+                    error_msg = "code: {}, {}".format(code, message)
                 else:
-                    logger.error(error)
+                    error_msg = error
 
             except (ValueError, KeyError):
-                logger.error(ex)
-            return 1
+                pass
+
+            az_error = AzCLIError(AzCLIErrorType.ServiceError, error_msg)
+
+        elif isinstance(ex, KeyboardInterrupt):
+            error_msg = 'Keyboard interrupt is captured.'
+            az_error = AzCLIError(AzCLIErrorType.ManualInterrupt, error_msg)
+
+        else:
+            error_msg = "The command failed with an unexpected error. Here is the traceback:"
+            az_error = AzCLIError(AzCLIErrorType.UnexpectedError, error_msg)
+            az_error.set_raw_exception(ex)
+            az_error.set_recommendation("To open an issue, please run: 'az feedback'")
 
-        logger.error("The command failed with an unexpected error. Here is the traceback:\n")
-        logger.exception(ex)
-        logger.warning("\nTo open an issue, please run: 'az feedback'")
+        az_error.print_error()
+        az_error.send_telemetry()
 
-        return 1
+        return exit_code
 
 
 # pylint: disable=inconsistent-return-statements
diff --git a/src/azure-cli/azure/cli/__main__.py b/src/azure-cli/azure/cli/__main__.py
index f26dceb4eb3..612737dba52 100644
--- a/src/azure-cli/azure/cli/__main__.py
+++ b/src/azure-cli/azure/cli/__main__.py
@@ -47,17 +47,13 @@ def cli_main(cli, args):
 
     exit_code = cli_main(az_cli, sys.argv[1:])
 
-    if exit_code and exit_code != 0:
-        if az_cli.result.error is not None and not telemetry.has_exceptions():
-            telemetry.set_exception(az_cli.result.error, fault_type='')
-        telemetry.set_failure()
-    else:
+    if exit_code == 0:
         telemetry.set_success()
 
     sys.exit(exit_code)
 
 except KeyboardInterrupt:
-    telemetry.set_user_fault('keyboard interrupt')
+    telemetry.set_user_fault('Keyboard interrupt is captured.')
     sys.exit(1)
 except SystemExit as ex:  # some code directly call sys.exit, this is to make sure command metadata is logged
     exit_code = ex.code if ex.code is not None else 1