cloud-init-dev team mailing list archive
-
cloud-init-dev team
-
Mailing list archive
-
Message #05228
[Merge] ~andyliuliming/cloud-init:reporting into cloud-init:master
Andy has proposed merging ~andyliuliming/cloud-init:reporting into cloud-init:master.
Requested reviews:
cloud-init commiters (cloud-init-dev)
For more details, see:
https://code.launchpad.net/~andyliuliming/cloud-init/+git/cloud-init/+merge/351742
added one option to send the cloud init events to the kvp in hyper-v.
--
Your team cloud-init commiters is requested to review the proposed merge of ~andyliuliming/cloud-init:reporting into cloud-init:master.
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..c0edee1 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:
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..e047767 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
diff --git a/cloudinit/reporting/events.py b/cloudinit/reporting/events.py
index e5dfab3..b47339e 100644
--- a/cloudinit/reporting/events.py
+++ b/cloudinit/reporting/events.py
@@ -29,7 +29,6 @@ class _nameset(set):
status = _nameset(("SUCCESS", "WARN", "FAIL"))
-
class ReportingEvent(object):
"""Encapsulation of event formatting."""
diff --git a/cloudinit/reporting/handlers.py b/cloudinit/reporting/handlers.py
index 4066076..df745e3 100644
--- a/cloudinit/reporting/handlers.py
+++ b/cloudinit/reporting/handlers.py
@@ -1,16 +1,30 @@
# This file is part of cloud-init. See LICENSE file for license information.
import abc
+import fcntl
+import hashlib
import json
+import os
import six
+if six.PY2:
+ import Queue as queue
+else:
+ import queue
+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
LOG = logging.getLogger(__name__)
+class ReportException(Exception):
+ def __init__(self, message):
+ super().__init__(message)
@six.add_metaclass(abc.ABCMeta)
class ReportingHandler(object):
@@ -85,9 +99,173 @@ 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'
+ def __init__(self,
+ kvp_file='/var/lib/hyperv/.kvp_pool_1',
+ event_types=None):
+ super(HyperVKvpReportingHandler, self).__init__()
+ self._kvp_file = kvp_file
+ self._event_types = event_types
+ self.running = False
+ self.queue_lock = threading.Lock()
+ self.running_lock = threading.Lock()
+ self.q = queue.Queue(100)
+ self.kvp_file = None
+ self.incarnation_no = self._get_incarnation_no()
+ # use the current time stamp as the incarnation number.
+ self.event_key_prefix = u"{0}|{1}".format(self.EVENT_PREFIX, self.incarnation_no)
+ self._current_offset = 0
+
+ def _get_incarnation_no(self):
+ try:
+ uptime_str = util.uptime()
+ return int(time.time() - float(uptime_str))
+ except ValueError:
+ LOG.warn("uptime not in correct format.")
+ return 0
+
+ def _open_kvp_file(self):
+ f = open(self._kvp_file, 'rb+')
+ return f
+
+ def _iterate_kvps(self, offset=0):
+ """
+ iterate the kvp file from the current offset.
+ """
+ try:
+ with self._open_kvp_file() 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)
+ 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):
+ if len(record_data) != self.HV_KVP_RECORD_SIZE:
+ raise ReportException("record_data len not correct.")
+ 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 == None:
+ raise ReportException("kvp file not opened.")
+ self.kvp_file.seek(-self.HV_KVP_RECORD_SIZE, 1)
+ self.kvp_file.write(record_data)
+
+ def _encode_event(self, event):
+ key = self._event_key(event)
+ # TODO trunc the description so the data would be in a correct json format.
+ data = {
+ "type": event.event_type,
+ "name": event.name,
+ "ts": (datetime.utcfromtimestamp(event.timestamp)
+ .isoformat() + 'Z'),
+ "msg": event.description
+ }
+ if hasattr(event, 'result'):
+ data['result'] = event.result
+ value = json.dumps(data, separators=(',', ':'))
+ data = self._encode_kvp_item(key, value)
+ return data
+
+ def _append_event(self, encoded_event):
+ with self._open_kvp_file() as f:
+ fcntl.flock(f, fcntl.LOCK_EX)
+ # seek to end of the file
+ f.seek(0, 2)
+ f.write(encoded_event)
+ f.flush()
+ self._current_offset = f.tell()
+
+ def _publish_event_routine(self):
+ while True:
+ try:
+ # acquire the lock.
+ event = self.q.get_nowait()
+ need_append = True
+ encoded_event = self._encode_event(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_event)
+ break
+ if need_append:
+ self._append_event(encoded_event)
+ self.q.task_done()
+ except queue.Empty:
+ with self.queue_lock:
+ # double check the queue is empty
+ if self.q.empty():
+ self.running = False
+ break
+
+ def trigger_publish_event(self):
+ if not self.running:
+ with self.running_lock:
+ if not self.running:
+ self.running = True
+ thread = threading.Thread(target = self._publish_event_routine)
+ thread.start()
+
+ # 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):
+ with self.queue_lock:
+ self.q.put(event)
+ self.trigger_publish_event()
+
+
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/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/util.py b/cloudinit/util.py
index 5068096..29fffb8 100644
--- a/cloudinit/util.py
+++ b/cloudinit/util.py
@@ -2031,6 +2031,24 @@ def subp(args, data=None, rcs=None, env=None, capture=True,
stderr=stderr, stdin=stdin,
env=env, shell=shell)
(out, err) = sp.communicate(data)
+
+ LOG.debug("out=%s", out)
+ LOG.debug("err=%s", err)
+ LOG.debug("rc=%d", sp.returncode)
+
+ # Just ensure blank instead of none.
+ if not out and capture:
+ out = b''
+ if not err and capture:
+ err = b''
+ if decode:
+ def ldecode(data, m='utf-8'):
+ if not isinstance(data, bytes):
+ return data
+ return data.decode(m, decode)
+
+ out = ldecode(out)
+ err = ldecode(err)
except OSError as e:
if status_cb:
status_cb('ERROR: End run command: invalid command provided\n')
diff --git a/tests/unittests/test_reporting_hyperv.py b/tests/unittests/test_reporting_hyperv.py
new file mode 100644
index 0000000..6801bfd
--- /dev/null
+++ b/tests/unittests/test_reporting_hyperv.py
@@ -0,0 +1,112 @@
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of cloud-init. See LICENSE file for license information.
+
+from cloudinit.reporting import events
+from cloudinit.reporting import handlers
+
+import mock
+import os
+import tempfile
+import time
+import uuid
+
+from cloudinit.tests.helpers import TestCase
+
+class TestKvpEncoding(TestCase):
+ 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(TestCase):
+ def test_event_type_can_be_filtered(self):
+ tmp_file = tempfile.NamedTemporaryFile(delete=False)
+ tmp_file.close()
+
+ reporter = handlers.HyperVKvpReportingHandler(
+ kvp_file=tmp_file.name,
+ 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())
+ self.assertEqual(1, len(kvps))
+
+ reporter.publish_event(
+ events.ReportingEvent('bar', 'name', 'description2'))
+ reporter.q.join()
+ kvps = list(reporter._iterate_kvps())
+ 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'])
+
+ os.remove(tmp_file.name)
+
+ def test_events_are_over_written(self):
+ tmp_file = tempfile.NamedTemporaryFile(delete=False)
+ tmp_file.close()
+ reporter = handlers.HyperVKvpReportingHandler(kvp_file=tmp_file.name)
+
+ self.assertEqual(0, len(list(reporter._iterate_kvps())))
+
+ 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())))
+
+ reporter2 = handlers.HyperVKvpReportingHandler(kvp_file=tmp_file.name)
+ 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())))
+
+ os.remove(tmp_file.name)
+
+ def test_events_with_higher_incarnation_not_over_written(self):
+ tmp_file = tempfile.NamedTemporaryFile(delete=False)
+ tmp_file.close()
+ reporter = handlers.HyperVKvpReportingHandler(kvp_file=tmp_file.name)
+
+ self.assertEqual(0, len(list(reporter._iterate_kvps())))
+
+ 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())))
+
+ reporter3 = handlers.HyperVKvpReportingHandler(kvp_file=tmp_file.name)
+ 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())))
+
+ os.remove(tmp_file.name)
+
+ def test_finish_event_result_is_logged(self):
+ tmp_file = tempfile.NamedTemporaryFile(delete=False)
+ tmp_file.close()
+ reporter = handlers.HyperVKvpReportingHandler(kvp_file=tmp_file.name)
+ reporter.publish_event(
+ events.FinishReportingEvent('name2', 'description1',
+ result=events.status.FAIL))
+ reporter.q.join()
+ self.assertIn('FAIL', list(reporter._iterate_kvps())[0]['value'])
+ os.remove(tmp_file.name)
\ No newline at end of file
diff --git a/tools/read-dependencies b/tools/read-dependencies
index b4656e6..84d5a16 100755
--- a/tools/read-dependencies
+++ b/tools/read-dependencies
@@ -223,6 +223,7 @@ def main(distro):
renames = deps_from_json.get('renames', {})
translated_pip_names = translate_pip_to_system_pkg(
pip_pkg_names, renames, args.python_version)
+
all_deps = []
if args.distro:
all_deps.extend(
@@ -233,6 +234,7 @@ def main(distro):
all_deps = translated_pip_names
else:
all_deps = pip_pkg_names
+
if args.install:
pkg_install(all_deps, args.distro, args.test_distro, args.dry_run)
else:
Follow ups
