Merge ~tamilmani1989/cloud-init:azure_networking into cloud-init:master

[Tamilmani Manoharan]

azure: detect vnet migration via netlink media change event

Replace Azure pre-provision polling on IMDS with a blocking call
which watches for netlink link state change messages. The media
change event happens when a pre-provisioned VM has been activated
and is connected to the users virtual network and cloud-init can
then resume operation to complete image instantiation.

This PR enables moving a virtual machine (VM) in Azure from one virtual network (Vnet) to another. It adds a new azure specific module that listens for Azure host-node to indicate that VM has now moved to a different network (by listening for nic connect/disconnect). Once VM move is detected (via netlink callback), the module triggers a re-dhcp to get ipaddress from the new Vnet.

[Robert Schweikert (rjschwei)]

The other question that's hidden in here is whether we should "just admit that we need dbus" and start properly depending on it. We already depend on dbus for "set_hostname". Problem is that we "don't really want to depend on dbus" and that creates other problems, i.e. hostname setting only works if it runs after dbus is already operational (for systemd based distributions). In this case the waiting on the socket could be replaces by using dbus notifications.

[Douglas Jordan (dojordan)]

Some comments inline

[Ryan Harper (raharper)]

Can you describe in what scenario the current code which polls the IMDS url for failures is inadequate?

Is there a way we can test the current code on Azure directly?

review: Needs Fixing

[Sushant Sharma (sushantsharma)]

> Can you describe in what scenario the current code which polls the IMDS url
> for failures is inadequate?
>
> Is there a way we can test the current code on Azure directly?

Thanks Ryan, This is a valid question, and we are in process of testing in Azure with all components involved (including updated host). Having said that, IMDS poll can fail for multiple reasons, e.g., IMDS in the middle of getting updated on the host, or any random communication failure (since this is a network call).
I think relying on a clean deterministic event should be preferred.
Furthermore, I believe the code that polls host for new provisioning data should just do that, and not worry about setting up networking. We should aim for clear and concise goals for every piece of code.

[Ryan Harper (raharper)]

> > Can you describe in what scenario the current code which polls the IMDS url
> > for failures is inadequate?
> >
> > Is there a way we can test the current code on Azure directly?
>
> Thanks Ryan, This is a valid question, and we are in process of testing in
> Azure with all components involved (including updated host). Having said that,
> IMDS poll can fail for multiple reasons, e.g., IMDS in the middle of getting
> updated on the host, or any random communication failure (since this is a
> network call).

Which of these failures are not handled already by the retry logic? If we have any failure attempting to fetch the URL from IMDS, then then the retry logic will acquire a new lease; so I'm struggling to see which scenario is not covered.

> I think relying on a clean deterministic event should be preferred.

I agree; I'd much rather see use of the HyperV Keystore here for indicating both that a VM has been tagged as pre-provision, and when its network has been moved.

> Furthermore, I believe the code that polls host for new provisioning data
> should just do that, and not worry about setting up networking.

The _poll_imds is not setting up networking for the system; rather it's setting up a temporary interface and DHCP'ing on to acquire information at a specific time for a specific use-case: preprovisioning.

[Sushant Sharma (sushantsharma)]

> > > Can you describe in what scenario the current code which polls the IMDS
> url
> > > for failures is inadequate?
> > >
> > > Is there a way we can test the current code on Azure directly?
> >
> > Thanks Ryan, This is a valid question, and we are in process of testing in
> > Azure with all components involved (including updated host). Having said
> that,
> > IMDS poll can fail for multiple reasons, e.g., IMDS in the middle of getting
> > updated on the host, or any random communication failure (since this is a
> > network call).
>
> Which of these failures are not handled already by the retry logic? If we
> have any failure attempting to fetch the URL from IMDS, then then the retry
> logic will acquire a new lease; so I'm struggling to see which scenario is not
> covered.
>

Let's remember that time to move VM in the new network is also critical for our customers. If we can save any amount of time (even if it is few seconds on average) that we spend on polling and retrying dhcp on failure, we should prefer that. Having said that, let us run some tests and see how much we are saving in the solution that depends on netlink versus the existing checked in code. Will report the results and hopefully we can make the case again for the current PR.

Thanks,
Sushant

[Douglas Jordan (dojordan)]

Added a few (mostly style) comments

[Tamilmani Manoharan (tamilmani1989)]

I addressed review comments

[Paul Meyer (paul-meyer)]

Overall LGTM, except for the lack of unit tests for the new code.

You could change the 0/-1 return code from create_bind_netlink_socket to be either an object property (netevent.is_listening) or throwing an exception with handling in the calling code.

review: Needs Fixing

[Tamilmani Manoharan (tamilmani1989)]

Hi Paul, I added unit test for network switch. Can you please review?

[Chad Smith (chad.smith)]

Updated my comment and patch suggestions, please feel free to disagree where you think points are irrelevant or wrong

  Here are a couple thoughts on your branch to aid in testability of the branch features:

The complete patch suggestion is here http://paste.ubuntu.com/p/pDYvtFQN8g/

I've itemized the general thoughts here for your review to see what you think
  1. generally let's limit what lives in a try/except block to code that we expect will actually generate exceptions. I moved general increment logic outside of the struct.unpack calls and moved the try/except socket.error checks to around the read_netlink_socket as I didn't think socket.errors would be raised by anything else in wait_for_media_disconnect_connect

  2. I suggest breaking up NetworkEvents class as it seems just a collection of functions more than a class with state. When I see one-line class methods that don't actually depend on any instance attributes, it's a hint that it might be an empty level of indirection we just don't need (like select_netlink_socket, read_socket and close_socket methods).

 3. renam3 networkevents.py -> netlink.py as a module to make it a bit more specific
 4 personal preference is to use namedtuples instead of classes to host simple structured attributes like your rta_attr definition
 5. I'd like to see simple unit tests on each function per the simple example I added in cloudinit/sources/helpers/tests/test_netlink.py

review: Needs Information

[Chad Smith (chad.smith)]

Also here's a cloudinit/sources/helpers/tests/test_netlink.py basic single test example

http://paste.ubuntu.com/p/SHTwRRZPD2/

[Server Team CI bot (server-team-bot)]

FAILED: Continuous integration, rev:f561d8f6a14a1136c99820fefdd78a3c65a405dd
https://jenkins.ubuntu.com/server/job/cloud-init-ci/429/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    FAILED: Ubuntu LTS: Build

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/429/rebuild

review: Needs Fixing (continuous-integration)

[Server Team CI bot (server-team-bot)]

PASSED: Continuous integration, rev:7712d6fa901ea6e132fd3bd7a526af566e5d3aa6
https://jenkins.ubuntu.com/server/job/cloud-init-ci/439/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    SUCCESS: Ubuntu LTS: Build
    SUCCESS: Ubuntu LTS: Integration
    IN_PROGRESS: Declarative: Post Actions

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/439/rebuild

review: Approve (continuous-integration)

[Joshua Powers (powersj)]

CI passed on latest commit, there are conflicts below that need to be worked out, moving it back to needs review at Tamilmani's request.

0c14e44... by Tamilmani Manoharan on 2018-11-13

Merge branch 'master' of https://git.launchpad.net/cloud-init into azure_networking

# Conflicts:
# cloudinit/sources/DataSourceAzure.py

144e6d8... by Tamilmani Manoharan on 2018-11-13

fixed indentation error

[Ryan Harper (raharper)]

Thanks for working on the refactor. This looks a lot better and it's much easier to read and write unittests.

I've quite a few inline comments, please look through those. The biggest query is around how do we know when the vnet migration has started to prevent us from getting stuck forever if it's already started before we attempt to listen for events.

review: Needs Fixing

[Tamilmani Manoharan (tamilmani1989)]

> Thanks for working on the refactor. This looks a lot better and it's much
> easier to read and write unittests.
>
> I've quite a few inline comments, please look through those. The biggest
> query is around how do we know when the vnet migration has started to prevent
> us from getting stuck forever if it's already started before we attempt to
> listen for events.

Thanks for reviewing it. Unless cloud-init report ready to azure fabric, vnet switch won't happen. Its way of informing the top level services that vm is ready for vnet switch. So I have created and bound netlink socket before that call so it won't miss any event that happens after that.

[Tamilmani Manoharan (tamilmani1989)]

I have few concern over some review comments which i replied inline. Other than that, I'm fine with other comments and will work on addressing those

[Ryan Harper (raharper)]

> > Thanks for working on the refactor. This looks a lot better and it's much
> > easier to read and write unittests.
> >
> > I've quite a few inline comments, please look through those. The biggest
> > query is around how do we know when the vnet migration has started to
> prevent
> > us from getting stuck forever if it's already started before we attempt to
> > listen for events.
>
> Thanks for reviewing it. Unless cloud-init report ready to azure fabric, vnet
> switch won't happen. Its way of informing the top level services that vm is
> ready for vnet switch. So I have created and bound netlink socket before that
> call so it won't miss any event that happens after that.

OK, so, the sequence then is:

1) crawl initial metadata and detect if pre-provision VM, if so write a REPROVISION marker file
2) in _poll_imds create nl socket
3) write a REPORT_READY marker file and call _report_ready()
4) _report_ready() will invoke the shim
5) wait_for_media_disconnect()

The window is now between (2) and (5); and the question is what does the netlink socket do with the incoming messages that we won't process until (5)?

From reading the upstream docs on netlink, the default netlink socket size is 32kb, and while the typical messages may be small, the payload for messages is arbitrary. I know that DELLINK and NEWLINK are small but other messages may come through and if we're not processing then the socket could become full.

I suspect that in practice there isn't a lot of netlink traffic on the node with only a single interface up at this point but I'm wondering if we shouldn't employ something that allows us to:

a) spawn a thread to start doing the "Wait" which would start reading messages as they come in on the socket *before* we send the READY flag to the fabric
b) after sending the ready flag, we would join() the thread which would require the main program to wait until the thread exited (ie, we detected the media change event).

Thoughts?

[Tamilmani Manoharan (tamilmani1989)]

> > > Thanks for working on the refactor. This looks a lot better and it's much
> > > easier to read and write unittests.
> > >
> > > I've quite a few inline comments, please look through those. The biggest
> > > query is around how do we know when the vnet migration has started to
> > prevent
> > > us from getting stuck forever if it's already started before we attempt to
> > > listen for events.
> >
> > Thanks for reviewing it. Unless cloud-init report ready to azure fabric,
> vnet
> > switch won't happen. Its way of informing the top level services that vm is
> > ready for vnet switch. So I have created and bound netlink socket before
> that
> > call so it won't miss any event that happens after that.
>
> OK, so, the sequence then is:
>
> 1) crawl initial metadata and detect if pre-provision VM, if so write a
> REPROVISION marker file
> 2) in _poll_imds create nl socket
> 3) write a REPORT_READY marker file and call _report_ready()
> 4) _report_ready() will invoke the shim
> 5) wait_for_media_disconnect()
>
> The window is now between (2) and (5); and the question is what does the
> netlink socket do with the incoming messages that we won't process until (5)?
>
> From reading the upstream docs on netlink, the default netlink socket size is
> 32kb, and while the typical messages may be small, the payload for messages is
> arbitrary. I know that DELLINK and NEWLINK are small but other messages may
> come through and if we're not processing then the socket could become full.
>
> I suspect that in practice there isn't a lot of netlink traffic on the node
> with only a single interface up at this point but I'm wondering if we
> shouldn't employ something that allows us to:
>
> a) spawn a thread to start doing the "Wait" which would start reading messages
> as they come in on the socket *before* we send the READY flag to the fabric
> b) after sending the ready flag, we would join() the thread which would
> require the main program to wait until the thread exited (ie, we detected the
> media change event).
>
> Thoughts?

I don't think we receive that many netlink messages since this happens before systemd-networkd starts. we manually setup an ephemeral network(one interface) to send report ready. Also netlink code listen for only RTMGRP_LINK(network interface create/delete/up/down events) and not for any ipaddress and route changes. There will be only one interface at that time and it reads only above events for that interface

[Ryan Harper (raharper)]

OK. Let's update the create_netlink_socket() method with some of these details, binding to only RTNLGRP_LINK (which only includes RTM_NEWLINK/RTM_DELLINK/RTN_GETLINK).

It would be nice if we could bind to those events and only on a specific ifindex (we could look up the currently up interface) but it's sufficient to filter in the wait_for_media_disconnect_connect().

[Tamilmani Manoharan (tamilmani1989)]

> OK. Let's update the create_netlink_socket() method with some of these
> details, binding to only RTNLGRP_LINK (which only includes
> RTM_NEWLINK/RTM_DELLINK/RTN_GETLINK).
>
> It would be nice if we could bind to those events and only on a specific
> ifindex (we could look up the currently up interface) but it's sufficient to
> filter in the wait_for_media_disconnect_connect().

ok will update it.

58ef420... by Tamilmani Manoharan on 2018-11-15

addressed review comments

3d3efee... by Tamilmani Manoharan on 2018-11-15

reverted back the line break change

148c88c... by Tamilmani Manoharan on 2018-11-15

added few more unit tests.

7e2bba6... by Tamilmani Manoharan on 2018-11-15

Added a test case to validate down up scenario

45806ec... by Tamilmani Manoharan on 2018-11-15

cleanup and retry on netlink socket creation error

cc64c51... by Tamilmani Manoharan on 2018-11-15

optimized test code

[Ryan Harper (raharper)]

I couple more in-line requests below. Thanks for incorporating the requested changes.

6e4cfb7... by Tamilmani Manoharan on 2018-11-16

addressed iteration 3 review comments

91a4304... by Tamilmani Manoharan on 2018-11-16

Merge branch 'master' of https://git.launchpad.net/cloud-init into azure_networking

# Conflicts:
# cloudinit/sources/DataSourceAzure.py
# tests/unittests/test_datasource/test_azure.py

feb9f10... by Tamilmani Manoharan on 2018-11-16

fixed azure UTs regression with recent changes in master

[Tamilmani Manoharan (tamilmani1989)]

> I couple more in-line requests below. Thanks for incorporating the requested
> changes.

Addressed your comments. Please take a look.

[Tamilmani Manoharan (tamilmani1989)]

thanks for reviewing. I will address your comments. Do you have any concerns other than these comments? Can you sign off once i address below comments?

59e5fe9... by Tamilmani Manoharan on 2018-11-17

addressed iteration 4 review comments

9c6eb4a... by Tamilmani Manoharan on 2018-11-17

added comments

b507379... by Tamilmani Manoharan on 2018-11-19

addressed iteration 5 review comments
Fixed code style error

[Server Team CI bot (server-team-bot)]

PASSED: Continuous integration, rev:b5073795740154066b791b2a9ab1f8a624a8bb3e
https://jenkins.ubuntu.com/server/job/cloud-init-ci/454/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    SUCCESS: Ubuntu LTS: Build
    SUCCESS: Ubuntu LTS: Integration
    IN_PROGRESS: Declarative: Post Actions

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/454/rebuild

review: Approve (continuous-integration)

[Chad Smith (chad.smith)]

Leaving this approved pending your comments regarding ignoring iterim state transitions for operstates != DOWN, DORMANT or UP.

review: Approve

2fc6e87... by Tamilmani Manoharan on 2018-11-21

addressed iteration 6 comments

75c8987... by Tamilmani Manoharan on 2018-11-21

dummy commit to sync with launchpad

[Tamilmani Manoharan (tamilmani1989)]

i addressed your comments

ad8ebf4... by Tamilmani Manoharan on 2018-11-21

moved log inside function as per review comment

[Chad Smith (chad.smith)]

Looks good, and unit tests pass. 2 Minor inline followups.
 Thanks Tamilmani.

[Ryan Harper (raharper)]

Please review the commit message I added.

I've a couple in-line comments on what we return when reading the attrs, please comment there as well.

review: Needs Information

f12813e... by Tamilmani Manoharan on 2018-11-26

fixed typo

[Ryan Harper (raharper)]

One more follow-up on the rta processing.

review: Needs Information

205864e... by Tamilmani Manoharan on 2018-11-26

return interfaceoperstate as none in case of failure

[Ryan Harper (raharper)]

Some comments on debug messages. And another occurred to me about the contents we get back from the socket, wondering if that is only ever a single message or possible a sequence of messages, and if so, don't we need to loop over the data?

review: Needs Information

[Tamilmani Manoharan (tamilmani1989)]

For netlink down/up events we won't get multipart message and i already tested it. Multipart messages will be received only if payload is too big.

dcf4c86... by Tamilmani Manoharan on 2018-11-27

Addressed reading multiple netlink messages in single receive.

978f7ed... by Tamilmani Manoharan on 2018-11-27

added test case for receiving multiple netlink messages in single receive call

d3fd194... by Tamilmani Manoharan on 2018-11-27

reverted changes in test_azure

fd355fc... by Tamilmani Manoharan on 2018-11-27

adhere to tox standard

[Ryan Harper (raharper)]

Thanks for handling multiple messages in the buffer. Some in-line comments, suggestions. I'd to add a test-case that ensure we parse the socket data buffer to completion and then a second read to the buffer to return the media change sequence so we validate that we handle entering and exiting both sets of loops.

[Scott Moser (smoser)]

1 small comment in line.
generally, i wonder what happens here on freebsd.

006bf63... by Tamilmani Manoharan on 2018-11-27

Added testcase for reading partial messages

[Tamilmani Manoharan (tamilmani1989)]

I responded for your comments.

437b25b... by Tamilmani Manoharan on 2018-11-27

modified logs

[Scott Moser (smoser)]

what about freebsd?
what happens there? do we hit your newly added code or is it avoided.

[Tamilmani Manoharan (tamilmani1989)]

> what about freebsd?
> what happens there? do we hit your newly added code or is it avoided.

Yes it will hit for freebsd also. Is there any problem with that?

85c5a6d... by Tamilmani Manoharan on 2018-11-27

1. Check for free bsd
2. Handle assertion errors

[Tamilmani Manoharan (tamilmani1989)]

Scott/Ryan, addressed your comments. Can you sign off?

c600588... by Tamilmani Manoharan on 2018-11-28

addressed reading partial messages from buffer issue and handled netlink socket close in case of exception

77dfb5f... by Tamilmani Manoharan on 2018-11-28

removed blank line at eof

8157985... by Tamilmani Manoharan on 2018-11-28

removed saving data to another variable

[Ryan Harper (raharper)]

Just a spelling fix in the unittests.

I think the core logic in wait_() is solid. Do you have a way to test this branch on an instances doing the pre-provision logic just to ensure we're not outsmarting ourselves with the additional parsing logic ?

[Tamilmani Manoharan (tamilmani1989)]

> Just a spelling fix in the unittests.
>
> I think the core logic in wait_() is solid. Do you have a way to test this
> branch on an instances doing the pre-provision logic just to ensure we're not
> outsmarting ourselves with the additional parsing logic ?

Yes i already tested that multiple times. I push image(with my changes) as vhd in azure marketplace and then deploy pps vm using that.

4d7dde8... by Tamilmani Manoharan on 2018-11-28

fixed spell error

[Ryan Harper (raharper)]

Thank you for the hard work and many changes.

review: Approve

ea309e3... by Tamilmani Manoharan on 2018-11-28

Fixed code style errors

[Sushant Sharma (sushantsharma)]

Thanks a lot Tamilmani!

review: Approve

[Scott Moser (smoser)]

review: Approve

[Server Team CI bot (server-team-bot)]

PASSED: Continuous integration, rev:ea309e371cb803edfe9f76a33261bfd02b4b5dda
https://jenkins.ubuntu.com/server/job/cloud-init-ci/467/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    SUCCESS: Ubuntu LTS: Build
    SUCCESS: Ubuntu LTS: Integration
    IN_PROGRESS: Declarative: Post Actions

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/467/rebuild

review: Approve (continuous-integration)

5917a00... by Tamilmani Manoharan on 2018-11-29

Merge branch 'master' of https://git.launchpad.net/cloud-init into azure_networking

9bc4a08... by Tamilmani Manoharan on 2018-11-29

There are 2 fixes:
1. if timeout happens, read returns none and concatenate data after that check
2. While resolving merge conflicts, removed else part accidentally. Restored the else condition.
3. Also reverted testcase changes

56559c2... by Tamilmani Manoharan on 2018-11-29

Fixed code style error

[Server Team CI bot (server-team-bot)]

FAILED: Continuous integration, rev:9bc4a08cbfb3cddc44e640d5543aaeef7fb186d5
https://jenkins.ubuntu.com/server/job/cloud-init-ci/469/
Executed test runs:
    SUCCESS: Checkout
    FAILED: Unit & Style Tests

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/469/rebuild

review: Needs Fixing (continuous-integration)

[Server Team CI bot (server-team-bot)]

PASSED: Continuous integration, rev:56559c2548f858046a0775d21190cb167463cdcb
https://jenkins.ubuntu.com/server/job/cloud-init-ci/470/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    SUCCESS: Ubuntu LTS: Build
    SUCCESS: Ubuntu LTS: Integration
    IN_PROGRESS: Declarative: Post Actions

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/470/rebuild

review: Approve (continuous-integration)

diff --git a/cloudinit/sources/DataSourceAzure.py b/cloudinit/sources/DataSourceAzure.py
index be82ec4..e076d5d 100644
--- a/cloudinit/sources/DataSourceAzure.py
+++ b/cloudinit/sources/DataSourceAzure.py
@@ -22,6 +22,7 @@ from cloudinit.event import EventType
 from cloudinit.net.dhcp import EphemeralDHCPv4
 from cloudinit import sources
 from cloudinit.sources.helpers.azure import get_metadata_from_fabric
+from cloudinit.sources.helpers import netlink
 from cloudinit.url_helper import UrlError, readurl, retry_on_url_exc
 from cloudinit import util
 
@@ -409,6 +410,10 @@ class DataSourceAzure(sources.DataSource):
 
             perform_reprovision = reprovision or self._should_reprovision(ret)
             if perform_reprovision:
+                if util.is_FreeBSD():
+                    msg = "Free BSD is not supported for PPS VMs"
+                    LOG.error(msg)
+                    raise sources.InvalidMetaDataException(msg)
                 ret = self._reprovision()
             imds_md = get_metadata_from_imds(
                 self.fallback_interface, retries=3)
@@ -523,8 +528,8 @@ class DataSourceAzure(sources.DataSource):
         response. Then return the returned JSON object."""
         url = IMDS_URL + "reprovisiondata?api-version=2017-04-02"
         headers = {"Metadata": "true"}
+        nl_sock = None
         report_ready = bool(not os.path.isfile(REPORTED_READY_MARKER_FILE))
-        LOG.debug("Start polling IMDS")
 
         def exc_cb(msg, exception):
             if isinstance(exception, UrlError) and exception.code == 404:
@@ -533,12 +538,19 @@ class DataSourceAzure(sources.DataSource):
             # call DHCP and setup the ephemeral network to acquire the new IP.
             return False
 
+        LOG.debug("Wait for vnetswitch to happen")
         while True:
             try:
                 # Save our EphemeralDHCPv4 context so we avoid repeated dhcp
                 self._ephemeral_dhcp_ctx = EphemeralDHCPv4()
                 lease = self._ephemeral_dhcp_ctx.obtain_lease()
                 if report_ready:
+                    try:
+                        nl_sock = netlink.create_bound_netlink_socket()
+                    except netlink.NetlinkCreateSocketError as e:
+                        LOG.warning(e)
+                        self._ephemeral_dhcp_ctx.clean_network()
+                        return
                     path = REPORTED_READY_MARKER_FILE
                     LOG.info(
                         "Creating a marker file to report ready: %s", path)
@@ -546,13 +558,24 @@ class DataSourceAzure(sources.DataSource):
                         pid=os.getpid(), time=time()))
                     self._report_ready(lease=lease)
                     report_ready = False
-                return readurl(url, timeout=1, headers=headers,
-                               exception_cb=exc_cb, infinite=True,
-                               log_req_resp=False).contents
+                    try:
+                        netlink.wait_for_media_disconnect_connect(
+                            nl_sock, lease['interface'])
+                    except AssertionError as error:
+                        LOG.error(error)
+                        return
+                    self._ephemeral_dhcp_ctx.clean_network()
+                else:
+                    return readurl(url, timeout=1, headers=headers,
+                                   exception_cb=exc_cb, infinite=True,
+                                   log_req_resp=False).contents
             except UrlError:
                 # Teardown our EphemeralDHCPv4 context on failure as we retry
                 self._ephemeral_dhcp_ctx.clean_network()
                 pass
+            finally:
+                if nl_sock:
+                    nl_sock.close()
 
     def _report_ready(self, lease):
         """Tells the fabric provisioning has completed """
diff --git a/cloudinit/sources/helpers/netlink.py b/cloudinit/sources/helpers/netlink.py
new file mode 100644
index 0000000..d377ae3
--- /dev/null
+++ b/cloudinit/sources/helpers/netlink.py
@@ -0,0 +1,250 @@
+# Author: Tamilmani Manoharan <[e-mail]>
+#
+# This file is part of cloud-init. See LICENSE file for license information.
+
+from cloudinit import log as logging
+from cloudinit import util
+from collections import namedtuple
+
+import os
+import select
+import socket
+import struct
+
+LOG = logging.getLogger(__name__)
+
+# http://man7.org/linux/man-pages/man7/netlink.7.html
+RTMGRP_LINK = 1
+NLMSG_NOOP = 1
+NLMSG_ERROR = 2
+NLMSG_DONE = 3
+RTM_NEWLINK = 16
+RTM_DELLINK = 17
+RTM_GETLINK = 18
+RTM_SETLINK = 19
+MAX_SIZE = 65535
+RTA_DATA_OFFSET = 32
+MSG_TYPE_OFFSET = 16
+SELECT_TIMEOUT = 60
+
+NLMSGHDR_FMT = "IHHII"
+IFINFOMSG_FMT = "BHiII"
+NLMSGHDR_SIZE = struct.calcsize(NLMSGHDR_FMT)
+IFINFOMSG_SIZE = struct.calcsize(IFINFOMSG_FMT)
+RTATTR_START_OFFSET = NLMSGHDR_SIZE + IFINFOMSG_SIZE
+RTA_DATA_START_OFFSET = 4
+PAD_ALIGNMENT = 4
+
+IFLA_IFNAME = 3
+IFLA_OPERSTATE = 16
+
+# https://www.kernel.org/doc/Documentation/networking/operstates.txt
+OPER_UNKNOWN = 0
+OPER_NOTPRESENT = 1
+OPER_DOWN = 2
+OPER_LOWERLAYERDOWN = 3
+OPER_TESTING = 4
+OPER_DORMANT = 5
+OPER_UP = 6
+
+RTAAttr = namedtuple('RTAAttr', ['length', 'rta_type', 'data'])
+InterfaceOperstate = namedtuple('InterfaceOperstate', ['ifname', 'operstate'])
+NetlinkHeader = namedtuple('NetlinkHeader', ['length', 'type', 'flags', 'seq',
+                                             'pid'])
+
+
+class NetlinkCreateSocketError(RuntimeError):
+    '''Raised if netlink socket fails during create or bind.'''
+    pass
+
+
+def create_bound_netlink_socket():
+    '''Creates netlink socket and bind on netlink group to catch interface
+    down/up events. The socket will bound only on RTMGRP_LINK (which only
+    includes RTM_NEWLINK/RTM_DELLINK/RTM_GETLINK events). The socket is set to
+    non-blocking mode since we're only receiving messages.
+
+    :returns: netlink socket in non-blocking mode
+    :raises: NetlinkCreateSocketError
+    '''
+    try:
+        netlink_socket = socket.socket(socket.AF_NETLINK,
+                                       socket.SOCK_RAW,
+                                       socket.NETLINK_ROUTE)
+        netlink_socket.bind((os.getpid(), RTMGRP_LINK))
+        netlink_socket.setblocking(0)
+    except socket.error as e:
+        msg = "Exception during netlink socket create: %s" % e
+        raise NetlinkCreateSocketError(msg)
+    LOG.debug("Created netlink socket")
+    return netlink_socket
+
+
+def get_netlink_msg_header(data):
+    '''Gets netlink message type and length
+
+    :param: data read from netlink socket
+    :returns: netlink message type
+    :raises: AssertionError if data is None or data is not >= NLMSGHDR_SIZE
+    struct nlmsghdr {
+               __u32 nlmsg_len;    /* Length of message including header */
+               __u16 nlmsg_type;   /* Type of message content */
+               __u16 nlmsg_flags;  /* Additional flags */
+               __u32 nlmsg_seq;    /* Sequence number */
+               __u32 nlmsg_pid;    /* Sender port ID */
+    };
+    '''
+    assert (data is not None), ("data is none")
+    assert (len(data) >= NLMSGHDR_SIZE), (
+        "data is smaller than netlink message header")
+    msg_len, msg_type, flags, seq, pid = struct.unpack(NLMSGHDR_FMT,
+                                                       data[:MSG_TYPE_OFFSET])
+    LOG.debug("Got netlink msg of type %d", msg_type)
+    return NetlinkHeader(msg_len, msg_type, flags, seq, pid)
+
+
+def read_netlink_socket(netlink_socket, timeout=None):
+    '''Select and read from the netlink socket if ready.
+
+    :param: netlink_socket: specify which socket object to read from
+    :param: timeout: specify a timeout value (integer) to wait while reading,
+            if none, it will block indefinitely until socket ready for read
+    :returns: string of data read (max length = <MAX_SIZE>) from socket,
+              if no data read, returns None
+    :raises: AssertionError if netlink_socket is None
+    '''
+    assert (netlink_socket is not None), ("netlink socket is none")
+    read_set, _, _ = select.select([netlink_socket], [], [], timeout)
+    # Incase of timeout,read_set doesn't contain netlink socket.
+    # just return from this function
+    if netlink_socket not in read_set:
+        return None
+    LOG.debug("netlink socket ready for read")
+    data = netlink_socket.recv(MAX_SIZE)
+    if data is None:
+        LOG.error("Reading from Netlink socket returned no data")
+    return data
+
+
+def unpack_rta_attr(data, offset):
+    '''Unpack a single rta attribute.
+
+    :param: data: string of data read from netlink socket
+    :param: offset: starting offset of RTA Attribute
+    :return: RTAAttr object with length, type and data. On error, return None.
+    :raises: AssertionError if data is None or offset is not integer.
+    '''
+    assert (data is not None), ("data is none")
+    assert (type(offset) == int), ("offset is not integer")
+    assert (offset >= RTATTR_START_OFFSET), (
+        "rta offset is less than expected length")
+    length = rta_type = 0
+    attr_data = None
+    try:
+        length = struct.unpack_from("H", data, offset=offset)[0]
+        rta_type = struct.unpack_from("H", data, offset=offset+2)[0]
+    except struct.error:
+        return None  # Should mean our offset is >= remaining data
+
+    # Unpack just the attribute's data. Offset by 4 to skip length/type header
+    attr_data = data[offset+RTA_DATA_START_OFFSET:offset+length]
+    return RTAAttr(length, rta_type, attr_data)
+
+
+def read_rta_oper_state(data):
+    '''Reads Interface name and operational state from RTA Data.
+
+    :param: data: string of data read from netlink socket
+    :returns: InterfaceOperstate object containing if_name and oper_state.
+              None if data does not contain valid IFLA_OPERSTATE and
+              IFLA_IFNAME messages.
+    :raises: AssertionError if data is None or length of data is
+             smaller than RTATTR_START_OFFSET.
+    '''
+    assert (data is not None), ("data is none")
+    assert (len(data) > RTATTR_START_OFFSET), (
+        "length of data is smaller than RTATTR_START_OFFSET")
+    ifname = operstate = None
+    offset = RTATTR_START_OFFSET
+    while offset <= len(data):
+        attr = unpack_rta_attr(data, offset)
+        if not attr or attr.length == 0:
+            break
+        # Each attribute is 4-byte aligned. Determine pad length.
+        padlen = (PAD_ALIGNMENT -
+                  (attr.length % PAD_ALIGNMENT)) % PAD_ALIGNMENT
+        offset += attr.length + padlen
+
+        if attr.rta_type == IFLA_OPERSTATE:
+            operstate = ord(attr.data)
+        elif attr.rta_type == IFLA_IFNAME:
+            interface_name = util.decode_binary(attr.data, 'utf-8')
+            ifname = interface_name.strip('\0')
+    if not ifname or operstate is None:
+        return None
+    LOG.debug("rta attrs: ifname %s operstate %d", ifname, operstate)
+    return InterfaceOperstate(ifname, operstate)
+
+
+def wait_for_media_disconnect_connect(netlink_socket, ifname):
+    '''Block until media disconnect and connect has happened on an interface.
+    Listens on netlink socket to receive netlink events and when the carrier
+    changes from 0 to 1, it considers event has happened and
+    return from this function
+
+    :param: netlink_socket: netlink_socket to receive events
+    :param: ifname: Interface name to lookout for netlink events
+    :raises: AssertionError if netlink_socket is None or ifname is None.
+    '''
+    assert (netlink_socket is not None), ("netlink socket is none")
+    assert (ifname is not None), ("interface name is none")
+    assert (len(ifname) > 0), ("interface name cannot be empty")
+    carrier = OPER_UP
+    prevCarrier = OPER_UP
+    data = bytes()
+    LOG.debug("Wait for media disconnect and reconnect to happen")
+    while True:
+        recv_data = read_netlink_socket(netlink_socket, SELECT_TIMEOUT)
+        if recv_data is None:
+            continue
+        LOG.debug('read %d bytes from socket', len(recv_data))
+        data += recv_data
+        LOG.debug('Length of data after concat %d', len(data))
+        offset = 0
+        datalen = len(data)
+        while offset < datalen:
+            nl_msg = data[offset:]
+            if len(nl_msg) < NLMSGHDR_SIZE:
+                LOG.debug("Data is smaller than netlink header")
+                break
+            nlheader = get_netlink_msg_header(nl_msg)
+            if len(nl_msg) < nlheader.length:
+                LOG.debug("Partial data. Smaller than netlink message")
+                break
+            padlen = (nlheader.length+PAD_ALIGNMENT-1) & ~(PAD_ALIGNMENT-1)
+            offset = offset + padlen
+            LOG.debug('offset to next netlink message: %d', offset)
+            # Ignore any messages not new link or del link
+            if nlheader.type not in [RTM_NEWLINK, RTM_DELLINK]:
+                continue
+            interface_state = read_rta_oper_state(nl_msg)
+            if interface_state is None:
+                LOG.debug('Failed to read rta attributes: %s', interface_state)
+                continue
+            if interface_state.ifname != ifname:
+                LOG.debug(
+                    "Ignored netlink event on interface %s. Waiting for %s.",
+                    interface_state.ifname, ifname)
+                continue
+            if interface_state.operstate not in [OPER_UP, OPER_DOWN]:
+                continue
+            prevCarrier = carrier
+            carrier = interface_state.operstate
+            # check for carrier down, up sequence
+            isVnetSwitch = (prevCarrier == OPER_DOWN) and (carrier == OPER_UP)
+            if isVnetSwitch:
+                LOG.debug("Media switch happened on %s.", ifname)
+                return
+        data = data[offset:]
+
+# vi: ts=4 expandtab
diff --git a/cloudinit/sources/helpers/tests/test_netlink.py b/cloudinit/sources/helpers/tests/test_netlink.py
new file mode 100644
index 0000000..c2898a1
--- /dev/null
+++ b/cloudinit/sources/helpers/tests/test_netlink.py
@@ -0,0 +1,373 @@
+# Author: Tamilmani Manoharan <[e-mail]>
+#
+# This file is part of cloud-init. See LICENSE file for license information.
+
+from cloudinit.tests.helpers import CiTestCase, mock
+import socket
+import struct
+import codecs
+from cloudinit.sources.helpers.netlink import (
+    NetlinkCreateSocketError, create_bound_netlink_socket, read_netlink_socket,
+    read_rta_oper_state, unpack_rta_attr, wait_for_media_disconnect_connect,
+    OPER_DOWN, OPER_UP, OPER_DORMANT, OPER_LOWERLAYERDOWN, OPER_NOTPRESENT,
+    OPER_TESTING, OPER_UNKNOWN, RTATTR_START_OFFSET, RTM_NEWLINK, RTM_SETLINK,
+    RTM_GETLINK, MAX_SIZE)
+
+
+def int_to_bytes(i):
+    '''convert integer to binary: eg: 1 to \x01'''
+    hex_value = '{0:x}'.format(i)
+    hex_value = '0' * (len(hex_value) % 2) + hex_value
+    return codecs.decode(hex_value, 'hex_codec')
+
+
+class TestCreateBoundNetlinkSocket(CiTestCase):
+
+    @mock.patch('cloudinit.sources.helpers.netlink.socket.socket')
+    def test_socket_error_on_create(self, m_socket):
+        '''create_bound_netlink_socket catches socket creation exception'''
+
+        """NetlinkCreateSocketError is raised when socket creation errors."""
+        m_socket.side_effect = socket.error("Fake socket failure")
+        with self.assertRaises(NetlinkCreateSocketError) as ctx_mgr:
+            create_bound_netlink_socket()
+        self.assertEqual(
+            'Exception during netlink socket create: Fake socket failure',
+            str(ctx_mgr.exception))
+
+
+class TestReadNetlinkSocket(CiTestCase):
+
+    @mock.patch('cloudinit.sources.helpers.netlink.socket.socket')
+    @mock.patch('cloudinit.sources.helpers.netlink.select.select')
+    def test_read_netlink_socket(self, m_select, m_socket):
+        '''read_netlink_socket able to receive data'''
+        data = 'netlinktest'
+        m_select.return_value = [m_socket], None, None
+        m_socket.recv.return_value = data
+        recv_data = read_netlink_socket(m_socket, 2)
+        m_select.assert_called_with([m_socket], [], [], 2)
+        m_socket.recv.assert_called_with(MAX_SIZE)
+        self.assertIsNotNone(recv_data)
+        self.assertEqual(recv_data, data)
+
+    @mock.patch('cloudinit.sources.helpers.netlink.socket.socket')
+    @mock.patch('cloudinit.sources.helpers.netlink.select.select')
+    def test_netlink_read_timeout(self, m_select, m_socket):
+        '''read_netlink_socket should timeout if nothing to read'''
+        m_select.return_value = [], None, None
+        data = read_netlink_socket(m_socket, 1)
+        m_select.assert_called_with([m_socket], [], [], 1)
+        self.assertEqual(m_socket.recv.call_count, 0)
+        self.assertIsNone(data)
+
+    def test_read_invalid_socket(self):
+        '''read_netlink_socket raises assert error if socket is invalid'''
+        socket = None
+        with self.assertRaises(AssertionError) as context:
+            read_netlink_socket(socket, 1)
+        self.assertTrue('netlink socket is none' in str(context.exception))
+
+
+class TestParseNetlinkMessage(CiTestCase):
+
+    def test_read_rta_oper_state(self):
+        '''read_rta_oper_state could parse netlink message and extract data'''
+        ifname = "eth0"
+        bytes = ifname.encode("utf-8")
+        buf = bytearray(48)
+        struct.pack_into("HH4sHHc", buf, RTATTR_START_OFFSET, 8, 3, bytes, 5,
+                         16, int_to_bytes(OPER_DOWN))
+        interface_state = read_rta_oper_state(buf)
+        self.assertEqual(interface_state.ifname, ifname)
+        self.assertEqual(interface_state.operstate, OPER_DOWN)
+
+    def test_read_none_data(self):
+        '''read_rta_oper_state raises assert error if data is none'''
+        data = None
+        with self.assertRaises(AssertionError) as context:
+            read_rta_oper_state(data)
+        self.assertTrue('data is none', str(context.exception))
+
+    def test_read_invalid_rta_operstate_none(self):
+        '''read_rta_oper_state returns none if operstate is none'''
+        ifname = "eth0"
+        buf = bytearray(40)
+        bytes = ifname.encode("utf-8")
+        struct.pack_into("HH4s", buf, RTATTR_START_OFFSET, 8, 3, bytes)
+        interface_state = read_rta_oper_state(buf)
+        self.assertIsNone(interface_state)
+
+    def test_read_invalid_rta_ifname_none(self):
+        '''read_rta_oper_state returns none if ifname is none'''
+        buf = bytearray(40)
+        struct.pack_into("HHc", buf, RTATTR_START_OFFSET, 5, 16,
+                         int_to_bytes(OPER_DOWN))
+        interface_state = read_rta_oper_state(buf)
+        self.assertIsNone(interface_state)
+
+    def test_read_invalid_data_len(self):
+        '''raise assert error if data size is smaller than required size'''
+        buf = bytearray(32)
+        with self.assertRaises(AssertionError) as context:
+            read_rta_oper_state(buf)
+        self.assertTrue('length of data is smaller than RTATTR_START_OFFSET' in
+                        str(context.exception))
+
+    def test_unpack_rta_attr_none_data(self):
+        '''unpack_rta_attr raises assert error if data is none'''
+        data = None
+        with self.assertRaises(AssertionError) as context:
+            unpack_rta_attr(data, RTATTR_START_OFFSET)
+        self.assertTrue('data is none' in str(context.exception))
+
+    def test_unpack_rta_attr_invalid_offset(self):
+        '''unpack_rta_attr raises assert error if offset is invalid'''
+        data = bytearray(48)
+        with self.assertRaises(AssertionError) as context:
+            unpack_rta_attr(data, "offset")
+        self.assertTrue('offset is not integer' in str(context.exception))
+        with self.assertRaises(AssertionError) as context:
+            unpack_rta_attr(data, 31)
+        self.assertTrue('rta offset is less than expected length' in
+                        str(context.exception))
+
+
[e-mail]('cloudinit.sources.helpers.netlink.socket.socket')
[e-mail]('cloudinit.sources.helpers.netlink.read_netlink_socket')
+class TestWaitForMediaDisconnectConnect(CiTestCase):
+    with_logs = True
+
+    def _media_switch_data(self, ifname, msg_type, operstate):
+        '''construct netlink data with specified fields'''
+        if ifname and operstate is not None:
+            data = bytearray(48)
+            bytes = ifname.encode("utf-8")
+            struct.pack_into("HH4sHHc", data, RTATTR_START_OFFSET, 8, 3,
+                             bytes, 5, 16, int_to_bytes(operstate))
+        elif ifname:
+            data = bytearray(40)
+            bytes = ifname.encode("utf-8")
+            struct.pack_into("HH4s", data, RTATTR_START_OFFSET, 8, 3, bytes)
+        elif operstate:
+            data = bytearray(40)
+            struct.pack_into("HHc", data, RTATTR_START_OFFSET, 5, 16,
+                             int_to_bytes(operstate))
+        struct.pack_into("=LHHLL", data, 0, len(data), msg_type, 0, 0, 0)
+        return data
+
+    def test_media_down_up_scenario(self, m_read_netlink_socket,
+                                    m_socket):
+        '''Test for media down up sequence for required interface name'''
+        ifname = "eth0"
+        # construct data for Oper State down
+        data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+        # construct data for Oper State up
+        data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+        m_read_netlink_socket.side_effect = [data_op_down, data_op_up]
+        wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertEqual(m_read_netlink_socket.call_count, 2)
+
+    def test_wait_for_media_switch_diff_interface(self, m_read_netlink_socket,
+                                                  m_socket):
+        '''wait_for_media_disconnect_connect ignores unexpected interfaces.
+
+        The first two messages are for other interfaces and last two are for
+        expected interface. So the function exit only after receiving last
+        2 messages and therefore the call count for m_read_netlink_socket
+        has to be 4
+        '''
+        other_ifname = "eth1"
+        expected_ifname = "eth0"
+        data_op_down_eth1 = self._media_switch_data(
+                                other_ifname, RTM_NEWLINK, OPER_DOWN)
+        data_op_up_eth1 = self._media_switch_data(
+                                other_ifname, RTM_NEWLINK, OPER_UP)
+        data_op_down_eth0 = self._media_switch_data(
+                                expected_ifname, RTM_NEWLINK, OPER_DOWN)
+        data_op_up_eth0 = self._media_switch_data(
+                                expected_ifname, RTM_NEWLINK, OPER_UP)
+        m_read_netlink_socket.side_effect = [data_op_down_eth1,
+                                             data_op_up_eth1,
+                                             data_op_down_eth0,
+                                             data_op_up_eth0]
+        wait_for_media_disconnect_connect(m_socket, expected_ifname)
+        self.assertIn('Ignored netlink event on interface %s' % other_ifname,
+                      self.logs.getvalue())
+        self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+    def test_invalid_msgtype_getlink(self, m_read_netlink_socket, m_socket):
+        '''wait_for_media_disconnect_connect ignores GETLINK events.
+
+        The first two messages are for oper down and up for RTM_GETLINK type
+        which netlink module will ignore. The last 2 messages are RTM_NEWLINK
+        with oper state down and up messages. Therefore the call count for
+        m_read_netlink_socket has to be 4 ignoring first 2 messages
+        of RTM_GETLINK
+        '''
+        ifname = "eth0"
+        data_getlink_down = self._media_switch_data(
+                                    ifname, RTM_GETLINK, OPER_DOWN)
+        data_getlink_up = self._media_switch_data(
+                                    ifname, RTM_GETLINK, OPER_UP)
+        data_newlink_down = self._media_switch_data(
+                                    ifname, RTM_NEWLINK, OPER_DOWN)
+        data_newlink_up = self._media_switch_data(
+                                    ifname, RTM_NEWLINK, OPER_UP)
+        m_read_netlink_socket.side_effect = [data_getlink_down,
+                                             data_getlink_up,
+                                             data_newlink_down,
+                                             data_newlink_up]
+        wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+    def test_invalid_msgtype_setlink(self, m_read_netlink_socket, m_socket):
+        '''wait_for_media_disconnect_connect ignores SETLINK events.
+
+        The first two messages are for oper down and up for RTM_GETLINK type
+        which it will ignore. 3rd and 4th messages are RTM_NEWLINK with down
+        and up messages. This function should exit after 4th messages since it
+        sees down->up scenario. So the call count for m_read_netlink_socket
+        has to be 4 ignoring first 2 messages of RTM_GETLINK and
+        last 2 messages of RTM_NEWLINK
+        '''
+        ifname = "eth0"
+        data_setlink_down = self._media_switch_data(
+                                    ifname, RTM_SETLINK, OPER_DOWN)
+        data_setlink_up = self._media_switch_data(
+                                    ifname, RTM_SETLINK, OPER_UP)
+        data_newlink_down = self._media_switch_data(
+                                    ifname, RTM_NEWLINK, OPER_DOWN)
+        data_newlink_up = self._media_switch_data(
+                                    ifname, RTM_NEWLINK, OPER_UP)
+        m_read_netlink_socket.side_effect = [data_setlink_down,
+                                             data_setlink_up,
+                                             data_newlink_down,
+                                             data_newlink_up,
+                                             data_newlink_down,
+                                             data_newlink_up]
+        wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+    def test_netlink_invalid_switch_scenario(self, m_read_netlink_socket,
+                                             m_socket):
+        '''returns only if it receives UP event after a DOWN event'''
+        ifname = "eth0"
+        data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+        data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+        data_op_dormant = self._media_switch_data(ifname, RTM_NEWLINK,
+                                                  OPER_DORMANT)
+        data_op_notpresent = self._media_switch_data(ifname, RTM_NEWLINK,
+                                                     OPER_NOTPRESENT)
+        data_op_lowerdown = self._media_switch_data(ifname, RTM_NEWLINK,
+                                                    OPER_LOWERLAYERDOWN)
+        data_op_testing = self._media_switch_data(ifname, RTM_NEWLINK,
+                                                  OPER_TESTING)
+        data_op_unknown = self._media_switch_data(ifname, RTM_NEWLINK,
+                                                  OPER_UNKNOWN)
+        m_read_netlink_socket.side_effect = [data_op_up, data_op_up,
+                                             data_op_dormant, data_op_up,
+                                             data_op_notpresent, data_op_up,
+                                             data_op_lowerdown, data_op_up,
+                                             data_op_testing, data_op_up,
+                                             data_op_unknown, data_op_up,
+                                             data_op_down, data_op_up]
+        wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertEqual(m_read_netlink_socket.call_count, 14)
+
+    def test_netlink_valid_inbetween_transitions(self, m_read_netlink_socket,
+                                                 m_socket):
+        '''wait_for_media_disconnect_connect handles in between transitions'''
+        ifname = "eth0"
+        data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+        data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+        data_op_dormant = self._media_switch_data(ifname, RTM_NEWLINK,
+                                                  OPER_DORMANT)
+        data_op_unknown = self._media_switch_data(ifname, RTM_NEWLINK,
+                                                  OPER_UNKNOWN)
+        m_read_netlink_socket.side_effect = [data_op_down, data_op_dormant,
+                                             data_op_unknown, data_op_up]
+        wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+    def test_netlink_invalid_operstate(self, m_read_netlink_socket, m_socket):
+        '''wait_for_media_disconnect_connect should handle invalid operstates.
+
+        The function should not fail and return even if it receives invalid
+        operstates. It always should wait for down up sequence.
+        '''
+        ifname = "eth0"
+        data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+        data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+        data_op_invalid = self._media_switch_data(ifname, RTM_NEWLINK, 7)
+        m_read_netlink_socket.side_effect = [data_op_invalid, data_op_up,
+                                             data_op_down, data_op_invalid,
+                                             data_op_up]
+        wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertEqual(m_read_netlink_socket.call_count, 5)
+
+    def test_wait_invalid_socket(self, m_read_netlink_socket, m_socket):
+        '''wait_for_media_disconnect_connect handle none netlink socket.'''
+        socket = None
+        ifname = "eth0"
+        with self.assertRaises(AssertionError) as context:
+            wait_for_media_disconnect_connect(socket, ifname)
+        self.assertTrue('netlink socket is none' in str(context.exception))
+
+    def test_wait_invalid_ifname(self, m_read_netlink_socket, m_socket):
+        '''wait_for_media_disconnect_connect handle none interface name'''
+        ifname = None
+        with self.assertRaises(AssertionError) as context:
+            wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertTrue('interface name is none' in str(context.exception))
+        ifname = ""
+        with self.assertRaises(AssertionError) as context:
+            wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertTrue('interface name cannot be empty' in
+                        str(context.exception))
+
+    def test_wait_invalid_rta_attr(self, m_read_netlink_socket, m_socket):
+        ''' wait_for_media_disconnect_connect handles invalid rta data'''
+        ifname = "eth0"
+        data_invalid1 = self._media_switch_data(None, RTM_NEWLINK, OPER_DOWN)
+        data_invalid2 = self._media_switch_data(ifname, RTM_NEWLINK, None)
+        data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+        data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+        m_read_netlink_socket.side_effect = [data_invalid1, data_invalid2,
+                                             data_op_down, data_op_up]
+        wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+    def test_read_multiple_netlink_msgs(self, m_read_netlink_socket, m_socket):
+        '''Read multiple messages in single receive call'''
+        ifname = "eth0"
+        bytes = ifname.encode("utf-8")
+        data = bytearray(96)
+        struct.pack_into("=LHHLL", data, 0, 48, RTM_NEWLINK, 0, 0, 0)
+        struct.pack_into("HH4sHHc", data, RTATTR_START_OFFSET, 8, 3,
+                         bytes, 5, 16, int_to_bytes(OPER_DOWN))
+        struct.pack_into("=LHHLL", data, 48, 48, RTM_NEWLINK, 0, 0, 0)
+        struct.pack_into("HH4sHHc", data, 48 + RTATTR_START_OFFSET, 8,
+                         3, bytes, 5, 16, int_to_bytes(OPER_UP))
+        m_read_netlink_socket.return_value = data
+        wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertEqual(m_read_netlink_socket.call_count, 1)
+
+    def test_read_partial_netlink_msgs(self, m_read_netlink_socket, m_socket):
+        '''Read partial messages in receive call'''
+        ifname = "eth0"
+        bytes = ifname.encode("utf-8")
+        data1 = bytearray(112)
+        data2 = bytearray(32)
+        struct.pack_into("=LHHLL", data1, 0, 48, RTM_NEWLINK, 0, 0, 0)
+        struct.pack_into("HH4sHHc", data1, RTATTR_START_OFFSET, 8, 3,
+                         bytes, 5, 16, int_to_bytes(OPER_DOWN))
+        struct.pack_into("=LHHLL", data1, 48, 48, RTM_NEWLINK, 0, 0, 0)
+        struct.pack_into("HH4sHHc", data1, 80, 8, 3, bytes, 5, 16,
+                         int_to_bytes(OPER_DOWN))
+        struct.pack_into("=LHHLL", data1, 96, 48, RTM_NEWLINK, 0, 0, 0)
+        struct.pack_into("HH4sHHc", data2, 16, 8, 3, bytes, 5, 16,
+                         int_to_bytes(OPER_UP))
+        m_read_netlink_socket.side_effect = [data1, data2]
+        wait_for_media_disconnect_connect(m_socket, ifname)
+        self.assertEqual(m_read_netlink_socket.call_count, 2)
diff --git a/tests/unittests/test_datasource/test_azure.py b/tests/unittests/test_datasource/test_azure.py
index 5ea7ae5..417d86a 100644
--- a/tests/unittests/test_datasource/test_azure.py
+++ b/tests/unittests/test_datasource/test_azure.py
@@ -564,6 +564,8 @@ fdescfs            /dev/fd          fdescfs rw              0 0
         self.assertEqual(1, report_ready_func.call_count)
 
     @mock.patch('cloudinit.sources.DataSourceAzure.util.write_file')
+    @mock.patch('cloudinit.sources.helpers.netlink.'
+                'wait_for_media_disconnect_connect')
     @mock.patch(
         'cloudinit.sources.DataSourceAzure.DataSourceAzure._report_ready')
     @mock.patch('cloudinit.net.dhcp.EphemeralIPv4Network')
@@ -572,7 +574,7 @@ fdescfs            /dev/fd          fdescfs rw              0 0
     def test_crawl_metadata_on_reprovision_reports_ready_using_lease(
                             self, m_readurl, m_dhcp,
                             m_net, report_ready_func,
-                            m_write):
+                            m_media_switch, m_write):
         """If reprovisioning, report ready using the obtained lease"""
         ovfenv = construct_valid_ovf_env(
                             platform_settings={"PreprovisionedVm": "True"})
@@ -586,6 +588,7 @@ fdescfs            /dev/fd          fdescfs rw              0 0
             'routers': '192.168.2.1', 'subnet-mask': '255.255.255.0',
             'unknown-245': '624c3620'}
         m_dhcp.return_value = [lease]
+        m_media_switch.return_value = None
 
         reprovision_ovfenv = construct_valid_ovf_env()
         m_readurl.return_value = url_helper.StringResponse(
@@ -1676,6 +1679,8 @@ class TestPreprovisioningShouldReprovision(CiTestCase):
 
 @mock.patch('cloudinit.net.dhcp.EphemeralIPv4Network')
 @mock.patch('cloudinit.net.dhcp.maybe_perform_dhcp_discovery')
[e-mail]('cloudinit.sources.helpers.netlink.'
+            'wait_for_media_disconnect_connect')
 @mock.patch('requests.Session.request')
 @mock.patch(MOCKPATH + 'DataSourceAzure._report_ready')
 class TestPreprovisioningPollIMDS(CiTestCase):
@@ -1689,7 +1694,8 @@ class TestPreprovisioningPollIMDS(CiTestCase):
 
     @mock.patch(MOCKPATH + 'EphemeralDHCPv4')
     def test_poll_imds_re_dhcp_on_timeout(self, m_dhcpv4, report_ready_func,
-                                          fake_resp, m_dhcp, m_net):
+                                          fake_resp, m_media_switch, m_dhcp,
+                                          m_net):
         """The poll_imds will retry DHCP on IMDS timeout."""
         report_file = self.tmp_path('report_marker', self.tmp)
         lease = {
@@ -1697,7 +1703,7 @@ class TestPreprovisioningPollIMDS(CiTestCase):
             'routers': '192.168.2.1', 'subnet-mask': '255.255.255.0',
             'unknown-245': '624c3620'}
         m_dhcp.return_value = [lease]
-
+        m_media_switch.return_value = None
         dhcp_ctx = mock.MagicMock(lease=lease)
         dhcp_ctx.obtain_lease.return_value = lease
         m_dhcpv4.return_value = dhcp_ctx
@@ -1723,11 +1729,12 @@ class TestPreprovisioningPollIMDS(CiTestCase):
             dsa._poll_imds()
         self.assertEqual(report_ready_func.call_count, 1)
         report_ready_func.assert_called_with(lease=lease)
-        self.assertEqual(2, m_dhcpv4.call_count, 'Expected 2 DHCP calls')
+        self.assertEqual(3, m_dhcpv4.call_count, 'Expected 3 DHCP calls')
         self.assertEqual(3, self.tries, 'Expected 3 total reads from IMDS')
 
-    def test_poll_imds_report_ready_false(self, report_ready_func,
-                                          fake_resp, m_dhcp, m_net):
+    def test_poll_imds_report_ready_false(self,
+                                          report_ready_func, fake_resp,
+                                          m_media_switch, m_dhcp, m_net):
         """The poll_imds should not call reporting ready
            when flag is false"""
         report_file = self.tmp_path('report_marker', self.tmp)
@@ -1736,6 +1743,7 @@ class TestPreprovisioningPollIMDS(CiTestCase):
             'interface': 'eth9', 'fixed-address': '192.168.2.9',
             'routers': '192.168.2.1', 'subnet-mask': '255.255.255.0',
             'unknown-245': '624c3620'}]
+        m_media_switch.return_value = None
         dsa = dsaz.DataSourceAzure({}, distro=None, paths=self.paths)
         with mock.patch(MOCKPATH + 'REPORTED_READY_MARKER_FILE', report_file):
             dsa._poll_imds()
@@ -1745,6 +1753,8 @@ class TestPreprovisioningPollIMDS(CiTestCase):
 @mock.patch(MOCKPATH + 'util.subp')
 @mock.patch(MOCKPATH + 'util.write_file')
 @mock.patch(MOCKPATH + 'util.is_FreeBSD')
[e-mail]('cloudinit.sources.helpers.netlink.'
+            'wait_for_media_disconnect_connect')
 @mock.patch('cloudinit.net.dhcp.EphemeralIPv4Network')
 @mock.patch('cloudinit.net.dhcp.maybe_perform_dhcp_discovery')
 @mock.patch('requests.Session.request')
@@ -1757,10 +1767,13 @@ class TestAzureDataSourcePreprovisioning(CiTestCase):
         self.paths = helpers.Paths({'cloud_dir': tmp})
         dsaz.BUILTIN_DS_CONFIG['data_dir'] = self.waagent_d
 
-    def test_poll_imds_returns_ovf_env(self, fake_resp, m_dhcp, m_net,
+    def test_poll_imds_returns_ovf_env(self, fake_resp,
+                                       m_dhcp, m_net,
+                                       m_media_switch,
                                        m_is_bsd, write_f, subp):
         """The _poll_imds method should return the ovf_env.xml."""
         m_is_bsd.return_value = False
+        m_media_switch.return_value = None
         m_dhcp.return_value = [{
             'interface': 'eth9', 'fixed-address': '192.168.2.9',
             'routers': '192.168.2.1', 'subnet-mask': '255.255.255.0'}]
@@ -1778,16 +1791,19 @@ class TestAzureDataSourcePreprovisioning(CiTestCase):
                                              'Cloud-Init/%s' % vs()
                                              }, method='GET', timeout=1,
                                     url=full_url)])
-        self.assertEqual(m_dhcp.call_count, 1)
+        self.assertEqual(m_dhcp.call_count, 2)
         m_net.assert_any_call(
             broadcast='192.168.2.255', interface='eth9', ip='192.168.2.9',
             prefix_or_mask='255.255.255.0', router='192.168.2.1')
-        self.assertEqual(m_net.call_count, 1)
+        self.assertEqual(m_net.call_count, 2)
 
-    def test__reprovision_calls__poll_imds(self, fake_resp, m_dhcp, m_net,
+    def test__reprovision_calls__poll_imds(self, fake_resp,
+                                           m_dhcp, m_net,
+                                           m_media_switch,
                                            m_is_bsd, write_f, subp):
         """The _reprovision method should call poll IMDS."""
         m_is_bsd.return_value = False
+        m_media_switch.return_value = None
         m_dhcp.return_value = [{
             'interface': 'eth9', 'fixed-address': '192.168.2.9',
             'routers': '192.168.2.1', 'subnet-mask': '255.255.255.0',
@@ -1811,11 +1827,11 @@ class TestAzureDataSourcePreprovisioning(CiTestCase):
                                              'User-Agent':
                                              'Cloud-Init/%s' % vs()},
                                     method='GET', timeout=1, url=full_url)])
-        self.assertEqual(m_dhcp.call_count, 1)
+        self.assertEqual(m_dhcp.call_count, 2)
         m_net.assert_any_call(
             broadcast='192.168.2.255', interface='eth9', ip='192.168.2.9',
             prefix_or_mask='255.255.255.0', router='192.168.2.1')
-        self.assertEqual(m_net.call_count, 1)
+        self.assertEqual(m_net.call_count, 2)
 
 
 class TestRemoveUbuntuNetworkConfigScripts(CiTestCase):