Repository: Azure/WALinuxAgent
Pull request #1743 (2020-01-06 19:18:22)

Make protocol util a singleton per thread
<!-- DO NOT DELETE THIS TEMPLATE -->

## Description
Recently we discovered a bug in the Guest Agent where the WireClient object different for different handlers which was due to the fact that a new ProtocolUtil object was being instantiated for every Hanlder (UpdateHandler, ExtHandler, etc). 
This PR addresses this issue by making sure that a single ProtocolUtil object is used for each thread. It was made a SingletonPerThread to avoid the concurrency issues that we face if we make it a generic singleton (and also because the majority of the processing is done in the ExtHandler thread and not in the other threads)

Issue # <!-- if any -->
<!--
Please add an informative description that covers that changes made by the pull request. 
This checklist is used to make sure that common issues in a pull request are addressed.
This will expedite the process of getting your pull request merged and avoid extra work on your part to fix issues discovered during the review process.
-->

---

### PR information
- [ ] The title of the PR is clear and informative.
- [ ] There are a small number of commits, each of which has an informative message. This means that previously merged commits do not appear in the history of the PR. For information on cleaning up the commits in your pull request, [see this page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).
- [ ] 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.

### Quality of Code and Contribution Guidelines
- [ ] I have read the [contribution guidelines](https://github.com/Azure/WALinuxAgent/blob/master/.github/CONTRIBUTING.md).

<!-- Reviewable:start -->
---
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/azure/walinuxagent/1743)
<!-- Reviewable:end -->


codecov[bot]: # [Codecov](https://codecov.io/gh/Azure/WALinuxAgent/pull/1743?src=pr&el=h1) Report
> Merging [#1743](https://codecov.io/gh/Azure/WALinuxAgent/pull/1743?src=pr&el=desc) into [develop](https://codecov.io/gh/Azure/WALinuxAgent/commit/dc13ed2ed2aa0891c1f7bcae11fd8c2e721c718b?src=pr&el=desc) will **increase** coverage by `0.05%`.
> The diff coverage is `85.05%`.

[![Impacted file tree graph](https://codecov.io/gh/Azure/WALinuxAgent/pull/1743/graphs/tree.svg?width=650&token=ae90E2KYCw&height=150&src=pr)](https://codecov.io/gh/Azure/WALinuxAgent/pull/1743?src=pr&el=tree)

```diff
@@             Coverage Diff             @@
##           develop    #1743      +/-   ##
===========================================
+ Coverage    68.51%   68.56%   +0.05%     
===========================================
  Files           81       82       +1     
  Lines        11734    11744      +10     
  Branches      1645     1647       +2     
===========================================
+ Hits          8039     8052      +13     
+ Misses        3350     3346       -4     
- Partials       345      346       +1
```


| [Impacted Files](https://codecov.io/gh/Azure/WALinuxAgent/pull/1743?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [azurelinuxagent/ga/monitor.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1743/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2dhL21vbml0b3IucHk=) | `89.24% <100%> (+1.93%)` | :arrow_up: |
| [azurelinuxagent/common/SingletonPerThread.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1743/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi9TaW5nbGV0b25QZXJUaHJlYWQucHk=) | `100% <100%> (ø)` | |
| [azurelinuxagent/common/protocol/wire.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1743/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi9wcm90b2NvbC93aXJlLnB5) | `76.48% <73.91%> (-0.17%)` | :arrow_down: |
| [azurelinuxagent/common/protocol/util.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1743/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi9wcm90b2NvbC91dGlsLnB5) | `83.41% <85.1%> (-0.9%)` | :arrow_down: |

------

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


Patch:
diff --git a/azurelinuxagent/common/protocol/util.py b/azurelinuxagent/common/protocol/util.py
index addb8d73..dceb60ec 100644
--- a/azurelinuxagent/common/protocol/util.py
+++ b/azurelinuxagent/common/protocol/util.py
@@ -27,6 +27,7 @@ import time
 import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.logger as logger
 import azurelinuxagent.common.utils.fileutil as fileutil
+from azurelinuxagent.common.singletonperthread import SingletonPerThread
 
 from azurelinuxagent.common.exception import ProtocolError, OSUtilError, \
                                       ProtocolNotFoundError, DhcpError
@@ -63,16 +64,17 @@ def get_protocol_util():
     return ProtocolUtil()
 
 
-class ProtocolUtil(object):
+class ProtocolUtil(SingletonPerThread):
     """
     ProtocolUtil handles initialization for protocol instance. 2 protocol types
     are invoked, wire protocol and metadata protocols.
+
+    Note: ProtocolUtil is a sub class of SingletonPerThread, this basically means that there would only be 1 single
+    instance of ProtocolUtil object per thread.
     """
 
     def __init__(self):
-        self.lock = threading.Lock()
-        self.lock_wireserver_endpoint = threading.Lock()
-        self.protocol = None
+        self._protocol = None
         self.endpoint = None
         self.osutil = get_osutil()
         self.dhcp_handler = get_dhcp_handler()
@@ -160,66 +162,55 @@ class ProtocolUtil(object):
             TAG_FILE_NAME)
 
     def get_wireserver_endpoint(self):
-        self.lock_wireserver_endpoint.acquire()
 
-        try:
-            if self.endpoint:
-                return self.endpoint
+        if self.endpoint:
+            return self.endpoint
 
-            file_path = self._get_wireserver_endpoint_file_path()
-            if os.path.isfile(file_path):
-                try:
-                    self.endpoint = fileutil.read_file(file_path)
+        file_path = self._get_wireserver_endpoint_file_path()
+        if os.path.isfile(file_path):
+            try:
+                self.endpoint = fileutil.read_file(file_path)
 
-                    if self.endpoint:
-                        logger.info("WireServer endpoint {0} read from file", self.endpoint)
-                        return self.endpoint
+                if self.endpoint:
+                    logger.info("WireServer endpoint {0} read from file", self.endpoint)
+                    return self.endpoint
 
-                    logger.error("[GetWireserverEndpoint] Unexpected empty file {0}: {1}", file_path, str(e))
-                except (IOError, OSError) as e:
-                    logger.error("[GetWireserverEndpoint] Error reading file {0}: {1}", file_path, str(e))
-            else:
-                logger.error("[GetWireserverEndpoint] Missing file {0}", file_path)
+                logger.error("[GetWireserverEndpoint] Unexpected empty file {0}: {1}", file_path, str(e))
+            except (IOError, OSError) as e:
+                logger.error("[GetWireserverEndpoint] Error reading file {0}: {1}", file_path, str(e))
+        else:
+            logger.error("[GetWireserverEndpoint] Missing file {0}", file_path)
 
-            self.endpoint = KNOWN_WIRESERVER_IP
-            logger.info("Using hardcoded Wireserver endpoint {0}", self.endpoint)
+        self.endpoint = KNOWN_WIRESERVER_IP
+        logger.info("Using hardcoded Wireserver endpoint {0}", self.endpoint)
 
-            return self.endpoint
-        finally:
-            self.lock_wireserver_endpoint.release()
+        return self.endpoint
 
     def _set_wireserver_endpoint(self, endpoint):
-        self.lock_wireserver_endpoint.acquire()
         try:
             self.endpoint = endpoint
             file_path = self._get_wireserver_endpoint_file_path()
             fileutil.write_file(file_path, endpoint)
         except (IOError, OSError) as e:
             raise OSUtilError(ustr(e))
-        finally:
-            self.lock_wireserver_endpoint.release()
 
     def _clear_wireserver_endpoint(self):
         """
         Cleanup previous saved wireserver endpoint.
         """
-        self.lock_wireserver_endpoint.acquire()
+
+        self.endpoint = None
+        endpoint_file_path = self._get_wireserver_endpoint_file_path()
+        if not os.path.isfile(endpoint_file_path):
+            return
 
         try:
-            self.endpoint = None
-            endpoint_file_path = self._get_wireserver_endpoint_file_path()
-            if not os.path.isfile(endpoint_file_path):
+            os.remove(endpoint_file_path)
+        except (IOError, OSError) as e:
+            # Ignore file-not-found errors (since the file is being removed)
+            if e.errno == errno.ENOENT:
                 return
-
-            try:
-                os.remove(endpoint_file_path)
-            except (IOError, OSError) as e:
-                # Ignore file-not-found errors (since the file is being removed)
-                if e.errno == errno.ENOENT:
-                    return
-                logger.error("Failed to clear wiresever endpoint: {0}", e)
-        finally:
-            self.lock_wireserver_endpoint.release()
+            logger.error("Failed to clear wiresever endpoint: {0}", e)
 
     def _detect_wire_protocol(self):
         endpoint = self.dhcp_handler.endpoint
@@ -313,7 +304,7 @@ class ProtocolUtil(object):
         """
         logger.info("Clean protocol and wireserver endpoint")
         self._clear_wireserver_endpoint()
-        self.protocol = None
+        self._protocol = None
         protocol_file_path = self._get_protocol_file_path()
         if not os.path.isfile(protocol_file_path):
             return
@@ -332,33 +323,28 @@ class ProtocolUtil(object):
         detect MetadataProtocol in priority.
         :returns: protocol instance
         """
-        self.lock.acquire()
 
-        try:
-            if self.protocol is not None:
-                return self.protocol
-
-            try:
-                self.protocol = self._get_protocol()
-                return self.protocol
-            except ProtocolNotFoundError:
-                pass
-            logger.info("Detect protocol endpoints")
-            protocols = [prots.WireProtocol]
-
-            if by_file:
-                tag_file_path = self._get_tag_file_path()
-                if os.path.isfile(tag_file_path):
-                    protocols.insert(0, prots.MetadataProtocol)
-            else:
-                protocols.append(prots.MetadataProtocol)
-            protocol_name, protocol = self._detect_protocol(protocols)
+        if self._protocol is not None:
+            return self._protocol
 
-            IOErrorCounter.set_protocol_endpoint(endpoint=protocol.get_endpoint())
-            self._save_protocol(protocol_name)
+        try:
+            self._protocol = self._get_protocol()
+            return self._protocol
+        except ProtocolNotFoundError:
+            pass
+        logger.info("Detect protocol endpoints")
+        protocols = [prots.WireProtocol]
+
+        if by_file:
+            tag_file_path = self._get_tag_file_path()
+            if os.path.isfile(tag_file_path):
+                protocols.insert(0, prots.MetadataProtocol)
+        else:
+            protocols.append(prots.MetadataProtocol)
+        protocol_name, protocol = self._detect_protocol(protocols)
 
-            self.protocol = protocol
-            return self.protocol
+        IOErrorCounter.set_protocol_endpoint(endpoint=protocol.get_endpoint())
+        self._save_protocol(protocol_name)
 
-        finally:
-            self.lock.release()
+        self._protocol = protocol
+        return self._protocol
diff --git a/azurelinuxagent/common/protocol/wire.py b/azurelinuxagent/common/protocol/wire.py
index 01d2f397..b65746e2 100644
--- a/azurelinuxagent/common/protocol/wire.py
+++ b/azurelinuxagent/common/protocol/wire.py
@@ -24,19 +24,20 @@ import re
 import time
 import xml.sax.saxutils as saxutils
 from datetime import datetime
+from threading import Lock
 
 import azurelinuxagent.common.conf as conf
-from azurelinuxagent.common.datacontract import validate_param, set_properties
-from azurelinuxagent.common.event import add_event, add_periodic, WALAEventOperation, CONTAINER_ID_ENV_VARIABLE
+import azurelinuxagent.common.logger as logger
 import azurelinuxagent.common.utils.textutil as textutil
+from azurelinuxagent.common.datacontract import validate_param, set_properties
+from azurelinuxagent.common.event import add_periodic, WALAEventOperation, CONTAINER_ID_ENV_VARIABLE
 from azurelinuxagent.common.exception import ProtocolNotFoundError, \
     ResourceGoneError, ExtensionDownloadError, InvalidContainerError, ProtocolError, HttpError
 from azurelinuxagent.common.future import httpclient, bytebuffer
-import azurelinuxagent.common.logger as logger
-from azurelinuxagent.common.utils import fileutil, restutil
 from azurelinuxagent.common.protocol.hostplugin import HostPluginProtocol
 from azurelinuxagent.common.protocol.restapi import *
 from azurelinuxagent.common.telemetryevent import TelemetryEventList
+from azurelinuxagent.common.utils import fileutil, restutil
 from azurelinuxagent.common.utils.archive import StateFlusher
 from azurelinuxagent.common.utils.cryptutil import CryptUtil
 from azurelinuxagent.common.utils.textutil import parse_doc, findall, find, \
@@ -524,11 +525,11 @@ def event_to_v1(event):
 
 
 class WireClient(object):
+
     def __init__(self, endpoint):
         logger.info("Wire server endpoint:{0}", endpoint)
         self.endpoint = endpoint
         self.goal_state = None
-        self.updated = None
         self.hosting_env = None
         self.shared_conf = None
         self.remote_access = None
@@ -669,19 +670,21 @@ class WireClient(object):
                 headers=headers,
                 use_proxy=use_proxy)
 
+            host_plugin = self.get_host_plugin()
             if restutil.request_failed(resp):
                 error_response = restutil.read_response_error(resp)
                 msg = "Fetch failed from [{0}]: {1}".format(uri, error_response)
                 logger.warn(msg)
-                if self.host_plugin is not None:
-                    self.host_plugin.report_fetch_health(uri,
-                                                         is_healthy=not restutil.request_failed_at_hostplugin(resp),
-                                                         source='WireClient',
-                                                         response=error_response)
+
+                if host_plugin is not None:
+                    host_plugin.report_fetch_health(uri,
+                                                     is_healthy=not restutil.request_failed_at_hostplugin(resp),
+                                                     source='WireClient',
+                                                     response=error_response)
                 raise ProtocolError(msg)
             else:
-                if self.host_plugin is not None:
-                    self.host_plugin.report_fetch_health(uri, source='WireClient')
+                if host_plugin is not None:
+                    host_plugin.report_fetch_health(uri, source='WireClient')
 
         except (HttpError, ProtocolError, IOError) as e:
             logger.verbose("Fetch failed from [{0}]: {1}", uri, e)
@@ -808,7 +811,6 @@ class WireClient(object):
                         self.save_cache(goal_state_file, xml_text)
 
                     self.goal_state_flusher.flush(datetime.utcnow())
-
                     self.goal_state = new_goal_state
                     save_goal_state(new_goal_state.incarnation, new_goal_state_xml)
                     self.update_hosting_env(new_goal_state)
@@ -819,7 +821,6 @@ class WireClient(object):
                     self.save_cache(incarnation_file, new_goal_state.incarnation)
 
                     update_host_plugin()
-
                 return
 
             except IOError as e:
@@ -1033,8 +1034,9 @@ class WireClient(object):
         try:
             ret = host_func()
         except (ResourceGoneError, InvalidContainerError) as e:
-            old_container_id = self.host_plugin.container_id
-            old_role_config_name = self.host_plugin.role_config_name
+            host_plugin = self.get_host_plugin()
+            old_container_id = host_plugin.container_id
+            old_role_config_name = host_plugin.role_config_name
 
             msg = "[PERIODIC] Request failed with the current host plugin configuration. " \
                   "ContainerId: {0}, role config file: {1}. Fetching new goal state and retrying the call." \
@@ -1043,8 +1045,8 @@ class WireClient(object):
 
             self.update_host_plugin_from_goal_state()
 
-            new_container_id = self.host_plugin.container_id
-            new_role_config_name = self.host_plugin.role_config_name
+            new_container_id = host_plugin.container_id
+            new_role_config_name = host_plugin.role_config_name
             msg = "[PERIODIC] Host plugin reconfigured with new parameters. " \
                   "ContainerId: {0}, role config file: {1}.".format(new_container_id, new_role_config_name)
             logger.periodic_info(logger.EVERY_SIX_HOURS, msg)
diff --git a/azurelinuxagent/common/singletonperthread.py b/azurelinuxagent/common/singletonperthread.py
new file mode 100644
index 00000000..0d913901
--- /dev/null
+++ b/azurelinuxagent/common/singletonperthread.py
@@ -0,0 +1,30 @@
+from threading import Lock, currentThread
+
+
+class _SingletonPerThreadMetaClass(type):
+    """ A metaclass that creates a SingletonPerThread base class when called. """
+    _instances = {}
+    _lock = Lock()
+
+    def __call__(cls, *args, **kwargs):
+        with cls._lock:
+            obj_name = "%s__%s" % (cls.__name__, currentThread().getName())  # Object Name = className__threadName
+            if obj_name not in cls._instances:
+                cls._instances[obj_name] = super(_SingletonPerThreadMetaClass, cls).__call__(*args, **kwargs)
+            return cls._instances[obj_name]
+
+
+class SingletonPerThread(_SingletonPerThreadMetaClass('SingleObjectPerThreadMetaClass', (object,), {})):
+    # This base class calls the metaclass above to create the singleton per thread object. This class provides an
+    # abstraction over how to invoke the Metaclass so just inheriting this class makes the
+    # child class a singleton per thread (As opposed to invoking the Metaclass separately for each derived classes)
+    # More info here - https://stackoverflow.com/questions/6760685/creating-a-singleton-in-python
+    #
+    # Usage:
+    # Inheriting this class will create a Singleton per thread for that class
+    # To delete the cached object of a class, call DerivedClassName.clear() to delete the object per thread
+    # Note: If the thread dies and is recreated with the same thread name, the existing object would be reused
+    # and no new object for the derived class would be created unless DerivedClassName.clear() is called explicitly to
+    # delete the cache
+    pass
+
diff --git a/azurelinuxagent/ga/monitor.py b/azurelinuxagent/ga/monitor.py
index b48f318d..77ed176e 100644
--- a/azurelinuxagent/ga/monitor.py
+++ b/azurelinuxagent/ga/monitor.py
@@ -277,6 +277,8 @@ class MonitorHandler(object):
                         MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD,
                         MonitorHandler.IMDS_HEARTBEAT_PERIOD).seconds
         while self.should_run:
+            # Updating the goal_state periodically for the monitor thread to keep the WireClient object upto-date always
+            self.protocol.update_goal_state()
             self.send_telemetry_heartbeat()
             self.poll_telemetry_metrics()
             self.send_telemetry_metrics()   # This will be removed in favor of poll_telemetry_metrics() and it'll directly send the perf data for each cgroup.


Test patch:
diff --git a/tests/common/test_singletonperthread.py b/tests/common/test_singletonperthread.py
new file mode 100644
index 00000000..f2371a3d
--- /dev/null
+++ b/tests/common/test_singletonperthread.py
@@ -0,0 +1,164 @@
+import uuid
+from multiprocessing import Queue
+from threading import Thread, currentThread
+
+from azurelinuxagent.common.singletonperthread import SingletonPerThread
+from tests.tools import AgentTestCase, clear_singleton_instances
+
+
+class TestClassToTestSingletonPerThread(SingletonPerThread):
+    """
+    Since these tests deal with testing in a multithreaded environment,
+    we employ the use of multiprocessing.Queue() to ensure that the data is consistent.
+
+     This test class uses a uuid to identify an object instead of directly using object reference because
+    Queue.get() returns a different object reference than what is put in it even though the object is same
+    (which is verified using uuid in this test class)
+
+    Eg:
+
+        obj1 = WireClient("obj1")
+        obj1
+        <__main__.WireClient object at 0x7f5e78476198>
+        q = Queue()
+        q.put(obj1)
+        test1 = q.get()
+        test1
+        <__main__.WireClient object at 0x7f5e78430630>
+
+        test1.endpoint == obj1.endpoint
+        True
+    """
+
+    def __init__(self):
+        # Set the name of the object to the current thread name
+        self.name = currentThread().getName()
+        # Unique identifier for a class object
+        self.uuid = str(uuid.uuid4())
+
+
+class TestSingletonPerThread(AgentTestCase):
+
+    THREAD_NAME_1 = 'thread-1'
+    THREAD_NAME_2 = 'thread-2'
+
+    def setUp(self):
+        super(TestSingletonPerThread, self).setUp()
+        # In a multi-threaded environment, exceptions thrown in the child thread will not be propagated to the parent
+        # thread. In order to achieve that, adding all exceptions to a Queue and then checking that in parent thread.
+        self.errors = Queue()
+        clear_singleton_instances(TestClassToTestSingletonPerThread)
+
+    def _setup_multithread_and_execute(self, func1, args1, func2, args2, t1_name=None, t2_name=None):
+
+        t1 = Thread(target=func1, args=args1)
+        t2 = Thread(target=func2, args=args2)
+        t1.setName(t1_name if t1_name else self.THREAD_NAME_1)
+        t2.setName(t2_name if t2_name else self.THREAD_NAME_2)
+        t1.start()
+        t2.start()
+        t1.join()
+        t2.join()
+
+        errs = []
+        while not self.errors.empty():
+            errs.append(self.errors.get())
+        if len(errs) > 0:
+            raise Exception("Errors: %s" % ' , '.join(errs))
+
+    @staticmethod
+    def _get_test_class_instance(q, err):
+        try:
+            obj = TestClassToTestSingletonPerThread()
+            q.put(obj)
+        except Exception as e:
+            err.put(str(e))
+
+    def _parse_instances_and_return_thread_objects(self, instances, t1_name=None, t2_name=None):
+        obj1, obj2 = instances.get(), instances.get()
+
+        def check_obj(name):
+            if obj1.name == name:
+                return obj1
+            elif obj2.name == name:
+                return obj2
+            else:
+                return None
+
+        t1_object = check_obj(t1_name if t1_name else self.THREAD_NAME_1)
+        t2_object = check_obj(t2_name if t2_name else self.THREAD_NAME_2)
+
+        return t1_object, t2_object
+
+    def test_it_should_have_only_one_instance_for_same_thread(self):
+        obj1 = TestClassToTestSingletonPerThread()
+        obj2 = TestClassToTestSingletonPerThread()
+
+        self.assertEqual(obj1.uuid, obj2.uuid)
+
+    def test_it_should_have_multiple_instances_for_multiple_threads(self):
+        instances = Queue()
+        self._setup_multithread_and_execute(func1=self._get_test_class_instance,
+                                            args1=(instances, self.errors),
+                                            func2=self._get_test_class_instance,
+                                            args2=(instances, self.errors))
+
+        self.assertEqual(2, instances.qsize())  # Assert that there are 2 objects in the queue
+        obj1, obj2 = instances.get(), instances.get()
+        self.assertNotEqual(obj1.uuid, obj2.uuid)
+
+    def test_it_should_return_existing_instance_for_new_thread_with_same_name(self):
+
+        instances = Queue()
+        self._setup_multithread_and_execute(func1=self._get_test_class_instance,
+                                            args1=(instances, self.errors),
+                                            func2=self._get_test_class_instance,
+                                            args2=(instances, self.errors))
+
+        t1_obj, t2_obj = self._parse_instances_and_return_thread_objects(instances)
+
+        new_instances = Queue()
+        # The 2nd call is to get new objects with the same thread name to verify if the objects are same
+        self._setup_multithread_and_execute(func1=self._get_test_class_instance,
+                                            args1=(new_instances, self.errors),
+                                            func2=self._get_test_class_instance,
+                                            args2=(new_instances, self.errors))
+
+        new_t1_obj, new_t2_obj = self._parse_instances_and_return_thread_objects(new_instances)
+
+        self.assertEqual(t1_obj.name, new_t1_obj.name)
+        self.assertEqual(t1_obj.uuid, new_t1_obj.uuid)
+        self.assertEqual(t2_obj.name, new_t2_obj.name)
+        self.assertEqual(t2_obj.uuid, new_t2_obj.uuid)
+
+    def test_singleton_object_should_match_thread_name(self):
+
+        instances = Queue()
+        t1_name = str(uuid.uuid4())
+        t2_name = str(uuid.uuid4())
+
+        test_class_obj_name = lambda t_name: "%s__%s" % (TestClassToTestSingletonPerThread.__name__, t_name)
+
+        self._setup_multithread_and_execute(func1=self._get_test_class_instance,
+                                            args1=(instances, self.errors),
+                                            func2=self._get_test_class_instance,
+                                            args2=(instances, self.errors),
+                                            t1_name=t1_name,
+                                            t2_name=t2_name)
+
+        singleton_instances = TestClassToTestSingletonPerThread._instances
+
+        # Assert instance names are consistent with the thread names
+        self.assertIn(test_class_obj_name(t1_name), singleton_instances)
+        self.assertIn(test_class_obj_name(t2_name), singleton_instances)
+
+        # Assert that the objects match their respective threads
+        # This function matches objects with their thread names and returns the respective object or None if not found
+        t1_obj, t2_obj = self._parse_instances_and_return_thread_objects(instances, t1_name, t2_name)
+        # Ensure that objects for both the threads were found
+        self.assertIsNotNone(t1_obj)
+        self.assertIsNotNone(t2_obj)
+        # Ensure that the objects match with their respective thread objects
+        self.assertEqual(singleton_instances[test_class_obj_name(t1_name)].uuid, t1_obj.uuid)
+        self.assertEqual(singleton_instances[test_class_obj_name(t2_name)].uuid, t2_obj.uuid)
+
diff --git a/tests/ga/test_exthandlers.py b/tests/ga/test_exthandlers.py
index 99a43735..fba789dc 100644
--- a/tests/ga/test_exthandlers.py
+++ b/tests/ga/test_exthandlers.py
@@ -1,24 +1,33 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the Apache License.
 import json
-import subprocess
 import os
+import subprocess
 import time
 
-from mock import MagicMock
+from azurelinuxagent.common.protocol.util import ProtocolUtil
 
+from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
+from azurelinuxagent.common.event import WALAEventOperation
+from azurelinuxagent.common.exception import ProtocolError, ExtensionError, ExtensionErrorCodes
 from azurelinuxagent.common.protocol.restapi import ExtensionStatus, Extension, ExtHandler, ExtHandlerProperties
+from azurelinuxagent.common.utils.extensionprocessutil import TELEMETRY_MESSAGE_MAX_LEN, format_stdout_stderr, \
+    read_output
+from mock import MagicMock
 from azurelinuxagent.common.protocol.wire import WireProtocol
 from azurelinuxagent.ga.exthandlers import parse_ext_status, ExtHandlerInstance, get_exthandlers_handler, \
     ExtCommandEnvVariable
-from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
-from azurelinuxagent.common.exception import ProtocolError, ExtensionError, ExtensionErrorCodes
-from azurelinuxagent.common.event import WALAEventOperation
-from azurelinuxagent.common.utils.extensionprocessutil import TELEMETRY_MESSAGE_MAX_LEN, format_stdout_stderr, read_output
-from tests.tools import AgentTestCase, patch, mock_sleep
+from tests.tools import AgentTestCase, patch, mock_sleep, clear_singleton_instances
 
 
 class TestExtHandlers(AgentTestCase):
+
+    def setUp(self):
+        super(TestExtHandlers, self).setUp()
+        # Since ProtocolUtil is a singleton per thread, we need to clear it to ensure that the test cases do not
+        # reuse a previous state
+        clear_singleton_instances(ProtocolUtil)
+
     def test_parse_extension_status00(self):
         """
         Parse a status report for a successful execution of an extension.
diff --git a/tests/ga/test_monitor.py b/tests/ga/test_monitor.py
index bcc2a78f..dcd3f4b2 100644
--- a/tests/ga/test_monitor.py
+++ b/tests/ga/test_monitor.py
@@ -26,6 +26,7 @@ import tempfile
 import time
 from datetime import timedelta
 
+from azurelinuxagent.common.protocol.util import ProtocolUtil
 from nose.plugins.attrib import attr
 
 import azurelinuxagent.common.conf as conf
@@ -52,7 +53,7 @@ from azurelinuxagent.ga.monitor import generate_extension_metrics_telemetry_dict
 from tests.common.test_cgroupstelemetry import make_new_cgroup
 from tests.protocol.mockwiredata import DATA_FILE, WireProtocolData
 from tests.tools import Mock, MagicMock, patch, load_data, AgentTestCase, data_dir, are_cgroups_enabled, \
-    i_am_root, skip_if_predicate_false, is_trusty_in_travis, skip_if_predicate_true
+    i_am_root, skip_if_predicate_false, is_trusty_in_travis, skip_if_predicate_true, clear_singleton_instances
 
 
 class ResponseMock(Mock):
@@ -117,6 +118,9 @@ class TestMonitor(AgentTestCase):
         AgentTestCase.setUp(self)
         prefix = "UnitTest"
         logger.DEFAULT_LOGGER = Logger(prefix=prefix)
+        # Since ProtocolUtil is a singleton per thread, we need to clear it to ensure that the test cases do not
+        # reuse a previous state
+        clear_singleton_instances(ProtocolUtil)
 
     def tearDown(self):
         AgentTestCase.tearDown(self)
@@ -342,18 +346,19 @@ class TestMonitor(AgentTestCase):
         self.assertEqual(0, patch_send_telemetry_metrics.call_count)
         self.assertEqual(0, patch_poll_telemetry_metrics.call_count)
 
-        monitor_handler.start()
-        time.sleep(1)
-        self.assertTrue(monitor_handler.is_alive())
+        with patch.object(monitor_handler, 'protocol'):
+            monitor_handler.start()
+            time.sleep(1)
+            self.assertTrue(monitor_handler.is_alive())
 
-        self.assertNotEqual(0, patch_hostplugin_heartbeat.call_count)
-        self.assertNotEqual(0, patch_send_events.call_count)
-        self.assertNotEqual(0, patch_telemetry_heartbeat.call_count)
-        self.assertNotEqual(0, patch_imds_heartbeat.call_count)
-        self.assertNotEqual(0, patch_send_telemetry_metrics.call_count)
-        self.assertNotEqual(0, patch_poll_telemetry_metrics.call_count)
+            self.assertNotEqual(0, patch_hostplugin_heartbeat.call_count)
+            self.assertNotEqual(0, patch_send_events.call_count)
+            self.assertNotEqual(0, patch_telemetry_heartbeat.call_count)
+            self.assertNotEqual(0, patch_imds_heartbeat.call_count)
+            self.assertNotEqual(0, patch_send_telemetry_metrics.call_count)
+            self.assertNotEqual(0, patch_poll_telemetry_metrics.call_count)
 
-        monitor_handler.stop()
+            monitor_handler.stop()
 
     @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_telemetry_metrics")
     @patch("azurelinuxagent.ga.monitor.MonitorHandler.poll_telemetry_metrics")
@@ -370,28 +375,29 @@ class TestMonitor(AgentTestCase):
         self.assertEqual(None, monitor_handler.last_telemetry_heartbeat)
         self.assertEqual(None, monitor_handler.last_imds_heartbeat)
 
-        monitor_handler.start()
-        time.sleep(0.2)
-        self.assertTrue(monitor_handler.is_alive())
+        with patch.object(monitor_handler, 'protocol'):
+            monitor_handler.start()
+            time.sleep(0.2)
+            self.assertTrue(monitor_handler.is_alive())
 
-        self.assertNotEqual(None, monitor_handler.last_host_plugin_heartbeat)
-        self.assertNotEqual(None, monitor_handler.last_event_collection)
-        self.assertNotEqual(None, monitor_handler.last_telemetry_heartbeat)
-        self.assertNotEqual(None, monitor_handler.last_imds_heartbeat)
+            self.assertNotEqual(None, monitor_handler.last_host_plugin_heartbeat)
+            self.assertNotEqual(None, monitor_handler.last_event_collection)
+            self.assertNotEqual(None, monitor_handler.last_telemetry_heartbeat)
+            self.assertNotEqual(None, monitor_handler.last_imds_heartbeat)
 
-        heartbeat_hostplugin = monitor_handler.last_host_plugin_heartbeat
-        heartbeat_imds = monitor_handler.last_imds_heartbeat
-        heartbeat_telemetry = monitor_handler.last_telemetry_heartbeat
-        events_collection = monitor_handler.last_event_collection
+            heartbeat_hostplugin = monitor_handler.last_host_plugin_heartbeat
+            heartbeat_imds = monitor_handler.last_imds_heartbeat
+            heartbeat_telemetry = monitor_handler.last_telemetry_heartbeat
+            events_collection = monitor_handler.last_event_collection
 
-        time.sleep(0.5)
+            time.sleep(0.5)
 
-        self.assertNotEqual(heartbeat_imds, monitor_handler.last_imds_heartbeat)
-        self.assertNotEqual(heartbeat_hostplugin, monitor_handler.last_host_plugin_heartbeat)
-        self.assertNotEqual(events_collection, monitor_handler.last_event_collection)
-        self.assertNotEqual(heartbeat_telemetry, monitor_handler.last_telemetry_heartbeat)
+            self.assertNotEqual(heartbeat_imds, monitor_handler.last_imds_heartbeat)
+            self.assertNotEqual(heartbeat_hostplugin, monitor_handler.last_host_plugin_heartbeat)
+            self.assertNotEqual(events_collection, monitor_handler.last_event_collection)
+            self.assertNotEqual(heartbeat_telemetry, monitor_handler.last_telemetry_heartbeat)
 
-        monitor_handler.stop()
+            monitor_handler.stop()
 
     @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_telemetry_metrics")
     @patch("azurelinuxagent.ga.monitor.MonitorHandler.poll_telemetry_metrics")
@@ -408,28 +414,29 @@ class TestMonitor(AgentTestCase):
         self.assertEqual(None, monitor_handler.last_telemetry_heartbeat)
         self.assertEqual(None, monitor_handler.last_imds_heartbeat)
 
-        monitor_handler.start()
-        time.sleep(0.2)
-        self.assertTrue(monitor_handler.is_alive())
+        with patch.object(monitor_handler, 'protocol'):
+            monitor_handler.start()
+            time.sleep(0.2)
+            self.assertTrue(monitor_handler.is_alive())
 
-        self.assertNotEqual(None, monitor_handler.last_host_plugin_heartbeat)
-        self.assertNotEqual(None, monitor_handler.last_event_collection)
-        self.assertNotEqual(None, monitor_handler.last_telemetry_heartbeat)
-        self.assertNotEqual(None, monitor_handler.last_imds_heartbeat)
+            self.assertNotEqual(None, monitor_handler.last_host_plugin_heartbeat)
+            self.assertNotEqual(None, monitor_handler.last_event_collection)
+            self.assertNotEqual(None, monitor_handler.last_telemetry_heartbeat)
+            self.assertNotEqual(None, monitor_handler.last_imds_heartbeat)
 
-        heartbeat_hostplugin = monitor_handler.last_host_plugin_heartbeat
-        heartbeat_imds = monitor_handler.last_imds_heartbeat
-        heartbeat_telemetry = monitor_handler.last_telemetry_heartbeat
-        events_collection = monitor_handler.last_event_collection
+            heartbeat_hostplugin = monitor_handler.last_host_plugin_heartbeat
+            heartbeat_imds = monitor_handler.last_imds_heartbeat
+            heartbeat_telemetry = monitor_handler.last_telemetry_heartbeat
+            events_collection = monitor_handler.last_event_collection
 
-        time.sleep(0.5)
+            time.sleep(0.5)
 
-        self.assertEqual(heartbeat_hostplugin, monitor_handler.last_host_plugin_heartbeat)
-        self.assertEqual(heartbeat_imds, monitor_handler.last_imds_heartbeat)
-        self.assertEqual(events_collection, monitor_handler.last_event_collection)
-        self.assertEqual(heartbeat_telemetry, monitor_handler.last_telemetry_heartbeat)
+            self.assertEqual(heartbeat_hostplugin, monitor_handler.last_host_plugin_heartbeat)
+            self.assertEqual(heartbeat_imds, monitor_handler.last_imds_heartbeat)
+            self.assertEqual(events_collection, monitor_handler.last_event_collection)
+            self.assertEqual(heartbeat_telemetry, monitor_handler.last_telemetry_heartbeat)
 
-        monitor_handler.stop()
+            monitor_handler.stop()
 
     @patch("azurelinuxagent.common.protocol.healthservice.HealthService.report_host_plugin_heartbeat")
     def test_heartbeat_creates_signal(self, patch_report_heartbeat, *args):
diff --git a/tests/ga/test_remoteaccess_handler.py b/tests/ga/test_remoteaccess_handler.py
index a54e4d54..329a2c80 100644
--- a/tests/ga/test_remoteaccess_handler.py
+++ b/tests/ga/test_remoteaccess_handler.py
@@ -17,12 +17,13 @@
 
 from datetime import timedelta
 
+from azurelinuxagent.common.protocol.util import ProtocolUtil
+
 from azurelinuxagent.common.exception import RemoteAccessError
 from azurelinuxagent.common.protocol.wire import *
 from azurelinuxagent.ga.remoteaccess import RemoteAccessHandler
 from tests.common.osutil.mock_osutil import MockOSUtil
-from tests.tools import AgentTestCase, load_data, patch
-
+from tests.tools import AgentTestCase, load_data, patch, clear_singleton_instances
 
 info_messages = []
 error_messages = []
@@ -52,6 +53,9 @@ class TestRemoteAccessHandler(AgentTestCase):
 
     def setUp(self):
         super(TestRemoteAccessHandler, self).setUp()
+        # Since ProtocolUtil is a singleton per thread, we need to clear it to ensure that the test cases do not
+        # reuse a previous state
+        clear_singleton_instances(ProtocolUtil)
         del info_messages[:]
         del error_messages[:]
         for data in TestRemoteAccessHandler.eventing_data:
diff --git a/tests/protocol/test_protocol_util.py b/tests/protocol/test_protocol_util.py
index 6f341f10..7ab38b5f 100644
--- a/tests/protocol/test_protocol_util.py
+++ b/tests/protocol/test_protocol_util.py
@@ -17,18 +17,57 @@
 
 import os
 import unittest
+from multiprocessing import Queue
+from threading import Thread
 
-from tests.tools import AgentTestCase, MagicMock, Mock, patch
+from tests.tools import AgentTestCase, MagicMock, Mock, patch, clear_singleton_instances
 from azurelinuxagent.common.exception import *
 from azurelinuxagent.common.protocol import get_protocol_util, \
                                             TAG_FILE_NAME
 from azurelinuxagent.common.utils.restutil import KNOWN_WIRESERVER_IP
-from azurelinuxagent.common.protocol.util import ENDPOINT_FILE_NAME
+from azurelinuxagent.common.protocol.util import ProtocolUtil
 from errno import ENOENT
 
 
 @patch("time.sleep")
 class TestProtocolUtil(AgentTestCase):
+
+    def setUp(self):
+        super(TestProtocolUtil, self).setUp()
+        # Since ProtocolUtil is a singleton per thread, we need to clear it to ensure that the test cases do not
+        # reuse a previous state
+        clear_singleton_instances(ProtocolUtil)
+
+    def test_get_protocol_util_should_return_same_object_for_same_thread(self, _):
+        protocol_util1 = get_protocol_util()
+        protocol_util2 = get_protocol_util()
+
+        self.assertEqual(protocol_util1, protocol_util2)
+
+    def test_get_protocol_util_should_return_different_object_for_different_thread(self, _):
+        def get_util_obj(q, err):
+            try:
+                q.put(get_protocol_util())
+            except Exception as e:
+                err.put(str(e))
+
+        queue = Queue()
+        errors = Queue()
+        t1 = Thread(target=get_util_obj, args=(queue, errors))
+        t2 = Thread(target=get_util_obj, args=(queue, errors))
+        t1.start()
+        t2.start()
+        t1.join()
+        t2.join()
+
+        errs = []
+        while not errors.empty():
+            errs.append(errors.get())
+        if len(errs) > 0:
+            raise Exception("Unable to fetch protocol_util. Errors: %s" % ' , '.join(errs))
+
+        self.assertEqual(2, queue.qsize())  # Assert that there are 2 objects in the queue
+        self.assertNotEqual(queue.get(), queue.get())
     
     @patch("azurelinuxagent.common.protocol.util.MetadataProtocol")
     @patch("azurelinuxagent.common.protocol.util.WireProtocol")
diff --git a/tests/protocol/test_wire.py b/tests/protocol/test_wire.py
index 19ae595f..1f634b39 100644
--- a/tests/protocol/test_wire.py
+++ b/tests/protocol/test_wire.py
@@ -21,8 +21,8 @@ import os
 import stat
 import tempfile
 import time
-import uuid
 import unittest
+import uuid
 import zipfile
 
 from azurelinuxagent.common.exception import InvalidContainerError, ResourceGoneError, ProtocolError, \
@@ -30,7 +30,7 @@ from azurelinuxagent.common.exception import InvalidContainerError, ResourceGone
 from azurelinuxagent.common.future import httpclient
 from azurelinuxagent.common.protocol.hostplugin import HostPluginProtocol
 from azurelinuxagent.common.protocol.wire import WireProtocol, WireClient, GoalState, ExtensionsConfig, \
-    InVMArtifactsProfile, VMAgentManifestUri, StatusBlob, VMStatus, INCARNATION_FILE_NAME, ExtHandlerVersionUri, \
+    InVMArtifactsProfile, VMAgentManifestUri, StatusBlob, VMStatus, ExtHandlerVersionUri, \
     DataContractList, socket
 from azurelinuxagent.common.telemetryevent import TelemetryEvent, TelemetryEventParam, TelemetryEventList
 from azurelinuxagent.common.utils import fileutil, restutil
diff --git a/tests/tools.py b/tests/tools.py
index 09ef0c4b..71d85613 100644
--- a/tests/tools.py
+++ b/tests/tools.py
@@ -26,15 +26,15 @@ import shutil
 import stat
 import sys
 import tempfile
+import time
 import unittest
 from functools import wraps
+from threading import currentThread
 
-import time
-
-from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
-import azurelinuxagent.common.event as event
 import azurelinuxagent.common.conf as conf
+import azurelinuxagent.common.event as event
 import azurelinuxagent.common.logger as logger
+from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
 from azurelinuxagent.common.osutil.factory import _get_osutil
 from azurelinuxagent.common.osutil.ubuntu import Ubuntu14OSUtil, Ubuntu16OSUtil
 from azurelinuxagent.common.utils import fileutil
@@ -544,3 +544,10 @@ def distros(distro_name=".*", distro_version=".*", distro_full_name=".*"):
     return decorator
 
 
+def clear_singleton_instances(cls):
+    # Adding this lock to avoid any race conditions
+    with cls._lock:
+        obj_name = "%s__%s" % (cls.__name__, currentThread().getName())  # Object Name = className__threadName
+        if obj_name in cls._instances:
+            del cls._instances[obj_name]
+