Repository: Azure/WALinuxAgent
Pull request #2479 (2022-01-20 12:41:51)

Upgrade/Downgrade agent if Requested Version available
<!-- DO NOT DELETE THIS TEMPLATE -->

## Description
This PR contains the next set of changes for GA Versioning. Certain key points of this PR - 
- The agent wont accept downgrades for requested versions < current daemon version
- Blacklist the Agent on downgrade
- AutoUpdate as soon as a requested version is available rather than waiting for the wait times (4 hrs/24 hrs)
- Remove dead code that is not relevant anymore
- A bit of a re-write of the auto-update logic (with emphasis on making sure the current scenario is not broken)

Issue # <!-- if any -->
<!--
Please add an informative description that covers that changes made by the pull request. 
This checklist is used to make sure that common issues in a pull request are addressed.
This will expedite the process of getting your pull request merged and avoid extra work on your part to fix issues discovered during the review process.
-->

---

### PR information
- [ ] The title of the PR is clear and informative.
- [ ] There are a small number of commits, each of which has an informative message. This means that previously merged commits do not appear in the history of the PR. For information on cleaning up the commits in your pull request, [see this page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).
- [ ] If applicable, the PR references the bug/issue that it fixes in the description.
- [ ] New Unit tests were added for the changes made

### Quality of Code and Contribution Guidelines
- [ ] I have read the [contribution guidelines](https://github.com/Azure/WALinuxAgent/blob/master/.github/CONTRIBUTING.md).

codecov[bot]: # [Codecov](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure) Report
> Merging [#2479](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure) (fec32f6) into [develop](https://codecov.io/gh/Azure/WALinuxAgent/commit/ec0609d673fa31ee4ad803885a02dbb55411c77c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure) (ec0609d) will **increase** coverage by `0.05%`.
> The diff coverage is `92.52%`.

[![Impacted file tree graph](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479/graphs/tree.svg?width=650&height=150&src=pr&token=ae90E2KYCw&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure)](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure)

```diff
@@             Coverage Diff             @@
##           develop    #2479      +/-   ##
===========================================
+ Coverage    71.96%   72.01%   +0.05%     
===========================================
  Files          101      101              
  Lines        15063    15098      +35     
  Branches      2393     2399       +6     
===========================================
+ Hits         10840    10873      +33     
  Misses        3737     3737              
- Partials       486      488       +2     
```


| [Impacted Files](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure) | Coverage Δ | |
|---|---|---|
| [azurelinuxagent/common/version.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi92ZXJzaW9uLnB5) | `70.23% <ø> (-0.36%)` | :arrow_down: |
| [azurelinuxagent/ga/update.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure#diff-YXp1cmVsaW51eGFnZW50L2dhL3VwZGF0ZS5weQ==) | `89.02% <92.52%> (+0.21%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure). Last update [ec0609d...fec32f6](https://codecov.io/gh/Azure/WALinuxAgent/pull/2479?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Azure).


Patch:
diff --git a/azurelinuxagent/common/version.py b/azurelinuxagent/common/version.py
index e35e28b9..ff9c903b 100644
--- a/azurelinuxagent/common/version.py
+++ b/azurelinuxagent/common/version.py
@@ -298,6 +298,3 @@ def set_goal_state_agent():
 
 GOAL_STATE_AGENT_VERSION = set_goal_state_agent()
 
-
-def is_current_agent_installed():
-    return CURRENT_AGENT == AGENT_LONG_VERSION
diff --git a/azurelinuxagent/ga/update.py b/azurelinuxagent/ga/update.py
index 83db4be7..d795d26b 100644
--- a/azurelinuxagent/ga/update.py
+++ b/azurelinuxagent/ga/update.py
@@ -29,7 +29,6 @@ import sys
 import time
 import uuid
 import zipfile
-
 from datetime import datetime, timedelta
 
 import azurelinuxagent.common.conf as conf
@@ -38,34 +37,33 @@ import azurelinuxagent.common.utils.fileutil as fileutil
 import azurelinuxagent.common.utils.restutil as restutil
 import azurelinuxagent.common.utils.textutil as textutil
 from azurelinuxagent.common.agent_supported_feature import get_supported_feature_by_name, SupportedFeatureNames
-from azurelinuxagent.common.persist_firewall_rules import PersistFirewallRulesHandler
 from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
-
 from azurelinuxagent.common.event import add_event, initialize_event_logger_vminfo_common_parameters, \
     WALAEventOperation, EVENTS_DIRECTORY
 from azurelinuxagent.common.exception import ResourceGoneError, UpdateError, ExitException, AgentUpgradeExitException
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil, systemd
-from azurelinuxagent.common.protocol.restapi import VMAgentUpdateStatus, VMAgentUpdateStatuses, ExtHandlerPackageList
-from azurelinuxagent.common.protocol.util import get_protocol_util
+from azurelinuxagent.common.osutil.default import get_firewall_drop_command, \
+    get_accept_tcp_rule
+from azurelinuxagent.common.persist_firewall_rules import PersistFirewallRulesHandler
 from azurelinuxagent.common.protocol.hostplugin import HostPluginProtocol
+from azurelinuxagent.common.protocol.restapi import VMAgentUpdateStatus, VMAgentUpdateStatuses, ExtHandlerPackageList, \
+    VERSION_0
+from azurelinuxagent.common.protocol.util import get_protocol_util
 from azurelinuxagent.common.utils import shellutil
 from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
 from azurelinuxagent.common.utils.networkutil import AddFirewallRules
 from azurelinuxagent.common.utils.shellutil import CommandError
-from azurelinuxagent.common.version import AGENT_NAME, AGENT_VERSION, AGENT_DIR_PATTERN, CURRENT_AGENT,\
-    CURRENT_VERSION, DISTRO_NAME, DISTRO_VERSION, is_current_agent_installed, get_lis_version, \
-    has_logrotate, PY_VERSION_MAJOR, PY_VERSION_MINOR, PY_VERSION_MICRO
+from azurelinuxagent.common.version import AGENT_NAME, AGENT_DIR_PATTERN, CURRENT_AGENT, \
+    CURRENT_VERSION, DISTRO_NAME, DISTRO_VERSION, get_lis_version, \
+    has_logrotate, PY_VERSION_MAJOR, PY_VERSION_MINOR, PY_VERSION_MICRO, get_daemon_version
 from azurelinuxagent.ga.collect_logs import get_collect_logs_handler, is_log_collection_allowed
-from azurelinuxagent.ga.env import get_env_handler
 from azurelinuxagent.ga.collect_telemetry_events import get_collect_telemetry_events_handler
-
-from azurelinuxagent.ga.exthandlers import HandlerManifest, ExtHandlersHandler, list_agent_lib_directory, ExtensionStatusValue, ExtHandlerStatusValue
+from azurelinuxagent.ga.env import get_env_handler
+from azurelinuxagent.ga.exthandlers import HandlerManifest, ExtHandlersHandler, list_agent_lib_directory, \
+    ExtensionStatusValue, ExtHandlerStatusValue
 from azurelinuxagent.ga.monitor import get_monitor_handler
-
 from azurelinuxagent.ga.send_telemetry_events import get_send_telemetry_events_handler
-from azurelinuxagent.common.osutil.default import get_firewall_drop_command, \
-    get_accept_tcp_rule
 
 AGENT_ERROR_FILE = "error.json"  # File name for agent error record
 AGENT_MANIFEST_FILE = "HandlerManifest.json"
@@ -199,7 +197,8 @@ class UpdateHandler(object):
         if self.signal_handler is None:
             self.signal_handler = signal.signal(signal.SIGTERM, self.forward_signal)
 
-        latest_agent = self.get_latest_agent()
+        latest_agent = None if not conf.get_autoupdate_enabled() else self.get_latest_agent_greater_than_daemon(
+            daemon_version=CURRENT_VERSION)
         if latest_agent is None:
             logger.info(u"Installed Agent {0} is the most current agent", CURRENT_AGENT)
             agent_cmd = "python -u {0} -run-exthandlers".format(sys.argv[0])
@@ -396,8 +395,7 @@ class UpdateHandler(object):
                 time.sleep(self._goal_state_period)
 
         except AgentUpgradeExitException as exitException:
-            add_event(AGENT_NAME, op=WALAEventOperation.AgentUpgrade, version=CURRENT_VERSION, is_success=True,
-                      message=exitException.reason, log_event=False)
+            add_event(op=WALAEventOperation.AgentUpgrade, message=exitException.reason, log_event=False)
             logger.info(exitException.reason)
         except ExitException as exitException:
             logger.info(exitException.reason)
@@ -449,6 +447,80 @@ class UpdateHandler(object):
             return False
         return True
 
+    def __update_guest_agent(self, protocol):
+        """
+        This function checks for new Agent updates and raises AgentUpgradeExitException if available.
+        There are 2 different ways the agent checks for an update -
+            1) Requested Version is specified in the Goal State.
+                - In this case, the Agent will download the requested version and upgrade/downgrade instantly.
+            2) No requested version.
+                - In this case, the agent will periodically check (1 hr) for new agent versions in GA Manifest.
+                - If available, it will download all versions > CURRENT_VERSION.
+                - Depending on the highest version > CURRENT_VERSION,
+                  the agent will update within 4 hrs (for a Hotfix update) or 24 hrs (for a Normal update)
+        """
+
+        def log_next_update_time():
+            next_normal_time, next_hotfix_time = self.__get_next_upgrade_times()
+            upgrade_type = self.__get_agent_upgrade_type(available_agent)
+            next_time = next_hotfix_time if upgrade_type == AgentUpgradeType.Hotfix else next_normal_time
+            message_ = "Discovered new {0} upgrade {1}; Will upgrade on or after {2}".format(
+                upgrade_type, available_agent.name,
+                datetime.utcfromtimestamp(next_time).strftime(logger.Logger.LogTimeFormatInUTC))
+            add_event(AGENT_NAME, op=WALAEventOperation.AgentUpgrade, version=CURRENT_VERSION, is_success=True,
+                      message=message_, log_event=False)
+            logger.info(message_)
+
+        def handle_updates_for_requested_version():
+            if requested_version < CURRENT_VERSION:
+                prefix = "downgrade"
+                # In case of a downgrade, we blacklist the current agent to avoid starting it back up ever again
+                # (the expectation here being that if RSM is asking us to a downgrade,
+                # there's a good reason for not wanting the current version).
+                try:
+                    # We should always have an agent directory for the CURRENT_VERSION
+                    # (unless the CURRENT_VERSION == daemon version, but since we don't support downgrading
+                    # below daemon version, we will never reach this code path if that's the scenario)
+                    current_agent = next(agent for agent in self.agents if agent.version == CURRENT_VERSION)
+                    logger.info(
+                        "Blacklisting the agent {0} since a downgrade was requested in the GoalState, "
+                        "suggesting that we really don't want to execute any extensions using this version".format(
+                            CURRENT_VERSION))
+                    current_agent.mark_failure(is_fatal=True)
+                except StopIteration:
+                    logger.warn(
+                        "Could not find a matching agent with current version {0} to blacklist, skipping it".format(
+                            CURRENT_VERSION))
+            else:
+                # In case of an upgrade, we don't need to blacklist anything as the daemon will automatically
+                # start the next available highest version which would be the requested version
+                prefix = "upgrade"
+            raise AgentUpgradeExitException(
+                "Exiting current process to {0} to the request Agent version {1}".format(prefix, requested_version))
+
+        # Ignore new agents if updating is disabled
+        if not conf.get_autoupdate_enabled():
+            return False
+
+        if self._download_agent_if_upgrade_available(protocol):
+            # The call to get_latest_agent_greater_than_daemon() also finds all agents in directory and sets the self.agents property.
+            # This state is used to find the GuestAgent object with the current version later if requested version is available in last GS.
+            available_agent = self.get_latest_agent_greater_than_daemon()
+            requested_version, _ = self.__get_requested_version_and_manifest_from_last_gs(protocol)
+            if requested_version is not None:
+                # If requested version specified, upgrade/downgrade to the specified version instantly as this is
+                # driven by the goal state (as compared to the agent periodically checking for new upgrades every hour)
+                handle_updates_for_requested_version()
+            elif available_agent is None:
+                # Legacy behavior: The current agent can become unavailable and needs to be reverted.
+                # In that case, self._upgrade_available() returns True and available_agent would be None. Handling it here.
+                raise AgentUpgradeExitException(
+                    "Agent {0} is reverting to the installed agent -- exiting".format(CURRENT_AGENT))
+            else:
+                log_next_update_time()
+
+        self.__upgrade_agent_if_permitted()
+
     def __goal_state_updated(self, incarnation):
         """
         This function returns if the Goal State updated.
@@ -459,6 +531,7 @@ class UpdateHandler(object):
         return incarnation != self.last_incarnation
 
     def _process_goal_state(self, exthandlers_handler, remote_access_handler):
+
         protocol = exthandlers_handler.protocol
         if not self._try_update_goal_state(protocol):
             self._heartbeat_update_goal_state_error_count += 1
@@ -466,25 +539,8 @@ class UpdateHandler(object):
             self._report_status(exthandlers_handler, incarnation_changed=False)
             return
 
-        if self._check_and_download_agent_if_upgrade_available(protocol):
-            available_agent = self.get_latest_agent()
-            # Legacy behavior: The current agent can become unavailable and needs to be reverted.
-            # In that case, self._upgrade_available() returns True and available_agent would be None. Handling it here.
-            if available_agent is None:
-                raise AgentUpgradeExitException("Agent {0} is reverting to the installed agent -- exiting".format(CURRENT_AGENT))
-            else:
-                next_normal_time, next_hotfix_time = self.__get_next_upgrade_times()
-                upgrade_type = self.__get_agent_upgrade_type(available_agent)
-                next_time = next_hotfix_time if upgrade_type == AgentUpgradeType.Hotfix else next_normal_time
-                message = "Discovered new {0} upgrade {1}; Will upgrade on or after {2}".format(
-                    upgrade_type, available_agent.name,
-                    datetime.utcfromtimestamp(next_time).strftime(logger.Logger.LogTimeFormatInUTC))
-                add_event(AGENT_NAME, op=WALAEventOperation.AgentUpgrade, version=CURRENT_VERSION, is_success=True,
-                          message=message, log_event=False)
-                logger.info(message)
-
-        self.__upgrade_agent_if_permitted()
-
+        # Update the Guest Agent if a new version is available
+        self.__update_guest_agent(protocol)
         incarnation = protocol.get_incarnation()
 
         try:
@@ -507,8 +563,7 @@ class UpdateHandler(object):
         finally:
             self.last_incarnation = incarnation
 
-    @staticmethod
-    def __get_vmagent_update_status(protocol, incarnation_changed):
+    def __get_vmagent_update_status(self, protocol, incarnation_changed):
         """
         This function gets the VMAgent update status as per the last GoalState.
         Returns: None if the last GS does not ask for requested version else VMAgentUpdateStatus
@@ -519,20 +574,14 @@ class UpdateHandler(object):
         update_status = None
 
         try:
-            agent_manifests, _ = protocol.get_vmagent_manifests()
-
-            try:
-                # Expectation here is that there will only be one manifest per family passed down from CRP
-                # (already verified during validations), we pick the first matching one here.
-                manifest = next(m for m in agent_manifests if m.family == conf.get_autoupdate_gafamily())
-            except StopIteration:
-                if incarnation_changed:
-                    logger.info("Unable to report update status as no matching manifest found for family: {0}".format(
-                        conf.get_autoupdate_gafamily()))
+            requested_version, manifest = self.__get_requested_version_and_manifest_from_last_gs(protocol)
+            if manifest is None and incarnation_changed:
+                logger.info("Unable to report update status as no matching manifest found for family: {0}".format(
+                    conf.get_autoupdate_gafamily()))
                 return None
 
-            if manifest.is_requested_version_specified:
-                if CURRENT_VERSION == manifest.requested_version:
+            if requested_version is not None:
+                if CURRENT_VERSION == requested_version:
                     status = VMAgentUpdateStatuses.Success
                     code = 0
                 else:
@@ -609,20 +658,30 @@ class UpdateHandler(object):
                 sys.exit(0)
         return
 
-    def get_latest_agent(self):
+    @staticmethod
+    def __get_daemon_version_for_update():
+        daemon_version = get_daemon_version()
+        if daemon_version != FlexibleVersion(VERSION_0):
+            return daemon_version
+        # We return 0.0.0.0 if daemon version is not specified. In that case,
+        # use the min version as 2.2.53 as we started setting the daemon version starting 2.2.53.
+        return FlexibleVersion("2.2.53")
+
+    def get_latest_agent_greater_than_daemon(self, daemon_version=None):
         """
         If autoupdate is enabled, return the most current, downloaded,
-        non-blacklisted agent which is not the current version (if any).
+        non-blacklisted agent which is not the current version (if any) and is greater than the `daemon_version`.
         Otherwise, return None (implying to use the installed agent).
+        If `daemon_version` is None, we fetch it from the environment variable set by the DaemonHandler
         """
 
-        if not conf.get_autoupdate_enabled():
-            return None
-
         self._find_agents()
+        daemon_version = self.__get_daemon_version_for_update() if daemon_version is None else daemon_version
+
+        # Fetch the downloaded agents that are different from the current version and greater than the daemon version
         available_agents = [agent for agent in self.agents
                             if agent.is_available
-                            and agent.version > FlexibleVersion(AGENT_VERSION)]
+                            and agent.version != CURRENT_VERSION and agent.version > daemon_version]
 
         return available_agents[0] if len(available_agents) >= 1 else None
 
@@ -770,7 +829,7 @@ class UpdateHandler(object):
         Load all non-blacklisted agents currently on disk.
         """
         try:
-            self._set_agents(self._load_agents())
+            self._set_and_sort_agents(self._load_agents())
             self._filter_blacklisted_agents()
         except Exception as e:
             logger.warn(u"Exception occurred loading available agents: {0}", ustr(e))
@@ -815,17 +874,6 @@ class UpdateHandler(object):
 
         return fileutil.read_file(conf.get_agent_pid_file_path()) != ustr(parent_pid)
 
-    def _is_version_eligible(self, version):
-        # Ensure the installed version is always eligible
-        if version == CURRENT_VERSION and is_current_agent_installed():
-            return True
-
-        for agent in self.agents:
-            if agent.version == version:
-                return agent.is_available
-
-        return False
-
     def _load_agents(self):
         path = os.path.join(conf.get_lib_dir(), "{0}-*".format(AGENT_NAME))
         return [GuestAgent(path=agent_dir)
@@ -867,7 +915,7 @@ class UpdateHandler(object):
                 logger.warn(u"Purging {0} raised exception: {1}", agent_path, ustr(e))
         return
 
-    def _set_agents(self, agents=None):
+    def _set_and_sort_agents(self, agents=None):
         if agents is None:
             agents = []
         self.agents = agents
@@ -910,7 +958,24 @@ class UpdateHandler(object):
                 str(e))
         return
 
-    def _check_and_download_agent_if_upgrade_available(self, protocol, base_version=CURRENT_VERSION):
+    @staticmethod
+    def __get_requested_version_and_manifest_from_last_gs(protocol):
+        """
+        Get the requested version and corresponding manifests from last GS if supported
+        Returns: (Requested Version, Manifest) if supported and available
+                 (None, None) if no manifests found in the last GS
+                 (None, manifest) if not supported or not specified in GS
+        """
+        family = conf.get_autoupdate_gafamily()
+        manifest_list, _ = protocol.get_vmagent_manifests()
+        manifests = [m for m in manifest_list if m.family == family and len(m.uris) > 0]
+        if len(manifests) == 0:
+            return None, None
+        if conf.get_enable_ga_versioning() and manifests[0].is_requested_version_specified:
+            return manifests[0].requested_version, manifests[0]
+        return None, manifests[0]
+
+    def _download_agent_if_upgrade_available(self, protocol, base_version=CURRENT_VERSION):
         """
         This function downloads the new agent if an update is available.
         If a requested version is available in goal state, then only that version is downloaded (new-update model)
@@ -919,46 +984,67 @@ class UpdateHandler(object):
         return: True if current agent is no longer available or an agent with a higher version number is available
         else False
         """
-        # Ignore new agents if updating is disabled
-        if not conf.get_autoupdate_enabled():
-            return False
 
-        def report_error(msg_, version=CURRENT_VERSION):
+        def report_error(msg_, version_=CURRENT_VERSION, op=WALAEventOperation.Download):
             logger.warn(msg_)
-            add_event(AGENT_NAME, op=WALAEventOperation.Download, version=version, is_success=False, message=msg_)
+            add_event(AGENT_NAME, op=op, version=version_, is_success=False, message=msg_, log_event=False)
+
+        def can_proceed_with_requested_version():
+            if not gs_updated:
+                # If incarnation didn't change, don't process anything.
+                return False
+
+            # With the new model, we will get a new GS when CRP wants us to auto-update using required version.
+            # If there's no new incarnation, don't proceed with anything
+            msg_ = "Found requested version in manifest: {0} for incarnation: {1}".format(
+                requested_version, incarnation)
+            logger.info(msg_)
+            add_event(AGENT_NAME, op=WALAEventOperation.AgentUpgrade, is_success=True, message=msg_, log_event=False)
+
+            if requested_version < daemon_version:
+                # Don't process the update if the requested version is lesser than daemon version,
+                # as we don't support downgrades below daemon versions.
+                report_error(
+                    "Can't process the upgrade as the requested version: {0} is < current daemon version: {1}".format(
+                        requested_version, daemon_version), op=WALAEventOperation.AgentUpgrade)
+                return False
+
+            return True
+
+        def agent_upgrade_time_elapsed(now_):
+            if self.last_attempt_time is not None:
+                next_attempt_time = self.last_attempt_time + conf.get_autoupdate_frequency()
+            else:
+                next_attempt_time = now_
+            if next_attempt_time > now_:
+                return False
+            return True
 
         family = conf.get_autoupdate_gafamily()
-        incarnation_changed = False
+        gs_updated = False
+        daemon_version = self.__get_daemon_version_for_update()
         try:
             # Fetch the agent manifests from the latest Goal State
-            manifest_list, incarnation = protocol.get_vmagent_manifests()
-            incarnation_changed = self.__goal_state_updated(incarnation)
-            manifests = [m for m in manifest_list if m.family == family and len(m.uris) > 0]
-            if len(manifests) == 0:
+            incarnation = protocol.get_incarnation()
+            gs_updated = self.__goal_state_updated(incarnation)
+            requested_version, manifest = self.__get_requested_version_and_manifest_from_last_gs(protocol)
+            if manifest is None:
                 logger.verbose(
                     u"No manifest links found for agent family: {0} for incarnation: {1}, skipping update check".format(
                         family, incarnation))
                 return False
         except Exception as err:
             # If there's some issues in fetching the agent manifests, report it only on incarnation change
-            if incarnation_changed:
-                msg = u"Exception retrieving agent manifests: {0}".format(textutil.format_exception(err))
+            msg = u"Exception retrieving agent manifests: {0}".format(textutil.format_exception(err))
+            if gs_updated:
                 report_error(msg)
+            else:
+                logger.verbose(msg)
             return False
 
-        requested_version = None
-        if conf.get_enable_ga_versioning() and manifests[0].is_requested_version_specified:
+        if requested_version is not None:
             # If GA versioning is enabled and requested version present in GS, and it's a new GS, follow new logic
-            if incarnation_changed:
-                # With the new model, we will get a new GS when CRP wants us to auto-update using required version.
-                # If there's no new incarnation, don't proceed with anything
-                requested_version = manifests[0].requested_version
-                msg = "Found requested version in manifest: {0} for incarnation: {1}".format(
-                    requested_version, incarnation)
-                logger.info(msg)
-                add_event(AGENT_NAME, op=WALAEventOperation.AgentUpgrade, is_success=True, message=msg)
-            else:
-                # If incarnation didn't change, don't process anything.
+            if not can_proceed_with_requested_version():
                 return False
         else:
             # If no requested version specified in the Goal State, follow the old auto-update logic
@@ -966,16 +1052,11 @@ class UpdateHandler(object):
             # If any subsequent goal state does not contain requested version, this timer will start then, and we will
             # download all versions available in PIR and auto-update to the highest available version on that goal state.
             now = time.time()
-            if self.last_attempt_time is not None:
-                next_attempt_time = self.last_attempt_time + conf.get_autoupdate_frequency()
-            else:
-                next_attempt_time = now
-            if next_attempt_time > now:
+            if not agent_upgrade_time_elapsed(now):
                 return False
 
             logger.info("No requested version specified, checking for all versions for agent update (family: {0})",
                         family)
-
             self.last_attempt_time = now
 
         try:
@@ -988,45 +1069,52 @@ class UpdateHandler(object):
             # In this case, no need to even fetch the GA family manifest as we don't need to download any agent.
             if requested_version is not None and requested_version == CURRENT_VERSION:
                 packages_to_download = []
-                logger.info("The requested version is running as the current version: {0}".format(requested_version))
+                msg = "The requested version is running as the current version: {0}".format(requested_version)
+                logger.info(msg)
+                add_event(AGENT_NAME, op=WALAEventOperation.AgentUpgrade, is_success=True, message=msg)
             else:
-                pkg_list = protocol.get_vmagent_pkgs(manifests[0])
+                pkg_list = protocol.get_vmagent_pkgs(manifest)
                 packages_to_download = pkg_list.versions
 
             # Verify the requested version is in GA family manifest (if specified)
             if requested_version is not None and requested_version != CURRENT_VERSION:
-                package_found = False
                 for pkg in pkg_list.versions:
                     if FlexibleVersion(pkg.version) == requested_version:
                         # Found a matching package, only download that one
                         packages_to_download = [pkg]
-                        package_found = True
                         break
-
-                if not package_found:
+                else:
                     msg = "No matching package found in the agent manifest for requested version: {0} in incarnation: {1}, skipping agent update".format(
                         requested_version, incarnation)
-                    report_error(msg, version=requested_version)
+                    report_error(msg, version_=requested_version)
                     return False
 
             # Set the agents to those available for download at least as current as the existing agent
             # or to the requested version (if specified)
             host = self._get_host_plugin(protocol=protocol)
-            self._set_agents([GuestAgent(pkg=pkg, host=host) for pkg in packages_to_download])
+            self._set_and_sort_agents([GuestAgent(pkg=pkg, host=host) for pkg in packages_to_download])
 
             # Remove from disk any agent no longer needed in the VM.
-            # If requested version is provided, this would delete all other agents present on the VM except the
-            # current one and the requested one if they're different, and only the current one if same.
+            # If requested version is provided, this would delete all other agents present on the VM except -
+            #   - the current version and the requested version if requested version != current version
+            #   - only the current version if requested version == current version
             # Note:
             #  The code leaves on disk available, but blacklisted, agents to preserve the state.
             #  Otherwise, those agents could be downloaded again and inappropriately retried.
             self._purge_agents()
             self._filter_blacklisted_agents()
 
-            # Return True if current agent is no longer available or an
-            # agent with a higher version number is available
-            return not self._is_version_eligible(base_version) \
-                   or (len(self.agents) > 0 and self.agents[0].version > base_version)
+            # If there are no agents available to upgrade/downgrade to, return False
+            if len(self.agents) == 0:
+                return False
+
+            if requested_version is not None:
+                # In case of requested version, return True if an agent with a different version number than the
+                # current version is available that is higher than the current daemon version
+                return self.agents[0].version != base_version and self.agents[0].version > daemon_version
+            else:
+                # Else, return True if the highest agent is > base_version (CURRENT_VERSION)
+                return self.agents[0].version > base_version
 
         except Exception as err:
             msg = u"Exception downloading agents for update: {0}".format(textutil.format_exception(err))
@@ -1232,8 +1320,8 @@ class UpdateHandler(object):
         self._last_hotfix_upgrade_time = now if next_hotfix_time <= now else self._last_hotfix_upgrade_time
         self._last_normal_upgrade_time = now if next_normal_time <= now else self._last_normal_upgrade_time
 
-        available_agent = self.get_latest_agent()
-        if available_agent is None:
+        available_agent = self.get_latest_agent_greater_than_daemon()
+        if available_agent is None or available_agent.version <= CURRENT_VERSION:
             logger.verbose("No agent upgrade discovered")
             return
 
@@ -1341,7 +1429,10 @@ class GuestAgent(object):
             self.error.mark_failure(is_fatal=is_fatal)
             self.error.save()
             if self.error.is_blacklisted:
-                logger.warn(u"Agent {0} is permanently blacklisted", self.name)
+                msg = u"Agent {0} is permanently blacklisted".format(self.name)
+                logger.warn(msg)
+                add_event(op=WALAEventOperation.AgentBlacklisted, is_success=False, message=msg, log_event=False,
+                          version=self.version)
         except Exception as e:
             logger.warn(u"Agent {0} failed recording error state: {1}", self.name, ustr(e))
 


Test patch:
diff --git a/tests/ga/test_update.py b/tests/ga/test_update.py
index e40c05ea..e1dbe55b 100644
--- a/tests/ga/test_update.py
+++ b/tests/ga/test_update.py
@@ -42,7 +42,8 @@ from azurelinuxagent.common.utils import fileutil, restutil, textutil
 from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
 from azurelinuxagent.common.utils.networkutil import FirewallCmdDirectCommands, AddFirewallRules
 from azurelinuxagent.common.version import AGENT_PKG_GLOB, AGENT_DIR_GLOB, AGENT_NAME, AGENT_DIR_PATTERN, \
-    AGENT_VERSION, CURRENT_AGENT, CURRENT_VERSION
+    AGENT_VERSION, CURRENT_AGENT, CURRENT_VERSION, set_daemon_version, \
+    __DAEMON_VERSION_ENV_VARIABLE as DAEMON_VERSION_ENV_VARIABLE
 from azurelinuxagent.ga.exthandlers import ExtHandlersHandler, ExtHandlerInstance, HandlerEnvironment, ExtensionStatusValue
 from azurelinuxagent.ga.update import GuestAgent, GuestAgentError, MAX_FAILURE, AGENT_MANIFEST_FILE, \
     get_update_handler, ORPHAN_POLL_INTERVAL, AGENT_PARTITION_FILE, AGENT_ERROR_FILE, ORPHAN_WAIT_INTERVAL, \
@@ -943,7 +944,7 @@ class TestUpdate(UpdateTestCase):
     def _test_evaluate_agent_health(self, child_agent_index=0):
         self.prepare_agents()
 
-        latest_agent = self.update_handler.get_latest_agent()
+        latest_agent = self.update_handler.get_latest_agent_greater_than_daemon()
         self.assertTrue(latest_agent.is_available)
         self.assertFalse(latest_agent.is_blacklisted)
         self.assertTrue(len(self.update_handler.agents) > 1)
@@ -982,7 +983,7 @@ class TestUpdate(UpdateTestCase):
     def test_filter_blacklisted_agents(self):
         self.prepare_agents()
 
-        self.update_handler._set_agents([GuestAgent(path=path) for path in self.agent_dirs()])
+        self.update_handler._set_and_sort_agents([GuestAgent(path=path) for path in self.agent_dirs()])
         self.assertEqual(len(self.agent_dirs()), len(self.update_handler.agents))
 
         kept_agents = self.update_handler.agents[::2]
@@ -1029,7 +1030,7 @@ class TestUpdate(UpdateTestCase):
     def test_get_latest_agent(self):
         latest_version = self.prepare_agents()
 
-        latest_agent = self.update_handler.get_latest_agent()
+        latest_agent = self.update_handler.get_latest_agent_greater_than_daemon()
         self.assertEqual(len(self._get_agents(self.tmp_dir)), len(self.update_handler.agents))
         self.assertEqual(latest_version, latest_agent.version)
 
@@ -1038,24 +1039,24 @@ class TestUpdate(UpdateTestCase):
         self.assertFalse(self._test_upgrade_available(
             versions=self.agent_versions(),
             count=1))
-        self.assertEqual(None, self.update_handler.get_latest_agent())
+        self.assertEqual(None, self.update_handler.get_latest_agent_greater_than_daemon())
 
     def test_get_latest_agent_no_updates(self):
-        self.assertEqual(None, self.update_handler.get_latest_agent())
+        self.assertEqual(None, self.update_handler.get_latest_agent_greater_than_daemon())
 
     def test_get_latest_agent_skip_updates(self):
         conf.get_autoupdate_enabled = Mock(return_value=False)
-        self.assertEqual(None, self.update_handler.get_latest_agent())
+        self.assertEqual(None, self.update_handler.get_latest_agent_greater_than_daemon())
 
     def test_get_latest_agent_skips_unavailable(self):
         self.prepare_agents()
-        prior_agent = self.update_handler.get_latest_agent()
+        prior_agent = self.update_handler.get_latest_agent_greater_than_daemon()
 
         latest_version = self.prepare_agents(count=self.agent_count() + 1, is_available=False)
         latest_path = os.path.join(self.tmp_dir, "{0}-{1}".format(AGENT_NAME, latest_version))
         self.assertFalse(GuestAgent(latest_path).is_available)
 
-        latest_agent = self.update_handler.get_latest_agent()
+        latest_agent = self.update_handler.get_latest_agent_greater_than_daemon()
         self.assertTrue(latest_agent.version < latest_version)
         self.assertEqual(latest_agent.version, prior_agent.version)
 
@@ -1100,34 +1101,6 @@ class TestUpdate(UpdateTestCase):
         with patch('os.getppid', return_value=42):
             self.assertTrue(self.update_handler._is_orphaned)
 
-    def test_is_version_available(self):
-        self.prepare_agents(is_available=True)
-        self.update_handler.agents = self.agents()
-
-        for agent in self.agents():
-            self.assertTrue(self.update_handler._is_version_eligible(agent.version))
-
-    @patch("azurelinuxagent.ga.update.is_current_agent_installed", return_value=False)
-    def test_is_version_available_rejects(self, mock_current):  # pylint: disable=unused-argument
-        self.prepare_agents(is_available=True)
-        self.update_handler.agents = self.agents()
-
-        self.update_handler.agents[0].mark_failure(is_fatal=True)
-        self.assertFalse(self.update_handler._is_version_eligible(self.agents()[0].version))
-
-    @patch("azurelinuxagent.ga.update.is_current_agent_installed", return_value=True)
-    def test_is_version_available_accepts_current(self, mock_current):  # pylint: disable=unused-argument
-        self.update_handler.agents = []
-        self.assertTrue(self.update_handler._is_version_eligible(CURRENT_VERSION))
-
-    @patch("azurelinuxagent.ga.update.is_current_agent_installed", return_value=False)
-    def test_is_version_available_rejects_by_default(self, mock_current):  # pylint: disable=unused-argument
-        self.prepare_agents()
-        self.update_handler.agents = []
-
-        v = self.agents()[0].version
-        self.assertFalse(self.update_handler._is_version_eligible(v))
-
     def test_purge_agents(self):
         self.prepare_agents()
         self.update_handler._find_agents()
@@ -1191,7 +1164,7 @@ class TestUpdate(UpdateTestCase):
     def test_run_latest(self):
         self.prepare_agents()
 
-        agent = self.update_handler.get_latest_agent()
+        agent = self.update_handler.get_latest_agent_greater_than_daemon()
         args, kwargs = self._test_run_latest()
         args = args[0]
         cmds = textutil.safe_shlex_split(agent.get_agent_cmd())
@@ -1209,8 +1182,8 @@ class TestUpdate(UpdateTestCase):
     def test_run_latest_passes_child_args(self):
         self.prepare_agents()
 
-        agent = self.update_handler.get_latest_agent()  # pylint: disable=unused-variable
-        args, kwargs = self._test_run_latest(child_args="AnArgument")  # pylint: disable=unused-variable
+        self.update_handler.get_latest_agent_greater_than_daemon()
+        args, _ = self._test_run_latest(child_args="AnArgument")
         args = args[0]
 
         self.assertTrue(len(args) > 1)
@@ -1252,7 +1225,7 @@ class TestUpdate(UpdateTestCase):
         self.assertEqual(1, mock_time.sleep_interval)
 
     def test_run_latest_defaults_to_current(self):
-        self.assertEqual(None, self.update_handler.get_latest_agent())
+        self.assertEqual(None, self.update_handler.get_latest_agent_greater_than_daemon())
 
         args, kwargs = self._test_run_latest()
 
@@ -1287,12 +1260,12 @@ class TestUpdate(UpdateTestCase):
         # logger.add_logger_appender(logger.AppenderType.STDOUT)
         self.prepare_agents()
 
-        latest_agent = self.update_handler.get_latest_agent()
+        latest_agent = self.update_handler.get_latest_agent_greater_than_daemon()
         self.assertTrue(latest_agent.is_available)
         self.assertEqual(0.0, latest_agent.error.last_failure)
         self.assertEqual(0, latest_agent.error.failure_count)
 
-        with patch('azurelinuxagent.ga.update.UpdateHandler.get_latest_agent', return_value=latest_agent):
+        with patch('azurelinuxagent.ga.update.UpdateHandler.get_latest_agent_greater_than_daemon', return_value=latest_agent):
             self._test_run_latest(mock_child=ChildMock(return_value=1))
 
         self.assertTrue(latest_agent.is_blacklisted)
@@ -1303,12 +1276,12 @@ class TestUpdate(UpdateTestCase):
     def test_run_latest_exception_blacklists(self):
         self.prepare_agents()
 
-        latest_agent = self.update_handler.get_latest_agent()
+        latest_agent = self.update_handler.get_latest_agent_greater_than_daemon()
         self.assertTrue(latest_agent.is_available)
         self.assertEqual(0.0, latest_agent.error.last_failure)
         self.assertEqual(0, latest_agent.error.failure_count)
 
-        with patch('azurelinuxagent.ga.update.UpdateHandler.get_latest_agent', return_value=latest_agent):
+        with patch('azurelinuxagent.ga.update.UpdateHandler.get_latest_agent_greater_than_daemon', return_value=latest_agent):
             self._test_run_latest(mock_child=ChildMock(side_effect=Exception("Force blacklisting")))
 
         self.assertFalse(latest_agent.is_available)
@@ -1319,12 +1292,12 @@ class TestUpdate(UpdateTestCase):
     def test_run_latest_exception_does_not_blacklist_if_terminating(self):
         self.prepare_agents()
 
-        latest_agent = self.update_handler.get_latest_agent()
+        latest_agent = self.update_handler.get_latest_agent_greater_than_daemon()
         self.assertTrue(latest_agent.is_available)
         self.assertEqual(0.0, latest_agent.error.last_failure)
         self.assertEqual(0, latest_agent.error.failure_count)
 
-        with patch('azurelinuxagent.ga.update.UpdateHandler.get_latest_agent', return_value=latest_agent):
+        with patch('azurelinuxagent.ga.update.UpdateHandler.get_latest_agent_greater_than_daemon', return_value=latest_agent):
             self.update_handler.is_running = False
             self._test_run_latest(mock_child=ChildMock(side_effect=Exception("Attempt blacklisting")))
 
@@ -1352,7 +1325,7 @@ class TestUpdate(UpdateTestCase):
             with open(error_file_path, 'w') as f:
                 f.write("")
 
-        latest_agent = self.update_handler.get_latest_agent()
+        latest_agent = self.update_handler.get_latest_agent_greater_than_daemon()
         self.assertEqual(latest_agent.version, dst_ver, "Latest agent version is invalid")
 
     def _test_run(self, invocations=1, calls=1, enable_updates=False, sleep_interval=(6,)):
@@ -1404,7 +1377,7 @@ class TestUpdate(UpdateTestCase):
         self._test_run()
 
     def test_run_stops_if_update_available(self):
-        self.update_handler._check_and_download_agent_if_upgrade_available = Mock(return_value=True)
+        self.update_handler._download_agent_if_upgrade_available = Mock(return_value=True)
         self._test_run(invocations=0, calls=0, enable_updates=True)
 
     def test_run_stops_if_orphaned(self):
@@ -1416,7 +1389,7 @@ class TestUpdate(UpdateTestCase):
         self.assertFalse(os.path.isfile(self.update_handler._sentinel_file_path()))
 
     def test_run_leaves_sentinel_on_unsuccessful_exit(self):
-        self.update_handler._check_and_download_agent_if_upgrade_available = Mock(side_effect=Exception)
+        self.update_handler._download_agent_if_upgrade_available = Mock(side_effect=Exception)
         self._test_run(invocations=1, calls=0, enable_updates=True)
         self.assertTrue(os.path.isfile(self.update_handler._sentinel_file_path()))
 
@@ -1428,14 +1401,14 @@ class TestUpdate(UpdateTestCase):
     def test_set_agents_sets_agents(self):
         self.prepare_agents()
 
-        self.update_handler._set_agents([GuestAgent(path=path) for path in self.agent_dirs()])
+        self.update_handler._set_and_sort_agents([GuestAgent(path=path) for path in self.agent_dirs()])
         self.assertTrue(len(self.update_handler.agents) > 0)
         self.assertEqual(len(self.agent_dirs()), len(self.update_handler.agents))
 
     def test_set_agents_sorts_agents(self):
         self.prepare_agents()
 
-        self.update_handler._set_agents([GuestAgent(path=path) for path in self.agent_dirs()])
+        self.update_handler._set_and_sort_agents([GuestAgent(path=path) for path in self.agent_dirs()])
 
         v = FlexibleVersion("100000")
         for a in self.update_handler.agents:
@@ -1487,7 +1460,7 @@ class TestUpdate(UpdateTestCase):
         self.update_handler.protocol_util = protocol
         conf.get_autoupdate_gafamily = Mock(return_value=protocol.family)
 
-        return self.update_handler._check_and_download_agent_if_upgrade_available(protocol, base_version=base_version)
+        return self.update_handler._download_agent_if_upgrade_available(protocol, base_version=base_version)
 
     def test_upgrade_available_returns_true_on_first_use(self):
         self.assertTrue(self._test_upgrade_available())
@@ -1500,7 +1473,7 @@ class TestUpdate(UpdateTestCase):
             self.update_handler.protocol_util = protocol
             with patch('azurelinuxagent.common.logger.warn') as mock_logger:
                 with patch('tests.ga.test_update.ProtocolMock.get_vmagent_pkgs', side_effect=ProtocolError):
-                    self.assertFalse(self.update_handler._check_and_download_agent_if_upgrade_available(protocol, base_version=CURRENT_VERSION))
+                    self.assertFalse(self.update_handler._download_agent_if_upgrade_available(protocol, base_version=CURRENT_VERSION))
                     self.assertEqual(0, mock_logger.call_count)
 
     def test_upgrade_available_includes_old_agents(self):
@@ -1529,28 +1502,19 @@ class TestUpdate(UpdateTestCase):
             agent_versions.append(CURRENT_VERSION)
         self.assertEqual(agent_versions, self.agent_versions())
 
-    def test_update_available_returns_true_if_current_gets_blacklisted(self):
-        self.update_handler._is_version_eligible = Mock(return_value=False)
-        self.assertTrue(self._test_upgrade_available())
-
     def test_upgrade_available_skips_if_too_frequent(self):
         conf.get_autoupdate_frequency = Mock(return_value=10000)
         self.update_handler.last_attempt_time = time.time()
         self.assertFalse(self._test_upgrade_available())
 
-    def test_upgrade_available_skips_if_when_no_new_versions(self):
+    def test_upgrade_available_skips_when_no_new_versions(self):
         self.prepare_agents()
         base_version = self.agent_versions()[0] + 1
-        self.update_handler._is_version_eligible = lambda x: x == base_version
         self.assertFalse(self._test_upgrade_available(base_version=base_version))
 
     def test_upgrade_available_skips_when_no_versions(self):
         self.assertFalse(self._test_upgrade_available(protocol=ProtocolMock()))
 
-    def test_upgrade_available_skips_when_updates_are_disabled(self):
-        conf.get_autoupdate_enabled = Mock(return_value=False)
-        self.assertFalse(self._test_upgrade_available())
-
     def test_upgrade_available_sorts(self):
         self.prepare_agents()
         self._test_upgrade_available()
@@ -1584,7 +1548,7 @@ class TestUpdate(UpdateTestCase):
         before an update is found, this test attempts to ensure that
         behavior never changes.
         """
-        self.update_handler._check_and_download_agent_if_upgrade_available = Mock(return_value=True)
+        self.update_handler._download_agent_if_upgrade_available = Mock(return_value=True)
         self._test_run(invocations=0, calls=0, enable_updates=True, sleep_interval=(300,))
 
     @patch("azurelinuxagent.common.logger.info")
@@ -1929,9 +1893,11 @@ class TestAgentUpgrade(UpdateTestCase):
 
         with _get_update_handler(iterations, test_data) as (update_handler, protocol):
 
+            protocol.aggregate_status = None
+
             def get_handler(url, **kwargs):
                 if reload_conf is not None:
-                    reload_conf(url, protocol.mock_wire_data)
+                    reload_conf(url, protocol)
 
                 if HttpRequestPredicates.is_agent_package_request(url):
                     agent_pkg = load_bin_data(self._get_agent_file_name(), self._agent_zip_dir)
@@ -1939,7 +1905,14 @@ class TestAgentUpgrade(UpdateTestCase):
                     return ResponseMock(response=agent_pkg)
                 return protocol.mock_wire_data.mock_http_get(url, **kwargs)
 
-            protocol.set_http_handlers(http_get_handler=get_handler)
+            def put_handler(url, *args, **_):
+                if HttpRequestPredicates.is_host_plugin_status_request(url):
+                    # Skip reading the HostGA request data as its encoded
+                    return MockHttpResponse(status=500)
+                protocol.aggregate_status = json.loads(args[0])
+                return MockHttpResponse(status=201)
+
+            protocol.set_http_handlers(http_get_handler=get_handler, http_put_handler=put_handler)
             with self.create_conf_mocks(hotfix_frequency, normal_frequency):
                 with patch("azurelinuxagent.ga.update.add_event") as mock_telemetry:
                     update_handler._protocol = protocol
@@ -1950,12 +1923,23 @@ class TestAgentUpgrade(UpdateTestCase):
         exit_args, _ = exit_mock.call_args
         self.assertEqual(exit_args[0], 0, "Exit code should be 0")
 
+    def __assert_upgrade_telemetry_emitted_for_requested_version(self, mock_telemetry, upgrade=True, version="99999.0.0.0"):
+        upgrade_event_msgs = [kwarg['message'] for _, kwarg in mock_telemetry.call_args_list if
+                              'Exiting current process to {0} to the request Agent version {1}'.format(
+                                  "upgrade" if upgrade else "downgrade", version) in kwarg['message'] and kwarg[
+                                  'op'] == WALAEventOperation.AgentUpgrade]
+        self.assertEqual(1, len(upgrade_event_msgs),
+                         "Did not find the event indicating that the agent was upgraded. Got: {0}".format(
+                             mock_telemetry.call_args_list))
+
     def __assert_upgrade_telemetry_emitted(self, mock_telemetry, upgrade_type=AgentUpgradeType.Normal):
         upgrade_event_msgs = [kwarg['message'] for _, kwarg in mock_telemetry.call_args_list if
                               '{0} Agent upgrade discovered, updating to WALinuxAgent-99999.0.0.0 -- exiting'.format(
                                   upgrade_type) in kwarg['message'] and kwarg[
                                   'op'] == WALAEventOperation.AgentUpgrade]
-        self.assertEqual(1, len(upgrade_event_msgs), "Agent not upgraded properly")
+        self.assertEqual(1, len(upgrade_event_msgs),
+                         "Did not find the event indicating that the agent was upgraded. Got: {0}".format(
+                             mock_telemetry.call_args_list))
 
     def __assert_agent_directories_available(self, versions):
         for version in versions:
@@ -1973,6 +1957,13 @@ class TestAgentUpgrade(UpdateTestCase):
                                  "Agent upgrade discovered, updating to" in kwarg['message'] and kwarg[
                                      'op'] == WALAEventOperation.AgentUpgrade]), "Unwanted upgrade")
 
+    def __assert_ga_version_in_status(self, aggregate_status, version=str(CURRENT_VERSION)):
+        self.assertIsNotNone(aggregate_status, "Status should be reported")
+        self.assertEqual(aggregate_status['aggregateStatus']['guestAgentStatus']['version'], version,
+                         "Status should be reported from the Current version")
+        self.assertEqual(aggregate_status['aggregateStatus']['guestAgentStatus']['status'], 'Ready',
+                         "Guest Agent should be reported as Ready")
+
     def test_it_should_upgrade_agent_on_process_start_if_auto_upgrade_enabled(self):
         with self.__get_update_handler(iterations=10) as (update_handler, mock_telemetry):
 
@@ -1988,7 +1979,8 @@ class TestAgentUpgrade(UpdateTestCase):
         data_file = DATA_FILE.copy()
         data_file['ga_manifest'] = "wire/ga_manifest_no_upgrade.xml"
 
-        def reload_conf(url, mock_wire_data):
+        def reload_conf(url, protocol):
+            mock_wire_data = protocol.mock_wire_data
             # This function reloads the conf mid-run to mimic an actual customer scenario
             if HttpRequestPredicates.is_ga_manifest_request(url) and mock_wire_data.call_counts["manifest_of_ga.xml"] >= no_of_iterations/2:
                 reload_conf.call_count += 1
@@ -2015,7 +2007,8 @@ class TestAgentUpgrade(UpdateTestCase):
         data_file = DATA_FILE.copy()
         data_file['ga_manifest'] = "wire/ga_manifest_no_upgrade.xml"
 
-        def reload_conf(url, mock_wire_data):
+        def reload_conf(url, protocol):
+            mock_wire_data = protocol.mock_wire_data
             # This function reloads the conf mid-run to mimic an actual customer scenario
             if HttpRequestPredicates.is_ga_manifest_request(url) and mock_wire_data.call_counts["manifest_of_ga.xml"] >= 2:
                 reload_conf.call_count += 1
@@ -2059,7 +2052,8 @@ class TestAgentUpgrade(UpdateTestCase):
         data_file = DATA_FILE.copy()
         data_file['ga_manifest'] = "wire/ga_manifest_no_upgrade.xml"
 
-        def reload_conf(url, mock_wire_data):
+        def reload_conf(url, protocol):
+            mock_wire_data = protocol.mock_wire_data
             # This function reloads the conf mid-run to mimic an actual customer scenario
             if HttpRequestPredicates.is_ga_manifest_request(url) and mock_wire_data.call_counts["manifest_of_ga.xml"] >= no_of_iterations / 2:
                 reload_conf.call_count += 1
@@ -2091,12 +2085,7 @@ class TestAgentUpgrade(UpdateTestCase):
                 update_handler.run(debug=True)
 
             self.__assert_exit_code_successful(update_handler.exit_mock)
-            upgrade_event_msgs = [kwarg['message'] for _, kwarg in mock_telemetry.call_args_list if
-                                  'Agent upgrade discovered, updating to WALinuxAgent-9.9.9.10 -- exiting' in kwarg[
-                                      'message'] and kwarg['op'] == WALAEventOperation.AgentUpgrade]
-            self.assertEqual(1, len(upgrade_event_msgs),
-                             "Did not find the event indicating that the agent was upgraded. Got: {0}".format(
-                                 mock_telemetry.call_args_list))
+            self.__assert_upgrade_telemetry_emitted_for_requested_version(mock_telemetry, version="9.9.9.10")
             self.__assert_agent_directories_exist_and_others_dont_exist(versions=["9.9.9.10"])
 
     def test_it_should_cleanup_all_agents_except_requested_version_and_current_version(self):
@@ -2112,10 +2101,7 @@ class TestAgentUpgrade(UpdateTestCase):
                 update_handler.run(debug=True)
 
             self.__assert_exit_code_successful(update_handler.exit_mock)
-            upgrade_event_msgs = [kwarg['message'] for _, kwarg in mock_telemetry.call_args_list if
-                                  'Agent upgrade discovered, updating to WALinuxAgent-9.9.9.10 -- exiting' in kwarg[
-                                      'message'] and kwarg['op'] == WALAEventOperation.AgentUpgrade]
-            self.assertEqual(1, len(upgrade_event_msgs), "Agent not upgraded properly")
+            self.__assert_upgrade_telemetry_emitted_for_requested_version(mock_telemetry, version="9.9.9.10")
             self.__assert_agent_directories_exist_and_others_dont_exist(versions=["9.9.9.10", str(CURRENT_VERSION)])
 
     def test_it_should_not_update_if_requested_version_not_found_in_manifest(self):
@@ -2148,9 +2134,10 @@ class TestAgentUpgrade(UpdateTestCase):
         self.prepare_agents()
         self.assertEqual(20, self.agent_count(), "Agent directories not set properly")
 
-        def reload_conf(url, mock_wire_data):
-            # This function reloads the conf mid-run to mimic an actual customer scenario
+        def reload_conf(url, protocol):
+            mock_wire_data = protocol.mock_wire_data
 
+            # This function reloads the conf mid-run to mimic an actual customer scenario
             if HttpRequestPredicates.is_goal_state_request(url) and mock_wire_data.call_counts[
              "goalstate"] >= 10 and mock_wire_data.call_counts["goalstate"] < 15:
 
@@ -2178,7 +2165,7 @@ class TestAgentUpgrade(UpdateTestCase):
 
             self.assertGreaterEqual(reload_conf.call_count, 1, "Reload conf not updated as expected")
             self.__assert_exit_code_successful(update_handler.exit_mock)
-            self.__assert_upgrade_telemetry_emitted(mock_telemetry)
+            self.__assert_upgrade_telemetry_emitted_for_requested_version(mock_telemetry)
             self.__assert_agent_directories_exist_and_others_dont_exist(versions=["99999.0.0.0", str(CURRENT_VERSION)])
             self.assertEqual(update_handler._protocol.mock_wire_data.call_counts['agentArtifact'], 1,
                              "only 1 agent should've been downloaded - 1 per incarnation")
@@ -2192,7 +2179,9 @@ class TestAgentUpgrade(UpdateTestCase):
         self.prepare_agents()
         self.assertEqual(20, self.agent_count(), "Agent directories not set properly")
 
-        def reload_conf(url, mock_wire_data):
+        def reload_conf(url, protocol):
+            mock_wire_data = protocol.mock_wire_data
+
             # This function reloads the conf mid-run to mimic an actual customer scenario
             if HttpRequestPredicates.is_goal_state_request(url) and mock_wire_data.call_counts[
              "goalstate"] >= 5:
@@ -2245,6 +2234,97 @@ class TestAgentUpgrade(UpdateTestCase):
             self.__assert_no_agent_upgrade_telemetry(mock_telemetry)
             self.__assert_agent_directories_exist_and_others_dont_exist(versions=[str(CURRENT_VERSION)])
 
+    def test_it_should_skip_wait_to_update_if_requested_version_available(self):
+        no_of_iterations = 100
+
+        def reload_conf(url, protocol):
+            mock_wire_data = protocol.mock_wire_data
+
+            # This function reloads the conf mid-run to mimic an actual customer scenario
+            if HttpRequestPredicates.is_goal_state_request(url) and mock_wire_data.call_counts["goalstate"] >= 5:
+                reload_conf.call_count += 1
+
+                # Assert GA version from status to ensure agent is running fine from the current version
+                self.__assert_ga_version_in_status(protocol.aggregate_status)
+
+                # Update the ext-conf and incarnation and add requested version from GS
+                mock_wire_data.data_files["ext_conf"] = "wire/ext_conf_requested_version.xml"
+                data_file['ga_manifest'] = "wire/ga_manifest.xml"
+                mock_wire_data.reload()
+                self._add_write_permission_to_goal_state_files()
+                mock_wire_data.set_incarnation(2)
+
+        reload_conf.call_count = 0
+
+        data_file = mockwiredata.DATA_FILE.copy()
+        data_file['ga_manifest'] = "wire/ga_manifest_no_upgrade.xml"
+        with self.__get_update_handler(iterations=no_of_iterations, test_data=data_file, reload_conf=reload_conf,
+                                       normal_frequency=10, hotfix_frequency=10) as (update_handler, mock_telemetry):
+            with patch.object(conf, "get_enable_ga_versioning", return_value=True):
+                update_handler.run(debug=True)
+
+            self.assertGreater(reload_conf.call_count, 0, "Reload conf not updated")
+            self.assertLess(update_handler.get_iterations(), no_of_iterations,
+                            "The code should've exited as soon as requested version was found")
+            self.__assert_exit_code_successful(update_handler.exit_mock)
+            self.__assert_upgrade_telemetry_emitted_for_requested_version(mock_telemetry, version="9.9.9.10")
+
+    def test_it_should_blacklist_current_agent_on_downgrade(self):
+        # Create Agent directory for current agent
+        self.prepare_agents(count=1)
+        self.assertTrue(os.path.exists(self.agent_dir(CURRENT_VERSION)))
+        self.assertFalse(next(agent for agent in self.agents() if agent.version == CURRENT_VERSION).is_blacklisted,
+                         "The current agent should not be blacklisted")
+        downgraded_version = "1.2.0"
+
+        data_file = mockwiredata.DATA_FILE.copy()
+        data_file["ext_conf"] = "wire/ext_conf_requested_version.xml"
+        with self.__get_update_handler(test_data=data_file) as (update_handler, mock_telemetry):
+            with patch.object(conf, "get_enable_ga_versioning", return_value=True):
+                update_handler._protocol.mock_wire_data.set_extension_config_requested_version(downgraded_version)
+                update_handler._protocol.mock_wire_data.set_incarnation(2)
+                try:
+                    set_daemon_version("1.0.0.0")
+                    update_handler.run(debug=True)
+                finally:
+                    os.environ.pop(DAEMON_VERSION_ENV_VARIABLE)
+
+            self.__assert_exit_code_successful(update_handler.exit_mock)
+            self.__assert_upgrade_telemetry_emitted_for_requested_version(mock_telemetry, upgrade=False,
+                                                                          version=downgraded_version)
+            self.assertTrue(next(agent for agent in self.agents() if agent.version == CURRENT_VERSION).is_blacklisted,
+                            "The current agent should be blacklisted")
+
+    def test_it_should_not_downgrade_below_daemon_version(self):
+        data_file = mockwiredata.DATA_FILE.copy()
+        data_file["ext_conf"] = "wire/ext_conf_requested_version.xml"
+        with self.__get_update_handler(test_data=data_file) as (update_handler, mock_telemetry):
+            with patch.object(conf, "get_enable_ga_versioning", return_value=True):
+                update_handler._protocol.mock_wire_data.set_extension_config_requested_version("1.0.0.0")
+                update_handler._protocol.mock_wire_data.set_incarnation(2)
+
+                try:
+                    set_daemon_version("1.2.3.4")
+                    update_handler.run(debug=True)
+                finally:
+                    os.environ.pop(DAEMON_VERSION_ENV_VARIABLE)
+
+            self.__assert_exit_code_successful(update_handler.exit_mock)
+            upgrade_msgs = [kwarg for _, kwarg in mock_telemetry.call_args_list if
+                            kwarg['op'] == WALAEventOperation.AgentUpgrade]
+            # This will throw if corresponding message not found so not asserting on that
+            requested_version_found = next(kwarg for kwarg in upgrade_msgs if
+                                           "Found requested version in manifest: 1.0.0.0 for incarnation: 2" in kwarg[
+                                               'message'])
+            self.assertTrue(requested_version_found['is_success'],
+                            "The requested version found op should be reported as a success")
+
+            skipping_update = next(kwarg for kwarg in upgrade_msgs if
+                                   "Can't process the upgrade as the requested version: 1.0.0.0 is < current daemon version: 1.2.3.4" in
+                                   kwarg['message'])
+            self.assertFalse(skipping_update['is_success'], "Failed Event should be reported as a failure")
+            self.__assert_ga_version_in_status(update_handler._protocol.aggregate_status)
+
 
 @patch('azurelinuxagent.ga.update.get_collect_telemetry_events_handler')
 @patch('azurelinuxagent.ga.update.get_send_telemetry_events_handler')
@@ -2406,6 +2486,9 @@ class ProtocolMock(object):
     def get_protocol(self):
         return self
 
+    def get_incarnation(self):
+        return self.etag
+
     def get_vmagent_manifests(self):
         self.call_counts["get_vmagent_manifests"] += 1
         if self.goal_state_is_stale:
@@ -2584,7 +2667,7 @@ def _create_update_handler():
     Creates an UpdateHandler in which agent updates are mocked as a no-op.
     """
     update_handler = get_update_handler()
-    update_handler._check_and_download_agent_if_upgrade_available = Mock(return_value=False)
+    update_handler._download_agent_if_upgrade_available = Mock(return_value=False)
     return update_handler