Merge ~calvinmwadime/ubuntu/+source/walinuxagent:mantic-merge-2.9.1.1 into ubuntu/+source/walinuxagent:ubuntu/mantic-devel

[Calvin Mwadime Makokha]

added sru tests

added manpages

debian/patches: update cgroup logic to include v2

fix_systemd_networkd_lease_file_path.patch: patch osutil/ubuntu.py

debian/control: Remove isc-dhcp-client from Depends

Remove upstart config

debian/rules: stop installing SysV init files and stop running tests

disable_udev_overrides.patch: patch setup.py

debian/patches: Remove deprecated patches

debian/install: add new udev rules

debian/docs: Remove Changelog

debian/watch: Fix package version regex pattern

New upstream version 2.9.1

[Bryce Harrington (bryce)]

Hi Calvin, it's a bit late in the release for merges, did you have an MRE or SRU bug to accompany this?

Also, I'm not spotting changes to debian/changelog, was that intentional to be excluded?

[Calvin Mwadime Makokha (calvinmwadime)]

Hi Bryce,

> Hi Calvin, it's a bit late in the release for merges, did you have an MRE or
> SRU bug to accompany this?
After discussion with Utkarsh, it was decided we shall revisit this MP later since I was late to raise it.

> Also, I'm not spotting changes to debian/changelog, was that intentional to be
> excluded?
There are changes made to it. I think the diff display might have been truncated, however, you can see the file listed in the list above showing modified files

[Lucas Kanashiro (lucaskanashiro)]

Hi Calvin,

Since you decided to work on this later, I'd ask you to check if there is a way to remove the review slot for ubuntu-sponsors. Otherwise, it will be kept in the general sponsorship queue and patch pilots will keep revisiting this.

[Calvin Mwadime Makokha (calvinmwadime)]

Hi Lucas, which is the best way to do so? I have changed its status from `needs review` to `work in progress` will that help?

[Calvin Mwadime Makokha (calvinmwadime)]

Hi Lucas, there has been a change in urgency for this merge. I would request a review on this merge request.Changing the status to `needs review`.

8a67b96... by Calvin Mwadime Makokha on 2023-10-30

fix_systemd_networkd_lease_file_path.patch: patch osutil/ubuntu.py

Use systemd_networkd for newer ubuntu servers

7d40672... by Calvin Mwadime Makokha on 2023-10-30

changelog: update version

0b1a8de... by Calvin Mwadime Makokha on 2023-10-30

debian/patches: update cgroup logic to include v2

Properly handle systemd using cgroup v2

c7bd079... by Calvin Mwadime Makokha on 2023-10-30

changelog update

dbbfc63... by Calvin Mwadime Makokha on 2023-10-30

Add manpages

fbe4dd5... by Calvin Mwadime Makokha on 2023-10-30

changelog update

1d131ad... by Calvin Mwadime Makokha on 2023-10-30

Added sru tests

9c007ee... by Calvin Mwadime Makokha on 2023-10-30

changelog update

Unmerged commits

9c007ee... by Calvin Mwadime Makokha on 2023-10-30

changelog update

1d131ad... by Calvin Mwadime Makokha on 2023-10-30

Added sru tests

fbe4dd5... by Calvin Mwadime Makokha on 2023-10-30

changelog update

dbbfc63... by Calvin Mwadime Makokha on 2023-10-30

Add manpages

c7bd079... by Calvin Mwadime Makokha on 2023-10-30

changelog update

0b1a8de... by Calvin Mwadime Makokha on 2023-10-30

debian/patches: update cgroup logic to include v2

Properly handle systemd using cgroup v2

7d40672... by Calvin Mwadime Makokha on 2023-10-30

changelog: update version

8a67b96... by Calvin Mwadime Makokha on 2023-10-30

fix_systemd_networkd_lease_file_path.patch: patch osutil/ubuntu.py

Use systemd_networkd for newer ubuntu servers

95c00e8... by Calvin Mwadime Makokha on 2023-09-11

debian/control: Remove isc-dhcp-client from Depends

9f935bf... by Calvin Mwadime Makokha on 2023-09-11

Remove upstart config

diff --git a/.flake8 b/.flake8
deleted file mode 100644
index 63303c3..0000000
--- a/.flake8
+++ /dev/null
@@ -1,32 +0,0 @@
-#
-# The project did not use flake8 since inception so there are a number
-# of time-consuming flake8-identified improvements that are just a lot
-# of busy work.  Each of these should be disabled and code cleaned up.
-#
-# W503: Line break occurred before a binary operator
-# W504: Line break occurred after a binary operator
-# E126: Continuation line over-indented for hanging indent
-# E127: Continuation line over-indented for visual indent
-# E128: Continuation line under-indented for visual indent
-# E201: Whitespace after '('
-# E202: Whitespace before ')'
-# E203: Whitespace before ':'
-# E221: Multiple spaces before operator
-# E225: Missing whitespace around operator
-# E226: Missing whitespace around arithmetic operator
-# E231: Missing whitespace after ',', ';', or ':'
-# E261: At least two spaces before inline comment
-# E265: Block comment should start with '# '
-# E302: Expected 2 blank lines, found 0
-# E501: Line too long (xx > yy characters)
-# E502: The backslash is redundant between brackets
-# F401: Module imported but unused
-# F403: 'from module import *' used; unable to detect undefined names
-# F405: Name may be undefined, or defined from star imports: module
-#
-
-[flake8]
-ignore = W503,W504,E126,E127,E128,E201,E202,E203,E221,E225,E226,E231,E261,E265,E302,E501,E502,F401,F403,F405
-exclude = .git,__pycache__,docs/source/conf.py,old,build,dist,tests
-max-complexity = 30
-max-line-length = 120
\ No newline at end of file
diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md
index edfa1e6..fdcc07c 100644
--- a/.github/PULL_REQUEST_TEMPLATE.md
+++ b/.github/PULL_REQUEST_TEMPLATE.md
@@ -14,9 +14,8 @@ This will expedite the process of getting your pull request merged and avoid ext
 ### 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).
-- [ ] Except for special cases involving multiple contributors, the PR is started from a fork of the main repository, not a branch.
 - [ ] If applicable, the PR references the bug/issue that it fixes in the description.
-- [ ] New Unit tests were added for the changes made and Travis.CI is passing.
+- [ ] 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).
\ No newline at end of file
diff --git a/.github/codecov.yml b/.github/codecov.yml
new file mode 100644
index 0000000..77707aa
--- /dev/null
+++ b/.github/codecov.yml
@@ -0,0 +1,2 @@
+github_checks:
+    annotations: false
diff --git a/.github/workflows/ci_pr.yml b/.github/workflows/ci_pr.yml
new file mode 100644
index 0000000..e559268
--- /dev/null
+++ b/.github/workflows/ci_pr.yml
@@ -0,0 +1,128 @@
+name: CI Unit tests
+
+on:
+  push:
+    branches: [ "*" ]
+  pull_request:
+    branches: [ "*" ]
+  workflow_dispatch:
+
+jobs:
+  test-legacy-python-versions:
+    
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - python-version: 2.6
+          - python-version: 3.4
+
+    name: "Python ${{ matrix.python-version }} Unit Tests"
+    runs-on: ubuntu-20.04
+    container:
+      image: ubuntu:16.04
+      volumes:
+        - /home/waagent:/home/waagent
+    defaults:
+      run:
+        shell: bash -l {0}
+
+    env:
+      NOSEOPTS: "--verbose"
+
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Install Python ${{ matrix.python-version }}
+      run: |
+        apt-get update
+        apt-get install -y curl bzip2 sudo python3
+        curl https://dcrdata.blob.core.windows.net/python/python-${{ matrix.python-version }}.tar.bz2 -o python-${{ matrix.python-version }}.tar.bz2
+        sudo tar xjvf python-${{ matrix.python-version }}.tar.bz2 --directory /
+
+    - name: Test with nosetests
+      run: |
+        if [[ ${{ matrix.python-version }} == 2.6 ]]; then
+          source /home/waagent/virtualenv/python2.6.9/bin/activate
+        else
+          source /home/waagent/virtualenv/python3.4.8/bin/activate
+        fi
+        ./ci/nosetests.sh
+        exit $?
+
+  test-current-python-versions:
+
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+
+          - python-version: 2.7
+            PYLINTOPTS: "--rcfile=ci/2.7.pylintrc --ignore=tests_e2e,makepkg.py"
+
+          - python-version: 3.5
+            PYLINTOPTS: "--rcfile=ci/3.6.pylintrc --ignore=tests_e2e,makepkg.py"
+
+          - python-version: 3.6
+            PYLINTOPTS: "--rcfile=ci/3.6.pylintrc --ignore=tests_e2e"
+
+          - python-version: 3.7
+            PYLINTOPTS: "--rcfile=ci/3.6.pylintrc --ignore=tests_e2e"
+
+          - python-version: 3.8
+            PYLINTOPTS: "--rcfile=ci/3.6.pylintrc --ignore=tests_e2e"
+
+          - python-version: 3.9
+            PYLINTOPTS: "--rcfile=ci/3.6.pylintrc"
+            additional-nose-opts: "--with-coverage --cover-erase --cover-inclusive --cover-branches --cover-package=azurelinuxagent"
+
+    name: "Python ${{ matrix.python-version }} Unit Tests"
+    runs-on: ubuntu-20.04
+
+    env:
+      PYLINTOPTS: ${{ matrix.PYLINTOPTS }}
+      PYLINTFILES: "azurelinuxagent setup.py makepkg.py tests tests_e2e"
+      NOSEOPTS: "--with-timer ${{ matrix.additional-nose-opts }}"
+      PYTHON_VERSION: ${{ matrix.python-version }}
+
+    steps:
+
+    - name: Checkout WALinuxAgent repo
+      uses: actions/checkout@v3
+
+    - name: Setup Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install dependencies
+      id: install-dependencies
+      run: |
+        sudo env "PATH=$PATH" python -m pip install --upgrade pip
+        sudo env "PATH=$PATH" pip install -r requirements.txt
+        sudo env "PATH=$PATH" pip install -r test-requirements.txt
+
+    - name: Run pylint
+      run: |
+        pylint $PYLINTOPTS --jobs=0 $PYLINTFILES
+
+    - name: Test with nosetests
+      if: success() || (failure() && steps.install-dependencies.outcome == 'success')
+      run: |
+        ./ci/nosetests.sh
+        exit $?
+
+    - name: Compile Coverage
+      if: matrix.python-version == 3.9
+      run: |
+        echo looking for coverage files :
+        ls -alh | grep -i coverage
+        sudo env "PATH=$PATH" coverage combine coverage.*.data
+        sudo env "PATH=$PATH" coverage xml
+        sudo env "PATH=$PATH" coverage report
+
+    - name: Upload Coverage
+      if: matrix.python-version == 3.9
+      uses: codecov/codecov-action@v2
+      with:
+        file: ./coverage.xml
\ No newline at end of file
diff --git a/.gitignore b/.gitignore
index 0a31340..fd64d33 100644
--- a/.gitignore
+++ b/.gitignore
@@ -17,8 +17,6 @@ develop-eggs/
 dist/
 downloads/
 eggs/
-lib/
-lib64/
 parts/
 sdist/
 var/
@@ -92,3 +90,4 @@ ENV/
 
 # pyenv
 .python-version
+.vscode/
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index fa672d3..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,43 +0,0 @@
----
-os: linux
-dist: xenial
-language: python
-env:
-  - NOSEOPTS="--verbose" SETUPOPTS=""
-  # Add SETUPOPTS="check flake8" to enable flake8 checks
-
-matrix:
-  # exclude the default "python" build - we're being specific here...
-  exclude:
-    - python:
-      env:
-        - NOSEOPTS="" SETUPOPTS="check flake8"
-
-  include:
-    - python: 2.6
-      dist: trusty
-      env:
-        - NOSEOPTS="--verbose" SETUPOPTS=""
-    - python: 2.7
-    - python: 3.4
-    - python: 3.6
-    - python: 3.7
-      env:
-        - >-
-          NOSEOPTS="--verbose --with-coverage --cover-inclusive
-          --cover-min-percentage=60 --cover-branches
-          --cover-package=azurelinuxagent --cover-xml"
-          SETUPOPTS=""
-
-install:
-  - pip install -r requirements.txt
-  - pip install -r test-requirements.txt
-
-script:
-  # future: - pylint setup.py makepkg.py azurelinuxagent/
-  - nosetests $NOSEOPTS --attr '!requires_sudo' tests
-  - sudo env "PATH=$PATH" nosetests $NOSEOPTS --verbose --attr 'requires_sudo' tests
-  - if [ ! -z "$SETUPOPTS" ]; then /usr/bin/env python setup.py $SETUPOPTS; fi
-
-after_success:
-  - if [[ $TRAVIS_PYTHON_VERSION == 3.7 ]]; then codecov; fi
\ No newline at end of file
diff --git a/CODEOWNERS b/CODEOWNERS
index a8f2d5d..8707e60 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -1,3 +1,4 @@
+1
 # See https://help.github.com/articles/about-codeowners/
 # for more info about CODEOWNERS file
 
@@ -9,6 +10,7 @@
 # when there are requests for changes in the provisioning agent. For any
 # questions, please feel free to reach out to [e-mail].
 /azurelinuxagent/pa/ @trstringer @anhvoms
+/tests/pa/ @trstringer @anhvoms
 
 #
 # RDMA
@@ -19,4 +21,4 @@
 #
 # Linux Agent team
 #
-* @narrieta @vrdmr @pgombar @larohra
+* @narrieta @ZhidongPeng @nagworld9 @maddieford
diff --git a/Changelog b/Changelog
deleted file mode 100644
index da68890..0000000
--- a/Changelog
+++ /dev/null
@@ -1,38 +0,0 @@
-WALinuxAgent                                                          Changelog
-|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
-
-Refer to releases WALinuxAgent release page: https://github.com/Azure/WALinuxAgent/releases for detailed changelog after v2.2.0
-
-12 August 2016, v2.1.6
-   . Improved RDMA support
-   . Extension state migration
-   . Alpine Linux support
-   . Fixes for #347, #351, #353
-
-15 July 2016, v2.1.5
-   . Goal state processing extension
-   . Multi-nic improvements
-   . Bug fixes for #145, #141, #133, #116, #187, #169, #104, #127, #163,
-    #190, #185, #174
-
-09 Mar 2016, WALinuxAgent 2.1.4
-   . Add support for FreeBSD
-   . Fix a bug for internal extension version resolving
-
-29 Jan 2016, WALinuxAgent 2.1.3
-   . Fixed endpoint probing for Azure Stack
-   . Multiple fixes for extension handling
-
-07 Dec 2015, WALinuxAgent 2.1.2
-   . Multiple fixes for extension handling and provisioning
-
-07 Aug 2015, WALinuxAgent 2.1.1
-   . Support python3
-   . Fixed bugs for metadata protocol
-   . Fixed a few pylint warnings
-   . Enabled travis-ci
-
-|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
-01 Jul 2015, WALinuxAgent 2.1.0
-   . Divide waagent into different modules
-
diff --git a/README.md b/README.md
index 0069d46..ae6a851 100644
--- a/README.md
+++ b/README.md
@@ -1,31 +1,15 @@
+
 # Microsoft Azure Linux Agent
 
-## Develop branch status
-
-[![Travis CI](https://travis-ci.org/Azure/WALinuxAgent.svg?branch=develop)](https://travis-ci.org/Azure/WALinuxAgent/branches)
-[![CodeCov](https://codecov.io/gh/Azure/WALinusAgent/branch/develop/graph/badge.svg)](https://codecov.io/gh/Azure/WALinuxAgent/branch/develop)
-
-Each badge below represents our basic validation tests for an image, which are executed several times each day. These include provisioning, user account, disk, extension and networking scenarios.
-
-Note: These badges represent testing to our develop branch which might not be stable. For a stable build please use master branch instead. 
-
-Image | Status |
-------|--------|
-Canonical UbuntuServer 14.04.5-LTS|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Canonical_UbuntuServer_14.04.5-LTS__agent--bvt.svg)
-Canonical UbuntuServer 14.04.5-DAILY-LTS|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Canonical_UbuntuServer_14.04.5-DAILY-LTS__agent--bvt.svg)
-Canonical UbuntuServer 16.04-LTS|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Canonical_UbuntuServer_16.04-LTS__agent--bvt.svg)
-Canonical UbuntuServer 16.04-DAILY-LTS|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Canonical_UbuntuServer_16.04-DAILY-LTS__agent--bvt.svg)
-Canonical UbuntuServer 18.04-LTS|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Canonical_UbuntuServer_18.04-LTS__agent--bvt.svg)
-Canonical UbuntuServer 18.04-DAILY-LTS|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Canonical_UbuntuServer_18.04-DAILY-LTS__agent--bvt.svg)
-Credativ Debian 8|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Credativ_Debian_8__agent--bvt.svg)
-Credativ Debian 8-DAILY|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Credativ_Debian_8-DAILY__agent--bvt.svg)
-Credativ Debian 9|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Credativ_Debian_9__agent--bvt.svg)
-Credativ Debian 9-DAILY|![badge](https://dcrbadges.blob.core.windows.net/scenarios/Credativ_Debian_9-DAILY__agent--bvt.svg)
-OpenLogic CentOS 6.9|![badge](https://dcrbadges.blob.core.windows.net/scenarios/OpenLogic_CentOS_6.9__agent--bvt.svg)
-OpenLogic CentOS 7.4|![badge](https://dcrbadges.blob.core.windows.net/scenarios/OpenLogic_CentOS_7.4__agent--bvt.svg)
-RedHat RHEL 6.9|![badge](https://dcrbadges.blob.core.windows.net/scenarios/RedHat_RHEL_6.9__agent--bvt.svg)
-RedHat RHEL 7-RAW|![badge](https://dcrbadges.blob.core.windows.net/scenarios/RedHat_RHEL_7-RAW__agent--bvt.svg)
-SUSE SLES 12-SP3|![badge](https://dcrbadges.blob.core.windows.net/scenarios/SUSE_SLES_12-SP3__agent--bvt.svg)
+## Linux distributions support
+
+Our daily automation tests most of the [Linux distributions supported by Azure](https://docs.microsoft.com/en-us/azure/virtual-machines/linux/endorsed-distros); the Agent can be
+used on other distributions as well, but development, testing and support for those are done by the open source community.
+
+Testing is done using the develop branch, which can be unstable. For a stable build please use the master branch instead.
+
+[![CodeCov](https://codecov.io/gh/Azure/WALinuxAgent/branch/develop/graph/badge.svg)](https://codecov.io/gh/Azure/WALinuxAgent/branch/develop)
+
 
 ## Introduction
 
@@ -49,7 +33,6 @@ functionality for Linux IaaS deployments:
 
 * Kernel
   * Configure virtual NUMA (disable for kernel <2.6.37)
-  * Consume Hyper-V entropy for /dev/random
   * Configure SCSI timeouts for the root device (which could be remote)
 
 * Diagnostics
@@ -79,13 +62,15 @@ The agent will use an HTTP proxy if provided via the `http_proxy` (for `http` re
 `https_proxy` (for `https` requests) environment variables. The `HttpProxy.Host` and
 `HttpProxy.Port` configuration variables (see below), if used, will override the environment
 settings. Due to limitations of Python, the agent *does not* support HTTP proxies requiring
-authentication.
+authentication. Note that when the agent service is managed by systemd, environment variables 
+such as `http_proxy` and `https_proxy` should be defined using one the mechanisms provided by 
+systemd (e.g. by using Environment or EnvironmentFile in the service file).
 
 ## Requirements
 
 The following systems have been tested and are known to work with the Azure
 Linux Agent.  Please note that this list may differ from the official list
-of supported systems on the Microsoft Azure Platform as described [here](http://support.microsoft.com/kb/2805216).
+of supported systems on the Microsoft Azure Platform as described [here](https://docs.microsoft.com/en-us/azure/virtual-machines/linux/endorsed-distros).
 
 Waagent depends on some system packages in order to function properly:
 
@@ -109,6 +94,12 @@ For more advanced installation options, such as installing to custom locations o
     sudo python setup.py install --register-service
 ```
 
+For Python 3, use:
+
+```bash
+    sudo python3 setup.py install --register-service
+```
+
 You can view more installation options by running:
 
 ```bash
@@ -177,6 +168,8 @@ For CoreOS, use:
 
 `-start`: Run waagent as a background process
 
+`-collect-logs [-full]`: Runs the log collector utility that collects relevant agent logs for debugging and stores them in the agent folder on disk. Exact location will be shown when run. Use flag `-full` for more exhaustive log collection. 
+
 ## Configuration
 
 A configuration file (/etc/waagent.conf) controls the actions of waagent. Blank lines and lines whose first character is a `#` are ignored (end-of-line comments are *not* supported).
@@ -185,6 +178,7 @@ A sample configuration file is shown below:
 
 ```yml
 Extensions.Enabled=y
+Extensions.GoalStatePeriod=6
 Provisioning.Agent=auto
 Provisioning.DeleteRootPassword=n
 Provisioning.RegenerateSshHostKeyPair=y
@@ -202,6 +196,8 @@ ResourceDisk.EnableSwap=n
 ResourceDisk.EnableSwapEncryption=n
 ResourceDisk.SwapSizeMB=0
 Logs.Verbose=n
+Logs.Collect=y
+Logs.CollectPeriod=3600
 OS.AllowHTTP=n
 OS.RootDeviceScsiTimeout=300
 OS.EnableFIPS=n
@@ -210,8 +206,6 @@ OS.SshClientAliveInterval=180
 OS.SshDir=/etc/ssh
 HttpProxy.Host=None
 HttpProxy.Port=None
-CGroups.EnforceLimits=y
-CGroups.Excluded=customscript,runcommand
 ```
 
 The various configuration options are described in detail below. Configuration
@@ -238,6 +232,32 @@ without the agent. In order to do that, the `provisionVMAgent` flag must be set 
 provisioning time, via whichever API is being used. We will provide more details on
 this on our wiki when it is generally available. 
 
+#### __Extensions.GoalStatePeriod__
+
+_Type: Integer_  
+_Default: 6_
+
+How often to poll for new goal states (in seconds) and report the status of the VM
+and extensions. Goal states describe the desired state of the extensions on the VM.
+
+_Note_: setting up this parameter to more than a few minutes can make the state of
+the VM be reported as unresponsive/unavailable on the Azure portal. Also, this 
+setting affects how fast the agent starts executing extensions. 
+
+#### __AutoUpdate.Enabled__
+
+_Type: Boolean_  
+_Default: y_
+
+Enables auto-update of the Extension Handler. The Extension Handler is responsible 
+for managing extensions and reporting VM status. The core functionality of the agent
+is contained in the Extension Handler, and we encourage users to enable this option 
+in order to maintain an up to date version.
+
+On most distros the default value is 'y'.
+
+For more information on the agent version, see our [FAQ](https://github.com/Azure/WALinuxAgent/wiki/FAQ#what-does-goal-state-agent-mean-in-waagent---version-output).
+
 #### __Provisioning.Agent__
 
 _Type: String_
@@ -261,7 +281,22 @@ _Note_: This configuration option has been removed and has no effect. waagent
 now auto-detects cloud-init as a provisioning agent (with an option to override
 with `Provisioning.Agent`).
 
-#### __Provisioning.UseCloudInit__ (*removed in 2.2.45*)
+#### __Provisioning.MonitorHostName__
+
+_Type: Boolean_ 
+_Default: n_
+
+Monitor host name changes and publish changes via DHCP requests.
+
+#### __Provisioning.MonitorHostNamePeriod__
+
+_Type: Integer_ 
+_Default: 30_
+
+How often to monitor host name changes (in seconds). This setting is ignored if
+MonitorHostName is not set.
+
+#### __Provisioning.UseCloudInit__
 
 _Type: Boolean_ 
 _Default: n_
@@ -397,7 +432,7 @@ system swap space.
 _Type: Boolean_  
 _Default: n_
 
-If set, the swap file (/swapfile) is mounted as an encrypted filesystem.
+If set, the swap file (/swapfile) is mounted as an encrypted filesystem (flag supported only on FreeBSD.)
 
 #### __ResourceDisk.SwapSizeMB__
 
@@ -414,6 +449,25 @@ _Default: n_
 If set, log verbosity is boosted. Waagent logs to /var/log/waagent.log and
 leverages the system logrotate functionality to rotate logs.
 
+
+#### __Logs.Collect__
+
+_Type: Boolean_  
+_Default: y_
+
+If set, agent logs will be periodically collected and uploaded to a secure location for improved supportability.
+
+NOTE: This feature relies on the agent's resource usage features (cgroups); this flag will not take effect on any distro not supported.
+
+#### __Logs.CollectPeriod__
+
+_Type: Integer_  
+_Default: 3600_
+
+This configures how frequently to collect and upload logs. Default is each hour.
+
+NOTE: This only takes effect if the Logs.Collect option is enabled.
+
 #### __OS.AllowHTTP__
 
 _Type: Boolean_  
@@ -442,6 +496,14 @@ OpenSSL commands. This signals OpenSSL to use any installed FIPS-compliant libra
 Note that the agent itself has no FIPS-specific code. _If no FIPS-compliant certificates are
 installed, then enabling this option will cause all OpenSSL commands to fail._
 
+#### __OS.MonitorDhcpClientRestartPeriod__
+
+_Type: Integer_
+_Default: 30_
+
+The agent monitor restarts of the DHCP client and restores network rules when it happens. This
+setting determines how often (in seconds) to monitor for restarts.
+
 #### __OS.RootDeviceScsiTimeout__
 
 _Type: Integer_  
@@ -450,6 +512,14 @@ _Default: 300_
 This configures the SCSI timeout in seconds on the root device. If not set, the
 system defaults are used.
 
+#### __OS.RootDeviceScsiTimeoutPeriod__
+
+_Type: Integer_  
+_Default: 30_
+
+How often to set the SCSI timeout on the root device (in seconds). This setting is
+ignored if RootDeviceScsiTimeout is not set.
+
 #### __OS.OpensslPath__
 
 _Type: String_  
@@ -458,6 +528,13 @@ _Default: None_
 This can be used to specify an alternate path for the openssl binary to use for
 cryptographic operations.
 
+#### __OS.RemovePersistentNetRulesPeriod__
+_Type: Integer_  
+_Default: 30_
+
+How often to remove the udev rules for persistent network interface names (75-persistent-net-generator.rules
+and /etc/udev/rules.d/70-persistent-net.rules) (in seconds) 
+
 #### __OS.SshClientAliveInterval__
 
 _Type: Integer_  
diff --git a/SECURITY.md b/SECURITY.md
new file mode 100644
index 0000000..e138ec5
--- /dev/null
+++ b/SECURITY.md
@@ -0,0 +1,41 @@
+<!-- BEGIN MICROSOFT SECURITY.MD V0.0.8 BLOCK -->
+
+## Security
+
+Microsoft takes the security of our software products and services seriously, which includes all source code repositories managed through our GitHub organizations, which include [Microsoft](https://github.com/microsoft), [Azure](https://github.com/Azure), [DotNet](https://github.com/dotnet), [AspNet](https://github.com/aspnet), [Xamarin](https://github.com/xamarin), and [our GitHub organizations](https://opensource.microsoft.com/).
+
+If you believe you have found a security vulnerability in any Microsoft-owned repository that meets [Microsoft's definition of a security vulnerability](https://aka.ms/opensource/security/definition), please report it to us as described below.
+
+## Reporting Security Issues
+
+**Please do not report security vulnerabilities through public GitHub issues.**
+
+Instead, please report them to the Microsoft Security Response Center (MSRC) at [https://msrc.microsoft.com/create-report](https://aka.ms/opensource/security/create-report).
+
+If you prefer to submit without logging in, send email to [[e-mail]](mailto:[e-mail]).  If possible, encrypt your message with our PGP key; please download it from the [Microsoft Security Response Center PGP Key page](https://aka.ms/opensource/security/pgpkey).
+
+You should receive a response within 24 hours. If for some reason you do not, please follow up via email to ensure we received your original message. Additional information can be found at [microsoft.com/msrc](https://aka.ms/opensource/security/msrc). 
+
+Please include the requested information listed below (as much as you can provide) to help us better understand the nature and scope of the possible issue:
+
+  * Type of issue (e.g. buffer overflow, SQL injection, cross-site scripting, etc.)
+  * Full paths of source file(s) related to the manifestation of the issue
+  * The location of the affected source code (tag/branch/commit or direct URL)
+  * Any special configuration required to reproduce the issue
+  * Step-by-step instructions to reproduce the issue
+  * Proof-of-concept or exploit code (if possible)
+  * Impact of the issue, including how an attacker might exploit the issue
+
+This information will help us triage your report more quickly.
+
+If you are reporting for a bug bounty, more complete reports can contribute to a higher bounty award. Please visit our [Microsoft Bug Bounty Program](https://aka.ms/opensource/security/bounty) page for more details about our active programs.
+
+## Preferred Languages
+
+We prefer all communications to be in English.
+
+## Policy
+
+Microsoft follows the principle of [Coordinated Vulnerability Disclosure](https://aka.ms/opensource/security/cvd).
+
+<!-- END MICROSOFT SECURITY.MD BLOCK -->
diff --git a/azurelinuxagent/agent.py b/azurelinuxagent/agent.py
index 6e65084..8c30348 100644
--- a/azurelinuxagent/agent.py
+++ b/azurelinuxagent/agent.py
@@ -24,21 +24,47 @@ Module agent
 from __future__ import print_function
 
 import os
-import sys
 import re
 import subprocess
+import sys
 import threading
-import traceback
+from azurelinuxagent.common import cgroupconfigurator, logcollector
+from azurelinuxagent.common.cgroupapi import SystemdCgroupsApi
 
-import azurelinuxagent.common.logger as logger
-import azurelinuxagent.common.event as event
 import azurelinuxagent.common.conf as conf
-from azurelinuxagent.common.version import AGENT_NAME, AGENT_LONG_VERSION, \
-                                     DISTRO_NAME, DISTRO_VERSION, \
-                                     PY_VERSION_MAJOR, PY_VERSION_MINOR, \
-                                     PY_VERSION_MICRO, GOAL_STATE_AGENT_VERSION
+import azurelinuxagent.common.event as event
+import azurelinuxagent.common.logger as logger
+from azurelinuxagent.common.future import ustr
+from azurelinuxagent.common.logcollector import LogCollector, OUTPUT_RESULTS_FILE_PATH
 from azurelinuxagent.common.osutil import get_osutil
-from azurelinuxagent.common.utils import fileutil
+from azurelinuxagent.common.utils import fileutil, textutil
+from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
+from azurelinuxagent.common.utils.networkutil import AddFirewallRules
+from azurelinuxagent.common.version import AGENT_NAME, AGENT_LONG_VERSION, AGENT_VERSION, \
+    DISTRO_NAME, DISTRO_VERSION, \
+    PY_VERSION_MAJOR, PY_VERSION_MINOR, \
+    PY_VERSION_MICRO, GOAL_STATE_AGENT_VERSION, \
+    get_daemon_version, set_daemon_version
+from azurelinuxagent.ga.collect_logs import CollectLogsHandler, get_log_collector_monitor_handler
+from azurelinuxagent.pa.provision.default import ProvisionHandler
+
+
+class AgentCommands(object):
+    """
+    This is the list of all commands that the Linux Guest Agent supports
+    """
+    DeprovisionUser = "deprovision+user"
+    Deprovision = "deprovision"
+    Daemon = "daemon"
+    Start = "start"
+    RegisterService = "register-service"
+    RunExthandlers = "run-exthandlers"
+    Version = "version"
+    ShowConfig = "show-configuration"
+    Help = "help"
+    CollectLogs = "collect-logs"
+    SetupFirewall = "setup-firewall"
+    Provision = "provision"
 
 
 class Agent(object):
@@ -49,24 +75,24 @@ class Agent(object):
         self.conf_file_path = conf_file_path
         self.osutil = get_osutil()
 
-        #Init stdout log
+        # Init stdout log
         level = logger.LogLevel.VERBOSE if verbose else logger.LogLevel.INFO
         logger.add_logger_appender(logger.AppenderType.STDOUT, level)
 
-        #Init config
+        # Init config
         conf_file_path = self.conf_file_path \
                 if self.conf_file_path is not None \
                     else self.osutil.get_agent_conf_file_path()
         conf.load_conf_from_file(conf_file_path)
 
-        #Init log
+        # Init log
         verbose = verbose or conf.get_logs_verbose()
         level = logger.LogLevel.VERBOSE if verbose else logger.LogLevel.INFO
-        logger.add_logger_appender(logger.AppenderType.FILE, level,
-                                 path="/var/log/waagent.log")
-        if conf.get_logs_console():
-            logger.add_logger_appender(logger.AppenderType.CONSOLE, level,
-                    path="/dev/console")
+        logger.add_logger_appender(logger.AppenderType.FILE, level, path=conf.get_agent_log_file())
+
+        # echo the log to /dev/console if the machine will be provisioned
+        if conf.get_logs_console() and not ProvisionHandler.is_provisioned():
+            self.__add_console_appender(level)
 
         if event.send_logs_to_telemetry():
             logger.add_logger_appender(logger.AppenderType.TELEMETRY,
@@ -84,22 +110,30 @@ class Agent(object):
                 "Exception occurred while creating extension "
                 "log directory {0}: {1}".format(ext_log_dir, e))
 
-        #Init event reporter
+        # Init event reporter
+        # Note that the reporter is not fully initialized here yet. Some telemetry fields are filled with data
+        # originating from the goal state or IMDS, which requires a WireProtocol instance. Once a protocol
+        # has been established, those fields must be explicitly initialized using
+        # initialize_event_logger_vminfo_common_parameters(). Any events created before that initialization
+        # will contain dummy values on those fields.
         event.init_event_status(conf.get_lib_dir())
-        event_dir = os.path.join(conf.get_lib_dir(), "events")
+        event_dir = os.path.join(conf.get_lib_dir(), event.EVENTS_DIRECTORY)
         event.init_event_logger(event_dir)
         event.enable_unhandled_err_dump("WALA")
 
+    def __add_console_appender(self, level):
+        logger.add_logger_appender(logger.AppenderType.CONSOLE, level, path="/dev/console")
+
     def daemon(self):
         """
         Run agent daemon
         """
+        set_daemon_version(AGENT_VERSION)
         logger.set_prefix("Daemon")
         threading.current_thread().setName("Daemon")
         child_args = None \
             if self.conf_file_path is None \
                 else "-configuration-path:{0}".format(self.conf_file_path)
-
         from azurelinuxagent.daemon import get_daemon_handler
         daemon_handler = get_daemon_handler()
         daemon_handler.run(child_args=child_args)
@@ -137,6 +171,21 @@ class Agent(object):
         """
         logger.set_prefix("ExtHandler")
         threading.current_thread().setName("ExtHandler")
+
+        #
+        # Agents < 2.2.53 used to echo the log to the console. Since the extension handler could have been started by
+        # one of those daemons, output a message indicating that output to the console will stop, otherwise users
+        # may think that the agent died if they noticed that output to the console stops abruptly.
+        #
+        # Feel free to remove this code if telemetry shows there are no more agents <= 2.2.53 in the field.
+        #
+        if conf.get_logs_console() and get_daemon_version() < FlexibleVersion("2.2.53"):
+            self.__add_console_appender(logger.LogLevel.INFO)
+            try:
+                logger.info(u"The agent will now check for updates and then will process extensions. Output to /dev/console will be suspended during those operations.")
+            finally:
+                logger.disable_console_output()
+
         from azurelinuxagent.ga.update import get_update_handler
         update_handler = get_update_handler()
         update_handler.run(debug)
@@ -146,91 +195,175 @@ class Agent(object):
         for k in sorted(configuration.keys()):
             print("{0} = {1}".format(k, configuration[k]))
 
+    def collect_logs(self, is_full_mode):
+        logger.set_prefix("LogCollector")
+
+        if is_full_mode:
+            logger.info("Running log collector mode full")
+        else:
+            logger.info("Running log collector mode normal")
+
+        # Check the cgroups unit
+        cpu_cgroup_path, memory_cgroup_path, log_collector_monitor = None, None, None
+        if CollectLogsHandler.should_validate_cgroups():
+            cgroups_api = SystemdCgroupsApi()
+            cpu_cgroup_path, memory_cgroup_path = cgroups_api.get_process_cgroup_paths("self")
 
-def main(args=[]):
+            cpu_slice_matches = (cgroupconfigurator.LOGCOLLECTOR_SLICE in cpu_cgroup_path)
+            memory_slice_matches = (cgroupconfigurator.LOGCOLLECTOR_SLICE in memory_cgroup_path)
+
+            if not cpu_slice_matches or not memory_slice_matches:
+                logger.info("The Log Collector process is not in the proper cgroups:")
+                if not cpu_slice_matches:
+                    logger.info("\tunexpected cpu slice")
+                if not memory_slice_matches:
+                    logger.info("\tunexpected memory slice")
+
+                sys.exit(logcollector.INVALID_CGROUPS_ERRCODE)
+
+        try:
+            log_collector = LogCollector(is_full_mode, cpu_cgroup_path, memory_cgroup_path)
+            log_collector_monitor = get_log_collector_monitor_handler(log_collector.cgroups)
+            log_collector_monitor.run()
+            archive = log_collector.collect_logs_and_get_archive()
+            logger.info("Log collection successfully completed. Archive can be found at {0} "
+                  "and detailed log output can be found at {1}".format(archive, OUTPUT_RESULTS_FILE_PATH))
+        except Exception as e:
+            logger.error("Log collection completed unsuccessfully. Error: {0}".format(ustr(e)))
+            logger.info("Detailed log output can be found at {0}".format(OUTPUT_RESULTS_FILE_PATH))
+            sys.exit(1)
+        finally:
+            if log_collector_monitor is not None:
+                log_collector_monitor.stop()
+
+    @staticmethod
+    def setup_firewall(firewall_metadata):
+
+        print("Setting up firewall for the WALinux Agent with args: {0}".format(firewall_metadata))
+        try:
+            AddFirewallRules.add_iptables_rules(firewall_metadata['wait'], firewall_metadata['dst_ip'],
+                                                firewall_metadata['uid'])
+            print("Successfully set the firewall rules")
+        except Exception as error:
+            print("Unable to add firewall rules. Error: {0}".format(ustr(error)))
+            sys.exit(1)
+
+
+def main(args=None):
     """
     Parse command line arguments, exit with usage() on error.
     Invoke different methods according to different command
     """
+    if args is None:
+        args = []
     if len(args) <= 0:
         args = sys.argv[1:]
-    command, force, verbose, debug, conf_file_path = parse_args(args)
-    if command == "version":
+    command, force, verbose, debug, conf_file_path, log_collector_full_mode, firewall_metadata = parse_args(args)
+    if command == AgentCommands.Version:
         version()
-    elif command == "help":
+    elif command == AgentCommands.Help:
         print(usage())
-    elif command == "start":
+    elif command == AgentCommands.Start:
         start(conf_file_path=conf_file_path)
     else:
         try:
             agent = Agent(verbose, conf_file_path=conf_file_path)
-            if command == "deprovision+user":
+            if command == AgentCommands.DeprovisionUser:
                 agent.deprovision(force, deluser=True)
-            elif command == "deprovision":
+            elif command == AgentCommands.Deprovision:
                 agent.deprovision(force, deluser=False)
-            elif command == "provision":
+            elif command == AgentCommands.Provision:
                 agent.provision()
-            elif command == "register-service":
+            elif command == AgentCommands.RegisterService:
                 agent.register_service()
-            elif command == "daemon":
+            elif command == AgentCommands.Daemon:
                 agent.daemon()
-            elif command == "run-exthandlers":
+            elif command == AgentCommands.RunExthandlers:
                 agent.run_exthandlers(debug)
-            elif command == "show-configuration":
+            elif command == AgentCommands.ShowConfig:
                 agent.show_configuration()
-        except Exception:
+            elif command == AgentCommands.CollectLogs:
+                agent.collect_logs(log_collector_full_mode)
+            elif command == AgentCommands.SetupFirewall:
+                agent.setup_firewall(firewall_metadata)
+        except Exception as e:
             logger.error(u"Failed to run '{0}': {1}",
                          command,
-                         traceback.format_exc())
+                         textutil.format_exception(e))
+
 
 def parse_args(sys_args):
     """
     Parse command line arguments
     """
-    cmd = "help"
+    cmd = AgentCommands.Help
     force = False
     verbose = False
     debug = False
     conf_file_path = None
-    for a in sys_args:
-        m = re.match("^(?:[-/]*)configuration-path:([\w/\.\-_]+)", a)
+    log_collector_full_mode = False
+    firewall_metadata = {
+        "dst_ip": None,
+        "uid": None,
+        "wait": ""
+    }
+
+    regex_cmd_format = "^([-/]*){0}"
+
+    for arg in sys_args:
+        if arg == "":
+            # Don't parse an empty parameter
+            continue
+        m = re.match("^(?:[-/]*)configuration-path:([\w/\.\-_]+)", arg)  # pylint: disable=W1401
         if not m is None:
             conf_file_path = m.group(1)
             if not os.path.exists(conf_file_path):
                 print("Error: Configuration file {0} does not exist".format(
                         conf_file_path), file=sys.stderr)
-                usage()
+                print(usage())
                 sys.exit(1)
-        
-        elif re.match("^([-/]*)deprovision\\+user", a):
-            cmd = "deprovision+user"
-        elif re.match("^([-/]*)deprovision", a):
-            cmd = "deprovision"
-        elif re.match("^([-/]*)daemon", a):
-            cmd = "daemon"
-        elif re.match("^([-/]*)start", a):
-            cmd = "start"
-        elif re.match("^([-/]*)register-service", a):
-            cmd = "register-service"
-        elif re.match("^([-/]*)run-exthandlers", a):
-            cmd = "run-exthandlers"
-        elif re.match("^([-/]*)version", a):
-            cmd = "version"
-        elif re.match("^([-/]*)verbose", a):
+        elif re.match("^([-/]*)deprovision\\+user", arg):
+            cmd = AgentCommands.DeprovisionUser
+        elif re.match(regex_cmd_format.format(AgentCommands.Deprovision), arg):
+            cmd = AgentCommands.Deprovision
+        elif re.match(regex_cmd_format.format(AgentCommands.Daemon), arg):
+            cmd = AgentCommands.Daemon
+        elif re.match(regex_cmd_format.format(AgentCommands.Start), arg):
+            cmd = AgentCommands.Start
+        elif re.match(regex_cmd_format.format(AgentCommands.RegisterService), arg):
+            cmd = AgentCommands.RegisterService
+        elif re.match(regex_cmd_format.format(AgentCommands.RunExthandlers), arg):
+            cmd = AgentCommands.RunExthandlers
+        elif re.match(regex_cmd_format.format(AgentCommands.Version), arg):
+            cmd = AgentCommands.Version
+        elif re.match(regex_cmd_format.format("verbose"), arg):
             verbose = True
-        elif re.match("^([-/]*)debug", a):
+        elif re.match(regex_cmd_format.format("debug"), arg):
             debug = True
-        elif re.match("^([-/]*)force", a):
+        elif re.match(regex_cmd_format.format("force"), arg):
             force = True
-        elif re.match("^([-/]*)show-configuration", a):
-            cmd = "show-configuration"
-        elif re.match("^([-/]*)(help|usage|\\?)", a):
-            cmd = "help"
+        elif re.match(regex_cmd_format.format(AgentCommands.ShowConfig), arg):
+            cmd = AgentCommands.ShowConfig
+        elif re.match("^([-/]*)(help|usage|\\?)", arg):
+            cmd = AgentCommands.Help
+        elif re.match(regex_cmd_format.format(AgentCommands.CollectLogs), arg):
+            cmd = AgentCommands.CollectLogs
+        elif re.match(regex_cmd_format.format("full"), arg):
+            log_collector_full_mode = True
+        elif re.match(regex_cmd_format.format(AgentCommands.SetupFirewall), arg):
+            cmd = AgentCommands.SetupFirewall
+        elif re.match(regex_cmd_format.format("dst_ip=(?P<dst_ip>[\\d.]{7,})"), arg):
+            firewall_metadata['dst_ip'] = re.match(regex_cmd_format.format("dst_ip=(?P<dst_ip>[\\d.]{7,})"), arg).group(
+                'dst_ip')
+        elif re.match(regex_cmd_format.format("uid=(?P<uid>[\\d]+)"), arg):
+            firewall_metadata['uid'] = re.match(regex_cmd_format.format("uid=(?P<uid>[\\d]+)"), arg).group('uid')
+        elif re.match(regex_cmd_format.format("(w|wait)$"), arg):
+            firewall_metadata['wait'] = "-w"
         else:
-            cmd = "help"
+            cmd = AgentCommands.Help
             break
 
-    return cmd, force, verbose, debug, conf_file_path
+    return cmd, force, verbose, debug, conf_file_path, log_collector_full_mode, firewall_metadata
 
 
 def version():
@@ -245,29 +378,33 @@ def version():
                                        PY_VERSION_MICRO))
     print("Goal state agent: {0}".format(GOAL_STATE_AGENT_VERSION))
 
+
 def usage():
     """
     Return agent usage message
     """
     s  = "\n"
     s += ("usage: {0} [-verbose] [-force] [-help] "
-           "-configuration-path:<path to configuration file>"
+           "-configuration-path:<path to configuration file>" 
            "-deprovision[+user]|-register-service|-version|-daemon|-start|"
-           "-run-exthandlers|-show-configuration]"
+           "-run-exthandlers|-show-configuration|-collect-logs [-full]|-setup-firewall [-dst_ip=<IP> -uid=<UID> [-w/--wait]]"
            "").format(sys.argv[0])
     s += "\n"
     return s
 
+
 def start(conf_file_path=None):
     """
     Start agent daemon in a background process and set stdout/stderr to
     /dev/null
     """
-    devnull = open(os.devnull, 'w')
     args = [sys.argv[0], '-daemon']
     if conf_file_path is not None:
         args.append('-configuration-path:{0}'.format(conf_file_path))
-    subprocess.Popen(args, stdout=devnull, stderr=devnull)
+
+    with open(os.devnull, 'w') as devnull:
+        subprocess.Popen(args, stdout=devnull, stderr=devnull)
+
 
 if __name__ == '__main__' :
     main()
diff --git a/azurelinuxagent/common/AgentGlobals.py b/azurelinuxagent/common/AgentGlobals.py
new file mode 100644
index 0000000..dbfda92
--- /dev/null
+++ b/azurelinuxagent/common/AgentGlobals.py
@@ -0,0 +1,39 @@
+# Microsoft Azure Linux Agent
+#
+# Copyright 2020 Microsoft Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.6+ and Openssl 1.0+
+
+
+class AgentGlobals(object):
+    """
+    This class is used for setting AgentGlobals which can be used all throughout the Agent.
+    """
+
+    GUID_ZERO = "00000000-0000-0000-0000-000000000000"
+
+    #
+    # Some modules (e.g. telemetry) require an up-to-date container ID. We update this variable each time we
+    # fetch the goal state.
+    #
+    _container_id = GUID_ZERO
+
+    @staticmethod
+    def get_container_id():
+        return AgentGlobals._container_id
+
+    @staticmethod
+    def update_container_id(container_id):
+        AgentGlobals._container_id = container_id
diff --git a/azurelinuxagent/common/agent_supported_feature.py b/azurelinuxagent/common/agent_supported_feature.py
new file mode 100644
index 0000000..d7f93e2
--- /dev/null
+++ b/azurelinuxagent/common/agent_supported_feature.py
@@ -0,0 +1,122 @@
+# Copyright 2018 Microsoft Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.6+ and Openssl 1.0+
+#
+
+
+class SupportedFeatureNames(object):
+    """
+    Enum for defining the Feature Names for all features that we the agent supports
+    """
+    MultiConfig = "MultipleExtensionsPerHandler"
+    ExtensionTelemetryPipeline = "ExtensionTelemetryPipeline"
+    FastTrack = "FastTrack"
+
+
+class AgentSupportedFeature(object):
+    """
+    Interface for defining all features that the Linux Guest Agent supports and reports their if supported back to CRP
+    """
+
+    def __init__(self, name, version="1.0", supported=False):
+        self.__name = name
+        self.__version = version
+        self.__supported = supported
+
+    @property
+    def name(self):
+        return self.__name
+
+    @property
+    def version(self):
+        return self.__version
+
+    @property
+    def is_supported(self):
+        return self.__supported
+
+
+class _MultiConfigFeature(AgentSupportedFeature):
+
+    __NAME = SupportedFeatureNames.MultiConfig
+    __VERSION = "1.0"
+    __SUPPORTED = True
+
+    def __init__(self):
+        super(_MultiConfigFeature, self).__init__(name=_MultiConfigFeature.__NAME,
+                                                  version=_MultiConfigFeature.__VERSION,
+                                                  supported=_MultiConfigFeature.__SUPPORTED)
+
+
+class _ETPFeature(AgentSupportedFeature):
+
+    __NAME = SupportedFeatureNames.ExtensionTelemetryPipeline
+    __VERSION = "1.0"
+    __SUPPORTED = True
+
+    def __init__(self):
+        super(_ETPFeature, self).__init__(name=self.__NAME,
+                                          version=self.__VERSION,
+                                          supported=self.__SUPPORTED)
+
+
+# This is the list of features that Agent supports and we advertise to CRP
+__CRP_ADVERTISED_FEATURES = {
+    SupportedFeatureNames.MultiConfig: _MultiConfigFeature()
+}
+
+
+# This is the list of features that Agent supports and we advertise to Extensions
+__EXTENSION_ADVERTISED_FEATURES = {
+    SupportedFeatureNames.ExtensionTelemetryPipeline: _ETPFeature()
+}
+
+
+def get_supported_feature_by_name(feature_name):
+    if feature_name in __CRP_ADVERTISED_FEATURES:
+        return __CRP_ADVERTISED_FEATURES[feature_name]
+
+    if feature_name in __EXTENSION_ADVERTISED_FEATURES:
+        return __EXTENSION_ADVERTISED_FEATURES[feature_name]
+
+    raise NotImplementedError("Feature with Name: {0} not found".format(feature_name))
+
+
+def get_agent_supported_features_list_for_crp():
+    """
+    List of features that the GuestAgent currently supports (like FastTrack, MultiConfig, etc).
+    We need to send this list as part of Status reporting to inform CRP of all the features the agent supports.
+    :return: Dict containing all CRP supported features with the key as their names and the AgentFeature object as
+             the value if they are supported by the Agent
+        Eg: {
+                MultipleExtensionsPerHandler: _MultiConfigFeature()
+            }
+    """
+
+    return dict((name, feature) for name, feature in __CRP_ADVERTISED_FEATURES.items() if feature.is_supported)
+
+
+def get_agent_supported_features_list_for_extensions():
+    """
+    List of features that the GuestAgent currently supports (like Extension Telemetry Pipeline, etc) needed by Extensions.
+    We need to send this list as environment variables when calling extension commands to inform Extensions of all the
+    features the agent supports.
+    :return: Dict containing all Extension supported features with the key as their names and the AgentFeature object as
+             the value if the feature is supported by the Agent.
+        Eg: {
+                CRPSupportedFeatureNames.ExtensionTelemetryPipeline: _ETPFeature()
+            }
+    """
+    return dict((name, feature) for name, feature in __EXTENSION_ADVERTISED_FEATURES.items() if feature.is_supported)
diff --git a/azurelinuxagent/common/cgroup.py b/azurelinuxagent/common/cgroup.py
index 2ad70c1..b2bf32f 100644
--- a/azurelinuxagent/common/cgroup.py
+++ b/azurelinuxagent/common/cgroup.py
@@ -13,47 +13,94 @@
 # limitations under the License.
 #
 # Requires Python 2.6+ and Openssl 1.0+
+
 import errno
 import os
 import re
+from datetime import timedelta
 
-from azurelinuxagent.common import logger
+from azurelinuxagent.common import logger, conf
 from azurelinuxagent.common.exception import CGroupsException
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil
 from azurelinuxagent.common.utils import fileutil
 
-re_user_system_times = re.compile(r'user (\d+)\nsystem (\d+)\n')
+_REPORT_EVERY_HOUR = timedelta(hours=1)
+_DEFAULT_REPORT_PERIOD = timedelta(seconds=conf.get_cgroup_check_period())
 
+AGENT_NAME_TELEMETRY = "walinuxagent.service"  # Name used for telemetry; it needs to be consistent even if the name of the service changes
+AGENT_LOG_COLLECTOR = "azure-walinuxagent-logcollector"
 
-class CGroupContollers(object):
-    CPU = "cpu"
-    MEMORY = "memory"
 
+class CounterNotFound(Exception):
+    pass
 
-class CGroup(object):
-    @staticmethod
-    def create(cgroup_path, controller, extension_name):
-        """
-        Factory method to create the correct CGroup.
-        """
-        if controller == CGroupContollers.CPU:
-            return CpuCgroup(extension_name, cgroup_path)
-        if controller == CGroupContollers.MEMORY:
-            return MemoryCgroup(extension_name, cgroup_path)
-        raise CGroupsException('CGroup controller {0} is not supported'.format(controller))
 
-    def __init__(self, name, cgroup_path, controller_type):
+class MetricValue(object):
+
+    """
+    Class for defining all the required metric fields to send telemetry.
+    """
+
+    def __init__(self, category, counter, instance, value, report_period=_DEFAULT_REPORT_PERIOD):
+        self._category = category
+        self._counter = counter
+        self._instance = instance
+        self._value = value
+        self._report_period = report_period
+
+    @property
+    def category(self):
+        return self._category
+
+    @property
+    def counter(self):
+        return self._counter
+
+    @property
+    def instance(self):
+        return self._instance
+
+    @property
+    def value(self):
+        return self._value
+
+    @property
+    def report_period(self):
+        return self._report_period
+
+
+class MetricsCategory(object):
+    MEMORY_CATEGORY = "Memory"
+    CPU_CATEGORY = "CPU"
+
+
+class MetricsCounter(object):
+    PROCESSOR_PERCENT_TIME = "% Processor Time"
+    TOTAL_MEM_USAGE = "Total Memory Usage"
+    MAX_MEM_USAGE = "Max Memory Usage"
+    THROTTLED_TIME = "Throttled Time"
+    SWAP_MEM_USAGE = "Swap Memory Usage"
+    AVAILABLE_MEM = "Available MBytes"
+    USED_MEM = "Used MBytes"
+
+
+re_user_system_times = re.compile(r'user (\d+)\nsystem (\d+)\n')
+
+
+class CGroup(object):
+    def __init__(self, name, cgroup_path):
         """
         Initialize _data collection for the Memory controller
         :param: name: Name of the CGroup
         :param: cgroup_path: Path of the controller
-        :param: controller_type:
         :return:
         """
         self.name = name
         self.path = cgroup_path
-        self.controller = controller_type
+
+    def __str__(self):
+        return "{0} [{1}]".format(self.name, self.path)
 
     def _get_cgroup_file(self, file_name):
         return os.path.join(self.path, file_name)
@@ -89,7 +136,7 @@ class CGroup(object):
             logger.error("File {0} is empty but should not be".format(parameter_filename))
             raise CGroupsException("File {0} is empty but should not be".format(parameter_filename))
         except Exception as e:
-            if isinstance(e, (IOError, OSError)) and e.errno == errno.ENOENT:
+            if isinstance(e, (IOError, OSError)) and e.errno == errno.ENOENT:  # pylint: disable=E1101
                 raise e
             parameter_filename = self._get_cgroup_file(parameter_name)
             raise CGroupsException("Exception while attempting to read {0}".format(parameter_filename), e)
@@ -114,42 +161,26 @@ class CGroup(object):
                                  ' Internal error: {1}'.format(self.path, ustr(e)))
         return False
 
-    def get_tracked_processes(self):
+    def get_tracked_metrics(self, **_):
         """
-        :return: List of Str (Pids). Will return an empty string if we couldn't fetch any tracked processes.
+        Retrieves the current value of the metrics tracked for this cgroup and returns them as an array.
+
+        Note: Agent won't track the metrics if the current cpu ticks less than previous value and returns empty array.
         """
-        procs = []
-        try:
-            procs = self._get_parameters("cgroup.procs")
-        except (IOError, OSError) as e:
-            if e.errno == errno.ENOENT:
-                # only suppressing file not found exceptions.
-                pass
-            else:
-                logger.periodic_warn(logger.EVERY_HALF_HOUR,
-                                     'Could not get list of procs from "cgroup.procs" file in the cgroup: {0}.'
-                                     ' Internal error: {1}'.format(self.path, ustr(e)))
-        except CGroupsException as e:
-            logger.periodic_warn(logger.EVERY_HALF_HOUR,
-                                 'Could not get list of tasks from "cgroup.procs" file in the cgroup: {0}.'
-                                 ' Internal error: {1}'.format(self.path, ustr(e)))
-        return procs
+        raise NotImplementedError()
 
 
 class CpuCgroup(CGroup):
     def __init__(self, name, cgroup_path):
-        super(CpuCgroup, self).__init__(name, cgroup_path, CGroupContollers.CPU)
+        super(CpuCgroup, self).__init__(name, cgroup_path)
 
         self._osutil = get_osutil()
         self._previous_cgroup_cpu = None
         self._previous_system_cpu = None
         self._current_cgroup_cpu = None
         self._current_system_cpu = None
-
-    def __str__(self):
-        return "cgroup: Name: {0}, cgroup_path: {1}; Controller: {2}".format(
-            self.name, self.path, self.controller
-        )
+        self._previous_throttled_time = None
+        self._current_throttled_time = None
 
     def _get_cpu_ticks(self, allow_no_such_file_or_directory_error=False):
         """
@@ -159,24 +190,54 @@ class CpuCgroup(CGroup):
         returns 0; this is useful when the function can be called before the cgroup has been created.
         """
         try:
-            cpu_stat = self._get_file_contents('cpuacct.stat')
+            cpuacct_stat = self._get_file_contents('cpuacct.stat')
         except Exception as e:
-            if not isinstance(e, (IOError, OSError)) or e.errno != errno.ENOENT:
+            if not isinstance(e, (IOError, OSError)) or e.errno != errno.ENOENT:  # pylint: disable=E1101
                 raise CGroupsException("Failed to read cpuacct.stat: {0}".format(ustr(e)))
             if not allow_no_such_file_or_directory_error:
                 raise e
-            cpu_stat = None
+            cpuacct_stat = None
 
         cpu_ticks = 0
 
-        if cpu_stat is not None:
-            match = re_user_system_times.match(cpu_stat)
+        if cpuacct_stat is not None:
+            #
+            # Sample file:
+            #     # cat /sys/fs/cgroup/cpuacct/azure.slice/walinuxagent.service/cpuacct.stat
+            #     user 10190
+            #     system 3160
+            #
+            match = re_user_system_times.match(cpuacct_stat)
             if not match:
-                raise CGroupsException("The contents of {0} are invalid: {1}".format(self._get_cgroup_file('cpuacct.stat'), cpu_stat))
+                raise CGroupsException(
+                    "The contents of {0} are invalid: {1}".format(self._get_cgroup_file('cpuacct.stat'), cpuacct_stat))
             cpu_ticks = int(match.groups()[0]) + int(match.groups()[1])
 
         return cpu_ticks
 
+    def get_throttled_time(self):
+        try:
+            with open(os.path.join(self.path, 'cpu.stat')) as cpu_stat:
+                #
+                # Sample file:
+                #
+                #   # cat /sys/fs/cgroup/cpuacct/azure.slice/walinuxagent.service/cpu.stat
+                #   nr_periods  51660
+                #   nr_throttled 19461
+                #   throttled_time 1529590856339
+                #
+                for line in cpu_stat:
+                    match = re.match(r'throttled_time\s+(\d+)', line)
+                    if match is not None:
+                        return int(match.groups()[0])
+                raise Exception("Cannot find throttled_time")
+        except (IOError, OSError) as e:
+            if e.errno == errno.ENOENT:
+                return 0
+            raise CGroupsException("Failed to read cpu.stat: {0}".format(ustr(e)))
+        except Exception as e:
+            raise CGroupsException("Failed to read cpu.stat: {0}".format(ustr(e)))
+
     def _cpu_usage_initialized(self):
         return self._current_cgroup_cpu is not None and self._current_system_cpu is not None
 
@@ -188,13 +249,14 @@ class CpuCgroup(CGroup):
             raise CGroupsException("initialize_cpu_usage() should be invoked only once")
         self._current_cgroup_cpu = self._get_cpu_ticks(allow_no_such_file_or_directory_error=True)
         self._current_system_cpu = self._osutil.get_total_cpu_ticks_since_boot()
+        self._current_throttled_time = self.get_throttled_time()
 
     def get_cpu_usage(self):
         """
         Computes the CPU used by the cgroup since the last call to this function.
 
-        The usage is measured as a percentage of utilization of all cores in the system. For example,
-        using 1 core at 100% on a 4-core system would be reported as 25%.
+        The usage is measured as a percentage of utilization of 1 core in the system. For example,
+        using 1 core all of the time on a 4-core system would be reported as 100%.
 
         NOTE: initialize_cpu_usage() must be invoked before calling get_cpu_usage()
         """
@@ -209,53 +271,122 @@ class CpuCgroup(CGroup):
         cgroup_delta = self._current_cgroup_cpu - self._previous_cgroup_cpu
         system_delta = max(1, self._current_system_cpu - self._previous_system_cpu)
 
-        return round(100.0 * float(cgroup_delta) / float(system_delta), 3)
+        return round(100.0 * self._osutil.get_processor_cores() * float(cgroup_delta) / float(system_delta), 3)
+
+    def get_cpu_throttled_time(self, read_previous_throttled_time=True):
+        """
+        Computes the throttled time (in seconds) since the last call to this function.
+        NOTE: initialize_cpu_usage() must be invoked before calling this function
+        Compute only current throttled time if read_previous_throttled_time set to False
+        """
+        if not read_previous_throttled_time:
+            return float(self.get_throttled_time() / 1E9)
+
+        if not self._cpu_usage_initialized():
+            raise CGroupsException(
+                "initialize_cpu_usage() must be invoked before the first call to get_throttled_time()")
+
+        self._previous_throttled_time = self._current_throttled_time
+        self._current_throttled_time = self.get_throttled_time()
+
+        return float(self._current_throttled_time - self._previous_throttled_time) / 1E9
+
+    def get_tracked_metrics(self, **kwargs):
+        tracked = []
+        cpu_usage = self.get_cpu_usage()
+        if cpu_usage >= float(0):
+            tracked.append(
+                MetricValue(MetricsCategory.CPU_CATEGORY, MetricsCounter.PROCESSOR_PERCENT_TIME, self.name, cpu_usage))
+
+        if 'track_throttled_time' in kwargs and kwargs['track_throttled_time']:
+            throttled_time = self.get_cpu_throttled_time()
+            if cpu_usage >= float(0) and throttled_time >= float(0):
+                tracked.append(
+                    MetricValue(MetricsCategory.CPU_CATEGORY, MetricsCounter.THROTTLED_TIME, self.name, throttled_time))
+
+        return tracked
 
 
 class MemoryCgroup(CGroup):
     def __init__(self, name, cgroup_path):
+        super(MemoryCgroup, self).__init__(name, cgroup_path)
+
+        self._counter_not_found_error_count = 0
+
+    def _get_memory_stat_counter(self, counter_name):
+        try:
+            with open(os.path.join(self.path, 'memory.stat')) as memory_stat:
+                # cat /sys/fs/cgroup/memory/azure.slice/memory.stat
+                # cache 67178496
+                # rss 42340352
+                # rss_huge 6291456
+                # swap 0
+                for line in memory_stat:
+                    re_memory_counter = r'{0}\s+(\d+)'.format(counter_name)
+                    match = re.match(re_memory_counter, line)
+                    if match is not None:
+                        return int(match.groups()[0])
+        except (IOError, OSError) as e:
+            if e.errno == errno.ENOENT:
+                raise
+            raise CGroupsException("Failed to read memory.stat: {0}".format(ustr(e)))
+        except Exception as e:
+            raise CGroupsException("Failed to read memory.stat: {0}".format(ustr(e)))
+
+        raise CounterNotFound("Cannot find counter: {0}".format(counter_name))
+
+    def get_memory_usage(self):
         """
-        Initialize _data collection for the Memory controller
+        Collect RSS+CACHE from memory.stat cgroup.
 
-        :return: MemoryCgroup
+        :return: Memory usage in bytes
+        :rtype: int
         """
-        super(MemoryCgroup, self).__init__(name, cgroup_path, CGroupContollers.MEMORY)
 
-    def __str__(self):
-        return "cgroup: Name: {0}, cgroup_path: {1}; Controller: {2}".format(
-            self.name, self.path, self.controller
-        )
+        cache = self._get_memory_stat_counter("cache")
+        rss = self._get_memory_stat_counter("rss")
+        return cache + rss
 
-    def get_memory_usage(self):
+    def try_swap_memory_usage(self):
         """
-        Collect memory.usage_in_bytes from the cgroup.
+        Collect SWAP from memory.stat cgroup.
 
         :return: Memory usage in bytes
         :rtype: int
+        Note: stat file is the only place to get the SWAP since other swap related file memory.memsw.usage_in_bytes is for total Memory+SWAP.
         """
-        usage = None
         try:
-            usage = self._get_parameters('memory.usage_in_bytes', first_line_only=True)
-        except Exception as e:
-            if isinstance(e, (IOError, OSError)) and e.errno == errno.ENOENT:
-                raise
-            raise CGroupsException("Exception while attempting to read {0}".format("memory.usage_in_bytes"), e)
-
-        return int(usage)
+            return self._get_memory_stat_counter("swap")
+        except CounterNotFound as e:
+            if self._counter_not_found_error_count < 1:
+                logger.periodic_info(logger.EVERY_HALF_HOUR,
+                                     '{0} from "memory.stat" file in the cgroup: {1}---[Note: This log for informational purpose only and can be ignored]'.format(ustr(e), self.path))
+                self._counter_not_found_error_count += 1
+            return 0
 
     def get_max_memory_usage(self):
         """
-        Collect memory.usage_in_bytes from the cgroup.
+        Collect memory.max_usage_in_bytes from the cgroup.
 
         :return: Memory usage in bytes
         :rtype: int
         """
-        usage = None
+        usage = 0
         try:
-            usage = self._get_parameters('memory.max_usage_in_bytes', first_line_only=True)
+            usage = int(self._get_parameters('memory.max_usage_in_bytes', first_line_only=True))
         except Exception as e:
-            if isinstance(e, (IOError, OSError)) and e.errno == errno.ENOENT:
+            if isinstance(e, (IOError, OSError)) and e.errno == errno.ENOENT:  # pylint: disable=E1101
                 raise
-            raise CGroupsException("Exception while attempting to read {0}".format("memory.usage_in_bytes"), e)
-
-        return int(usage)
+            raise CGroupsException("Exception while attempting to read {0}".format("memory.max_usage_in_bytes"), e)
+
+        return usage
+
+    def get_tracked_metrics(self, **_):
+        return [
+            MetricValue(MetricsCategory.MEMORY_CATEGORY, MetricsCounter.TOTAL_MEM_USAGE, self.name,
+                        self.get_memory_usage()),
+            MetricValue(MetricsCategory.MEMORY_CATEGORY, MetricsCounter.MAX_MEM_USAGE, self.name,
+                        self.get_max_memory_usage(), _REPORT_EVERY_HOUR),
+            MetricValue(MetricsCategory.MEMORY_CATEGORY, MetricsCounter.SWAP_MEM_USAGE, self.name,
+                        self.try_swap_memory_usage(), _REPORT_EVERY_HOUR)
+        ]
diff --git a/azurelinuxagent/common/cgroupapi.py b/azurelinuxagent/common/cgroupapi.py
index c671a2e..ca0ef3b 100644
--- a/azurelinuxagent/common/cgroupapi.py
+++ b/azurelinuxagent/common/cgroupapi.py
@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 # Copyright 2018 Microsoft Corporation
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -14,102 +15,65 @@
 #
 # Requires Python 2.6+ and Openssl 1.0+
 
-import errno
 import os
+import re
 import shutil
 import subprocess
+import threading
 import uuid
 
 from azurelinuxagent.common import logger
-from azurelinuxagent.common.cgroup import CGroup
+from azurelinuxagent.common.cgroup import CpuCgroup, MemoryCgroup
 from azurelinuxagent.common.cgroupstelemetry import CGroupsTelemetry
 from azurelinuxagent.common.conf import get_agent_pid_file_path
-from azurelinuxagent.common.event import add_event, WALAEventOperation
 from azurelinuxagent.common.exception import CGroupsException, ExtensionErrorCodes, ExtensionError, \
     ExtensionOperationError
 from azurelinuxagent.common.future import ustr
+from azurelinuxagent.common.osutil import systemd
 from azurelinuxagent.common.utils import fileutil, shellutil
-from azurelinuxagent.common.utils.extensionprocessutil import handle_process_completion, read_output
-from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
+from azurelinuxagent.common.utils.extensionprocessutil import handle_process_completion, read_output, \
+    TELEMETRY_MESSAGE_MAX_LEN
+from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
+from azurelinuxagent.common.version import get_distro
 
 CGROUPS_FILE_SYSTEM_ROOT = '/sys/fs/cgroup'
 CGROUP_CONTROLLERS = ["cpu", "memory"]
-VM_AGENT_CGROUP_NAME = "walinuxagent.service"
-EXTENSIONS_ROOT_CGROUP_NAME = "walinuxagent.extensions"
-UNIT_FILES_FILE_SYSTEM_PATH = "/etc/systemd/system"
+EXTENSION_SLICE_PREFIX = "azure-vmextensions"
 
 
-class CGroupsApi(object):
+class SystemdRunError(CGroupsException):
     """
-    Interface for the cgroups API
+    Raised when systemd-run fails
     """
-    def create_agent_cgroups(self):
-        raise NotImplementedError()
-
-    def create_extension_cgroups_root(self):
-        raise NotImplementedError()
-
-    def create_extension_cgroups(self, extension_name):
-        raise NotImplementedError()
-
-    def remove_extension_cgroups(self, extension_name):
-        raise NotImplementedError()
 
-    def get_extension_cgroups(self, extension_name):
-        raise NotImplementedError()
+    def __init__(self, msg=None):
+        super(SystemdRunError, self).__init__(msg)
 
-    def start_extension_command(self, extension_name, command, timeout, shell, cwd, env, stdout, stderr, error_code):
-        raise NotImplementedError()
 
-    def cleanup_legacy_cgroups(self):
-        raise NotImplementedError()
+class CGroupsApi(object):
+    @staticmethod
+    def cgroups_supported():
+        distro_info = get_distro()
+        distro_name = distro_info[0]
+        try:
+            distro_version = FlexibleVersion(distro_info[1])
+        except ValueError:
+            return False
+        return distro_name.lower() == 'ubuntu' and distro_version.major >= 16
 
     @staticmethod
     def track_cgroups(extension_cgroups):
         try:
             for cgroup in extension_cgroups:
                 CGroupsTelemetry.track_cgroup(cgroup)
-        except Exception as e:
+        except Exception as exception:
             logger.warn("Cannot add cgroup '{0}' to tracking list; resource usage will not be tracked. "
-                        "Error: {1}".format(cgroup.path, ustr(e)))
+                        "Error: {1}".format(cgroup.path, ustr(exception)))
 
     @staticmethod
-    def _get_extension_cgroup_name(extension_name):
-        # Since '-' is used as a separator in systemd unit names, we replace it with '_' to prevent side-effects.
-        return extension_name.replace('-', '_')
-
-    @staticmethod
-    def create():
-        """
-        Factory method to create the correct API for the current platform
-        """
-        return SystemdCgroupsApi() if CGroupsApi._is_systemd() else FileSystemCgroupsApi()
-
-    @staticmethod
-    def _is_systemd():
-        """
-        Determine if systemd is managing system services; the implementation follows the same strategy as, for example,
-        sd_booted() in libsystemd, or /usr/sbin/service
-        """
-        return os.path.exists('/run/systemd/system/')
-
-    @staticmethod
-    def _foreach_controller(operation, message):
-        """
-        Executes the given operation on all controllers that need to be tracked; outputs 'message' if the controller
-        is not mounted or if an error occurs in the operation
-        :return: Returns a list of error messages or an empty list if no errors occurred
-        """
-        mounted_controllers = os.listdir(CGROUPS_FILE_SYSTEM_ROOT)
-
-        for controller in CGROUP_CONTROLLERS:
-            try:
-                if controller not in mounted_controllers:
-                    logger.warn('Cgroup controller "{0}" is not mounted. {1}', controller, message)
-                else:
-                    operation(controller)
-            except Exception as e:
-                logger.warn('Error in cgroup controller "{0}": {1}. {2}', controller, ustr(e), message)
+    def get_processes_in_cgroup(cgroup_path):
+        with open(os.path.join(cgroup_path, "cgroup.procs"), "r") as cgroup_procs:
+            return [int(pid) for pid in cgroup_procs.read().split()]
 
     @staticmethod
     def _foreach_legacy_cgroup(operation):
@@ -138,429 +102,250 @@ class CGroupsApi(object):
 
                 if os.path.exists(procs_file):
                     procs_file_contents = fileutil.read_file(procs_file).strip()
-                    daemon_pid = fileutil.read_file(get_agent_pid_file_path()).strip()
+                    daemon_pid = CGroupsApi.get_daemon_pid()
 
-                    if daemon_pid in procs_file_contents:
+                    if ustr(daemon_pid) in procs_file_contents:
                         operation(controller, daemon_pid)
         finally:
             for _, cgroup in legacy_cgroups:
                 logger.info('Removing {0}', cgroup)
                 shutil.rmtree(cgroup, ignore_errors=True)
+        return len(legacy_cgroups)
 
-
-class FileSystemCgroupsApi(CGroupsApi):
-    """
-    Cgroups interface using the cgroups file system directly
-    """
     @staticmethod
-    def _try_mkdir(path):
-        """
-        Try to create a directory, recursively. If it already exists as such, do nothing. Raise the appropriate
-        exception should an error occur.
-
-        :param path: str
-        """
-        if not os.path.isdir(path):
-            try:
-                os.makedirs(path, 0o755)
-            except OSError as e:
-                if e.errno == errno.EEXIST:
-                    if not os.path.isdir(path):
-                        raise CGroupsException("Create directory for cgroup {0}: normal file already exists with that name".format(path))
-                    else:
-                        pass  # There was a race to create the directory, but it's there now, and that's fine
-                elif e.errno == errno.EACCES:
-                    # This is unexpected, as the agent runs as root
-                    raise CGroupsException("Create directory for cgroup {0}: permission denied".format(path))
-                else:
-                    raise
-
-    @staticmethod
-    def _get_agent_cgroup_path(controller):
-        return os.path.join(CGROUPS_FILE_SYSTEM_ROOT, controller, VM_AGENT_CGROUP_NAME)
-
-    @staticmethod
-    def _get_extension_cgroups_root_path(controller):
-        return os.path.join(CGROUPS_FILE_SYSTEM_ROOT, controller, EXTENSIONS_ROOT_CGROUP_NAME)
-
-    def _get_extension_cgroup_path(self, controller, extension_name):
-        extensions_root = self._get_extension_cgroups_root_path(controller)
-
-        if not os.path.exists(extensions_root):
-            logger.warn("Root directory {0} does not exist.".format(extensions_root))
-
-        cgroup_name = self._get_extension_cgroup_name(extension_name)
+    def get_daemon_pid():
+        return int(fileutil.read_file(get_agent_pid_file_path()).strip())
 
-        return os.path.join(extensions_root, cgroup_name)
 
-    def _create_extension_cgroup(self, controller, extension_name):
-        return CGroup.create(self._get_extension_cgroup_path(controller, extension_name), controller, extension_name)
+class SystemdCgroupsApi(CGroupsApi):
+    """
+    Cgroups interface via systemd
+    """
 
-    @staticmethod
-    def _add_process_to_cgroup(pid, cgroup_path):
-        tasks_file = os.path.join(cgroup_path, 'cgroup.procs')
-        fileutil.append_file(tasks_file, "{0}\n".format(pid))
-        logger.info("Added PID {0} to cgroup {1}".format(pid, cgroup_path))
+    def __init__(self):
+        self._cgroup_mountpoints = None
+        self._agent_unit_name = None
+        self._systemd_run_commands = []
+        self._systemd_run_commands_lock = threading.RLock()
 
-    def cleanup_legacy_cgroups(self):
+    def get_systemd_run_commands(self):
         """
-        Previous versions of the daemon (2.2.31-2.2.40) wrote their PID to /sys/fs/cgroup/{cpu,memory}/WALinuxAgent/WALinuxAgent;
-        starting from version 2.2.41 we track the agent service in walinuxagent.service instead of WALinuxAgent/WALinuxAgent. This
-        method moves the daemon's PID from the legacy cgroups to the newer cgroups.
+        Returns a list of the systemd-run commands currently running (given as PIDs)
         """
-        def move_daemon_pid(controller, daemon_pid):
-            new_path = FileSystemCgroupsApi._get_agent_cgroup_path(controller)
-            logger.info("Writing daemon's PID ({0}) to {1}", daemon_pid, new_path)
-            fileutil.append_file(os.path.join(new_path, "cgroup.procs"), daemon_pid)
-            msg = "Moved daemon's PID from legacy cgroup to {0}".format(new_path)
-            add_event(AGENT_NAME, version=CURRENT_VERSION, op=WALAEventOperation.CGroupsCleanUp, is_success=True, message=msg)
+        with self._systemd_run_commands_lock:
+            return self._systemd_run_commands[:]
 
-        CGroupsApi._foreach_legacy_cgroup(move_daemon_pid)
-
-    def create_agent_cgroups(self):
+    def get_cgroup_mount_points(self):
         """
-        Creates a cgroup for the VM Agent in each of the controllers we are tracking; returns the created cgroups.
+        Returns a tuple with the mount points for the cpu and memory controllers; the values can be None
+        if the corresponding controller is not mounted
         """
-        cgroups = []
-
-        pid = int(os.getpid())
-
-        def create_cgroup(controller):
-            path = FileSystemCgroupsApi._get_agent_cgroup_path(controller)
-
-            if not os.path.isdir(path):
-                FileSystemCgroupsApi._try_mkdir(path)
-                logger.info("Created cgroup {0}".format(path))
-
-            self._add_process_to_cgroup(pid, path)
-
-            cgroups.append(CGroup.create(path, controller, VM_AGENT_CGROUP_NAME))
-
-        self._foreach_controller(create_cgroup, 'Failed to create a cgroup for the VM Agent; resource usage will not be tracked')
-
-        if len(cgroups) == 0:
-            raise CGroupsException("Failed to create any cgroup for the VM Agent")
+        # the output of mount is similar to
+        #     $ mount -t cgroup
+        #     cgroup on /sys/fs/cgroup/systemd type cgroup (rw,nosuid,nodev,noexec,relatime,xattr,name=systemd)
+        #     cgroup on /sys/fs/cgroup/cpu,cpuacct type cgroup (rw,nosuid,nodev,noexec,relatime,cpu,cpuacct)
+        #     cgroup on /sys/fs/cgroup/memory type cgroup (rw,nosuid,nodev,noexec,relatime,memory)
+        #     etc
+        #
+        if self._cgroup_mountpoints is None:
+            cpu = None
+            memory = None
+            for line in shellutil.run_command(['mount', '-t', 'cgroup']).splitlines():
+                match = re.search(r'on\s+(?P<path>/\S+(memory|cpuacct))\s', line)
+                if match is not None:
+                    path = match.group('path')
+                    if 'cpuacct' in path:
+                        cpu = path
+                    else:
+                        memory = path
+            self._cgroup_mountpoints = {'cpu': cpu, 'memory': memory}
 
-        return cgroups
+        return self._cgroup_mountpoints['cpu'], self._cgroup_mountpoints['memory']
 
-    def create_extension_cgroups_root(self):
+    @staticmethod
+    def get_process_cgroup_relative_paths(process_id):
         """
-        Creates the directory within the cgroups file system that will contain the cgroups for the extensions.
+        Returns a tuple with the path of the cpu and memory cgroups for the given process (relative to the mount point of the corresponding
+        controller).
+        The 'process_id' can be a numeric PID or the string "self" for the current process.
+        The values returned can be None if the process is not in a cgroup for that controller (e.g. the controller is not mounted).
         """
-        def create_cgroup(controller):
-            path = self._get_extension_cgroups_root_path(controller)
-
-            if not os.path.isdir(path):
-                FileSystemCgroupsApi._try_mkdir(path)
-                logger.info("Created {0}".format(path))
+        # The contents of the file are similar to
+        #    # cat /proc/1218/cgroup
+        #    10:memory:/system.slice/walinuxagent.service
+        #    3:cpu,cpuacct:/system.slice/walinuxagent.service
+        #    etc
+        cpu_path = None
+        memory_path = None
+        for line in fileutil.read_file("/proc/{0}/cgroup".format(process_id)).splitlines():
+            match = re.match(r'\d+:(?P<controller>(memory|.*cpuacct.*)):(?P<path>.+)', line)
+            if match is not None:
+                controller = match.group('controller')
+                path = match.group('path').lstrip('/') if match.group('path') != '/' else None
+                if controller == 'memory':
+                    memory_path = path
+                else:
+                    cpu_path = path
 
-        self._foreach_controller(create_cgroup, 'Failed to create a root cgroup for extensions')
+        return cpu_path, memory_path
 
-    def create_extension_cgroups(self, extension_name):
+    def get_process_cgroup_paths(self, process_id):
         """
-        Creates a cgroup for the given extension in each of the controllers we are tracking; returns the created cgroups.
+        Returns a tuple with the path of the cpu and memory cgroups for the given process. The 'process_id' can be a numeric PID or the string "self" for the current process.
+        The values returned can be None if the process is not in a cgroup for that controller (e.g. the controller is not mounted).
         """
-        cgroups = []
-
-        def create_cgroup(controller):
-            cgroup = self._create_extension_cgroup(controller, extension_name)
+        cpu_cgroup_relative_path, memory_cgroup_relative_path = self.get_process_cgroup_relative_paths(process_id)
 
-            if not os.path.isdir(cgroup.path):
-                FileSystemCgroupsApi._try_mkdir(cgroup.path)
-                logger.info("Created cgroup {0}".format(cgroup.path))
+        cpu_mount_point, memory_mount_point = self.get_cgroup_mount_points()
 
-            cgroups.append(cgroup)
+        cpu_cgroup_path = os.path.join(cpu_mount_point, cpu_cgroup_relative_path) \
+            if cpu_mount_point is not None and cpu_cgroup_relative_path is not None else None
 
-        self._foreach_controller(create_cgroup, 'Failed to create a cgroup for extension {0}'.format(extension_name))
+        memory_cgroup_path = os.path.join(memory_mount_point, memory_cgroup_relative_path) \
+            if memory_mount_point is not None and memory_cgroup_relative_path is not None else None
 
-        return cgroups
+        return cpu_cgroup_path, memory_cgroup_path
 
-    def remove_extension_cgroups(self, extension_name):
+    def get_unit_cgroup_paths(self, unit_name):
         """
-        Deletes the cgroups for the given extension.
+        Returns a tuple with the path of the cpu and memory cgroups for the given unit.
+        The values returned can be None if the controller is not mounted.
+        Ex: ControlGroup=/azure.slice/walinuxagent.service
+        controlgroup_path[1:] = azure.slice/walinuxagent.service
         """
-        def remove_cgroup(controller):
-            path = self._get_extension_cgroup_path(controller, extension_name)
-
-            if os.path.exists(path):
-                try:
-                    os.rmdir(path)
-                    logger.info('Deleted cgroup "{0}".'.format(path))
-                except OSError as exception:
-                    if exception.errno == 16:  # [Errno 16] Device or resource busy
-                        logger.warn('CGroup "{0}" still has active tasks; will not remove it.'.format(path))
-
-        self._foreach_controller(remove_cgroup, 'Failed to delete cgroups for extension {0}'.format(extension_name))
-
-    def get_extension_cgroups(self, extension_name):
-        """
-        Returns the cgroups for the given extension.
-        """
-
-        cgroups = []
+        controlgroup_path = systemd.get_unit_property(unit_name, "ControlGroup")
+        cpu_mount_point, memory_mount_point = self.get_cgroup_mount_points()
 
-        def get_cgroup(controller):
-            cgroup = self._create_extension_cgroup(controller, extension_name)
-            cgroups.append(cgroup)
+        cpu_cgroup_path = os.path.join(cpu_mount_point, controlgroup_path[1:]) \
+            if cpu_mount_point is not None else None
 
-        self._foreach_controller(get_cgroup, 'Failed to retrieve cgroups for extension {0}'.format(extension_name))
+        memory_cgroup_path = os.path.join(memory_mount_point, controlgroup_path[1:]) \
+            if memory_mount_point is not None else None
 
-        return cgroups
+        return cpu_cgroup_path, memory_cgroup_path
 
-    def start_extension_command(self, extension_name, command, timeout, shell, cwd, env, stdout, stderr,
-                                error_code=ExtensionErrorCodes.PluginUnknownFailure):
+    @staticmethod
+    def get_cgroup2_controllers():
         """
-        Starts a command (install/enable/etc) for an extension and adds the command's PID to the extension's cgroup
-        :param extension_name: The extension executing the command
-        :param command: The command to invoke
-        :param timeout: Number of seconds to wait for command completion
-        :param cwd: The working directory for the command
-        :param env: The environment to pass to the command's process
-        :param stdout: File object to redirect stdout to
-        :param stderr: File object to redirect stderr to
-        :param error_code: Extension error code to raise in case of error
+        Returns a tuple with the mount point for the cgroups v2 controllers, and the currently mounted controllers;
+        either value can be None if cgroups v2 or its controllers are not mounted
         """
-        try:
-            extension_cgroups = self.create_extension_cgroups(extension_name)
-        except Exception as exception:
-            extension_cgroups = []
-            logger.warn("Failed to create cgroups for extension '{0}'; resource usage will not be tracked. "
-                        "Error: {1}".format(extension_name, ustr(exception)))
-
-        def pre_exec_function():
-            os.setsid()
-
-            try:
-                pid = os.getpid()
-
-                for cgroup in extension_cgroups:
-                    try:
-                        self._add_process_to_cgroup(pid, cgroup.path)
-                    except Exception as exception:
-                        logger.warn("Failed to add PID {0} to the cgroups for extension '{1}'. "
-                                    "Resource usage will not be tracked. Error: {2}".format(pid,
-                                                                                            extension_name,
-                                                                                            ustr(exception)))
-            except Exception as e:
-                logger.warn("Failed to add extension {0} to its cgroup. Resource usage will not be tracked. "
-                            "Error: {1}".format(extension_name, ustr(e)))
-
-        process = subprocess.Popen(command,
-                                   shell=shell,
-                                   cwd=cwd,
-                                   env=env,
-                                   stdout=stdout,
-                                   stderr=stderr,
-                                   preexec_fn=pre_exec_function)
-
-        self.track_cgroups(extension_cgroups)
-        process_output = handle_process_completion(process=process,
-                                                   command=command,
-                                                   timeout=timeout,
-                                                   stdout=stdout,
-                                                   stderr=stderr,
-                                                   error_code=error_code)
-
-        return extension_cgroups, process_output
-
-
-class SystemdCgroupsApi(CGroupsApi):
-    """
-    Cgroups interface via systemd
-    """
-
-    @staticmethod
-    def create_and_start_unit(unit_filename, unit_contents):
-        try:
-            unit_path = os.path.join(UNIT_FILES_FILE_SYSTEM_PATH, unit_filename)
-            fileutil.write_file(unit_path, unit_contents)
-            shellutil.run_command(["systemctl", "daemon-reload"])
-            shellutil.run_command(["systemctl", "start", unit_filename])
-        except Exception as e:
-            raise CGroupsException("Failed to create and start {0}. Error: {1}".format(unit_filename, ustr(e)))
+        # the output of mount is similar to
+        #     $ mount -t cgroup2
+        #     cgroup2 on /sys/fs/cgroup/unified type cgroup2 (rw,nosuid,nodev,noexec,relatime,nsdelegate)
+        #
+        for line in shellutil.run_command(['mount', '-t', 'cgroup2']).splitlines():
+            match = re.search(r'on\s+(?P<path>/\S+)\s', line)
+            if match is not None:
+                mount_point = match.group('path')
+                controllers = None
+                controllers_file = os.path.join(mount_point, 'cgroup.controllers')
+                if os.path.exists(controllers_file):
+                    controllers = fileutil.read_file(controllers_file)
+                return mount_point, controllers
+        return None, None
 
     @staticmethod
-    def _get_extensions_slice_root_name():
-        return "system-{0}.slice".format(EXTENSIONS_ROOT_CGROUP_NAME)
-
-    def _get_extension_slice_name(self, extension_name):
-        return "system-{0}-{1}.slice".format(EXTENSIONS_ROOT_CGROUP_NAME, self._get_extension_cgroup_name(extension_name))
-
-    def create_agent_cgroups(self):
-        try:
-            cgroup_unit = None
-            cgroup_paths = fileutil.read_file("/proc/self/cgroup")
-            for entry in cgroup_paths.splitlines():
-                fields = entry.split(':')
-                if fields[1] == "name=systemd":
-                    cgroup_unit = fields[2].lstrip(os.path.sep)
-
-            cpu_cgroup_path = os.path.join(CGROUPS_FILE_SYSTEM_ROOT, 'cpu', cgroup_unit)
-            memory_cgroup_path = os.path.join(CGROUPS_FILE_SYSTEM_ROOT, 'memory', cgroup_unit)
-
-            return [CGroup.create(cpu_cgroup_path, 'cpu', VM_AGENT_CGROUP_NAME),
-                    CGroup.create(memory_cgroup_path, 'memory', VM_AGENT_CGROUP_NAME)]
-        except Exception as e:
-            raise CGroupsException("Failed to get paths of agent's cgroups. Error: {0}".format(ustr(e)))
-
-    def create_extension_cgroups_root(self):
-        unit_contents = """
-[Unit]
-Description=Slice for walinuxagent extensions
-DefaultDependencies=no
-Before=slices.target
-Requires=system.slice
-After=system.slice"""
-        unit_filename = self._get_extensions_slice_root_name()
-        self.create_and_start_unit(unit_filename, unit_contents)
-        logger.info("Created slice for walinuxagent extensions {0}".format(unit_filename))
-
-    def create_extension_cgroups(self, extension_name):
-        # TODO: The slice created by this function is not used currently. We need to create the extension scopes within
-        #  this slice and use the slice to monitor the cgroups. Also see comment in get_extension_cgroups.
-        # the slice.
-        unit_contents = """
-[Unit]
-Description=Slice for extension {0}
-DefaultDependencies=no
-Before=slices.target
-Requires=system-{1}.slice
-After=system-{1}.slice""".format(extension_name, EXTENSIONS_ROOT_CGROUP_NAME)
-        unit_filename = self._get_extension_slice_name(extension_name)
-        self.create_and_start_unit(unit_filename, unit_contents)
-        logger.info("Created slice for {0}".format(unit_filename))
-
-        return self.get_extension_cgroups(extension_name)
-
-    def remove_extension_cgroups(self, extension_name):
-        # For transient units, cgroups are released automatically when the unit stops, so it is sufficient
-        # to call stop on them. Persistent cgroups are released when the unit is disabled and its configuration
-        # file is deleted.
-        # The assumption is that this method is called after the extension has been uninstalled. For now, since
-        # we're running extensions within transient scopes which clean up after they finish running, no removal
-        # of units is needed. In the future, when the extension is running under its own slice,
-        # the following clean up is needed.
-        unit_filename = self._get_extension_slice_name(extension_name)
-        try:
-            unit_path = os.path.join(UNIT_FILES_FILE_SYSTEM_PATH, unit_filename)
-            shellutil.run_command(["systemctl", "stop", unit_filename])
-            fileutil.rm_files(unit_path)
-            shellutil.run_command(["systemctl", "daemon-reload"])
-        except Exception as e:
-            raise CGroupsException("Failed to remove {0}. Error: {1}".format(unit_filename, ustr(e)))
-
-    def get_extension_cgroups(self, extension_name):
-        # TODO: The slice returned by this function is not used currently. We need to create the extension scopes within
-        #  this slice and use the slice to monitor the cgroups. Also see comment in create_extension_cgroups.
-        slice_name = self._get_extension_cgroup_name(extension_name)
-
-        cgroups = []
-
-        def create_cgroup(controller):
-            cpu_cgroup_path = os.path.join(CGROUPS_FILE_SYSTEM_ROOT, controller, 'system.slice', slice_name)
-            cgroups.append(CGroup.create(cpu_cgroup_path, controller, extension_name))
-
-        self._foreach_controller(create_cgroup, 'Cannot retrieve cgroup for extension {0}; resource usage will not be tracked.'.format(extension_name))
-
-        return cgroups
+    def _is_systemd_failure(scope_name, stderr):
+        stderr.seek(0)
+        stderr = ustr(stderr.read(TELEMETRY_MESSAGE_MAX_LEN), encoding='utf-8', errors='backslashreplace')
+        unit_not_found = "Unit {0} not found.".format(scope_name)
+        return unit_not_found in stderr or scope_name not in stderr
 
     @staticmethod
-    def _is_systemd_failure(scope_name, process_output):
-        unit_not_found = "Unit {0} not found.".format(scope_name)
-        return unit_not_found in process_output or scope_name not in process_output
+    def get_extension_slice_name(extension_name, old_slice=False):
+        # The old slice makes it difficult for user to override the limits because they need to place drop-in files on every upgrade if extension slice is different for each version.
+        # old slice includes <HandlerName>.<ExtensionName>-<HandlerVersion>
+        # new slice without version <HandlerName>.<ExtensionName>
+        if not old_slice:
+            extension_name = extension_name.rsplit("-", 1)[0]
+        # Since '-' is used as a separator in systemd unit names, we replace it with '_' to prevent side-effects.
+        return EXTENSION_SLICE_PREFIX + "-" + extension_name.replace('-', '_') + ".slice"
 
-    def start_extension_command(self, extension_name, command, timeout, shell, cwd, env, stdout, stderr,
+    def start_extension_command(self, extension_name, command, cmd_name, timeout, shell, cwd, env, stdout, stderr,
                                 error_code=ExtensionErrorCodes.PluginUnknownFailure):
-        scope_name = "{0}_{1}".format(self._get_extension_cgroup_name(extension_name), uuid.uuid4())
-
-        process = subprocess.Popen(
-            "systemd-run --unit={0} --scope {1}".format(scope_name, command),
-            shell=shell,
-            cwd=cwd,
-            stdout=stdout,
-            stderr=stderr,
-            env=env,
-            preexec_fn=os.setsid)
+        scope = "{0}_{1}".format(cmd_name, uuid.uuid4())
+        extension_slice_name = self.get_extension_slice_name(extension_name)
+        with self._systemd_run_commands_lock:
+            process = subprocess.Popen(  # pylint: disable=W1509
+                # Some distros like ubuntu20 by default cpu and memory accounting enabled. Thus create nested cgroups under the extension slice
+                # So disabling CPU and Memory accounting prevents from creating nested cgroups, so that all the counters will be present in extension Cgroup
+                # since slice unit file configured with accounting enabled.
+                "systemd-run --property=CPUAccounting=no --property=MemoryAccounting=no --unit={0} --scope --slice={1} {2}".format(scope, extension_slice_name, command),
+                shell=shell,
+                cwd=cwd,
+                stdout=stdout,
+                stderr=stderr,
+                env=env,
+                preexec_fn=os.setsid)
+
+            # We start systemd-run with shell == True so process.pid is the shell's pid, not the pid for systemd-run
+            self._systemd_run_commands.append(process.pid)
+
+        scope_name = scope + '.scope'
+
+        logger.info("Started extension in unit '{0}'", scope_name)
+
+        cpu_cgroup = None
+        try:
+            cgroup_relative_path = os.path.join('azure.slice/azure-vmextensions.slice', extension_slice_name)
 
-        logger.info("Started extension using scope '{0}'", scope_name)
-        extension_cgroups = []
+            cpu_cgroup_mountpoint, memory_cgroup_mountpoint = self.get_cgroup_mount_points()
 
-        def create_cgroup(controller):
-            cgroup_path = os.path.join(CGROUPS_FILE_SYSTEM_ROOT, controller, 'system.slice', scope_name + ".scope")
-            extension_cgroups.append(CGroup.create(cgroup_path, controller, extension_name))
+            if cpu_cgroup_mountpoint is None:
+                logger.info("The CPU controller is not mounted; will not track resource usage")
+            else:
+                cpu_cgroup_path = os.path.join(cpu_cgroup_mountpoint, cgroup_relative_path)
+                cpu_cgroup = CpuCgroup(extension_name, cpu_cgroup_path)
+                CGroupsTelemetry.track_cgroup(cpu_cgroup)
 
-        self._foreach_controller(create_cgroup, 'Cannot create cgroup for extension {0}; '
-                                                'resource usage will not be tracked.'.format(extension_name))
-        self.track_cgroups(extension_cgroups)
+            if memory_cgroup_mountpoint is None:
+                logger.info("The Memory controller is not mounted; will not track resource usage")
+            else:
+                memory_cgroup_path = os.path.join(memory_cgroup_mountpoint, cgroup_relative_path)
+                memory_cgroup = MemoryCgroup(extension_name, memory_cgroup_path)
+                CGroupsTelemetry.track_cgroup(memory_cgroup)
+
+        except IOError as e:
+            if e.errno == 2:  # 'No such file or directory'
+                logger.info("The extension command already completed; will not track resource usage")
+            logger.info("Failed to start tracking resource usage for the extension: {0}", ustr(e))
+        except Exception as e:
+            logger.info("Failed to start tracking resource usage for the extension: {0}", ustr(e))
 
         # Wait for process completion or timeout
         try:
-            process_output = handle_process_completion(process=process,
-                                                       command=command,
-                                                       timeout=timeout,
-                                                       stdout=stdout,
-                                                       stderr=stderr,
-                                                       error_code=error_code)
+            return handle_process_completion(process=process, command=command, timeout=timeout, stdout=stdout,
+                                             stderr=stderr, error_code=error_code, cpu_cgroup=cpu_cgroup)
         except ExtensionError as e:
             # The extension didn't terminate successfully. Determine whether it was due to systemd errors or
             # extension errors.
-            process_output = read_output(stdout, stderr)
-            systemd_failure = self._is_systemd_failure(scope_name, process_output)
-
-            if not systemd_failure:
+            if not self._is_systemd_failure(scope, stderr):
                 # There was an extension error; it either timed out or returned a non-zero exit code. Re-raise the error
                 raise
+
+            # There was an issue with systemd-run. We need to log it and retry the extension without systemd.
+            process_output = read_output(stdout, stderr)
+            # Reset the stdout and stderr
+            stdout.truncate(0)
+            stderr.truncate(0)
+
+            if isinstance(e, ExtensionOperationError):
+                # no-member: Instance of 'ExtensionError' has no 'exit_code' member (no-member) - Disabled: e is actually an ExtensionOperationError
+                err_msg = 'Systemd process exited with code %s and output %s' % (
+                    e.exit_code, process_output)  # pylint: disable=no-member
             else:
-                # There was an issue with systemd-run. We need to log it and retry the extension without systemd.
-                err_msg = 'Systemd process exited with code %s and output %s' % (e.exit_code, process_output) \
-                    if isinstance(e, ExtensionOperationError) else "Systemd timed-out, output: %s" % process_output
-                event_msg = 'Failed to run systemd-run for unit {0}.scope. ' \
-                            'Will retry invoking the extension without systemd. ' \
-                            'Systemd-run error: {1}'.format(scope_name, err_msg)
-                add_event(AGENT_NAME,
-                          version=CURRENT_VERSION,
-                          op=WALAEventOperation.InvokeCommandUsingSystemd,
-                          is_success=False,
-                          log_event=False,
-                          message=event_msg)
-                logger.warn(event_msg)
-
-                # Reset the stdout and stderr
-                stdout.truncate(0)
-                stderr.truncate(0)
-
-                # Try invoking the process again, this time without systemd-run
-                logger.info('Extension invocation using systemd failed, falling back to regular invocation '
-                            'without cgroups tracking.')
-                process = subprocess.Popen(command,
-                                           shell=shell,
-                                           cwd=cwd,
-                                           env=env,
-                                           stdout=stdout,
-                                           stderr=stderr,
-                                           preexec_fn=os.setsid)
-
-                process_output = handle_process_completion(process=process,
-                                                           command=command,
-                                                           timeout=timeout,
-                                                           stdout=stdout,
-                                                           stderr=stderr,
-                                                           error_code=error_code)
-
-                return [], process_output
-
-        # The process terminated in time and successfully
-        return extension_cgroups, process_output
+                err_msg = "Systemd timed-out, output: %s" % process_output
+            raise SystemdRunError(err_msg)
+        finally:
+            with self._systemd_run_commands_lock:
+                self._systemd_run_commands.remove(process.pid)
 
     def cleanup_legacy_cgroups(self):
         """
         Previous versions of the daemon (2.2.31-2.2.40) wrote their PID to /sys/fs/cgroup/{cpu,memory}/WALinuxAgent/WALinuxAgent;
         starting from version 2.2.41 we track the agent service in walinuxagent.service instead of WALinuxAgent/WALinuxAgent. If
-        we find that any of the legacy groups include the PID of the daemon then we disable data collection for this instance
-        (under systemd, moving PIDs across the cgroup file system can produce unpredictable results)
+        we find that any of the legacy groups include the PID of the daemon then we need to disable data collection for this
+        instance (under systemd, moving PIDs across the cgroup file system can produce unpredictable results)
         """
-        def report_error(_, daemon_pid):
-            raise CGroupsException(
-                "The daemon's PID ({0}) was already added to the legacy cgroup; this invalidates resource usage data.".format(daemon_pid))
-
-        CGroupsApi._foreach_legacy_cgroup(report_error)
+        return CGroupsApi._foreach_legacy_cgroup(lambda *_: None)
diff --git a/azurelinuxagent/common/cgroupconfigurator.py b/azurelinuxagent/common/cgroupconfigurator.py
index ea6983f..767786f 100644
--- a/azurelinuxagent/common/cgroupconfigurator.py
+++ b/azurelinuxagent/common/cgroupconfigurator.py
@@ -1,3 +1,4 @@
+# -*- encoding: utf-8 -*-
 # Copyright 2018 Microsoft Corporation
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -13,157 +14,870 @@
 # limitations under the License.
 #
 # Requires Python 2.6+ and Openssl 1.0+
-
+import glob
+import json
 import os
+import re
 import subprocess
+import threading
 
+from azurelinuxagent.common import conf
 from azurelinuxagent.common import logger
-from azurelinuxagent.common.cgroupapi import CGroupsApi
+from azurelinuxagent.common.cgroup import CpuCgroup, AGENT_NAME_TELEMETRY, MetricsCounter, MemoryCgroup
+from azurelinuxagent.common.cgroupapi import CGroupsApi, SystemdCgroupsApi, SystemdRunError, EXTENSION_SLICE_PREFIX
 from azurelinuxagent.common.cgroupstelemetry import CGroupsTelemetry
-from azurelinuxagent.common.exception import CGroupsException, ExtensionErrorCodes
+from azurelinuxagent.common.exception import ExtensionErrorCodes, CGroupsException, AgentMemoryExceededException
 from azurelinuxagent.common.future import ustr
-from azurelinuxagent.common.osutil import get_osutil
+from azurelinuxagent.common.osutil import get_osutil, systemd
+from azurelinuxagent.common.version import get_distro
+from azurelinuxagent.common.utils import shellutil, fileutil
 from azurelinuxagent.common.utils.extensionprocessutil import handle_process_completion
-from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
 from azurelinuxagent.common.event import add_event, WALAEventOperation
 
+AZURE_SLICE = "azure.slice"
+_AZURE_SLICE_CONTENTS = """
+[Unit]
+Description=Slice for Azure VM Agent and Extensions
+DefaultDependencies=no
+Before=slices.target
+"""
+_VMEXTENSIONS_SLICE = EXTENSION_SLICE_PREFIX + ".slice"
+_AZURE_VMEXTENSIONS_SLICE = AZURE_SLICE + "/" + _VMEXTENSIONS_SLICE
+_VMEXTENSIONS_SLICE_CONTENTS = """
+[Unit]
+Description=Slice for Azure VM Extensions
+DefaultDependencies=no
+Before=slices.target
+[Slice]
+CPUAccounting=yes
+MemoryAccounting=yes
+"""
+_EXTENSION_SLICE_CONTENTS = """
+[Unit]
+Description=Slice for Azure VM extension {extension_name}
+DefaultDependencies=no
+Before=slices.target
+[Slice]
+CPUAccounting=yes
+CPUQuota={cpu_quota}
+MemoryAccounting=yes
+"""
+LOGCOLLECTOR_SLICE = "azure-walinuxagent-logcollector.slice"
+# More info on resource limits properties in systemd here:
+# https://access.redhat.com/documentation/en-us/red_hat_enterprise_linux/7/html/resource_management_guide/sec-modifying_control_groups
+_LOGCOLLECTOR_SLICE_CONTENTS_FMT = """
+[Unit]
+Description=Slice for Azure VM Agent Periodic Log Collector
+DefaultDependencies=no
+Before=slices.target
+[Slice]
+CPUAccounting=yes
+CPUQuota={cpu_quota}
+MemoryAccounting=yes
+"""
+_LOGCOLLECTOR_CPU_QUOTA = "5%"
+LOGCOLLECTOR_MEMORY_LIMIT = 30 * 1024 ** 2  # 30Mb
+
+_AGENT_DROP_IN_FILE_SLICE = "10-Slice.conf"
+_AGENT_DROP_IN_FILE_SLICE_CONTENTS = """
+# This drop-in unit file was created by the Azure VM Agent.
+# Do not edit.
+[Service]
+Slice=azure.slice
+"""
+_DROP_IN_FILE_CPU_ACCOUNTING = "11-CPUAccounting.conf"
+_DROP_IN_FILE_CPU_ACCOUNTING_CONTENTS = """
+# This drop-in unit file was created by the Azure VM Agent.
+# Do not edit.
+[Service]
+CPUAccounting=yes
+"""
+_DROP_IN_FILE_CPU_QUOTA = "12-CPUQuota.conf"
+_DROP_IN_FILE_CPU_QUOTA_CONTENTS_FORMAT = """
+# This drop-in unit file was created by the Azure VM Agent.
+# Do not edit.
+[Service]
+CPUQuota={0}
+"""
+_DROP_IN_FILE_MEMORY_ACCOUNTING = "13-MemoryAccounting.conf"
+_DROP_IN_FILE_MEMORY_ACCOUNTING_CONTENTS = """
+# This drop-in unit file was created by the Azure VM Agent.
+# Do not edit.
+[Service]
+MemoryAccounting=yes
+"""
+
+
+class DisableCgroups(object):
+    ALL = "all"
+    AGENT = "agent"
+    EXTENSIONS = "extensions"
+
+
+def _log_cgroup_info(format_string, *args):
+    message = format_string.format(*args)
+    logger.info("[CGI] " + message)
+    add_event(op=WALAEventOperation.CGroupsInfo, message=message)
+
+
+def _log_cgroup_warning(format_string, *args):
+    message = format_string.format(*args)
+    logger.info("[CGW] " + message)  # log as INFO for now, in the future it should be logged as WARNING
+    add_event(op=WALAEventOperation.CGroupsInfo, message=message, is_success=False, log_event=False)
+
 
 class CGroupConfigurator(object):
     """
     This class implements the high-level operations on CGroups (e.g. initialization, creation, etc)
 
-    NOTE: with the exception of start_extension_command, none of the methods in this class raise exceptions (cgroup operations should not block extensions)
+    NOTE: with the exception of start_extension_command, none of the methods in this class
+    raise exceptions (cgroup operations should not block extensions)
     """
-    class __impl(object):
+
+    class _Impl(object):
         def __init__(self):
+            self._initialized = False
+            self._cgroups_supported = False
+            self._agent_cgroups_enabled = False
+            self._extensions_cgroups_enabled = False
+            self._cgroups_api = None
+            self._agent_cpu_cgroup_path = None
+            self._agent_memory_cgroup_path = None
+            self._agent_memory_cgroup = None
+            self._check_cgroups_lock = threading.RLock()  # Protect the check_cgroups which is called from Monitor thread and main loop.
+
+        def initialize(self):
+            try:
+                if self._initialized:
+                    return
+                # This check is to reset the quotas if agent goes from cgroup supported to unsupported distros later in time.
+                if not CGroupsApi.cgroups_supported():
+                    agent_drop_in_path = systemd.get_agent_drop_in_path()
+                    try:
+                        if os.path.exists(agent_drop_in_path) and os.path.isdir(agent_drop_in_path):
+                            files_to_cleanup = []
+                            agent_drop_in_file_slice = os.path.join(agent_drop_in_path, _AGENT_DROP_IN_FILE_SLICE)
+                            agent_drop_in_file_cpu_accounting = os.path.join(agent_drop_in_path,
+                                                                             _DROP_IN_FILE_CPU_ACCOUNTING)
+                            agent_drop_in_file_memory_accounting = os.path.join(agent_drop_in_path,
+                                                                                _DROP_IN_FILE_MEMORY_ACCOUNTING)
+                            agent_drop_in_file_cpu_quota = os.path.join(agent_drop_in_path, _DROP_IN_FILE_CPU_QUOTA)
+                            files_to_cleanup.extend([agent_drop_in_file_slice, agent_drop_in_file_cpu_accounting,
+                                                     agent_drop_in_file_memory_accounting, agent_drop_in_file_cpu_quota])
+                            self.__cleanup_all_files(files_to_cleanup)
+                            self.__reload_systemd_config()
+                            logger.info("Agent reset the quotas if distro: {0} goes from supported to unsupported list", get_distro())
+                    except Exception as err:
+                        logger.warn("Unable to delete Agent drop-in files while resetting the quotas: {0}".format(err))
+
+                # check whether cgroup monitoring is supported on the current distro
+                self._cgroups_supported = CGroupsApi.cgroups_supported()
+                if not self._cgroups_supported:
+                    logger.info("Cgroup monitoring is not supported on {0}", get_distro())
+                    return
+
+                # check that systemd is detected correctly
+                self._cgroups_api = SystemdCgroupsApi()
+                if not systemd.is_systemd():
+                    _log_cgroup_warning("systemd was not detected on {0}", get_distro())
+                    return
+
+                _log_cgroup_info("systemd version: {0}", systemd.get_version())
+
+                # This is temporarily disabled while we analyze telemetry. Likely it will be removed.
+                # self.__collect_azure_unit_telemetry()
+                # self.__collect_agent_unit_files_telemetry()
+
+                if not self.__check_no_legacy_cgroups():
+                    return
+
+                agent_unit_name = systemd.get_agent_unit_name()
+                agent_slice = systemd.get_unit_property(agent_unit_name, "Slice")
+                if agent_slice not in (AZURE_SLICE, "system.slice"):
+                    _log_cgroup_warning("The agent is within an unexpected slice: {0}", agent_slice)
+                    return
+
+                self.__setup_azure_slice()
+
+                cpu_controller_root, memory_controller_root = self.__get_cgroup_controllers()
+                self._agent_cpu_cgroup_path, self._agent_memory_cgroup_path = self.__get_agent_cgroups(agent_slice,
+                                                                                                       cpu_controller_root,
+                                                                                                       memory_controller_root)
+
+                if self._agent_cpu_cgroup_path is not None or self._agent_memory_cgroup_path is not None:
+                    self.enable()
+
+                if self._agent_cpu_cgroup_path is not None:
+                    _log_cgroup_info("Agent CPU cgroup: {0}", self._agent_cpu_cgroup_path)
+                    self.__set_cpu_quota(conf.get_agent_cpu_quota())
+                    CGroupsTelemetry.track_cgroup(CpuCgroup(AGENT_NAME_TELEMETRY, self._agent_cpu_cgroup_path))
+
+                if self._agent_memory_cgroup_path is not None:
+                    _log_cgroup_info("Agent Memory cgroup: {0}", self._agent_memory_cgroup_path)
+                    self._agent_memory_cgroup = MemoryCgroup(AGENT_NAME_TELEMETRY, self._agent_memory_cgroup_path)
+                    CGroupsTelemetry.track_cgroup(self._agent_memory_cgroup)
+
+                _log_cgroup_info('Agent cgroups enabled: {0}', self._agent_cgroups_enabled)
+
+            except Exception as exception:
+                _log_cgroup_warning("Error initializing cgroups: {0}", ustr(exception))
+            finally:
+                self._initialized = True
+
+        @staticmethod
+        def __collect_azure_unit_telemetry():
+            azure_units = []
+
+            try:
+                units = shellutil.run_command(['systemctl', 'list-units', 'azure*', '-all'])
+                for line in units.split('\n'):
+                    match = re.match(r'\s?(azure[^\s]*)\s?', line, re.IGNORECASE)
+                    if match is not None:
+                        azure_units.append((match.group(1), line))
+            except shellutil.CommandError as command_error:
+                _log_cgroup_warning("Failed to list systemd units: {0}", ustr(command_error))
+
+            for unit_name, unit_description in azure_units:
+                unit_slice = "Unknown"
+                try:
+                    unit_slice = systemd.get_unit_property(unit_name, "Slice")
+                except Exception as exception:
+                    _log_cgroup_warning("Failed to query Slice for {0}: {1}", unit_name, ustr(exception))
+
+                _log_cgroup_info("Found an Azure unit under slice {0}: {1}", unit_slice, unit_description)
+
+            if len(azure_units) == 0:
+                try:
+                    cgroups = shellutil.run_command('systemd-cgls')
+                    for line in cgroups.split('\n'):
+                        if re.match(r'[^\x00-\xff]+azure\.slice\s*', line, re.UNICODE):
+                            logger.info(ustr("Found a cgroup for azure.slice\n{0}").format(cgroups))
+                            # Don't add the output of systemd-cgls to the telemetry, since currently it does not support Unicode
+                            add_event(op=WALAEventOperation.CGroupsInfo, message="Found a cgroup for azure.slice")
+                except shellutil.CommandError as command_error:
+                    _log_cgroup_warning("Failed to list systemd units: {0}", ustr(command_error))
+
+        @staticmethod
+        def __collect_agent_unit_files_telemetry():
+            agent_unit_files = []
+            agent_service_name = get_osutil().get_service_name()
+            try:
+                fragment_path = systemd.get_unit_property(agent_service_name, "FragmentPath")
+                if fragment_path != systemd.get_agent_unit_file():
+                    agent_unit_files.append(fragment_path)
+            except Exception as exception:
+                _log_cgroup_warning("Failed to query the agent's FragmentPath: {0}", ustr(exception))
+
+            try:
+                drop_in_paths = systemd.get_unit_property(agent_service_name, "DropInPaths")
+                for path in drop_in_paths.split():
+                    agent_unit_files.append(path)
+            except Exception as exception:
+                _log_cgroup_warning("Failed to query the agent's DropInPaths: {0}", ustr(exception))
+
+            for unit_file in agent_unit_files:
+                try:
+                    with open(unit_file, "r") as file_object:
+                        _log_cgroup_info("Found a custom unit file for the agent: {0}\n{1}", unit_file,
+                                         file_object.read())
+                except Exception as exception:
+                    _log_cgroup_warning("Can't read {0}: {1}", unit_file, ustr(exception))
+
+        def __check_no_legacy_cgroups(self):
+            """
+            Older versions of the daemon (2.2.31-2.2.40) wrote their PID to /sys/fs/cgroup/{cpu,memory}/WALinuxAgent/WALinuxAgent. When running
+            under systemd this could produce invalid resource usage data. Cgroups should not be enabled under this condition.
+            """
+            legacy_cgroups = self._cgroups_api.cleanup_legacy_cgroups()
+            if legacy_cgroups > 0:
+                _log_cgroup_warning("The daemon's PID was added to a legacy cgroup; will not monitor resource usage.")
+                return False
+            return True
+
+        def __get_cgroup_controllers(self):
+            #
+            # check v1 controllers
+            #
+            cpu_controller_root, memory_controller_root = self._cgroups_api.get_cgroup_mount_points()
+
+            if cpu_controller_root is not None:
+                logger.info("The CPU cgroup controller is mounted at {0}", cpu_controller_root)
+            else:
+                _log_cgroup_warning("The CPU cgroup controller is not mounted")
+
+            if memory_controller_root is not None:
+                logger.info("The memory cgroup controller is mounted at {0}", memory_controller_root)
+            else:
+                _log_cgroup_warning("The memory cgroup controller is not mounted")
+
+            #
+            # check v2 controllers
+            #
+            cgroup2_mount_point, cgroup2_controllers = self._cgroups_api.get_cgroup2_controllers()
+            if cgroup2_mount_point is not None:
+                _log_cgroup_info("cgroups v2 mounted at {0}.  Controllers: [{1}]", cgroup2_mount_point,
+                                 cgroup2_controllers)
+
+            return cpu_controller_root, memory_controller_root
+
+        @staticmethod
+        def __setup_azure_slice():
             """
-            Ensures the cgroups file system is mounted and selects the correct API to interact with it
+            The agent creates "azure.slice" for use by extensions and the agent. The agent runs under "azure.slice" directly and each
+            extension runs under its own slice ("Microsoft.CPlat.Extension.slice" in the example below). All the slices for
+            extensions are grouped under "vmextensions.slice".
+
+            Example:  -.slice
+                      ├─user.slice
+                      ├─system.slice
+                      └─azure.slice
+                        ├─walinuxagent.service
+                        │ ├─5759 /usr/bin/python3 -u /usr/sbin/waagent -daemon
+                        │ └─5764 python3 -u bin/WALinuxAgent-2.2.53-py2.7.egg -run-exthandlers
+                        └─azure-vmextensions.slice
+                          └─Microsoft.CPlat.Extension.slice
+                              └─5894 /usr/bin/python3 /var/lib/waagent/Microsoft.CPlat.Extension-1.0.0.0/enable.py
+
+            This method ensures that the "azure" and "vmextensions" slices are created. Setup should create those slices
+            under /lib/systemd/system; but if they do not exist, __ensure_azure_slices_exist will create them.
+
+            It also creates drop-in files to set the agent's Slice and CPUAccounting if they have not been
+            set up in the agent's unit file.
+
+            Lastly, the method also cleans up unit files left over from previous versions of the agent.
             """
-            osutil = get_osutil()
 
-            self._cgroups_supported = osutil.is_cgroups_supported()
+            # Older agents used to create this slice, but it was never used. Cleanup the file.
+            CGroupConfigurator._Impl.__cleanup_unit_file("/etc/systemd/system/system-walinuxagent.extensions.slice")
+
+            unit_file_install_path = systemd.get_unit_file_install_path()
+            azure_slice = os.path.join(unit_file_install_path, AZURE_SLICE)
+            vmextensions_slice = os.path.join(unit_file_install_path, _VMEXTENSIONS_SLICE)
+            logcollector_slice = os.path.join(unit_file_install_path, LOGCOLLECTOR_SLICE)
+            agent_unit_file = systemd.get_agent_unit_file()
+            agent_drop_in_path = systemd.get_agent_drop_in_path()
+            agent_drop_in_file_slice = os.path.join(agent_drop_in_path, _AGENT_DROP_IN_FILE_SLICE)
+            agent_drop_in_file_cpu_accounting = os.path.join(agent_drop_in_path, _DROP_IN_FILE_CPU_ACCOUNTING)
+            agent_drop_in_file_memory_accounting = os.path.join(agent_drop_in_path, _DROP_IN_FILE_MEMORY_ACCOUNTING)
+
+            files_to_create = []
+
+            if not os.path.exists(azure_slice):
+                files_to_create.append((azure_slice, _AZURE_SLICE_CONTENTS))
+
+            if not os.path.exists(vmextensions_slice):
+                files_to_create.append((vmextensions_slice, _VMEXTENSIONS_SLICE_CONTENTS))
+
+            # Update log collector slice contents
+            slice_contents = _LOGCOLLECTOR_SLICE_CONTENTS_FMT.format(cpu_quota=_LOGCOLLECTOR_CPU_QUOTA)
+            files_to_create.append((logcollector_slice, slice_contents))
+
+            if fileutil.findre_in_file(agent_unit_file, r"Slice=") is not None:
+                CGroupConfigurator._Impl.__cleanup_unit_file(agent_drop_in_file_slice)
+            else:
+                if not os.path.exists(agent_drop_in_file_slice):
+                    files_to_create.append((agent_drop_in_file_slice, _AGENT_DROP_IN_FILE_SLICE_CONTENTS))
+
+            if fileutil.findre_in_file(agent_unit_file, r"CPUAccounting=") is not None:
+                CGroupConfigurator._Impl.__cleanup_unit_file(agent_drop_in_file_cpu_accounting)
+            else:
+                if not os.path.exists(agent_drop_in_file_cpu_accounting):
+                    files_to_create.append((agent_drop_in_file_cpu_accounting, _DROP_IN_FILE_CPU_ACCOUNTING_CONTENTS))
+
+            if fileutil.findre_in_file(agent_unit_file, r"MemoryAccounting=") is not None:
+                CGroupConfigurator._Impl.__cleanup_unit_file(agent_drop_in_file_memory_accounting)
+            else:
+                if not os.path.exists(agent_drop_in_file_memory_accounting):
+                    files_to_create.append(
+                        (agent_drop_in_file_memory_accounting, _DROP_IN_FILE_MEMORY_ACCOUNTING_CONTENTS))
+
+            if len(files_to_create) > 0:
+                # create the unit files, but if 1 fails remove all and return
+                try:
+                    for path, contents in files_to_create:
+                        CGroupConfigurator._Impl.__create_unit_file(path, contents)
+                except Exception as exception:
+                    _log_cgroup_warning("Failed to create unit files for the azure slice: {0}", ustr(exception))
+                    for unit_file in files_to_create:
+                        CGroupConfigurator._Impl.__cleanup_unit_file(unit_file)
+                    return
+
+                CGroupConfigurator._Impl.__reload_systemd_config()
 
-            if self._cgroups_supported:
-                self._enabled = True
+        @staticmethod
+        def __reload_systemd_config():
+            # reload the systemd configuration; the new slices will be used once the agent's service restarts
+            try:
+                logger.info("Executing systemctl daemon-reload...")
+                shellutil.run_command(["systemctl", "daemon-reload"])
+            except Exception as exception:
+                _log_cgroup_warning("daemon-reload failed (create azure slice): {0}", ustr(exception))
+
+        @staticmethod
+        def __create_unit_file(path, contents):
+            parent, _ = os.path.split(path)
+            if not os.path.exists(parent):
+                fileutil.mkdir(parent, mode=0o755)
+            exists = os.path.exists(path)
+            fileutil.write_file(path, contents)
+            _log_cgroup_info("{0} {1}", "Updated" if exists else "Created", path)
+
+        @staticmethod
+        def __cleanup_unit_file(path):
+            if os.path.exists(path):
                 try:
-                    osutil.mount_cgroups()
-                    self._cgroups_api = CGroupsApi.create()
-                    status = "The cgroup filesystem is ready to use"
-                except Exception as e:
-                    status = ustr(e)
-                    self._enabled = False
+                    os.remove(path)
+                    _log_cgroup_info("Removed {0}", path)
+                except Exception as exception:
+                    _log_cgroup_warning("Failed to remove {0}: {1}", path, ustr(exception))
+
+        @staticmethod
+        def __cleanup_all_files(files_to_cleanup):
+            for path in files_to_cleanup:
+                if os.path.exists(path):
+                    try:
+                        os.remove(path)
+                        _log_cgroup_info("Removed {0}", path)
+                    except Exception as exception:
+                        _log_cgroup_warning("Failed to remove {0}: {1}", path, ustr(exception))
+
+        @staticmethod
+        def __create_all_files(files_to_create):
+            # create the unit files, but if 1 fails remove all and return
+            try:
+                for path, contents in files_to_create:
+                    CGroupConfigurator._Impl.__create_unit_file(path, contents)
+            except Exception as exception:
+                _log_cgroup_warning("Failed to create unit files : {0}", ustr(exception))
+                for unit_file in files_to_create:
+                    CGroupConfigurator._Impl.__cleanup_unit_file(unit_file)
+                return
+
+        def is_extension_resource_limits_setup_completed(self, extension_name, cpu_quota=None):
+            unit_file_install_path = systemd.get_unit_file_install_path()
+            old_extension_slice_path = os.path.join(unit_file_install_path, SystemdCgroupsApi.get_extension_slice_name(extension_name, old_slice=True))
+            # clean up the old slice from the disk
+            if os.path.exists(old_extension_slice_path):
+                CGroupConfigurator._Impl.__cleanup_unit_file(old_extension_slice_path)
+
+            extension_slice_path = os.path.join(unit_file_install_path,
+                                                SystemdCgroupsApi.get_extension_slice_name(extension_name))
+            cpu_quota = str(
+                cpu_quota) + "%" if cpu_quota is not None else ""  # setting an empty value resets to the default (infinity)
+            slice_contents = _EXTENSION_SLICE_CONTENTS.format(extension_name=extension_name,
+                                                              cpu_quota=cpu_quota)
+            if os.path.exists(extension_slice_path):
+                with open(extension_slice_path, "r") as file_:
+                    if file_.read() == slice_contents:
+                        return True
+            return False
+
+        def __get_agent_cgroups(self, agent_slice, cpu_controller_root, memory_controller_root):
+            agent_unit_name = systemd.get_agent_unit_name()
+
+            expected_relative_path = os.path.join(agent_slice, agent_unit_name)
+            cpu_cgroup_relative_path, memory_cgroup_relative_path = self._cgroups_api.get_process_cgroup_relative_paths(
+                "self")
+
+            if cpu_cgroup_relative_path is None:
+                _log_cgroup_warning("The agent's process is not within a CPU cgroup")
+            else:
+                if cpu_cgroup_relative_path == expected_relative_path:
+                    _log_cgroup_info('CPUAccounting: {0}', systemd.get_unit_property(agent_unit_name, "CPUAccounting"))
+                    _log_cgroup_info('CPUQuota: {0}', systemd.get_unit_property(agent_unit_name, "CPUQuotaPerSecUSec"))
+                else:
+                    _log_cgroup_warning(
+                        "The Agent is not in the expected CPU cgroup; will not enable monitoring. Cgroup:[{0}] Expected:[{1}]",
+                        cpu_cgroup_relative_path,
+                        expected_relative_path)
+                    cpu_cgroup_relative_path = None  # Set the path to None to prevent monitoring
+
+            if memory_cgroup_relative_path is None:
+                _log_cgroup_warning("The agent's process is not within a memory cgroup")
             else:
-                self._enabled = False
-                self._cgroups_api = None
-                status = "Cgroups are not supported by the platform"
+                if memory_cgroup_relative_path == expected_relative_path:
+                    memory_accounting = systemd.get_unit_property(agent_unit_name, "MemoryAccounting")
+                    _log_cgroup_info('MemoryAccounting: {0}', memory_accounting)
+                else:
+                    _log_cgroup_info(
+                        "The Agent is not in the expected memory cgroup; will not enable monitoring. CGroup:[{0}] Expected:[{1}]",
+                        memory_cgroup_relative_path,
+                        expected_relative_path)
+                    memory_cgroup_relative_path = None  # Set the path to None to prevent monitoring
 
-            logger.info("CGroups Status: {0}".format(status))
+            if cpu_controller_root is not None and cpu_cgroup_relative_path is not None:
+                agent_cpu_cgroup_path = os.path.join(cpu_controller_root, cpu_cgroup_relative_path)
+            else:
+                agent_cpu_cgroup_path = None
+
+            if memory_controller_root is not None and memory_cgroup_relative_path is not None:
+                agent_memory_cgroup_path = os.path.join(memory_controller_root, memory_cgroup_relative_path)
+            else:
+                agent_memory_cgroup_path = None
 
-            add_event(
-                AGENT_NAME,
-                version=CURRENT_VERSION,
-                op=WALAEventOperation.InitializeCGroups,
-                is_success=self._enabled,
-                message=status,
-                log_event=False)
+            return agent_cpu_cgroup_path, agent_memory_cgroup_path
+
+        def supported(self):
+            return self._cgroups_supported
 
         def enabled(self):
-            return self._enabled
+            return self._agent_cgroups_enabled or self._extensions_cgroups_enabled
+
+        def agent_enabled(self):
+            return self._agent_cgroups_enabled
+
+        def extensions_enabled(self):
+            return self._extensions_cgroups_enabled
 
         def enable(self):
-            if not self._cgroups_supported:
-                raise CGroupsException("cgroups are not supported on the current platform")
+            if not self.supported():
+                raise CGroupsException(
+                    "Attempted to enable cgroups, but they are not supported on the current platform")
+            self._agent_cgroups_enabled = True
+            self._extensions_cgroups_enabled = True
 
-            self._enabled = True
+        def disable(self, reason, disable_cgroups):
+            if disable_cgroups == DisableCgroups.ALL:  # disable all
+                # Reset quotas
+                self.__reset_agent_cpu_quota()
+                extension_services = self.get_extension_services_list()
+                for extension in extension_services:
+                    logger.info("Resetting extension : {0} and it's services: {1} CPUQuota".format(extension, extension_services[extension]))
+                    self.__reset_extension_cpu_quota(extension_name=extension)
+                    self.__reset_extension_services_cpu_quota(extension_services[extension])
+                self.__reload_systemd_config()
 
-        def disable(self):
-            self._enabled = False
-            CGroupsTelemetry.reset()
+                CGroupsTelemetry.reset()
+                self._agent_cgroups_enabled = False
+                self._extensions_cgroups_enabled = False
+            elif disable_cgroups == DisableCgroups.AGENT:  # disable agent
+                self._agent_cgroups_enabled = False
+                self.__reset_agent_cpu_quota()
+                CGroupsTelemetry.stop_tracking(CpuCgroup(AGENT_NAME_TELEMETRY, self._agent_cpu_cgroup_path))
 
-        def _invoke_cgroup_operation(self, operation, error_message, on_error=None):
+            message = "[CGW] Disabling resource usage monitoring. Reason: {0}".format(reason)
+            logger.info(message)  # log as INFO for now, in the future it should be logged as WARNING
+            add_event(op=WALAEventOperation.CGroupsDisabled, message=message, is_success=False, log_event=False)
+
+        @staticmethod
+        def __set_cpu_quota(quota):
             """
-            Ensures the given operation is invoked only if cgroups are enabled and traps any errors on the operation.
+            Sets the agent's CPU quota to the given percentage (100% == 1 CPU)
+
+            NOTE: This is done using a dropin file in the default dropin directory; any local overrides on the VM will take precedence
+            over this setting.
             """
-            if not self.enabled():
-                return
+            quota_percentage = "{0}%".format(quota)
+            _log_cgroup_info("Ensuring the agent's CPUQuota is {0}", quota_percentage)
+            if CGroupConfigurator._Impl.__try_set_cpu_quota(quota_percentage):
+                CGroupsTelemetry.set_track_throttled_time(True)
+
+        @staticmethod
+        def __reset_agent_cpu_quota():
+            """
+            Removes any CPUQuota on the agent
 
+            NOTE: This resets the quota on the agent's default dropin file; any local overrides on the VM will take precedence
+            over this setting.
+            """
+            logger.info("Resetting agent's CPUQuota")
+            if CGroupConfigurator._Impl.__try_set_cpu_quota(''):  # setting an empty value resets to the default (infinity)
+                _log_cgroup_info('CPUQuota: {0}',
+                                 systemd.get_unit_property(systemd.get_agent_unit_name(), "CPUQuotaPerSecUSec"))
+
+        @staticmethod
+        def __try_set_cpu_quota(quota):
             try:
-                return operation()
-            except Exception as e:
-                logger.warn("{0} Error: {1}".format(error_message, ustr(e)))
-                if on_error is not None:
-                    try:
-                        on_error(e)
-                    except Exception as ex:
-                        logger.warn("CGroupConfigurator._invoke_cgroup_operation: {0}".format(ustr(e)))
+                drop_in_file = os.path.join(systemd.get_agent_drop_in_path(), _DROP_IN_FILE_CPU_QUOTA)
+                contents = _DROP_IN_FILE_CPU_QUOTA_CONTENTS_FORMAT.format(quota)
+                if os.path.exists(drop_in_file):
+                    with open(drop_in_file, "r") as file_:
+                        if file_.read() == contents:
+                            return True  # no need to update the file; return here to avoid doing a daemon-reload
+                CGroupConfigurator._Impl.__create_unit_file(drop_in_file, contents)
+            except Exception as exception:
+                _log_cgroup_warning('Failed to set CPUQuota: {0}', ustr(exception))
+                return False
+            try:
+                logger.info("Executing systemctl daemon-reload...")
+                shellutil.run_command(["systemctl", "daemon-reload"])
+            except Exception as exception:
+                _log_cgroup_warning("daemon-reload failed (set quota): {0}", ustr(exception))
+                return False
+            return True
+
+        def check_cgroups(self, cgroup_metrics):
+            self._check_cgroups_lock.acquire()
+            try:
+                if not self.enabled():
+                    return
+
+                errors = []
+
+                process_check_success = False
+                try:
+                    self._check_processes_in_agent_cgroup()
+                    process_check_success = True
+                except CGroupsException as exception:
+                    errors.append(exception)
 
-        def create_agent_cgroups(self, track_cgroups):
+                quota_check_success = False
+                try:
+                    if cgroup_metrics:
+                        self._check_agent_throttled_time(cgroup_metrics)
+                    quota_check_success = True
+                except CGroupsException as exception:
+                    errors.append(exception)
+
+                reason = "Check on cgroups failed:\n{0}".format("\n".join([ustr(e) for e in errors]))
+
+                if not process_check_success and conf.get_cgroup_disable_on_process_check_failure():
+                    self.disable(reason, DisableCgroups.ALL)
+
+                if not quota_check_success and conf.get_cgroup_disable_on_quota_check_failure():
+                    self.disable(reason, DisableCgroups.AGENT)
+            finally:
+                self._check_cgroups_lock.release()
+
+        def _check_processes_in_agent_cgroup(self):
             """
-            Creates and returns the cgroups needed to track the VM Agent
+            Verifies that the agent's cgroup includes only the current process, its parent, commands started using shellutil and instances of systemd-run
+            (those processes correspond, respectively, to the extension handler, the daemon, commands started by the extension handler, and the systemd-run
+            commands used to start extensions on their own cgroup).
+            Other processes started by the agent (e.g. extensions) and processes not started by the agent (e.g. services installed by extensions) are reported
+            as unexpected, since they should belong to their own cgroup.
+
+            Raises a CGroupsException if the check fails
             """
-            def __impl():
-                cgroups = self._cgroups_api.create_agent_cgroups()
+            unexpected = []
+            agent_cgroup_proc_names = []
+            try:
+                daemon = os.getppid()
+                extension_handler = os.getpid()
+                agent_commands = set()
+                agent_commands.update(shellutil.get_running_commands())
+                systemd_run_commands = set()
+                systemd_run_commands.update(self._cgroups_api.get_systemd_run_commands())
+                agent_cgroup = CGroupsApi.get_processes_in_cgroup(self._agent_cpu_cgroup_path)
+                # get the running commands again in case new commands started or completed while we were fetching the processes in the cgroup;
+                agent_commands.update(shellutil.get_running_commands())
+                systemd_run_commands.update(self._cgroups_api.get_systemd_run_commands())
 
-                if track_cgroups:
-                    for cgroup in cgroups:
-                        CGroupsTelemetry.track_cgroup(cgroup)
+                for process in agent_cgroup:
+                    agent_cgroup_proc_names.append(self.__format_process(process))
+                    # Note that the agent uses systemd-run to start extensions; systemd-run belongs to the agent cgroup, though the extensions don't.
+                    if process in (daemon, extension_handler) or process in systemd_run_commands:
+                        continue
+                    # check shell systemd_run process if above process check didn't catch it
+                    if self._check_systemd_run_process(process):
+                        continue
+                    # systemd_run_commands contains the shell that started systemd-run, so we also need to check for the parent
+                    if self._get_parent(process) in systemd_run_commands and self._get_command(
+                            process) == 'systemd-run':
+                        continue
+                    # check if the process is a command started by the agent or a descendant of one of those commands
+                    current = process
+                    while current != 0 and current not in agent_commands:
+                        current = self._get_parent(current)
+                    # Verify if Process started by agent based on the marker found in process environment or process is in Zombie state.
+                    # If so, consider it as valid process in agent cgroup.
+                    if current == 0 and not (self.__is_process_descendant_of_the_agent(process) or self.__is_zombie_process(process)):
+                        unexpected.append(self.__format_process(process))
+                        if len(unexpected) >= 5:  # collect just a small sample
+                            break
+            except Exception as exception:
+                _log_cgroup_warning("Error checking the processes in the agent's cgroup: {0}".format(ustr(exception)))
 
-                return cgroups
+            if len(unexpected) > 0:
+                self._report_agent_cgroups_procs(agent_cgroup_proc_names, unexpected)
+                raise CGroupsException("The agent's cgroup includes unexpected processes: {0}".format(unexpected))
 
-            self._invoke_cgroup_operation(__impl, "Failed to create a cgroup for the VM Agent; resource usage for the Agent will not be tracked.")
+        @staticmethod
+        def _get_command(pid):
+            try:
+                with open('/proc/{0}/comm'.format(pid), "r") as file_:
+                    comm = file_.read()
+                    if comm and comm[-1] == '\x00':  # if null-terminated, remove the null
+                        comm = comm[:-1]
+                    return comm.rstrip()
+            except Exception:
+                return "UNKNOWN"
 
-        def cleanup_legacy_cgroups(self):
-            def __impl():
-                self._cgroups_api.cleanup_legacy_cgroups()
+        @staticmethod
+        def __format_process(pid):
+            """
+            Formats the given PID as a string containing the PID and the corresponding command line truncated to 64 chars
+            """
+            try:
+                cmdline = '/proc/{0}/cmdline'.format(pid)
+                if os.path.exists(cmdline):
+                    with open(cmdline, "r") as cmdline_file:
+                        return "[PID: {0}] {1:64.64}".format(pid, cmdline_file.read())
+            except Exception:
+                pass
+            return "[PID: {0}] UNKNOWN".format(pid)
 
-            message = 'Failed to process legacy cgroups. Collection of resource usage data will be disabled.'
+        @staticmethod
+        def __is_process_descendant_of_the_agent(pid):
+            """
+            Returns True if the process is descendant of the agent by looking at the env flag(AZURE_GUEST_AGENT_PARENT_PROCESS_NAME)
+            that we set when the process starts otherwise False.
+            """
+            try:
+                env = '/proc/{0}/environ'.format(pid)
+                if os.path.exists(env):
+                    with open(env, "r") as env_file:
+                        environ = env_file.read()
+                        if environ and environ[-1] == '\x00':
+                            environ = environ[:-1]
+                        return "{0}={1}".format(shellutil.PARENT_PROCESS_NAME, shellutil.AZURE_GUEST_AGENT) in environ
+            except Exception:
+                pass
+            return False
 
-            def disable_cgroups(exception):
-                self.disable()
-                add_event(
-                    AGENT_NAME,
-                    version=CURRENT_VERSION,
-                    op=WALAEventOperation.CGroupsCleanUp,
-                    is_success=False,
-                    log_event=False,
-                    message='{0} {1}'.format(message, ustr(exception)))
+        @staticmethod
+        def __is_zombie_process(pid):
+            """
+            Returns True if process is in Zombie state otherwise False.
 
-            self._invoke_cgroup_operation(__impl, message, on_error=disable_cgroups)
+            Ex: cat /proc/18171/stat
+            18171 (python3) S 18103 18103 18103 0 -1 4194624 57736 64902 0 3
+            """
+            try:
+                stat = '/proc/{0}/stat'.format(pid)
+                if os.path.exists(stat):
+                    with open(stat, "r") as stat_file:
+                        return stat_file.read().split()[2] == 'Z'
+            except Exception:
+                pass
+            return False
 
-        def create_extension_cgroups_root(self):
+        @staticmethod
+        def _check_systemd_run_process(process):
             """
-            Creates the container (directory/cgroup) that includes the cgroups for all extensions (/sys/fs/cgroup/*/walinuxagent.extensions)
+            Returns True if process is shell systemd-run process started by agent otherwise False.
+
+            Ex: sh,7345 -c systemd-run --unit=enable_7c5cab19-eb79-4661-95d9-9e5091bd5ae0 --scope --slice=azure-vmextensions-Microsoft.OSTCExtensions.VMAccessForLinux_1.5.11.slice /var/lib/waagent/Microsoft.OSTCExtensions.VMAccessForLinux-1.5.11/processes.sh
             """
-            def __impl():
-                self._cgroups_api.create_extension_cgroups_root()
+            try:
+                process_name = "UNKNOWN"
+                cmdline = '/proc/{0}/cmdline'.format(process)
+                if os.path.exists(cmdline):
+                    with open(cmdline, "r") as cmdline_file:
+                        process_name = "{0}".format(cmdline_file.read())
+                match = re.search(r'systemd-run.*--unit=.*--scope.*--slice=azure-vmextensions.*', process_name)
+                if match is not None:
+                    return True
+            except Exception:
+                pass
+            return False
+
+        @staticmethod
+        def _report_agent_cgroups_procs(agent_cgroup_proc_names, unexpected):
+            for proc_name in unexpected:
+                if 'UNKNOWN' in proc_name:
+                    msg = "Agent includes following processes when UNKNOWN process found: {0}".format("\n".join([ustr(proc) for proc in agent_cgroup_proc_names]))
+                    add_event(op=WALAEventOperation.CGroupsInfo, message=msg)
 
-            self._invoke_cgroup_operation(__impl, "Failed to create a root cgroup for extensions; resource usage for extensions will not be tracked.")
+        @staticmethod
+        def _check_agent_throttled_time(cgroup_metrics):
+            for metric in cgroup_metrics:
+                if metric.instance == AGENT_NAME_TELEMETRY and metric.counter == MetricsCounter.THROTTLED_TIME:
+                    if metric.value > conf.get_agent_cpu_throttled_time_threshold():
+                        raise CGroupsException("The agent has been throttled for {0} seconds".format(metric.value))
 
-        def create_extension_cgroups(self, name):
+        def check_agent_memory_usage(self):
+            if self.enabled() and self._agent_memory_cgroup:
+                metrics = self._agent_memory_cgroup.get_tracked_metrics()
+                current_usage = 0
+                for metric in metrics:
+                    if metric.counter == MetricsCounter.TOTAL_MEM_USAGE:
+                        current_usage += metric.value
+                    elif metric.counter == MetricsCounter.SWAP_MEM_USAGE:
+                        current_usage += metric.value
+
+                if current_usage > conf.get_agent_memory_quota():
+                    raise AgentMemoryExceededException("The agent memory limit {0} bytes exceeded. The current reported usage is {1} bytes.".format(conf.get_agent_memory_quota(), current_usage))
+
+        @staticmethod
+        def _get_parent(pid):
             """
-            Creates and returns the cgroups for the given extension
+            Returns the parent of the given process. If the parent cannot be determined returns 0 (which is the PID for the scheduler)
             """
-            def __impl():
-                return self._cgroups_api.create_extension_cgroups(name)
+            try:
+                stat = '/proc/{0}/stat'.format(pid)
+                if os.path.exists(stat):
+                    with open(stat, "r") as stat_file:
+                        return int(stat_file.read().split()[3])
+            except Exception:
+                pass
+            return 0
 
-            return self._invoke_cgroup_operation(__impl, "Failed to create a cgroup for extension '{0}'; resource usage will not be tracked.".format(name))
+        def start_tracking_unit_cgroups(self, unit_name):
+            """
+            TODO: Start tracking Memory Cgroups
+            """
+            try:
+                cpu_cgroup_path, memory_cgroup_path = self._cgroups_api.get_unit_cgroup_paths(unit_name)
+
+                if cpu_cgroup_path is None:
+                    logger.info("The CPU controller is not mounted; will not track resource usage")
+                else:
+                    CGroupsTelemetry.track_cgroup(CpuCgroup(unit_name, cpu_cgroup_path))
+
+                if memory_cgroup_path is None:
+                    logger.info("The Memory controller is not mounted; will not track resource usage")
+                else:
+                    CGroupsTelemetry.track_cgroup(MemoryCgroup(unit_name, memory_cgroup_path))
+
+            except Exception as exception:
+                logger.info("Failed to start tracking resource usage for the extension: {0}", ustr(exception))
 
-        def remove_extension_cgroups(self, name):
+        def stop_tracking_unit_cgroups(self, unit_name):
             """
-            Deletes the cgroup for the given extension
+            TODO: remove Memory cgroups from tracked list.
             """
-            def __impl():
-                cgroups = self._cgroups_api.remove_extension_cgroups(name)
-                return cgroups
+            try:
+                cpu_cgroup_path, memory_cgroup_path = self._cgroups_api.get_unit_cgroup_paths(unit_name)
+
+                if cpu_cgroup_path is not None:
+                    CGroupsTelemetry.stop_tracking(CpuCgroup(unit_name, cpu_cgroup_path))
+
+                if memory_cgroup_path is not None:
+                    CGroupsTelemetry.stop_tracking(MemoryCgroup(unit_name, memory_cgroup_path))
 
-            self._invoke_cgroup_operation(__impl, "Failed to delete cgroups for extension '{0}'.".format(name))
+            except Exception as exception:
+                logger.info("Failed to stop tracking resource usage for the extension service: {0}", ustr(exception))
 
-        def start_extension_command(self, extension_name, command, timeout, shell, cwd, env, stdout, stderr,
+        def stop_tracking_extension_cgroups(self, extension_name):
+            """
+            TODO: remove extension Memory cgroups from tracked list
+            """
+            try:
+                extension_slice_name = SystemdCgroupsApi.get_extension_slice_name(extension_name)
+                cgroup_relative_path = os.path.join(_AZURE_VMEXTENSIONS_SLICE,
+                                                    extension_slice_name)
+
+                cpu_cgroup_mountpoint, memory_cgroup_mountpoint = self._cgroups_api.get_cgroup_mount_points()
+                cpu_cgroup_path = os.path.join(cpu_cgroup_mountpoint, cgroup_relative_path)
+                memory_cgroup_path = os.path.join(memory_cgroup_mountpoint, cgroup_relative_path)
+
+                if cpu_cgroup_path is not None:
+                    CGroupsTelemetry.stop_tracking(CpuCgroup(extension_name, cpu_cgroup_path))
+
+                if memory_cgroup_path is not None:
+                    CGroupsTelemetry.stop_tracking(MemoryCgroup(extension_name, memory_cgroup_path))
+
+            except Exception as exception:
+                logger.info("Failed to stop tracking resource usage for the extension service: {0}", ustr(exception))
+
+        def start_extension_command(self, extension_name, command, cmd_name, timeout, shell, cwd, env, stdout, stderr,
                                     error_code=ExtensionErrorCodes.PluginUnknownFailure):
             """
             Starts a command (install/enable/etc) for an extension and adds the command's PID to the extension's cgroup
             :param extension_name: The extension executing the command
             :param command: The command to invoke
+            :param cmd_name: The type of the command(enable, install, etc.)
             :param timeout: Number of seconds to wait for command completion
             :param cwd: The working directory for the command
             :param env:  The environment to pass to the command's process
@@ -172,39 +886,207 @@ class CGroupConfigurator(object):
             :param stderr: File object to redirect stderr to
             :param error_code: Extension error code to raise in case of error
             """
-            if not self.enabled():
-                process = subprocess.Popen(command,
-                                           shell=shell,
-                                           cwd=cwd,
-                                           env=env,
-                                           stdout=stdout,
-                                           stderr=stderr,
-                                           preexec_fn=os.setsid)
-
-                process_output = handle_process_completion(process=process,
-                                                           command=command,
-                                                           timeout=timeout,
-                                                           stdout=stdout,
-                                                           stderr=stderr,
-                                                           error_code=error_code)
-            else:
-                extension_cgroups, process_output = self._cgroups_api.start_extension_command(extension_name,
-                                                                                              command,
-                                                                                              timeout,
-                                                                                              shell=shell,
-                                                                                              cwd=cwd,
-                                                                                              env=env,
-                                                                                              stdout=stdout,
-                                                                                              stderr=stderr,
-                                                                                              error_code=error_code)
-
-            return process_output
-
-    # unique instance for the singleton (TODO: find a better pattern for a singleton)
+            if self.enabled():
+                try:
+                    return self._cgroups_api.start_extension_command(extension_name, command, cmd_name, timeout,
+                                                                     shell=shell, cwd=cwd, env=env, stdout=stdout,
+                                                                     stderr=stderr, error_code=error_code)
+                except SystemdRunError as exception:
+                    reason = 'Failed to start {0} using systemd-run, will try invoking the extension directly. Error: {1}'.format(
+                        extension_name, ustr(exception))
+                    self.disable(reason, DisableCgroups.ALL)
+                    # fall-through and re-invoke the extension
+
+            # subprocess-popen-preexec-fn<W1509> Disabled: code is not multi-threaded
+            process = subprocess.Popen(command, shell=shell, cwd=cwd, env=env, stdout=stdout, stderr=stderr, preexec_fn=os.setsid)  # pylint: disable=W1509
+            return handle_process_completion(process=process, command=command, timeout=timeout, stdout=stdout, stderr=stderr, error_code=error_code)
+
+        def __reset_extension_cpu_quota(self, extension_name):
+            """
+            Removes any CPUQuota on the extension
+
+            NOTE: This resets the quota on the extension's slice; any local overrides on the VM will take precedence
+            over this setting.
+            """
+            if self.enabled():
+                self.setup_extension_slice(extension_name, cpu_quota=None)
+
+        def setup_extension_slice(self, extension_name, cpu_quota):
+            """
+            Each extension runs under its own slice (Ex "Microsoft.CPlat.Extension.slice"). All the slices for
+            extensions are grouped under "azure-vmextensions.slice.
+
+            This method ensures that the extension slice is created. Setup should create
+            under /lib/systemd/system if it is not exist.
+            TODO: set memory quotas
+            """
+            if self.enabled():
+                unit_file_install_path = systemd.get_unit_file_install_path()
+                extension_slice_path = os.path.join(unit_file_install_path,
+                                                    SystemdCgroupsApi.get_extension_slice_name(extension_name))
+                try:
+                    cpu_quota = str(cpu_quota) + "%" if cpu_quota is not None else ""  # setting an empty value resets to the default (infinity)
+                    if cpu_quota == "":
+                        _log_cgroup_info("CPUQuota not set for {0}", extension_name)
+                    else:
+                        _log_cgroup_info("Ensuring the {0}'s CPUQuota is {1}", extension_name, cpu_quota)
+                    slice_contents = _EXTENSION_SLICE_CONTENTS.format(extension_name=extension_name,
+                                                                      cpu_quota=cpu_quota)
+                    CGroupConfigurator._Impl.__create_unit_file(extension_slice_path, slice_contents)
+                except Exception as exception:
+                    _log_cgroup_warning("Failed to set the extension {0} slice and quotas: {1}", extension_name,
+                                        ustr(exception))
+                    CGroupConfigurator._Impl.__cleanup_unit_file(extension_slice_path)
+
+        def remove_extension_slice(self, extension_name):
+            """
+            This method ensures that the extension slice gets removed from /lib/systemd/system if it exist
+            Lastly stop the unit. This would ensure the cleanup the /sys/fs/cgroup controller paths
+            """
+            if self.enabled():
+                unit_file_install_path = systemd.get_unit_file_install_path()
+                extension_slice_name = SystemdCgroupsApi.get_extension_slice_name(extension_name)
+                extension_slice_path = os.path.join(unit_file_install_path, extension_slice_name)
+                if os.path.exists(extension_slice_path):
+                    self.stop_tracking_extension_cgroups(extension_name)
+                    CGroupConfigurator._Impl.__cleanup_unit_file(extension_slice_path)
+
+        def set_extension_services_cpu_memory_quota(self, services_list):
+            """
+            Each extension service will have name, systemd path and it's quotas.
+            This method ensures that drop-in files are created under service.d folder if quotas given.
+            ex: /lib/systemd/system/extension.service.d/11-CPUAccounting.conf
+            TODO: set memory quotas
+            """
+            if self.enabled() and services_list is not None:
+                for service in services_list:
+                    service_name = service.get('name', None)
+                    unit_file_path = systemd.get_unit_file_install_path()
+                    if service_name is not None and unit_file_path is not None:
+                        files_to_create = []
+                        drop_in_path = os.path.join(unit_file_path, "{0}.d".format(service_name))
+                        drop_in_file_cpu_accounting = os.path.join(drop_in_path,
+                                                                   _DROP_IN_FILE_CPU_ACCOUNTING)
+                        files_to_create.append((drop_in_file_cpu_accounting, _DROP_IN_FILE_CPU_ACCOUNTING_CONTENTS))
+                        drop_in_file_memory_accounting = os.path.join(drop_in_path,
+                                                                      _DROP_IN_FILE_MEMORY_ACCOUNTING)
+                        files_to_create.append(
+                            (drop_in_file_memory_accounting, _DROP_IN_FILE_MEMORY_ACCOUNTING_CONTENTS))
+
+                        cpu_quota = service.get('cpuQuotaPercentage', None)
+                        if cpu_quota is not None:
+                            cpu_quota = str(cpu_quota) + "%"
+                            _log_cgroup_info("Ensuring the {0}'s CPUQuota is {1}", service_name, cpu_quota)
+                            drop_in_file_cpu_quota = os.path.join(drop_in_path, _DROP_IN_FILE_CPU_QUOTA)
+                            cpu_quota_contents = _DROP_IN_FILE_CPU_QUOTA_CONTENTS_FORMAT.format(cpu_quota)
+                            files_to_create.append((drop_in_file_cpu_quota, cpu_quota_contents))
+
+                        self.__create_all_files(files_to_create)
+                        self.__reload_systemd_config()
+
+        def __reset_extension_services_cpu_quota(self, services_list):
+            """
+            Removes any CPUQuota on the extension service
+
+            NOTE: This resets the quota on the extension service's default dropin file; any local overrides on the VM will take precedence
+            over this setting.
+            """
+            if self.enabled() and services_list is not None:
+                service_name = None
+                try:
+                    for service in services_list:
+                        service_name = service.get('name', None)
+                        unit_file_path = systemd.get_unit_file_install_path()
+                        if service_name is not None and unit_file_path is not None:
+                            files_to_create = []
+                            drop_in_path = os.path.join(unit_file_path, "{0}.d".format(service_name))
+                            cpu_quota = ""  # setting an empty value resets to the default (infinity)
+                            drop_in_file_cpu_quota = os.path.join(drop_in_path, _DROP_IN_FILE_CPU_QUOTA)
+                            cpu_quota_contents = _DROP_IN_FILE_CPU_QUOTA_CONTENTS_FORMAT.format(cpu_quota)
+                            if os.path.exists(drop_in_file_cpu_quota):
+                                with open(drop_in_file_cpu_quota, "r") as file_:
+                                    if file_.read() == cpu_quota_contents:
+                                        return
+                                files_to_create.append((drop_in_file_cpu_quota, cpu_quota_contents))
+                            self.__create_all_files(files_to_create)
+                except Exception as exception:
+                    _log_cgroup_warning('Failed to reset CPUQuota for {0} : {1}', service_name, ustr(exception))
+
+        def remove_extension_services_drop_in_files(self, services_list):
+            """
+            Remove the dropin files from service .d folder for the given service
+            """
+            if services_list is not None:
+                for service in services_list:
+                    service_name = service.get('name', None)
+                    unit_file_path = systemd.get_unit_file_install_path()
+                    if service_name is not None and unit_file_path is not None:
+                        files_to_cleanup = []
+                        drop_in_path = os.path.join(unit_file_path, "{0}.d".format(service_name))
+                        drop_in_file_cpu_accounting = os.path.join(drop_in_path,
+                                                                   _DROP_IN_FILE_CPU_ACCOUNTING)
+                        files_to_cleanup.append(drop_in_file_cpu_accounting)
+                        drop_in_file_memory_accounting = os.path.join(drop_in_path,
+                                                                      _DROP_IN_FILE_MEMORY_ACCOUNTING)
+                        files_to_cleanup.append(drop_in_file_memory_accounting)
+                        cpu_quota = service.get('cpuQuotaPercentage', None)
+                        if cpu_quota is not None:
+                            drop_in_file_cpu_quota = os.path.join(drop_in_path, _DROP_IN_FILE_CPU_QUOTA)
+                            files_to_cleanup.append(drop_in_file_cpu_quota)
+
+                        CGroupConfigurator._Impl.__cleanup_all_files(files_to_cleanup)
+                        _log_cgroup_info("Drop in files removed for {0}".format(service_name))
+
+        def stop_tracking_extension_services_cgroups(self, services_list):
+            """
+            Remove the cgroup entry from the tracked groups to stop tracking.
+            """
+            if self.enabled() and services_list is not None:
+                for service in services_list:
+                    service_name = service.get('name', None)
+                    if service_name is not None:
+                        self.stop_tracking_unit_cgroups(service_name)
+
+        def start_tracking_extension_services_cgroups(self, services_list):
+            """
+            Add the cgroup entry to start tracking the services cgroups.
+            """
+            if self.enabled() and services_list is not None:
+                for service in services_list:
+                    service_name = service.get('name', None)
+                    if service_name is not None:
+                        self.start_tracking_unit_cgroups(service_name)
+
+        @staticmethod
+        def get_extension_services_list():
+            """
+            ResourceLimits for extensions are coming from <extName>/HandlerManifest.json file.
+            Use this pattern to determine all the installed extension HandlerManifest files and
+            read the extension services if ResourceLimits are present.
+            """
+            extensions_services = {}
+            for manifest_path in glob.iglob(os.path.join(conf.get_lib_dir(), "*/HandlerManifest.json")):
+                match = re.search("(?P<extname>[\\w+\\.-]+).HandlerManifest\\.json", manifest_path)
+                if match is not None:
+                    extensions_name = match.group('extname')
+                    if not extensions_name.startswith('WALinuxAgent'):
+                        try:
+                            data = json.loads(fileutil.read_file(manifest_path))
+                            resource_limits = data[0].get('resourceLimits', None)
+                            services = resource_limits.get('services') if resource_limits else None
+                            extensions_services[extensions_name] = services
+                        except (IOError, OSError) as e:
+                            _log_cgroup_warning(
+                                'Failed to load manifest file ({0}): {1}'.format(manifest_path, e.strerror))
+                        except ValueError:
+                            _log_cgroup_warning('Malformed manifest file ({0}).'.format(manifest_path))
+            return extensions_services
+
+    # unique instance for the singleton
     _instance = None
 
     @staticmethod
     def get_instance():
         if CGroupConfigurator._instance is None:
-            CGroupConfigurator._instance = CGroupConfigurator.__impl()
+            CGroupConfigurator._instance = CGroupConfigurator._Impl()
         return CGroupConfigurator._instance
diff --git a/azurelinuxagent/common/cgroupstelemetry.py b/azurelinuxagent/common/cgroupstelemetry.py
index 4bbcba1..7b6bba0 100644
--- a/azurelinuxagent/common/cgroupstelemetry.py
+++ b/azurelinuxagent/common/cgroupstelemetry.py
@@ -15,101 +15,26 @@
 # Requires Python 2.6+ and Openssl 1.0+
 import errno
 import threading
-from collections import namedtuple
-from datetime import datetime as dt
 
 from azurelinuxagent.common import logger
-from azurelinuxagent.common.cgroup import CpuCgroup, CGroupContollers
-from azurelinuxagent.common.exception import CGroupsException
+from azurelinuxagent.common.cgroup import CpuCgroup
 from azurelinuxagent.common.future import ustr
-from azurelinuxagent.common.logger import EVERY_SIX_HOURS
-from azurelinuxagent.common.resourceusage import MemoryResourceUsage, ProcessInfo
-
-MetricValue = namedtuple('Metric', ['category', 'counter', 'instance', 'value'])
-StatmMetricValue = namedtuple('StatmMetricValue', ['pid_name_cmdline', 'resource_metric'])
-
-DELIM = " | "
-DEFAULT_PROCESS_NAME = "NO_PROCESS_FOUND"
-DEFAULT_PROCESS_COMMANDLINE = "NO_CMDLINE_FOUND"
-
-
-class MetricsCategory(object):
-    MEMORY_CATEGORY = "Memory"
-    PROCESS_CATEGORY = "Process"
-
-
-class MetricsCounter(object):
-    PROCESSOR_PERCENT_TIME = "% Processor Time"
-    TOTAL_MEM_USAGE = "Total Memory Usage"
-    MAX_MEM_USAGE = "Max Memory Usage"
-    MEM_USED_BY_PROCESS = "Memory Used by Process"
 
 
 class CGroupsTelemetry(object):
     """
     """
-    _tracked = []
-    _cgroup_metrics = {}
+    _tracked = {}
+    _track_throttled_time = False
     _rlock = threading.RLock()
 
     @staticmethod
-    def get_process_info_summary(process_id):
-        process_cmdline = DEFAULT_PROCESS_COMMANDLINE
-        process_name = DEFAULT_PROCESS_NAME
-
-        # The ProcessName and ProcessCommandLine can generate Exception if the file /proc/<pid>/{comm,cmdline} cease to
-        # exist; eg: the process can die, or finish. Which is why we need Default Names, in case we fail to fetch the
-        # details from those files.
-        try:
-            process_cmdline = ProcessInfo.get_proc_cmdline(process_id) if not None else DEFAULT_PROCESS_COMMANDLINE
-        except Exception as e:
-            logger.periodic_info(EVERY_SIX_HOURS, "[PERIODIC] {0}", ustr(e))
-
-        try:
-            process_name = ProcessInfo.get_proc_name(process_id) if not None else DEFAULT_PROCESS_NAME
-        except Exception as e:
-            logger.periodic_info(EVERY_SIX_HOURS, "[PERIODIC] {0}", ustr(e))
-
-        return process_id + DELIM + process_name + DELIM + process_cmdline
+    def set_track_throttled_time(value):
+        CGroupsTelemetry._track_throttled_time = value
 
     @staticmethod
-    def _get_metrics_list(metric):
-        return [metric.average(), metric.min(), metric.max(), metric.median(), metric.count(),
-                metric.first_poll_time(), metric.last_poll_time()]
-
-    @staticmethod
-    def _process_cgroup_metric(cgroup_metrics):
-        memory_usage = cgroup_metrics.get_memory_metrics()
-        max_memory_usage = cgroup_metrics.get_max_memory_metrics()
-        cpu_usage = cgroup_metrics.get_cpu_metrics()
-        memory_usage_per_process = cgroup_metrics.get_proc_statm_memory_metrics()
-
-        processed_extension = {}
-
-        if cpu_usage.count() > 0:
-            processed_extension["cpu"] = {"cur_cpu": CGroupsTelemetry._get_metrics_list(cpu_usage)}
-
-        if memory_usage.count() > 0:
-            if "memory" in processed_extension:
-                processed_extension["memory"]["cur_mem"] = CGroupsTelemetry._get_metrics_list(memory_usage)
-            else:
-                processed_extension["memory"] = {"cur_mem": CGroupsTelemetry._get_metrics_list(memory_usage)}
-
-        if max_memory_usage.count() > 0:
-            if "memory" in processed_extension:
-                processed_extension["memory"]["max_mem"] = CGroupsTelemetry._get_metrics_list(max_memory_usage)
-            else:
-                processed_extension["memory"] = {"max_mem": CGroupsTelemetry._get_metrics_list(max_memory_usage)}
-
-        for pid_process_memory in memory_usage_per_process:
-            if "proc_statm_memory" in processed_extension:
-                processed_extension["proc_statm_memory"][pid_process_memory.pid_name_cmdline] = \
-                    CGroupsTelemetry._get_metrics_list(pid_process_memory.resource_metric)
-            else:
-                processed_extension["proc_statm_memory"] = {pid_process_memory.pid_name_cmdline:
-                    CGroupsTelemetry._get_metrics_list(pid_process_memory.resource_metric)}
-
-        return processed_extension
+    def get_track_throttled_time():
+        return CGroupsTelemetry._track_throttled_time
 
     @staticmethod
     def track_cgroup(cgroup):
@@ -122,221 +47,56 @@ class CGroupsTelemetry(object):
 
         with CGroupsTelemetry._rlock:
             if not CGroupsTelemetry.is_tracked(cgroup.path):
-                CGroupsTelemetry._tracked.append(cgroup)
-                logger.info("Started tracking new cgroup: {0}, path: {1}".format(cgroup.name, cgroup.path))
+                CGroupsTelemetry._tracked[cgroup.path] = cgroup
+                logger.info("Started tracking cgroup {0}", cgroup)
 
     @staticmethod
     def is_tracked(path):
         """
         Returns true if the given item is in the list of tracked items
-        O(n) operation. But limited to few cgroup objects we have.
+        O(1) operation.
         """
         with CGroupsTelemetry._rlock:
-            for cgroup in CGroupsTelemetry._tracked:
-                if path == cgroup.path:
-                    return True
+            if path in CGroupsTelemetry._tracked:
+                return True
 
         return False
 
     @staticmethod
     def stop_tracking(cgroup):
         """
-        Stop tracking the cgroups for the given name
+        Stop tracking the cgroups for the given path
         """
         with CGroupsTelemetry._rlock:
-            CGroupsTelemetry._tracked.remove(cgroup)
-            logger.info("Stopped tracking cgroup: {0}, path: {1}".format(cgroup.name, cgroup.path))
-
-    @staticmethod
-    def report_all_tracked():
-        """
-        The report_all_tracked's purpose is to collect the data from the tracked cgroups and process the metric into a
-        data structure by _process_cgroup_metric. The perf metric is added into the data structure and returned to the
-        caller.
-
-        The report_all_tracked would be removed soon - in favor of sending report_metric directly, when polling the data
-        from tracked groups.
-
-        :return collected_metrics: dictionary of cgroups metrics.
-        """
-        collected_metrics = {}
-
-        for name, cgroup_metrics in CGroupsTelemetry._cgroup_metrics.items():
-            perf_metric = CGroupsTelemetry._process_cgroup_metric(cgroup_metrics)
-
-            if perf_metric:
-                collected_metrics[name] = perf_metric
-
-            cgroup_metrics.clear()
-
-        # Doing cleanup after the metrics have already been collected.
-        for key in [key for key in CGroupsTelemetry._cgroup_metrics if
-                    CGroupsTelemetry._cgroup_metrics[key].marked_for_delete]:
-            del CGroupsTelemetry._cgroup_metrics[key]
-
-        return collected_metrics
+            if cgroup.path in CGroupsTelemetry._tracked:
+                CGroupsTelemetry._tracked.pop(cgroup.path)
+                logger.info("Stopped tracking cgroup {0}", cgroup)
 
     @staticmethod
     def poll_all_tracked():
         metrics = []
-
+        inactive_cgroups = []
         with CGroupsTelemetry._rlock:
-            for cgroup in CGroupsTelemetry._tracked[:]:
-                if cgroup.name not in CGroupsTelemetry._cgroup_metrics:
-                    CGroupsTelemetry._cgroup_metrics[cgroup.name] = CgroupMetrics()
+            for cgroup in CGroupsTelemetry._tracked.values():
                 try:
-                    if cgroup.controller == CGroupContollers.CPU:
-                        current_cpu_usage = cgroup.get_cpu_usage()
-                        CGroupsTelemetry._cgroup_metrics[cgroup.name].add_cpu_usage(current_cpu_usage)
-                        metrics.append(MetricValue(MetricsCategory.PROCESS_CATEGORY, MetricsCounter.
-                                                   PROCESSOR_PERCENT_TIME, cgroup.name, current_cpu_usage))
-                    elif cgroup.controller == CGroupContollers.MEMORY:
-                        current_memory_usage = cgroup.get_memory_usage()
-                        CGroupsTelemetry._cgroup_metrics[cgroup.name].add_memory_usage(current_memory_usage)
-                        metrics.append(MetricValue(MetricsCategory.MEMORY_CATEGORY, MetricsCounter.
-                                                   TOTAL_MEM_USAGE, cgroup.name, current_memory_usage))
-
-                        max_memory_usage = cgroup.get_max_memory_usage()
-                        CGroupsTelemetry._cgroup_metrics[cgroup.name].add_max_memory_usage(max_memory_usage)
-                        metrics.append(MetricValue(MetricsCategory.MEMORY_CATEGORY, MetricsCounter.MAX_MEM_USAGE,
-                                                   cgroup.name, max_memory_usage))
-
-                        pids = cgroup.get_tracked_processes()
-                        for pid in pids:
-                            try:
-                                mem_usage_from_procstatm = MemoryResourceUsage.get_memory_usage_from_proc_statm(pid)
-                                metrics.append(MetricValue(MetricsCategory.MEMORY_CATEGORY, MetricsCounter.
-                                                           MEM_USED_BY_PROCESS, CGroupsTelemetry.get_process_info_summary(pid),
-                                                           mem_usage_from_procstatm))
-                                CGroupsTelemetry._cgroup_metrics[cgroup.name].add_proc_statm_memory(
-                                    CGroupsTelemetry.get_process_info_summary(pid), mem_usage_from_procstatm)
-                            except Exception as e:
-                                if not isinstance(e, (IOError, OSError)) or e.errno != errno.ENOENT:
-                                    logger.periodic_warn(logger.EVERY_HOUR, "[PERIODIC] Could not collect proc_statm "
-                                                                            "for pid {0}. Error : {1}", pid, ustr(e))
-                    else:
-                        raise CGroupsException('CGroup controller {0} is not supported for cgroup {1}'.format(
-                            cgroup.controller, cgroup.name))
+                    metrics.extend(cgroup.get_tracked_metrics(track_throttled_time=CGroupsTelemetry._track_throttled_time))
                 except Exception as e:
                     # There can be scenarios when the CGroup has been deleted by the time we are fetching the values
                     # from it. This would raise IOError with file entry not found (ERRNO: 2). We do not want to log
                     # every occurrences of such case as it would be very verbose. We do want to log all the other
                     # exceptions which could occur, which is why we do a periodic log for all the other errors.
-                    if not isinstance(e, (IOError, OSError)) or e.errno != errno.ENOENT:
+                    if not isinstance(e, (IOError, OSError)) or e.errno != errno.ENOENT:  # pylint: disable=E1101
                         logger.periodic_warn(logger.EVERY_HOUR, '[PERIODIC] Could not collect metrics for cgroup '
                                                                 '{0}. Error : {1}'.format(cgroup.name, ustr(e)))
                 if not cgroup.is_active():
-                    CGroupsTelemetry.stop_tracking(cgroup)
-                    CGroupsTelemetry._cgroup_metrics[cgroup.name].marked_for_delete = True
+                    inactive_cgroups.append(cgroup)
+            for inactive_cgroup in inactive_cgroups:
+                CGroupsTelemetry.stop_tracking(inactive_cgroup)
 
         return metrics
 
     @staticmethod
-    def prune_all_tracked():
-        with CGroupsTelemetry._rlock:
-            for cgroup in CGroupsTelemetry._tracked[:]:
-                if not cgroup.is_active():
-                    CGroupsTelemetry.stop_tracking(cgroup)
-
-    @staticmethod
     def reset():
         with CGroupsTelemetry._rlock:
-            CGroupsTelemetry._tracked *= 0  # emptying the list
-            CGroupsTelemetry._cgroup_metrics = {}
-
-
-class CgroupMetrics(object):
-    def __init__(self):
-        self._memory_usage = Metric()
-        self._max_memory_usage = Metric()
-        self._cpu_usage = Metric()
-        self._proc_statm_mem = {}
-
-        self.marked_for_delete = False
-
-    def add_memory_usage(self, usage):
-        if not self.marked_for_delete:
-            self._memory_usage.append(usage)
-
-    def add_max_memory_usage(self, usage):
-        if not self.marked_for_delete:
-            self._max_memory_usage.append(usage)
-
-    def add_cpu_usage(self, usage):
-        if not self.marked_for_delete:
-            self._cpu_usage.append(usage)
-
-    def add_proc_statm_memory(self, pid, usage):
-        if not self.marked_for_delete:
-            if pid not in self._proc_statm_mem:
-                self._proc_statm_mem[pid] = Metric()
-            self._proc_statm_mem[pid].append(usage)
-
-    def get_memory_metrics(self):
-        return self._memory_usage
-
-    def get_max_memory_metrics(self):
-        return self._max_memory_usage
-
-    def get_cpu_metrics(self):
-        return self._cpu_usage
-
-    def get_proc_statm_memory_metrics(self):
-        """
-        :return: StatmMetricValue tuples of pid and metric
-        """
-        return [StatmMetricValue(pid_name_cmdline, metric) for pid_name_cmdline, metric in self._proc_statm_mem.items()]
-
-    def clear(self):
-        self._memory_usage.clear()
-        self._max_memory_usage.clear()
-        self._cpu_usage.clear()
-        self._proc_statm_mem.clear()
-
-
-class Metric(object):
-    def __init__(self):
-        self._data = []
-        self._first_poll_time = None
-        self._last_poll_time = None
-
-    def append(self, data):
-        if not self._first_poll_time:
-            #  We only want to do it first time.
-            self._first_poll_time = dt.utcnow()
-
-        self._data.append(data)
-        self._last_poll_time = dt.utcnow()
-
-    def clear(self):
-        self._first_poll_time = None
-        self._last_poll_time = None
-        self._data *= 0
-
-    def average(self):
-        return float(sum(self._data)) / float(len(self._data)) if self._data else None
-
-    def max(self):
-        return max(self._data) if self._data else None
-
-    def min(self):
-        return min(self._data) if self._data else None
-
-    def median(self):
-        data = sorted(self._data)
-        l_len = len(data)
-        if l_len < 1:
-            return None
-        if l_len % 2 == 0:
-            return (data[int((l_len - 1) / 2)] + data[int((l_len + 1) / 2)]) / 2.0
-        else:
-            return data[int((l_len - 1) / 2)]
-
-    def count(self):
-        return len(self._data)
-
-    def first_poll_time(self):
-        return str(self._first_poll_time)
-
-    def last_poll_time(self):
-        return str(self._last_poll_time)
+            CGroupsTelemetry._tracked.clear()  # emptying the dictionary
+            CGroupsTelemetry._track_throttled_time = False
diff --git a/azurelinuxagent/common/conf.py b/azurelinuxagent/common/conf.py
index bfc61f0..46765ea 100644
--- a/azurelinuxagent/common/conf.py
+++ b/azurelinuxagent/common/conf.py
@@ -19,11 +19,11 @@
 
 """
 Module conf loads and parses configuration file
-"""
+"""  # pylint: disable=W0105
 import os
 import os.path
 
-import azurelinuxagent.common.utils.fileutil as fileutil
+from azurelinuxagent.common.utils.fileutil import read_file #pylint: disable=R0401
 from azurelinuxagent.common.exception import AgentConfigError
 
 DISABLE_AGENT_FILE = 'disable_agent'
@@ -49,25 +49,43 @@ class ConfigurationProvider(object):
                 value = parts[1].split('#')[0].strip("\" ").strip()
                 self.values[key] = value if value != "None" else None
 
-    def get(self, key, default_val):
+    @staticmethod
+    def _get_default(default):
+        if hasattr(default, '__call__'):
+            return default()
+        return default
+
+    def get(self, key, default_value):
+        """
+        Retrieves a string parameter by key and returns its value. If not found returns the default value,
+        or if the default value is a callable returns the result of invoking the callable.
+        """
         val = self.values.get(key)
-        return val if val is not None else default_val
+        return val if val is not None else self._get_default(default_value)
 
-    def get_switch(self, key, default_val):
+    def get_switch(self, key, default_value):
+        """
+        Retrieves a switch parameter by key and returns its value as a boolean. If not found returns the default value,
+        or if the default value is a callable returns the result of invoking the callable.
+        """
         val = self.values.get(key)
         if val is not None and val.lower() == 'y':
             return True
         elif val is not None and val.lower() == 'n':
             return False
-        return default_val
+        return self._get_default(default_value)
 
-    def get_int(self, key, default_val):
+    def get_int(self, key, default_value):
+        """
+        Retrieves an int parameter by key and returns its value. If not found returns the default value,
+        or if the default value is a callable returns the result of invoking the callable.
+        """
         try:
             return int(self.values.get(key))
         except TypeError:
-            return default_val
+            return self._get_default(default_value)
         except ValueError:
-            return default_val
+            return self._get_default(default_value)
 
 
 __conf__ = ConfigurationProvider()
@@ -81,7 +99,7 @@ def load_conf_from_file(conf_file_path, conf=__conf__):
         raise AgentConfigError(("Missing configuration in {0}"
                                 "").format(conf_file_path))
     try:
-        content = fileutil.read_file(conf_file_path)
+        content = read_file(conf_file_path)
         conf.load(content)
     except IOError as err:
         raise AgentConfigError(("Failed to load conf file:{0}, {1}"
@@ -97,6 +115,7 @@ __SWITCH_OPTIONS__ = {
     "OS.CheckRdmaDriver": False,
     "Logs.Verbose": False,
     "Logs.Console": True,
+    "Logs.Collect": True,
     "Extensions.Enabled": True,
     "Provisioning.AllowResetSysUser": False,
     "Provisioning.RegenerateSshHostKeyPair": False,
@@ -110,7 +129,16 @@ __SWITCH_OPTIONS__ = {
     "ResourceDisk.EnableSwapEncryption": False,
     "AutoUpdate.Enabled": True,
     "EnableOverProvisioning": True,
-    "CGroups.EnforceLimits": False,
+    #
+    # "Debug" options are experimental and may be removed in later
+    # versions of the Agent.
+    #
+    "Debug.CgroupLogMetrics": False,
+    "Debug.CgroupDisableOnProcessCheckFailure": True,
+    "Debug.CgroupDisableOnQuotaCheckFailure": True,
+    "Debug.EnableAgentMemoryUsageCheck": False,
+    "Debug.EnableFastTrack": True,
+    "Debug.EnableGAVersioning": False
 }
 
 
@@ -133,16 +161,37 @@ __STRING_OPTIONS__ = {
     "ResourceDisk.MountOptions": None,
     "ResourceDisk.Filesystem": "ext3",
     "AutoUpdate.GAFamily": "Prod",
-    "CGroups.Excluded": "customscript,runcommand",
+    "Debug.CgroupMonitorExpiryTime": "2022-03-31",
+    "Debug.CgroupMonitorExtensionName": "Microsoft.Azure.Monitor.AzureMonitorLinuxAgent",
 }
 
 
 __INTEGER_OPTIONS__ = {
+    "Extensions.GoalStatePeriod": 6,
+    "Extensions.InitialGoalStatePeriod": 6,
+    "OS.EnableFirewallPeriod": 300,
+    "OS.RemovePersistentNetRulesPeriod": 30,
+    "OS.RootDeviceScsiTimeoutPeriod": 30,
+    "OS.MonitorDhcpClientRestartPeriod": 30,
     "OS.SshClientAliveInterval": 180,
+    "Provisioning.MonitorHostNamePeriod": 30,
     "Provisioning.PasswordCryptSaltLength": 10,
     "HttpProxy.Port": None,
     "ResourceDisk.SwapSizeMB": 0,
-    "Autoupdate.Frequency": 3600
+    "Autoupdate.Frequency": 3600,
+    "Logs.CollectPeriod": 3600,
+    #
+    # "Debug" options are experimental and may be removed in later
+    # versions of the Agent.
+    #
+    "Debug.CgroupCheckPeriod": 300,
+    "Debug.AgentCpuQuota": 50,
+    "Debug.AgentCpuThrottledTimeThreshold": 120,
+    "Debug.AgentMemoryQuota": 30 * 1024 ** 2,
+    "Debug.EtpCollectionPeriod": 300,
+    "Debug.AutoUpdateHotfixFrequency": 14400,
+    "Debug.AutoUpdateNormalFrequency": 86400,
+    "Debug.FirewallRulesLogPeriod": 86400
 }
 
 
@@ -160,10 +209,40 @@ def get_configuration(conf=__conf__):
     return options
 
 
+def get_default_value(option):
+    if option in __STRING_OPTIONS__:
+        return __STRING_OPTIONS__[option]
+    raise ValueError("{0} is not a valid configuration parameter.".format(option))
+
+
+def get_int_default_value(option):
+    if option in __INTEGER_OPTIONS__:
+        return int(__INTEGER_OPTIONS__[option])
+    raise ValueError("{0} is not a valid configuration parameter.".format(option))
+
+
+def get_switch_default_value(option):
+    if option in __SWITCH_OPTIONS__:
+        return __SWITCH_OPTIONS__[option]
+    raise ValueError("{0} is not a valid configuration parameter.".format(option))
+
+
 def enable_firewall(conf=__conf__):
     return conf.get_switch("OS.EnableFirewall", False)
 
 
+def get_enable_firewall_period(conf=__conf__):
+    return conf.get_int("OS.EnableFirewallPeriod", 300)
+
+
+def get_remove_persistent_net_rules_period(conf=__conf__):
+    return conf.get_int("OS.RemovePersistentNetRulesPeriod", 30)
+
+
+def get_monitor_dhcp_client_restart_period(conf=__conf__):
+    return conf.get_int("OS.MonitorDhcpClientRestartPeriod", 30)
+
+
 def enable_rdma(conf=__conf__):
     return conf.get_switch("OS.EnableRDMA", False) or \
            conf.get_switch("OS.UpdateRdmaDriver", False) or \
@@ -186,11 +265,20 @@ def get_logs_console(conf=__conf__):
     return conf.get_switch("Logs.Console", True)
 
 
+def get_collect_logs(conf=__conf__):
+    return conf.get_switch("Logs.Collect", True)
+
+
+def get_collect_logs_period(conf=__conf__):
+    return conf.get_int("Logs.CollectPeriod", 3600)
+
+
 def get_lib_dir(conf=__conf__):
     return conf.get("Lib.Dir", "/var/lib/waagent")
 
 
 def get_published_hostname(conf=__conf__):
+    # Some applications rely on this file; do not remove this setting
     return os.path.join(get_lib_dir(conf), 'published_hostname')
 
 
@@ -206,6 +294,10 @@ def get_ext_log_dir(conf=__conf__):
     return conf.get("Extension.LogDir", "/var/log/azure")
 
 
+def get_agent_log_file():
+    return "/var/log/waagent.log"
+
+
 def get_fips_enabled(conf=__conf__):
     return conf.get_switch("OS.EnableFIPS", False)
 
@@ -244,18 +336,22 @@ def get_ssh_key_glob(conf=__conf__):
 
 def get_ssh_key_private_path(conf=__conf__):
     return os.path.join(get_ssh_dir(conf),
-        'ssh_host_{0}_key'.format(get_ssh_host_keypair_type(conf)))
+        'ssh_host_{0}_key'.format(get_ssh_host_keypair_type(conf))) 
 
 
 def get_ssh_key_public_path(conf=__conf__):
     return os.path.join(get_ssh_dir(conf),
-        'ssh_host_{0}_key.pub'.format(get_ssh_host_keypair_type(conf)))
+        'ssh_host_{0}_key.pub'.format(get_ssh_host_keypair_type(conf))) 
 
 
 def get_root_device_scsi_timeout(conf=__conf__):
     return conf.get("OS.RootDeviceScsiTimeout", None)
 
 
+def get_root_device_scsi_timeout_period(conf=__conf__):
+    return conf.get_int("OS.RootDeviceScsiTimeoutPeriod", 30)
+
+
 def get_ssh_host_keypair_type(conf=__conf__):
     keypair_type = conf.get("Provisioning.SshHostKeyPairType", "rsa")
     if keypair_type == "auto":
@@ -275,6 +371,14 @@ def get_extensions_enabled(conf=__conf__):
     return conf.get_switch("Extensions.Enabled", True)
 
 
+def get_goal_state_period(conf=__conf__):
+    return conf.get_int("Extensions.GoalStatePeriod", 6)
+
+
+def get_initial_goal_state_period(conf=__conf__):
+    return conf.get_int("Extensions.InitialGoalStatePeriod", default_value=lambda: get_goal_state_period(conf=conf))
+
+
 def get_allow_reset_sys_user(conf=__conf__):
     return conf.get_switch("Provisioning.AllowResetSysUser", False)
 
@@ -322,6 +426,10 @@ def get_monitor_hostname(conf=__conf__):
     return conf.get_switch("Provisioning.MonitorHostName", False)
 
 
+def get_monitor_hostname_period(conf=__conf__):
+    return conf.get_int("Provisioning.MonitorHostNamePeriod", 30)
+
+
 def get_httpproxy_host(conf=__conf__):
     return conf.get("HttpProxy.Host", None)
 
@@ -340,10 +448,12 @@ def get_resourcedisk_format(conf=__conf__):
 
 def get_resourcedisk_enable_swap(conf=__conf__):
     return conf.get_switch("ResourceDisk.EnableSwap", False)
-    
+
+
 def get_resourcedisk_enable_swap_encryption(conf=__conf__):
     return conf.get_switch("ResourceDisk.EnableSwapEncryption", False)
 
+
 def get_resourcedisk_mountpoint(conf=__conf__):
     return conf.get("ResourceDisk.MountPoint", "/mnt/resource")
 
@@ -384,10 +494,151 @@ def get_disable_agent_file_path(conf=__conf__):
     return os.path.join(get_lib_dir(conf), DISABLE_AGENT_FILE)
 
 
-def get_cgroups_enforce_limits(conf=__conf__):
-    return conf.get_switch("CGroups.EnforceLimits", False)
+def get_cgroups_enabled(conf=__conf__):
+    return conf.get_switch("CGroups.Enabled", True)
+
+
+def get_monitor_network_configuration_changes(conf=__conf__):
+    return conf.get_switch("Monitor.NetworkConfigurationChanges", False)
+
+
+def get_cgroup_check_period(conf=__conf__):
+    """
+    How often to perform checks on cgroups (are the processes in the cgroups as expected,
+    has the agent exceeded its quota, etc)
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_int("Debug.CgroupCheckPeriod", 300)
+
 
+def get_cgroup_log_metrics(conf=__conf__):
+    """
+    If True, resource usage metrics are written to the local log
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_switch("Debug.CgroupLogMetrics", False)
+
+
+def get_cgroup_disable_on_process_check_failure(conf=__conf__):
+    """
+    If True, cgroups will be disabled if the process check fails
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_switch("Debug.CgroupDisableOnProcessCheckFailure", True)
+
+
+def get_cgroup_disable_on_quota_check_failure(conf=__conf__):
+    """
+    If True, cgroups will be disabled if the CPU quota check fails
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_switch("Debug.CgroupDisableOnQuotaCheckFailure", True)
+
+
+def get_agent_cpu_quota(conf=__conf__):
+    """
+    CPU quota for the agent as a percentage of 1 CPU (100% == 1 CPU)
 
-def get_cgroups_excluded(conf=__conf__):
-    excluded_value = conf.get("CGroups.Excluded", "customscript, runcommand")
-    return [s for s in [i.strip().lower() for i in excluded_value.split(',')] if len(s) > 0] if excluded_value else []
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_int("Debug.AgentCpuQuota", 50)
+
+
+def get_agent_cpu_throttled_time_threshold(conf=__conf__):
+    """
+    Throttled time threshold for agent cpu in seconds.
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_int("Debug.AgentCpuThrottledTimeThreshold", 120)
+
+
+def get_agent_memory_quota(conf=__conf__):
+    """
+    Memory quota for the agent in bytes.
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_int("Debug.AgentMemoryQuota", 30 * 1024 ** 2)
+
+
+def get_enable_agent_memory_usage_check(conf=__conf__):
+    """
+    If True, Agent checks it's Memory usage.
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_switch("Debug.EnableAgentMemoryUsageCheck", False)
+
+
+def get_cgroup_monitor_expiry_time(conf=__conf__):
+    """
+    cgroups monitoring for pilot extensions disabled after expiry time
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get("Debug.CgroupMonitorExpiryTime", "2022-03-31")
+
+
+def get_cgroup_monitor_extension_name (conf=__conf__):
+    """
+    cgroups monitoring extension name
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get("Debug.CgroupMonitorExtensionName", "Microsoft.Azure.Monitor.AzureMonitorLinuxAgent")
+
+
+def get_enable_fast_track(conf=__conf__):
+    """
+    If True, the agent use FastTrack when retrieving goal states
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_switch("Debug.EnableFastTrack", True)
+
+
+def get_etp_collection_period(conf=__conf__):
+    """
+    Determines the frequency to perform ETP collection on extensions telemetry events.
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_int("Debug.EtpCollectionPeriod", 300)
+
+
+def get_hotfix_upgrade_frequency(conf=__conf__):
+    """
+    Determines the frequency to check for Hotfix upgrades (<Patch>.<Build> version changed in new upgrades).
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_int("Debug.AutoUpdateHotfixFrequency", 4 * 60 * 60)
+
+
+def get_normal_upgrade_frequency(conf=__conf__):
+    """
+    Determines the frequency to check for Normal upgrades (<Major>.<Minor> version changed in new upgrades).
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_int("Debug.AutoUpdateNormalFrequency", 24 * 60 * 60)
+
+
+def get_enable_ga_versioning(conf=__conf__):
+    """
+    If True, the agent uses GA Versioning for auto-updating the agent vs automatically auto-updating to the highest version.
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_switch("Debug.EnableGAVersioning", False)
+
+
+def get_firewall_rules_log_period(conf=__conf__):
+    """
+    Determine the frequency to perform the periodic operation of logging firewall rules.
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_int("Debug.FirewallRulesLogPeriod", 86400)
diff --git a/azurelinuxagent/common/datacontract.py b/azurelinuxagent/common/datacontract.py
index c69bebc..b6d1f3c 100644
--- a/azurelinuxagent/common/datacontract.py
+++ b/azurelinuxagent/common/datacontract.py
@@ -20,9 +20,11 @@
 from azurelinuxagent.common.exception import ProtocolError
 import azurelinuxagent.common.logger as logger
 
+# pylint: disable=W0105
 """
 Base class for data contracts between guest and host and utilities to manipulate the properties in those contracts
-"""
+""" 
+# pylint: enable=W0105
 
 
 class DataContract(object):
@@ -30,7 +32,7 @@ class DataContract(object):
 
 
 class DataContractList(list):
-    def __init__(self, item_cls):
+    def __init__(self, item_cls):  # pylint: disable=W0231
         self.item_cls = item_cls
 
 
diff --git a/azurelinuxagent/common/dhcp.py b/azurelinuxagent/common/dhcp.py
index 5974965..3db58d3 100644
--- a/azurelinuxagent/common/dhcp.py
+++ b/azurelinuxagent/common/dhcp.py
@@ -14,23 +14,23 @@
 #
 # Requires Python 2.6+ and Openssl 1.0+
 
+import array
 import os
 import socket
-import array
 import time
+
 import azurelinuxagent.common.logger as logger
+from azurelinuxagent.common.exception import DhcpError
+from azurelinuxagent.common.osutil import get_osutil
+from azurelinuxagent.common.utils.restutil import KNOWN_WIRESERVER_IP
 from azurelinuxagent.common.utils.textutil import hex_dump, hex_dump2, \
     hex_dump3, \
     compare_bytes, str_to_ord, \
     unpack_big_endian, \
     int_to_ip4_addr
-from azurelinuxagent.common.exception import DhcpError
-from azurelinuxagent.common.osutil import get_osutil
-
 
 # the kernel routing table representation of 168.63.129.16
 KNOWN_WIRESERVER_IP_ENTRY = '10813FA8'
-from azurelinuxagent.common.utils.restutil import KNOWN_WIRESERVER_IP
 
 
 def get_dhcp_handler():
@@ -86,7 +86,7 @@ class DhcpHandler(object):
         logger.info("Test for route to {0}".format(KNOWN_WIRESERVER_IP))
         try:
             route_table = self.osutil.read_route_table()
-            if any([(KNOWN_WIRESERVER_IP_ENTRY in route) for route in route_table]):
+            if any((KNOWN_WIRESERVER_IP_ENTRY in route) for route in route_table):
                 # reset self.gateway and self.routes
                 # we do not need to alter the routing table
                 self.endpoint = KNOWN_WIRESERVER_IP
@@ -100,7 +100,7 @@ class DhcpHandler(object):
             logger.error(
                 "Could not determine whether route exists to {0}: {1}".format(
                     KNOWN_WIRESERVER_IP, e))
-                    
+
         return route_exists
 
     @property
@@ -116,7 +116,7 @@ class DhcpHandler(object):
         exists = False
 
         logger.info("Checking for dhcp lease cache")
-        cached_endpoint = self.osutil.get_dhcp_lease_endpoint()
+        cached_endpoint = self.osutil.get_dhcp_lease_endpoint()  # pylint: disable=E1128
         if cached_endpoint is not None:
             self.endpoint = cached_endpoint
             exists = True
@@ -157,11 +157,14 @@ class DhcpHandler(object):
             self.endpoint = KNOWN_WIRESERVER_IP
             return
 
+        # pylint: disable=W0105
         """
         Build dhcp request with mac addr
         Configure route to allow dhcp traffic
         Stop dhcp service if necessary
         """
+        # pylint: enable=W0105
+
         logger.info("Send dhcp request")
         mac_addr = self.osutil.get_mac_addr()
 
@@ -194,7 +197,7 @@ class DhcpHandler(object):
         self.endpoint, self.gateway, self.routes = parse_dhcp_resp(resp)
 
 
-def validate_dhcp_resp(request, response):
+def validate_dhcp_resp(request, response):  # pylint: disable=R1710
     bytes_recv = len(response)
     if bytes_recv < 0xF6:
         logger.error("HandleDhcpResponse: Too few bytes received:{0}",
@@ -228,7 +231,7 @@ def validate_dhcp_resp(request, response):
                         "doesn't match the request")
 
 
-def parse_route(response, option, i, length, bytes_recv):
+def parse_route(response, option, i, length, bytes_recv):  # pylint: disable=W0613
     # http://msdn.microsoft.com/en-us/library/cc227282%28PROT.10%29.aspx
     logger.verbose("Routes at offset: {0} with length:{1}", hex(i),
                    hex(length))
@@ -386,7 +389,7 @@ def build_dhcp_request(mac_addr, request_broadcast):
         # set broadcast flag to true to request the dhcp server
         # to respond to a boradcast address,
         # this is useful when user dhclient fails.
-        request[0x0A] = 0x80;
+        request[0x0A] = 0x80
 
     # fill in ClientHardwareAddress
     for a in range(0, 6):
diff --git a/azurelinuxagent/common/event.py b/azurelinuxagent/common/event.py
index 5274b80..1f903a9 100644
--- a/azurelinuxagent/common/event.py
+++ b/azurelinuxagent/common/event.py
@@ -18,30 +18,35 @@
 import atexit
 import json
 import os
+import platform
 import re
 import sys
 import threading
 import time
 import traceback
-from collections import namedtuple
 from datetime import datetime
 
 import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.logger as logger
-from azurelinuxagent.common.exception import EventError
-from azurelinuxagent.common.future import ustr, OrderedDict
-from azurelinuxagent.common.datacontract import get_properties, DataContractList
-from azurelinuxagent.common.telemetryevent import TelemetryEventParam, TelemetryEvent
+from azurelinuxagent.common.AgentGlobals import AgentGlobals
+from azurelinuxagent.common.exception import EventError, OSUtilError
+from azurelinuxagent.common.future import ustr
+from azurelinuxagent.common.datacontract import get_properties, set_properties
+from azurelinuxagent.common.osutil import get_osutil
+from azurelinuxagent.common.telemetryevent import TelemetryEventParam, TelemetryEvent, CommonTelemetryEventSchema, \
+    GuestAgentGenericLogsSchema, GuestAgentExtensionEventsSchema, GuestAgentPerfCounterEventsSchema
 from azurelinuxagent.common.utils import fileutil, textutil
-from azurelinuxagent.common.version import CURRENT_VERSION, CURRENT_AGENT
+from azurelinuxagent.common.utils.textutil import parse_doc, findall, find, getattrib, str_to_encoded_ustr
+from azurelinuxagent.common.version import CURRENT_VERSION, CURRENT_AGENT, AGENT_NAME, DISTRO_NAME, DISTRO_VERSION, DISTRO_CODE_NAME, AGENT_EXECUTION_MODE
+from azurelinuxagent.common.protocol.imds import get_imds_client
+
+EVENTS_DIRECTORY = "events"
 
 _EVENT_MSG = "Event: name={0}, op={1}, message={2}, duration={3}"
 TELEMETRY_EVENT_PROVIDER_ID = "69B669B9-4AF8-4C50-BDC4-6006FA76E975"
+TELEMETRY_EVENT_EVENT_ID = 1
 TELEMETRY_METRICS_EVENT_ID = 4
 
-# Store the last retrieved container id as an environment variable to be shared between threads for telemetry purposes
-CONTAINER_ID_ENV_VARIABLE = "AZURE_GUEST_AGENT_CONTAINER_ID"
-
 TELEMETRY_LOG_PROVIDER_ID = "FFF0196F-EE4C-4EAF-9AA5-776F622DEB4F"
 TELEMETRY_LOG_EVENT_ID = 7
 
@@ -53,33 +58,39 @@ SEND_LOGS_TO_TELEMETRY = False
 
 MAX_NUMBER_OF_EVENTS = 1000
 
+AGENT_EVENT_FILE_EXTENSION = '.waagent.tld'
+EVENT_FILE_REGEX = re.compile(r'(?P<agent_event>\.waagent)?\.tld$')
 
 def send_logs_to_telemetry():
     return SEND_LOGS_TO_TELEMETRY
 
 
-def get_container_id_from_env():
-    return os.environ.get(CONTAINER_ID_ENV_VARIABLE, "UNINITIALIZED")
-
-
 class WALAEventOperation:
     ActivateResourceDisk = "ActivateResourceDisk"
     AgentBlacklisted = "AgentBlacklisted"
     AgentEnabled = "AgentEnabled"
+    AgentMemory = "AgentMemory"
+    AgentUpgrade = "AgentUpgrade"
     ArtifactsProfileBlob = "ArtifactsProfileBlob"
-    AutoUpdate = "AutoUpdate"
-    CustomData = "CustomData"
     CGroupsCleanUp = "CGroupsCleanUp"
-    CGroupsLimitsCrossed = "CGroupsLimitsCrossed"
-    ExtensionMetricsData = "ExtensionMetricsData"
+    CGroupsDisabled = "CGroupsDisabled"
+    CGroupsInfo = "CGroupsInfo"
+    CollectEventErrors = "CollectEventErrors"
+    CollectEventUnicodeErrors = "CollectEventUnicodeErrors"
+    ConfigurationChange = "ConfigurationChange"
+    CustomData = "CustomData"
+    DefaultChannelChange = "DefaultChannelChange"
     Deploy = "Deploy"
     Disable = "Disable"
     Downgrade = "Downgrade"
     Download = "Download"
     Enable = "Enable"
     ExtensionProcessing = "ExtensionProcessing"
+    ExtensionTelemetryEventProcessing = "ExtensionTelemetryEventProcessing"
+    FetchGoalState = "FetchGoalState"
     Firewall = "Firewall"
-    GetArtifactExtended = "GetArtifactExtended"
+    GoalState = "GoalState"
+    GoalStateUnsupportedFeatures = "GoalStateUnsupportedFeatures"
     HealthCheck = "HealthCheck"
     HealthObservation = "HealthObservation"
     HeartBeat = "HeartBeat"
@@ -87,29 +98,35 @@ class WALAEventOperation:
     HostPluginHeartbeat = "HostPluginHeartbeat"
     HostPluginHeartbeatExtended = "HostPluginHeartbeatExtended"
     HttpErrors = "HttpErrors"
+    HttpGet = "HttpGet"
     ImdsHeartbeat = "ImdsHeartbeat"
     Install = "Install"
-    InitializeCGroups = "InitializeCGroups"
     InitializeHostPlugin = "InitializeHostPlugin"
-    InvokeCommandUsingSystemd = "InvokeCommandUsingSystemd"
     Log = "Log"
+    LogCollection = "LogCollection"
     OSInfo = "OSInfo"
     Partition = "Partition"
-    ProcessGoalState = "ProcessGoalState"
+    PersistFirewallRules = "PersistFirewallRules"
+    PluginSettingsVersionMismatch = "PluginSettingsVersionMismatch"
+    InvalidExtensionConfig = "InvalidExtensionConfig"
     Provision = "Provision"
     ProvisionGuestAgent = "ProvisionGuestAgent"
     RemoteAccessHandling = "RemoteAccessHandling"
+    ReportEventErrors = "ReportEventErrors"
+    ReportEventUnicodeErrors = "ReportEventUnicodeErrors"
     ReportStatus = "ReportStatus"
     ReportStatusExtended = "ReportStatusExtended"
     Restart = "Restart"
     SequenceNumberMismatch = "SequenceNumberMismatch"
     SetCGroupsLimits = "SetCGroupsLimits"
     SkipUpdate = "SkipUpdate"
+    StatusProcessing = "StatusProcessing"
     UnhandledError = "UnhandledError"
     UnInstall = "UnInstall"
     Unknown = "Unknown"
-    Upgrade = "Upgrade"
     Update = "Update"
+    VmSettings = "VmSettings"
+    VmSettingsSummary = "VmSettingsSummary"
 
 
 SHOULD_ENCODE_MESSAGE_LEN = 80
@@ -173,11 +190,57 @@ class EventStatus(object):
 
 __event_status__ = EventStatus()
 __event_status_operations__ = [
-        WALAEventOperation.AutoUpdate,
         WALAEventOperation.ReportStatus
     ]
 
 
+def parse_json_event(data_str):
+    data = json.loads(data_str)
+    event = TelemetryEvent()
+    set_properties("TelemetryEvent", event, data)
+    event.file_type = "json"
+    return event
+
+
+def parse_event(data_str):
+    try:
+        try:
+            return parse_json_event(data_str)
+        except ValueError:
+            return parse_xml_event(data_str)
+    except Exception as e:
+        raise EventError("Error parsing event: {0}".format(ustr(e)))
+
+
+def parse_xml_param(param_node):
+    name = getattrib(param_node, "Name")
+    value_str = getattrib(param_node, "Value")
+    attr_type = getattrib(param_node, "T")
+    value = value_str
+    if attr_type == 'mt:uint64':
+        value = int(value_str)
+    elif attr_type == 'mt:bool':
+        value = bool(value_str)
+    elif attr_type == 'mt:float64':
+        value = float(value_str)
+    return TelemetryEventParam(name, value)
+
+
+def parse_xml_event(data_str):
+    try:
+        xml_doc = parse_doc(data_str)
+        event_id = getattrib(find(xml_doc, "Event"), 'id')
+        provider_id = getattrib(find(xml_doc, "Provider"), 'id')
+        event = TelemetryEvent(event_id, provider_id)
+        param_nodes = findall(xml_doc, 'Param')
+        for param_node in param_nodes:
+            event.parameters.append(parse_xml_param(param_node))
+        event.file_type = "xml"
+        return event
+    except Exception as e:
+        raise ValueError(ustr(e))
+
+
 def _encode_message(op, message):
     """
     Gzip and base64 encode a message based on the operation.
@@ -214,20 +277,158 @@ def _encode_message(op, message):
 
 
 def _log_event(name, op, message, duration, is_success=True):
-    global _EVENT_MSG
+    global _EVENT_MSG  # pylint: disable=W0603
 
-    message = _encode_message(op, message)
     if not is_success:
         logger.error(_EVENT_MSG, name, op, message, duration)
     else:
         logger.info(_EVENT_MSG, name, op, message, duration)
 
 
+class CollectOrReportEventDebugInfo(object):
+    """
+    This class is used for capturing and reporting debug info that is captured during event collection and
+    reporting to wireserver.
+    It captures the count of unicode errors and any unexpected errors and also a subset of errors with stacks to help
+    with debugging any potential issues.
+    """
+    __MAX_ERRORS_TO_REPORT = 5
+    OP_REPORT = "Report"
+    OP_COLLECT = "Collect"
+
+    def __init__(self, operation=OP_REPORT):
+        self.__unicode_error_count = 0
+        self.__unicode_errors = set()
+        self.__op_error_count = 0
+        self.__op_errors = set()
+
+        if operation == self.OP_REPORT:
+            self.__unicode_error_event = WALAEventOperation.ReportEventUnicodeErrors
+            self.__op_errors_event = WALAEventOperation.ReportEventErrors
+        elif operation == self.OP_COLLECT:
+            self.__unicode_error_event = WALAEventOperation.CollectEventUnicodeErrors
+            self.__op_errors_event = WALAEventOperation.CollectEventErrors
+
+    def report_debug_info(self):
+
+        def report_dropped_events_error(count, errors, operation_name):
+            err_msg_format = "DroppedEventsCount: {0}\nReasons (first {1} errors): {2}"
+            if count > 0:
+                add_event(op=operation_name,
+                          message=err_msg_format.format(count, CollectOrReportEventDebugInfo.__MAX_ERRORS_TO_REPORT, ', '.join(errors)),
+                          is_success=False)
+
+        report_dropped_events_error(self.__op_error_count, self.__op_errors, self.__op_errors_event)
+        report_dropped_events_error(self.__unicode_error_count, self.__unicode_errors, self.__unicode_error_event)
+
+    @staticmethod
+    def _update_errors_and_get_count(error_count, errors, error):
+        error_count += 1
+        if len(errors) < CollectOrReportEventDebugInfo.__MAX_ERRORS_TO_REPORT:
+            errors.add("{0}: {1}".format(ustr(error), traceback.format_exc()))
+        return error_count
+
+    def update_unicode_error(self, unicode_err):
+        self.__unicode_error_count = self._update_errors_and_get_count(self.__unicode_error_count, self.__unicode_errors,
+                                                                       unicode_err)
+
+    def update_op_error(self, op_err):
+        self.__op_error_count = self._update_errors_and_get_count(self.__op_error_count, self.__op_errors, op_err)
+
+
 class EventLogger(object):
     def __init__(self):
         self.event_dir = None
         self.periodic_events = {}
 
+        #
+        # All events should have these parameters.
+        #
+        # The first set comes from the current OS and is initialized here. These values don't change during
+        # the agent's lifetime.
+        #
+        # The next two sets come from the goal state and IMDS and must be explicitly initialized using
+        # initialize_vminfo_common_parameters() once a protocol for communication with the host has been
+        # created. Their values  don't change during the agent's lifetime. Note that we initialize these
+        # parameters here using dummy values (*_UNINITIALIZED) since events sent to the host should always
+        # match the schema defined for them in the telemetry pipeline.
+        #
+        # There is another set of common parameters that must be computed at the time the event is created
+        # (e.g. the timestamp and the container ID); those are added to events (along with the parameters
+        # below) in _add_common_event_parameters()
+        #
+        # Note that different kinds of events may also include other parameters; those are added by the
+        # corresponding add_* method (e.g. add_metric for performance metrics).
+        #
+        self._common_parameters = []
+
+        # Parameters from OS
+        osutil = get_osutil()
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.OSVersion, EventLogger._get_os_version()))
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.ExecutionMode, AGENT_EXECUTION_MODE))
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.RAM, int(EventLogger._get_ram(osutil))))
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.Processors, int(EventLogger._get_processors(osutil))))
+
+        # Parameters from goal state
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.TenantName, "TenantName_UNINITIALIZED"))
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.RoleName, "RoleName_UNINITIALIZED"))
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.RoleInstanceName, "RoleInstanceName_UNINITIALIZED"))
+        #
+        # # Parameters from IMDS
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.Location, "Location_UNINITIALIZED"))
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.SubscriptionId, "SubscriptionId_UNINITIALIZED"))
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.ResourceGroupName, "ResourceGroupName_UNINITIALIZED"))
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.VMId, "VMId_UNINITIALIZED"))
+        self._common_parameters.append(TelemetryEventParam(CommonTelemetryEventSchema.ImageOrigin, 0))
+
+    @staticmethod
+    def _get_os_version():
+        return "{0}:{1}-{2}-{3}:{4}".format(platform.system(), DISTRO_NAME, DISTRO_VERSION, DISTRO_CODE_NAME, platform.release())
+
+    @staticmethod
+    def _get_ram(osutil):
+        try:
+            return osutil.get_total_mem()
+        except OSUtilError as e:
+            logger.warn("Failed to get RAM info; will be missing from telemetry: {0}", ustr(e))
+        return 0
+
+    @staticmethod
+    def _get_processors(osutil):
+        try:
+            return osutil.get_processor_cores()
+        except OSUtilError as e:
+            logger.warn("Failed to get Processors info; will be missing from telemetry: {0}", ustr(e))
+        return 0
+
+    def initialize_vminfo_common_parameters(self, protocol):
+        """
+        Initializes the common parameters that come from the goal state and IMDS
+        """
+        # create an index of the event parameters for faster updates
+        parameters = {}
+        for p in self._common_parameters:
+            parameters[p.name] = p
+
+        try:
+            vminfo = protocol.get_vminfo()
+            parameters[CommonTelemetryEventSchema.TenantName].value = vminfo.tenantName
+            parameters[CommonTelemetryEventSchema.RoleName].value = vminfo.roleName
+            parameters[CommonTelemetryEventSchema.RoleInstanceName].value = vminfo.roleInstanceName
+        except Exception as e:
+            logger.warn("Failed to get VM info from goal state; will be missing from telemetry: {0}", ustr(e))
+
+        try:
+            imds_client = get_imds_client(protocol.get_endpoint())
+            imds_info = imds_client.get_compute()
+            parameters[CommonTelemetryEventSchema.Location].value = imds_info.location
+            parameters[CommonTelemetryEventSchema.SubscriptionId].value = imds_info.subscriptionId
+            parameters[CommonTelemetryEventSchema.ResourceGroupName].value = imds_info.resourceGroupName
+            parameters[CommonTelemetryEventSchema.VMId].value = imds_info.vmId
+            parameters[CommonTelemetryEventSchema.ImageOrigin].value = int(imds_info.image_origin)
+        except Exception as e:
+            logger.warn("Failed to get IMDS info; will be missing from telemetry: {0}", ustr(e))
+
     def save_event(self, data):
         if self.event_dir is None:
             logger.warn("Cannot save event -- Event reporter is not initialized.")
@@ -258,7 +459,7 @@ class EventLogger(object):
         try:
             with open(filename + ".tmp", 'wb+') as hfile:
                 hfile.write(data.encode("utf-8"))
-            os.rename(filename + ".tmp", filename + ".tld")
+            os.rename(filename + ".tmp", filename + AGENT_EVENT_FILE_EXTENSION)
         except (IOError, OSError) as e:
             msg = "Failed to write events to file: {0}".format(e)
             raise EventError(msg)
@@ -271,38 +472,31 @@ class EventLogger(object):
             (self.periodic_events[h] + delta) <= datetime.now()
 
     def add_periodic(self, delta, name, op=WALAEventOperation.Unknown, is_success=True, duration=0,
-                     version=str(CURRENT_VERSION), message="", evt_type="", is_internal=False, log_event=True,
-                     force=False):
+                     version=str(CURRENT_VERSION), message="", log_event=True, force=False):
         h = hash(name + op + ustr(is_success) + message)
 
         if force or self.is_period_elapsed(delta, h):
             self.add_event(name, op=op, is_success=is_success, duration=duration,
-                           version=version, message=message, evt_type=evt_type,
-                           is_internal=is_internal, log_event=log_event)
+                           version=version, message=message, log_event=log_event)
             self.periodic_events[h] = datetime.now()
 
     def add_event(self, name, op=WALAEventOperation.Unknown, is_success=True, duration=0, version=str(CURRENT_VERSION),
-                  message="", evt_type="", is_internal=False, log_event=True):
+                  message="", log_event=True):
 
         if (not is_success) and log_event:
             _log_event(name, op, message, duration, is_success=is_success)
 
-        self._add_event(duration, evt_type, is_internal, is_success, message, name, op, version, event_id=1)
-
-    def _add_event(self, duration, evt_type, is_internal, is_success, message, name, op, version, event_id):
-        event = TelemetryEvent(event_id, TELEMETRY_EVENT_PROVIDER_ID)
-
-        event.parameters.append(TelemetryEventParam('Name', str(name)))
-        event.parameters.append(TelemetryEventParam('Version', str(version)))
-        event.parameters.append(TelemetryEventParam('IsInternal', bool(is_internal)))
-        event.parameters.append(TelemetryEventParam('Operation', str(op)))
-        event.parameters.append(TelemetryEventParam('OperationSuccess', bool(is_success)))
-        event.parameters.append(TelemetryEventParam('Message', str(message)))
-        event.parameters.append(TelemetryEventParam('Duration', int(duration)))
-        event.parameters.append(TelemetryEventParam('ExtensionType', str(evt_type)))
+        event = TelemetryEvent(TELEMETRY_EVENT_EVENT_ID, TELEMETRY_EVENT_PROVIDER_ID)
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Name, str_to_encoded_ustr(name)))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Version, str_to_encoded_ustr(version)))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Operation, str_to_encoded_ustr(op)))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.OperationSuccess, bool(is_success)))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Message, str_to_encoded_ustr(message)))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Duration, int(duration)))
+        self.add_common_event_parameters(event, datetime.utcnow())
 
-        event.parameters = self.add_default_parameters_to_event(event.parameters)
         data = get_properties(event)
+
         try:
             self.save_event(json.dumps(data))
         except EventError as e:
@@ -310,13 +504,13 @@ class EventLogger(object):
 
     def add_log_event(self, level, message):
         event = TelemetryEvent(TELEMETRY_LOG_EVENT_ID, TELEMETRY_LOG_PROVIDER_ID)
-        event.parameters.append(TelemetryEventParam('EventName', WALAEventOperation.Log))
-        event.parameters.append(TelemetryEventParam('CapabilityUsed', logger.LogLevel.STRINGS[level]))
-        event.parameters.append(TelemetryEventParam('Context1', self._clean_up_message(message)))
-        event.parameters.append(TelemetryEventParam('Context2', ''))
-        event.parameters.append(TelemetryEventParam('Context3', ''))
+        event.parameters.append(TelemetryEventParam(GuestAgentGenericLogsSchema.EventName, WALAEventOperation.Log))
+        event.parameters.append(TelemetryEventParam(GuestAgentGenericLogsSchema.CapabilityUsed, logger.LogLevel.STRINGS[level]))
+        event.parameters.append(TelemetryEventParam(GuestAgentGenericLogsSchema.Context1, str_to_encoded_ustr(self._clean_up_message(message))))
+        event.parameters.append(TelemetryEventParam(GuestAgentGenericLogsSchema.Context2, datetime.utcnow().strftime(logger.Logger.LogTimeFormatInUTC)))
+        event.parameters.append(TelemetryEventParam(GuestAgentGenericLogsSchema.Context3, ''))
+        self.add_common_event_parameters(event, datetime.utcnow())
 
-        event.parameters = self.add_default_parameters_to_event(event.parameters)
         data = get_properties(event)
         try:
             self.save_event(json.dumps(data))
@@ -334,17 +528,16 @@ class EventLogger(object):
         :param bool log_event: If true, log the collected metric in the agent log
         """
         if log_event:
-            from azurelinuxagent.common.version import AGENT_NAME
             message = "Metric {0}/{1} [{2}] = {3}".format(category, counter, instance, value)
             _log_event(AGENT_NAME, "METRIC", message, 0)
 
         event = TelemetryEvent(TELEMETRY_METRICS_EVENT_ID, TELEMETRY_EVENT_PROVIDER_ID)
-        event.parameters.append(TelemetryEventParam('Category', str(category)))
-        event.parameters.append(TelemetryEventParam('Counter', str(counter)))
-        event.parameters.append(TelemetryEventParam('Instance', str(instance)))
-        event.parameters.append(TelemetryEventParam('Value', float(value)))
+        event.parameters.append(TelemetryEventParam(GuestAgentPerfCounterEventsSchema.Category, str_to_encoded_ustr(category)))
+        event.parameters.append(TelemetryEventParam(GuestAgentPerfCounterEventsSchema.Counter, str_to_encoded_ustr(counter)))
+        event.parameters.append(TelemetryEventParam(GuestAgentPerfCounterEventsSchema.Instance, str_to_encoded_ustr(instance)))
+        event.parameters.append(TelemetryEventParam(GuestAgentPerfCounterEventsSchema.Value, float(value)))
+        self.add_common_event_parameters(event, datetime.utcnow())
 
-        event.parameters = self.add_default_parameters_to_event(event.parameters)
         data = get_properties(event)
         try:
             self.save_event(json.dumps(data))
@@ -392,56 +585,34 @@ class EventLogger(object):
             else:
                 return message
 
-    @staticmethod
-    def add_default_parameters_to_event(event_parameters, set_values_for_agent=True):
+    def add_common_event_parameters(self, event, event_timestamp):
         """
-        Default fields are only populated by Agent and not the extension. Agent will fill up any event if they don't
-        have the default params. Example: GAVersion and ContainerId are populated for agent events on the fly,
-        but not for extension events. Add it if it's missing.
-
-        We write the GAVersion here rather than add it in azurelinuxagent.ga.monitor.MonitorHandler.add_sysinfo
-        as there could be a possibility of events being sent with newer version of the agent, rather than the agent
-        version generating the event.
-        # Old behavior example: V1 writes the event on the disk and finds an update immediately, and updates. Now the
-        new monitor thread would pick up the events from the disk and send it with the CURRENT_AGENT, which would have
-        newer version of the agent. This causes confusion.
-
-        ContainerId can change due to live migration and we want to preserve the container Id of the container writing
-        the event, rather than sending the event.
-        OpcodeName - This is used as the actual time of event generation.
-
-        :param event_parameters: List of parameters of the event.
-        :param set_values_for_agent: Need default values populated or not. Extensions need only GAVersion and
-                                            ContainerId to be populated and others should be
-        :return: Event with default parameters populated (either values for agent or extension)
+        This method is called for all events and ensures all telemetry fields are added before the event is sent out.
+        Note that the event timestamp is saved in the OpcodeName field.
         """
-        DefaultParameter = namedtuple('DefaultParameter', ['name', 'value'])
-        default_parameters = [DefaultParameter("GAVersion", CURRENT_AGENT),
-                              DefaultParameter('ContainerId', get_container_id_from_env()),
-                              DefaultParameter('OpcodeName', datetime.utcnow().__str__() if set_values_for_agent else ""),
-                              DefaultParameter('EventTid', threading.current_thread().ident if set_values_for_agent else 0),
-                              DefaultParameter('EventPid', os.getpid() if set_values_for_agent else 0),
-                              DefaultParameter("TaskName", threading.current_thread().getName() if set_values_for_agent else ""),
-                              DefaultParameter("KeywordName", '')]
-