Merge ~chad.smith/cloud-init:ubuntu/devel into cloud-init:ubuntu/devel

[Chad Smith]

New-upstream-snapshot with azure kvp reporting feature

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

PASSED: Continuous integration, rev:00362e3be579a8157e0a915b250888c39eb408fc
https://jenkins.ubuntu.com/server/job/cloud-init-ci/284/
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/284/rebuild

review: Approve (continuous-integration)

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

PASSED: Continuous integration, rev:1c047bd2a9661be9a82e072723f15560eebcafdd
https://jenkins.ubuntu.com/server/job/cloud-init-ci/285/
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/285/rebuild

review: Approve (continuous-integration)

diff --git a/cloudinit/cloud.py b/cloudinit/cloud.py
index 6d12c43..7ae98e1 100644
--- a/cloudinit/cloud.py
+++ b/cloudinit/cloud.py
@@ -47,7 +47,7 @@ class Cloud(object):
 
     @property
     def cfg(self):
-        # Ensure that not indirectly modified
+        # Ensure that cfg is not indirectly modified
         return copy.deepcopy(self._cfg)
 
     def run(self, name, functor, args, freq=None, clear_on_fail=False):
@@ -61,7 +61,7 @@ class Cloud(object):
             return None
         return fn
 
-    # The rest of thes are just useful proxies
+    # The rest of these are just useful proxies
     def get_userdata(self, apply_filter=True):
         return self.datasource.get_userdata(apply_filter)
 
diff --git a/cloudinit/cmd/main.py b/cloudinit/cmd/main.py
index d6ba90f..4ea4fe7 100644
--- a/cloudinit/cmd/main.py
+++ b/cloudinit/cmd/main.py
@@ -315,7 +315,7 @@ def main_init(name, args):
                 existing = "trust"
 
         init.purge_cache()
-        # Delete the non-net file as well
+        # Delete the no-net file as well
         util.del_file(os.path.join(path_helper.get_cpath("data"), "no-net"))
 
     # Stage 5
@@ -339,7 +339,7 @@ def main_init(name, args):
                               " Likely bad things to come!"))
         if not args.force:
             init.apply_network_config(bring_up=not args.local)
-            LOG.debug("[%s] Exiting without datasource in local mode", mode)
+            LOG.debug("[%s] Exiting without datasource", mode)
             if mode == sources.DSMODE_LOCAL:
                 return (None, [])
             else:
@@ -877,9 +877,11 @@ def main(sysv_args=None):
         rname, rdesc, reporting_enabled=report_on)
 
     with args.reporter:
-        return util.log_time(
+        retval = util.log_time(
             logfunc=LOG.debug, msg="cloud-init mode '%s'" % name,
             get_uptime=True, func=functor, args=(name, args))
+        reporting.flush_events()
+        return retval
 
 
 if __name__ == '__main__':
diff --git a/cloudinit/config/tests/test_ssh.py b/cloudinit/config/tests/test_ssh.py
new file mode 100644
index 0000000..7441d9e
--- /dev/null
+++ b/cloudinit/config/tests/test_ssh.py
@@ -0,0 +1,147 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+
+from cloudinit.config import cc_ssh
+from cloudinit.tests.helpers import CiTestCase, mock
+
+MODPATH = "cloudinit.config.cc_ssh."
+
+
[e-mail](MODPATH + "ssh_util.setup_user_keys")
+class TestHandleSsh(CiTestCase):
+    """Test cc_ssh handling of ssh config."""
+
+    def test_apply_credentials_with_user(self, m_setup_keys):
+        """Apply keys for the given user and root."""
+        keys = ["key1"]
+        user = "clouduser"
+        options = cc_ssh.DISABLE_ROOT_OPTS
+        cc_ssh.apply_credentials(keys, user, False, options)
+        self.assertEqual([mock.call(set(keys), user),
+                          mock.call(set(keys), "root", options="")],
+                         m_setup_keys.call_args_list)
+
+    def test_apply_credentials_with_no_user(self, m_setup_keys):
+        """Apply keys for root only."""
+        keys = ["key1"]
+        user = None
+        options = cc_ssh.DISABLE_ROOT_OPTS
+        cc_ssh.apply_credentials(keys, user, False, options)
+        self.assertEqual([mock.call(set(keys), "root", options="")],
+                         m_setup_keys.call_args_list)
+
+    def test_apply_credentials_with_user_disable_root(self, m_setup_keys):
+        """Apply keys for the given user and disable root ssh."""
+        keys = ["key1"]
+        user = "clouduser"
+        options = cc_ssh.DISABLE_ROOT_OPTS
+        cc_ssh.apply_credentials(keys, user, True, options)
+        options = options.replace("$USER", user)
+        self.assertEqual([mock.call(set(keys), user),
+                          mock.call(set(keys), "root", options=options)],
+                         m_setup_keys.call_args_list)
+
+    def test_apply_credentials_with_no_user_disable_root(self, m_setup_keys):
+        """Apply keys no user and disable root ssh."""
+        keys = ["key1"]
+        user = None
+        options = cc_ssh.DISABLE_ROOT_OPTS
+        cc_ssh.apply_credentials(keys, user, True, options)
+        options = options.replace("$USER", "NONE")
+        self.assertEqual([mock.call(set(keys), "root", options=options)],
+                         m_setup_keys.call_args_list)
+
+    @mock.patch(MODPATH + "glob.glob")
+    @mock.patch(MODPATH + "ug_util.normalize_users_groups")
+    @mock.patch(MODPATH + "os.path.exists")
+    def test_handle_no_cfg(self, m_path_exists, m_nug,
+                           m_glob, m_setup_keys):
+        """Test handle with no config ignores generating existing keyfiles."""
+        cfg = {}
+        keys = ["key1"]
+        m_glob.return_value = []  # Return no matching keys to prevent removal
+        # Mock os.path.exits to True to short-circuit the key writing logic
+        m_path_exists.return_value = True
+        m_nug.return_value = ([], {})
+        cloud = self.tmp_cloud(
+            distro='ubuntu', metadata={'public-keys': keys})
+        cc_ssh.handle("name", cfg, cloud, None, None)
+        options = cc_ssh.DISABLE_ROOT_OPTS.replace("$USER", "NONE")
+        m_glob.assert_called_once_with('/etc/ssh/ssh_host_*key*')
+        self.assertIn(
+            [mock.call('/etc/ssh/ssh_host_rsa_key'),
+             mock.call('/etc/ssh/ssh_host_dsa_key'),
+             mock.call('/etc/ssh/ssh_host_ecdsa_key'),
+             mock.call('/etc/ssh/ssh_host_ed25519_key')],
+            m_path_exists.call_args_list)
+        self.assertEqual([mock.call(set(keys), "root", options=options)],
+                         m_setup_keys.call_args_list)
+
+    @mock.patch(MODPATH + "glob.glob")
+    @mock.patch(MODPATH + "ug_util.normalize_users_groups")
+    @mock.patch(MODPATH + "os.path.exists")
+    def test_handle_no_cfg_and_default_root(self, m_path_exists, m_nug,
+                                            m_glob, m_setup_keys):
+        """Test handle with no config and a default distro user."""
+        cfg = {}
+        keys = ["key1"]
+        user = "clouduser"
+        m_glob.return_value = []  # Return no matching keys to prevent removal
+        # Mock os.path.exits to True to short-circuit the key writing logic
+        m_path_exists.return_value = True
+        m_nug.return_value = ({user: {"default": user}}, {})
+        cloud = self.tmp_cloud(
+            distro='ubuntu', metadata={'public-keys': keys})
+        cc_ssh.handle("name", cfg, cloud, None, None)
+
+        options = cc_ssh.DISABLE_ROOT_OPTS.replace("$USER", user)
+        self.assertEqual([mock.call(set(keys), user),
+                          mock.call(set(keys), "root", options=options)],
+                         m_setup_keys.call_args_list)
+
+    @mock.patch(MODPATH + "glob.glob")
+    @mock.patch(MODPATH + "ug_util.normalize_users_groups")
+    @mock.patch(MODPATH + "os.path.exists")
+    def test_handle_cfg_with_explicit_disable_root(self, m_path_exists, m_nug,
+                                                   m_glob, m_setup_keys):
+        """Test handle with explicit disable_root and a default distro user."""
+        # This test is identical to test_handle_no_cfg_and_default_root,
+        # except this uses an explicit cfg value
+        cfg = {"disable_root": True}
+        keys = ["key1"]
+        user = "clouduser"
+        m_glob.return_value = []  # Return no matching keys to prevent removal
+        # Mock os.path.exits to True to short-circuit the key writing logic
+        m_path_exists.return_value = True
+        m_nug.return_value = ({user: {"default": user}}, {})
+        cloud = self.tmp_cloud(
+            distro='ubuntu', metadata={'public-keys': keys})
+        cc_ssh.handle("name", cfg, cloud, None, None)
+
+        options = cc_ssh.DISABLE_ROOT_OPTS.replace("$USER", user)
+        self.assertEqual([mock.call(set(keys), user),
+                          mock.call(set(keys), "root", options=options)],
+                         m_setup_keys.call_args_list)
+
+    @mock.patch(MODPATH + "glob.glob")
+    @mock.patch(MODPATH + "ug_util.normalize_users_groups")
+    @mock.patch(MODPATH + "os.path.exists")
+    def test_handle_cfg_without_disable_root(self, m_path_exists, m_nug,
+                                             m_glob, m_setup_keys):
+        """Test handle with disable_root == False."""
+        # When disable_root == False, the ssh redirect for root is skipped
+        cfg = {"disable_root": False}
+        keys = ["key1"]
+        user = "clouduser"
+        m_glob.return_value = []  # Return no matching keys to prevent removal
+        # Mock os.path.exits to True to short-circuit the key writing logic
+        m_path_exists.return_value = True
+        m_nug.return_value = ({user: {"default": user}}, {})
+        cloud = self.tmp_cloud(
+            distro='ubuntu', metadata={'public-keys': keys})
+        cloud.get_public_ssh_keys = mock.Mock(return_value=keys)
+        cc_ssh.handle("name", cfg, cloud, None, None)
+
+        self.assertEqual([mock.call(set(keys), user),
+                          mock.call(set(keys), "root", options="")],
+                         m_setup_keys.call_args_list)
diff --git a/cloudinit/distros/__init__.py b/cloudinit/distros/__init__.py
index ab0b077..fde054e 100755
--- a/cloudinit/distros/__init__.py
+++ b/cloudinit/distros/__init__.py
@@ -157,7 +157,7 @@ class Distro(object):
                     distro)
         header = '\n'.join([
             "# Converted from network_config for distro %s" % distro,
-            "# Implmentation of _write_network_config is needed."
+            "# Implementation of _write_network_config is needed."
         ])
         ns = network_state.parse_net_config_data(netconfig)
         contents = eni.network_state_to_eni(
diff --git a/cloudinit/reporting/__init__.py b/cloudinit/reporting/__init__.py
index 1ed2b48..ed5c703 100644
--- a/cloudinit/reporting/__init__.py
+++ b/cloudinit/reporting/__init__.py
@@ -18,7 +18,7 @@ DEFAULT_CONFIG = {
 
 
 def update_configuration(config):
-    """Update the instanciated_handler_registry.
+    """Update the instantiated_handler_registry.
 
     :param config:
         The dictionary containing changes to apply.  If a key is given
@@ -37,6 +37,12 @@ def update_configuration(config):
         instantiated_handler_registry.register_item(handler_name, instance)
 
 
+def flush_events():
+    for _, handler in instantiated_handler_registry.registered_items.items():
+        if hasattr(handler, 'flush'):
+            handler.flush()
+
+
 instantiated_handler_registry = DictRegistry()
 update_configuration(DEFAULT_CONFIG)
 
diff --git a/cloudinit/reporting/handlers.py b/cloudinit/reporting/handlers.py
index 4066076..6d23558 100644
--- a/cloudinit/reporting/handlers.py
+++ b/cloudinit/reporting/handlers.py
@@ -1,17 +1,32 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
 import abc
+import fcntl
 import json
 import six
+import os
+import re
+import struct
+import threading
+import time
 
 from cloudinit import log as logging
 from cloudinit.registry import DictRegistry
 from cloudinit import (url_helper, util)
+from datetime import datetime
 
+if six.PY2:
+    from multiprocessing.queues import JoinableQueue as JQueue
+else:
+    from queue import Queue as JQueue
 
 LOG = logging.getLogger(__name__)
 
 
+class ReportException(Exception):
+    pass
+
+
 @six.add_metaclass(abc.ABCMeta)
 class ReportingHandler(object):
     """Base class for report handlers.
@@ -24,6 +39,10 @@ class ReportingHandler(object):
     def publish_event(self, event):
         """Publish an event."""
 
+    def flush(self):
+        """Ensure ReportingHandler has published all events"""
+        pass
+
 
 class LogHandler(ReportingHandler):
     """Publishes events to the cloud-init log at the ``DEBUG`` log level."""
@@ -85,9 +104,236 @@ class WebHookHandler(ReportingHandler):
             LOG.warning("failed posting event: %s", event.as_string())
 
 
+class HyperVKvpReportingHandler(ReportingHandler):
+    """
+    Reports events to a Hyper-V host using Key-Value-Pair exchange protocol
+    and can be used to obtain high level diagnostic information from the host.
+
+    To use this facility, the KVP user-space daemon (hv_kvp_daemon) has to be
+    running. It reads the kvp_file when the host requests the guest to
+    enumerate the KVP's.
+
+    This reporter collates all events for a module (origin|name) in a single
+    json string in the dictionary.
+
+    For more information, see
+    https://technet.microsoft.com/en-us/library/dn798287.aspx#Linux%20guests
+    """
+    HV_KVP_EXCHANGE_MAX_VALUE_SIZE = 2048
+    HV_KVP_EXCHANGE_MAX_KEY_SIZE = 512
+    HV_KVP_RECORD_SIZE = (HV_KVP_EXCHANGE_MAX_KEY_SIZE +
+                          HV_KVP_EXCHANGE_MAX_VALUE_SIZE)
+    EVENT_PREFIX = 'CLOUD_INIT'
+    MSG_KEY = 'msg'
+    RESULT_KEY = 'result'
+    DESC_IDX_KEY = 'msg_i'
+    JSON_SEPARATORS = (',', ':')
+    KVP_POOL_FILE_GUEST = '/var/lib/hyperv/.kvp_pool_1'
+
+    def __init__(self,
+                 kvp_file_path=KVP_POOL_FILE_GUEST,
+                 event_types=None):
+        super(HyperVKvpReportingHandler, self).__init__()
+        self._kvp_file_path = kvp_file_path
+        self._event_types = event_types
+        self.q = JQueue()
+        self.kvp_file = None
+        self.incarnation_no = self._get_incarnation_no()
+        self.event_key_prefix = u"{0}|{1}".format(self.EVENT_PREFIX,
+                                                  self.incarnation_no)
+        self._current_offset = 0
+        self.publish_thread = threading.Thread(
+                target=self._publish_event_routine)
+        self.publish_thread.daemon = True
+        self.publish_thread.start()
+
+    def _get_incarnation_no(self):
+        """
+        use the time passed as the incarnation number.
+        the incarnation number is the number which are used to
+        distinguish the old data stored in kvp and the new data.
+        """
+        uptime_str = util.uptime()
+        try:
+            return int(time.time() - float(uptime_str))
+        except ValueError:
+            LOG.warning("uptime '%s' not in correct format.", uptime_str)
+            return 0
+
+    def _iterate_kvps(self, offset):
+        """iterate the kvp file from the current offset."""
+        try:
+            with open(self._kvp_file_path, 'rb+') as f:
+                self.kvp_file = f
+                fcntl.flock(f, fcntl.LOCK_EX)
+                f.seek(offset)
+                record_data = f.read(self.HV_KVP_RECORD_SIZE)
+                while len(record_data) == self.HV_KVP_RECORD_SIZE:
+                    self._current_offset += self.HV_KVP_RECORD_SIZE
+                    kvp_item = self._decode_kvp_item(record_data)
+                    yield kvp_item
+                    record_data = f.read(self.HV_KVP_RECORD_SIZE)
+                fcntl.flock(f, fcntl.LOCK_UN)
+        finally:
+            self.kvp_file = None
+
+    def _event_key(self, event):
+        """
+        the event key format is:
+        CLOUD_INIT|<incarnation number>|<event_type>|<event_name>
+        """
+        return u"{0}|{1}|{2}".format(self.event_key_prefix,
+                                     event.event_type, event.name)
+
+    def _encode_kvp_item(self, key, value):
+        data = (struct.pack("%ds%ds" % (
+                self.HV_KVP_EXCHANGE_MAX_KEY_SIZE,
+                self.HV_KVP_EXCHANGE_MAX_VALUE_SIZE),
+                key.encode('utf-8'), value.encode('utf-8')))
+        return data
+
+    def _decode_kvp_item(self, record_data):
+        record_data_len = len(record_data)
+        if record_data_len != self.HV_KVP_RECORD_SIZE:
+            raise ReportException(
+                "record_data len not correct {0} {1}."
+                .format(record_data_len, self.HV_KVP_RECORD_SIZE))
+        k = (record_data[0:self.HV_KVP_EXCHANGE_MAX_KEY_SIZE].decode('utf-8')
+                                                             .strip('\x00'))
+        v = (
+            record_data[
+                self.HV_KVP_EXCHANGE_MAX_KEY_SIZE:self.HV_KVP_RECORD_SIZE
+                ].decode('utf-8').strip('\x00'))
+
+        return {'key': k, 'value': v}
+
+    def _update_kvp_item(self, record_data):
+        if self.kvp_file is None:
+            raise ReportException(
+                "kvp file '{0}' not opened."
+                .format(self._kvp_file_path))
+        self.kvp_file.seek(-self.HV_KVP_RECORD_SIZE, 1)
+        self.kvp_file.write(record_data)
+
+    def _append_kvp_item(self, record_data):
+        with open(self._kvp_file_path, 'rb+') as f:
+            fcntl.flock(f, fcntl.LOCK_EX)
+            # seek to end of the file
+            f.seek(0, 2)
+            f.write(record_data)
+            f.flush()
+            fcntl.flock(f, fcntl.LOCK_UN)
+            self._current_offset = f.tell()
+
+    def _break_down(self, key, meta_data, description):
+        del meta_data[self.MSG_KEY]
+        des_in_json = json.dumps(description)
+        des_in_json = des_in_json[1:(len(des_in_json) - 1)]
+        i = 0
+        result_array = []
+        message_place_holder = "\"" + self.MSG_KEY + "\":\"\""
+        while True:
+            meta_data[self.DESC_IDX_KEY] = i
+            meta_data[self.MSG_KEY] = ''
+            data_without_desc = json.dumps(meta_data,
+                                           separators=self.JSON_SEPARATORS)
+            room_for_desc = (
+                self.HV_KVP_EXCHANGE_MAX_VALUE_SIZE -
+                len(data_without_desc) - 8)
+            value = data_without_desc.replace(
+                message_place_holder,
+                '"{key}":"{desc}"'.format(
+                    key=self.MSG_KEY, desc=des_in_json[:room_for_desc]))
+            result_array.append(self._encode_kvp_item(key, value))
+            i += 1
+            des_in_json = des_in_json[room_for_desc:]
+            if len(des_in_json) == 0:
+                break
+        return result_array
+
+    def _encode_event(self, event):
+        """
+        encode the event into kvp data bytes.
+        if the event content reaches the maximum length of kvp value.
+        then it would be cut to multiple slices.
+        """
+        key = self._event_key(event)
+        meta_data = {
+                "name": event.name,
+                "type": event.event_type,
+                "ts": (datetime.utcfromtimestamp(event.timestamp)
+                       .isoformat() + 'Z'),
+                }
+        if hasattr(event, self.RESULT_KEY):
+            meta_data[self.RESULT_KEY] = event.result
+        meta_data[self.MSG_KEY] = event.description
+        value = json.dumps(meta_data, separators=self.JSON_SEPARATORS)
+        # if it reaches the maximum length of kvp value,
+        # break it down to slices.
+        # this should be very corner case.
+        if len(value) > self.HV_KVP_EXCHANGE_MAX_VALUE_SIZE:
+            return self._break_down(key, meta_data, event.description)
+        else:
+            data = self._encode_kvp_item(key, value)
+            return [data]
+
+    def _publish_event_routine(self):
+        while True:
+            try:
+                event = self.q.get(block=True)
+                need_append = True
+                try:
+                    if not os.path.exists(self._kvp_file_path):
+                        LOG.warning(
+                            "skip writing events %s to %s. file not present.",
+                            event.as_string(),
+                            self._kvp_file_path)
+                    encoded_event = self._encode_event(event)
+                    # for each encoded_event
+                    for encoded_data in (encoded_event):
+                        for kvp in self._iterate_kvps(self._current_offset):
+                            match = (
+                                re.match(
+                                    r"^{0}\|(\d+)\|.+"
+                                    .format(self.EVENT_PREFIX),
+                                    kvp['key']
+                                ))
+                            if match:
+                                match_groups = match.groups(0)
+                                if int(match_groups[0]) < self.incarnation_no:
+                                    need_append = False
+                                    self._update_kvp_item(encoded_data)
+                                    continue
+                        if need_append:
+                            self._append_kvp_item(encoded_data)
+                except IOError as e:
+                    LOG.warning(
+                        "failed posting event to kvp: %s e:%s",
+                        event.as_string(), e)
+                finally:
+                    self.q.task_done()
+
+            # when main process exits, q.get() will through EOFError
+            # indicating we should exit this thread.
+            except EOFError:
+                return
+
+    # since the saving to the kvp pool can be a time costing task
+    # if the kvp pool already contains a chunk of data,
+    # so defer it to another thread.
+    def publish_event(self, event):
+        if (not self._event_types or event.event_type in self._event_types):
+            self.q.put(event)
+
+    def flush(self):
+        LOG.debug('HyperVReportingHandler flushing remaining events')
+        self.q.join()
+
+
 available_handlers = DictRegistry()
 available_handlers.register_item('log', LogHandler)
 available_handlers.register_item('print', PrintHandler)
 available_handlers.register_item('webhook', WebHookHandler)
+available_handlers.register_item('hyperv', HyperVKvpReportingHandler)
 
 # vi: ts=4 expandtab
diff --git a/cloudinit/sources/helpers/vmware/imc/config_nic.py b/cloudinit/sources/helpers/vmware/imc/config_nic.py
index 3ef8c62..e1890e2 100644
--- a/cloudinit/sources/helpers/vmware/imc/config_nic.py
+++ b/cloudinit/sources/helpers/vmware/imc/config_nic.py
@@ -164,7 +164,7 @@ class NicConfigurator(object):
             return ([subnet], route_list)
 
         # Add routes if there is no primary nic
-        if not self._primaryNic:
+        if not self._primaryNic and v4.gateways:
             route_list.extend(self.gen_ipv4_route(nic,
                                                   v4.gateways,
                                                   v4.netmask))
diff --git a/cloudinit/stages.py b/cloudinit/stages.py
index c132b57..8874d40 100644
--- a/cloudinit/stages.py
+++ b/cloudinit/stages.py
@@ -510,7 +510,7 @@ class Init(object):
                 # The default frequency if handlers don't have one
                 'frequency': frequency,
                 # This will be used when new handlers are found
-                # to help write there contents to files with numbered
+                # to help write their contents to files with numbered
                 # names...
                 'handlercount': 0,
                 'excluded': excluded,
diff --git a/cloudinit/tests/helpers.py b/cloudinit/tests/helpers.py
index 5bfe7fa..de24e25 100644
--- a/cloudinit/tests/helpers.py
+++ b/cloudinit/tests/helpers.py
@@ -10,7 +10,6 @@ import shutil
 import sys
 import tempfile
 import time
-import unittest
 
 import mock
 import six
@@ -28,11 +27,15 @@ except ImportError:
 
 from cloudinit.config.schema import (
     SchemaValidationError, validate_cloudconfig_schema)
+from cloudinit import cloud
+from cloudinit import distros
 from cloudinit import helpers as ch
+from cloudinit.sources import DataSourceNone
 from cloudinit import util
 
 # Used for skipping tests
 SkipTest = unittest2.SkipTest
+skipIf = unittest2.skipIf
 
 # Used for detecting different python versions
 PY2 = False
@@ -187,6 +190,29 @@ class CiTestCase(TestCase):
         """
         raise SystemExit(code)
 
+    def tmp_cloud(self, distro, sys_cfg=None, metadata=None):
+        """Create a cloud with tmp working directory paths.
+
+        @param distro: Name of the distro to attach to the cloud.
+        @param metadata: Optional metadata to set on the datasource.
+
+        @return: The built cloud instance.
+        """
+        self.new_root = self.tmp_dir()
+        if not sys_cfg:
+            sys_cfg = {}
+        tmp_paths = {}
+        for var in ['templates_dir', 'run_dir', 'cloud_dir']:
+            tmp_paths[var] = self.tmp_path(var, dir=self.new_root)
+            util.ensure_dir(tmp_paths[var])
+        self.paths = ch.Paths(tmp_paths)
+        cls = distros.fetch(distro)
+        mydist = cls(distro, sys_cfg, self.paths)
+        myds = DataSourceNone.DataSourceNone(sys_cfg, mydist, self.paths)
+        if metadata:
+            myds.metadata.update(metadata)
+        return cloud.Cloud(myds, self.paths, sys_cfg, mydist, None)
+
 
 class ResourceUsingTestCase(CiTestCase):
 
@@ -426,21 +452,6 @@ def readResource(name, mode='r'):
 
 
 try:
-    skipIf = unittest.skipIf
-except AttributeError:
-    # Python 2.6.  Doesn't have to be high fidelity.
-    def skipIf(condition, reason):
-        def decorator(func):
-            def wrapper(*args, **kws):
-                if condition:
-                    return func(*args, **kws)
-                else:
-                    print(reason, file=sys.stderr)
-            return wrapper
-        return decorator
-
-
-try:
     import jsonschema
     assert jsonschema  # avoid pyflakes error F401: import unused
     _missing_jsonschema_dep = False
diff --git a/debian/changelog b/debian/changelog
index ea38c3b..f303d70 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,17 @@
+cloud-init (18.3-35-g3f6d0972-0ubuntu1) cosmic; urgency=medium
+
+  * New upstream snapshot.
+    - Add unit tests for config/cc_ssh.py [Francis Ginther]
+    - Fix the built-in cloudinit/tests/helpers:skipIf
+    - read-version: enhance error message [Joshua Powers]
+    - hyperv_reporting_handler: simplify threaded publisher
+    - VMWare: Fix a network config bug in vm with static IPv4 and no gateway.
+      [Pengpeng Sun]
+    - logging: Add logging config type hyperv for reporting via Azure KVP
+      [Andy Liu]
+
+ -- Chad Smith <[e-mail]>  Sat, 01 Sep 2018 12:08:52 -0600
+
 cloud-init (18.3-29-gdab59087-0ubuntu1) cosmic; urgency=medium
 
   * New upstream snapshot.
diff --git a/tests/unittests/test_reporting_hyperv.py b/tests/unittests/test_reporting_hyperv.py
new file mode 100644
index 0000000..2e64c6c
--- /dev/null
+++ b/tests/unittests/test_reporting_hyperv.py
@@ -0,0 +1,134 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+from cloudinit.reporting import events
+from cloudinit.reporting import handlers
+
+import json
+import os
+
+from cloudinit import util
+from cloudinit.tests.helpers import CiTestCase
+
+
+class TestKvpEncoding(CiTestCase):
+    def test_encode_decode(self):
+        kvp = {'key': 'key1', 'value': 'value1'}
+        kvp_reporting = handlers.HyperVKvpReportingHandler()
+        data = kvp_reporting._encode_kvp_item(kvp['key'], kvp['value'])
+        self.assertEqual(len(data), kvp_reporting.HV_KVP_RECORD_SIZE)
+        decoded_kvp = kvp_reporting._decode_kvp_item(data)
+        self.assertEqual(kvp, decoded_kvp)
+
+
+class TextKvpReporter(CiTestCase):
+    def setUp(self):
+        super(TextKvpReporter, self).setUp()
+        self.tmp_file_path = self.tmp_path('kvp_pool_file')
+        util.ensure_file(self.tmp_file_path)
+
+    def test_event_type_can_be_filtered(self):
+        reporter = handlers.HyperVKvpReportingHandler(
+            kvp_file_path=self.tmp_file_path,
+            event_types=['foo', 'bar'])
+
+        reporter.publish_event(
+            events.ReportingEvent('foo', 'name', 'description'))
+        reporter.publish_event(
+            events.ReportingEvent('some_other', 'name', 'description3'))
+        reporter.q.join()
+
+        kvps = list(reporter._iterate_kvps(0))
+        self.assertEqual(1, len(kvps))
+
+        reporter.publish_event(
+            events.ReportingEvent('bar', 'name', 'description2'))
+        reporter.q.join()
+        kvps = list(reporter._iterate_kvps(0))
+        self.assertEqual(2, len(kvps))
+
+        self.assertIn('foo', kvps[0]['key'])
+        self.assertIn('bar', kvps[1]['key'])
+        self.assertNotIn('some_other', kvps[0]['key'])
+        self.assertNotIn('some_other', kvps[1]['key'])
+
+    def test_events_are_over_written(self):
+        reporter = handlers.HyperVKvpReportingHandler(
+            kvp_file_path=self.tmp_file_path)
+
+        self.assertEqual(0, len(list(reporter._iterate_kvps(0))))
+
+        reporter.publish_event(
+            events.ReportingEvent('foo', 'name1', 'description'))
+        reporter.publish_event(
+            events.ReportingEvent('foo', 'name2', 'description'))
+        reporter.q.join()
+        self.assertEqual(2, len(list(reporter._iterate_kvps(0))))
+
+        reporter2 = handlers.HyperVKvpReportingHandler(
+            kvp_file_path=self.tmp_file_path)
+        reporter2.incarnation_no = reporter.incarnation_no + 1
+        reporter2.publish_event(
+            events.ReportingEvent('foo', 'name3', 'description'))
+        reporter2.q.join()
+
+        self.assertEqual(2, len(list(reporter2._iterate_kvps(0))))
+
+    def test_events_with_higher_incarnation_not_over_written(self):
+        reporter = handlers.HyperVKvpReportingHandler(
+            kvp_file_path=self.tmp_file_path)
+
+        self.assertEqual(0, len(list(reporter._iterate_kvps(0))))
+
+        reporter.publish_event(
+            events.ReportingEvent('foo', 'name1', 'description'))
+        reporter.publish_event(
+            events.ReportingEvent('foo', 'name2', 'description'))
+        reporter.q.join()
+        self.assertEqual(2, len(list(reporter._iterate_kvps(0))))
+
+        reporter3 = handlers.HyperVKvpReportingHandler(
+            kvp_file_path=self.tmp_file_path)
+        reporter3.incarnation_no = reporter.incarnation_no - 1
+        reporter3.publish_event(
+            events.ReportingEvent('foo', 'name3', 'description'))
+        reporter3.q.join()
+        self.assertEqual(3, len(list(reporter3._iterate_kvps(0))))
+
+    def test_finish_event_result_is_logged(self):
+        reporter = handlers.HyperVKvpReportingHandler(
+            kvp_file_path=self.tmp_file_path)
+        reporter.publish_event(
+            events.FinishReportingEvent('name2', 'description1',
+                                        result=events.status.FAIL))
+        reporter.q.join()
+        self.assertIn('FAIL', list(reporter._iterate_kvps(0))[0]['value'])
+
+    def test_file_operation_issue(self):
+        os.remove(self.tmp_file_path)
+        reporter = handlers.HyperVKvpReportingHandler(
+            kvp_file_path=self.tmp_file_path)
+        reporter.publish_event(
+            events.FinishReportingEvent('name2', 'description1',
+                                        result=events.status.FAIL))
+        reporter.q.join()
+
+    def test_event_very_long(self):
+        reporter = handlers.HyperVKvpReportingHandler(
+            kvp_file_path=self.tmp_file_path)
+        description = 'ab' * reporter.HV_KVP_EXCHANGE_MAX_VALUE_SIZE
+        long_event = events.FinishReportingEvent(
+            'event_name',
+            description,
+            result=events.status.FAIL)
+        reporter.publish_event(long_event)
+        reporter.q.join()
+        kvps = list(reporter._iterate_kvps(0))
+        self.assertEqual(3, len(kvps))
+
+        # restore from the kvp to see the content are all there
+        full_description = ''
+        for i in range(len(kvps)):
+            msg_slice = json.loads(kvps[i]['value'])
+            self.assertEqual(msg_slice['msg_i'], i)
+            full_description += msg_slice['msg']
+        self.assertEqual(description, full_description)
diff --git a/tests/unittests/test_vmware_config_file.py b/tests/unittests/test_vmware_config_file.py
index 036f687..602dedb 100644
--- a/tests/unittests/test_vmware_config_file.py
+++ b/tests/unittests/test_vmware_config_file.py
@@ -2,11 +2,15 @@
 # Copyright (C) 2016 VMware INC.
 #
 # Author: Sankar Tanguturi <[e-mail]>
+#         Pengpeng Sun <[e-mail]>
 #
 # This file is part of cloud-init. See LICENSE file for license information.
 
 import logging
+import os
 import sys
+import tempfile
+import textwrap
 
 from cloudinit.sources.DataSourceOVF import get_network_config_from_conf
 from cloudinit.sources.DataSourceOVF import read_vmware_imc
@@ -343,4 +347,115 @@ class TestVmwareConfigFile(CiTestCase):
         conf = Config(cf)
         self.assertEqual("test-script", conf.custom_script_name)
 
+
+class TestVmwareNetConfig(CiTestCase):
+    """Test conversion of vmware config to cloud-init config."""
+
+    def _get_NicConfigurator(self, text):
+        fp = None
+        try:
+            with tempfile.NamedTemporaryFile(mode="w", dir=self.tmp_dir(),
+                                             delete=False) as fp:
+                fp.write(text)
+                fp.close()
+            cfg = Config(ConfigFile(fp.name))
+            return NicConfigurator(cfg.nics, use_system_devices=False)
+        finally:
+            if fp:
+                os.unlink(fp.name)
+
+    def test_non_primary_nic_without_gateway(self):
+        """A non primary nic set is not required to have a gateway."""
+        config = textwrap.dedent("""\
+            [NETWORK]
+            NETWORKING = yes
+            BOOTPROTO = dhcp
+            HOSTNAME = myhost1
+            DOMAINNAME = eng.vmware.com
+
+            [NIC-CONFIG]
+            NICS = NIC1
+
+            [NIC1]
+            MACADDR = 00:50:56:a6:8c:08
+            ONBOOT = yes
+            IPv4_MODE = BACKWARDS_COMPATIBLE
+            BOOTPROTO = static
+            IPADDR = 10.20.87.154
+            NETMASK = 255.255.252.0
+            """)
+        nc = self._get_NicConfigurator(config)
+        self.assertEqual(
+            [{'type': 'physical', 'name': 'NIC1',
+              'mac_address': '00:50:56:a6:8c:08',
+              'subnets': [
+                  {'control': 'auto', 'type': 'static',
+                   'address': '10.20.87.154', 'netmask': '255.255.252.0'}]}],
+            nc.generate())
+
+    def test_non_primary_nic_with_gateway(self):
+        """A non primary nic set can have a gateway."""
+        config = textwrap.dedent("""\
+            [NETWORK]
+            NETWORKING = yes
+            BOOTPROTO = dhcp
+            HOSTNAME = myhost1
+            DOMAINNAME = eng.vmware.com
+
+            [NIC-CONFIG]
+            NICS = NIC1
+
+            [NIC1]
+            MACADDR = 00:50:56:a6:8c:08
+            ONBOOT = yes
+            IPv4_MODE = BACKWARDS_COMPATIBLE
+            BOOTPROTO = static
+            IPADDR = 10.20.87.154
+            NETMASK = 255.255.252.0
+            GATEWAY = 10.20.87.253
+            """)
+        nc = self._get_NicConfigurator(config)
+        self.assertEqual(
+            [{'type': 'physical', 'name': 'NIC1',
+              'mac_address': '00:50:56:a6:8c:08',
+              'subnets': [
+                  {'control': 'auto', 'type': 'static',
+                   'address': '10.20.87.154', 'netmask': '255.255.252.0'}]},
+             {'type': 'route', 'destination': '10.20.84.0/22',
+              'gateway': '10.20.87.253', 'metric': 10000}],
+            nc.generate())
+
+    def test_a_primary_nic_with_gateway(self):
+        """A primary nic set can have a gateway."""
+        config = textwrap.dedent("""\
+            [NETWORK]
+            NETWORKING = yes
+            BOOTPROTO = dhcp
+            HOSTNAME = myhost1
+            DOMAINNAME = eng.vmware.com
+
+            [NIC-CONFIG]
+            NICS = NIC1
+
+            [NIC1]
+            MACADDR = 00:50:56:a6:8c:08
+            ONBOOT = yes
+            IPv4_MODE = BACKWARDS_COMPATIBLE
+            BOOTPROTO = static
+            IPADDR = 10.20.87.154
+            NETMASK = 255.255.252.0
+            PRIMARY = true
+            GATEWAY = 10.20.87.253
+            """)
+        nc = self._get_NicConfigurator(config)
+        self.assertEqual(
+            [{'type': 'physical', 'name': 'NIC1',
+              'mac_address': '00:50:56:a6:8c:08',
+              'subnets': [
+                  {'control': 'auto', 'type': 'static',
+                   'address': '10.20.87.154', 'netmask': '255.255.252.0',
+                   'gateway': '10.20.87.253'}]}],
+            nc.generate())
+
+
 # vi: ts=4 expandtab
diff --git a/tools/read-version b/tools/read-version
index 3ea9e66..e69c2ce 100755
--- a/tools/read-version
+++ b/tools/read-version
@@ -76,6 +76,12 @@ if is_gitdir(_tdir) and which("git"):
     if not version.startswith(src_version):
         sys.stderr.write("git describe version (%s) differs from "
                          "cloudinit.version (%s)\n" % (version, src_version))
+        sys.stderr.write(
+            "Please get the latest upstream tags.\n"
+            "As an example, this can be done with the following:\n"
+            "$ git remote add upstream https://git.launchpad.net/cloud-init\n"
+            "$ git fetch upstream --tags\n"
+        )
         sys.exit(1)
 
     version_long = tiny_p(cmd + ["--long"]).strip()